peter-toth commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3734474820
##########
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.** @ulysses-you to your question — I think Spark indeed doesn't
promise a merge order, so I wouldn't call the `first`/`last` change a broken
contract on its own. But there's a second consequence that I think does have to
be fixed: **the two execution paths of this feature disagree with each other**,
so results depend on `spark.sql.codegen.wholeStage`.
On the interpreted path `nextPassThroughOutput()` is the last branch of
`next()`, so the map (or sort) output always drains first. On the codegen path
`emitPassThroughRow` runs at the end of `doConsumeWithKeys`, so the first
bypassed row is appended from inside the build loop and `adaptiveResumeBuild`
only calls `outputMap()` on the *next* `processNext()`. Exactly one bypassed
row (or one fan-out batch, with `Expand`/`Generate` below) therefore jumps
ahead of the frozen map, and if its key is already in the map the `Final` merge
sees that group's buffers in the opposite order.
Measured on `f2616ecb556`, single partition, `minRows = 8`,
`enableTwoLevelAggMap = false`, all else default — 8 distinct keys fill the map
and trip the periodic check, then `id = 8` repeats key `0`:
```scala
spark.range(0, 40, 1, 1)
.select(when($"id" === 8, lit(0L)).otherwise($"id") as "k", $"id" as "v")
.groupBy($"k").agg(first($"v") as "f", last($"v") as "l")
```
| adaptive | wholeStage | row for `k = 0` |
| --- | --- | --- |
| true | true | `f = 8, l = 0` |
| true | false | `f = 0, l = 8` |
| false | either | `f = 0, l = 8` |
The suite can't see this: `checkAnswer` is order-insensitive, and the one
test with `first`/`last` ("a mix of many aggregate functions and buffer types")
groups on fully distinct keys, so every group has a single row.
So whichever order you decide the feature has, please make both paths
implement it and add a test that pins the *values* (not just the set) for a
group split across the map and the stream. If you keep the map-first order
@sunchao asked for, the codegen side needs the interpreted path's trick — stash
the flip row the way `pendingPassThroughRow` does and emit it after
`outputMap()` has finished, remembering that a fan-out child can produce
several such rows in one `doConsume` call.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -626,13 +788,27 @@ case class HashAggregateExec(
| long $beforeAgg = System.nanoTime();
| $doAggFuncName(partitionIndex);
| $aggTime.add((System.nanoTime() - $beforeAgg) / $NANOS_PER_MILLIS);
+ | $adaptiveStopCheck
|}
- |// output the result
- |$outputFromFastHashMap
- |$outputFromRegularHashMap
+ |$adaptiveResumeBuild
+ |$adaptiveFinalOutput
""".stripMargin
}
+ // Blocking operators normally suppress the child's `shouldStop()` check
because they buffer all
+ // output. With adaptive partial aggregation, pass-through rows are appended
to the output buffer
+ // while consuming child input, so the stop check must be re-enabled to keep
the buffer bounded.
+ override def needStopCheck: Boolean = adaptivePartialAggEnabled
Review Comment:
**Finding 10.** I reached the same conclusion independently and I don't
think this is addressed on `f2616ecb556`. The only change since your review is
the `results unchanged below a generator that cannot yield mid-fan-out` test
added in `dc2091212e1`, which asserts results, not the buffer bound.
`GenerateExec.codeGenCollection`/`codeGenIterableOnce` still emit `for (index
…) { consume(…) }` and `while (iterator.hasNext()) { consume(…) }` with no
`shouldStopCheckCode`, so once pass-through is active every generated row for
one input row is appended before the leaf's stop check can run — and because
`GenerateExec.needCopyResult = true`, each one is a fresh `UnsafeRow.copy()` on
the heap.
The part I'd emphasise over the raw count: before this PR those rows landed
in `UnsafeFixedWidthAggregationMap`, which is spillable, and now they land in
`BufferedRowIterator.currentRows`, which is not. So the failure mode is an OOM
with no fallback rather than extra spill I/O.
@ulysses-you was something else intended by "Addressed" here? If the answer
is that this is acceptable, could the comment at `:811` say so explicitly,
since it currently reads as though `needStopCheck` bounds the buffer?
--
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]
