ohad-definity opened a new pull request, #59706:
URL: https://github.com/apache/spark/pull/59706
### What changes were proposed in this pull request?
Nested schema pruning stops at a `Generate`: `SchemaPruning` (V1) and
`V2ScanRelationPushDown` only see the `Project`/`Filter` chain directly over a
scan, and the only path through an `explode` is
`GeneratorNestedColumnAliasing`, which pushes a *single* field accessor into
the generator input. So:
- reading two or more fields of an exploded element reads the whole element
(SPARK-34956, the `TODO` in `GeneratorNestedColumnAliasing`);
- with two chained explodes, e.g. `explode(items) AS item` then
`explode(item.parts) AS part`, reading `part.pid` reads all of `parts`
(SPARK-47230);
- an exploded element that is not read at all still reads the whole array
(SPARK-47230).
This PR lets both scan pruning rules look through a chain of `explode` /
`posexplode` generators (inner or outer) over a scan:
1. **Collect.** Going down from the top generator, the fields read from each
generator's element (above it, through nested field accesses, in projections
and filters) become a requirement on its input, built with
`SelectedField.withDataType`. The input is a scan column, or the element of a
generator below, which then adds those fields to its own requirement. An
element that is not read keeps its smallest field, for the array sizes (as in
`GenerateOptimization`); an element used whole needs its input whole.
2. **Narrow.** The scan is pruned by the existing code
(`prunePhysicalColumns` for V1, `PushDownUtils.pruneColumns` for V2), with the
generators' root fields as extra requested fields.
3. **Re-resolve.** The chain is re-resolved bottom-up against the narrower
types: references take their child's new type, nested field accesses are
re-resolved with `ProjectionOverSchema`, generator outputs take the new element
type. If the result is unresolved or the output schema would change, the plan
is left unchanged.
The last commit handles the `size(arr) > 0 AND isnotnull(arr)` filters
inferred for a non-outer explode, which otherwise read the whole element of a
generator in the chain (`RewriteSizeOfArrayStruct` only rewrites scan columns)
or a whole nested field like `items.parts`:
- within the chain, `size(x)` reads the smallest field of the outermost
array of structs of the same length, e.g. `size(items.parts)` →
`size(items.qty)`;
- `identifyRootFields` takes the generators' root fields, so they also drop
optional reads such as `isnotnull(items.parts)`.
Without it, a plain (non-outer) `explode(items) … explode(item.parts)`
reading `part.pid` still reads all of `parts`: 63 MB without it vs 0.2 MB with
it (two inner fields: 63 MB vs 0.4 MB). Outer chains and single explodes do not
need it.
It can be split into a follow-up PR if preferred.
The behavior is guarded by a new internal flag
`spark.sql.optimizer.nestedSchemaPruning.throughGenerate.enabled` (default
`true`). `GeneratorNestedColumnAliasing` is unchanged; for a single field it
still rewrites the generator first, and this path then has nothing left to
prune.
#### Related work
- Builds on `SelectedField.withDataType` from SPARK-57022 (nested pruning
through `transform`) to express what a generator's input needs, and on
`ProjectionOverSchema` to re-resolve.
- Closes the `TODO(SPARK-34956)` left in `GeneratorNestedColumnAliasing`
(SPARK-34638, which itself follows SPARK-27707).
- Uses the same "smallest field" choice as `GenerateOptimization` and
`RewriteSizeOfArrayStruct` (SPARK-58735) when only array sizes are needed.
- Earlier attempts at SPARK-47230, all closed without review: #45953 (a
`Project`/`Filter`/`Generate` pattern in `NestedColumnAliasing`, one field per
attribute), and #52288, #52598, #54283 by @IgorBerman. #52288 and #54283
rebuild the generator input per row (`transform`/`named_struct`, or a new
`arrays_zip`-style expression); #52598 narrows the scan like this PR, but
repairs the plan with a separate ordinal-fixing rule and runtime name-based
ordinal lookup in codegen. This PR keeps the change inside the two existing
scan pruning rules, adds no expression, and re-resolves with the existing
`ProjectionOverSchema`. Co-authored with @IgorBerman: many of the test shapes
here follow his suites in #52598 and #54283.
#### Alternatives considered
The `TODO(SPARK-34956)` in `GeneratorNestedColumnAliasing` suggests
rebuilding an array of structs from the fields read: `explode(items)` →
`explode(transform(items, x -> named_struct('a', x.a, 'b', x.b)))`, which
nested pruning through `transform` (SPARK-57022) would then narrow at the scan.
We did not take that route because:
- **It costs work on every row.** `transform` builds a new struct for every
element and a new array for every row, on top of reading the narrowed column.
Narrowing the scan type and re-resolving the plan costs nothing per row: the
scan returns rows already in their final shape.
- **It grows with depth.** Two chained explodes need a nested `transform`
and a struct rebuilt at each level, and every rebuilt struct's output type must
still match what the operators above expect.
- **The plan changes for everyone.** The rewrite inserts new expressions
into the user's plan even where the scan cannot prune (e.g. a source without
nested pushdown), whereas this change only alters types when the scan is
actually narrowed.
- **It does not cover the remaining shapes.** An unread element (only array
sizes needed) and the inferred `size()` / `isnotnull()` filters would still
need the handling in this PR.
Measured on master today, a hand-written `explode(transform(...))` is not
pruned at all: nested pruning through `transform` only applies in a projection
directly over the scan, not in a generator's input. Two fields of one explode
read 179 MB with the rewrite vs 27 MB with this change; one field through two
chained explodes reads 63 MB vs 0.2 MB. So the rewrite would also need to move
the `transform` into a projection below the generator. Even written that way by
hand:
| Query (Parquet, bytes read, median of 3) | `transform` in a projection
over the scan | This change |
|---|---|---|
| two fields of one `explode` | 27.1 MB, 1016 ms | 27.1 MB, 734 ms |
| one field through two chained explodes (nested `transform`) | 63 MB, 1618
ms | 0.2 MB, 608 ms |
At equal bytes the struct rebuild costs time (about 1.4× here, on a shared
CI runner, so treat it as indicative), and the nested `transform` is not pruned
below the first level.
### Why are the changes needed?
Exploding arrays of structs is common (event and log data, nested ML
features), and the inner structs are often wide. Today such queries read every
field of the exploded structs. With this change (Parquet, `/proc/self/io` bytes
read, flag off → on):
| Query | Bytes read | Time |
|---|---|---|
| two fields of one `explode`, 1M rows × 5 | 179 MB → 27 MB | 2.7 s → 0.7 s |
| `explode` output unused | 183 MB → 14 MB | 1.6 s → 0.4 s |
| one field through two chained explodes, 200k × 5 × 5 | 63 MB → 0.2 MB |
1.2 s → 0.6 s |
| two inner fields through two chained explodes | 63 MB → 0.4 MB | 1.1 s →
0.6 s |
On cloud storage, where time is dominated by bytes read, the same pruning
(implemented as an optimizer extension, not this patch) gave on Databricks
Runtime, Delta tables on S3 with the disk cache off, for two fields read across
two chained explodes (median of 3):
| Runtime | Bytes read | Time |
|---|---|---|
| DBR 13.3 (Spark 3.4) | 557 MB → 15.6 MB | 7.08 s → 1.18 s |
| DBR 17.3 (Spark 4.0) | 314 MB → 4.9 MB | 3.60 s → 1.05 s |
| DBR 18.0 (Spark 4.1) | 314 MB → 4.9 MB | 2.84 s → 1.10 s |
### Does this PR introduce _any_ user-facing change?
No change in query results. Queries that read nested fields through
`explode` / `posexplode` read fewer columns from Parquet and ORC (V1 and V2),
visible in `ReadSchema`. The existing SPARK-34638 / SPARK-41961 tests that
documented "Currently we don't prune multiple field case" now expect the
narrower schema.
### How was this patch tested?
New tests in `SchemaPruningSuite` (run for Parquet and ORC, V1 and V2,
vectorized and not):
- one level: multiple fields, a filter between the generator and the
projection, outer `posexplode`, unused element (outer and not), aggregate,
case-insensitive resolution, element used whole (not pruned), flag off, single
field still rewritten by `GeneratorNestedColumnAliasing`;
- two chained levels (`orders.items[].parts[]`): one inner field (outer and
not), multiple and non-adjacent inner fields, fields of both levels, two
`posexplode`s and mixed `explode`/`posexplode` chains, aggregate, inner element
used whole and outer element used whole (not pruned), a generator whose input
is not a field path (`explode(array(item.sku, 'x'))`);
- references written in another case than the generator alias
(`FRIEND.middle` for `friend`), at one and two levels, since re-resolution
matches references by `exprId`;
- other shapes: a filter on the exploded column itself, an
`array<array<struct>>` exploded twice, an array inside a struct, an explode
above an aggregate (not pruned through), an explode over a CTE selecting all
columns, and a streaming query over a file source. Several of these follow test
cases from #52598 and #54283.
Updated expectations in the SPARK-34638 / SPARK-41961 tests. New `Explode`
cases in `NestedSchemaPruningBenchmark` (flag off / on). Benchmark result files
will be regenerated with the `Run benchmarks` workflow.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
