peter-toth commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3735867587
##########
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 eight cells do agree now, but they are eight variations
of one plan shape. `spark.range(0, 9, 1, 1)` gives a single input partition, so
`SinglePartition` satisfies the Final's `ClusteredDistribution`, no `Exchange`
is inserted, and `Partial` and `Final` fuse into one `WholeStageCodegen`
(finding 12). In that shape the partial aggregate's `outputFunc` feeds the
Final's `doConsume` — a map insert — so nothing ever reaches
`BufferedRowIterator.currentRows`, `shouldStop()` stays false for the whole
build, `doAgg` consumes the partition in one call, and `outputMap()` is only
reached from `adaptiveFinalOutput`, i.e. after every pass-through row. That is
why the generated path looks like "maps drain afterwards" there.
With an `Exchange` it does not. `shouldStop()` becomes true as soon as the
first bypassed row is appended, `doAgg` returns, and `adaptiveResumeBuild`
drains the map on the next `processNext()` — so the generated order is `[first
bypassed row] [map] [rest]`, i.e. map-first for every group except the one that
first bypassed row belongs to.
Measured on both heads, `minRows = 8`, `enableTwoLevelAggMap = false`,
everything else default, reading the `k = -1` row:
```scala
spark.range(0, 40, 1, splits)
.select(when($"id" === 0 || $"id" === dupAt, lit(-1L)).otherwise($"id") as
"k", $"id" as "v")
.groupBy($"k").agg(first($"v") as "f", last($"v") as "l")
```
| head | plan shape | colliding row | reference `(f,l)` | generated |
interpreted |
| --- | --- | --- | --- | --- | --- |
| `f2616ecb556` | no `Exchange` | `id=8`, first bypassed | `(0,8)` | `(8,0)`
| `(0,8)` |
| `f2616ecb556` | no `Exchange` | `id=9`, later | `(0,9)` | `(9,0)` |
`(0,9)` |
| `f2616ecb556` | `Exchange` | `id=8`, first bypassed | `(0,8)` | `(8,0)` |
`(0,8)` |
| `f2616ecb556` | `Exchange` | `id=9`, later | `(0,9)` | `(0,9)` | `(0,9)` |
| `87937dbd6d0` | no `Exchange` | `id=8`, first bypassed | `(0,8)` | `(8,0)`
| `(8,0)` |
| `87937dbd6d0` | no `Exchange` | `id=9`, later | `(0,9)` | `(9,0)` |
`(9,0)` |
| `87937dbd6d0` | `Exchange` | `id=8`, first bypassed | `(0,8)` | `(8,0)` |
`(8,0)` |
| `87937dbd6d0` | `Exchange` | `id=9`, later | `(0,9)` | **`(0,9)`** |
**`(9,0)`** |
`numBypassingRows` is 32 (`splits=1`) or 24 (`splits=2`) in every cell, so
the bypass fires throughout.
Two things follow. The paths agreed in one of these four cases before and in
three now, but the case they now disagree in is the one with an `Exchange` and
a collision that is not adjacent to the flip — the shape the feature is for.
And the interpreted order you replaced matched a non-bypassed run in **all
four** cases, while the generated order matches in one; after the change the
interpreted order matches in none. So the target to align on is the order the
interpreted path already had, not the generated one — and it is the only order
that can match, because a group's map buffer holds its *earlier* rows, so its
streamed rows have to merge after it.
On why the other direction did not work — I do not think either obstacle is
load-bearing:
- **No `return` from `doConsume` is needed.** Call `outputMap()` inline in
`emitPassThroughRow`, before `outputFunc(...)`, and drop the `if (shouldStop())
return;` for that one call. The map is already frozen at that point and both
probes are skipped once `adaptivePassThroughTerm` is set (`:1003-1010`,
`:1046`), so nothing touches it afterwards, and the child's loop and resume
state are untouched because control never leaves `doConsume`.
- **"Several rows in one `doConsume` call" is not a contract the buffer
relies on.** `GenerateExec.codeGenCollection`/`codeGenIterableOnce` append an
entire collection per `doConsume` with no stop check — that is finding 10. The
cost of draining inline is a one-off `currentRows` spike bounded by the size of
the map you were about to free.
The dropped rows under a fused `UnionExec` sound like the bug `dc2091212e1`
fixed rather than a consequence of draining, but if it reproduces with
`postChildProduce` in place I would like to see it.
Failing that, finding 15 removes the choice entirely: if a group is never
split between the map and the stream, there is no merge order to pick.
##########
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.** Re-measured on `87937dbd6d0` — unchanged, 100,000 partial
output rows with the feature off against 500,000 with it on, `numBypassingRows`
400,000, all settings at their defaults.
On your question: of the two levers, I would take the first (scale the first
check point to what the task has seen) as the near-term answer and not the
second. Un-bypassing needs the map back, and the honest version of that is not
"rebuild" — it is a fresh epoch, exactly like the one `processInputs` already
starts after a spill, which means the same machinery twice and a
`mapOutputDone` flag that has to become repeatable in codegen. That is a lot of
moving parts for a decision that is still a bet on one window.
What I would rather compare both against is finding 15: stop *growing* the
map at the flip instead of abandoning it, and stream only rows whose key the
map does not already hold. On this exact input the 100 tail keys are already
resident when the flip fires, so they keep aggregating and the output goes back
to ~100k rows. But I want to be straight that it bounds this rather than fixing
it — invert the input to 100k distinct keys followed by 400k rows over 100
*brand-new* keys and all 400k still stream. Its real appeal is finding 1: a
group is never split between the map and the stream, so there is no merge order
to pick. And its real cost is that it keeps the per-row probe, which your own
no-spill benchmark says is worth 1.7x on its own (473.8 -> 283.8 ns/row with no
spill anywhere), so it would need re-benchmarking before anyone commits to it.
Thanks for the `adaptiveNextCheckRow *= 2` pointer — I had not looked at
`a62ca30775c`. Having read it, I do not think dropping it cost you anything
here, and it is worth saying why so it does not get proposed back as the fix.
The doubling only made *later* checks rarer; the first one still landed at
`sampleRows`, and with a one-way flip that first check is the one that commits
the task, so the exposure was identical. Its stated purpose was cost control
("a low-cardinality input is re-evaluated only rarely"), but the per-row work
is the counter increment, which doubling does not remove — it saves one
comparison per 100k rows, against the ~1.5 ns/row the low-cardinality benchmark
attributes to the whole check. And because a later check can only ever *start*
a bypass, never undo one, rarer later checks made finding 6 slightly worse. So
cloud-fan's simplification looks right on the merits as well as on complexity.
Whichever lever lands, while the flip stays one-way I would default
`spark.sql.execution.aggregate.adaptivePartialAggregation.enabled` to `false`
for one release, since the downside here is unbounded while the upside is
bounded by the cost of maintaining the map.
##########
docs/sql-performance-tuning.md:
##########
@@ -181,6 +181,58 @@ Missing or inaccurate statistics will hinder Spark's
ability to select an optima
- **Query plan estimates**: You can inspect Spark's cost estimates in the
optimized query plan via [`EXPLAIN COST`](sql-ref-syntax-qry-explain.html) or
`DataFrame.explain(mode="cost")`.
- **Runtime statistics**: You can inspect these statistics in the [SQL
UI](web-ui.html#sql-tab) under the "Details" section as a query is running.
Look for `Statistics(..., isRuntime=true)` in the plan.
+## Optimizing the Aggregate
+
+### Adaptive Partial Aggregation
+
+A grouping aggregation normally runs in two phases: a partial aggregation
before the shuffle and a
+final aggregation after it. The partial aggregation is only worthwhile when it
actually reduces the
+number of rows; when the grouping keys are close to unique it maintains -- and
possibly spills -- an
+aggregation map roughly as large as its input while emitting almost as many
rows as it consumed.
+
+When adaptive partial aggregation is enabled, hash aggregation measures the
compaction ratio (the
+number of processed rows divided by the number of keys held in its aggregation
maps) at runtime
+and, if the partial aggregation is not collapsing enough rows to be
worthwhile, stops populating
+the aggregation map and passes the remaining rows through as single-row
partial aggregation buffers
+for the final aggregation to merge. Query results are unchanged. The ratio is
evaluated
Review Comment:
**Finding 6.** Agreed on the sequencing — settle the policy under finding 2
and let this wording follow, rather than rewording now and changing behaviour
after.
One note on the tension you describe: your option 2 and finding 2 stop
pulling against each other under finding 15. If the flip only stops the map
*growing*, an input that turns high-cardinality late is handled because its new
keys stream, so "a query that only becomes ineffective later in its input is
still caught" becomes true as written without making any single window
decisive. That is the one part of finding 15 I would call a clean fix rather
than a trade. If the policy lands on either of the levers instead, option 1 is
the right call and I would narrow the sentence to the spill check.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregationIterator.scala:
##########
@@ -354,6 +410,47 @@ class TungstenAggregationIterator(
}
}
+ ///////////////////////////////////////////////////////////////////////////
+ // Part 5b: Methods and fields used by adaptive partial aggregation
pass-through.
+ ///////////////////////////////////////////////////////////////////////////
+
+ // Indicates that partial aggregation has been bypassed and the remaining
input rows should be
+ // passed through as single-row partial buffers. Set in `processInputs` by
either check point.
+ // It may coexist with earlier spills, so the output order is: sort-based
(or map) output first,
+ // then the pass-through rows.
+ private[this] var passThrough: Boolean = false
+
+ // The row that could not be inserted at the spill check. It is stashed here
(as
+ // a copy) so it becomes the first pass-through row rather than being lost.
+ private[this] var pendingPassThroughRow: InternalRow = null
+
+ // A reused aggregation buffer for building single-row partial buffers
during pass-through. It is
+ // re-initialized from `initialAggregationBuffer` for every passed-through
row.
+ private[this] lazy val passThroughAggregationBuffer: UnsafeRow =
createNewAggregationBuffer()
+
+ // Whether there are remaining pass-through rows to emit.
+ private def passThroughHasNext: Boolean =
+ passThrough && (pendingPassThroughRow != null || inputIter.hasNext)
+
+ // Emits the next input row as a single-row partial aggregation buffer, i.e.
a group of size one.
+ // The output (grouping key ++ buffer) is a valid partial buffer that the
downstream Final
+ // aggregation merges, so the result is identical to running partial
aggregation on this row.
+ private def nextPassThroughOutput(): UnsafeRow = {
Review Comment:
**Finding 7.** A separate PR is fine by me. Timing the work where it happens
changes what `aggTime` means for the sort-fallback path too, which predates
this PR, so it deserves its own change rather than riding along here — and the
codegen half is already correct at this head, so the gap is confined to the
interpreted path.
##########
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.** The comment at `:827-838` reads correctly now, thanks — and
`GenerateExec` has no `shouldStop()` anywhere in the file, so what it says is
accurate.
On the `buffered=1` measurement: I think it read the wrong buffer. If the
probe used a single-partition input (the way the suite's generator test does),
the plan has no `Exchange` — `SinglePartition` satisfies the Final's
`ClusteredDistribution`, so `Partial` and `Final` fuse into one
`WholeStageCodegen`, which I measured on this head (finding 12). In that plan
the partial aggregate's `outputFunc` feeds the Final's `doConsume`, i.e. a
hash-map insert, so no bypassed row ever enters
`BufferedRowIterator.currentRows`; the `currentRows` the probe samples belongs
to the stage `WholeStageCodegenExec` terminates, which there is the Final
aggregate. `buffered=1` is the Final's own output row, and `shouldStop()` is
false throughout the build — which is also why `bypassed=998` and `buffered=1`
can hold at once.
Re-running it with two or more input partitions should give the `Exchange`,
and then `currentRows` is the partial aggregate's own buffer. That is also the
only shape where the bound you documented can be exceeded, so it is where the
number is meaningful.
Either way I am fine with the outcome — documenting it is a reasonable call
for a Non-blocking item, and it now says the failure mode is heap rather than
spill I/O.
--
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]
