apache / apache/datafusion-comet
Native Parquet scan multiplies rows for a struct with duplicate field names
- 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
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