fix: decline Parquet scans whose struct repeats a field id

[comphead]

Which issue does this PR close?

Closes #5801.

Related to #5605, whose duplicate-name check was missing from the Arrow cache serializer. This does not close #5783; see the note at the end.

Rationale for this change

Duplicate Parquet field ids (#5801). 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.

The requested schema carries the ids in each StructField.metadata, so this is decidable from what the plan already holds. No footer read, no execution-path gate.

Duplicate struct field names in the cache serializer. 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 for the scan type checker, the native shuffle predicate and the two row-conversion sinks, but ArrowCachedBatchSerializer.supportsType accepted it, so

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.

What changes are included in this PR?

1. DataTypeSupport gains hasDuplicateFieldNames, which the existing trait check and the cache serializer both call so the definition of "duplicate name" lives in one place, and findDuplicateStructFieldIds, which descends through structs, arrays and maps and describes the first offending struct with its path.

2. 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, x and y are told apart by name and the scan stays native.

3. ArrowCachedBatchSerializer.supportsType rejects duplicate child names, so such a relation is cached in Spark's default format instead, alongside the interval types already excluded there.

Nothing in the operator conversion path changes.

How are these changes tested?

• DataTypeSupportSuite (new): hasDuplicateFieldNames including that names differing only by case are not duplicates (Java Arrow tells a and A apart, and Spark's own Parquet reader raises on that shape before Comet is asked); findDuplicateStructFieldIds at every nesting level (struct child, array element, map key, map value, two levels deep); distinct and absent ids; a non-integral id, which is left for Spark's ParquetUtils.getFieldId to reject rather than misreported as a duplicate.
• CometNativeReaderSuite: Duplicate field ids inside a struct are not validated when the file schema equals the requested schema and no predicate is pushed #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, which is what makes the opener skip the adapter. 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, pinning that the gate is scoped to field id matching.
• CometInMemoryCacheSuite: the duplicate-named struct relation is cached as DefaultCachedBatch, reads back correctly, and produces no CometInMemoryTableScan. This test fails on main.

Existing suites run locally and green on the Spark 4.1 profile: CometExecRuleSuite, CometScanRuleSuite, CometNativeShuffleSuite, CometInMemoryCacheSuite, CometNativeReaderSuite, CometShuffleSuite, DisableAQECometShuffleSuite, ParquetReadV1Suite, CometFuzzTestSuite. cargo fmt --check and cargo clippy --all-targets --workspace -- -D warnings are clean (no Rust changes here).

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 scan rule.

Why this does not close #5783

#5783 is the duplicate-name case on the read path: a Parquet file whose struct has two byte-identical child names, read with a declared schema naming one of them.

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 is in the file, not in the plan. Everything a planning rule can reach holds the declared schema:

LogicalRelation.output            = s:struct<dup:bigint>
HadoopFsRelation.dataSchema       = struct<s:struct<dup:bigint>>
FileSourceScanExec.requiredSchema = struct<s:struct<dup:bigint>>
footer schema                     = message spark_schema {
                                      required group s {
                                        required int64 dup; required int64 dup; } }

The file schema is obtainable at plan time — ParquetFileFormat.inferSchema returns struct<s:struct<dup:bigint,dup:bigint>>, and the COLUMN_ALREADY_EXISTS you get from spark.read.parquet(p) comes later, from DataSource.resolveRelation's own duplicate check, not from inference. But getting it means Comet reading a footer on the driver, which it does not do today, and with the default mergeSchema=false only the first part-file would be touched, so the check would not hold for a table whose files disagree the way Spark's per-file clipping does. That is a bigger decision than this fix, and the airtight version belongs in the native reader, which is what #5786 and #5654 are for.

[andygrove]

@comphead is this different from #5786?

[comphead]

@comphead is this different from #5786?

this PR is supposed to be more generic, not only Parquet scan, but if UDF or any operator produces a schema with a struct having duplicated column we gonna fallback. Other PRs likely to be built on top of this? WDYT? @andygrove @dwsmith1983 @ErikBPF

I expect the change should be straightforward as we know about the schema attached to the plan nodes on the planning side.

[sunchao]

Correctness

This change extends duplicate-struct fallback to operator conversion and the Arrow cache serializer. Previously the shared type checker protected scans, shuffle, and row conversion, but an operator or cache could still expose a struct with repeated child names to Java Arrow. The new operator check covers both outputs and data-producing inputs, and the cache check consistently selects Spark's default cache format for unsupported structs.

Maintained Spark 3.5 and 4.0 resolve Parquet field IDs among siblings and raise when a requested ID matches multiple physical fields. That rule also applies at the file-schema root. The new scan guard catches nested duplicates but misses duplicate IDs on top-level fields, because the inherited schema entry point passes only each field's data type. This leaves a declared-schema case of the same equal-schema/no-predicate reader bypass unguarded. The first inline P2 describes the missing root check and regression case.

The second P2 is the missing CI registration for DataTypeSupportSuite. Both preflight runs fail on that omission, and the Spark runtime jobs are skipped. The new reader fixture usefully covers metadata-free files with field-ID matching enabled and disabled, while the cache test checks the selected batch format and readback. Those runtime results are author-reported, not independently established here.

Validation and scope

Reviewed head 204b6574 against base 74725611. The seven authored file changes are identical in merge 3dc05091, which also includes the newer base shuffle commit. Local python3 -B dev/ci/check-suites.py reproduces the preflight failure. An isolated compile of the exact shared type checker and extracted scan checker confirms that top-level duplicate IDs are accepted while nested duplicate IDs are rejected. That probe uses configuration/type-shim stubs and cached Spark type classes. Its later flag-disabled control stops on a missing Kryo class, so this is component evidence, not a passing Spark/JNI suite.

The maintained Spark 3.4 and 4.1 branches were unavailable for semantic comparison. Physical-only duplicate names hidden by a declared schema remain outside this PR's stated scope.

Performance

The added checks run during planning and cache-format selection rather than per row. Falling back prevents constructing Arrow batches whose struct children cannot be represented faithfully. The operator check short-circuits after the first duplicate and preserves ordinary repeated top-level output names.

The scan check recursively walks a type and then the existing support checker recursively visits it again, so deeply nested schemas repeat some work. Moving the ID validation to a single schema-level pass would also address the correctness gap. I have no benchmark evidence of a material planning regression, and no execution-speed claim follows from this review.

Design

A common conversion gate is a sensible place to enforce the Arrow boundary invariant. Inspecting child outputs matters for operators that consume a duplicate-named struct without returning it, and unwrapping WriteFilesExec follows the existing input handling. Using the serializer's support predicate for both cache creation and reading keeps the chosen cache format consistent.

Name duplication and Parquet ID ambiguity need different entry points. Repeated top-level output names can be valid positional columns, but duplicate field IDs at the file root still require the Parquet read check. Keep the current nested-name behavior and apply ID validation to the complete requested scan schema when field-ID matching is enabled.

Abstraction & complexity

The small recursive helper keeps struct, array, and map traversal together and produces a useful path in fallback explanations. It does not require a new framework or configuration surface. Malformed field-ID metadata is excluded from duplicate detection, while the existing Spark-backed schema serialization still validates the ID value.

The main simplification is to validate field IDs once at the schema boundary instead of recursively re-entering the same check through each type. The two inline P2s are the actionable changes from this review.

[sunchao]

Correctness

[P2] Check duplicate field IDs on the complete requested schema

DataTypeSupport.isSchemaSupported calls this method with each field's dataType, so StructType(x: Long id=1, y: Long id=1) reaches this helper twice as LongType and is accepted. I confirmed that with an isolated compile of the exact checker. Wrapping the same fields inside s correctly falls back.

For a metadata-free Parquet file with those two top-level fields and the same requested schema, this leaves the same no-predicate bypass as the new nested fixture: DataFusion 55.1 skips the expression adapter when the schemas are equal. Spark's field-ID ambiguity check applies at the root as well. Please run this check once on the complete requested schema and add the flattened version of the new reader fixture. The exception for repeated top-level output names does not apply to Parquet field IDs.

[sunchao]

Correctness

[P2] Register the new suite in both CI workflow matrices

dev/ci/check-suites.py requires every non-excluded suite to appear in both .github/workflows/pr_build_linux.yml and .github/workflows/pr_build_macos.yml. Neither contains org.apache.comet.DataTypeSupportSuite. The current preflight exits 255 with Suite not found in workflow .github/workflows/pr_build_linux.yml: org.apache.comet.DataTypeSupportSuite, which I also reproduced locally. This stops CI before the runtime test jobs run. Please add the suite to both matrices so the guard succeeds and the new unit tests execute.

[andygrove]

The force-push changed what this PR is, so flagging that up front. The operator conversion gate is gone, which means the "not only Parquet scan, any operator that produces a duplicate-named struct falls back" part that @comphead described above as the difference from #5786 is no longer in here. What is left is the scan check for duplicate field ids plus the cache serializer check. That is a reasonable PR on its own, but it means the "any operator" half of #5605 is now covered by nothing that is open. Is that coming back here, or should we file it separately so it does not get lost?

One note if it does come back. I ran the earlier head 204b6574 and that gate broke CometArrayExpressionSuite "folded map value with duplicate struct field names falls back (multirow)". The plan still falls back and the answer is still right, but the gate fires before the serde runs, so the recorded reason becomes Native operators do not support v.map value: struct with duplicate field names (x) and the expected Unsupported data type MapType is never recorded. That suite was not in the list you ran locally.

Preflight is red on both runs with Suite not found in workflow .github/workflows/pr_build_linux.yml: org.apache.comet.DataTypeSupportSuite. dev/ci/check-suites.py scans pr_build_macos.yml too, so it needs adding to both. Everything downstream is skipped right now, so there is no runtime coverage of this change in CI yet.

One more small thing that did not fit on a line in the diff. The description says the point of hasDuplicateFieldNames is that the definition of "duplicate name" lives in one place, but CometShuffleExchangeExec.scala:483 and CometShuffleExchangeExec.scala:610 still spell it out by hand as fields.map(f => f.name).distinct.length == fields.length. Could those use the new helper while you are in here?

I ran this locally on the Spark 4.1 profile, JDK 17, macOS aarch64, at a406fc75. Both of your new runtime tests pass, and so does CometExpressionSuite "named_struct with duplicate field names". The root-level field id case in my inline comment below fails.

[andygrove]

I confirmed @sunchao's point about top-level fields with a real run rather than a compile probe, and it does reproduce on this head.

Same construction as your new test in CometNativeReaderSuite, but with the two id-1 fields at the top level instead of inside s:

message spark_schema {
  optional int64 x = 1;
  optional int64 y = 1;
}

read back with StructType(Seq(withFieldId("x", 1), withFieldId("y", 1))) and spark.sql.parquet.fieldId.read.enabled=true:

COMET: ROWS=[10,20]      (CometNativeScan in the plan)
SPARK: ERROR=Found duplicate field(s) "1": [x, y] in id mapping mode.

That is #5801 exactly, one level up. The reason is structural rather than an oversight in the predicate: DataTypeSupport.isSchemaSupported hands isTypeSupported each field's dataType, so the root struct's fields are never compared against each other and this override cannot see them.

Would you consider overriding isSchemaSupported here instead and running findDuplicateStructFieldIds once over the whole requested schema? That covers the root, and it also drops the repeated traversal, since findDuplicateStructFieldIds already recurses while isTypeSupported re-enters it at every nesting level. Could the root case get a test alongside the nested one?

[andygrove]

Small thing about this doc comment. ParquetReadSupport.matchIdField raises when the id a requested field asks for matches more than one field in the file, whereas this checks whether two requested fields share an id. Those are different predicates, and they only coincide under the equal-schema condition #5801 describes.

That is enough for the gap you are closing, but the comment reads as though the second follows from the first, and someone will rely on that later. Could it say which one is implemented and why the narrower check is sufficient here?

[andygrove]

Comet already uses ParquetUtils.hasFieldId and ParquetUtils.getFieldId for this in serde/operator/package.scala:48-51 and QueryPlanSerde.scala:700. Any reason not to use ParquetUtils.FIELD_ID_METADATA_KEY and ParquetUtils.hasFieldId here rather than redeclaring the key and hand-rolling the metadata read? It would keep the two from drifting if Spark ever changes it.