fix: match container field ids that the INT96 coercion drops

[dwsmith1983]

Which issue does this PR close?

Closes #6131.

Rationale for this change

With spark.sql.parquet.fieldId.read.enabled=true, the native scan matches requested fields to file fields by the PARQUET:field_id on the Arrow schema DataFusion hands the schema adapter. When a file holds an INT96 column, DataFusion 55.1.0's INT96 coercion rebuilds every struct, list and map field of that schema without its metadata, so an id that sits only on a container is gone and the container is null filled. Spark's ParquetReadSupport.clipParquetGroupFields matches the id on the raw Parquet group and reads it. Spark writes timestamps as INT96 by default, so any id-bearing container next to a timestamp column was affected.

The upstream fix, apache/datafusion#24790, keeps the metadata but is not in 55.1.0. This is a workaround for the time until Comet moves to a DataFusion release that carries it.

What changes are included in this PR?

• EagerPageIndexReader::get_metadata records the name and id of every field from the file's own Parquet schema under a comet.parquet.field_ids key in the footer it returns, only for scans that match by field id, only when the file has an INT96 leaf and a struct, list or map carries an id, and only in the metadata returned for that open. The shared metadata cache keeps the footer as written. A key of that name already in the file is removed so the stamp is always the one computed from the Parquet schema.
• SparkPhysicalExprAdapterFactory::create puts the dropped container ids back on the physical schema before any id matching runs. It checks the field count, every name and every id the schema still carries and leaves the schema unchanged on any mismatch.
• CometCastColumnExpr relabels a decoded array to the restored physical type, so the scan's batches match its schema and nested conversions resolve fields by those ids. Columns whose type gained nested ids but need no other conversion get wrapped for the same reason.
• A rebuilt footer loses the decryptor its column chunks need, so a scan that matches field ids while Parquet encryption is configured falls back to Spark at planning time, with the reason in the plan, as native Variant scans already do. The planner cannot see whether a file has INT96 or container ids, so every such scan falls back. The native reader still refuses such a file with an error naming it, as a backstop. The user guide lists the fallback.
• The Variant footer rewrite in the reader factory is split into computing the key-value list and rebuilding the metadata, so both rewrites share one rebuild.

The new field_id_stamp module documents that it should be deleted once DataFusion carries apache/datafusion#24790.

How are these changes tested?

• Unit tests in field_id_stamp.rs cover the stamp contents for structs, lists, maps and a legacy repeated primitive, the cases where no stamp is needed, restoring ids on a coerced schema, every mismatch shape returning the schema unchanged, and replacing a stamp key the file already carries. One test drives DataFusion's Int96Coercer directly and asserts that it drops container ids. It fails once a DataFusion upgrade keeps them, which is the signal to remove the workaround.
• Reader factory tests check that only a field-id scan gets the stamp, that the cached footer stays as written, and that the two encrypted cases are refused with the documented message while a scan without field id matching still reads the file.
• CometScanRuleSuite checks that an encrypted scan matching field ids falls back to Spark with the reason, and stays native when field id reads are off or the read schema carries no ids.
• Scan tests in parquet_exec.rs read files with an id on a root struct, a renamed nested struct and a legacy repeated primitive next to INT96 and check the values.
• ParquetReadSuite writes struct, list and map columns holding INT96 timestamps with ids on the containers only, on containers and children, at several depths and with containers renamed in the read schema, plus a timestamp outside the nanosecond range and a TIMESTAMP_MICROS control, and compares each column with Spark. The renamed case is compared with Spark from 4.1 on, because the vectorized reader in 3.5 and 4.0 rejects a renamed nested struct in ParquetColumnVector. The pinned rows are checked on every version. These suites pass on Spark 3.5, 4.0 and 4.1.

[andygrove]

apache/datafusion#24790 is three .with_metadata(...) calls, and you've already proposed it for 55.2.0 on apache/datafusion#25758. If it lands there, main and branch-1.1 only need the version bump. We'd avoid the footer rewrite, the reserved comet.parquet.field_ids key and the new fallback that sends every encrypted field-id scan to Spark, including files that never needed the stamp. Could we open the branch-55 backport first and hold this PR until we know whether 55.2.0 carries it? If we want main correct before then, would a CometScanRule fallback for scans that request an id on a struct, array or map field be enough as a stopgap? It's a few lines and comes out with the bump.

[andygrove]

This list is what someone will follow when a DataFusion upgrade makes this test fail. It leaves out the encryption fallback in CometScanRule, its test in CometScanRuleSuite, the entry in scans.md and the with_use_field_id wiring in parquet_exec.rs. If those stay behind, encrypted field-id scans keep falling back for no reason. Since this PR closes #6131, the code comments and the user guide also end up pointing at a closed issue. Could you open an issue for removing the workaround once DataFusion carries apache/datafusion#24790, list every site there, and point these references at it?

[dwsmith1983]

If 55.2.0 carries apache/datafusion#24790, this file goes away with the version bump. If it doesn't, I'll open the removal issue listing the encryption fallback in CometScanRule, its CometScanRuleSuite test, the scans.md entry and the with_use_field_id wiring in parquet_exec.rs, and point the code comments and the user guide at it instead of #6131.

[andygrove]

I ran this test with the gate replaced by true on 3.4.3, 3.5.9 and 4.0.4, and it passes, so Spark reads the renamed INT96 case on those versions too. Could we drop the gate so every version compares with Spark, and trim the comment above to match? I did see Spark 3.4 reject a struct renamed under a list or map when the file was TIMESTAMP_MICROS, which may be what you ran into, but this case only writes INT96.

[dwsmith1983]

Thanks for running it on all three. I'll drop the gate and trim the comment. The tests stay in this PR whichever way 55.2.0 goes.

[andygrove]

Every case here selects a whole root column, so only the projection path is covered. On main, a filter through one of these containers drops rows. where s.inner.x = 1 and where s is not null both return nothing, and count(s) is 0. select s.inner.x comes back null through the pruned cast. All of these match Spark with this PR. Could you add a filter and a pruned nested projection so the predicate rewrite is pinned too?

[dwsmith1983]

I'll add where s.inner.x = 1, where s is not null, count(s) and select s.inner.x, so the predicate rewrite and the pruned cast are pinned alongside the projection.

[dwsmith1983]

Could we open the branch-55 backport first and hold this PR until we know whether 55.2.0 carries it?

Yes. The backport is apache/datafusion#25895, a clean cherry-pick of #24790 onto branch-55, linked from #25758. I'll hold this PR until we know whether 55.2.0 carries it, and if it does, this PR shrinks to the tests after the version bump.

would a CometScanRule fallback for scans that request an id on a struct, array or map field be enough as a stopgap?

I think so, and it's up as #6444: when field id reads are on and the requested schema has an id on a struct, array or map field, the scan falls back to Spark. Variant fields on Spark 4.x lose their id the same way, so they're included too. A native test fails once DataFusion keeps those ids, which flags when the fallback can come out.

[sunchao]

Summary

• Prior state and problem: DataFusion’s INT96 coercion drops container field IDs, causing valid Parquet structs, lists and maps to read as null under field-ID matching.
• Design approach: Carry IDs from the raw Parquet schema through a per-open footer stamp, then restore them before schema adaptation.
• Correctness / compatibility analysis: Compared Spark’s field-ID resolution across 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0, including missing IDs, renamed fields and nested containers. No introduced P1/P2 issues found within this review.
• Key design decisions: Preserve the shared metadata cache, reject mismatched stamps, and fall back to Spark when encryption and field-ID matching are enabled. The workaround spans several layers, as the existing upstream-backport discussion already addresses.
• Implementation sketch: with_field_ids_stamp records IDs, restore_field_ids restores container metadata, and CometCastColumnExpr relabels decoded arrays. Stamping adds per-open schema traversal. Relabeling shares underlying buffers. No performance benchmark was run.
• Behavioral changes worth calling out: Affected containers resolve by ID instead of becoming null. Encrypted field-ID scans fall back even when their files would not require restoration, as documented.
• Suggested improvements: No additional P1/P2 requests. Existing discussions about the upstream fix, workaround removal and regression coverage remain open and are not duplicated here.

Reviewed the full 10-file diff from 1628c520096e12ba3fb13988d6ee235070e77e07 to 3763f40e3a56d47d2988c0968a6ce693aa8dbc43. The PR remains non-draft. Applied review-comet-pr and review-comet-expression-pr for the native cast changes. Read all supplied reviews, comments and threads.

Exact-head CI: labeling passed. Comet CI and CodeQL report action_required, so neither provides a passing validation verdict.

Validation: native Parquet tests passed with 243 passed and one ignored. Four disposable probes also passed, covering reordered IDs with distinct values, pruning, missing IDs, null parents, predicates and nested lists/maps. All disposable source changes were removed. JVM compilation and Spark suites could not run: required Maven dependencies were unavailable locally, and Maven Central access was blocked by network policy. No exact-head Spark SQL suite verdict was available.

[dwsmith1983]

@andygrove @sunchao
The backport merged as apache/datafusion#25895, so 55.2.0 will carry the fix (RC planned this weekend per apache/datafusion#25758). Closing this in favor of picking up 55.2.0; #6444 covers the gap until then.