Skip to content

feat: support _metadata constant columns in native Parquet scan - #5237

Draft
mbutrovich wants to merge 6 commits into
apache:mainfrom
mbutrovich:parquet_metadata_columns
Draft

feat: support _metadata constant columns in native Parquet scan#5237
mbutrovich wants to merge 6 commits into
apache:mainfrom
mbutrovich:parquet_metadata_columns

Conversation

@mbutrovich

@mbutrovich mbutrovich commented Aug 3, 2026

Copy link
Copy Markdown
Contributor

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?

Seven 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 with a range predicate (>), not just equality.
  • 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.
  • Projecting metadata columns together with a real Hive partition column, which exercises the schema ordering that appends metadata columns after partition columns (CometNativeScan.scala's partitionSchemaFields).
  • Filtering on a real Hive partition column and a metadata column in one predicate, which exercises column-index resolution across all three schema segments (data, partition, metadata) at once, in planner.rs's filter_schema.

The two-file tests (writeTwoFilesAndDiscoverMetadata) assert rdd.getNumPartitions == 1 to confirm both files land in the same Spark task, so partition2Proto's per-file loop is exercised with more than one file, not just across separate tasks. Getting two tiny files into one partition isn't the default: spark.sql.files.minPartitionNum defaults to the session's target parallelism, which pushes maxSplitBytes down to spark.sql.files.openCostInBytes, and a single file's own virtual open-cost surcharge already consumes that budget, so Spark schedules one file per task by default. These tests force spark.sql.files.minPartitionNum=1 to get both files packed into one partition instead.

…e, file_block_start, file_block_length, file_modification_time)
@mbutrovich mbutrovich self-assigned this Aug 3, 2026
@mbutrovich
mbutrovich marked this pull request as draft August 3, 2026 20:20
@mbutrovich mbutrovich added this to the 1.1.0 milestone Aug 3, 2026
@mbutrovich
mbutrovich marked this pull request as ready for review August 3, 2026 21:03
@mbutrovich
mbutrovich marked this pull request as draft August 4, 2026 00:21
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant