dongjoon-hyun commented on code in PR #59823:
URL: https://github.com/apache/spark/pull/59823#discussion_r4389483397
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -4685,16 +4685,39 @@ 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 demands at least this value
too, see " +
+
"'spark.sql.execution.aggregate.adaptivePartialAggregation.minCompactionToSpill'.")
.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 then means " +
+ "aggregating the rest of the input through spills, which pays off only
if that input " +
+ "keeps repeating keys, so this evaluation measures the recent rows:
those processed " +
+ "since the periodic evaluation before last, against the keys they
added. When keys " +
+ "repeat across the input, that ratio climbs as the map fills. When it
is below the " +
+ "larger of this value and " +
+ s"'${ADAPTIVE_PARTIAL_AGGREGATION_MIN_COMPACTION.key}', the map is not
spilled and the " +
+ "rest of the input is passed through. Like the periodic evaluation, it
only runs once " +
+ "the current map has processed " +
+ s"'${ADAPTIVE_PARTIAL_AGGREGATION_MIN_ROWS.key}' rows.")
+ .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:
**[1] Above the `[minRows, 2 * minRows)` band, the recent-rows ratio reaches
2.0 only when the map holds more than about 57% of a task's distinct keys,
however often each key repeats.**
[ulysses-you's
comment](https://github.com/apache/spark/pull/59823#issuecomment-6436941615)
shows that inside the band the decision depends on the map capacity. It still
does above the band, where the window is in effect. For keys drawn uniformly
from D distinct keys, a row adds a key with probability 1 - k/D while the map
holds k keys, so the recent rows compact by roughly 1/(1 - k/D) just before the
fill. That measures how much of the key space fits in the map, not how often
the keys repeat.
Take the example of [item
12](https://github.com/apache/spark/pull/59764#discussion_r4371926323) of my
review on PR 59764: 6M rows over 600K distinct keys per task, so 10 rows per
key, with `minRows = 100000`, but with room for 240K keys instead of 400K:
- The map fills after about 306K rows, past three periodic checks, so the
spill check judges rows 200K to 306K: about 106K rows over 70K new keys, 1.52.
- 1.52 is below 2.0, so the task passes the rest through and shuffles about
5.9M rows. Keeping the aggregation would shuffle about 0.6M, after about 19
epochs that spill about 4.5M rows in total. Before this PR the task kept
aggregating, because the epoch ratio, 1.28, is above 1.05.
With room for 400K keys, as in item 12, the window reads 2.63 and the task
keeps aggregating. The `keys recur` benchmark case is on that side too: its map
fills at 220K rows and holds about 67% of the 200K keys, so the window reads
about 2.2. By expected-value arithmetic, the window stays below 2.0 until the
map holds about 57% of the keys, whether each key repeats 10 or 100 times per
task.
Could the description and the docs say that the spill check keeps
aggregating recurring keys only when the map holds most of a task's distinct
keys? Or, since the description's own signal is that the recent ratio "climbs
above the epoch's" when keys repeat and matches it when each key's rows arrive
together, could the check compare the two ratios instead of the recent one with
an absolute 2.0? In the example above the recent ratio is about 1.19 times the
epoch's (1.52 against 1.28), while for `floor(2 * id / 3)` the two match.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregationIterator.scala:
##########
@@ -271,10 +288,14 @@ class TungstenAggregationIterator(
// Periodic check: if the aggregation is not collapsing enough rows,
bypass it for the
// rest of the input.
if (adaptivePartialAggEnabled && processedRows == nextCheckRow) {
- if (ineffective()) {
+ if (ineffective(processedRows, hashMap.getNumKeys(),
adaptiveMinCompaction)) {
passThrough = true
} else {
nextCheckRow += minRows
+ windowRows = checkRows
Review Comment:
**[2] `minRows` now also sets how many rows the spill check samples, from
`minRows` to `2 * minRows - 1` of the latest rows, and neither config doc says
so.**
The window starts at the periodic check before last (L295-298 here and
HashAggregateExec.scala:1188-1191), so its length follows the periodic cadence
rather than the size of the epoch. With the default and a map that fills after
several million rows, the decision that cannot be undone rests on the last 100K
to 200K rows. With `minRows = 1000` it rests on the last 1000 to 1999 rows, so
a short run of new keys right before the fill can pull the window below 2.0
even when the epoch as a whole keeps repeating keys. The suite's forced-spill
cells, with `minRows = 8`, judge 8 to 15 rows.
The `minRows` entry (SQLConf.scala:4672-4676 and sql-performance-tuning.md
L248-254) describes it only as the periodic interval and the floor of the spill
check, and the advice to raise it (sql-performance-tuning.md L226-229) does not
say that this also widens the spill-check sample. ulysses-you's item 2 already
asks to align the same entry for `minRows = 0`. Could the entry also say that
the spill check judges the rows since the check before last, or could the
window be sized as a fraction of the epoch instead?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/AdaptivePartialAggregationBenchmark.scala:
##########
@@ -135,6 +135,55 @@ object AdaptivePartialAggregationBenchmark extends
SqlBasedBenchmark {
extraConf = spillCheckConf)
benchmark.run()
}
+
+ // Inputs that compact by about 1.5 within each map epoch, read from four
partitions so an
+ // `Exchange` separates the Partial from the Final, and with a map forced
to fill every few
+ // hundred thousand rows. Whether keeping the aggregation through the
spills pays off depends
+ // on whether the rest of the input keeps repeating keys:
+ // - each key's rows arrive together, so later epochs share no keys and
passing the rest
+ // through at the first fill avoids spilling every epoch;
+ // - keys recur uniformly across the task (10 rows per key), so the
final merge collapses
+ // them and keeping the aggregation shuffles a tenth of the rows;
+ // - 10 hot keys hold 30% of the rows and the rest are unique, so
passing through avoids the
+ // spills but sends every hot-key row to the reducer of its key.
+ // `minCompactionToSpill = 1.05` judges the spill check by
`minCompaction`; the default, 2.0,
+ // demands that the recent rows compact by 2.0.
+ val spillCheckCases = Seq(
+ ("keys arrive together", "cast(id * 2 / 3 as bigint)", 300000),
+ ("keys recur", "pmod(xxhash64(id), 200000)", 220000),
+ ("hot keys", "if(id % 100 < 30, id % 10, id + 1000)", 300000))
+ spillCheckCases.foreach { case (name, key, fillRows) =>
+ runBenchmark(s"spill check with an Exchange, $name") {
Review Comment:
**[5] The result files were not regenerated for these cases.**
`sql/core/benchmarks/AdaptivePartialAggregationBenchmark-results.txt` and
its `-jdk21` and `-jdk25` variants still hold only the four groups from
5bcd5f59b9f, so the 20% and 35% numbers in the description are not recorded
anywhere a later run can be compared with. The class doc at L31-40 also still
says that every scenario runs the `adaptive = T/F` matrix over the
high/low-cardinality and spill grid, while these groups compare
`minCompactionToSpill` values. The four existing groups decide the same at 2.0,
so only the new groups change. Could you regenerate the three files with the
"Run benchmarks" workflow and extend the class doc?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -1278,30 +1377,40 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
// failed insertion that becomes the first pass-through row, so both
execution paths must judge
// the same row set and reach the same decision. `id % 40` over 400 rows
gives 40 keys when the
// map fills at 50 aggregated rows, i.e. a compaction ratio of exactly
1.25: demanding 1.25 the
- // aggregation is kept (`50 < 40 * 1.25` is false), demanding 1.3 it is
bypassed.
- Seq(1.25 -> false, 1.3 -> true).foreach { case (minCompaction,
shouldBypass) =>
+ // aggregation is kept (`50 < 40 * 1.25` is false), demanding 1.3 it is
bypassed. The spill
+ // check demands the larger of `minCompaction` and `minCompactionToSpill`,
so either one can
+ // set the boundary.
+ Seq(
+ (1.0, 1.25, false),
+ (1.0, 1.3, true),
+ (1.3, 1.0, true)
Review Comment:
**[4] Nit: the boundary test does not cover `minCompaction` on the keep
side, and with `minRows = 0` it never reaches the windowed comparison.**
The cells cover `minCompactionToSpill` at the boundary from both sides, but
`minCompaction` only from the bypass side. A `(1.25, 1.0, false)` cell would
pin that the larger value applies at equality too.
The windowed form, `processedRows - windowRows` against `(keys - windowKeys)
* threshold`, meets its boundary only through the tie in [3]. A cell with
`minRows = 10` could reach it on purpose. For example, 10 rows over 9 keys
followed by keys that each take two rows (ids 10-29 add 10 keys) fill at 30
rows with a window of exactly 2.0 over an epoch of 1.58. The task then keeps
aggregating at 2.0 and passes through at 2.1, while a whole-epoch check would
pass through at both.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregationIterator.scala:
##########
@@ -225,10 +230,16 @@ class TungstenAggregationIterator(
// The spill check waits for `minRows` rows of the current epoch, and
for at least one row
// when `minRows = 0`.
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
+ // The epoch's processed rows and keys at its last periodic check, and
at the check before
+ // that, where the window of recent rows that the spill check judges
starts.
+ var checkRows = 0L
+ var checkKeys = 0L
+ var windowRows = 0L
Review Comment:
**[6] Nit: `checkRows` and `windowRows` follow from `nextCheckRow`.**
The periodic checks run at exact multiples of `minRows` within an epoch, and
`nextCheckRow` advances only when a check passes. So at any spill check
`checkRows == nextCheckRow - minRows` and `windowRows == max(0, nextCheckRow -
2 * minRows)`, which is also 0 when `minRows = 0`. Only the two key snapshots
need state. Deriving the row side would drop two variables with their resets
and rotations here (L235-238, L273-276 and L295-298), and two mutable states in
HashAggregateExec.scala (L591-594, L1081-1084 and L1188-1191). This is
optional, since the explicit snapshots are easy to follow.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -1227,15 +1325,16 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
"spark.sql.TungstenAggregate.testFallbackStartsAt" -> s"$fastCap,
40") ++
fixedPlanConfs): _*) {
val df = () => spark.range(0, 400, 1, 1)
- .select(when($"id" < 100, $"id" % 5).otherwise($"id") as "k", $"id"
as "v")
+ .select(when($"id" < 80, $"id" % 5).otherwise($"id") as "k", $"id"
as "v")
.groupBy($"k")
.agg(sum($"v") as "s")
withClue(s"wholeStage=$wholeStage twoLevelMap=$twoLevelMap") {
val c = runAndReadCounters(df)
assert(c.tasksFallBacked > 0,
"the low-cardinality prefix should still spill and fall back to
sort")
- assert(c.passedThrough > 0,
- "the high-cardinality tail after the spill should be passed
through")
+ val firstPassedThrough = if (wholeStage && twoLevelMap) 92 else 88
Review Comment:
**[3] In the two-level codegen cell, the expected 92 depends on a window of
exactly 2.0.**
In that cell the second epoch starts at id 44, and its periodic checks run
at 32 rows (id 75, where the epoch has one key, 4) and at 40 rows (id 83, five
keys: 4 and 80-83). When id 84 finds the map full, the window is 8 rows over 4
keys, exactly 2.0, and the task keeps aggregating only because `8 < 4 * 2.0` is
false. Over the whole epoch it would see 40 rows over 5 keys, so the tie comes
from the window.
With a default of 2.1, or `<=` in the comparison, this cell passes through
at id 84 and the count becomes 316 instead of 308, which the comment at
L1305-1313 does not prepare a reader for. The other cells see 8 rows over 0
keys at id 80 and do not depend on the threshold. Could the test pin
`minCompactionToSpill` below 2.0, e.g. 1.5, which keeps all four counts, or
mention the tie in the comment?
--
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]
