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.