peter-toth commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3755078003
##########
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:
**Finding 1.** The non-fan-out shape agrees now, thanks — but the contract
still breaks when the child produces more than one row per `doConsume`.
`GenerateExec` and `ExpandExec` have no `shouldStop()` inside their fan-out
loop, so in the exchange-split shape the whole batch belonging to the trigger
input row is appended before the leaf's stop check returns, and only then does
`adaptiveResumeBuild` drain the maps. The interpreted path never sees the
fan-out — it emits exactly one row, then the map, then the rest. A group whose
rows straddle that boundary therefore merges in a different order on the two
paths.
Measured on `388a2d76`, with `minRows = 8`, `enableTwoLevelAggMap = false`,
everything else default:
```scala
spark.range(0, 20, 1, 2)
.select($"id", explode(array(
$"id" * 2,
when($"id" === 4, lit(0L)).otherwise($"id" * 2 + 1))) as "k")
.select($"k", ($"id" * 100 + $"k") as "v")
.groupBy($"k").agg(first($"v") as "f", last($"v") as "l")
```
Plan is `HashAggregate(Final)` / `Exchange` / `HashAggregate(partial)` /
`Project` / `Generate explode(...)` / `Range`. Reading the `k = 0` row:
| run | (f, l) | numBypassingRows |
| --- | --- | --- |
| feature off, codegen | (0, 400) | 0 |
| feature off, interpreted | (0, 400) | 0 |
| feature on, codegen | **(400, 0)** | 24 |
| feature on, interpreted | (0, 400) | 24 |
Every id fans out into two rows, all keys distinct except id 4's second row,
which collides with id 0's first (`k = 0`). The periodic check flips on the 8th
accepted row, so id 4's batch is fully bypassed; codegen puts both of its rows
ahead of the frozen map, so the streamed `v = 400` reaches the `Final` before
the map's `v = 0`.
Note which side is right: the interpreted path matches the non-bypassed run,
codegen does not. That is the same direction I argued at R2 — draining the maps
before the first bypassed row is the only stable order, because a group's map
buffer always holds its *earlier* rows. Map-first gives `(0, 400)` on both
paths here and matches the reference; trigger-first can only match when the
collision happens to miss the trigger batch.
The new order test cannot see this because its child emits one row per input
row. A cell with a generator below the aggregate at `splits = 2` would cover it.
--
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]
