peter-toth opened a new pull request, #60174:
URL: https://github.com/apache/spark/pull/60174
### What changes were proposed in this pull request?
Backport of #60030 to branch-4.0, in the shape of the branch-4.3, branch-4.2
and branch-4.1 backports #60125, #60172 and #60173. The fix is the same as on
master, but its shape keeps the signature of `BatchScanExec` in this
maintenance branch:
- `BatchScanExec` keeps its `scan` field. Each node holds a `PlannedScan` in
a private field instead, built from `scan`.
- `makeCopy` and `clone` hand the node's `PlannedScan` over to a copy over
the same `Scan`. The rules that rewrite the node do that through `makeCopy`.
`clone` needs its own override, because `TreeNode.clone` does not go through
`makeCopy`.
- Any other new node plans its own, including one the case-class `copy`
makes. Nothing in Spark copies an executable `BatchScanExec` with `copy`.
- `PlannedScan` is package private here.
A `PlannedScan` holds the `Scan`, with the `Batch`, its reader factory and
the input partitions planned from it once, when a node first asks for them. So
the copies read the partitions the original node planned, with the reader
factory of the `Batch` that planned them. A pushed runtime filter still replans
from a new `Batch`, as before.
Nothing is planned earlier than before.
- The first node that asks plans the partitions, as `inputPartitions` did.
- A copy reuses them instead of planning again, so a connector's
`planInputPartitions()` runs less often.
Handing mutable state over in `makeCopy` and `clone` is what
`InMemoryRelation` does with its `statsOfPlanToCache`.
The branch-4.1 commit applies here with these changes:
- `BatchScanExec` adds only `SparkPlan` to its imports. `PlannedScan` comes
after `StoragePartitionJoinParams`, which this branch defines in the same file.
- `UnsupportedColumnarModeCatalog` overrides this branch's `createTable`,
which has no constraints, and builds its `InMemoryTable` from a schema.
Like the branch-4.1 backport, compared with the branch-4.2 commit:
- `BatchScanExec` keeps this branch's `spjParams` field and
`keyGroupedPartitioning` override. The fix touches neither.
- `checkStoragePartitionedJoin` runs each case with
`spark.sql.sources.v2.bucketing.pushPartValues.enabled` on and off. With it on,
this branch hides most of the bug, see below.
- `UnsupportedColumnarModeCatalog` copies no pushed filters into its scan,
since this branch's in-memory scan has none.
- `SparkPlanSuite` gets only the new test. A test comment no longer mentions
`GroupPartitionsExec`, which this branch does not have.
### Why are the changes needed?
A storage-partitioned join can fail when the build side of a DPP subquery
has a runtime filter of its own. With AQE off:
1. A DPP subquery that does not reuse a broadcast plans its build side
again. Its `BatchScanExec` wraps the same `Scan` instance as the join's scan on
that side.
2. The build side has a runtime filter of its own, a DPP filter from another
join. The DPP subquery runs first, and its `filter()` call prunes the shared
`Scan`. The runtime filtering contracts allow that.
3. The join's `BatchScanExec` is a copy made after planning.
`ReuseExchangeAndSubquery` made it when it replaced a subquery in the node's
runtime filters with a reused one. The copy plans its input partitions again,
from the already pruned `Scan`.
4. So the copy reports fewer partitions than the join was planned for.
On this branch, as on branch-4.1, the scan pads its partitions to the common
partition values the join pushed down, when
`spark.sql.sources.v2.bucketing.pushPartValues.enabled` is on (the default).
That hides the missing partitions of the copy. The join fails with `Can't zip
RDDs with unequal numbers of partitions` in two cases:
- With `pushPartValues` off, e.g. `List(7, 3)`.
- When the build side's filter prunes every split, also with
`pushPartValues` on. The copy then reports no `KeyGroupedPartitioning`, so
nothing is padded, e.g. `List(7, 0)`.
For example, take `items` partitioned by `id`, `purchases` partitioned by
`item_id`, ids 1 to 7, and a dimension table `dim(k, flag)` with no row where
`flag = 2`, over a connector whose `filter()` prunes on `IN`. With DPP on,
`spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly=false`,
`spark.sql.optimizer.dynamicPartitionPruning.fallbackFilterRatio=10`,
`spark.sql.autoBroadcastJoinThreshold=-1` and AQE off:
```sql
SELECT i.id, p.price FROM items i JOIN
(SELECT /*+ BROADCAST(d) */ p.item_id, p.price FROM purchases p
JOIN dim d ON p.item_id = d.k WHERE d.flag = 2) p
ON i.id = p.item_id
```
fails with `Can't zip RDDs with unequal numbers of partitions: List(7, 0)`.
The scalar subquery shape of #60030 does not fail on this branch, because
scalar subquery partition filters reach DSv2 runtime filtering only from
SPARK-56467, in 4.2.0. Its test is kept.
With AQE on, a scan in the default `PARTITION_DEFINED` columnar support mode
is not affected. AQE asks the copies `supportsColumnar` when it creates the
result stage, and in that mode this plans their partitions before the DPP
subquery runs. A scan that declares `SUPPORTED` or `UNSUPPORTED` plans nothing
there, so it fails the same way with AQE on.
### Does this PR introduce _any_ user-facing change?
Yes. Such a query now returns its rows instead of failing.
### How was this patch tested?
The tests of #60030, in `KeyGroupedPartitioningSuite` as in #60125. Each
checks the rows and that the join plans no shuffle, with `pushPartValues` on
and off.
- A scalar subquery filter, with one split per key, over
`InMemoryTableWithV2FilterCatalog`, whose scans prune on `=`. AQE off. It
passes without the fix on this branch, see above.
- A DPP filter from a broadcast dimension, with two splits per key, once
keeping 3 keys and once pruning every split. AQE off.
- The DPP filter case over a catalog whose scans declare `UNSUPPORTED`, with
AQE on and off.
Without the fix the last two fail with `Can't zip RDDs with unequal numbers
of partitions`, the last one with AQE on too.
A new test in `SparkPlanSuite`, over a `Scan` that returns a new `Batch`
from each `toBatch`. The copies `makeCopy` and `clone` make share the node's
`Batch` and partitions, and a `makeCopy` over another `Scan` does not. Without
the fix it fails at the first assertion.
The nested shape also passes, with the right rows and no shuffle, over
`InMemoryTableCatalog` and `InMemoryTableWithV2FilterCatalog`, with
`pushPartValues` and AQE on and off, and with or without a dimension match.
On this branch, also ran `SameResultSuite`, `SQLAppStatusListenerSuite`,
`EnsureRequirementsSuite`, the `DynamicPartitionPruningV2*` suites,
`DataSourceV2SQLSuiteV2Filter`, `DataSourceV2Suite`,
`FileBasedDataSourceSuite`, `DataSourceV2OptionSuite`,
`PruneFileSourcePartitionsSuite`, `OrcV2SchemaPruningSuite`, the
`GroupBasedRowLevelOperation*` suites, `AvroRowReaderSuite` and `AvroV2Suite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code with Claude Opus 5.5
--
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]
