mixermt opened a new issue, #4859:
URL: https://github.com/apache/datafusion-comet/issues/4859
### 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:
```text
is_flagged
source_type
events
event_hour
```
The Spark UI shows the same pruned nested schema under `events` for both
plans:
```text
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:
```text
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:
```scala
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.
--
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]
