cloud-fan commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3749347731
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregationIterator.scala:
##########
@@ -355,7 +417,54 @@ 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()
+
+ // 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 = {
+ val row = if (pendingPassThroughRow != null) {
+ val stashed = pendingPassThroughRow
+ pendingPassThroughRow = null
+ stashed
+ } else {
+ inputIter.next()
+ }
+ val groupingKey = groupingProjection.apply(row)
+ // Reset the buffer to initial values, then update it with this single row.
+ passThroughAggregationBuffer.copyFrom(initialAggregationBuffer)
+ processRow(passThroughAggregationBuffer, row)
+ numBypassingRows += 1
+ generateOutput(groupingKey, passThroughAggregationBuffer)
+ }
+
+ ///////////////////////////////////////////////////////////////////////////
+ // Part 7: Loads input rows and setup aggregationBufferMapIterator if we
Review Comment:
[nit] `setup` is a noun here; please use the verb phrase `sets up`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregationIterator.scala:
##########
@@ -57,9 +57,10 @@ import org.apache.spark.util.ArrayImplicits._
* - Part 3: Methods and fields used by hash-based aggregation.
* - Part 4: Methods and fields used when we switch to sort-based aggregation.
* - Part 5: Methods and fields used by sort-based aggregation.
- * - Part 6: Loads input and process input rows.
- * - Part 7: Public methods of this iterator.
- * - Part 8: A utility function used to generate a result when there is no
+ * - Part 6: Methods and fields used by adaptive partial aggregation
pass-through.
+ * - Part 7: Loads input and process input rows.
Review Comment:
[nit] Please use parallel verb agreement here: `Loads input and processes
input rows`.
--
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]
