feat: support _metadata constant columns in native Parquet scan

[mbutrovich]

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. _metadata.row_index). Removed the now-redundant fileConstantMetadataColumns.nonEmpty fallback check below it, since the narrowed gate already covers that case. Kept the _tmp_metadata_row_index schema check: that column is Spark's internal magic-named temporary column, not an isMetadataCol-flagged attribute, so the metadata-column gate never sees it and the scan needs its own check.
• 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.
• ShimFileFormat.scala (per Spark version): added fileConstantMetadataExtractors / getFileConstantMetadataColumnValue. Spark 3.4 predates per-format metadata extractors (SPARK-43868), so its shim replicates FileFormat.BASE_METADATA_EXTRACTORS inline instead of delegating. Spark 4.2 backported SPARK-56931, which added a required dataType argument to getFileConstantMetadataColumnValue; the shim signature is unified to 4 args across all five supported versions so the call site stays version-independent, with only the 4.2 shim using the extra argument. This required splitting the previously-shared spark-4.x/ShimFileFormat.scala into per-minor-version copies, matching how other Spark 4.x shims in this repo already handle behavior that differs by minor version.
• 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.
• CometNativeScan.scala: constant metadata columns are wired under a _comet_metadata_ prefix rather than their bare Spark name. DataFusion's table_partition_cols literal substitution matches by name, not position, so a bare name like file_size could collide with a real data or Hive partition column of the same name and get silently substituted. Spark itself has no such hazard (metadata attributes always resolve by ExprId/ordinal, never by name), so the prefix reproduces that guarantee instead of deviating from it.

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.

Three more tests cover review feedback: file_path/file_name are correctly url-encoded on Spark 3.4 for a directory containing a space (that logic is hand-replicated on 3.4 and untested elsewhere in PR CI), and two collision reproductions (a data column and, separately, a partition column both named file_size) confirming the rename fix above. A CometScanRuleSuite test also pins that _metadata (whole struct) and _metadata.row_index still fall back to Spark.

[andygrove]

Thanks for this, and for the unusually detailed PR description. The insight that the six file-constant columns are exactly shaped like Hive partition columns, and so can ride the existing table_partition_cols machinery with no proto or DataFusion changes, is a nice one.

I built the branch locally and probed a number of query shapes beyond the seven new tests. Most of what I tried behaves correctly:

• The schema-ordering assumption holds. FileSourceStrategy builds outputAttributes as readDataColumns ++ generatedMetadataColumns ++ partitionColumns ++ constantMetadataColumns, and partitionColumns there is the full partition schema rather than a pruned subset. So filter_schema = required_schema ++ partition_schema in planner.rs really does line up with Scala's exprToProto(filter, scan.output) numbering, as long as generatedMetadataColumns is empty. The narrowed gate guarantees that, and the _tmp_metadata_row_index check is a second layer.
• Non-UTC session timezone with file_modification_time matches Spark.
• spark.comet.parquet.rowFilterPushdown.enabled=true with a _metadata.file_size predicate matches Spark.
• A metadata-only projection (empty requiredSchema, e.g. SELECT _metadata.file_name FROM t) runs natively and returns correct values.
• A directory with a space in it produces the correct URL-encoded file_path and file_name on Spark 4.1.
• Selecting the whole _metadata struct still falls back to Spark, since row_index is in the pruned struct.

I did find one correctness problem that I think needs addressing before merge, plus two coverage gaps.

1. Metadata column names can collide with real column names, giving wrong results

Reusing table_partition_cols also opts into DataFusion's partition-value literal substitution. In ParquetOpener::prepare, DataFusion builds a literal_columns: HashMap<String, ScalarValue> from table_partition_cols zipped with partition_values, then runs replace_columns_with_literals over both the projection and the predicate. That helper matches purely on Column::name() and ignores the index (datafusion-physical-expr-adapter/src/schema_rewriter.rs).

Hive partition column names can never collide with the data schema, so this has always been safe. But file_size, file_path, file_name, file_block_start, file_block_length and file_modification_time are ordinary identifiers a user's table can legitimately contain, and they now land in table_partition_cols.

Three failures I reproduced on this branch (Spark 4.1, ParquetReadV1Suite):

Silent wrong results in a plain projection.

Seq((1L, 10), (2L, 20)).toDF("file_size", "x").repartition(1).write.parquet(dir)
spark.read.parquet(dir).select($"file_size", $"x", $"_metadata.file_size".as("meta_size"))

Spark: [1,10,719], [2,20,719]. Comet: [719,10,719], [719,20,719]. The real data column is silently overwritten with the file's byte size.

Silent wrong results in a filter. Same table, add .filter($"file_size" === 1L). Spark returns 1 row, Comet returns 0. The data column reference becomes the literal 719, so the predicate prunes the whole file.

Hard failure when a Hive partition column carries the name.

Seq((1, 100L), (2, 200L)).toDF("id", "file_size").repartition(1)
  .write.partitionBy("file_size").parquet(dir)
spark.read.parquet(dir).select($"id", $"file_size", $"_metadata.file_size")

fails with CometNativeException: Invalid argument error: column types must match schema types, expected Int32 but found Int64 at column index 1, because the Int32 partition column reference gets replaced by the Int64 metadata literal.

There are two ways out. The cheap one is a collision guard in CometScanRule.scala:151: only accept a constant metadata column when its name does not collide with relation.dataSchema or relation.partitionSchema, respecting spark.sql.caseSensitive, and fall back otherwise. The better one is to rename the fields on the wire, say _comet_metadata_file_size, in both CometNativeScan.scala:176 and operator/package.scala:77, so a collision is structurally impossible and native execution is retained. Everything downstream is positional (the projection vector and the Bound indices), so the rename should be contained.

Either way it would be good to have tests in both directions: a data column named file_size, and a partition column named file_size.

2. The Spark 3.4 shim is hand-replicated and has the thinnest CI coverage

spark/src/main/spark-3.4/org/apache/comet/shims/ShimFileFormat.scala:39 replicates Spark 3.5's BASE_METADATA_EXTRACTORS rather than Spark 3.4's own FileFormat.updateMetadataInternalRow. Those are close but not textually identical: 3.4 derives file_name from pf.filePath.toPath.toUri.getRawPath, because FileScanRDD passes currentFile.toPath (a Hadoop Path), whereas the shim uses pf.filePath.toUri.getRawPath (SparkPath's own URI, one fewer round trip through Path).

I expect those agree for ordinary paths, and the new tests do pass on 3.4. But every temp dir in them is free of characters that would exercise the difference, and the job that would catch it (Spark's metadata file path is url encoded and metadata file name is url encoded tests in FileMetadataStructSuite) lives in Spark SQL Tests (Spark 3.4), which does not run in PR CI.

Could you add a ParquetReadSuite test that writes into a directory with a space in the name and projects both file_path and file_name? That suite runs on all five versions in PR CI, so it would cover the one version where this logic is hand-written. I confirmed such a test passes on 4.1, so this is about pinning 3.4 down rather than a known break.

3. Nothing covers the fallback branch of the narrowed gate

The change narrows a fallback gate in CometScanRule.scala:154 and removes another one entirely, but none of the seven new tests assert that the remaining unsupported shape still falls back. Selecting the whole _metadata struct, or _metadata.row_index directly, is what has to keep going through Spark. I checked by hand that it does today. Since the gate is now a name-set subtraction rather than an unconditional check, could we pin that with a test?

[comphead]

its not a problem of this specific PR, but prob we can fix the error message to

Native Parquet Scan doesn't support functions `input_file_name`, .....

[mbutrovich]

Do we use markdown semantics in error messages?

[mbutrovich]

I did a quick pass, and I don't see us ever using markdown formatting in errors.

[mbutrovich]

Thanks @andygrove! I think I addressed all of the feedback in the latest commit.

[andygrove]

Thanks for addressing my feedback @mbutrovich. Could you take a look at the test failure in CI?

- CometScanRule should fallback to Spark for unsupported _metadata columns *** FAILED *** (206 milliseconds)
    org.apache.spark.sql.catalyst.ExtendedAnalysisException: [UNRESOLVED_COLUMN.WITH_SUGGESTION] A column, variable, or function parameter with name `_metadata` cannot be resolved. Did you mean one of the following? [`id`, `name`]. SQLSTATE: 42703; line 1 pos 11;
  'Project [id#257713, '_metadata]
  +- SubqueryAlias test_data
     +- View (`test_data`, [id#257713, name#257714])
        +- Relation [id#257713,name#257714] parquet

[mbutrovich]

Thanks for addressing my feedback @mbutrovich. Could you take a look at the test failure in CI?

- CometScanRule should fallback to Spark for unsupported _metadata columns *** FAILED *** (206 milliseconds)
    org.apache.spark.sql.catalyst.ExtendedAnalysisException: [UNRESOLVED_COLUMN.WITH_SUGGESTION] A column, variable, or function parameter with name `_metadata` cannot be resolved. Did you mean one of the following? [`id`, `name`]. SQLSTATE: 42703; line 1 pos 11;
  'Project [id#257713, '_metadata]
  +- SubqueryAlias test_data
     +- View (`test_data`, [id#257713, name#257714])
        +- Relation [id#257713,name#257714] parquet

Thanks @andygrove! Should be fixed now.

[andygrove]

LGTM pending CI. Thanks @mbutrovich