ulysses-you opened a new pull request, #59950:
URL: https://github.com/apache/spark/pull/59950
### What changes were proposed in this pull request?
Dynamic partition pruning (DPP) builds a subquery that collects the distinct
build-side join keys. Both `PlanDynamicPruningFilters` and
`PlanAdaptiveDynamicPruningFilters` constructed it as `Aggregate(aliases,
aliases, buildPlan)`, i.e. the grouping expressions were aliases of the join
keys.
This PR moves those aliases into a `Project` below the aggregate and groups
by the attributes that projection produces:
```scala
val groupingAttrs = aliases.map(_.toAttribute)
val aggregate = Aggregate(groupingAttrs, groupingAttrs, Project(aliases,
buildPlan))
```
The subquery's output schema is unchanged; only where the alias is evaluated
changes.
Grouping by attributes produced by a lower `Project` is the shape that
`PullOutGroupingExpressions` establishes for complex grouping expressions. DPP
builds its subquery after logical optimization, so that rule never sees it;
this change applies the same shape when the subquery is constructed.
### Why are the changes needed?
When the build side is a storage-partitioned-join scan, it reports a
`KeyedPartitioning`. Physical planning splits the aggregate into partial and
final. The partial aggregate's `resultExpressions` use the grouping attributes
(`groupingExpressions.map(_.toAttribute)`), so the `Alias` object is no longer
part of `outputExpressions`. `BaseAggregateExec` extends
`PartitioningPreservingUnaryExecNode`, whose alias map is built from the
`Alias` nodes in `outputExpressions`, so it cannot project the scan's keyed
expressions onto the aggregate's output attribute, drops the
`KeyedPartitioning` and reports `UnknownPartitioning`. `EnsureRequirements`
then inserts an `Exchange hashpartitioning(...)` between the partial and final
aggregate.
With a plain attribute as the grouping expression the partial aggregate
keeps the keyed partitioning, and `EnsureRequirements` coalesces duplicate keys
with `GroupPartitionsExec` instead of shuffling:
```
Before
HashAggregate (Final)
+- Exchange hashpartitioning(id, 5)
+- HashAggregate (Partial)
+- Project / Filter / BatchScan (KeyedPartitioning)
After
HashAggregate (Final)
+- GroupPartitionsExec
+- HashAggregate (Partial)
+- Project [id AS id]
+- Filter
+- BatchScan (KeyedPartitioning)
```
JIRA: https://issues.apache.org/jira/browse/SPARK-60741
### Does this PR introduce _any_ user-facing change?
No. Query results are unchanged. The physical plan of the DPP build-side
subquery no longer contains a shuffle when the build side reports
`KeyedPartitioning`.
### How was this patch tested?
A new test, `SPARK-60741: dynamic partition filtering keeps the build side's
keyed partitioning`, in `KeyGroupedPartitioningRuntimeFilterTests` (it runs
against both the `SupportsRuntimeFiltering` and the
`SupportsRuntimeCatalystFiltering` fixtures), with `ADAPTIVE_EXECUTION_ENABLED`
both off and on, covering `PlanDynamicPruningFilters` and
`PlanAdaptiveDynamicPruningFilters` respectively. It asserts that the DPP
subquery exists, that its plan has no `ShuffleExchangeLike` and contains a
`GroupPartitionsExec`, that the outer storage-partitioned join adds no shuffle,
and that the result is correct.
Regression:
* All 8 concrete `DynamicPartitionPruning*` suites: 341 passed, 11 ignored,
0 failed.
* `KeyGroupedPartitioningRuntimeFilterSuite` and its Catalyst variant, full
run: 10 passed, 0 failed.
```bash
build/sbt 'sql/testOnly *DynamicPartitionPruningV1SuiteAEOff
*DynamicPartitionPruningV1SuiteAEOn *DynamicPartitionPruningV2SuiteAEOff
*DynamicPartitionPruningV2SuiteAEOn *DynamicPartitionPruningV2FilterSuiteAEOff
*DynamicPartitionPruningV2FilterSuiteAEOn
*DynamicPartitionPruningV2CatalystFilterSuiteAEOff
*DynamicPartitionPruningV2CatalystFilterSuiteAEOn'
build/sbt 'sql/testOnly
org.apache.spark.sql.connector.KeyGroupedPartitioningRuntimeFilterSuite
org.apache.spark.sql.connector.KeyGroupedPartitioningCatalystRuntimeFilterSuite'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (deepseek-flash[1m])
🤖 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]
