dongjoon-hyun commented on code in PR #59764:
URL: https://github.com/apache/spark/pull/59764#discussion_r4371926326
##########
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:
**13. With 2.0, a spill-check flip also passes through the hot keys that the
frozen map already holds, so their reduction moves to one reducer per hot
key.** Once pass-through fires, both probes are skipped (L1040, L1080-1085, and
the loop exit at TungstenAggregationIterator.scala:235), so every later row of
a hot key is shuffled on its own. With a few hot keys holding a fraction h of
the rows and a nearly unique tail, which is the motivating case in the
description, the ratio at a fill is about 1/(1 - h). The old threshold passed
such a task through only when h < 4.8%; 2.0 does so for any h < 50%.
For example, with 10 hot keys holding 30% of 20M rows per task, 500 tasks
and 200 shuffle partitions, each task now shuffles about 5.6M hot-key rows, and
the reducer of each hot key receives about 280M of them against about 35M rows
for a normal partition, a 9x straggler (about 150x for a single hot key at h =
0.45). Before this PR, the same task kept aggregating and sent one row per hot
key. AQE splits skewed partitions only for joins and rebalances
(`OptimizeSkewedJoin`, `OptimizeSkewInRebalancePartitions`), not for aggregates.
Could you state this trade-off in the description and the docs, or, after a
spill-check flip, keep aggregating the rows whose key the frozen map already
holds (a probe-only variant of peter-toth's [Finding
15](https://github.com/apache/spark/pull/57742#discussion_r3735865431) in
#57742)?
##########
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:
**11. This timer adds the map finish to `aggTime` a second time whenever a
row is passed through.** The guarded setup runs on the first call of
`outputMapAndFlush`. When pass-through emits any row, that call comes from
`handlePassThroughRow` (L1337), inside the consume code of the first `doAgg`.
Nothing is appended before the first passed-through row, so `shouldStop()`
stays false and that `doAgg` is still running under the `beforeAgg` timer
(L883-885). The finish is then added here and again at L885, in both the split
and the fused shape. Only when no row is passed through does the setup run from
`adaptiveFinalOutput` (L869-870), outside the build timer, so the comment at
L781-783 and the description's "outside every timer" describe that case only.
At the merge-base the pass-through case was already counted once, so this
commit fixes the no-pass-through case (0 to 1) but doubles the pass-through
case (1 to 2).
The doubled part is large after an earlier spill: `finishAggregate`
(L348-352) then sorts the whole frozen map, writes it to disk and opens the
merge, and the new default makes pass-through after a spill more common. The
interpreted path counts the same work once
(TungstenAggregationIterator.scala:287-293, in the constructor timed at
L114/L151).
The finish needs exactly one timer. If item 2 is resolved by stopping the
outer timer when pass-through fires, this timer should stay. Otherwise, the
guarded setup could run inside `doAgg`, in `postChildProduce` (L697-698) as the
non-adaptive path does at L700, and this timer could go; that also keeps a
single `System.nanoTime()` pair (item 6) and lets L784 move back (item 22).
Could you pick one, and update the comment at L781-783 and the description to
match?
##########
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 =
Review Comment:
14. Minor: with the second commit, this PR adds a new public config with its
own default, and the title changed from the rename and timing fix it started as
(d20fe6c). AGENTS.md asks to use `[FOLLOWUP]` "only for small PRs that directly
modify or correct unreleased earlier PRs", and to create a separate JIRA ticket
"for separately planned work, non-trivial changes, or work outside the earlier
JIRA's scope". The config also brings back the separate spill threshold that
#57742 removed after you asked for one policy at both check points, since
separate thresholds add complexity "without showing that these dimensions must
be independently tunable"
([discussion_r3711016966](https://github.com/apache/spark/pull/57742#discussion_r3711016966)).
Could this config get its own JIRA, and could the description show why the two
thresholds now need to be tuned separately?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -1199,6 +1205,47 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
}
}
+ test("the spill check passes through an epoch that compacts too little to
spill") {
+ // Keys `floor(2 * id / 3)` collapse every three rows into two keys, a
steady ratio of about
+ // 1.5. That is above `minCompaction` (1.05), so the periodic checks at 10
and 20 rows keep
+ // aggregating. The forced fallback fills the map when id 30 arrives,
after 30 rows over 20
+ // keys: keeping the aggregation would pay to spill the map and merge it
back for a ratio of
+ // only 1.5, so the default `minCompactionToSpill` (2.0) passes the rest
through instead, and
+ // the sorter is never created. Lowering it to 1.4 keeps the old behavior:
every epoch spills,
+ // and nothing is passed through.
+ Seq(None -> true, Some("1.4") -> false).foreach { case
(minCompactionToSpill, shouldBypass) =>
+ Seq(true, false).foreach { wholeStage =>
+ withSQLConf(
+ (Seq(
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "true",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> wholeStage.toString,
+ SQLConf.ENABLE_TWOLEVEL_AGG_MAP.key -> "false",
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS.key -> "10",
+ "spark.sql.TungstenAggregate.testFallbackStartsAt" -> "1, 30") ++
+ minCompactionToSpill.map(
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION_TO_SPILL.key
-> _) ++
+ fixedPlanConfs): _*) {
+ val df = () => spark.range(0, 400, 1, 1)
+ .select(($"id" * 2 / 3).cast("long").cast("string") as "k", $"id"
as "v")
+ .groupBy($"k")
+ .agg(sum($"v") as "s")
+ withClue(s"minCompactionToSpill=$minCompactionToSpill
wholeStage=$wholeStage") {
+ val c = runAndReadCounters(df)
+ if (shouldBypass) {
+ assert(c.skipped == 400 - 30,
+ "the spill check should pass through from the row that found
the map full")
+ assert(c.tasksFallBacked == 0,
+ "the spill check must replace the sort fallback, not trigger
it")
+ } else {
+ assert(c.skipped == 0, "an epoch above minCompactionToSpill must
keep aggregating")
+ assert(c.tasksFallBacked > 0, "an epoch that keeps aggregating
must spill")
+ }
+ }
+ }
+ }
+ }
+ }
+
test("a new in-memory map epoch after a spill can still bypass") {
Review Comment:
15. Minor: under the new default, this test no longer covers what its
comment describes. In the interpreted path and the regular-map-only codegen
cell, the third epoch (ids 80-119) has 20 repeating rows and 20 distinct ones,
40 rows over 25 keys = 1.6. Before this PR that epoch spilled at id 120, and
the next, fully distinct epoch flipped at its first periodic check on id 127,
so 272 rows were passed through after three spills. Now the spill check at id
120 passes through (40 < 25 * 2.0), so 280 rows are passed through after two
spills (268 vs 276 in the two-level codegen cell). Both satisfy
`tasksFallBacked > 0` and `skipped > 0`, and no other test still has a
periodic-check flip in an epoch after a spill. Could you move the boundary to
`id < 80`, so that every cell flips at a periodic check after two spills under
either threshold (312 rows, 308 in the two-level codegen cell), and assert that
exact count? That would also pin the `nextCheckRow` reset
(HashAggregateExec.scala:1032,
TungstenAggregationIterator.scala:263), which even the old assertions did not.
##########
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 " +
Review Comment:
19. Minor: this is about docs/sql-performance-tuning.md L249-252, which is
outside the diff. The `minCompaction` row there still says "when an evaluation
finds the ratio below this value the partial aggregation is bypassed for the
rest of the input", while this doc now limits the config to the periodic
evaluation and points to `minCompactionToSpill` for the one before a spill.
Both texts are published (this one through the generated SQL config table in
configuration.md), so the same config is now described in two ways, and the row
does not say that `minCompaction` alone no longer governs the spill check. For
example, `minCompaction=1.0` used to turn off both evaluations, because `rows <
keys * 1.0` never holds, while now the spill check still passes through below
2.0. Could the docs row use the same wording as this doc?
##########
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 " +
Review Comment:
18. Minor: "the spill is skipped" holds only until the task first spills.
After an earlier spill, passing through still writes the frozen map to disk and
merges it back: `finishAggregate` runs
`sorter.merge(hashMap.destructAndCreateExternalSorter())`
(HashAggregateExec.scala:349-350; TungstenAggregationIterator.scala:287-293 on
the interpreted path), which spills the map in
`UnsafeExternalSorter.createWithExistingInMemorySorter` and counts a sort
fallback. The test at AdaptivePartialAggregationSuite.scala:1249 asserts
`tasksFallBacked > 0` for exactly this case. From then on the threshold only
decides the later epochs, and the reason given here, that keeping "also pays to
write the map to disk and merge it back later", applies to the current map only
before the first spill. A map that fills before `minRows` rows spills without
an evaluation, so the first evaluated spill check may already have a sorter.
The same claim appears in sql-performance-tuning.md L202-204 and L261-263, Hash
AggregateExec.scala:213-214 and 963-964,
TungstenAggregationIterator.scala:197-199 and the `AggCounters` doc
(AdaptivePartialAggregationSuite.scala:169-172). "Pays to write" can also read
as "it is worthwhile to write"; the description's "also means writing the map
to disk" is clearer. Could you qualify these sentences for a task that has
already spilled?
##########
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:
**12. The cost model behind the 2.0 default assumes that spill epochs share
no keys.** The description argues that at a ratio r, keeping the aggregation
saves shuffling 1 - 1/r of the rows and spills about 1/r of them, so the two
balance near r = 2. But the sort fallback merges each key across all epochs at
the end of the task (HashAggregateExec.scala:348-405), so what keeping shuffles
is the task's distinct key count, N/R with R = N / distinct keys, not N/r.
Keeping then costs about N/r + N/R rows against N for pass-through, and it is
cheaper whenever r > R/(R-1). That threshold is 2 only when R = r, that is,
when no key repeats across epochs.
For example, take uniformly random keys with 600K distinct keys per task,
room for 400K keys in the map, and 6M rows. The periodic check sees 1.09 at
100K rows and keeps aggregating, and the map first fills after about 659K rows
at r = 1.65, while R = 10. Before this PR the task spills 9 times (about 3.66M
rows written) and shuffles 0.60M rows. With this PR it passes through at the
first fill and shuffles about 5.74M rows: 9.6x the shuffle, about 35% more rows
written in total, and 9.6x the hashing in the Final. For uniform keys the
decision flips whenever the map holds roughly 9% to 80% of the task's distinct
keys. The boundary test's own input (`id % 40`, r = 1.25, R = 10) is such a
case, while the new test's `floor(2 * id / 3)` has disjoint epochs, the one
shape where the model holds.
Could you keep the spill-time default equal to `minCompaction`, so that 2.0
is opt-in, or base the decision on an estimate of the key overlap across
epochs, e.g. a sketch of the keys filled while spilling?
##########
docs/sql-performance-tuning.md:
##########
@@ -251,6 +253,18 @@ start later.
</td>
<td>4.4.0</td>
</tr>
+ <tr>
+
<td><code>spark.sql.execution.aggregate.adaptivePartialAggregation.minCompactionToSpill</code></td>
+ <td>2.0</td>
Review Comment:
17. Minor: nothing measures this default. The spill-check cases of
AdaptivePartialAggregationBenchmark.scala:113-137 compact at r = 1.0 and r ~
1049, which decide the same way at 1.05 and at 2.0. The harness also runs
`local[1]` with one shuffle partition (SqlBasedBenchmark.scala:42-44), so a
single-slice `spark.range(N)` fuses Partial and Final with no `Exchange`, and
the shuffle cost that 2.0 weighs against spilling is not part of any
measurement. Could you add a spill-check case with two or more input partitions
and a derived key at r ~ 1.5, once with disjoint epochs (`(id * 2 /
3).cast("long")`) and once with keys that repeat across epochs, run it with
`minCompactionToSpill` at 1.05 and 2.0, and put the numbers in the description?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -102,7 +102,8 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
*/
private def checkAdaptiveMatchesReference(
build: Int => DataFrame,
- expectBypass: Boolean = true): Unit = {
+ expectBypass: Boolean = true,
+ extraConfs: Seq[(String, String)] = Nil): Unit = {
Review Comment:
23. Nit: `extraConfs` is used only by ROLLUP and CUBE, with the same literal
(L726, L740), and the scaladoc above (L80-102) does not describe it. Since
`withSQLConf` restores only its own keys and this helper never sets
`MIN_COMPACTION_TO_SPILL`, wrapping those two calls in
`withSQLConf(SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION_TO_SPILL.key
-> "1.0") { ... }` would do the same. Could you drop the parameter, or document
it?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -1199,6 +1205,47 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
}
}
+ test("the spill check passes through an epoch that compacts too little to
spill") {
+ // Keys `floor(2 * id / 3)` collapse every three rows into two keys, a
steady ratio of about
+ // 1.5. That is above `minCompaction` (1.05), so the periodic checks at 10
and 20 rows keep
+ // aggregating. The forced fallback fills the map when id 30 arrives,
after 30 rows over 20
+ // keys: keeping the aggregation would pay to spill the map and merge it
back for a ratio of
+ // only 1.5, so the default `minCompactionToSpill` (2.0) passes the rest
through instead, and
+ // the sorter is never created. Lowering it to 1.4 keeps the old behavior:
every epoch spills,
+ // and nothing is passed through.
+ Seq(None -> true, Some("1.4") -> false).foreach { case
(minCompactionToSpill, shouldBypass) =>
+ Seq(true, false).foreach { wholeStage =>
+ withSQLConf(
+ (Seq(
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_ENABLED.key -> "true",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> wholeStage.toString,
+ SQLConf.ENABLE_TWOLEVEL_AGG_MAP.key -> "false",
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS.key -> "10",
+ "spark.sql.TungstenAggregate.testFallbackStartsAt" -> "1, 30") ++
+ minCompactionToSpill.map(
+ SQLConf.ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION_TO_SPILL.key
-> _) ++
+ fixedPlanConfs): _*) {
+ val df = () => spark.range(0, 400, 1, 1)
Review Comment:
16. Minor: the spill-check flip between `minCompaction` and 2.0, which this
PR adds, is tested only where Partial and Final fuse. This test, the `(1.0,
1.3, true)` boundary cell and the test at L1249 all read `spark.range(0, 400,
1, 1)`, so passed-through rows feed the Final's `doConsume` and never go
through the output buffer, `shouldStop()` and the resumed build. The cells that
would take this path with an `Exchange` in between, ROLLUP and CUBE under
forced spill (their `Expand` always splits the plan), are pinned to 1.0 at L726
and L740. Could this test also run with two input partitions? The cast key adds
an `Exchange`, each 200-row partition fills at 30 rows over 20 keys, and
`skipped` would be `2 * (200 - 30)`.
##########
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")
Review Comment:
22. Nit: moving this line before `doAgg`, outside the adaptive branch,
changes the generated code of every keyed aggregate, including those with the
feature off. `metricTerm` registers a reference object, and `addReferenceObj`
appends in registration order (CodeGenerator.scala:276-281), so `aggTime` now
takes an earlier `references[]` slot and every reference that the child's
produce and the output function register shifts by one. There is no functional
effect, but L678-682 and L818-820 keep the codegen layout of plans the feature
never touches unchanged, and this is the only feature-off codegen change in the
PR. Could this line go back to its old place? With the `postChildProduce`
option in item 11, nothing here needs it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregationIterator.scala:
##########
@@ -230,8 +230,8 @@ class TungstenAggregationIterator(
val spillCheckMinRows = math.max(minRows, 1L)
// The partial aggregation is ineffective when it does not collapse
`minCompaction` rows into
// one key. There is no fast map on this path, so the map's keys are all
the operator holds.
- def ineffective(): Boolean =
- processedRows < hashMap.getNumKeys().toDouble * adaptiveMinCompaction
+ def ineffective(minCompaction: Double): Boolean =
Review Comment:
24. Nit: this parameter is still named `minCompaction`, but L247 passes
`adaptiveMinCompactionToSpill`, which is the larger of the two thresholds, and
the same holds for `adaptiveIneffective` (HashAggregateExec.scala:969, called
with it at HashAggregateExec.scala:1044). The unchanged comment at L231-232
("does not collapse `minCompaction` rows into one key") now describes only one
of the two calls, while the block comment at L193-195 calls the value
`threshold`. Could you rename the parameter to `threshold` in both functions
and update L231-232?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -1199,6 +1205,47 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
}
}
+ test("the spill check passes through an epoch that compacts too little to
spill") {
+ // Keys `floor(2 * id / 3)` collapse every three rows into two keys, a
steady ratio of about
+ // 1.5. That is above `minCompaction` (1.05), so the periodic checks at 10
and 20 rows keep
Review Comment:
21. Nit: a few comments around the spill-check tests now credit the wrong
threshold or skip a step.
- Here, there is also a periodic check at 30 rows, on id 29 (30 rows over 20
keys = 1.5), right before id 30 finds the map full. It sees the same ratio as
the spill check, so it is the clearest contrast between the two thresholds.
- At L1184, `minCompaction -> "2.0"` now equals the default spill threshold,
so the test at L1170 passes with or without it (19 rows over 11 keys = 1.73 <
2.0), and its comment at L1171-1176 still credits `minCompaction` for both
decisions. Setting `MIN_COMPACTION_TO_SPILL -> "2.0"` instead keeps the intent
and leaves the periodic check at 1.05, where 5.0 still keeps aggregating.
- At L709, "about 1.4x" fits ROLLUP (1.33-1.45) but not CUBE, whose epochs
compact about 1.23x.
Could you update them?
##########
docs/sql-performance-tuning.md:
##########
@@ -200,6 +200,8 @@ 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.
+Keeping the aggregation past that point also pays to write the map to disk and
merge it back later,
+so the evaluation before a spill demands a higher ratio,
`minCompactionToSpill`.
Review Comment:
20. Minor: the tuning advice was not updated for the new config. L209-217
still tells a user whose distinct-heavy prefix trips pass-through early to
raise `minRows`, which helps at a fill only if it exceeds the rows the map
takes to fill. After this PR, the spill check with its 2.0 default is the
evaluation most likely to commit the task early, and lowering
`minCompactionToSpill` is the direct fix, which the paragraph does not mention.
The new entry (SQLConf.scala:4701-4710 and L256-267 here) also does not say
that the spill-time evaluation runs only once the current map has processed
`minRows` rows, so with the default 100000 a map that fills sooner never
consults it. Last, "demands a higher ratio, `minCompactionToSpill`" here holds
only by default; the rule is the larger of the two values, as L263-264 says.
Could you mention `minCompactionToSpill` in L209-217, the `minRows` floor in
its entry, and "the larger of `minCompaction` and `minCompactionToSpill`" here?
--
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]
