ulysses-you commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3735236805
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -845,29 +1158,56 @@ case class HashAggregateExec(
}
}
- val declareRowBuffer: String = if (isFastHashMapEnabled) {
- val fastRowType = if (isVectorizedHashMapEnabled) {
- classOf[MutableColumnarRow].getName
+ val declareRowBuffer: String = {
+ val declareBuffers = if (isFastHashMapEnabled) {
+ val fastRowType = if (isVectorizedHashMapEnabled) {
+ classOf[MutableColumnarRow].getName
+ } else {
+ "UnsafeRow"
+ }
+ s"""
+ |UnsafeRow $unsafeRowBuffer = null;
+ |$fastRowType $fastRowBuffer = null;
+ """.stripMargin
+ } else {
+ s"UnsafeRow $unsafeRowBuffer = null;"
+ }
+ val declareBypassed = if (adaptivePartialAggEnabled) {
+ s"boolean $adaptiveRowBypassedTerm = false;"
} else {
- "UnsafeRow"
+ ""
}
s"""
- |UnsafeRow $unsafeRowBuffer = null;
- |$fastRowType $fastRowBuffer = null;
+ |$declareBuffers
+ |$declareBypassed
""".stripMargin
- } else {
- s"UnsafeRow $unsafeRowBuffer = null;"
}
// We try to do hash map based in-memory aggregation first. If there is
not enough memory (the
// hash map will return null for new key), we spill the hash map to disk
to free memory, then
// continue to do in-memory aggregation and spilling until all the rows
had been processed.
// Finally, sort the spilled aggregate buffers by key, and merge them
together for same key.
+ //
+ // With adaptive partial aggregation, once pass-through is active
`updateRowInHashMap` fills the
+ // single-row buffer built above; we then emit `key ++ buffer` straight to
the parent so the row
+ // skips both the fast map and the regular map.
+ val emitPassThroughRow = if (adaptivePartialAggEnabled) {
+ val numBypassingRows = metricTerm(ctx, "numBypassingRows")
+ s"""
+ |if ($adaptiveRowBypassedTerm) {
+ | $numBypassingRows.add(1);
+ | $outputFunc(${unsafeRowKeyCode.value}, $unsafeRowBuffer);
Review Comment:
Fixed in 87937dbd6d0. You and @peter-toth are right that this had to be
resolved regardless of whether Spark promises a merge order --
`spark.sql.codegen.wholeStage` must not be observable in results.
I aligned the interpreted path to the generated one rather than the reverse.
In `TungstenAggregationIterator.next()` the pass-through branch now comes
first, so streamed rows precede the frozen map contents, which is what codegen
already does (a bypassed row is emitted from inside the build loop; the maps
drain afterwards).
Going the other way -- making codegen drain the maps before the triggering
row -- is what I tried first, and it does not work cleanly: the emit site sits
inside the child throughput loop, so returning from there skips the child own
stop check and corrupts its resume state, and draining appends several rows in
one `doConsume` call, which breaks the one-row-per-call contract the output
buffer relies on and dropped rows under a fused `UnionExec`.
Verified across codegen x two-level-map x spill (8 cells): all agree now.
Added `both execution paths agree on an order-sensitive aggregate`, which I
confirmed fails on the previous code with exactly the two divergent results:
```
the execution paths disagree: List([[-1,8,0],...], [[-1,0,8],...])
```
Note this settles path agreement, not the merge order itself: both paths now
return `first=8, last=0` where a non-bypassed run returns `(0, 8)`. That
question is still open.
--
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]
