dongjoon-hyun commented on code in PR #59758:
URL: https://github.com/apache/spark/pull/59758#discussion_r4361856053
##########
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
Review Comment:
**[5] The JIT-limit motivation has no numbers, and the default config bounds
it.**
`CollapseCodegenStages` keeps whole-stage codegen only when the output of an
operator has at most `spark.sql.codegen.maxFields` fields, 100 by default
(WholeStageCodegenExec.scala:1214), and a partial aggregate outputs its
grouping keys plus its buffer slots (AggUtils.scala:174-176). So by default the
projection moved here writes fewer than 100 slots, on the order of 1-1.5KB of
bytecode. That can still tip a consume method over 8000 bytes, but neither this
comment nor the PR description shows a case that does.
Minor: could the PR description give the largest method size before and
after for an example aggregation, or say that the gain matters mainly when
`maxFields` is raised?
##########
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:
**[2] The extracted method still grows with the number of buffer slots.**
`GenerateUnsafeProjection` splits the field writes of a top-level projection
only when `ctx.currentVars` is null (GenerateUnsafeProjection.scala:188-189,
"TODO: support whole stage codegen"). Here `currentVars` still holds the
consume input, so `makeEmptyAggBuffer` gets every write in one body, on the
order of 10-15 bytes of bytecode per slot. With the default
`spark.sql.codegen.maxFields` this stays small (item 5), but when that limit is
raised, a partial aggregate with several hundred buffer slots gets a
`makeEmptyAggBuffer` past the 8000-byte limit of HotSpot. That method then
stays interpreted while it runs for every bypassed row, so the hot path only
moves from one interpreted method to another.
Could item 1 address this as well? If the projection stays, could the
comment at L921-923 say that the method itself is not split?
##########
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:
**[3] Under a fused `UnionExec`, each child adds its own copy of this
method.**
`doConsumeWithKeys` runs once per `consume`, and `UnionExec.doProduce` calls
`produce(ctx, this)` once per child (basicPhysicalOperators.scala:1566-1570).
So `groupBy(...).agg(...)` over a fused union of N children generates N
identical `makeEmptyAggBuffer_<i>` methods and N `UnsafeRowWriter` states, all
counting toward the class size. The inline projection was duplicated the same
way before this PR, but now that it is a method that reads no input, one copy
would do.
Minor: could it be generated once in `doProduceWithKeys`, before `doAgg` is
generated, and kept in a field like `adaptiveOutputMapAndFlushFuncName`?
##########
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:
**[4] No test pins the new layout.**
`AdaptivePartialAggregationSuite` compares query results only, and the
results were the same before this PR, so a later change that inlines the
projection again would pass every test. The path where `addNewFunction` returns
a `nestedClassInstance.`-qualified name, which the inline code could not reach,
is not exercised by this suite either.
Minor: could a test check the generated code of a pass-through-eligible
aggregate, e.g. that it contains `makeEmptyAggBuffer`, or compare the largest
method size with the helper at WholeStageCodegenSuite.scala:1849?
##########
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:
**[1] Each bypassed row still re-projects a constant initial buffer.**
The initial values do not depend on the input row, so every call to
`makeEmptyAggBuffer()` writes the same bytes again: `rowWriter.reset()`,
`zeroOutNullBytes()`, and one `write` or `setNullAt` per buffer slot. Once
pass-through fires, this runs for every remaining row of the task. The
interpreted path already resets the same buffer with a single memory copy,
`passThroughAggregationBuffer.copyFrom(initialAggregationBuffer)`
(TungstenAggregationIterator.scala:466), and the generated code already calls
`getEmptyAggregationBuffer()` (L302) to build the fast hash map (L623).
Could the generated code do the same, e.g. keep the row from
`$thisPlan.getEmptyAggregationBuffer()` and a reusable `.copy()` of it (backed
by a byte array, as `copyFrom` requires) as two mutable states, and reset the
copy with `copyFrom` for each bypassed row? Reusing one row is as safe as
reusing the writer row today, since `handlePassThroughRow` copies the buffer
before queuing it. The code size would then no longer depend on the number of
buffer slots, which also covers item 2.
##########
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) {
Review Comment:
**[6] `makeEmptyAggBufferFunc` departs from the `*FuncName` naming of this
file.**
The value returned by `ctx.addNewFunction` can be a
`nestedClassInstance.`-qualified name, and this file names such values
`doAggFuncName` (L799) and `adaptiveOutputMapAndFlushFuncName` (L776), as
`SampleExec` and `RangeExec` do with `initSamplerFuncName` and
`initRangeFuncName`.
Nit: could it be `makeEmptyAggBufferFuncName`?
--
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]
