peter-toth opened a new pull request, #60172: URL: https://github.com/apache/spark/pull/60172
### What changes were proposed in this pull request? Backport of #60030 to branch-4.2, in the shape of the branch-4.3 backport #60125. 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.3 commit applies here with two changes: - `BatchScanExec` adds only `SparkPlan` to its imports. This branch's file does not import `EmptyRDDWithPartitions`. - The test catalog `UnsupportedColumnarModeCatalog` overrides `InMemoryScanBuilder.build`, and copies the scan it builds into one that declares `UNSUPPORTED`. This branch's `InMemoryScanBuilder` has no `createScan` hook, which came with SPARK-58523. ### 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. It can be a DPP filter from another join, or a scalar subquery filter routed there by SPARK-56467. 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, and the join fails with `Can't zip RDDs with unequal numbers of partitions`. For example, take a connector whose `filter()` drops the partitions an `=` predicate rules out, with `items` partitioned by `id` and `purchases` partitioned by `item_id`, ids 1 to 7 with one split per key. 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, i.name, p.price FROM items i JOIN purchases p ON i.id = p.item_id WHERE i.id = (SELECT max(item_id) FROM purchases WHERE price > 4.0) ``` fails with `Can't zip RDDs with unequal numbers of partitions: List(7, 1)`. The same happens when `purchases` is first joined with a broadcast dimension, whose DPP filter is then the build side's own runtime filter. A connector that prunes on `IN` only is enough for that. 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, adapted to this branch. It has neither the Catalyst runtime filter fixture nor the runtime filter test trait, so they are in `KeyGroupedPartitioningSuite`. Each checks the rows and that the join plans no shuffle. - A scalar subquery filter, with one split per key, over `InMemoryTableWithV2FilterCatalog`, whose scans prune on `=`. AQE off. - 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 each fails 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. On this branch, also ran `GroupPartitionsExecSuite`, `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]
