ulysses-you commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3727551391
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -535,19 +613,30 @@ case class HashAggregateExec(
// `addNewFunction` spills this helper into a nested class (as can happen
// once the outer class passes the code-size threshold), the bare field
// reference fails with `IllegalAccessError`.
+
+ // Generate code for output. This must happen before the `doAgg` helper
below, because with
+ // adaptive partial aggregation enabled, `doConsumeWithKeys` (invoked from
the child's produce
+ // inside `doAgg`) emits pass-through rows by calling this output function
directly.
+ val keyTerm = ctx.freshName("aggKey")
+ val bufferTerm = ctx.freshName("aggBuffer")
+ outputFunc = generateResultFunction(ctx)
+
+ // After the child input is consumed, finish the build: with adaptive
partial aggregation mark
+ // that the child is fully consumed (to support re-entry; the map
iterators are set up inside
+ // the map-output function), otherwise set up the map iterators for the
output below.
+ val postChildProduce = if (adaptivePartialAggEnabled) {
+ s"$adaptiveChildrenConsumedTerm = true;"
Review Comment:
Good catch! Confirmed and fixed in dc2091212e1. `spark.range(0, 100, 1,
2).union(spark.range(100, 200, 1, 2))` grouped by `id` returned 12 of 200
groups before the fix.
--
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]
