cloud-fan commented on code in PR #59764:
URL: https://github.com/apache/spark/pull/59764#discussion_r4373736017
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -200,15 +200,15 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
val counters = AggCounters(
// The metric is only registered on aggregates the feature applies to;
an aggregate without
// it bypassed nothing.
- skipped =
partialAggs.map(_.metrics.get("numBypassingRows").map(_.value).getOrElse(0L)).sum,
+ skipped =
partialAggs.map(_.metrics.get("numPassthroughRows").map(_.value).getOrElse(0L)).sum,
partialOutputRows =
partialAggs.map(_.metrics("numOutputRows").value).sum,
spillBytes = partialAggs.map(_.metrics("spillSize").value).sum,
tasksFallBacked =
partialAggs.map(_.metrics("numTasksFallBacked").value).sum)
checkAgainstReference(df, build)
counters
}
- private def numBypassingRows(build: () => DataFrame): Long =
runAndReadCounters(build).skipped
+ private def numPassthroughRows(build: () => DataFrame): Long =
runAndReadCounters(build).skipped
Review Comment:
Added "a split adaptive aggregate times only its build, never the
pass-through rows". It counts the `System.nanoTime()` calls in the partial
aggregate's generated code: one start and the two places the build can stop.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -79,7 +79,7 @@ case class HashAggregateExec(
// Only the aggregates that can actually bypass report this, so the rest
do not show a
// constant 0 in the SQL UI (see `UnionExec.metrics` for the same
approach).
if (adaptivePartialAggEnabled) {
- Map("numBypassingRows" -> SQLMetrics.createMetric(sparkContext, "number
of bypassing rows"))
+ Map("numPassthroughRows" -> SQLMetrics.createMetric(sparkContext,
"passthrough output rows"))
Review Comment:
Done. `docs/web-ui.md` lists the metric and says "time in aggregation build"
excludes the passed-through rows, and the Adaptive Partial Aggregation section
of `sql-performance-tuning.md` says the same.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -841,14 +841,14 @@ case class HashAggregateExec(
} else {
""
}
+ // The resumed build is not timed: split by an exchange, every
passed-through row returns and
+ // re-enters here, so a timer would cost two `System.nanoTime()` calls per
row while the
Review Comment:
Narrowed. The comment now says that a child that checks `shouldStop()`
between rows re-enters once per row, and a one-to-many child such as
`GenerateExec` once per input row.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -79,7 +79,7 @@ case class HashAggregateExec(
// Only the aggregates that can actually bypass report this, so the rest
do not show a
Review Comment:
Renamed this comment and the suite's names: `AggCounters.passedThrough`,
`passThroughRowsByGroupingKeyCount`, the assertion messages and the test title.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -137,7 +137,7 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
df.collect()
val skipped = collect(df.queryExecution.executedPlan) {
case agg: HashAggregateExec if isPartialPhase(agg) =>
- agg.metrics.get("numBypassingRows").map(_.value).getOrElse(0L)
+ agg.metrics.get("numPassthroughRows").map(_.value).getOrElse(0L)
Review Comment:
Added `passThroughRows(agg)` and `passThroughRows(df)` and used them at all
four sites, so the key now appears once in the suite.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -772,15 +778,21 @@ case class HashAggregateExec(
// the maps must precede the held rows so the downstream Final merges a
group's map buffer
// before its pass-through buffers (a group can straddle the freeze, since
the maps only freeze
// once pass-through fires), and the copies were made when the row was
queued, so emitting them
- // here cannot corrupt the build's reusable rows.
+ // here cannot corrupt the build's reusable rows. The iterator setup sorts
and merges any
+ // spilled data, which the non-adaptive path times as part of the build,
so it is timed here
+ // too; it runs once per task.
+ val aggTime = metricTerm(ctx, "aggTime")
adaptiveOutputMapAndFlushFuncName = if (adaptivePartialAggEnabled) {
val name = ctx.freshName("outputMapAndFlush")
val pair = ctx.freshName("pendingRow")
+ val beforeFinish = ctx.freshName("beforeFinish")
ctx.addNewFunction(name,
s"""
|private void $name() throws java.io.IOException {
| if (!$adaptiveMapSetupDoneTerm) {
+ | long $beforeFinish = System.nanoTime();
| $finishHashMap
+ | $aggTime.add((System.nanoTime() - $beforeFinish) /
$NANOS_PER_MILLIS);
Review Comment:
Thanks, fixed, with both of the parts you describe. When pass-through never
starts, the map setup now runs at the end of the build, inside the build timer,
as with the feature off. When pass-through starts, the timer stops right after
the setup. So the finish is counted once in every case, each task makes two
`System.nanoTime()` calls, and the `aggTime` term is back in its old place
(item 22).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -4685,16 +4685,36 @@ object SQLConf {
.doc("The minimum compaction ratio required to keep the pre-shuffle
partial aggregation " +
s"(see '${ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key}'). The compaction
ratio is the " +
"number of processed rows divided by the number of keys held in the
aggregation maps, " +
- "so a ratio of 10 means the partial aggregation collapses ten rows
into one. When an " +
- "evaluation finds the ratio below this value, the partial aggregation
is bypassed for " +
- "the rest of the input. A larger value bypasses more aggressively.")
+ "so a ratio of 10 means the partial aggregation collapses ten rows
into one. When the " +
+ "periodic evaluation finds the ratio below this value, the partial
aggregation is " +
+ "bypassed for the rest of the input. A larger value bypasses more
aggressively. The " +
+ "evaluation when the map is about to spill uses " +
+
"'spark.sql.execution.aggregate.adaptivePartialAggregation.minCompactionToSpill'
instead " +
+ "when that is larger.")
.version("4.4.0")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
.doubleConf
.checkValue(v => v >= 1.0 && v.isFinite,
"The minimum compaction ratio must be a finite value of at least 1.0.")
.createWithDefault(1.05)
+ val ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION_TO_SPILL =
+
buildConf("spark.sql.execution.aggregate.adaptivePartialAggregation.minCompactionToSpill")
+ .doc("The minimum compaction ratio required to keep the pre-shuffle
partial aggregation " +
+ "when its map is about to spill (see " +
+ s"'${ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key}'). Keeping the
aggregation past that " +
+ "point also pays to write the map to disk and merge it back later, so
it must collapse " +
+ "more rows into each key to be worthwhile than it must between spills.
When the ratio " +
+ "is below this value, the spill is skipped and the rest of the input
is passed " +
+ "through. The evaluation uses the larger of this value and " +
+ s"'${ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION.key}'.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .doubleConf
+ .checkValue(v => v >= 1.0 && v.isFinite,
+ "The minimum compaction ratio must be a finite value of at least 1.0.")
+ .createWithDefault(2.0)
Review Comment:
You're right: the description's model assumes no key repeats across spills.
With repeats, keeping the aggregation is cheaper whenever 1/r + 1/R < 1, so 2.0
is only the worst case. I've reverted the threshold from this PR and will take
it to its own JIRA, with a benchmark of the cases you describe and a design
that accounts for keys that repeat across spills.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -1030,7 +1040,8 @@ case class HashAggregateExec(
|if (!$adaptivePassThroughTerm) {
| $probeRegularMap
| if ($unsafeRowBuffer == null) {
- | if ($processedRowsTerm >= ${spillCheckMinRows}L &&
$adaptiveIneffective) {
+ | if ($processedRowsTerm >= ${spillCheckMinRows}L &&
Review Comment:
Agreed that passing the hot keys through moves their reduction to one
reducer per key. This moves to the new JIRA along with item 12. Its benchmark
will include a hot-key case with an `Exchange`, so the skew shows up in the
numbers.
--
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]
