comphead opened a new issue, #6724: URL: https://github.com/apache/datafusion-comet/issues/6724
### What is the problem the feature request solves? The native Iceberg scan reads and decodes every leaf of a projected struct, list, or map column, even when Spark's nested schema pruning keeps only some of its fields. Results are correct, because the batch adaptation after the read drops the extra fields. The cost is I/O and decode work for data the query never uses. **Cause.** For a task without deletes, `CometIcebergNativeScan.serializePartitions` sends iceberg-rust the full `metadata.tableSchema` as the `FileScanTask` schema, with top-level `project_field_ids`. iceberg-rust's `include_leaf_field_id` (`arrow/reader/projection.rs`) expands a struct, list, or map field id to every leaf it finds in that schema. The pruned read schema, `metadata.scanSchema` (`SparkScan.expectedSchema()`), is used only when `VERSION AS OF` reads a dropped column. A task with deletes uses the full `FileScanTask.schema()`. The code passed `scanSchema` until #2729. That PR switched to `tableSchema` because a `count(*)` scan schema lacked the partition source columns that `constants_map` needs. `schemaWithRequiredFields` now adds partition source columns explicitly, so that reason no longer applies. `CometIcebergNativeSuite` already notes the behavior: "The native read asks for the column's full nested type even when Spark prunes it." **Impact.** These numbers come from Spark event logs of TPC-H SF1000 with a nested schema on Iceberg (Spark 4.0, 32 executors x 16 cores). In that schema each table keeps its non-key columns in an `array<struct>` column, such as `partsupp_data`. - Q20 needs only `ps_availqty` from `partsupp_data`, and Spark prunes the struct to it. Comet's partsupp scan still reads 26.4 GiB, which is every byte of the table's files. Q16 reads only the top-level keys of the same table and scans 1.2 GiB. The partsupp scan is about 62% of Comet's Q20 time, and its task time is only 14% below Spark's. In that run Q20 was level with Spark (28.6 s vs 28.1 s), while Comet was faster on most other queries. - The gap matters only where Spark itself prunes a nested column. An `explode` adds a `size(col) > 0` filter on the whole column, which blocks Spark's pruning unless the explode was rewritten to a single field. Across the 22 queries, Spark prunes 12 of 66 scans. They carry 18% of Comet's scanned bytes. On them Comet's scan-stage task time is 0.53x Spark's, against 0.38x on scans where both engines read the whole column and 0.25x on flat scans. - Q2 scans `partsupp` twice, once pruned by Spark and once not. Spark spends 724 s of task time on the pruned scan and 1,309 s on the other. Comet spends about 695 s on each. - The affected scan is on the critical path of Q14 (`part`, 6.7 of 7.6 s), Q18 (`orders`), and Q20 (`partsupp`). ### Describe the potential solution Pass the pruned scan schema to iceberg-rust instead of the table schema. I have a working prototype: - The task schema is `metadata.scanSchema` without its metadata columns, for every task, including tasks with deletes. Metadata columns are field ids from `Integer.MAX_VALUE - 200` up, which a table schema never has. Partition source columns and equality-delete keys are still added through `schemaWithRequiredFields`. - `schemaWithRequiredFields` is memoized by base schema and field ids. Without that, a partition source the query does not project makes every task build its own schema, and the schema pool grows to one entry per task. - A task keeps the full schema when a missing partition source or equality-delete key is nested inside a struct the query pruned. That field would otherwise be appended at the top level. A table with a top-level `region` and an `s.region` partition source then fails with `Invalid schema: multiple fields for name region`. - A new config, `spark.comet.scan.icebergNative.nestedSchemaPruning.enabled` (default `true`), turns it off. The nested-evolution fallback from #6543 stays as it is. It checks the full column type in the table schema, so it still falls back when the evolved field is pruned away. A follow-up could check only the pruned fields and keep more queries native. **Prototype results.** - Four new tests pass. They cover struct, nested struct, list, and map value pruning with NULL validity, checked against Spark. They also cover deletes, partitions, `VERSION AS OF`, the nested partition source case, one schema in the pool, and data files without field ids. - `CometIcebergNativeSuite` passes 120 of 121, with one version-gated cancel. `CometFuzzIcebergSuite`, `CometIcebergResidualPushdownSuite`, and `CometIcebergRewriteActionSuite` pass. - A new case in `CometIcebergReadBenchmark` reads a 4M-row nested `partsupp` table (zstd, comments of about 128 characters). It ran on one local core with a warm page cache, using a release native build: | Query | Spark | Comet, every nested field | Comet, pruned | |---|---:|---:|---:| | `explode(partsupp_data.ps_availqty)` (Q20 shape) | 413 ms | 624 ms | 62 ms | | `explode(partsupp_data.ps_comment)` (control) | 2,800 ms | 590 ms | 542 ms | Bytes scanned for the first query drop from 124.0 MiB to 7.5 MiB. Locally the gain is all decode work, since the data sits in the page cache. On object storage the smaller read should add to it. That part is not measured yet. ### Additional context - iceberg-rust matches nested children by position rather than by field id (apache/iceberg-rust#2617). The pruned read projects the same leaves from the file as the task schema holds, so positions line up for every file the #6543 fallback lets through. - For data files without field ids and no name mapping, iceberg-rust projects whole top-level columns by position. Arrow's struct cast then narrows each one to the pruned type by field name. Results are correct, but there is no I/O saving. - iceberg-rust merges reads whose gaps are under 1 MiB (`DEFAULT_RANGE_COALESCE_BYTES`). On tables with small files the I/O saving is smaller than the size of the pruned fields suggests. The decode saving remains. - The cluster check still to do is TPC-H Q14, Q18, and Q20 on the nested schema, expecting Q20's partsupp scan to drop from 26.4 GiB to a few GiB. ### Willingness to contribute I can contribute a fix for this bug independently -- 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]
