Comet native Parquet scan reads full nested columns despite pruned Spark ReadSchema

[mixermt]

Describe the bug

Comet's native Parquet scan appears to ignore nested schema pruning for complex columns. In a production Spark 3.5.6 comparison job, plain Spark and Comet produced the same Spark physical-plan ReadSchema, same input row count, same selected files, and same downstream shuffle output, but Comet read/decompressed far more data during the scan.

Spark 3.5.6, Comet 0.17.0

The query reads Parquet from HDFS and writes Parquet back to HDFS. Column names below are anonymized. The relevant scan reads four top-level fields:

is_flagged
source_type
events
event_hour

The Spark UI shows the same pruned nested schema under events for both plans:

ReadSchema: struct<
  is_flagged:boolean,
  source_type:string,
  events:array<struct<
    is_available:boolean,
    event_time_ms:bigint,
    is_active:boolean,
    items:array<struct<
      group_id:bigint,
      entity_id:bigint,
      has_amount:boolean,
      item_type:string,
      metric_value:double,
      actor_id:bigint,
      is_skipped:boolean
    >>
  >>,
  event_hour:timestamp
>

The physical input schema is much wider than the requested schema. A representative anonymized shape is:

InputSchema: struct<
  is_flagged:boolean,
  source_type:string,
  event_hour:timestamp,
  dimension_id:bigint,
  region_code:string,
  client_type:string,
  ingestion_time:timestamp,
  events:array<struct<
    is_available:boolean,
    event_time_ms:bigint,
    is_active:boolean,
    event_token:string,
    source_name:string,
    secondary_id:bigint,
    debug_flags:array<string>,
    latency_parts:struct<
      queue_time_ms:bigint,
      fetch_time_ms:bigint,
      render_time_ms:bigint,
      retry_count:int
    >,
    items:array<struct<
      group_id:bigint,
      entity_id:bigint,
      has_amount:boolean,
      item_type:string,
      metric_value:double,
      actor_id:bigint,
      is_skipped:boolean,
      auxiliary_id:bigint,
      class_code:string,
      mode:string,
      raw_score:double,
      normalized_score:double,
      reason_codes:array<string>,
      audit_blob:binary,
      feature_map:map<string,double>,
      diagnostics:struct<
        module_id:string,
        module_version:string,
        trace_id:string,
        extra_payload:binary
      >
    >>
  >>,
  raw_payload:binary,
  metadata_json:string
>

The expected behavior is to read only the requested nested leaves, not all child fields under events / items.

However, the observed scan metrics were very different:

Metric	Plain Spark scan	Comet native scan
Spark version	3.5.6	3.5.6
SQL duration	120,642 ms	1,688,780 ms
Scan tasks	20,012	20,012
Files selected	10,300	10,300
Input records	310,897,758	310,897,758
Stage input bytes	30,945,027,031	1,351,791,550,588
Shuffle records written	157,270,190	157,270,190
Shuffle bytes written	3,216,674,929	3,216,674,929
Comet bytes_scanned metric	n/a	1259.0 GiB

size of files read = 2.2 TiB appeared in both UIs, but that is the selected file length metric, not physical bytes fetched. The physical read evidence is Spark stage inputBytes and Comet's native bytes_scanned metric.

This strongly suggests the logical Spark-side pruning is present, but Comet's native reader still reads a much wider nested Parquet schema for the projected complex column.

Steps to reproduce

1. Use Spark 3.5.x with Comet native Parquet scans enabled.
2. Read a Parquet dataset with a wide nested column, for example an array<struct<...>> with many child fields.
3. Run a query that only needs a subset of nested fields from that complex column.
4. Compare against plain Spark using the same query and selected files.
5. In Spark UI, verify both plans show the same pruned ReadSchema.
6. Compare stage input bytes and Comet native scan Number of bytes scanned.

Minimal shape of the reproducer:

val df = spark.read.schema(prunedSchema).parquet(path)

df
  .filter(!$"is_flagged")
  .selectExpr(
    "event_hour",
    "posexplode_outer(events) as (event_idx, event)"
  )
  .filter("event.is_available and (event.is_active is null or event.is_active)")
  .selectExpr(
    "event_hour",
    "posexplode_outer(event.items) as (item_idx, item)",
    "event.event_time_ms"
  )
  .filter("item.entity_id is not null")
  .groupBy(
    $"event_hour",
    $"item.actor_id",
    $"item.group_id",
    $"item.entity_id"
  )
  .count()
  .write.mode("overwrite").parquet(outputPath)

The important part is that the query projects only selected child fields from a nested Parquet column while the original physical Parquet files contain many more children under that same top-level column.

Expected behavior

Comet native Parquet scan should honor Spark's pruned nested ReadSchema and avoid reading/decompressing unrequested nested child fields, matching Spark's Parquet scan behavior as closely as possible.

For this workload, Comet should not read ~1.35 TB of stage input when plain Spark reads ~30.9 GB for the same files, same rows, and same displayed ReadSchema.

Additional context

Local code inspection points to the native V1 Parquet scan path using full data_schema plus a top-level projection vector, instead of giving DataFusion a nested-pruned Parquet read schema.

Relevant code path:

• Spark writes the pruned requiredSchema into Parquet read config:
  • sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
  • ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA
• Spark clips nested Parquet schemas in:
  • sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala
  • clipParquetSchema, clipParquetType, clipParquetListType, clipParquetMapType
• Comet serializes both the pruned schema and the full data schema:
  • spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala
  • requiredSchema = schema2Proto(scan.requiredSchema.fields)
  • dataSchema = schema2Proto(scan.relation.dataSchema.fields)
  • projectionVector is built with scan.relation.dataSchema.fieldIndex(field.name), so it is top-level only.
• Native planner passes both schemas into the native scan:
  • native/core/src/execution/planner.rs
  • init_datasource_exec(required_schema, Some(data_schema), ..., Some(projection_vector), ...)
• Native Parquet setup chooses the full data_schema as the base schema when present:
  • native/core/src/parquet/parquet_exec.rs
  • (Some(schema), Some(proj)) => (Arc::clone(schema), Some(proj.clone()))
  • then applies with_projection_indices(Some(projection))

That can prune top-level columns, but it does not appear to prune nested leaves inside a projected complex column such as events.

Possible fixes / workarounds:

1. Workaround: set spark.comet.scan.enabled=false for this workload. That keeps Spark's Parquet scan path and avoids the native scan behavior.
2. Optional workaround to test: combine spark.comet.scan.enabled=false with spark.comet.convert.parquet.enabled=true, so Spark performs the Parquet read and Comet can still convert the resulting batches after the scan.
3. Code fix direction: for native V1 Parquet scan, initialize DataFusion's Parquet reader with the nested-pruned required_schema when Spark has already produced the read schema, instead of using full data_schema with only a top-level projection vector.
4. Add a regression test with array<struct<...many fields...>>, select only a subset of nested fields, verify CometNativeScanExec is used, and assert the native scan does not read unrequested nested leaves.

[mixermt]

Planing to work on the fix and verify similar for Iceberg table highly-nested input

[sandugood]

Could you, potentially, test it against the latest main branch?

[mixermt]

Could you, potentially, test it against the latest main branch?

Not sure I'll have the capacity to test against the main branch, but the fix I prepare with LLMs clearly points out that the issue is present under the main branch.

[mixermt]

Some observations after LLM iterations
Generated by LLM, reviewed by human

I tested the fix direction suggested here (option 3: pass the nested-pruned required_schema to DataFusion as the scan's file schema, instead of full data_schema + top-level projection vector) and measured native bytes_scanned directly. Results with DataFusion 54, a ~2 MB file of 2000 rows with a ~1 KB payload leaf next to a small id leaf:

shape	bytes_scanned, pruned schema	bytes_scanned, full schema
event: struct<id,payload> → struct<id>	2,045,604	2,045,604
events: array<struct<id,payload>> → array<struct<id>>	2,045,618	2,045,618
top-level control: (id, payload) → id	18,794	2,045,597

The pruned read schema fixes the scan's output: batches now carry exactly Spark's pruned ReadSchema (this also removes a latent wrong-results hazard, since downstream expressions bind struct fields by ordinal against the pruned schema, which was only accidentally correct when the pruned fields were a positional prefix of the file struct, as they are in this report). But it does not reduce Parquet I/O. Unrequested nested leaves are still fetched and decoded, then dropped in memory by the schema adapter.

Root cause: DataFusion's ParquetOpener derives the leaf-level ProjectionMask from the projection and filter expressions (build_projection_read_plan in datafusion-datasource-parquet/src/row_filter.rs), and only get_field(column, ...) chains produce leaf masks. A pruned-struct-vs-full-struct schema difference is instead handled by the PhysicalExprAdapter as a whole-column cast, which the pushdown checker cannot see through, so it falls back to ProjectionMask::roots and reads the entire column chunk.

The reader is not the limitation. parquet-rs ProjectionMask::leaves handles nested pruning, and DataFusion already uses it for pushed-down struct filters. The gap is purely that nothing translates a clipped logical file schema into a leaf mask. So fully fixing this issue needs one of:

1. Upstream DataFusion: derive ProjectionMask::leaves from the logical-vs-physical file schema diff (the equivalent of Spark's clipParquetSchema). Benefits all DataFusion users.
2. Comet-side: a custom opener/FileSource that computes the leaf mask from required_schema against the file's SchemaDescriptor by path matching. No upstream dependency, more maintenance.
3. Partial only: DF 54's FileSource::try_pushdown_projection accepts expression projections, so plain struct roots could be rewritten as struct(get_field(col,'x'), ...). But there is no expression form for pruning through array<struct> or map values, which is exactly the shape in this report, so this cannot fix it alone.

Passing the pruned schema is still worth doing for the schema-correctness fix, just with the expectation set that stage input bytes will not improve until the mask-level work lands.

@andygrove @mbutrovich any guide from you will be appreciated.

[mbutrovich]

I seem to remember when we first implemented the (at the time) ParquetExec (now DataSourceExec) scan, we couldn’t push projections into nested columns without scanning the whole parent column. It sounds like we might be able to revisit that. I’ll take a look. Thanks for reporting @mixermt!

[adriangb]

I've spent a bit exploring this issue with Claude and these are the conclusions we came to. I reviewed them and they make sense to me, I'm not sure what the path forward is though. Keep in mind I am not familiar with Spark or Comet.

This largely confirms @mixermt's analysis and adds some detail on why the gap exists and why the obvious workarounds don't apply.

Why nested pruning works in plain DataFusion but not here

DataFusion can prune nested leaves — SELECT s.foo FROM t in datafusion-cli only reads the s.foo leaf. But that works because the pruning is carried in the expression: the projection is get_field(Column("s"), "foo"), and build_projection_read_plan in datafusion-datasource-parquet/src/row_filter.rs translates literal get_field chains into a ProjectionMask::leaves.

Spark communicates nested pruning completely differently. The SchemaPruning optimizer rule consumes the GetStructField expressions and bakes the result into the scan node as a narrower requiredSchema (a Column("events") whose type is a narrower array<struct<...>> than the file's); the remaining downstream field accesses are rebased against the pruned scan output. So at the scan boundary the pruning exists only as a type, not as expressions — there is nothing for Comet to translate into get_field.

What DataFusion does with a pruned schema today (DF 54, still true on current main)

When the logical file schema says events: array<struct<3 fields>> but the physical file has array<struct<18 fields>>:

1. The DefaultPhysicalExprAdapter rewrites the projection against the physical file schema, turning the type mismatch into a whole-column CastExpr(Column("events"), narrow_type) (datafusion-physical-expr-adapter/src/schema_rewriter.rs).
2. build_projection_read_plan derives the projection mask from the rewritten expressions. Its only sources of leaf-level masks are literal get_field chains (and struct-field filter accesses). A CastExpr(Column) falls through to the generic path, which records the root column and expands it to every physical leaf under events.
3. The full column chunks are fetched and decoded; the cast drops the unwanted subfields in memory. This is exactly why @mixermt measured identical bytes_scanned with the pruned schema — the output is fixed, the I/O is not.

The parquet reader itself is not the limitation: parquet-rs ProjectionMask::leaves handles arbitrary nesting, and DataFusion already uses it for pushed-down struct filters. The gap is purely that nothing translates a schema-level diff (logical file schema narrower than physical) into a leaf mask.

Why Comet can't just re-synthesize get_field expressions

Even setting aside the array<struct> problem, translating the pruned schema back into DataFusion expressions is lossy:

• Placement: the leaf-mask derivation only looks at expressions inside the DataSourceExec's own projection. Comet builds the scan with an index projection and keeps its projection logic in a separate ProjectionExec using its own ordinal-based GetStructField expression, which DataFusion's PushdownChecker can't recognize (it downcasts specifically to GetFieldFunc). Comet would have to push DF-native get_field exprs into the scan via FileSource::try_pushdown_projection.
• Ordinal vs. name resolution: Spark resolves struct fields by ordinal and Parquet structs may contain duplicate or case-colliding field names; get_field is name-based. (This appears to be why Comet's GetStructField is ordinal-based in the first place.)
• Struct-level nullability: the scan must still emit the struct column, so the expression encoding is named_struct('foo', get_field(s, 'foo')) — but if s is NULL that yields a non-null {foo: NULL} instead of NULL, silently changing s IS NULL semantics. A clipped-schema read preserves struct nullness via the surviving leaves' definition levels.
• It stops at structs: there is no expression form at all for pruning subfields of array<struct> elements or map values — which is the shape in this report. (Element access like events[1].foo is orthogonal: pruning list<struct<foo>> means "only foo, for every element"; Parquet shreds nested data into one column chunk per leaf, so leaf selection is the only prunable axis at the schema level.)

What seems needed upstream in DataFusion

Schema-driven nested pruning in the parquet opener: when a required root column's logical file-schema type is a structural subset of its physical type, compute the leaf mask by walking the two type trees and matching fields by name — the equivalent of Spark's ParquetReadSupport.clipParquetSchema — instead of expanding the root to all leaves. After clipping, the decoder emits the narrow type directly and the adapter-inserted cast degrades to a cheap adjustment, while still handling schema evolution (missing subfield → null fill, reordering, promotion). Edge cases to spec: map key/value handling, case sensitivity, and field-id-based matching for Iceberg (possibly making the clipping policy pluggable, like the expr adapter factory already is).

This can be reproduced and tested in pure DataFusion with no Comet involved: write a parquet file with a wide struct (or array<struct>) column, CREATE EXTERNAL TABLE over it declaring a narrower nested type, select that column, and observe bytes_scanned — the full column chunk is read despite the narrow table schema. That isolates the schema-diff path from the get_field path that already works.

An interim alternative would be an API for ParquetSource callers to supply the leaf-level projection directly (per-file ProjectionMask hook or an authoritative "read schema"), which Comet could drive itself — but the diff-based approach would benefit every embedder that hands DataFusion a pre-pruned schema (delta-rs, Iceberg integrations, etc.).

Happy to help move the DataFusion side forward if there's agreement on the direction.

[adriangb]

So in conclusion: I think we could specialize struct adaptation to physical files by reading only the fields the logical schema needs instead of reading the whole thing and then dropping them.

cc @kosiew since you've work on struct <-> struct casting and IO optimization.