ulysses-you commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3763419013
##########
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:
Fixed in 77a8aad0997. The drain is now a loop, so the whole frozen map comes
out right after the trigger row, before the fan-out batch continues (previously
one map row per bypassed row). The lit(6L) reproduction passes. Two new tests
cover the wider shape: "a wide fan-out drains the whole frozen map before
streaming the batch" (batch four rows wide, two collisions at different drain
positions), and "a frozen map larger than the fan-out batch preserves the merge
order" (minRows=32; compares the generated and interpreted paths per cell,
because in the forced-fallback cells the colliding key rows straddle the Final
own spill). Full suite: 51 tests pass.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregationIterator.scala:
##########
@@ -355,7 +420,61 @@ class TungstenAggregationIterator(
}
///////////////////////////////////////////////////////////////////////////
- // Part 6: Loads input rows and setup aggregationBufferMapIterator if we
+ // Part 6: 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. The output order matches the
generated path: the first
+ // pass-through row is emitted before the frozen map (or sort-based) output,
and the remaining
+ // rows stream afterwards.
+ 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
+
+ // Whether the first pass-through row has not been emitted yet. It is
emitted before the frozen
+ // map (or sort) output -- mirroring the generated path, where the first
bypassed row is appended
+ // to the output buffer from inside the build loop and the maps only drain
afterwards.
+ private[this] var passThroughTriggerPending: Boolean = false
+
+ // 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()
Review Comment:
Fixed in b039db81c79. The createNewAggregationBuffer comment now lists
passThroughAggregationBuffer as the third caller.
--
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]
