ulysses-you commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3735264049
##########
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:
Confirmed. `doExecute` wraps only the `TungstenAggregationIterator`
constructor, and once pass-through is active most of the input is consumed
lazily from `next()` via `nextPassThroughOutput`, so that time is attributed to
nothing. The gap predates this PR for the sort-fallback path, but pass-through
widens it considerably.
Not fixed yet: the honest fix times the work where it happens -- around
`processInputs` and the pass-through output -- rather than around iterator
construction, which is a metric change with a scope of its own. Happy to do it
in this PR if you would rather not leave it, otherwise I will file it
separately.
--
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]
