fix: validate parquet field ids when the opener skips the expression adapter

[dwsmith1983]

Which issue does this PR close?

Closes #5801.

This branch is stacked on #5654 and shows its four commits until that merges; the change under review here is the last commit, fix: validate parquet field ids when the opener skips the expression adapter.

Rationale for this change

DataFusion's Parquet opener creates the physical expression adapter only when a predicate is pushed or the file schema differs from the requested schema. #5654 validates duplicate requested field ids inside that adapter once per file, so a file with no key-value metadata whose schema equals the requested schema, read with field id matching on and no data filter, skipped the validation and read a struct with two children sharing a requested id positionally. Spark rejects that during schema clipping. Spark-written files always carry key-value metadata that arrow-rs folds into the file schema, so they always reach the adapter; the gap is metadata-free files.

What changes are included in this PR?

• The Parquet reader factory, which fetches every file's footer regardless of what the opener decides, gains an optional field id check installed at plan time only when field id matching is on and the requested schema carries a field id at any depth, so ordinary reads pay nothing. After the footer is fetched it converts the file schema to Arrow the way the opener does and resolves the requested fields with the same mapping resolver the adapter uses, raising the same duplicate field id error. The adapter path is unchanged; this makes the two agree.
• A per-file memo keyed on the cached footer's identity means a file is validated once per distinct footer, not once per batch. The check costs one schema conversion and a linear walk per file, for field id reads only.
• The JNI error bridge walks the cause chain so a Spark error raised inside the reader factory keeps its Spark exception class instead of surfacing as a generic read failure.

One difference from the adapter path, deliberate: this check validates the columns the read projects, as Spark's clipping does, while the adapter's root-level check runs over the full data schema it is handed. A duplicate id inside a struct the read does not project is therefore reported only on the adapter path; a test pins the projected behavior.

How are these changes tested?

Rust: five new scan tests through DataSourceExec on a file written without key-value metadata, asserting first that the file schema equals the requested schema so the opener skips the adapter: the duplicate id is rejected, unique ids read, the check does not run with field id matching off, and the same two cases through the planner's data schema plus projection wiring. The rejection test read the struct as [42, 43] before the fix. Two tests for the error chain walk in the JNI bridge. Core crate 359 tests, bridge crate 30, clippy and fmt clean.

Scala: a ParquetReadV1Suite case writes the file with parquet-mr and no key-value metadata, asserts the footer's key-value map is empty and that the plan carries the native scan, and expects Spark's duplicate field id error; it reported no exception against the previous native library. The suite passes at 65 on Spark 3.5.

[sunchao]

Correctness

Reviewed c051b271016a93b13f3681605020f9c64ccf5541 against 424c31aa79d13fddf743ffa29bae3c6f146e6c5e, including all seven changed files and the four prerequisite commits currently in the diff. The final commit closes the gap where DataFusion skips the expression adapter for an equal-schema, predicate-free Parquet read. Previously that path could return a struct whose requested field ID matched multiple physical children. The new reader-factory check resolves the requested fields when the footer loads, before that shortcut can bypass validation.

I compared the lookup rules with the maintained Spark 3.5/4.0 ParquetReadSupport and ParquetUtils sources. Spark resolves ID-bearing requested fields strictly by ID, reports multiple matches for a requested ID, and uses name matching for fields without IDs. The shared resolver follows those rules through structs, lists and maps. It keeps missing ID matches separate from same-name physical fields. The footer check is enabled only for an ID-enabled read whose required schema carries an ID, including nested IDs, and checks that required schema rather than every unprojected column. The existing adapter stores nested ambiguities per column and raises them when that column is referenced. Its separate root-level validation remains unchanged in scope.

The prerequisite changes also preserve the distinctions between case-insensitive ambiguity and case-sensitive exact-name lookup, and prevent a metadata-only relabel from bypassing a non-positional field-ID mapping. Mappings are retained when expression children are replaced. The conversion continues to carry parent null visibility into checked timestamp conversion and to preserve the configured all-fields-missing struct behavior.

For the new footer error path, ParquetError::External and DataFusion's wrappers expose the typed error through their source chains. The bridge now finds that Spark payload before generic file-read classification. CometExecIterator passes it to the existing version-specific Spark error converter. Plain Parquet errors continue through the existing classification path. I found no actionable P1/P2 issue in the reviewed changes.

The added tests cover duplicate and unique IDs without footer key-value metadata, field-ID matching disabled, projected data-schema wiring, typed versus ordinary Parquet errors, and a Scala native-scan assertion for the metadata-free file. The author reports 359 core tests, 30 bridge tests, a 65-test Spark 3.5 ParquetReadV1Suite run, clippy and formatting checks. I did not independently execute those results. At the 2026-09-09 18:08 UTC refresh, CI, CodeQL and the Delta build gate all require workflow approval and have no jobs. Only labeling ran, checking out the base commit. There is therefore no current-head compile or runtime CI result to credit. Local verification was source review, dependency-source inspection and the whitespace check. Maintained Spark 3.4/4.1 sources were unavailable.

Performance

The new work is restricted to field-ID reads. On a footer that needs checking it performs an Arrow-schema conversion and mapping resolution. Successful checks are memoized by path and footer identity, avoiding the work on repeated opens of the same cached footer. The memo does not keep footer contents alive through a strong reference. Concurrent first opens can perform duplicate validation, but neither can publish success before validation completes.

The prerequisite mapping cache moves nested lookup work out of batch conversion. Normal conversion uses the stored indices rather than rebuilding name and ID maps for each batch. The additional footer check still duplicates some resolution when the adapter also runs, and the memo retains an entry per observed path for the factory's lifetime. Those are bounded per-file costs within the chosen design. No measured speedup or benchmark result was independently established, and I found no material performance issue to raise.

Design

Footer loading is an appropriate hook for this check because it precedes the opener's decision to skip rewriting. Reusing the mapping resolver keeps ambiguity handling aligned between the two paths. Installing the check with the required schema also makes projection scope explicit.

The memo uses the actual footer identity rather than path alone, so replacing cached metadata causes revalidation. Failed validation is not cached. Error translation preserves the typed Spark error instead of introducing a special message parser for this new reader path. These choices keep the additional behavior localized to validation and its error transport.

Abstraction & complexity

FieldIdCheck has a narrow responsibility: retain the requested schema and options, validate a footer, and remember successful identities. FieldMapping separates per-file resolution from per-batch conversion and is shared with the existing adapter. FieldMatch records one index and an ambiguity bit, constructing the matched-name list only for an error. The revision adds some state and a second validation entry point, but they serve the opener shortcut directly. I found no additional abstraction change needed before merge.

[andygrove]

I built this locally and ran the native parquet tests, the jni-bridge tests, ParquetReadV1Suite, CometNativeReaderSuite, CometIcebergNativeSuite and SparkErrorConverterSuite. All green. I also went looking for a way the new footer check could reject a read Spark accepts and could not find one. A requested id absent from the file still null-fills, and resolve_field_mapping only ever raises the two ambiguity errors, so it cannot turn a working read into a failure.

One thing I think deserves a sentence in the description. I went hunting for the same opener-skip hole on the case-insensitive duplicate name error and there isn't one, because for the file and requested schemas to compare equal the requested schema would have to hold two fields folding to the same name, and Spark rejects that at analysis with COLUMN_ALREADY_EXISTS. The duplicate id error is reachable precisely because two fields can share an id while having distinct names. That is what makes the narrow use_field_id && schema_holds_field_ids gate defensible, and right now a reader has to work it out for themselves.

Two things I would like to talk through before this goes in.

The first is the scope asymmetry the description calls deliberate. It says a duplicate id inside a struct the read does not project is reported only on the adapter path. I do not think that is what happens. resolve_nested_mappings stores the ambiguity per column and rewrite only raises it for columns the expression actually references, so an unprojected nested duplicate is silent on both paths. I checked with a scan wired the way the planner wires one, three-column data schema, projection of one, the duplicate sitting in an unprojected struct, adapter forced by a column the file lacks. It reads clean.

The asymmetry that is real sits at the root, and it runs the other way. remap_physical_schema loops over all of logical_schema.fields() rather than the projection, so a duplicate id on columns the read never asks for fails the query, while Spark answers it because clipParquetSchema only ever sees the requested schema:

val writeSchema = new StructType()
  .add("a", LongType, true, withId(20))
  .add("x", LongType, true, withId(1))
  .add("y", LongType, true, withId(1))
// write one row, then with spark.sql.parquet.fieldId.read.enabled = true
spark.read.schema(writeSchema).parquet(dir).select("a").collect()

Spark returns [7]. Comet raises Found duplicate field(s) "1": x, y in id mapping mode. It reproduces on main too, so this is not something you introduced, but this PR is the first place the two scopes sit next to each other and the footer check is the one that matches Spark. Would you narrow the root check in remap_physical_schema to the requested columns, or file it as a follow-up and reword the description so the asymmetry reads the way it actually behaves?

The second is that scan_bare_file_projected hardcodes case_sensitive = true, so none of the five new tests run in the mode Spark defaults to. That matters more than usual here because resolve_struct_mapping takes a different branch when case sensitivity is off and the footer check can now reach it. A metadata-free file holding s<x: long, X: long> where s carries field id 10 fails at footer load with [_LEGACY_ERROR_TEMP_2093] Found duplicate field(s) "x": [[x, X]] in case-insensitive mode. That is a different error class raised from a new place, the adapter used to raise it from rewrite and only for referenced columns. Could you thread case_sensitive through the helper and cover it?

While I was in there I noticed the check covers more than the tests show. A duplicate id on root fields is caught, which closes the same opener-skip hole for remap_physical_schema's own root check, and so is one inside a list<struct<...>> element. Both work today. Two more cases reusing write_bare_parquet would pin them, and the root one seems worth a line in the description because it widens what the PR fixes.

On the JNI change, spark_error_json_in_chain runs ahead of try_classify_file_read_error for every non-External CometError::DataFusion, so any typed SparkError reachable through a Context or Shared wrapper anywhere in the plan now changes exception class, not only one raised inside the parquet reader. I think that is the right behaviour. I checked the case I expected to be at risk, an ANSI failure in a pushed-down filter, and it is safe, because DataFusion stringifies that one into an ArrowError::ComputeError and nothing typed survives the chain. Worth mentioning in the description, since the change is wider than the field id path that motivates it.

Last, a process question rather than a code one. This closes an issue filed against #5654, which is still open, and the commit cannot be rebased onto main alone because validate_field_mapping needs resolve_field_mapping. Is there a reason not to fold it into #5654? That would leave one reviewable change instead of two, and this cannot merge before #5654 lands anyway.

[andygrove]

parking_lot::Mutex is already a dependency of native/core and is what every other mutex in the crate uses, in fair_pool.rs, task_shared.rs and jni_api.rs. Switching validated to it would let the lock() helper and the MutexGuard and PoisonError imports go away. PoisonError::into_inner is currently the only occurrence in the whole native tree.

[andygrove]

This makes four copies of throw_new("org/apache/comet/exceptions/CometQueryExecutionException", json_message) in this file, at 543, 549, 578 and 667. Would you pull out a throw_spark_error_json(env, json_message) and have throw_spark_error_as_json call it too? The External arm just above also hand-rolls the same two downcasts spark_error_json_in_chain does, only without the chain walk, so pointing it at the new function would remove that duplication and stop the two arms disagreeing about how deep they look.

[andygrove]

The reason the opener skips the adapter is spelled out here, on FieldIdCheck, and again in init_datasource_exec. Keeping it once on FieldIdCheck and having the other two point at it would mean it only has to be corrected in one place if DataFusion changes that condition.