peter-toth commented on code in PR #58106:
URL: https://github.com/apache/spark/pull/58106#discussion_r3819949540
##########
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:
**Finding 5.** This sentence claims more than the check delivers. Take the
commonest DISTINCT shape, `count(DISTINCT v)` with no non-distinct aggregate:
`functionsWithoutDistinct` is empty, so
`AggUtils.planAggregateWithOneDistinct`'s de-duplication phase carries *no*
aggregate functions at all. Dumped on this head:
```
*(1) HashAggregate(keys=[k], functions=[count(distinct v)]) Final
reqDist=true
+- *(1) HashAggregate(keys=[k], functions=[partial_count(distinct v)])
Partial reqDist=false
+- *(1) HashAggregate(keys=[k, v], functions=[]) <no
modes> reqDist=true <- de-duplication
+- *(1) HashAggregate(keys=[k, v], functions=[]) <no
modes> reqDist=false
```
`aggregateExpressions.isEmpty` is true for that phase, so the `||
aggregateExpressions.isEmpty` disjunct on line 202 admits it and `exists(_.mode
== Partial)` never runs. I confirmed it: taking that phase and dropping its
required distribution, `metrics.contains("numBypassingRows")` comes back
`true`. So for exactly the shape this paragraph is about, the guard adds
nothing and the distribution check is the only thing keeping the phase out —
same as before this PR.
Not a live bug, but the wording should say what it covers:
```scala
* `requiredChildDistributionExpressions` check below already keeps it
out. The
* `exists(_.mode == Partial)` check is a defensive guard on top of it
for a de-duplication
* phase that carries non-distinct aggregates; one with none at all
has an empty
* `aggregateExpressions`, is admitted by the `isEmpty` disjunct, and
still relies on the
* distribution check alone.
```
The same overstatement is in the PR description ("kept out by its required
distribution plus a defensive `exists(_.mode == Partial)` guard") and in the
new test helper comment at `AdaptivePartialAggregationSuite.scala:224-226`
("the one phase the `exists(_.mode == Partial)` guard exists to exclude"). That
last one has a practical consequence: `dedupPhases` is empty in the three
`count distinct: ...` tests, which have no non-distinct aggregate, so the
assertion is vacuous there.
Optional, if you would rather close the hole than document it: narrow the
disjunct to `aggregateExpressions.isEmpty && initialInputBufferOffset == 0`. A
pre-shuffle group-by-only partial aggregate has offset 0 (`AggUtils.scala:178`)
while the de-duplication phase has `(groupingAttributes ++
distinctAttributes).length` (`:248`). I have not stress-tested that, so the
test in finding 4 plus honest wording is enough for me.
##########
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:
**Finding 4.** Ablated on this head: I deleted this line, rebuilt, and ran
the whole suite — 54/54 still green, including the `dedupPhases` assertion
added for finding 2. That assertion checks the pure-`PartialMerge`
de-duplication phase reports no `numBypassingRows` metric, but that phase is
kept out by `requiredChildDistributionExpressions.isEmpty` on the line below,
which was already there before this PR. So nothing in the suite tells the two
checks apart, and this one is free to rot. That is on my R1 ask, not on your
fix.
A shape that does fail without the line has to build the operator the
built-in planner never emits — the de-duplication phase with its required
distribution dropped:
```scala
test("a pure PartialMerge phase without a required distribution is not
eligible") {
// The built-in planner always sets
`requiredChildDistributionExpressions` on the
// de-duplication phase, so drop it to get the shape this guard is for:
bypassing that phase
// would emit duplicate (k, v) rows and over-count the DISTINCT.
withSQLConf(
(Seq(SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "true") ++
fixedPlanConfs): _*) {
val df = spark.range(0, 400, 1, 1)
.select(($"id" % 4).cast("string") as "k", $"id" as "v")
.groupBy($"k").agg(countDistinct($"v") as "cd", sum($"v") as "s")
df.collect()
val dedup = collect(df.queryExecution.executedPlan) {
case agg: HashAggregateExec if agg.aggregateExpressions.nonEmpty &&
agg.aggregateExpressions.forall(_.mode == PartialMerge) => agg
}
assert(dedup.size == 1 &&
dedup.head.requiredChildDistributionExpressions.isDefined)
val unconstrained =
dedup.head.copy(requiredChildDistributionExpressions = None)
assert(!unconstrained.metrics.contains("numBypassingRows"),
"a pure PartialMerge phase must never be eligible for the bypass")
}
}
```
I ran it both ways: it fails with this line deleted (`contained key
"numBypassingRows"`) and passes with it in place. The `copy` has to sit inside
the `withSQLConf` block — `adaptivePartialAggEnabled` is computed in the
constructor from `SQLConf.get`.
--
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]
