cloud-fan commented on code in PR #59823:
URL: https://github.com/apache/spark/pull/59823#discussion_r4395119528
##########
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:
Thanks, your arithmetic is right, and I reproduced it. With keys spread
evenly, the recent ratio depends only on the share of the task's keys the maps
hold, and reaches 2.0 at about 57% of 600K keys (about 50% of 6M). Comparing
the recent ratio with the epoch's measures the same share on another scale. It
also reads 1.0 whenever the map fills before its second periodic check, because
the window is then the whole epoch. Nothing inside one epoch shows how often
keys come back after the map fills. So in d74b8d0 the tuning guide and the
`minCompactionToSpill` entry say that the default keeps aggregating such keys
only when the map holds about half of them or more, and the guide says that
lowering `minCompactionToSpill` keeps more.
Whether passing through is then the costlier call depends on how a spilled
row compares with a shuffled row, so I added your case to the benchmark: 10
rows per key, with the maps holding 40% of the keys, run with the fast map off
and on. Against 1.05, the default ranged from 8% faster to 7% slower on the
runners that produced the results in 6a5684c. In one earlier run on an AMD EPYC
9V45 it was about 21% slower, which matches your break-even when a spilled row
costs about as much as a shuffled one. In the same runs, the default was about
1.3x faster when keys arrive together or concentrate on a few hot keys. Since
no threshold on the epoch's own rows wins both, I'd keep 2.0. I filed
SPARK-60601 for a signal that sees keys recur across epochs, for example a
small hash-selected sample of the first epoch's keys, or the fast map's keys,
which outlive every spill.
##########
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:
Done in d74b8d0. The `minRows` entry, in `SQLConf` and the tuning guide, now
says that it also sets how many recent rows the spill check measures: at least
`minRows` and fewer than twice that, or the whole map epoch before the map's
second periodic check, and the whole epoch with `minRows = 0`. The tuning
advice says that raising `minRows` widens that sample. I kept the window on the
periodic cadence rather than a fraction of the epoch: the window is meant to
measure the rate of new keys near the fill, and a fraction of a long epoch
drifts toward the epoch's own ratio.
##########
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:
Done in d74b8d0. The test pins `minCompactionToSpill` to 1.5, which keeps
all four counts, and the comment mentions the tie. I also confirmed that 2.1
turns the two-level codegen cell into 316, as you said.
##########
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:
Done in d74b8d0. The boundary test has the `(1.25, 1.0, false)` cell, and a
new test, "the spill check judges the recent rows at their exact ratio
boundary", uses your input: on both paths it keeps aggregating at 2.0 and
passes the last 10 rows through at 2.1.
--
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]
