[sam-1112]

Which issue does this PR close?

Closes #6994.

Rationale for this change

When one Parquet field is requested by both ID and name, Comet returns NULL for the losing name match at the root but duplicates the values inside a struct.

Apply shared claim arbitration at both levels so the ID match receives the data and the name match reads NULL, following Spark parquet-mr’s behavior for this case.

What changes are included in this PR?

• Move field claim precedence into FieldResolver, shared by root remapping, nested conversion, and plan-time conversion checks.
• Resolve repeated requested ID claims with last-wins semantics, matching Spark parquet-mr. Retain the existing JVM fallback for these schemas.
• Preserve shadowing, placeholder naming, ambiguity checks, and error ordering.
• Prevent generic structural casts from bypassing claim arbitration.
• Add regressions covering nested containers, projections, pruning, filter pushdown, case sensitivity, ID configuration, off-heap reads, and repeated requested IDs.

How are these changes tested?

Local validation:

• Native library tests: 982 passed, 5 ignored.
• Final focused regression tests on Spark 4.1.3: 69 passed.
• Focused regression matrix passed on Spark 3.4.3, 3.5.9, 4.0.4, and 4.2.0.
• Full ParquetReadV1Suite and CometNativeReaderSuite passed on Spark 4.1.3.
• Formatting checks passed.

The cross-profile and full-suite runs preceded the final test cleanup, which removed three redundant root pruning cases and reused the parquet-mr helper. The resulting 69 focused cases were rerun successfully on Spark 4.1.3.

[sunchao]

Summary

• Prior state and problem
  When x matched a Parquet field by ID and a matched the same field by name, the root gave the value to x. Inside a struct, Comet returned it twice.
• Design approach
  FieldResolver now assigns each physical field one owner. Root remapping and nested conversion share that decision.
• Correctness
  The original nested fixture now returns (7, NULL), matching parquet-mr. I reproduced two regressions on separate reads. Projecting one repeated-ID alias can silently return NULL, and a losing struct request can hide a schema error that Spark raises.
• Compatibility analysis
  Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0 clip each requested field independently. Parquet-mr then maps physical IDs to output positions, with the last requested field winning a repeated ID. Both Spark readers agreed on the single-alias projection and the incompatible struct in runs on 4.1.3 and 4.2.0. Spark schema clipping · Spark row conversion
• Key design decisions
  ID matches outrank shadows, which outrank name matches. Equal ID claims keep the last requester and equal name claims keep the first. The JVM fallback checks for repeated IDs in Spark’s required schema.
• Implementation sketch
  file_claims() drives root renaming. resolve_all() clears losing indices for nested conversion and file-open validation. is_pure_structural_narrowing() routes competing target names through Comet’s converter.
• Performance
  On a conflict-free struct with 64 renamed, ID-bearing children, nine paired runs measured 6.8% more CPU for Parquet decoding plus conversion at 64 rows per batch. A single-lookup variant recovered 5.5% against the head. The 8,192-row control showed no repeatable slowdown.
• Design
  Sharing precedence removes the original root-versus-nested inconsistency. The integration tests cover structs, arrays, maps, pruning and filters. The repeated-ID Scala tests request every alias together, so they exercise the fallback and miss the native single-alias projection.
• Abstraction & complexity
  FieldClaim makes ownership explicit. The nested result uses None for both an absent field and a selected field that lost ownership. Retaining the validated selection alongside its owner would preserve the information needed for structural checks and avoid selecting each child twice.
• Behavioral changes worth calling out
  All four touched files matched branch-1.2 at the supplied base. Relative to that release, the PR fixes nested duplication and changes repeated root IDs from first-wins to last-wins. The new reader diagnostic also writes indeterminate vectorized values to ScalaTest output.
• Suggested improvements
  Add native-versus-Spark regressions that project each repeated-ID alias separately and that request a struct through the losing name match of a primitive. Keep the vectorized diagnostic at reader selection and compare row values in the parquet-mr and Comet branches.

Reviewed the full four-file diff from 330b29cf1013351cd9d7e93d3d37a6bc0b6d956b to b5cb98bf1b6f194797c133fd9f170c8bc4aa2b92 using review-comet-pr. The changes stay within Parquet schema adaptation and its reader tests, so no subsystem-specific sibling skill applied. The PR was open and non-draft. Its reviews, conversation and inline threads were empty on recheck.

At this SHA, labeling passed. Comet CI and CodeQL both reported action_required and ran zero jobs. Labeling · Comet CI · CodeQL

Validation included Spark-written Parquet files, base/head DataSourceExec scans, 11 passing isolated head tests and four passing base controls. The scan probes used the reviewed adapter sources with DataFusion 55.2.0 and cached Arrow/Parquet 59.3.0. The optimized timing harness used the locked Arrow/Parquet 59.3.1. The configured Cargo registry lacked locked aws-config 1.13.0, preventing a full workspace build and local CometNativeReaderSuite run. The logging finding was verified from source and the documented reader behavior.

[sunchao]

[P2] An unprojected alias can take the selected column’s value. With field-ID reads enabled, write A = 7 with ID 1, supply a read schema with x and y both carrying ID 1, and select only x. The JVM guard sees only x in requiredSchema. Root remapping sees both fields and this tie-break assigns A to unread y, so SELECT x silently returns NULL. Extend the duplicate-ID fallback to cover a required root whose ID also occurs on another root in the serialized schema.

Evidence: Spark 4.1.3 and 4.2.0 each showed requiredSchema=[x] and relation.dataSchema=[x,y]. Parquet-mr and the vectorized reader both returned 7 for either alias projected separately. On the same files, the base native adapter returned 7 for x. The head rewrote x to a literal NULL and its DataSourceExec scan returned NULL.

[sam-1112]

Thanks!! I have fixed by checking required roots against the actual data schema produced by nativeDataSchema(). With field-ID reads enabled, the native scan now falls back when a required root shares its ID with another serialized data root. Added regressions that project x and y separately in both schema orders, comparing against both Spark readers and asserting the fallback reason. Both aliases return 7. Control cases confirm that selecting an unrelated unique-ID root or disabling field-ID matching still uses the native scan.

[sam-1112]

@sunchao After committing the extended fallback check, I tried a small local PoC that limits claim arbitration to the required roots. For this example, both SELECT x and SELECT y stay on the native path and return 7. However, I haven't fully validated the approach yet.

Would it make sense to track this in a separate follow-up issue?

[andygrove]

A follow-up makes sense to me, and I filed #7206 for it. The name-match version of this shape is still wrong after the new fallback, and on main too. With [x id1, a, z id2] over a file [A id1, z id2], select("a", "z") returns [7, 9] in both Spark readers and [null, 9] in Comet, because the unread x still takes A by id. No id repeats there, so the fallback can't see it. Limiting claims to the required roots, as in your PoC, should fix both shapes, and then the fallback could go.

[sunchao]

[P2] Using the arbitrated indices for file-open validation hides an incompatible nested shape. With case-insensitive field-ID reads, write s.A as BIGINT with ID 1, then request s.x BIGINT with ID 1 and s.a STRUCT<v:BIGINT> without an ID. Spark clips the physical primitive against the requested struct and rejects the read. resolve_all() clears the losing index before check_conversion sees it, so Comet returns (7, NULL). Validate independently selected container shapes before masking losers for value conversion.

Evidence: On the same one-row fixture, both readers on Spark 4.1.3 and 4.2.0 raised ClassCastException: optional int64 A = 1 is not a group. The base native adapter rejected INT64 at [s, A]. At head, check_file_conversions accepted the schema and the native scan produced x=7, a=NULL.

[sam-1112]

I have fixed by retaining independent selections for validation and masking losers only for value conversion. Losing container requests now raise structural errors, while losing leaf conversions retain their NULL behavior.

[sunchao]

[P2] resolve_all() selects every requested child through resolve() and selects it again through file_claims(). A nested conversion repeats the ID parsing and lookups on every batch, including when every child has a unique match. Record claims from the first validated selection, preserving requested-order ambiguity errors, then derive the output indices from the owners.

Evidence: An optimized, CPU-pinned probe decoded 32,768 rows with 64 constant INT32 children, unique IDs and renamed targets. At 64 rows per batch, converter CPU rose from 27.8 to 33.1 µs per batch. Decoding plus conversion was 3.6–8.7% slower in all nine alternating base/head pairs, with a 6.8% median increase and identical output. A single-selection variant lowered the combined cost by 5.5% against head.

[sam-1112]

resolve_all() now selects each child once and records claims from that validated selection, preserving requested-order ambiguity errors.

I added parquet_field_resolution.rs to benchmark conversion and decode-plus-conversion with 64 unique-ID and renamed children at 64/8192 rows per batch. Nine alternating paired runs showed ~9.2% and ~4.2% lower elapsed time respectively at 64 rows/batch; the 8192-row combined result was essentially flat.

[andygrove]

I ran parquet_field_resolution against main and this head, alternating the two binaries six times, with LTO off and 16 codegen units for both. At 64 rows per batch the head is still slower than main. convert/64 averaged 15.20 µs against 14.47 µs, between 3% and 6% slower in every pair. decode_and_convert/64 averaged 15.95 ms against 15.61 ms, between 1% and 3% slower. At 8192 rows the difference was within noise. So the gain in your reply looks like it's measured against the earlier head rather than main.

The work that's new on every batch is the claim vector and a name lookup for each id-bearing field. Could a group without competing claims skip that, so it costs what it did before?

[sam-1112]

I added a fast path for nested groups resolved entirely by ID. These groups skip both the claim vector and the shadow-name lookups. Mixed groups initialize ownership bookkeeping at the first name selection, reusing the preceding validated ID selections. Independent type validation and ambiguity checks remain intact.

I reran six alternating pairs against main (7deeba21a), using the same harness and lockfile, with LTO off and 16 codegen units for both builds:

Benchmark	main	Fast-path head	Change
convert/64	13.880 µs	13.694 µs	−1.34%
decode_and_convert/64	15.337 ms	15.199 ms	−0.90%
convert/8192	13.990 µs	13.782 µs	−1.49%
decode_and_convert/8192	1.112 ms	1.110 ms	Within noise

[sunchao]

[P2] The vectorized off-heap case formats every collected row into a ScalaTest info event. Issue #6994 documents that the losing field can contain stale JVM memory, so a passing test can write those values to CI output. Keep the plan assertion for that reader and move collection and row formatting inside the branches that check defined results.

Evidence: df.collect() and rows.mkString both run before the if (comet || !vectorized) guard. Spark 4.1.3 honors the test’s off-heap setting for task reads. The suite runs in the Linux and macOS workflows, and the ScalaTest configuration writes to stdout and SparkTestSuite.txt. Reader behavior documented in #6994

[sam-1112]

The affected Spark vectorized branch now checks only the plan. Collection, row formatting, and value assertions are inside the if (comet || !vectorized) guard, so only Comet and parquet-mr results are collected and logged.

[andygrove]

I ran the new tests locally on Spark 4.1 and all 94 pass. I also checked the #6994 shapes against both Spark readers and against main, and the struct case now returns NULL for the name match as parquet-mr does.

There's one change I think this needs before it merges, which I left inline on check_conversion. With this PR, a nested field that loses its claim no longer gets its deferred type rejection, so a read that fails in Spark's default reader, and in Comet on main, now returns NULL.

The user guide still lists #6994 under "Reads that differ from Spark", at docs/source/user-guide/latest/parquet.md line 274, and after this PR the struct half of that entry is wrong. Could you move it to "Differences that are not Comet bugs" and say that Comet now returns NULL for the sibling at every level, as parquet-mr does, while Spark's vectorized reader returns zeros, or stale memory with off-heap column vectors?

While testing I found two root-level cases that are also wrong on main, so I filed them rather than asking for them here. #7206 is an unread id alias taking the file column from a read column that matches it by name. #7207 is a root field that loses its claim skipping the type check Spark runs on it.

[andygrove]

Masking a losing leaf's deferred rejection changes the answer from Spark's default reader here. With s struct<x id1 bigint, a int> over a file s struct<A id1 bigint>, Spark's vectorized reader fails with SchemaColumnConvertNotSupportedException: column: [s, A], physicalType: INT64, logicalType: int in either field order. That's because VectorizedColumnReader.readBatch calls getUpdater for every requested leaf, including the one whose pages the other reader already drained. Main raises the same exception through RejectOnNonEmpty. With this change Comet returns (7, null), which is parquet-mr's answer. An empty file still reads in all three.

Comet otherwise follows the vectorized reader's conversion checks and doesn't replicate parquet-mr's null-on-narrowing. So could the losing leaf keep its deferred rejection here? It would still read NULL whenever Spark can read the column as its type. That would flip the Int32 cases at depth 1 and 2 in conflicting_field_claims_read_null_at_every_depth, and the leaf case in losing_field_claims_preserve_open_errors_but_not_deferred_rejections. The root has the same gap on main, which I filed as #7207.

[sam-1112]

Losing nested selections now retain their RejectOnNonEmpty result; ownership only determines whether runtime conversion supplies the value or NULL.

I updated the depth 1/2 Int32 cases and the losing-leaf validation test accordingly. Regression coverage confirms that both field orders fail with the expected [s, A] conversion error on non-empty input, while empty files and fully pruned row groups still read successfully. Compatible losing fields continue to return NULL.

[comphead]

Sharing one FieldClaim resolver between root remapping and nested conversion, and removing RootResolution rather than copying it, is the right shape. andygrove's three requests are still open at this head, so these are separate from them. They are about code and tests that I could not find a read path for, which matters for a change of this size.

[comphead]

With the new required-root fallback, I could not find a read that gives one file field two id claims, two name claims, or a shadow plus a name claim. A repeated required id falls back in CometScanTypeChecker (top-level and nested), a required root sharing an id with an unread root falls back in the new hasDuplicateFieldIds(fields, requiredFieldNames), and siblings whose names fold together fall back through hasCaseInsensitiveDuplicateFieldNames and hasDuplicateFieldNames. A tie between two unread roots changes no projected column. That leaves id against name, the #6994 shape. If that holds, this last-wins arm only runs in unit tests, and main's tie order could stay. Then repeated_requested_ids_read_last_field_at_root_and_nested, id_ties_keep_last_and_name_ties_keep_first, three_or_more_id_claims_keep_last_over_name_and_shadow and shadowed_struct_fields_follow_all_missing_null_policy, about 200 lines, could go, or shrink to one test if parquet-mr parity is the reason to keep it. I traced the guards and have not run them.

[comphead]

I think this check cannot fire. An id-bearing target already returns false at the use_field_id check above. Two targets that fold together but differ in case leave one without an exact-name source, which returns false at the find(|f| f.name() == target_field.name()) below. What remains is byte-identical sibling names, which DataTypeSupport.isTypeSupported declines (struct with duplicate field names). is_pure_structural_narrowing_denies_conflicting_name_claims builds exactly that ("a", "a") shape. Could the check, its per-target String clone and that test go?

[comphead]

All 12 cases request every alias together, so on the Comet side they assert the fallback main already has: duplicate Parquet field ids among top-level fields at the root and struct with duplicate Parquet field ids when nested. Those are already pinned by "native scan declines a nested struct whose fields repeat a Parquet field id" and "native scan declines top-level fields that repeat a Parquet field id" (lines 1826 and 1861). count and reverse do not change which guard fires. Could this block go, or shrink to one root and one nested case if the parquet-mr reference check is worth keeping?

[comphead]

Now that the Spark vectorized branch only checks the plan, offheap no longer changes anything this test asserts. Only Spark's vectorized reader reads spark.sql.columnVector.offheap.enabled, and I could not find it under spark/src/main or native. So two of the four cases repeat the other two. The rows are also covered by the "Parquet field claims prefer id over name" matrix, which adds both field orders and parent nulls. Could this block go, with IntegerType added to that matrix if the type matters?

[sam-1112]

Thanks, @comphead . I checked the guards and ran the corresponding read-path tests. They support your analysis, so I made the following simplifications:

• Removed the repeated-ID last-wins arm, retaining main’s first-wins tie order, and deleted the Rust tests you listed.
• Removed claimed_names, its per-target String clone, and is_pure_structural_narrowing_denies_conflicting_name_claims.
• Reduced the 12-case repeated-ID Scala matrix to one root and one nested case, preserving the parquet-mr reference check and explicit fallback assertions.
• Removed the separate off-heap reproduction block and added IntegerType coverage to the existing id-over-name matrix, retaining both field orders and parent nulls.

The required/unread-root projection tests and existing duplicate-ID/name fallback guards remain in place.

[sunchao]

The revised reader keeps file-open compatibility checks separate from deciding which requested field receives a value. That fixes the losing-container case from my earlier review while preserving the all-ID fast path. The nested-conflict tests cover both the error and the returned values, and the native tests and Spark 4.1 scan shard passed on this head. Tracking projection-only arbitration separately in #7206 makes sense to me.

[sam-1112]

Hi @andygrove, I've addressed your review comments, and the PR is ready for another review. The remaining CI failures appear unrelated to this PR. Thanks for taking another look!

[andygrove]

Thanks @sam-1112, all three of my earlier points are addressed. I ran the head merged with current main: the 376 native parquet tests and the 92 "Parquet field claims" Scala tests pass on Spark 4.1. The losing-leaf test covers my shape in both field orders and compares the native scan with Spark's vectorized reader. The parquet.md entry now sits under differences that are not Comet bugs. In a six-pair A/B of parquet_field_resolution against main, convert/64 and decode_and_convert/64 are at parity (about 1–2% faster), so the +5% at 64 rows is gone. decode_and_convert/8192 is about 1% slower in every pair, which I don't think needs anything.

[sam-1112]

Thanks for the approval, @andygrove !

Could you please rerun the CI checks when you have a chance?