ulysses-you commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3735257517
##########
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:
Confirmed and fixed in 87937dbd6d0. `AggUtils.planAggregateWithOneDistinct`
builds `partialDistinctAggregate` through `createAggregate` without passing
`requiredChildDistributionExpressions`, so it defaults to `None` while sitting
after a shuffle -- exactly as you say, and exactly the aggregate the `count
distinct: the distinct partial aggregate bypasses` test exercises.
The comment now says what actually carries the gate: every function must be
in `Partial` mode, which is what excludes that aggregate (its modes are
`PartialMerge ++ Partial`). `requiredChildDistributionExpressions.isEmpty` is
kept for the one case the mode check cannot decide -- a group-by-only aggregate
has no functions at all, so `forall` is vacuously true for both its phases --
and the comment now notes it is not a general pre-shuffle test, naming your
counterexample.
--
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]
