peter-toth commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3755079421
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -711,6 +995,54 @@ case class HashAggregateExec(
} else {
findOrInsertRegularHashMap
}
+
+ // Every row is either accepted by an aggregation map or streamed
through -- the fast map
+ // serves a row without it ever reaching the regular map, so both
buffers are consulted to
+ // tell the two apart.
+ //
+ // An accepted row counts toward the compaction ratio, so the numerator
matches the
+ // operator-level denominator. Counting inside the regular-map branch
alone would drop the
+ // rows the fast map absorbed from the ratio and bypass an aggregation
that is in fact
+ // reducing. A row no map holds is streamed once pass-through is active:
both probes are
+ // skipped (guarded above), so neither buffer is set, and `rowBypassed`
marks exactly those
+ // rows. The row that fails to insert at the spill boundary lands here
too, while the row
+ // that merely flipped pass-through at the check point is already
aggregated in the map that
+ // took it and must not be re-emitted.
+ val countOrPassThroughRow = if (adaptivePartialAggEnabled) {
+ val heldByAMap = if (isFastHashMapEnabled) {
+ s"($fastRowBuffer != null || $unsafeRowBuffer != null)"
+ } else {
+ s"($unsafeRowBuffer != null)"
+ }
+ // The grouping key was already projected in
`findOrInsertRegularHashMap`
+ // (`unsafeRowKeyCode.code`), so `unsafeRowKeyCode.value` holds this
row's key. Only build
+ // the single-row partial buffer here.
+ s"""
+ |if ($heldByAMap) {
+ | if (!$adaptivePassThroughTerm) {
+ | $processedRowsTerm += 1;
+ | if ($processedRowsTerm == $adaptiveNextCheckRowTerm) {
Review Comment:
**Finding 2.** Thanks for confirming the repro. I'm not pushing this as a
merge blocker. Re-arming the flip is the only thing that covers it in general,
and that's more surgery than I made it sound: the flip also frees the maps
(`outputMap()` drains and frees them), so re-arming means a fresh
`createHashMap()` plus resetting `adaptiveMapOutputDone`/`adaptiveMapSetupDone`
— a per-epoch map lifecycle in both execution paths. Fair to leave for later.
Two corrections to what I wrote before.
`minCompaction` does not bound this, so the conservative default doesn't
help here: the prefix in the repro is fully distinct, ratio 1.0, so it flips at
any threshold at or above 1.0. Worth being explicit about, because the
threshold is the knob the docs point users at.
And my finding 15 suggestion was overstated. Probe-retention only rescues a
tail whose keys the prefix already inserted — which is this repro, since `id %
100` reuses keys 0-99 that are already in the map — and does nothing when the
tail's hot keys are new, because those are absent from the frozen map and every
one of their rows still streams. Partial mitigation, not a fix.
What I'd ask for instead is one sentence in
`docs/sql-performance-tuning.md`, since `enabled` is the only real opt-out and
nothing currently tells a user this behaviour exists:
> The decision is made from the rows observed so far and is not reversed for
the rest of the task, so a partition whose leading rows are much more distinct
than the rest may keep passing rows through even where aggregating them would
have reduced more. Set
`spark.sql.execution.aggregate.adaptivePartialAggregation.enabled` to `false`
to opt out.
--
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]
