feat: support _metadata constant columns in native Parquet scan - #5237
Draft
mbutrovich wants to merge 6 commits into
Draft
feat: support _metadata constant columns in native Parquet scan#5237mbutrovich wants to merge 6 commits into
mbutrovich wants to merge 6 commits into
Conversation
…e, file_block_start, file_block_length, file_modification_time)
mbutrovich
marked this pull request as ready for review
August 3, 2026 21:03
mbutrovich
marked this pull request as draft
August 4, 2026 00:21
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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_metadatacolumns, which turned out to be unblocked already and unrelated to that dependency.Rationale for this change
Spark's
FileSourceScanExecexposes six file-source constant_metadatacolumns:file_path,file_name,file_size,file_block_start,file_block_length,file_modification_time. Unlikerow_index, all six are known before opening the file and are constant for every row read from it, exactly like Hive partition columns.CometScanRulewas falling back to Spark unconditionally whenever any of these appeared in a query, with no distinction fromrow_index. The value-delivery mechanism for this already exists for partition columns (DataFusion'stable_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 infileConstantMetadataColumns(i.e.row_index). Removed two fallback checks further down (fileConstantMetadataColumns.nonEmpty, and therow_indexschema 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 ownscan.outputordering (data columns, then partition columns, then constant metadata columns), and extends the projection vector and length assertion to match.operator/package.scala:partition2Prototakes the constant metadata attributes and the relation'sfileConstantMetadataExtractors, and derives each value via Spark's ownFileFormat.getFileConstantMetadataColumnValue(covers custom per-format extractor overrides), retyped against the attribute's declareddataTypethe same way Spark's ownupdateMetadataInternalRowdoes. Values are appended to the existingpartition_valuesproto field alongside real partition values.CometNativeScanExec.scala: threadsfileConstantMetadataColumnsandfileFormat.fileConstantMetadataExtractorsthrough topartition2Proto.contraintExpressions.scala/QueryPlanSerde.scala: added a serde for Spark'sKnownNotNulltagging expression. Spark'sFileSourceStrategywraps the reconstructed_metadatastruct inKnownNotNullto 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 existingCometKnownNullable; both now share one helper).native/core/src/execution/planner.rs:data_filterswere bound only againstrequired_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 combinedrequired_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 throughdataFilters).docs/.../compatibility/scans.md: narrowed the "Spark metadata columns" limitation entry to_metadata.row_indexonly.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):_metadata.file_sizealone. 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._metadata.file_sizewith a range predicate (>), not just equality._metadata.file_sizecombined with an ordinary data column in one predicate._metadata.file_pathfor exact string equality.file_pathis 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.CometNativeScan.scala'spartitionSchemaFields).planner.rs'sfilter_schema.The two-file tests (
writeTwoFilesAndDiscoverMetadata) assertrdd.getNumPartitions == 1to confirm both files land in the same Spark task, sopartition2Proto'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.minPartitionNumdefaults to the session's target parallelism, which pushesmaxSplitBytesdown tospark.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 forcespark.sql.files.minPartitionNum=1to get both files packed into one partition instead.