peter-toth commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3767754630
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -845,29 +1229,86 @@ 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.
+ //
+ // The maps were frozen when pass-through fired, so they are drained to
preserve the merge order
+ // (trigger row, then the maps, then the rest of the input). Where the
drain runs depends on
+ // the plan shape. Split by an exchange, the partial is the whole-stage
root and its output
+ // function appends a reference to the reusable output row; draining in
the same call would
+ // overwrite that row while still buffered and silently drop the trigger,
so the drain is left
+ // to `doProduceWithKeys` on the next `processNext` (the buffered trigger
is pulled first).
+ // A one-to-many child that cannot yield mid-fan-out changes the trade:
without a stop check
+ // its whole batch is appended before the deferred drain runs, flipping
the merge order for a
+ // group that straddles the freeze point. Such children report
`needCopyResult`, so every row
+ // is a copy, the aliasing hazard is gone, and the drain runs here right
after the trigger,
+ // draining the whole frozen map before the fan-out batch continues. Fused
with the Final (no
+ // exchange), the output is consumed directly by the Final's `doConsume`
and never buffered, so
+ // there is no aliasing hazard either; the drain runs here too, because
nothing else would
+ // yield the build loop (with no append there is no `shouldStop()`), and
the maps would
+ // otherwise come out only after all the remaining streamed rows, flipping
the merge order.
+ val emitPassThroughRow = if (adaptivePartialAggEnabled) {
+ val numBypassingRows = metricTerm(ctx, "numBypassingRows")
+ val drainFused = if (isWholeStageRoot && !needCopyResult) {
+ ""
+ } else {
+ // `outputMap` returns on `shouldStop()`, already true here because
the bypassed row was
+ // just appended, so drain in a loop to emit the whole frozen map
before the fan-out batch
+ // continues. Fused, nothing is buffered and `shouldStop()` stays
false, so the loop runs
+ // once.
+ s"""
+ |while (!$adaptiveMapOutputDoneTerm) {
Review Comment:
**Finding 20.** The loop settles finding 1 and is the right shape for it.
Worth recording what it costs, though: the whole frozen map is now materialised
in `BufferedRowIterator.currentRows` before the fan-out batch continues, and
that buffer sits outside the memory manager.
`shouldStop()` is `!currentRows.isEmpty()` and the bypassed row was just
appended, so each `outputMap()` call emits one row and returns; the loop
therefore runs once per map entry, all inside this `doConsume`, and nothing can
remove from `currentRows` until it and the rest of the batch return to
`processNext`. It only arises in the whole-stage-root + `needCopyResult` shape
— an `Exchange` above the partial and a join, `Expand` or `Generate` below it —
which is most real aggregations.
Measured on this head, `Generate` child and `splits = 2`, reading the
partial's `numOutputRows - numBypassingRows`:
| `minRows` | map rows resident at once, per task |
| --- | --- |
| `100000` (default) | 99,999 |
| `200000` | 199,999 |
Linear in the knob, which follows from the predicate: the flip needs
`processedRows < keys * minCompaction`, so the map holds at least `minRows /
1.05` entries when it freezes. Each entry becomes a `row.copy()` — a `byte[]`,
an `UnsafeRow` and a `LinkedList` node, around 105 bytes for this two-column
output — so roughly 11 MB per task at the default, more as the output row
widens.
The map side is already handled and I want to be clear about that:
`UnsafeFixedWidthAggregationMap.iterator()` returns
`BytesToBytesMap.destructiveIterator()`, so the `longArray` is released on
construction and each data page is freed as the iterator moves off it. This is
not a doubling of managed memory. What it is: packed, tracked, spillable
records become ~1.2-2x as many bytes in a plain `LinkedList` that
`TaskMemoryManager` cannot see, cannot spill, and does not count against the
execution pool. The spill check is the worse end, since there the map is at the
execution-memory limit by definition, so the row count is bounded by available
memory rather than by `minRows`.
I'm not asking for a behavioural change — the merge order you settled at R4
depends on this drain, and the interpreted path has no equivalent because
`next()` hands back one row at a time, so any bound applied here alone would
put `wholeStage` back in play. What I would ask is that it be stated where
someone will find it:
- the `needStopCheck` comment at `:853-858` documents the neighbouring case,
a fan-out child's own batch accumulating in `currentRows`, which is bounded by
one input row's fan-out; the drain adds the whole map on top, and as written
the comment reads as if the fan-out width were the bound;
- the `minRows` row in `sql-performance-tuning.md` presents it purely as a
decision-frequency knob, but it also sizes this buffer.
One note against the finding-15 retention shape, which I closed as
declined-and-accepted: it avoids this by construction, because a probing map is
never output mid-build, so `currentRows` never holds more than one map row and
there is no order to establish before the batch continues. "The map is freed
early" was on the plus side of the ledger for freezing; this buffer is the
price of that early free. I'm not reopening it, but if the buffer ever turns
out to matter in practice, that is where the fix lives.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -4156,6 +4156,47 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val ADAPTIVE_PARTIAL_AGGREGATION_ENABLED =
+
buildConf("spark.sql.execution.aggregate.adaptivePartialAggregation.enabled")
+ .doc("When true, hash aggregation adaptively bypasses the pre-shuffle
partial aggregation " +
+ "at runtime when it observes that the partial aggregation is not
reducing the number of " +
+ "rows enough to be worthwhile. Once bypassed, the remaining input rows
are passed " +
+ "through as single-row partial aggregation buffers for the final
aggregation to merge, " +
+ "which avoids the cost of maintaining and spilling a large aggregation
map with little " +
+ "reduction benefit. Disabled by default. This applies only to hash
aggregation with " +
+ "grouping keys.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
Review Comment:
**Finding 21.** `ConfigBindingPolicy`'s own decision procedure
(`common/utils/src/main/scala/org/apache/spark/internal/config/ConfigBindingPolicy.scala:33-38`)
is "can the config change the result of resolving the body of a
view/UDF/procedure? If not, use `NOT_APPLICABLE`", and it spells out that
merely being *read* during resolution does not count. All three new entries
only change how an already-planned `HashAggregateExec` executes, so
`NOT_APPLICABLE` is the label; `USE_HASH_AGG` two entries up (`:4138`) is the
matching precedent. `BYPASS_PARTIAL_AGGREGATION` uses `SESSION`, but it does
not change resolution either, so it is not a precedent worth following.
No behaviour difference — `NOT_APPLICABLE` reads from the active session at
runtime — so this is metadata only. Applies to `minRows` and `minCompaction` as
well.
--
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]
