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.

[dwsmith1983]

The opener short circuit from #5801 is fixed in #5808, stacked on this branch: the reader factory validates the requested field ids from the footer with the same resolver, so metadata-free files with identical schemas and no filter now raise the same error.

[andygrove]

Thanks for the rework. All four points from my earlier review are addressed and I traced each one against the head rather than the summary.

convert_struct now resolves the child with array.columns().get(from_index) and returns an error naming the requested field, the index and the child count, so a struct arriving with fewer children than the mapping was resolved against is a DataFusionError rather than a panic in the executor. The root last-wins gap is closed properly: has_duplicate_names widens needs_remap so a case-sensitive read with duplicate root names no longer falls through to DefaultPhysicalExprAdapter::index_of, shadowed_by_later_duplicate hides the earlier ones, and I checked the interaction with the ID path in both orders. [d(id 5), d] against logical d(id 5) keeps d on the ID match (the ID block returns before the shadow rename) and shields the sibling, and [d, d(id 5)] shadows the first and keeps the second. Your note that parquet-mr keys two root d columns to one chunk, so there is no Spark-comparable end-to-end assertion for that shape, is worth having in the record. The case-insensitive ambiguity companion test is there, and #5801 tracks the opener short circuit with #5808 stacked on it.

One thing I would like resolved before this merges, and it is about merge order rather than the code in isolation.

convert_array's list arms collide with #5681

#5681 is approved and touches the same match in parquet_convert_array_impl. It widens the single (List, List) arm to the whole array family on both sides, including cross-representation pairs, specifically so a List<Struct> read as a LargeList<Struct> resolves its element fields with Spark's rules before Arrow changes the list layout. Without that, cast_struct_to_struct gives up on name matching and falls back to positional, so a requested field absent from the file reads the neighbouring column's values.

This branch rewrites the same arms into convert_array and covers (List, List) and (LargeList, LargeList) only. That is not a regression against main on its own. The problem is what a mechanical conflict resolution produces if #5681 lands first. resolve_field_mapping matches (List, List) | (LargeList, LargeList) and falls to _ => Ok(FieldMapping::Leaf) for every other pair, and type_holds_struct only walks Struct, List, LargeList and Map. So a List<Struct> read as a LargeList<Struct> resolves to Leaf, #5681's widened arm calls mapping.list_element()?, which returns Ok(&Leaf), and the struct child then lands on the new (Struct(_), Struct(_), other) arm and fails with struct column resolved to a non-struct field mapping. A read that works on main today would start erroring, and it would do so from the arm you added as an internal assertion, so it would read like a planner bug rather than a missing case in the resolver.

Whichever of the two goes second has to carry the other's coverage, and the part that a conflict marker will not point at is that resolve_field_mapping and type_holds_struct need widening too, not just the convert_array arms. Could you and @peterxcli agree an order, and could the one that rebases extend all three places together? A cross-representation List<Struct> case in struct_field_matching would pin it.

Related and smaller: the three places that decide list handling (type_holds_struct, resolve_field_mapping, convert_array's arms) currently agree only because they all default to leaf-or-passthrough for FixedSizeList and the two view types. That is fine today, but it is three lists that have to move together and nothing says so. Would a comment on type_holds_struct pointing at the other two help, or is there a way to derive one from another?

One last thing I checked and am happy with, since it is the subtle part of the new design. Gating the relabel shortcut in CometCastColumnExpr on mapping.is_positional() is a real fix, not just plumbing: swapped field IDs give names and types that satisfy types_differ_only_in_field_names, so before this the relabel would have returned the file's columns under the requested names and silently swapped the values. test_swapped_field_ids_bypass_relabel_shortcut pins exactly that, and the companion test keeps the shortcut alive for the ordinary item/element relabel with ID read mode on. Since the mapping is resolved per file in create and read per batch, and spark_parquet_convert's public entry still resolves from the runtime type, I do not see a way for a batch to reach a mapping resolved against a different schema.

[dwsmith1983]

Is there a reason not to fold it into #5654?

No. #5808 is now the last three commits here and is closed, so this is the one reviewable change; the description covers both.

Could you and @peterxcli agree an order, and could the one that rebases extend all three places together?

Extended all three here so the order stops mattering: resolve_field_mapping, type_holds_struct and convert_array share one list_element_field helper covering List, LargeList, FixedSizeList and the two view types, and the converter now converts the element values through the mapping first, rebuilds the array in the file's representation around them, and only then lets Arrow's cast change the layout, the same shape as #5681's arm. Three cases in struct_field_matching pin it: a List<Struct> requested as LargeList<Struct> resolves to a non-positional element mapping, converts by id with swapped values coming back swapped, and null-fills a requested element field the file lacks instead of reading its neighbour. Left @peterxcli a note on #5681 with that, so either rebase is taking one arm over the other.

Would a comment on type_holds_struct pointing at the other two help, or is there a way to derive one from another?

Derived: all three call the helper, and its doc names them.

Would you narrow the root check in remap_physical_schema to the requested columns, or file it as a follow-up and reword the description so the asymmetry reads the way it actually behaves?

Narrowed here, and you are right about the description; it now says what happens. remap_physical_schema records the ambiguity per logical field instead of failing, leaves the colliding physical fields out of the id rename so the shield fake-names them, and rewrite raises for referenced columns the way nested ambiguities already are. Your three-column example reads [7] through the adapter now (test unrequested_duplicate_root_field_ids_read_through_adapter, with a column the file lacks forcing the adapter), and rewrite of the ambiguous column still raises _LEGACY_ERROR_TEMP_2094. Both paths validate the same scope, the columns a read requests.

Could you thread case_sensitive through the helper and cover it?

Threaded, with the case the description now explains: s(id 10)<x, X> in a metadata-free file fails at footer load with _LEGACY_ERROR_TEMP_2093 in case-insensitive mode and reads in case-sensitive mode. The root duplicate id and the list<struct> element duplicate are pinned as well, and the description says the root check is closed by the same footer check.

The three inline items are in: the memo is a parking_lot::Mutex with the helper and the poison imports gone, throw_spark_error_json is the one throw site with the External arm going through spark_error_json_in_chain (a new test drives a SparkError behind External and Context across the JNI boundary and checks the class), and the opener-skip explanation lives on FieldIdCheck with the other two pointing at it. The description also carries the COLUMN_ALREADY_EXISTS reasoning and the note that the chain walk widens which typed errors keep their class.

Core crate 367 tests, bridge 32, each commit compiles on its own, and ParquetReadV1Suite, CometNativeReaderSuite and SparkErrorConverterSuite pass on Spark 3.5 against the rebuilt library.