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 in schema conversion is too late because decoding has already combined the leaves. The issue explicitly accepts a clear error instead of Spark-compatible duplicate selection.

What changes are included in this PR?

• Validate sibling names recursively when the native reader loads metadata, before decoder construction. Validation also runs after cache hits and covers structs inside arrays/maps.
• Reject the entire file, including unprojected duplicates and reads using field IDs, in either case-sensitivity mode. This deliberately conservative behavior is documented in the Parquet compatibility guide. Users can disable Comet to use Spark's duplicate-name resolution with an explicit schema.
• Preserve the validation requirement in comments describing eventual replacement of the page-index reader workaround.
• Add 12 regression/control cases to the existing native-reader suite. Duplicate fixtures contain three rows in one file, covering batch sizes 1 and 4096. Other checks cover repeated reads, unprojected duplicates, distinct-case names, and repeated names in separate groups.

How are these changes tested?

Validated commit 513d6fc26 on Apollo with Spark 4.1, JDK 17, and Rust 1.97.1, after rebasing onto 17f54da8c (#5751). The native library was rebuilt from that commit.

• RED on unpatched fefee03d9: all 11 negative cases fail with the single-file fixtures; the valid-name control passes.
• GREEN: 139 Scala tests passed across CometNativeReaderSuite and ParquetReadV1Suite, including all 12 added cases and the newly merged case-insensitive cases. One existing NullType test is canceled (Spark 4.1 NullType parquet: parquet-rs rejects BOOLEAN + Unknown logical type #4199), and one existing optional-struct-field test is ignored.
• All 18 ParquetEncryptionITCase tests passed.
• Two native reader/cache tests and the upstream structural-narrowing compatibility test passed.
• Full Maven verify, including packaging, formatting/style, and Apache RAT checks, passed. Changed Rust files pass rustfmt; git diff --check passes.

make core
./mvnw verify -Pspark-4.1 -Dtest=none \
  -Dsuites=org.apache.comet.exec.CometNativeReaderSuite,org.apache.comet.parquet.ParquetReadV1Suite
export LD_LIBRARY_PATH="$JAVA_HOME/lib/server${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}"
(cd native && cargo test -p datafusion-comet parquet::parquet_exec::tests --lib)
(cd native && cargo test -p datafusion-comet structural_narrowing_requires_unambiguous_exact_match --lib)

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