cloud-fan commented on code in PR #59758:
URL: https://github.com/apache/spark/pull/59758#discussion_r4366321426
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -918,11 +918,22 @@ case class HashAggregateExec(
// For adaptive partial aggregation pass-through, each bypassed row is
emitted as a single-row
// partial buffer: start from the initial aggregation buffer, apply the
update expressions once,
// and output `key ++ buffer` for the Final aggregate to merge. This
projects the initial
- // buffer.
- val emptyAggBufferCode = if (adaptivePartialAggEnabled) {
- GenerateUnsafeProjection.createCode(ctx, declFunctions.flatMap(f =>
f.initialValues))
+ // buffer, in its own method: the projection's code grows with the number
of buffer slots, and
+ // inlining it into the per-row consume method could push a wide
aggregation's consume method
+ // past the JIT's huge-method limit.
+ val makeEmptyAggBufferFunc = if (adaptivePartialAggEnabled) {
+ val emptyAggBufferCode =
Review Comment:
Done in bbf464a. The initial buffer is now built once per task, and each
passed-through row resets a reusable copy with `copyFrom`, as
`TungstenAggregationIterator` does. `makeEmptyAggBuffer` is gone.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -918,11 +918,22 @@ case class HashAggregateExec(
// For adaptive partial aggregation pass-through, each bypassed row is
emitted as a single-row
// partial buffer: start from the initial aggregation buffer, apply the
update expressions once,
// and output `key ++ buffer` for the Final aggregate to merge. This
projects the initial
- // buffer.
- val emptyAggBufferCode = if (adaptivePartialAggEnabled) {
- GenerateUnsafeProjection.createCode(ctx, declFunctions.flatMap(f =>
f.initialValues))
+ // buffer, in its own method: the projection's code grows with the number
of buffer slots, and
+ // inlining it into the per-row consume method could push a wide
aggregation's consume method
+ // past the JIT's huge-method limit.
+ val makeEmptyAggBufferFunc = if (adaptivePartialAggEnabled) {
+ val emptyAggBufferCode =
+ GenerateUnsafeProjection.createCode(ctx, declFunctions.flatMap(f =>
f.initialValues))
Review Comment:
Fixed by the change for item 1: the reset is now a single `copyFrom` call,
so none of the generated code grows with the number of buffer slots.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -918,11 +918,22 @@ case class HashAggregateExec(
// For adaptive partial aggregation pass-through, each bypassed row is
emitted as a single-row
// partial buffer: start from the initial aggregation buffer, apply the
update expressions once,
// and output `key ++ buffer` for the Final aggregate to merge. This
projects the initial
- // buffer.
- val emptyAggBufferCode = if (adaptivePartialAggEnabled) {
- GenerateUnsafeProjection.createCode(ctx, declFunctions.flatMap(f =>
f.initialValues))
+ // buffer, in its own method: the projection's code grows with the number
of buffer slots, and
+ // inlining it into the per-row consume method could push a wide
aggregation's consume method
+ // past the JIT's huge-method limit.
+ val makeEmptyAggBufferFunc = if (adaptivePartialAggEnabled) {
+ val emptyAggBufferCode =
+ GenerateUnsafeProjection.createCode(ctx, declFunctions.flatMap(f =>
f.initialValues))
+ val makeEmptyAggBuffer = ctx.freshName("makeEmptyAggBuffer")
Review Comment:
Done in bbf464a. Both buffers are registered in `doProduceWithKeys`, so the
children of a fused Union share them. Each child's consume code keeps only its
own one-line `copyFrom` call.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -918,11 +918,22 @@ case class HashAggregateExec(
// For adaptive partial aggregation pass-through, each bypassed row is
emitted as a single-row
// partial buffer: start from the initial aggregation buffer, apply the
update expressions once,
// and output `key ++ buffer` for the Final aggregate to merge. This
projects the initial
- // buffer.
- val emptyAggBufferCode = if (adaptivePartialAggEnabled) {
- GenerateUnsafeProjection.createCode(ctx, declFunctions.flatMap(f =>
f.initialValues))
+ // buffer, in its own method: the projection's code grows with the number
of buffer slots, and
+ // inlining it into the per-row consume method could push a wide
aggregation's consume method
+ // past the JIT's huge-method limit.
+ val makeEmptyAggBufferFunc = if (adaptivePartialAggEnabled) {
+ val emptyAggBufferCode =
+ GenerateUnsafeProjection.createCode(ctx, declFunctions.flatMap(f =>
f.initialValues))
+ val makeEmptyAggBuffer = ctx.freshName("makeEmptyAggBuffer")
+ ctx.addNewFunction(makeEmptyAggBuffer,
Review Comment:
Added `pass-through resets its buffer with copyFrom from buffers registered
once`. It checks the generated code of an aggregate over a fused two-child
Union: each child resets the buffer with `copyFrom`, and both use the same
buffers.
--
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]
