mbutrovich opened a new pull request, #5237: URL: https://github.com/apache/datafusion-comet/pull/5237
## Which issue does this PR close? Closes #. This is motivated by #3432 but does not close it. #3432 is specifically about `_metadata.row_index`, which is generated per row by the Parquet reader and needs DataFusion's virtual-column plumbing (DataFusion 55, not yet picked up by Comet). This PR adds the other six `_metadata` columns, which turned out to be unblocked already and unrelated to that dependency. ## Rationale for this change Spark's `FileSourceScanExec` exposes six file-source constant `_metadata` columns: `file_path`, `file_name`, `file_size`, `file_block_start`, `file_block_length`, `file_modification_time`. Unlike `row_index`, all six are known before opening the file and are constant for every row read from it, exactly like Hive partition columns. `CometScanRule` was falling back to Spark unconditionally whenever any of these appeared in a query, with no distinction from `row_index`. The value-delivery mechanism for this already exists for partition columns (DataFusion's `table_partition_cols` / `PartitionedFile.partition_values`, generic over field position), so these six columns can reuse it directly with no DataFusion or proto changes. ## What changes are included in this PR? - `CometScanRule.scala`: narrowed the metadata-column fallback gate to only reject columns that are not in `fileConstantMetadataColumns` (i.e. `row_index`). Removed two fallback checks further down (`fileConstantMetadataColumns.nonEmpty`, and the `row_index` schema check) that were unreachable once the earlier gate covers both cases. - `CometNativeScan.scala`: appends the constant metadata columns' schema after the real Hive partition schema, matching Spark's own `scan.output` ordering (data columns, then partition columns, then constant metadata columns), and extends the projection vector and length assertion to match. - `operator/package.scala`: `partition2Proto` takes the constant metadata attributes and the relation's `fileConstantMetadataExtractors`, and derives each value via Spark's own `FileFormat.getFileConstantMetadataColumnValue` (covers custom per-format extractor overrides), retyped against the attribute's declared `dataType` the same way Spark's own `updateMetadataInternalRow` does. Values are appended to the existing `partition_values` proto field alongside real partition values. - `CometNativeScanExec.scala`: threads `fileConstantMetadataColumns` and `fileFormat.fileConstantMetadataExtractors` through to `partition2Proto`. - `contraintExpressions.scala` / `QueryPlanSerde.scala`: added a serde for Spark's `KnownNotNull` tagging expression. Spark's `FileSourceStrategy` wraps the reconstructed `_metadata` struct in `KnownNotNull` to force non-nullability on the schema; it is a runtime no-op, so the serde serializes the child and drops the tag (same approach as the existing `CometKnownNullable`; both now share one helper). - `native/core/src/execution/planner.rs`: `data_filters` were bound only against `required_schema`, which excludes partition and metadata columns. A filter referencing one of these columns failed with a column-index-out-of-bounds error. Filters are now bound against the combined `required_schema` + `partition_schema`, matching how Scala numbers columns when building the filter proto. This also fixes filtering on real Hive partition columns pushed down as a data filter, which shared the same latent bug (previously never triggered, since Spark's planner never routes a pure partition-column predicate through `dataFilters`). - `docs/.../compatibility/scans.md`: narrowed the "Spark metadata columns" limitation entry to `_metadata.row_index` only. ## How are these changes tested? Four new tests in `ParquetReadSuite.scala`, each checking both correct results (against Comet-disabled Spark) and that the native scan actually runs (no fallback): - Projecting all six constant metadata columns. - Filtering on `_metadata.file_size` alone. A pure metadata-column predicate can never become a Spark partition-pruning filter, since these columns are not part of the Hive partition schema, so this exercises per-file value correctness during the scan itself, not just projection. - Filtering on `_metadata.file_size` combined with an ordinary data column in one predicate. - Filtering on `_metadata.file_path` for exact string equality. `file_path` is derived to match Spark's own qualified-URI string exactly, including its single-slash local-path form, so an equality filter is a direct check on that derivation. Not yet covered: a table with both real Hive partition columns and selected metadata columns together (the ordering logic that appends metadata columns after partition columns has not been exercised with a non-empty partition schema), and a metadata-column filter using a range predicate rather than equality. -- 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]
