fix: reject duplicate Parquet field names before decoding

[ErikBPF]

Which issue does this PR close?

Closes #5783.

Rationale for this change

Native Parquet scans can silently multiply rows when a struct contains byte-identical sibling names. Resolving or rejecting duplicates after decoding is too late because the decoder has already combined distinct leaves. Reject ambiguous fields before decoding while allowing unrelated duplicate siblings that the reader safely prunes.

What changes are included in this PR?

• Validate sibling names when loading native-reader metadata, including cache hits, before constructing the decoder. Cover nested structs, arrays, and maps in both case-sensitivity modes.
• Preserve the required schema through validation. Skip unselected top-level columns, and skip unrequested nested fields only when the shared structural-narrowing check establishes that they will be pruned. Keep full-subtree validation when missing fields or type conversions require full decoding. Field-ID reads validate the entire file schema.
• Keep full nested validation for embedded Arrow schema hints and synthesized Spark variant schemas because they can change the schema the decoder sees.
• Document the supported projection behavior and retain explicit rejection of ambiguous decoded fields. Spark-compatible duplicate-name resolution remains separate work.

How are these changes tested?

The new nested-projection regression reproduced the reported duplicate-field error on the previous implementation. It reads the unique other child from a struct containing two dup children, compares Spark and Comet results, and checks the exact three rows in both case-sensitivity modes. Additional controls retain duplicate rejection when the reader must decode the full subtree.

A native regression writes a real Parquet footer whose embedded Arrow schema restores a dictionary type. It failed before the conservative schema-hint guard and passes with it.

Orion verification on Spark 4.1.3 / JDK 17 passed:

• Full CometNativeReaderSuite: 71 succeeded, 0 failed, 1 existing cancellation for the NullType limitation tracked by Spark 4.1 NullType parquet: parquet-rs rejects BOOLEAN + Unknown logical type #4199 / SPARK-54220.
• Four focused projected_fields Rust tests, including the dictionary-hint regression.
• Native build, Rust formatting, Maven formatting checks, and git diff --check.

(cd native && cargo test -p datafusion-comet projected_fields --lib --offline)
(cd native && cargo build)
./mvnw -B test -Dtest=none \
  -Dsuites=org.apache.comet.exec.CometNativeReaderSuite

These are local Orion results for the follow-up changes. Earlier full CI results belong to the previously published head.

[ErikBPF]

@andygrove created this pr to address your recent issue. When you have the time could you please check the provided solution?

[sunchao]

Correctness

Reviewed 23efb2437d868916b313c4c2405bb98d26ce293d against 4eeb1f80f0541f72389a11e6e2d0ee269d648c23. I found no actionable correctness issue in this change.

The prior reader could decode byte-identical sibling names into multiplied rows or fail after its column readers lost synchronization, as reported in #5783. This adds a recursive physical-schema check in EagerPageIndexReader::get_metadata, after metadata retrieval and before Arrow schema construction and decoding. The locked DataFusion 55.0.0 / Parquet 59.3.0 sources confirm that cache hits still pass through this check. Decryption options and the existing page-index policy remain intact. Filter pushdown retains the factory. Files eliminated before metadata loading are never decoded.

On the maintained Spark 3.5 and 4.0 branches, case-sensitive name lookup selects the last identical sibling, case-insensitive lookup rejects multiple matches, and enabled field-ID lookup can resolve fields independently of names. This PR deliberately chooses the clear-error option accepted in #5783: it rejects duplicate physical names even if they are unprojected or have distinct IDs. The compatibility guide states this narrower behavior and the option to disable Comet. Unique sibling names are unaffected by this check. Case-distinct names are allowed here and remain subject to the existing case-insensitive ambiguity checks. Each group has its own name set, including nested LIST/MAP groups, so names in separate structs do not collide. Since validation precedes values, nulls, batch boundaries and numeric conversions cannot bypass it. Maintained Spark 3.4/4.1 source branches were unavailable. No source-level compatibility claim is made for those versions.

Validation

The 12 added cases cover two/three identical children, an additional distinct sibling, array elements and map values at batch sizes 1 and 4096, plus repeated reads, unprojected duplicates in both case modes, and a valid separate-group/case-distinct control. The failure cases assert a native scan and the specific new error. Repeated reads exercise the path but do not independently prove a cache hit. The cache guarantee follows from the inspected call chain.

The author reports 139 Scala tests and 18 encryption tests passing at 513d6fc26, plus native reader/cache and structural-narrowing checks. The reader factory, scan setup, regression suite and Cargo lock are unchanged between that commit and this head, but inherited timestamp-conversion changes make the overall trees different. Those reports are historical evidence. At the September 9, 10:30 UTC refresh, CI, CodeQL and the Delta gate were action_required. Only labeling had succeeded. Current product compilation/execution is therefore unverified. I ran source/whitespace checks, not a local product build or test.

Performance

The new work is an expected linear walk over physical schema nodes for each metadata request, using one HashSet per group and borrowed names. It adds no per-row or per-batch work, column copies, or object-store reads. Cache hits repeat this walk intentionally so cached metadata cannot bypass validation. Allocation depends on schema width and nesting. No benchmark was supplied or run, so this review does not claim a measured throughput improvement or quantify the cost for very wide schemas.

Design

The metadata boundary is the appropriate place to prevent this decoder failure: resolving names later in the schema adapter cannot undo rows already combined by decoding. Checking the entire physical schema also keeps the safety rule independent of projection and field-ID adaptation. This is a conservative compatibility tradeoff, explicitly documented, rather than an implementation of Spark's duplicate selection. The existing page-index factory already owns this metadata path, and both its module documentation and installation site now require preserving validation when that workaround is replaced. Future Spark-compatible selection would need safe duplicate handling before decoder construction. No additional abstraction is needed for this error-based fix.

Abstraction & complexity

The change adds one private recursive helper and reuses the existing Parquet error channel. A separate set per group directly expresses sibling uniqueness, without normalization or cross-group state. Tests extend the existing native-reader suite, and the two preservation comments explain the otherwise easy-to-miss lifetime of the guard. I found no actionable complexity or abstraction issue.

[andygrove]

Thanks for picking this up. I checked the branch out locally, built it, and ran CometNativeReaderSuite (74 passed, plus the one pre-existing NullType cancel). To get a baseline I commented out the single validate_field_names call and rebuilt, which reproduces main's behavior for this path exactly, then ran the same probes against both builds.

The thing I keep coming back to is that the guard rejects the whole file regardless of what the query projects, so a query that returns the right answer today starts failing. On a file written as spark.range(3).selectExpr("id", "named_struct('dup', id, 'dup', id + 100) as s"):

query	Spark	main	this branch
spark.read.schema("id bigint")	3 rows	3 rows, correct	error
same plus where id > 1000, so every row is pruned	empty	empty	error

Since an explicit read schema is the only way to read one of these files at all, one bad struct makes the entire file unreadable by Comet, and the only escape is turning Comet off for the query. I don't think that follows from #5783. I said a clear error was acceptable for the case that returns wrong results, not for queries that are correct today.

Would you consider scoping the walk to the subtree reachable from the required schema? The required schema is right there in init_datasource_exec, so the factory could be constructed with the folded top-level names and skip root children outside that set while still recursing fully into the selected ones. When use_field_id is set names don't identify the projection, so that case would keep the current whole-schema behavior. That still closes #5783 and leaves the currently-correct queries working.

Second thing. validate_field_names runs on root_schema(), so duplicates in the root group are one of the two branches it guards, but every new test builds its duplicate with named_struct and can only reach the nested branch. I said in the issue that top-level duplicates were unreachable, which is true of Spark's writer but not of Parquet, and this suite already has writeDirect at line 1036 for writing an arbitrary MessageType through a raw RecordConsumer. I tried it with

message spark_schema {
  optional int64 a;
  optional int64 a;
  optional int64 b;
}

and a single row a=1, a=2, b=3. On main, reading schema("a bigint") returns two rows from a one-row file where Spark returns [1], and reading schema("b bigint") is correct on main but errors here. So this PR is also fixing a root-level wrong-results case that nothing currently asserts. Could you add it? A handful of Rust unit tests directly on validate_field_names would be cheap too, and would cover shapes Scala can't write: a LIST element group, a MAP key_value group, and same-name-in-separate-groups. I wrote six against this branch and they all pass in under a millisecond.

On the batch-size dimension, that was clearly load-bearing for your RED run, where 1 vs 4096 decided whether you got multiplied rows or a desync error. Now that the check fires in get_metadata before any decoder exists, both arms run identical code and assert the identical message. Would you swap those five duplicates for the root-group case above? Same test count, more of the function covered.

Dropping the #5783 link from the docs makes sense since this closes it, but could you file a follow-up for the Spark-compatible resolution and link that instead? The datetime rebasing entry just above links #5010 the same way, and as written the limitation reads as permanent with nowhere to track it. Worth capturing in that follow-up: matching Spark isn't one rule. On the two-a file above Spark resolved the root-level duplicate to the first child, while #5783 found last-wins for the nested case through caseSensitiveParquetFieldMap. That's a good argument for erroring first, which is what you've done.

Last, this needs a rebase and eager_page_index_reader_factory.rs has moved a lot underneath it, from 224 lines to about 1050 on main via the scan I/O metrics work (#5453) and the Variant projection work (#5794). get_metadata now binds the fetch as a Result, records metrics off it, unwraps with let metadata = metadata?;, and ends in if spark_variant_schema { with_spark_arrow_schema(metadata) } else { Ok(metadata) }. The validation wants to go straight after that unwrap and before the branch so both arms are covered. Please re-run the new suite after the merge, that placement is easy to get subtly wrong in a conflict resolution.

A few things I checked that are fine, so you don't have to. The factory is installed at the only production ParquetSource::new site, so every native scan is covered. Encrypted opens go through the same get_metadata. The error reaches the user with the file path attached, since Spark wraps it in FAILED_READ_FILE.NO_HINT, so there's no need to add the location to the message. And I measured the cost of the walk on a wide schema (1000 leaf fields, 20 files, every open a metadata cache hit): median 56.3ms without the validation against 57.2 to 59.3ms across three runs with it, which is inside the run-to-run noise. No perf concern.

[ErikBPF]

Thanks for the detailed review and reproductions. Addressed the requests in b696cc3 and rebased onto main, preserving the updated reader metrics and Variant handling.

• Restrict duplicate validation to required top-level columns, recursively checking each selected subtree. Field-ID reads retain conservative whole-schema validation; empty projections skip all roots. Metadata-cache hits remain validated.
• Added raw root-duplicate coverage plus unrelated-column, case-insensitive, repeated-read, pruning, count-only, and renamed field-ID cases. Added Rust LIST/MAP/separate-group coverage and removed the redundant batch-size dimension.
• Updated the compatibility documentation and opened Support Spark-compatible duplicate Parquet field resolution #5884 for Spark-compatible duplicate resolution. Independent Spark 4.1.3 vectorized testing found reader-dependent nested behavior; the follow-up includes that reproduction rather than assuming universal last-wins semantics.

Validation: reproduced both valid-projection failures before the fix. Afterward, the full Spark 4.1 native-reader suite passed 70 tests (one existing NullType cancellation), and all 8 focused cases passed on Spark 3.5. Rust Parquet tests: 188 passed, one existing ignored benchmark. Native build, whole-reactor packaging, all-target workspace Clippy with warnings denied, semantic/syntactic Scalafix, Spotless, formatting, and whitespace checks passed.

[sunchao]

Re-reviewed b696cc31951c223d9d68a0768fb3958970d77753 against de1eb4f86c12af0895784c93e1f152c705f6ef0e. No new or remaining P1/P2 findings.

The update addresses the unprojected-column regression in the earlier review: name-based reads check required top-level roots with the existing case-folding rules, recurse through each selected subtree, and skip empty projections. Field-ID reads retain the documented whole-schema check. Validation runs after metadata retrieval and metrics recording, before the Variant branch and Arrow decoding; cached metadata follows the same path.

The new coverage includes raw root duplicates, unrelated columns, repeated reads, pruning, count-only reads, renamed field IDs, and Rust LIST/MAP/separate-group cases. The compatibility guide links #5884 for reader-dependent Spark resolution. The added work remains per metadata request; I did not run a performance benchmark.

At the September 12, 22:29 UTC refresh, CI had 56 successful and 10 skipped checks. I inspected logs confirming all eight duplicate-name cases passed on Spark 3.5 and Spark 4.1, plus all six new Rust cases. These jobs checked out merge commit 68fc20a8d75e524b3e5c80e550e7d5497692a02e; all four changed files and inspected supporting sources match the reviewed head. Five inherited base files make the complete trees different. No local product build was run. Canonical Spark source checks covered maintained 3.5/4.0 branches; maintained 3.4/4.1 branches were unavailable.

[andygrove]

The scoping to projected roots fixes the case I raised, but recursing into each selected root with projected_fields set to None still rejects a read that works today.

Take a file whose s group is dup, dup, other, read as spark.read.schema("s struct<other: bigint>"). Spark's clipParquetGroupFields iterates the requested fields only, so other resolves to a single child and the clipped read schema never mentions dup. I checked ParquetReadSupport on 3.4, 3.5, 4.0 and 4.1 and the matcher is the same on all of them. Comet gets there the same way today. is_pure_structural_narrowing returns true because other has exactly one folded match, replace_with_spark_cast leaves DataFusion's CastExpr in place, and build_projection_read_plan clips that cast down to the single other leaf, so the duplicate leaves are never decoded. That is the same leaf pruning the two issue #4859 tests in this suite assert. On this branch validate_field_names errors before any of it runs, and it holds in both case-sensitivity modes, so it is reachable under the default spark.sql.caseSensitive=false.

The comment on validate_field_names says nested projection does not safely separate duplicate leaves. That is true when the duplicate name is itself requested, because resolver_matches is then 2, is_pure_structural_narrowing returns false, and the whole root is decoded. It is not true when the duplicate is only a sibling of what was asked for. Would it make sense to carry the required schema down the recursion instead of dropping it at the root, and reject only when a requested field name matches more than one physical sibling? That is the rule Spark applies, and #5884 already describes the guard as covering selected ambiguous groups. A test for the shape that should keep working would be worth having too. A file written as named_struct('dup', id, 'dup', id + 100, 'other', id + 900) read back as s struct<other: bigint> returns the right answer on main and errors here, and nothing in the suite catches it.

This also lands on top of #5654, which is open and takes the opposite position on the same files. resolve_struct_mapping there resolves byte-identical siblings last-wins in case-sensitive mode, and shadowed_by_later_duplicate extends that to the root group. If this merges first none of that is reachable for a native Parquet scan, because get_metadata errors before the adapter runs. The two conflict textually as well. git merge-tree reports eight hunks in eager_page_index_reader_factory.rs and one in parquet_exec.rs, because #5654 hangs its own with_field_id_check validator off the same builder and the same get_metadata call site. Both are clean against main on their own, so neither CI run shows it. Your #5884 records that Spark 4.1.3 with the vectorized reader returned {0, 100, 1} for the nested fixture rather than last-wins, which argues against #5654's resolution as written. Could you and @dwsmith1983 settle an order, and note on #5654 whether its duplicate-name resolution should give way to the error here?

One more ordering point. fold_name and fold_schema_names become fallible in #5845, and this PR adds the only two new callers outside name_fold.rs. Both sit inside closures with nowhere to put an error. with_required_schema is a -> Self builder, and the fold_name call in validate_field_names is inside an is_some_and closure whose error type is ParquetError rather than DataFusionError. Whichever lands second will need to thread the Result through rather than reach for an unwrap.

[ErikBPF]

The validator now preserves nested projection information and reuses the reader's structural-narrowing check. The regression reads the unique other field beside duplicate dup siblings in both case-sensitivity modes, compares Spark results, and checks the exact rows. Selected ambiguity and full-subtree decoding still fail before decoding. Embedded Arrow schema hints and synthesized Spark variant schemas retain full nested validation because they can change the decoded schema.

The nested-projection regression failed before the fix. A second regression with a real dictionary-encoded Arrow schema hint failed before the conservative hint guard. Final Orion verification passed: four focused Rust tests and the full CometNativeReaderSuite with 71 succeeded, 0 failed, and one existing NullType cancellation (#4199 / SPARK-54220). The native build and formatting checks also passed.

For merge order, I suggest landing this decoder safety guard before #5654, then rebasing #5654 and preserving rejection until its last-wins path has evidence that ambiguous leaves decode correctly. #5845 is also still open; whichever lands second needs to propagate fallible name folding through the validator, projection decision, and shared structural-narrowing helper without unwrap. Please coordinate that order before merging.

[andygrove]

Thanks for the rework. The scoping does fix the case I raised, and I confirmed the pruned reads work: s struct<other>, a struct nested one level deeper, and s array<struct<other>> all return the right rows against main's answer, with and without a pushed filter and with rowFilterPushdown on. I also reproduced your run on the head: CometNativeReaderSuite 71 passed with the one NullType cancel, 190 Rust parquet tests, clippy and fmt clean.

Before anything else, eager_page_index_reader_factory.rs:596 starts with // ponytail:. That looks like a tooling marker rather than something meant for the file. It's the only occurrence in the repo.

The thing I can't get past is cost. The walk in the previous revision was cheap, but reconstructing the projection on every get_metadata is not. Both the physical.fields().iter().find(...) in the selected map and the projected.iter().find(...) inside validate_field_names are O(n²) with a fresh fold_name allocation on both sides of every comparison. Measured on a release build over a flat file with every column projected:

leaf columns	previous revision	this revision
100	1.6 µs	363 µs
250	3.3 µs	2.2 ms
500	6.7 µs	8.8 ms
1000	13.6 µs	36.4 ms

That is 36 ms per file open on a thousand-column table, on every open including metadata cache hits, which is the exact workload the factory's own doc comment is about. parquet_to_arrow_schema is only about 270 µs of it, so the conversion is fine and the matching loops are the problem. Could the cheap walk run first? validate_field_names(root, None, ..) is strictly stricter than the projection-aware call, since the projected version only ever skips sibling pairs the full walk also checks. So if the walk passes you can return immediately and never build selected at all, which puts the common path back at 13.6 µs and confines the expensive analysis to the rare file that actually has a duplicate. When you do need it, is_pure_structural_narrowing right next door already folds each name once and says why: "O(sources), not O(targets x sources), matching this file's bulk-fold convention."

Second, I want to revisit the field-ID case. I accepted whole-schema validation there last round, but that was on the premise that names can't identify the projection. Field IDs can, and Comet already resolves them in remap_physical_schema via id_to_phys_names. On your own root-duplicate fixture, reading renamed_b by field id 3 returns [3] in Spark and on main, and errors on this branch. Your test asserts that error two lines after asserting that the same b read by name works, so the same column in the same file succeeds or fails depending only on whether the conf is on. Could the walk be restricted to the IDs the required schema resolves, the same way it's restricted by name?

On validate_field_type, the Map and FixedSizeList arms permit pruning that DataFusion never performs. nested_schema_pruning::clip_type clips (Struct, Struct), (List, List) and (LargeList, LargeList), and its own comment says "maps, dictionaries, fixed-size lists, views, is kept wholesale". Those arms are unreachable today only because is_pure_structural_narrowing returns false for Map and requires exact equality for FixedSizeList, but projected_fields_skip_unselected_nested_duplicates asserts a pruned DataType::Map is fine, so the helper's contract now records map pruning as safe. If someone extends is_pure_structural_narrowing to maps later, which is a natural follow-up to #4859, the guard silently starts skipping duplicates the decoder will read. Dropping both arms costs nothing, since they fall through to _ => validate_field_names(schema, None, ..), which is what happens today anyway. While you're there, could you add a line at is_pure_structural_narrowing's definition noting the second caller? Its doc comment reads as a pure optimization allow list, and it's now load-bearing for correctness.

A few test and doc points. The array shape the guard newly permits has no end-to-end coverage. I checked and it does work, but the Rust unit test exercises the validator in isolation and would keep passing even if clip_type stopped clipping through a list. The Scala case would catch that. In duplicate Parquet field names outside a nested projection remain readable, val name = "other" is fixed inside the Seq(true, false) loop, so both iterations are identical and the case-sensitivity dimension isn't exercised. The shape that would exercise it fails: S struct<OTHER: bigint> under caseSensitive=false gives the duplicate error while Spark returns three rows, because is_pure_structural_narrowing needs an exact name match. That's not a regression, main gives StructArrayReader out of sync, but the loop reads as coverage it doesn't provide. Same for maps: a pruned s map<string, struct<other: bigint>> read errors while Spark returns rows, and nothing asserts it. That matters for the scans.md wording, which says the check covers "structs, arrays, and maps" and that "safely pruned nested fields are skipped" and "applies in both case-sensitivity modes". A user reads that as "select only the unique sibling and you're fine", and maps, case-differing read schemas, and field-ID reads all contradict it.

Last, and I realize this is late to raise, but could we talk about the layer? #5783 traces back to #5602, which replaced the unconditional assert_eq!(field_name_to_index_map.len(), from_fields.len()) in parquet_support.rs with a duplicate error gated on !parquet_options.case_sensitive, so the byte-identical case now falls through to indices[0]. I raised that on #5602 itself.

I don't think a revert is the answer. #5602 is the Unicode fold fix for #5495, so reverting brings back silent NULLs for a file column like MÜNCHEN read as münchen, and it introduced name_fold.rs, which is now used throughout parquet_exec.rs, parquet_support.rs and schema_adapter.rs, including the two call sites this PR adds. #5845 exists only to make those folds fallible, so a revert takes it with it, and six commits have landed on schema_adapter.rs since. It also wouldn't give us what we want, because the assert was a panic rather than an error and, living inside parquet_convert_struct_to_struct, it never covered the root group. No 1.0.x exposure either, since #5602 isn't on branch-1.0.

What #5602 really did was split one blunt unconditional check into a proper Spark-matching error for the case-insensitive half and nothing for the byte-identical half. So the narrow fix is to give the other half an error too. I tried exactly that: change the guard so a collision also errors when case-sensitive, worded so it doesn't claim case-insensitive mode, and disable this PR's validate_field_names call so the metadata layer behaves like main. All five nested shapes in #5783 error cleanly, including the array and map ones. Every pruned read keeps working with no projection reconstruction at all, including the field-ID read of renamed_b. CometNativeReaderSuite gives 64 passed, with the only failures being this PR's seven new tests. The appeal is that checking the struct the decoder actually produced gets projection-awareness for free, so there's no parquet_to_arrow_schema per open, no coupling to is_pure_structural_narrowing, no Arrow-schema-hint conservatism, and no field-ID false rejection.

The gap is the root group, and that one isn't #5602's doing. message spark_schema { optional int64 a=1; optional int64 a=2; optional int64 b=3; } read as a bigint still returns two rows where Spark returns [1], so it needs its own check, and remap_physical_schema already folds every root name so it would be O(n) there. I'd also want to confirm the adapter path is reached when the required type happens to equal the physical type and no cast is inserted. The PR description says rejecting after decoding is too late because the decoder has already combined the leaves, and that's right for resolving, but in every shape I tried the rejection fired before any row came back. Would you be willing to try that direction before we spend more rounds on the projection reconstruction?

On ordering, #5654 is still open and still resolves byte-identical siblings last-wins in resolve_struct_mapping, and git merge-tree still reports conflicts in eager_page_index_reader_factory.rs and parquet_exec.rs. #5845 is also still open. Worth settling with @dwsmith1983 before either lands. And CI hasn't run at this head, only the label job, so the green run in the earlier review was the previous commit.