dongjoon-hyun commented on code in PR #59764:
URL: https://github.com/apache/spark/pull/59764#discussion_r4379787809
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -811,7 +839,20 @@ case class HashAggregateExec(
}
val aggTime = metricTerm(ctx, "aggTime")
- val beforeAgg = ctx.freshName("beforeAgg")
+ // The build timer. With adaptive partial aggregation, once pass-through
has started the map
+ // setup stops it instead (see `aggTime` above), so its start is kept in a
field.
+ val (startBuildTimer, stopBuildTimer) = if (adaptivePartialAggEnabled) {
Review Comment:
25. Minor: since the setup always runs exactly once inside the first
`doAgg`, the two places that can stop the build timer always record what one
stop there would. `setupMapOutput` adds the time when `passThrough` is true
(L702-704), and `stopBuildTimer` adds it after the first `doAgg` returns when
`passThrough` is false (L916). The aggregate appends nothing before the first
passed-through row, so the first `doAgg` either runs the setup for that row
(L1368, then L813) or reaches L722 with `shouldStop()` false. When no row is
passed through, the setup call at L722 is the last statement of `doAgg`, so
nothing runs between it and L916.
Could you drop `stopBuildTimer` on the adaptive path, together with the `if
($adaptivePassThroughTerm)` around L703? The timer then has one start and one
stop, the two guards on `passThrough` that have to agree go away, and the test
from item 6 can check where the time is taken (item 26).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregationIterator.scala:
##########
@@ -551,16 +541,11 @@ class TungstenAggregationIterator(
}
} else {
// Adaptive partial aggregation bypassed partial aggregation: emit the
remaining input
- // rows as single-row partial buffers, after the frozen map output
above.
- if (passThroughDrainStart == -1L) {
- passThroughDrainStart = System.nanoTime()
- }
- val out = nextPassThroughOutput()
- if (!passThroughHasNext) {
- aggTime += NANOSECONDS.toMillis(System.nanoTime() -
passThroughDrainStart)
- passThroughDrainStart = -1L
- }
- out
+ // rows as single-row partial buffers, after the frozen map output
above. The drain is
Review Comment:
30. Nit: this is about the PR description. "Both cost two
`System.nanoTime()` calls per row" holds for the codegen timer around each
resumed build, but not for the interpreted drain timer that dd0f15465e9 added.
That timer started on the first passed-through row and stopped once
`passThroughHasNext` was false, so it made two calls per task. What was wrong
with it is what the next sentence of the description and this comment say:
`next()` is pulled lazily, so it also counted the downstream work between the
calls. Since the description becomes the commit message, could you limit the
per-row cost to the codegen timer in that sentence?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -1375,16 +1382,46 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
.groupBy($"k")
.agg(sum($"v") as "s")
withClue(clue) {
- assert(numBypassingRows(df) == 0,
+ assert(numPassthroughRows(df) == 0,
"no bypass expected before `minRows` rows have been processed")
}
}
}
- test("partial aggregate output row count reflects the bypass (independent of
the skip metric)") {
+ test("a split adaptive aggregate times only its build, never the
pass-through rows") {
+ // Split by an exchange, a passed-through row returns from the build and
re-enters it, so a
+ // timer around the resumed build or around each passed-through row would
cost two
+ // `System.nanoTime()` calls per row. The partial aggregate's generated
code must hold only the
+ // build timer: its start, its stop once the build returns, and its stop
at the map setup where
+ // the first passed-through row ends the build.
+ withSQLConf(
+ (Seq(
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "true",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS.key -> "8") ++
fixedPlanConfs): _*) {
+ val df = spark.range(0, 400, 1, 2)
+ .select($"id".cast("string") as "k", $"id" as "v")
+ .groupBy($"k")
+ .agg(sum($"v") as "s")
+ df.collect()
+ assert(passThroughRows(df) > 0, "the high-cardinality input should be
passed through")
+ val partialStages = collect(df.queryExecution.executedPlan) {
+ case stage: WholeStageCodegenExec if (stage.child match {
+ case agg: HashAggregateExec => isPartialPhase(agg)
+ case _ => false
+ }) => stage
+ }
+ assert(partialStages.length == 1, "expected the partial aggregate in a
stage of its own")
+ val code = partialStages.head.doCodeGen()._2.body
+ assert("System\\.nanoTime\\(\\)".r.findAllMatchIn(code).length == 3,
code)
Review Comment:
27. Nit: the count covers the generated code of the whole stage, including
`Range`, `Project` and the fast hash map class, so a timing metric added to any
of them would fail this test although the aggregate did not change. The clue is
the whole generated source. Could you say in the comment above that the count
is for the stage, and pass only the matching lines as the clue, e.g.
`code.split("\n").filter(_.contains("System.nanoTime()")).mkString("\n")`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -680,16 +682,44 @@ case class HashAggregateExec(
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.
+ // `aggTime` times the build up to the point where pass-through starts,
plus the map iterator
+ // setup (`finishHashMap`, which sorts and merges any spilled data), but
never the pass-through
Review Comment:
29. Nit: this comment and the one at L800-801 name two cases: the setup runs
at the end of the input if pass-through never starts, and for the first
passed-through row otherwise. There is a third. When the periodic check flips
pass-through on the last input row, no row is passed through, so the setup runs
at the end of the input (L722) with `passThrough` true, and the time is added
there rather than at L916. The result is right, but a reader who takes
`passThrough` being true at L702 to mean that a row reached `outputMapAndFlush`
would be wrong. With the single stop from item 25, these comments only need to
say that the setup runs once, inside the first `doAgg`.
##########
docs/sql-performance-tuning.md:
##########
@@ -200,6 +200,9 @@ queue behind it and flushed only after it drains, so a
duplicate of a key alread
merges after that key's accumulated rows, keeping an order-sensitive aggregate
such as
`first`/`last` consistent with a run that never bypasses. The ratio is
evaluated periodically, and
again right before the aggregation map would spill, in which case the spill is
skipped entirely.
+The SQL UI counts the passed-through rows in `number of pass-through output
rows`; they are not
Review Comment:
31. Nit: this sentence sits between "...again right before the aggregation
map would spill, in which case the spill is skipped entirely" and "Because
pass-through cannot be reversed, the evaluation before a spill also waits...",
which continues about that same evaluation. Could you move it after "...instead
of committing the task on a small sample.", so the description of the two
evaluations stays together?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -1375,16 +1382,46 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
.groupBy($"k")
.agg(sum($"v") as "s")
withClue(clue) {
- assert(numBypassingRows(df) == 0,
+ assert(numPassthroughRows(df) == 0,
"no bypass expected before `minRows` rows have been processed")
}
}
}
- test("partial aggregate output row count reflects the bypass (independent of
the skip metric)") {
+ test("a split adaptive aggregate times only its build, never the
pass-through rows") {
Review Comment:
26. Minor: this pins the split shape that item 6 asked about, but not the
two fixes made since: the stop in `setupMapOutput` for the fused shape (item
2), and the setup running at the end of the build, inside the timer, when
pass-through never starts (your fix for item 11). Each of these changes keeps
three `System.nanoTime()` calls, so this test still passes:
- Inverting `if ($adaptivePassThroughTerm)` at HashAggregateExec.scala L702
drops the build time whenever pass-through starts, and counts it twice
otherwise.
- Removing the `!passThrough` guard at L847 adds the build time a second
time whenever pass-through starts, which in the fused shape brings item 2 back.
- Removing the `setupMapOutput()` call from `postChildProduce` (L722) moves
the setup out of every timer again when pass-through never starts.
The generated code does not depend on the data, so a count cannot tell these
apart. With the single stop from item 25, checking that the generated
`setupMapOutput` function holds the stop, and that the build calls it after the
child's loop, would pin both fixes.
##########
docs/web-ui.md:
##########
@@ -379,7 +379,8 @@ Here is the list of SQL metrics:
<tr><td> <code>sort time</code> </td><td> the time spent on sorting </td><td>
Sort </td></tr>
<tr><td> <code>peak memory</code> </td><td> the peak memory usage in the
operator </td><td> Sort, HashAggregate </td></tr>
<tr><td> <code>spill size</code> </td><td> number of bytes spilled to disk
from memory in the operator </td><td> Sort, HashAggregate </td></tr>
-<tr><td> <code>time in aggregation build</code> </td><td> the time spent on
aggregation </td><td> HashAggregate, ObjectHashAggregate </td></tr>
+<tr><td> <code>time in aggregation build</code> </td><td> the time spent on
aggregation. For a HashAggregate that passes rows through (see adaptive partial
aggregation), it excludes the passed-through rows </td><td> HashAggregate,
ObjectHashAggregate </td></tr>
Review Comment:
32. Nit: "(see adaptive partial aggregation)" names a section without
linking it, and the new row below it does not point anywhere either. This table
is raw HTML, so a Markdown link would not render here, but `<a
href="sql-performance-tuning.html#adaptive-partial-aggregation">adaptive
partial aggregation</a>` would, as in the table at `rdd-programming-guide.md`
L1123. Could you link the section from both rows?
##########
docs/sql-performance-tuning.md:
##########
@@ -200,6 +200,9 @@ queue behind it and flushed only after it drains, so a
duplicate of a key alread
merges after that key's accumulated rows, keeping an order-sensitive aggregate
such as
`first`/`last` consistent with a run that never bypasses. The ratio is
evaluated periodically, and
again right before the aggregation map would spill, in which case the spill is
skipped entirely.
+The SQL UI counts the passed-through rows in `number of pass-through output
rows`; they are not
+timed, so `time in aggregation build` covers only the aggregation up to the
point where
Review Comment:
28. Minor: "covers only the aggregation up to the point where pass-through
starts" leaves out the map setup, which your reply to item 1 and the comment at
HashAggregateExec.scala L685-686 include. After an earlier spill, the setup
also writes the frozen map to disk: `finishAggregate` calls
`destructAndCreateExternalSorter()` (HashAggregateExec.scala L346), which
spills it (UnsafeExternalSorter.java L131), as item 18 described, and
TungstenAggregationIterator.scala L283-289 does the same on the interpreted
path. With "only", a reader cannot account for that part of the metric. Could
you add the setup here, e.g. "covers the aggregation up to the point where
pass-through starts, and setting up the output of the aggregated map"?
--
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]
