peter-toth commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3755080190
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -141,6 +148,40 @@ case class HashAggregateExec(
.map(_.asInstanceOf[DeclarativeAggregate])
private val bufferSchema =
DataTypeUtils.fromAttributes(aggregateBufferAttributes)
+ /**
+ * Whether adaptive partial aggregation applies to this operator. When it
does, the aggregation
+ * may bypass partial aggregation at runtime and pass the remaining input
rows through as
+ * single-row partial buffers (see
[[SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED]]). It only
+ * applies to a pre-shuffle partial aggregation with grouping keys:
+ * - `Partial` mode only: the downstream `Final` aggregation merges the
passed-through
+ * single-row buffers, so the output contract is unchanged.
`Final`/`Complete` produce the
+ * result themselves and have no such downstream. `PartialMerge` does
have one and could be
+ * supported by passing its incoming buffer through unchanged, but that
is left for later.
+ * - grouping keys present: a global aggregation produces a single output
row, so partial
+ * aggregation achieves the maximum reduction and must never be bypassed.
+ * - DISTINCT aggregate functions are allowed: the intermediate
`PartialMerge` phase of the
+ * multi-phase distinct plan is not `Partial` mode (and requires a
distribution), so it always
+ * aggregates and de-duplicates, and the passed-through rows from the
distinct `Partial` phase
+ * therefore carry exactly one distinct value each.
+ */
+ private val adaptivePartialAggEnabled: Boolean = {
+ conf.adaptivePartialAggregationEnabled &&
+ groupingExpressions.nonEmpty &&
+ // Only the pre-shuffle partial aggregation has a downstream `Final` to
merge passed-through
+ // single-row buffers. `requiredChildDistributionExpressions` is `None`
exactly for that
+ // pre-shuffle phase and `Some` for the `Final`/`Complete` phase. This
check is what keeps a
+ // group-by-only aggregate (no aggregate functions, so an empty
`aggregateExpressions`) from
+ // being admitted vacuously: `aggregateExpressions.forall(_.mode ==
Partial)` alone is true
+ // for the empty list, which would wrongly make the `Final` phase
eligible as well.
+ requiredChildDistributionExpressions.isEmpty &&
Review Comment:
**Finding 8.** The new wording fixed the claim I raised but added another
one that is also false. It now says `planAggregateWithOneDistinct` leaves
`requiredChildDistributionExpressions` as `None` on a post-shuffle aggregate
"but the mode check covers that case". It doesn't — in exactly the shape your
own test exercises.
For `count(DISTINCT v) GROUP BY k` there are no non-distinct functions, so:
- step 2 `partialMergeAggregate` gets
`functionsWithoutDistinct.map(_.copy(mode = PartialMerge))` = empty, so it is a
group-by-only aggregate, not a `PartialMerge ++ Partial` mix. It is excluded by
`requiredChildDistributionExpressions = Some(...)`, not by the mode check.
- step 3 `partialDistinctAggregate` gets `Seq() ++
distinctAggregateExpressions`, i.e. all `Partial`, with no required
distribution. It passes both checks and is eligible — intended, and `count
distinct: the distinct partial aggregate bypasses` asserts it.
What makes step 3 safe is not the mode check: its output partitioning `(k,
v)` does not satisfy the `Final`'s `ClusteredDistribution(k)`, so
`EnsureRequirements` puts another `Exchange` between them and there is still a
`Final` to merge the passed-through buffers. The same two sentences in the
class doc at `:170-173` and in the PR description have the same problem.
```scala
// - DISTINCT aggregate functions are allowed. The phase that must
not bypass is the one
// de-duplicating on (keys ++ distinct columns); it requires a
distribution, so
// `requiredChildDistributionExpressions` keeps it out. The
distinct `Partial` phase that
// groups on the keys alone is eligible even though it sits after
a shuffle: another
// `Exchange` and a `Final` follow it, so its passed-through
buffers are still merged.
```
--
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]
