apache / apache/datafusion-comet

Native Parquet scan multiplies rows for a struct with duplicate field names

Open
#5,783 0 comments 0 reactions 0 assignees View on GitHub
area:scan bug correctness native_datafusion priority:critical requires-triage
Dominant language
Scala
Stars
1.3k
Forks
373
Avg merge
2d 4h
Merged PRs (30d)
198

Description

### Describe the bug

When a Parquet file contains a struct with two or more byte-identical child field names, Comet's native scan returns one output row per (input row × matching leaf) instead of one row per input row. The declared output schema still has a single field, so the extra rows appear with no error and no warning.

Spark returns the correct row count, resolving the duplicate name to a single child.

Three shapes, all under `spark.sql.caseSensitive=true`:

| Struct in the file | Spark | Comet |
| --- | --- | --- |
| `struct` | 3 rows | **6 rows** |
| `struct` | 3 rows | **9 rows** |
| `struct` | 3 rows | **error** (`StructArrayReader out of sync`) |

### Steps to reproduce

Reproduced on `bc74cc79f` (current `main`), Spark 4.1, JDK 17, macOS aarch64.

```scala
withTempPath { path =>
spark
.range(3)
.selectExpr("named_struct('dup', id, 'dup', id + 100) as s")
.write
.mode("overwrite")
.parquet(path.toString)

// An explicit read schema is required. Spark blocks schema *inference* on duplicate
// nested names with COLUMN_ALREADY_EXISTS, so this is only reachable when the schema
// is declared (spark.read.schema(...), or a table with a declared schema).
spark.read.schema("s struct").parquet(path.toString).collect()
}
```

Observed, with `CometNativeScanExec` in the plan:

```
spark (comet disabled) -> count=3 rows=[[2]] [[1]] [[0]]
comet -> count=6 rows=[[2]] [[102]] [[1]] [[101]] [[0]] [[100]]
```

The multiplier tracks the number of duplicate children, so three `dup` children give nine rows:

```
named_struct('dup', id, 'dup', id + 100, 'dup', id + 200) as s // read as s struct

spark -> count=3 rows=[[1]] [[2]] [[0]]
comet -> count=9 rows=[[1]] [[101]] [[201]] [[2]] [[102]] [[202]] [[0]] [[100]] [[200]]
```

Adding a non-duplicate sibling turns it into a hard error rather than wrong results:

```
named_struct('dup', id, 'dup', id + 100, 'other', id + 900) as s
// read as s struct

spark -> count=3 rows=[[2,902]] [[1,901]] [[0,900]]
comet -> SparkException: Arrow error: Parquet argument error: Parquet error:
StructArrayReader out of sync in read_records, expected 1 read, got 0
```

### Expected behavior

Match Spark: one row per input row, resolving the duplicate name to a single child. Spark's `ParquetReadSupport` builds `caseSensitiveParquetFieldMap` with `.toMap`, so the last child with a given name wins.

Failing with a clear error would also be acceptable, and is strictly better than silently changing the row count.

### Additional context

- **Scope is duplicates inside a struct.** Top-level duplicate column names are not reachable this way, because Spark's own writer rejects them with `COLUMN_ALREADY_EXISTS`. `named_struct` permits duplicate field names and the writer accepts them, so a plain Spark job can produce a triggering file.
- Comet's nested convert in `native/core/src/parquet/parquet_support.rs` does have a duplicate-match check, but it is deliberately gated on `!case_sensitive` (a case-sensitive collision means byte-identical names, where an error saying "in case-insensitive mode" would be wrong). The case-sensitive path falls through to `indices[0]`, i.e. first-wins, which is itself a smaller divergence from Spark's last-wins. The row multiplication happens before that, so the projection appears to match both leaves independently.
- Found while reviewing #5751 and #5707, which cover the *case-insensitive* ambiguity in the same match arm. Not caused by that PR, whose diff is test-only.
- Related but a different code path: #5605 (native shuffle accepts structs with duplicate field names, then fails importing the batch back to the JVM).
- I have applied `priority:critical` because the label covers "silent wrong results", but the trigger is narrow (duplicate names inside a struct plus an explicit read schema). Happy for triage to re-rank.

Contributor guide

Open the contributing guide

Research direction

Start in native/core/src/parquet/parquet_support.rs, especially the nested convert duplicate-match check and the native Parquet scan path. Run the Scala reproduction with an explicitly declared nested schema and compare Comet with Spark. Done means duplicate children no longer multiply rows, and the mixed-duplicate case no longer produces StructArrayReader out of sync or silently incorrect results.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust, scala, spark
Domain
backend, data-engineering
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Clearly specified
Newbie friendliness
68/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.