Native scan path doesn't honour Parquet field-ID matching when spark.sql.parquet.fieldId.read.enabled=true

[schenksj]

Summary

Comet's native scan paths (SCAN_NATIVE_DATAFUSION and the new
SCAN_NATIVE_DELTA_COMPAT in the delta-kernel-phase-1 work) read parquet
columns by name. When the user enables Spark's parquet field-ID-based
column resolution via spark.sql.parquet.fieldId.read.enabled=true,
Spark's parquet reader matches columns by parquet.field.id metadata
on each StructField rather than by name. DataFusion's parquet path
does not honour that metadata, so columns are still resolved by name --
silently producing wrong results when names and IDs disagree.

Repro (Delta column-mapping id mode)

The Delta id column-mapping mode relies on field-ID matching to
decouple the table's logical column name from the parquet file's
physical name. Tests that exercise the rename-detection semantics
(e.g. DeltaColumnMappingSuite "column mapping batch scan should
detect physical name changes" and "explicit id matching") expect
nulls when a field's ID is changed in Delta metadata such that it
no longer matches the file's stored ID. Vanilla Spark + Delta returns
nulls; Comet returns the actual data because its by-name resolver
finds the column whose name didn't change.

Workaround

nativeDataFusionScan already declines when both
spark.sql.parquet.fieldId.read.enabled=true and the requiredSchema
has field-IDs (ParquetUtils.hasFieldIds). The same gate has now been
mirrored in nativeDeltaScan. However, the check returns false for
Delta because Delta's HadoopFsRelation strips the field-ID metadata
from requiredSchema -- the IDs live on the snapshot's metadata,
which the Comet rule doesn't consult. So the gate never fires for
Delta column-mapping id mode.

Proposed fix

Extend Comet's parquet-read path to honour parquet.field.id /
field_id Arrow metadata for column resolution when the session's
PARQUET_FIELD_ID_READ_ENABLED is true, mirroring Spark's
ParquetReadSupport.matchByName/matchByID selection. Track per-field
IDs on data_schema and pass them through to the native parquet
reader so the schema adapter prefers ID-match.

Filed against: branch delta-kernel-phase-1 (PR #3932)
Related Spark behavior: org.apache.spark.sql.execution.datasources.parquet.ParquetReadSupport

[mbutrovich]

Could you clarify if this is a native_datafusion issue right now with reading back Parquet files with field IDs? If so, is there a repro test you could provide?

[mbutrovich]

Investigating with Claude:

Thanks for the write-up. I traced the field-ID handling across our scan paths to make sure I understood the scope before responding.

Field-ID handling today

• Legacy V1 (CometParquetFileFormat): honors field IDs via CometParquetReadSupport (see useFieldId at lines 126, 203, 245, 351).
• SCAN_NATIVE_DATAFUSION: gate at CometScanRule.scala#L235-L239 falls back to Spark when spark.sql.parquet.fieldId.read.enabled=true and requiredSchema has IDs.
• SCAN_NATIVE_ICEBERG_COMPAT: also honors field IDs. The relation's fileFormat is swapped to CometParquetFileFormat in CometScanExec.scala#L558, and useFieldId is plumbed into NativeBatchReader at CometParquetFileFormat.scala#L97,L158. The reader rewrites the Spark schema by ID at NativeBatchReader.java#L398-L401 (getSparkSchemaByFieldId, derived from CometParquetReadSupport.matchFieldId).
• SCAN_NATIVE_DELTA_COMPAT (PR feat: Native Delta Lake scan via delta-kernel-rs #3932): this is the actual gap. The mirrored gate never fires because Delta strips field-ID metadata from requiredSchema before it reaches Comet.

Test coverage

No field-ID exclusions in dev/diffs/{3.4.3,3.5.8,4.0.2,4.1.1}.diff, so Spark's ParquetFieldIdIOSuite and ParquetFieldIdSchemaSuite run unmodified. They pass via the legacy V1 reader or the native_datafusion fallback. The Comet-side test ParquetReadSuite.scala#L1356 writes and reads with matching names, so it would pass even without by-ID matching and does not protect against the rename-detection scenario.

Suggested fix for SCAN_NATIVE_DATAFUSION

Rather than just plugging the Delta-specific gate, we could honor field IDs end-to-end in the native path and drop the fallback. arrow-rs only round-trips the metadata; it does not resolve columns by ID:

• PARQUET:field_id is copied from parquet→Arrow field metadata when reading the footer (parquet/src/arrow/schema/complex.rs#L766-L770, constant at mod.rs#L227) and read back when writing.
• ProjectionMask::leaves(parquet_schema, indices) projects by physical leaf index. ProjectionMask::columns(...) and ArrowReaderOptions::with_schema(...) are both name/position-based - no ID resolution.

iceberg-rust builds ID-based projection on top of those primitives in reader/projection.rs#L144-L213: walk the file's Arrow schema, filter leaves whose PARQUET:field_id metadata matches a requested ID, build a ProjectionMask::leaves(...) from the matched indices, and let a downstream transformer reorder/rename to the requested schema. Comet's native side already understands PARQUET:field_id in native/core/src/parquet/cast_column.rs#L572-L575, so the post-projection rename/cast layer largely works.

Concrete hookup:

1. JVM side: when serializing the scan, attach the Spark field IDs as proto metadata (we already pass parquet.field.id through for the Iceberg path).
2. Native side in native/core/src/parquet/parquet_exec.rs: when useFieldId is set and the required schema carries IDs, build the projection from the file's footer using leaf field IDs (mirroring iceberg-rust's get_arrow_projection_mask_with_field_ids) instead of by name.

Net: one new helper in the Rust scan path plus ID propagation in the proto. No JVM-side footer reads, no arrow-rs work, no per-file schema rewrites at planning time. This also fixes the Delta gap reported here as a side effect, since correctness no longer depends on the JVM-side gate seeing IDs on requiredSchema.