fix: match Spark's duplicate field and field id semantics in parquet field lookup

[dwsmith1983]

Which issue does this PR close?

Closes #5801.

Part of the restructuring of #5365 requested in review: this extracts the duplicate field and field id matching semantics that previously traveled with the Delta contrib work, re-derived on top of the folding that #5602 added. It also absorbs the reader-factory validation that was stacked on it as #5808, since that commit cannot rebase onto main on its own and the two could not merge separately.

Rationale for this change

Four places where the native parquet field lookup diverges from Spark:

1. remap_physical_schema only shields id-bearing logical fields whose id is missing from the file. When a logical field's id matches one physical field but a stray physical column carries that logical field's name, the stray column can still name-match through the expression adapter fallback and hijack the read.
2. parquet_convert_struct_to_struct silently resolves a requested field id that matches more than one physical field to the first match. Spark raises the duplicate field error in field id lookup mode.
3. Case sensitive exact name lookup on duplicate names resolves to the first field, while Spark builds its name map with .toMap, where the last entry wins.
4. DataFusion's Parquet opener creates the physical expression adapter only when a predicate is pushed or the file schema differs from the requested schema. A file with no key-value metadata whose schema equals the requested schema, read with field id matching on and no data filter, skipped the validation above and read a struct with two children sharing a requested id positionally. Spark rejects that during schema clipping. Spark-written files always carry key-value metadata that arrow-rs folds into the file schema, so they always reach the adapter; the gap is metadata-free files.

What changes are included in this PR?

• All id-bearing logical fields are shielded from name matching, the shield runs after the name match pass so a successful match claims the field first, and fake placeholder names skip a reserved set built from both schemas so they can never collide with real columns.
• A requested field id resolving to more than one physical field raises SparkError::DuplicateFieldByFieldId (_LEGACY_ERROR_TEMP_2094). Duplicate ids that no requested field references remain harmless, at the root as well as inside nested types: the adapter records a root ambiguity per logical field and raises it from rewrite only for the columns a read references, the same way nested ambiguities are handled. Before this, a duplicate id on columns the read never asked for failed the query, while Spark answers it because clipParquetSchema only sees the requested schema.
• Exact name lookup on duplicate names resolves to the last field.
• The field mapping is resolved once per file in the adapter factory and reused for every batch; a mapping that reorders fields (ids resolved out of position) disables the metadata-only relabel shortcut, which would otherwise hand back the file's columns under the requested names with the values swapped.
• The mapping resolver, the struct-holding type walk and the array converter now share one definition of which Arrow types are lists, so a List<Struct> read as a LargeList<Struct> (or any other pair of list representations) resolves its element fields with Spark's rules before Arrow changes the list layout. This is the same shape fix: apply Spark's Parquet conversion rules to nested struct/list/map fields #5681 takes for the converter, so whichever of the two lands second rebases mechanically.
• The Parquet reader factory, which fetches every file's footer regardless of what the opener decides, gains an optional field id check installed at plan time only when field id matching is on and the requested schema carries a field id at any depth, so ordinary reads pay nothing. After the footer is fetched it converts the file schema to Arrow the way the opener does and resolves the requested fields with the same mapping resolver the adapter uses, raising the same errors. A per-file memo keyed on the cached footer's identity means a file is validated once per distinct footer, not once per batch.
• The JNI error bridge walks the cause chain so a Spark error raised inside the reader factory keeps its Spark exception class instead of surfacing as a generic read failure. This is wider than the reader path: any typed SparkError reachable through a Context or Shared wrapper anywhere in the plan now keeps its class. An ANSI failure in a pushed-down filter is unaffected, because DataFusion stringifies it into an ArrowError::ComputeError and nothing typed survives the chain.

Both validation paths now validate the same scope, the columns a read requests. The footer check covers the required schema, and the adapter raises only for referenced columns. The check also catches a duplicate id on root fields and inside a list<struct> element, so it closes the same opener-skip gap for the root-level check, not only the nested one.

There is no equivalent gap for the case-insensitive duplicate name error: for the file and requested schemas to compare equal, the requested schema would have to hold two fields folding to the same name, and Spark rejects that at analysis with COLUMN_ALREADY_EXISTS. Two fields can share an id while having distinct names, which is what makes the duplicate id error reachable and the use_field_id && schema_holds_field_ids gate on the footer check sufficient.

How are these changes tested?

Rust, all written before the change they pin and failing on the previous head unless noted:

• Adapter and conversion: stray column hijacking the remap, case insensitive sibling null-filled, duplicate id silently reading the first match, first-wins name resolution, swapped field ids bypassing the relabel shortcut, root duplicate id raised only for referenced columns (rewrite of an unreferenced column succeeds, of the ambiguous one errors).
• List representations: a List<Struct> requested as LargeList<Struct> resolves to a non-positional element mapping, converts by id, and null-fills a requested element field the file lacks instead of reading its neighbour.
• Scans through DataSourceExec on a file written without key-value metadata, asserting first that the file schema equals the requested schema so the opener skips the adapter: duplicate struct id rejected, unique ids read, check inert with field id matching off, the same through the planner's data schema plus projection wiring, duplicate id at the root, duplicate id inside a list<struct> element, case-insensitive duplicate name (_LEGACY_ERROR_TEMP_2093) rejected at footer load with the case-sensitive read of the same file succeeding, and unrequested root duplicate ids read through the adapter path.
• JNI bridge: the chain walk finds a SparkError behind DataFusion's Context and External wrappers; a plain Parquet error still classifies as a file read failure.

Core crate 367 tests, bridge crate 32, clippy with -D warnings and fmt clean. Each commit compiles and passes the parquet module tests on its own.

Scala: a ParquetReadV1Suite case writes the file with parquet-mr and no key-value metadata, asserts the footer's key-value map is empty and that the plan carries the native scan, and expects Spark's duplicate field id error; it reported no exception against the previous native library. ParquetReadV1Suite, CometNativeReaderSuite and SparkErrorConverterSuite on Spark 3.5 against the rebuilt native library: 137 succeeded, 0 failed.

[sunchao]

This fixes stray-name matching and placeholder collisions, and adds nested duplicate-ID rejection and last-wins exact-name lookup. One gap remains: metadata-only struct relabeling bypasses the duplicate-ID check, as detailed inline.

I compared the code with maintained Spark 3.5 and 4.0 sources. Eight component-check groups passed using extracted Comet helpers with Arrow/Parquet 58.4.0 and DataFusion 54.1.0. A separate probe reproduced the cast bypass and verified a renamed-child control. These probes use limited scaffolding and are not full Comet scan, JNI or Spark query tests. The reported 246 native and 58 Spark 3.5 tests are the author's results.

At 04:51 UTC, current-head CI had 29 successful, 32 running and 7 skipped checks. Full CI validation was still pending.

[sunchao]

[P2] Run duplicate-ID validation before metadata-only struct relabeling

Could you route metadata-only struct adaptations through this validation too? For file struct s<x: int id=1, y: int id=1, z: int id=2> and requested s<x: int id=1, y: int id=3, z: int id=2>, Spark rejects requested ID 1 as ambiguous. DataFusion emits a struct cast, but CometCastColumnExpr::evaluate takes types_differ_only_in_field_names and calls relabel_array, because that predicate ignores field-ID metadata. The new lookup never runs and leaves all three physical values in place. A focused probe using the current cast expression and a real Arrow/Parquet round trip returned [42, 43, 44], while renaming requested x made the same input reach the duplicate-ID error. Could you guard the relabel shortcut for ID-based reads and add a cast-expression or scan regression with unchanged child names?

[dwsmith1983]

Good catch, the shortcut sailed right past the new validation. Fixed in 3d68f22: the relabel arm is now guarded so that when use_field_id is set and the requested type carries field id metadata, evaluation falls through to the struct conversion where the duplicate id lookup runs. Chose the guard at the call site rather than inside types_differ_only_in_field_names since that predicate is a pure structural comparison with no access to the parquet options. Your exact probe is now a regression test (unchanged child names, duplicate id 1, asserts the 2094 error) plus a companion pinning that the fast path survives for name only differences without ids and for the flag alone.

[sunchao]

Reviewed head 58e67fee against base 55ae4f20. The PR is focused, but I found one new regression, an incomplete validation fix, and avoidable batch-processing overhead.

1. [P2] Placeholder collisions can suppress column defaults.
   schema_adapter.rs:177 reserves exact names, while missing-column detection uses case-folded names. A generated __comet_unmatched_field_id_1 therefore collides with requested __COMET_UNMATCHED_FIELD_ID_1. In my reproduction, the base returns the configured default 7; this PR returns NULL.

   Reserve names using the existing folded schema names. That change passed the reproduction. Constructing the reservation set only when shielding needs it would also avoid extra hashing on ordinary reads without field IDs.

2. [P2] Duplicate-ID validation still depends on whether a cast occurs.
   The new guard in cast_column.rs:292 misses identical physical/requested schemas. Such reads can omit the cast entirely or return before the guard.

   A real native Parquet scan of identical s<x: long id=1, y: long id=1> schemas returned [42, 43]; Spark 4.1.3’s schema-clipping check rejected the duplicate ID. This also occurs on the base, so it is an incomplete fix, not a new regression. Validation needs to cover reads that require no conversion.

3. Avoid allocating a vector for every unique ID on every struct conversion.
   parquet_support.rs:270 changes the index to HashMap<i32, Vec<usize>>. An allocation probe using the old and new construction loops measured 8 → 264 allocations for 256 unique IDs.

   A compact unique/duplicate entry would preserve the behavior. Collect matching field names only when reporting an ambiguity. The new contains_field_id_metadata predicate also depends on immutable expression state and can be computed once.

The strongest design improvement is to resolve and validate requested fields once per file schema, then reuse the mapping across batches. That addresses the validation bypasses and repeated lookup work together. Metadata-only relabeling remains safe when the resolved mapping is positional. A small mapping object is a useful abstraction here.

Validation: 100 native Parquet tests passed, with default HDFS features disabled. Additional head/base probes confirmed both correctness cases. Performance evidence measures component allocations, not overall scan speed. CI snapshot: 57 passed, 7 running, 7 skipped. Nothing was posted to GitHub.

[dwsmith1983]

Thanks, all three are addressed in a69b5f5, following the once-per-file design you suggested.

The physical expression adapter factory already runs once per file schema, so it now resolves a small FieldMapping tree (struct sources, list and large list elements, map key and value, leaf) for every logical field that holds a struct, mirroring Spark's clipParquetSchema at each nesting level. Duplicate requested ids and ambiguous case-insensitive names are detected there, stored per logical field, and raised when the column is referenced, whether or not a cast is later emitted. The per-batch conversion receives the resolved mapping and applies it positionally, so there is no hashing or per-id allocation on the batch path; the index type is a compact entry that records an index and an ambiguity flag, and matching names are collected only when the error is built. The contains_field_id_metadata predicate is gone; the relabel shortcut is gated on the mapping being positional instead.

Your repros: the identical s<x: long id=1, y: long id=1> schema now raises the duplicate id error with no cast in the plan, pinned in Rust through the exec path and in ParquetReadV1Suite (the Scala case fails against the previous native library and passes now). The folded placeholder collision returns the configured default again, pinned in the adapter tests with Spark-style key-value metadata on the file.

One residual worth naming: DataFusion's opener skips the adapter entirely when the logical and physical schemas compare equal and no predicate exists. Spark-written files always carry key-value metadata that arrow-rs folds into the physical schema, so they always go through the adapter, but a file with no metadata at all and duplicated ids inside a struct would still read positionally. Happy to cover that in a follow-up if you think it matters.

[dwsmith1983]

@sunchao the once-per-file mapping round covering your three findings is pushed. Ready for another look.

[andygrove]

convert_struct in native/core/src/parquet/parquet_support.rs around line 563 now calls array.column(from_index) with an index resolved at planning time against adapted_physical_schema, where the old code derived it from the runtime array's own DataType. The only guard is sources.len() != to_fields.len(), which checks the target side. What guarantees the struct array the reader hands back always carries the same children in the same order as the physical field the mapping was resolved against? If that can drift at all, this is an index panic in the executor rather than a DataFusionError. Checking from_index against array.num_columns() would bound the worst case.

The description lists last-wins exact-name resolution as one of the three fixes and resolve_struct_mapping does it for struct children. At the top level in case-sensitive mode with no field ids, needs_remap is false in schema_adapter.rs around line 520, so resolution falls to DefaultPhysicalExprAdapter, which goes through Schema::index_of and returns the first match. Spark builds caseSensitiveParquetFieldMap at the root message level with the same .toMap it uses for nested groups. Was the top level deliberately left out of scope?

The field-id ambiguity path is covered from several angles now. The case-insensitive name ambiguity that resolve_struct_mapping raises around line 388 does not appear to have a companion test in the new struct_field_matching module. It might be worth pinning that half too, since it is the branch that decides between an error and a silently wrong column.

On the residual you named where DataFusion skips the adapter when the two schemas compare equal and there is no predicate, I confirmed that short circuit in the 55.0.0 opener. Could you open a tracking issue and link it here so it does not get lost? The branch also conflicts with main right now and needs a rebase before anything meaningful runs against it.

[dwsmith1983]

Rebased onto main and the three points are in the head (b438dce).

Bounds: convert_struct now looks the source index up with columns().get and returns an error naming the requested field, the index, and the child count instead of indexing; a test feeds a struct with fewer children than the mapping expects and asserts the error. A struct column arriving with a non-struct mapping errors the same way rather than passing through. I did not add a type-equality check there, since convert_array dispatches on the runtime child type and strict equality would reject coercions it handles.

Root last-wins: not deliberate, the top level had simply fallen to the default adapter. With duplicate exact names at the root in case-sensitive mode the remap path now runs, the shadowed earlier fields get a placeholder name so the default adapter's index_of lands on the last one, and the nested resolver does the same; Spark 3.5's clipParquetGroupFields uses one .toMap for root and nested groups. Tests at the adapter level and through DataSourceExec, plus one that a field id match still wins over the duplicate. One thing worth knowing: parquet-mr writes two root columns named d into a single column chunk keyed by path, so a Spark-written file with that shape reads six interleaved values in Spark and arrow-rs alike, and no Spark-comparable end-to-end assertion exists for it; the Rust scan test uses arrow-rs to write the file.

The case-insensitive ambiguity now has its companion test in struct_field_matching: A and a against a requested a errors naming both in case-insensitive mode and reads a in case-sensitive mode.

The opener short circuit is tracked in #5801.

[andygrove]

All four points from my earlier review are in, and the #5681 merge-order concern turned out fine since that landed first and this branch rebased onto a base containing it. I spot-checked the resolution rules against ParquetReadSupport.clipParquetGroupFields rather than taking them on trust, and they line up. Field id wins with no name fallback, a non-id requested field matches by name in the same group, caseSensitiveParquetFieldMap is .toMap so exact-name collisions resolve last-wins silently, and caseInsensitiveParquetFieldMap is a groupBy on toLowerCase(Locale.ROOT) that raises on more than one hit. resolve_struct_mapping matches all of it, including raising only for the requested id rather than validating the whole group.

One naming point. This adds schema_holds_field_ids in parquet_support.rs, which walks nested types, and it now sits one word away from schema_has_field_ids in schema_adapter.rs, which only looks at root fields. They are easy to confuse and they are not interchangeable. Could the new one be named so the difference is visible at the call site, or could the old one gain a line saying it is deliberately root only?

There is a behavior question behind that name which I think is follow-up work rather than something for this PR. schema_has_field_ids is what gates the ParquetMissingFieldIds rejection at the top of remap_physical_schema. Spark's equivalent in ParquetReadSupport.getRequestedSchema uses containsFieldIds on the file schema and ParquetUtils.hasFieldIds on the requested schema, both recursive, and it does not consult spark.sql.parquet.fieldId.read.enabled at all. So with the default use_field_id of false, a requested schema carrying parquet.field.id read against a file with no ids returns rows in Comet where Spark raises, and a file whose ids sit only on nested fields is rejected by Comet where Spark null-fills through matchIdField. Both reproduce on main so neither is from this change, but this is the PR that puts a correct recursive predicate one module away from the one making that decision. Could you file a tracking issue and link it here?

Separately, this needs sequencing with #5786, which I have also commented on. That one adds a validate_field_names footer check that errors on byte-identical sibling names, which is the opposite of the last-wins resolution here, and it hangs a second validator off the same get_metadata call site. git merge-tree reports eight conflicting hunks in eager_page_index_reader_factory.rs and one in parquet_exec.rs, and both PRs are clean against main on their own so neither CI run shows it. #5884 also records Spark 4.1.3 returning {0, 100, 1} for the nested duplicate fixture rather than last-wins, which is worth reconciling against what clipParquetGroupFields does, since the clip and the reader may not agree. Could you and @ErikBPF settle which behavior we want before either lands?

[sunchao]

Reviewed head 1ea1be161 against base db790673d. I found one P2 correctness regression to fix before merging.

P2: Duplicate root names now select the wrong column

The new root-field shadowing hides earlier duplicate names and selects the last column.

I tested one Parquet file containing two separate d columns, reading it with explicit schema d bigint and case sensitivity enabled:

Reader	Result
Spark 4.1.3, vectorized	[1, 2, 3]
Base native scan harness	[1, 2, 3]
PR native scan harness	[10, 20, 30]

This changes previously matching results. Spark’s schema-clipping .toMap behavior does not establish what its reader actually returns. Preserve the working root-column selection and add a comparison against Spark using the same file.

This supplies concrete differential evidence for the concern already discussed in #5884.

Validation and remaining assessment

• CI: 56 successful, 10 skipped; Required Checks passed.
• Local: 22 component tests passed. Additional scan probes verified the earlier default-value and duplicate-ID fixes.
• Build limit: The registry lacks pinned DataFusion 55.1. Native probes used current changed modules with DataFusion 55.0 / Arrow 59.3, rather than a full current-head Comet build. Relevant upstream root-read paths were checked against 55.1.
• No other actionable findings emerged from mapping, projection, ownership, footer-cache, or JNI-error review.

Evidence: Spark result, base result, PR result (local reproduction logs).

Nothing was published to GitHub.

[dwsmith1983]

Preserve the working root-column selection and add a comparison against Spark using the same file.

The comparison is in: test-data/duplicate-root-names.parquet carries two root d columns, and a new ParquetReadSuite test reads it with d bigint under case sensitivity and asserts Comet's answer and plan against Spark. It passes on 3.5 and 4.0, both returning [1, 2, 3], because the JVM-side projection binds the first d before the native adapter sees the file, so the adapter's last-wins rule only decides between columns that are both projected. Your harness drove the adapter alone, which is where [10, 20, 30] comes from. The end-to-end selection is unchanged from base and now pinned.

Could the new one be named so the difference is visible at the call site, or could the old one gain a line saying it is deliberately root only?

Both: the recursive predicate is now any_nested_field_has_id, and schema_has_field_ids says it is root only on purpose and names the other.

Could you file a tracking issue and link it here?

Will do once we settle the nested-duplicate behaviour with @ErikBPF on #5786, since the same issue should record both the field-id gating gap and the nested resolution we want; I would rather not open two overlapping ones.

[ErikBPF]

Thanks—the root parity test and naming clarification address those points. My proposed order remains #5786 first, then rebase #5654.

Preserve the pre-decoder guard for selected ambiguous fields until correct physical-leaf selection is demonstrated. Unrequested nested duplicates can remain readable where structural narrowing proves they are pruned, as #5786 now covers. The root JVM projection result does not settle nested decoding.

Can we agree on that order and behavior? #5884 already tracks duplicate resolution; please explicitly track the field-ID gating gap and link both PRs.

[dwsmith1983]

Preserve the working root-column selection and add a comparison against Spark using the same file.

Correction to my earlier note: after merging today's main, the end-to-end read does return the last d, so your P2 stands on the current base. Root duplicates now resolve first-wins: a later exact duplicate takes the fake name instead of the earlier one, so the adapter binds the first file column the way Spark's reader does. The two adapter tests assert [1, 2, 3], and the new ParquetReadSuite comparison against Spark over the two-column file is the guard that caught it.

[dwsmith1983]

Can we agree on that order and behavior? #5884 already tracks duplicate resolution; please explicitly track the field-ID gating gap and link both PRs.

The field-id gating gap is #5936, linked from both PRs. On behavior, this PR now does what Andy proposed on #5786 for the nested case rather than last-wins: a requested field that matches more than one byte-identical sibling is refused with a message naming the field, while reading the unique sibling beside them still works, both pinned in Rust and by a Scala read of a file whose struct carries dup, dup, other. Root exact duplicates read the first file column, as Spark's reader does, pinned end to end against Spark.

On order: since the refusal now lives in the resolver this PR owns, I would land this one first and rebase #5786 onto it for the shapes that remain, the metadata-time check included if it is still needed. The fallible name folding from #5845 is already propagated here.

[sunchao]

Reviewed head 1fa518b54 against base 1d0ce5fe8. One P2 regression remains.

P2: Footer validation rejects supported duplicate-root reads

parquet_support.rs:352 treats the root schema as a nested struct, applying the new duplicate-name rejection before the adapter can select the first root column.

Trigger: a file contains d, d, and a(id=7). Request d by name and a by ID, with case sensitivity and field-ID reads enabled.

Reader	Result
Spark 4.1.3 vectorized	(1,100), (2,200), (3,300)
Base native scan harness	Same rows
PR native scan harness	Duplicate-field error for d

Disabling field-ID reads makes the PR return the correct rows. The unrelated ID-bearing column activates the footer check and causes the failure.

Fix: preserve root first-wins behavior in footer validation while retaining nested rejection. Add a mixed ID/name regression through the reader factory.

Validation

• 120 component tests passed, plus Spark comparisons covering 32 read combinations.
• Reviewed mapping, projection, array lifetimes, footer caching, and JNI errors. No other introduced actionable defects found.
• Build limitation: the registry lacks DataFusion 55.1. Component probes used current sources with DataFusion 55.0 / Arrow 59.3; full Comet integration tests could not run.
• CI: 7 passed, 12 queued/running, 9 skipped.

Full review and evidence (local review report and reproduction logs). Nothing posted to GitHub.

[dwsmith1983]

Fix: preserve root first-wins behavior in footer validation while retaining nested rejection. Add a mixed ID/name regression through the reader factory.

Done. validate_field_mapping now renames every later exact duplicate at the root to a name no request can match, keeping its type and metadata, before resolving, so the footer check binds the first root column the way the adapter does while a requested nested duplicate is still refused. Two regressions: a unit test on the check itself with d, d, a(id=7) requested by name and by id under case sensitivity and field ids (passes), the same under case-insensitivity (Spark's ambiguity error) and the nested shape (refused); and a reader-factory test that writes that file, loads its footer and runs FieldIdCheck.validate on it. Both fail without the change. Merged with main at a8e8157 first.

[ErikBPF]

Agreed on landing this first, then integrating the remaining nested-only checks in #5786. The separate mixed-type root investigation in #5964 now has parquet-java PR apache/parquet-java#3796. Spark descriptor/filter changes have passed targeted experimental checks, but still need upstream JIRA tracking, a current-master port and an available parquet dependency. That work remains separate from these Comet PRs.

[comphead]

The solution might be overcomplicated here. I would expect a small check, that checks all the plan nodes and related schema and if the struct found with duplicated name, then fallback

[comphead]

Let me try to come up with more light weight solution

[andygrove]

Triage note: #5786 is working the other side of duplicate Parquet field names in the native scan. It rejects ambiguous sibling names before the decoder is built, closing #5783, and explicitly leaves "Spark-compatible duplicate-name resolution" as separate work — which is this PR. You share parquet_exec.rs, schema_adapter.rs and eager_page_index_reader_factory.rs.

The risk is that the two disagree on the outcome for the same file: this PR resolves case-sensitive duplicates last-wins the way Spark's .toMap does, while #5786 would reject some of those reads before the lookup ever runs. Could you and @ErikBPF work out which cases each of you is claiming, and in what order they land?

[ErikBPF]

@dwsmith1983 yours have a lot more work implemented. I see no problem in landing it first and I adapt mine after. @andygrove when do you intent to close 1.1.0? Just want to know the feasibility to have both on this release

[andygrove]

@dwsmith1983 yours have a lot more work implemented. I see no problem in landing it first and I adapt mine after. @andygrove when do you intent to close 1.1.0? Just want to know the feasibility to have both on this release

There is no hard deadline, but I was hoping to have 1.1.0 released before the end of the month.