cloud-fan commented on code in PR #59764:
URL: https://github.com/apache/spark/pull/59764#discussion_r4381035037
##########
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:
Done in 72ae198. The map setup runs once and nothing is output before it, so
it now stops the build timer unconditionally. `stopBuildTimer` is empty on the
adaptive path, and both `passThrough` guards are gone.
With the stop in the setup, where the setup runs no longer affects
`aggTime`. So I also moved the setup back inline into the output function, and
`postChildProduce` back to its form before this PR. When no row is passed
through, the setup runs from the output right after the first `doAgg`, as it
did before this PR, and the stop follows it. Only the `shouldStop()` and
`childrenConsumed` checks run in between. The timer still has one start and one
stop per task.
##########
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:
Done in 72ae198. With the single stop, the two guards are gone, and where
the setup runs no longer changes what is timed. The test now checks that the
stage makes exactly two `System.nanoTime()` calls, and that one of them is in
the generated `outputMapAndFlush` function, which holds the setup. Moving the
stop out of the setup, or adding a timer anywhere in the stage, fails it.
##########
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:
Done in 72ae198. The comment says the count is for the whole stage, whose
other operators take no time, and the clue now shows only the matching lines.
##########
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:
Done in 72ae198. The sentence now says `time in aggregation build` also
covers preparing the aggregated map for output, which in a task that has
spilled also writes the map to disk and merges it with the spilled data. The
`aggTime` comment in `HashAggregateExec` says the same.
--
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]
