comphead opened a new pull request, #6004:
URL: https://github.com/apache/datafusion-comet/pull/6004
## Which issue does this PR close?
Closes #5801.
Also closes the duplicate-name half of #5605 in the two places it was still
reachable (any operator, and the Arrow cache serializer), so I have referenced
rather than closed that one.
This does **not** close #5783. See the note at the end for why a plan-time
check cannot reach it.
## Rationale for this change
Two separate things make a struct's children impossible to resolve
one-to-one, and Comet handled each of them in only part of the code.
**Duplicate child names.** Java Arrow keys a struct vector's children by
name, so `struct<a, a>` loses a child on the way back across the C data
interface (#5605). `DataTypeSupport` declines the shape, which covers the scan
type checker, the native shuffle predicate and the two row-conversion sinks.
Nothing covers the remaining operators, and nothing covered the Arrow cache
serializer: `ArrowCachedBatchSerializer.supportsType` accepted `struct<a, a>`,
so
```scala
spark
.sql("SELECT id AS key, named_struct('a', id, 'a', id + 1) AS st FROM
range(1000)")
.createOrReplaceTempView("t")
spark.catalog.cacheTable("t")
```
stored the relation as a `CometCachedBatch`, a format the native scan over
that cache can never import back.
**Duplicate Parquet field ids.** Under
`spark.sql.parquet.fieldId.read.enabled` Spark resolves each requested field to
the one Parquet field carrying its id, and raises
`FOUND_DUPLICATE_FIELD_IN_FIELD_ID_LOOKUP_MODE` when more than one answers.
Comet never looked at field ids at all. DataFusion 55's opener skips the
expression adapter when the file's physical schema compares equal to the
logical schema and no predicate is pushed, so the file in #5801 was read
positionally and Comet returned rows where Spark raises.
## What changes are included in this PR?
1. **`DataTypeSupport`** gains `hasDuplicateFieldNames` plus two recursive
checks, `findDuplicateStructFieldNames` and `findDuplicateStructFieldIds`, that
descend through structs, arrays and maps and describe the first offending
struct with its path. The existing trait check now calls the shared predicate,
so the definition of "duplicate" lives in one place.
Only nested structs are inspected. A plan's top-level output attributes
routinely repeat a name (a self-join gives two `id` columns, told apart by
expression id) and Comet matches top-level columns positionally, so those are
not duplicates in this sense.
2. **`CometExecRule.tryConvertToComet`** declines any operator whose own
output, or a data-producing child's output, carries a duplicate-named struct.
This sits beside the existing `VariantType` gate, on the one path every
operator conversion takes, so the invariant holds for every serde rather than
for the four places that happened to check.
3. **`CometScanTypeChecker`** declines a scan whose requested schema repeats
a Parquet field id, when field id matching is on. The read goes back to Spark,
which raises the ambiguity error. With field id matching off the two fields are
told apart by name and the scan stays native.
4. **`ArrowCachedBatchSerializer.supportsType`** rejects duplicate child
names, so such a relation is cached in Spark's default format instead of a
Comet format that can never be read back natively.
## How are these changes tested?
New and updated tests, all on the Spark 4.1 profile:
- **`DataTypeSupportSuite`** (new): both recursive checks at every nesting
level (struct child, array element, map key, map value, two levels deep), the
accepted cases including names that differ only by case, a non-integral field
id, and that the name check and the id check are independent of each other.
- **`CometNativeReaderSuite`**: #5801 end to end. A Parquet file written
with no key-value metadata and schema `s<x id=1, y id=1>`, read back with that
same schema. Comet now raises Spark's `Found duplicate field(s) "1": [x, y] in
id mapping mode` and the plan carries no `CometNativeScanExec`. With
`spark.sql.parquet.fieldId.read.enabled=false` the scan stays native and
returns the row, which pins that the new gate is scoped to field id matching.
- **`CometInMemoryCacheSuite`**: a relation with `named_struct('a', id, 'a',
id + 1)` is cached as `DefaultCachedBatch`, reads back correctly, and produces
no `CometInMemoryTableScan`. This test fails on `main`, where the relation is
cached in Comet's Arrow format.
Existing suites run locally and green: `CometExecRuleSuite`,
`CometScanRuleSuite`, `CometNativeShuffleSuite`, `CometInMemoryCacheSuite`,
`CometNativeReaderSuite`, `CometShuffleSuite`, `DisableAQECometShuffleSuite`,
`ParquetReadV1Suite`, `CometFuzzTestSuite`. `CometExpressionSuite` has two
pre-existing `DatePart`/`dayofweek` failures in my environment that reproduce
unchanged on the base commit and are unrelated to this change.
Applying `run-spark-4.1-tests`, since this touches the conversion path every
plan goes through.
## Why this does not close #5783
#5783 is the other half of the duplicate-name story: a Parquet file whose
struct has two byte-identical child names, read with a declared schema naming
one of them.
```scala
spark.range(3).selectExpr("named_struct('dup', id, 'dup', id + 100) as
s").write.parquet(p)
spark.read.schema("s struct<dup: bigint>").parquet(p).collect() // Spark 3
rows, Comet 6
```
The duplicate exists only in the file. Spark's analyzer rejects a declared
read schema that repeats a nested name with `COLUMN_ALREADY_EXISTS`, and its
Parquet schema inference rejects such a file for the same reason, so the
requested schema Comet sees is `struct<dup>` with nothing in it to detect. I
confirmed this against all three shapes in the issue: none of them puts the
duplicate anywhere a planning rule can see it, so no plan-time fallback reaches
them. That fix belongs in the native reader, which is what #5786 and #5654 are
for.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
