peter-toth opened a new pull request, #60030: URL: https://github.com/apache/spark/pull/60030
### What changes were proposed in this pull request? A `BatchScanExec` now keeps its input partitions in a new `PlannedInputPartitions` holder, which plans them once, when a node first asks for them. `DataSourceV2Strategy` gives each node it builds its own holder. The copies later rules make of the node share it, so they use the partitions the original node planned. Nothing is planned earlier than before. - The first node that asks plans the partitions with its own `Batch`, as `inputPartitions` did. - A copy reuses them instead of planning again, so a connector's `planInputPartitions()` runs less often. This follows SPARK-59289, which made the derived values of `GroupPartitionsExec` constructor fields for the same reason. A lazy val belongs to one instance, and a copy computes it again. Sharing a holder between copies is what `InMemoryRelation` does with its `CachedRDDBuilder`. ### Why are the changes needed? A storage-partitioned join can fail with `Can't zip RDDs with unequal numbers of partitions` 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 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 the scalar subquery in the node's runtime filters with a `ReusedSubquery`. The copy plans its input partitions again, from the already pruned `Scan`. 4. `replanWithRuntimeFilters` pads the filtered partitions to the keys of those input partitions. So the join's scan reports 1 partition, while the other side reports 7. For example, with `items` partitioned by `id` and `purchases` partitioned by `item_id`, ids 1 to 7 with one split per key, 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)`. branch-4.1 and branch-4.0 are not affected, because they do not route scalar subquery filters into runtime filtering. ### Does this PR introduce _any_ user-facing change? Yes. Such a query now returns its rows instead of failing. ### How was this patch tested? New test in `KeyGroupedPartitioningRuntimeFilterTests`, with an `=` and a `>=` scalar subquery filter, AQE on and off. It checks the rows and that the join plans no shuffle. - Without the fix it fails in `KeyGroupedPartitioningCatalystRuntimeFilterSuite` with `List(7, 1)`. - `KeyGroupedPartitioningRuntimeFilterSuite` runs it too, but its fixture prunes on `IN` only, so it passes there without the fix. Also ran `KeyGroupedPartitioningSuite`, `GroupPartitionsExecSuite`, `SameResultSuite`, `SparkPlanSuite`, the `DynamicPartitionPruningV2*` suites, `DataSourceV2CatalystRuntimeFilterSuite`, `DataSourceV2SQLSuiteV2Filter`, `DSv2PlanMergingSuite`, and the suites whose `BatchScanExec` patterns got the new field. ### 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]
