ulysses-you commented on code in PR #58106:
URL: https://github.com/apache/spark/pull/58106#discussion_r3820517577
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -191,7 +198,8 @@ case class HashAggregateExec(
groupingExpressions.nonEmpty &&
!isStreaming &&
!groupingExpressions.exists(_.metadata.contains(SessionWindow.marker)) &&
- aggregateExpressions.forall(a => a.mode == Partial) &&
+ aggregateExpressions.forall(a => a.mode == Partial || a.mode ==
PartialMerge) &&
+ (aggregateExpressions.exists(_.mode == Partial) ||
aggregateExpressions.isEmpty) &&
Review Comment:
Thanks for the ablation; fair that the two checks are indistinguishable
today. I'd rather not pin it with a `copy(requiredChildDistributionExpressions
= None)` test: it fabricates a shape no built-in planner emits, so it asserts
the guard's implementation rather than an observable path, and in every real
plan the distribution check alone keeps the de-duplication phase out. I've left
the guard in as a defensive layer and scoped its comment to say exactly that
(see Finding 5). Happy to add the pin if you'd still prefer it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -162,12 +162,18 @@ case class HashAggregateExec(
* 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. `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.
- * The mode check is also what excludes the intermediate phase of a
DISTINCT plan, whose
- * modes are `PartialMerge ++ Partial`.
+ * - `Partial` and `PartialMerge` modes only: the downstream `Final`
aggregation merges the
+ * passed-through single-row buffers. `Final`/`Complete` produce the
result themselves and
+ * have no such downstream. A `PartialMerge` member is the non-distinct
aggregate of the
+ * DISTINCT intermediate phase
(`AggUtils.planAggregateWithOneDistinct`): its input row is
+ * already a partial buffer, so the pass-through applies the merge to an
empty buffer, which
+ * leaves the incoming buffer unchanged, and the downstream `Final`
re-merges it. A pure
+ * `PartialMerge` phase (the de-duplication on keys ++ distinct columns)
must not bypass, or
+ * duplicate (key, distinct column) rows would over-count DISTINCT. The
built-in planner
+ * never emits such a phase without a required distribution, so the
+ * `requiredChildDistributionExpressions` check below already keeps it
out; the
+ * `exists(_.mode == Partial)` check is a defensive guard against
third-party or future
Review Comment:
Good catch; for `count(DISTINCT v)` alone the de-duplication phase has an
empty `aggregateExpressions`, so `isEmpty` admits it and `exists(_.mode ==
Partial)` never runs. Scoped the comment (plus the test-helper comment and the
PR description) to what the guard actually covers, and kept the disjunct as-is
rather than narrowing to `initialInputBufferOffset == 0` since that path isn't
stress-tested.
--
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]
