peter-toth commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3758034452
##########
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 `lit(0L)` repro passes now, thanks — but the residual is
wider than "the first bypassed row colliding with a key in the map". Change
that one literal in your own new test and it fails:
```scala
// AdaptivePartialAggregationSuite, "fan-out below a split aggregate
preserves the merge order"
when($"id" === 4, lit(6L)).otherwise($"id" * 2 + 1)
```
```
- fan-out below a split aggregate preserves the merge order *** FAILED ***
inputPartitions=2 wholeStage=true twoLevelMap=true forceSpill=true
```
Reading the `k = 6` row on `cb5b6a9`, `minRows = 8`, `inputPartitions = 2`,
no forced spill, both `twoLevelMap` values:
| run | (f, l) |
| --- | --- |
| feature off | (306, 406) |
| on, interpreted | (306, 406) |
| on, codegen | **(406, 306)** |
`inputPartitions = 1` fuses and is unaffected.
Mechanism: `outputMap()` returns at its first `shouldStop()`, and
`shouldStop()` is already true when `emitPassThroughRow` calls it in the
whole-stage-root shape, because the bypassed row was just appended. So the
drain emits **exactly one** map row per bypassed row, and the frozen map
interleaves with the trigger row's fan-out batch — `r1, m1, r2, m2, …`, then
the rest of the map on the following `processNext` calls, then the rest of the
input. With `explode(array(a, b))` that is two rows, so `r2` overtakes `m2 …
mN`; the window is the fan-out width, so a wider generator or a join match set
widens it. The interpreted path emits `r1`, then the whole map, then the rest —
which is also the order a non-bypassed run gives that key.
Your comment at `:1272-1276` says the drain "runs here right after the
trigger, between fan-out rows", so the interleaving is deliberate; what it
costs is the path-agreement guarantee that the suite class doc, the order
test's comment, `sql-performance-tuning.md` and the PR description all state.
Smallest fix that keeps trigger-first is to finish the drain instead of
taking one row from it:
```suggestion
val drainFused = if (isWholeStageRoot && !needCopyResult) {
""
} else {
s"""
|while (!$adaptiveMapOutputDoneTerm) {
| $adaptiveOutputMapFuncName();
|}
""".stripMargin
}
```
The maps are frozen and both probes are gated off by then, so nothing else
touches them; fused, the first call already sets the flag so the loop runs
once; in the `needCopyResult` branch every appended row is a copy, so the
aliasing hazard stays gone. `limitNotReachedCond` is empty for a blocking
operator, so the loop always terminates. The cost is the frozen map sitting in
`currentRows` at once — the same non-spillable buffering you already document
for fan-out under pass-through.
If you would rather not buffer the map, the other option is to keep the
partial drain and drop the agreement claim from all four places. That makes
`spark.sql.codegen.wholeStage` observable on `first`/`last`, and it flips on
its own when generated code exceeds the compile limits, so I would not pick
that one.
And whichever way it goes, please make the regression test pin a collision
with a key that is *not* the first one inserted — `lit(0L)` passes only because
the map drains in insertion order and `k = 0` is drained first.
##########
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.** No need to put the sentence back — with `enabled` defaulting
to `false` the config table carries the opt-out, and the two new paragraphs say
what a reader needs. Resolved from my side.
--
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]
