ulysses-you commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3735265751
##########
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:
Confirmed, and I do not have a fix I am confident in yet.
Your measurement matches what I saw from a different direction while
checking the ROLLUP tests: a five-column rollup bypasses even though its
overall compaction ratio is 1.2, because the first check lands before the
expansion has covered every grouping set. Same failure -- a prefix that does
not represent the task decides it -- just reached from another angle.
Two levers, and they pull against each other:
- Scale the first check point to what the task has seen, so a small window
cannot commit a large task. Cheap, but it only delays the same mistake on a big
enough input.
- Let a later check undo the bypass. That addresses it properly, but the
maps are frozen and freed once pass-through fires, so resuming means rebuilding
-- and it also interacts with item 6, since a window that can reverse wants to
judge its own rows rather than the epoch cumulative.
Do you have a preference? I would rather agree the shape before
implementing, given the last few rounds here.
--
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]
