HyukjinKwon commented on code in PR #56777:
URL: https://github.com/apache/spark/pull/56777#discussion_r3478154691
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala:
##########
@@ -127,11 +127,34 @@ object AggUtils {
aggregateExpressions: Seq[AggregateExpression],
resultExpressions: Seq[NamedExpression],
child: SparkPlan): Seq[SparkPlan] = {
- // Check if we can use HashAggregate.
+
+ val groupingAttributes = groupingExpressions.map(_.toAttribute)
+
+ // When partial aggregation is disabled, skip the pre-shuffle partial
aggregation and run a
+ // single Complete-mode aggregation after the shuffle. This can improve
performance when the
+ // group cardinality is high and the pre-shuffle reduction ratio is low.
+ //
+ // session_window requires MergingSessionsExec (inserted below via
mayAppendMergingSessionExec)
+ // to sort and merge overlapping sessions before the final aggregation.
The bypass is skipped
+ // when a session_window grouping key is present so that the normal
Partial+Merge+Final path
+ // runs and MergingSessionsExec is correctly inserted.
+ val hasSessionWindow =
groupingExpressions.exists(_.metadata.contains(SessionWindow.marker))
+ if (child.conf.bypassPartialAggregation && !hasSessionWindow) {
+ val completeAggregateExpressions = aggregateExpressions.map(_.copy(mode
= Complete))
+ val completeAggregateAttributes =
completeAggregateExpressions.map(_.resultAttribute)
+ val completeAggregate = createAggregate(
+ requiredChildDistributionExpressions = Some(groupingAttributes),
Review Comment:
`requiredChildDistributionExpressions = Some(groupingAttributes)` is applied
over the **raw** child here, but `groupingAttributes =
groupingExpressions.map(_.toAttribute)`. For any grouping key that isn't a
plain child attribute — e.g. `GROUP BY v % 10` — `SparkStrategies` wraps it as
`Alias(v % 10, "k")` (SparkStrategies.scala ~L708), so `.toAttribute` is a
synthetic `AttributeReference` that is **not in the raw child's output**. The
resulting `ClusteredDistribution` → `HashPartitioning(thatAttr)` then fails to
bind against the child at execution.
The normal Final path (L186) can use `Some(groupingAttributes)` only because
its child is the Partial agg that *produces* those attributes; the bypass's
child is the raw input, so it should distribute on the expressions themselves:
```suggestion
requiredChildDistributionExpressions = Some(groupingExpressions),
```
This is currently untested: the first test groups by `(v % 10).as("k")` but
only checks `executedPlan` structure (no execution), and the SUM/COUNT/AVG
tests group by a materialized plain column `k`. An executing `GROUP BY
<expression>` test (collect + `checkAnswer` under the config) would catch it.
(Side note: with the fix, the bypass evaluates the grouping expression twice —
in the shuffle partitioning and in the Complete agg — vs once in the two-phase
path; fine for deterministic keys, worth a thought for nondeterministic ones.)
--
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]
