cloud-fan commented on code in PR #59823:
URL: https://github.com/apache/spark/pull/59823#discussion_r4395364082
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/AdaptivePartialAggregationSuite.scala:
##########
@@ -1310,6 +1423,42 @@ class AdaptivePartialAggregationSuite extends QueryTest
with SharedSparkSession
}
}
+ test("the spill check judges the recent rows at their exact ratio boundary")
{
+ // With `minRows = 10`, ids 0-9 add 9 keys and ids 10-29 add 10 more, two
rows each, so when
+ // the forced fallback fills the map at 30 rows, the rows since the check
before last (at 10
Review Comment:
Thanks, you're right. The check at 30 rows runs on id 29, so the window is
ids 20-29, and the old input read 2.0 from either start. 40163fb now uses your
input: the window reads exactly 2.0, a window starting one check earlier reads
1.33, and the whole epoch reads 1.5. So the test keeps aggregating at 2.0 and
passes through at 2.1 on both paths. I also checked that the test fails when
the window starts one check earlier.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregationIterator.scala:
##########
@@ -189,14 +190,18 @@ class TungstenAggregationIterator(
// When adaptive partial aggregation is enabled
(`adaptivePartialAggEnabled`), the processing may
// stop early and switch to pass-through mode: the remaining input rows are
not added to the map
// but are instead emitted as single-row partial buffers by the output stage
(see
- // `nextPassThroughOutput`). One predicate decides both check points -- the
aggregation is
- // ineffective when `processedRows < distinctKeys * minCompaction`, i.e. it
does not collapse
- // `minCompaction` rows into one key:
- // - periodically, every `minRows` processed rows;
- // - when the map is about to spill, in which case the spill is skipped
entirely and the row
- // that could not be inserted becomes the first pass-through row.
Pass-through cannot be
- // undone within the task, so like the periodic check this waits until
the current epoch has
- // processed `minRows` rows; a map that fills sooner spills as usual.
+ // `nextPassThroughOutput`). The aggregation is ineffective when `rows`
processed rows do not
+ // collapse into their `keys` keys by at least `threshold`, checked at two
points:
+ // - periodically, every `minRows` processed rows, over the whole epoch
against
+ // `minCompaction`;
+ // - when the map is about to spill, over the recent rows (those since the
periodic check
+ // before last) against the higher `minCompactionToSpill`. Keeping the
aggregation then
+ // means aggregating the rest of the input through spills, which pays
off only if that input
+ // keeps repeating keys; when keys repeat across the input, the ratio of
the recent rows
+ // climbs above the epoch's as the map fills. If the check passes, the
map does not spill
Review Comment:
Done in 40163fb, with your wording.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala:
##########
@@ -1148,10 +1178,12 @@ case class HashAggregateExec(
| if (!$adaptivePassThroughTerm) {
| $processedRowsTerm += 1;
| if ($processedRowsTerm == $adaptiveNextCheckRowTerm) {
- | if ($adaptiveIneffective) {
+ | if ($epochIneffective) {
| $adaptivePassThroughTerm = true;
| } else {
| $adaptiveNextCheckRowTerm += ${adaptiveMinRows}L;
+ | $adaptiveWindowKeysTerm = $adaptiveCheckKeysTerm;
+ | $adaptiveCheckKeysTerm = $adaptiveEpochKeys;
Review Comment:
Done in 40163fb, with two tests:
- "the spill check of a new map epoch does not count the previous epoch's
keys" fills every epoch after one periodic check. The second epoch compacts by
19 / 12 and passes its last 12 rows through, while carrying over the first
epoch's 5 keys would judge 19 rows over 7 keys and keep aggregating. It runs on
both paths.
- "the spill check's window counts the keys of both maps" uses the two-level
map on the codegen path, with a 16-key fast map that holds 10 keys when the
window starts. The window reads 1.83, so the task keeps aggregating at 1.5 and
passes through at 1.9.
I checked that the tests fail when the code skips the key-count reset at a
spill, snapshots only the regular map's keys, or counts the fast map's keys
twice.
--
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]
