dwsmith1983 opened a new pull request, #6405: URL: https://github.com/apache/datafusion-comet/pull/6405
## 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
