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

[dwsmith1983]

Which issue does this PR close?

Closes #6192: the native scan read a nested struct by position when the requested fields matched the file's by name and type but not by id, so a nested column dropped and added back under the same name came back with its old values where Spark returns null. The relabel shortcut is now taken only when the resolved mapping is positional.

Relates to #5884, the sibling-name half of duplicate field resolution, and to #5936, whose gate is addressed in #6116. #5801, which an earlier revision closed, was closed by #6004.

Rationale for this change

Spark's ParquetReadSupport.clipParquetGroupFields resolves requested fields by id at every nesting level and raises _LEGACY_ERROR_TEMP_2094 when one requested id matches more than one file field. The native scan raised it only for root columns and silently read the first match below the root.

Spark resolves exact names through a map whose last entry wins inside a struct, while its reader binds the first file column at the root. The native scan did neither consistently.

An id-bearing requested field must never fall back to a name match, and the field mapping between the requested and file schemas is a per-file decision, not a per-batch one.

What changes are included in this PR?

• resolve_field_mapping and resolve_struct_mapping raise Spark's duplicate-id error for an ambiguous requested id at any depth, and remap_physical_schema defers a root ambiguity until a read references that column, so an unrequested duplicate stays harmless.
• Exact-name duplicates are rejected the way main rejects them since duplicate names are refused before decoding: a requested nested name two siblings carry, a referenced root name the file repeats, and an id whose physical root name is duplicated. Case-insensitive ambiguity raises _LEGACY_ERROR_TEMP_2093 for case variants and byte-identical siblings alike, while an exact match still reads. One divergence from Spark is kept from main and recorded by a test: a requested id that matches exactly one field is still rejected when its root name repeats, where Spark's matchIdField reads it.
• The id shield in remap_physical_schema runs after the name match, so a plain name match claims its column and only a genuine collision with an id-bearing request is renamed away; placeholder names skip every name either schema holds.
• One list_element_field decides what a list is for the resolver, the converter and the adapter's struct walk, so a List<Struct> read as a LargeList<Struct> resolves its element fields the same way.
• The field mapping is resolved once per file and applied positionally per batch through CometCastColumnExpr::with_parquet_options; a mapping that reorders fields disables the metadata-only relabel shortcut.

What this PR does not do: a Parquet file without key-value metadata whose schema equals the requested schema is still read positionally, because DataFusion's opener skips the expression adapter for it, so none of the resolution above runs for that file. #6004 covers the case where the requested schema itself repeats an id. Nested duplicate names in such a file, including two names that collide case-insensitively, still read by position where Spark raises; that is tracked in #6136.

How are these changes tested?

Rust, in parquet_support.rs and schema_adapter.rs, each written before the change it pins: an ambiguous requested id inside a struct, inside a list element and at the root raises, with the unrequested duplicate as the control; exact duplicate names are refused when requested and skipped otherwise; case-insensitive ambiguity raises for case variants and for byte-identical siblings while an exact match reads; an opaque Variant decode with duplicate physical children is refused by the physical name check; the id shield beats a stray same-name column and its placeholder never folds onto a requested name; list representations resolve by id across List and LargeList; a reordering mapping bypasses the relabel shortcut. Two scans through DataSourceExec exercise the adapter end to end.

Scala, in ParquetReadSuite: the Spark port multiple id matches, a duplicate id inside a struct on a Spark-written file asserting the native scan and Spark's error, and a differential read over the checked-in duplicate-nested-names.parquet file comparing against Spark.

Locally on Spark 3.5: the parquet modules of the core crate, the JNI bridge crate, clippy, fmt, ParquetReadV1Suite and CometNativeReaderSuite through the full reactor, test-compile on 3.4, 4.0 and 4.1, spotless and the semantic scalafix check. Spark's own ParquetFieldIdIOSuite runs in the Spark SQL job and needs run-spark-4.1-tests from a maintainer.

[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]

@comphead would you like me to cut it down?

[dwsmith1983]

The initial reason I was slightly concerned is 2k LOC to patch an extreme scenario

Cut down in 24030e5, as offered above. The footer-time validator is gone with its tests, so the reader factory and the scan setup are back to main and the Rust test bulk with them. What stays is what #5786 does not cover: the resolver raising Spark's duplicate-id error for an ambiguous requested id at any nesting level, exact-name lookups that bind the way Spark's do, the id shield against a stray same-name column, one list definition across the mapping paths, and the field mapping resolved once per file instead of per batch. The JNI cause-chain walk went too: restoring main's error conversion left every kept test green, because the adapter raises its errors at the top level, so that piece only served the footer check. The description is rewritten to match, with the remaining gap stated in full: a metadata-free file whose schema equals the requested one is still read positionally because the opener skips the adapter, #6004 covers the case where the requested schema itself repeats an id, and nested duplicate names in such a file are tracked in #6136.

Against main the diff is now 6 files changed, 2102 insertions(+), 356 deletions(-), down from 9 files with 2898 insertions and 384 deletions; the reader factory, the scan setup and the JNI error conversion are back to main byte for byte.

[andygrove]

@comphead could you take another look?

[comphead]

Thanks @dwsmith1983 @andygrove
#5786 is about to be merged, lets see what is still to be fixed and address it in this PR

[dwsmith1983]

#5786 is about to be merged, lets see what is still to be fixed and address it in this PR
@andygrove @comphead
#5786 now owns the duplicate name case at every level. match_struct_fields rejects a requested nested name that two siblings carry, rewrite rejects a referenced root name the file repeats, id_duplicate_roots rejects an id whose physical root name is duplicated, and check_decoded_field_names refuses any duplicate inside a subtree a Comet cast decodes in full. This branch sits on top of all of that. The first-wins root binding is gone, since main rejects that read and pins the rejection in CometNativeReaderSuite, and the branch's own duplicate-name message is gone with it. Where the resolver here replaced match_struct_fields, it raises main's duplicate Parquet field name error for byte-identical siblings in both case modes, so every path reports that shape the same way.

What this PR still fixes is field ids, which #5786 does not touch.

A requested id that two file fields carry inside a struct. File s (id 2) <x: bigint (id 1), y: bigint (id 1)>, read with spark.sql.parquet.fieldId.read.enabled=true and schema s (id 2) <x: bigint (id 1)>. Spark raises _LEGACY_ERROR_TEMP_2094 from matchIdField. On main match_struct_fields keeps the first field per id (map.entry(id).or_insert(i)) and returns x = 42 with no error, and none of the duplicate-name checks fire because the names differ. The resolver here raises Spark's error at any depth, and only for the ids a read references, so an unrequested duplicate id at the root no longer fails the whole scan the way remap_physical_schema did. ParquetReadSuite has this as duplicate field id inside a struct is rejected when a requested id matches two fields.

Swapped ids inside a struct. File s <x (id 1), y (id 2)>, requested s <x (id 2), y (id 1)>. Names and types agree, so main's relabel shortcut in CometCastColumnExpr hands back the file's columns under the requested names and the values come out swapped. The mapping resolved once per file is not positional for that shape, so the relabel is skipped and each field reads by id.

The id shield in remap_physical_schema runs after the name match. Main renames a physical column away whenever its folded name equals an id-bearing requested name, before an id-less requested field can claim it by name. File κ (id 2), requested Κ (id 1) plus id-less κ, case-insensitive: main null-fills both, Spark and this branch return (NULL, 7).

The rest is the once-per-file FieldMapping handed to every cast instead of re-resolving per batch, and one list_element_field helper shared by the resolver, the converter and the adapter's struct walk, so a List<Struct> read as LargeList<Struct> resolves its elements by id.

In the merge, main's needs_remap, the collision check in rewrite, id_duplicate_roots and the checked_decoded_expr wrapping stand as written. Removed from the branch: has_duplicate_names, shadowed_by_earlier_duplicate, the three root first-wins tests, the matching ParquetReadSuite read and duplicate-root-names.parquet. issue_5783_nested_name_duplicate is ported onto resolve_field_mapping because match_struct_fields no longer exists.

One divergence from Spark remains after the merge, and it comes from #5786 rather than from this branch. id_duplicate_roots in schema_adapter.rs rejects a requested id whenever its physical root name repeats in the file, even when the id itself matches exactly one field. Physical [d (id 1), d (id 2)] read as d (id 1) with field-id reads on: Spark's matchIdField in ParquetReadSupport.clipParquetGroupFields (branch-3.5 lines 438-450) finds one field for id 1 and reads it, and main raises duplicate Parquet field name 'd'. CometNativeReaderSuite's duplicate Parquet field names - root group and unprojected root duplicates pins that rejection, so the merge keeps it and a Rust test named known_divergence_repeated_root_name_rejects_unambiguous_field_id records it. The one way to close it is to narrow id_duplicate_roots to fire only when the requested id is ambiguous, in this PR, which reverses that test.

[andygrove]

This gate fixes a silent wrong-result bug on main, and I filed #6192 for it. Your summary covers swapped ids, but the shape I'd expect users to actually hit is a nested column dropped and added back under the same name. A file with s struct<x (id 1), y (id 2)> read as s struct<x (id 3), y (id 2)> returns the old x values on main, where Spark returns null. The same happens inside a list element and a map value. All of those match Spark 4.1.3 on this branch, with the native scan in the plan.

Could you add Closes #6192 to the description, plus a ParquetReadSuite test that runs the drop-and-re-add read through checkSparkAnswerAndOperator for a struct, a list element and a map value? At the moment only test_swapped_field_ids_bypass_relabel_shortcut covers this, at the expression level.

[dwsmith1983]

Could you add Closes #6192 to the description, plus a ParquetReadSuite test that runs the drop-and-re-add read through checkSparkAnswerAndOperator for a struct, a list element and a map value?

Added Closes #6192 to the description and three ParquetReadSuite tests, nested column dropped and re-added under the same name with a new field id: struct, : list element and : map value. They share one file, written with field ids on, holding struct<x (id 1), y (id 2)> as a top level column, as a list element and as a map value, and each reads its column with x carrying id 3 and y keeping id 2. Each runs through checkSparkAnswerAndOperator, asserts CometNativeScanExec is in the plan, and pins the expected rows explicitly: x is null and y keeps its values, including a null struct inside the list and the map, empty containers, and a null column.

One thing the run turned up: Spark 3.5's own vectorized reader raises on the list and map reads (ParquetColumnVector rejects the _fake_name_ field that clipParquetSchema generates for the unmatched id below a list or map), and Spark 4.0.4 still has that check, so for those two the comparison with Spark runs from 4.1 on and the pinned rows carry the check on 3.5 and 4.0. The struct case compares with Spark on every version.

With the gate in cast_column.rs forced to always take the relabel shortcut, all three fail with the old x values: the struct test returns [1,[1,10]] and [2,[5,50]] where Spark returns [1,[null,10]] and [2,[null,50]], the list test returns [2,20], [3,30] and the map test k -> [4,40] in place of nulls. With the gate as pushed all three pass, on Spark 3.5 and 4.1.3.

[andygrove]

#6004 now declines a requested schema that repeats an id at planning, so this test covers a read Spark can no longer hand to the native scan, and the reasoning in its doc comment no longer applies. The reachable version is @sunchao's shape from the first round, s<x (id 1), y (id 1), z (id 2)> read as s<x (id 1), y (id 3), z (id 2)>. main returns all three values there, Spark raises, and this branch raises too. Its test (test_field_id_read_rejects_duplicate_ids_despite_matching_names) was dropped in b4549c9d2. Could this one use that shape instead?

[dwsmith1983]

Could this one use that shape instead?

Yes. The test is now parquet_duplicate_file_field_id_rejected_when_requested and uses the shape from the first round: the file holds s<x (id 1), y (id 1), z (id 2)> and the read asks for s<x (id 1), y (id 3), z (id 2)>, so the duplicate sits only in the file and a planner that declines repeated ids in the requested schema still hands this read to the native scan. The scan raises the duplicate id error for requested id 1 matching x and y, where a positional read would hand back all three values. The doc comment says why the identical-schema shape is no longer reachable and what this one proves instead.

[andygrove]

Spark brackets this list in matchIdField with mkString("[", ", ", "]"), so its message reads Found duplicate field(s) "1": [x, y] in id mapping mode and ours reads "1": x, y. Now that this helper formats the list for root and nested fields alike, could it add the brackets? The Scala test could then compare the whole message with Spark's instead of just the prefix.

[dwsmith1983]

Now that this helper formats the list for root and nested fields alike, could it add the brackets?

Done. field_names_with_id now returns the list bracketed and comma joined the way matchIdField renders it, so the message Comet hands to foundDuplicateFieldInFieldIdLookupModeError reads Found duplicate field(s) "1": [x, y] in id mapping mode on both sides. The Rust display string dropped its own brackets so the list is not wrapped twice, and the Rust assertions pin id=1 matches [x, y] rather than only the id.

On the Scala side, multiple id matches and duplicate field id inside a struct is rejected when a requested id matches two fields run the same read with Comet off, take the duplicate id message Spark raises, check that it lists the expected fields ("1": [a, rand2] for the root case, "1": [x, y] for the struct), and assert that Comet's message equals it in full.

[andygrove]

Thanks for the quick turnaround on today's comments. The brackets, the drop-and-re-add tests and sunchao's shape all look right, and I checked the 3.5 and 4.0 gating against ParquetColumnVector in 3.5.9 and 4.0.4, which do have that check. Since your runs were on 3.5, I ran everything on the default Spark 4.1 profile, and it's all green. I also disabled each of the new guards in turn, the relabel gate, the nested and root duplicate-id errors, and the nested raise in rewrite, and every one of them has a test that fails without it.

On narrowing id_duplicate_roots, I'd keep it as it is. I measured the shape on a file with roots a (id 1) = 1, a (id 2) = 2 and b (id 3) = 3, in both case modes. With the guard removed, Comet returns [1, 2] for a read of id 1 or id 2, which is two rows from a one-row file. Spark doesn't give a usable answer either. Its vectorized reader returns [1] for both ids, so id 2 comes back with id 1's value, and the parquet-mr reader fails outright. That's the same name-based leaf lookup #5964 describes. So the rejection protects the decoder rather than diverging from a Spark behavior we could match. Could the doc comment on known_divergence_repeated_root_name_rejects_unambiguous_field_id say that, and link #5964? As written it reads as a gap waiting to be closed, and closing it would bring #5783 back at the root.

[andygrove]

This turns main's error into an INTERNAL_ERROR. SparkError::Internal goes through SparkErrorConverter as SparkException.internalError, so a requested nested duplicate now reaches the user as [INTERNAL_ERROR] Found duplicate Parquet field name 'dup' SQLSTATE: XX000. That tells them they've hit a bug in the engine, for a limitation scans.md documents. On main the same read raises CometNativeException with the plain message. On this branch the root check and the non-pruning path still do, so the same limitation surfaces as two exception classes depending on nesting. In the test log, all ten duplicate Parquet field names fail clearly cases, the schema-merge test and the fixture test take the INTERNAL_ERROR path. The tests don't notice because they only match the message text. Could the resolver raise the same error main does here? And would you add an assertion on the exception class in one of those tests, so the paths can't drift apart again?

[andygrove]

Is there something about duplicate-nested-names.parquet that a Spark-written file can't reproduce? It was written by pyarrow 25.0.1, and nothing in the repo says how to regenerate it. The two reads it covers, s.other succeeding and s.dup refusing, are the pair CometNativeReaderSuite already covers on main with a file written from named_struct('dup', id, 'dup', id + 100, 'other', id + 900). If the pyarrow writer is the point, could the test say why? Otherwise I'd drop the test and the binary, which also takes a little off the diff comphead was concerned about.