spark-test commented on code in PR #59823:
URL: https://github.com/apache/spark/pull/59823#discussion_r4392846978
##########
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:
**Nit (P3):** Two transitions of the new key snapshots have no assertion
that depends on them.
1. The reset at a spill: `$adaptiveCheckKeysTerm = 0L` at L1079, and
`checkKeys = 0L` in TungstenAggregationIterator L275. It matters only when a
post-spill epoch fills between its first and second periodic checks, because
`windowKeys` takes the value `checkKeys` held when the first check rotated it.
Every forced-spill test refills at `2 * minRows` rows or later: the `4, 16`
cells with `minRows = 8`, the 1.3 cell filling at 30 with `minRows = 10`, and
the post-spill test filling at 40 with `minRows = 8`. So dropping the reset
passes the suite, although such an epoch would then subtract the previous
epoch's key count from its own.
2. The fast-map term that `adaptiveEpochKeys` adds to the snapshot here.
Every window test disables the two-level map. The one two-level exact-count
test ('a new in-memory map epoch after a spill can still bypass') fills its
4-key fast map within the first 4 rows, before any check. So snapshotting only
`$hashMapTerm.getNumKeys()` changes its first spill check (id 44) from 12 rows
over 0 keys to 12 over 4, and the `400 - 92` count still holds. With the
default 2^16-key fast map and `minRows = 100000`, the fast map can still be
gaining keys inside a window, so this term matters on the default codegen
configuration.
Could the suite cover a post-spill epoch that fills after exactly one
periodic check, and a two-level codegen case whose fast map gains keys inside
the judged window?
**Verification:**
- **Regression:** If the window's key snapshot were not reset at a spill, a
run whose next epoch fills between its first and second periodic checks would
produce a different passed-through count on both execution paths, and the
coverage fails.
- **Regression:** If the generated-code snapshot counted only regular-map
keys, or counted fast-map keys twice, a two-level-map run with fast-map keys
added inside the judged window would produce a different passed-through count,
and the coverage fails.
##########
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:
**Nit (P3):** When id 30 finds the map full, the periodic check at 30 rows
has already run on id 29. So `nextCheckRow` is 40, and the spill check judges
the rows since the check at 20 rows: ids 20-29, 10 rows over 5 keys (`10 < 5 *
2.0` is false). It does not judge 20 rows over 10 keys since the check at 10.
Ids 10-29 all add one key per two rows, so a window starting at 10 also
gives exactly 2.0. Both assertions (0 and 10 passed-through rows) hold either
way. The repeat-keys and 'compacts too little' tests do not tell those two
starts apart either, so a regression that moved the window start one interval
earlier would pass the suite.
Could the comment describe rows 20-29, and could the input make ids 10-19
add keys at a different rate from ids 20-29? For example:
- ids 0-9 over 5 keys (two rows each);
- ids 10-19 over 10 new keys (one per row);
- ids 20-29 over 5 new keys (two rows each).
The periodic checks still keep aggregating (2.0, 1.33 and 1.5), and the
judged window stays at exactly 2.0. But a window from the check at 10 would see
20 rows over 15 keys and pass through at 2.0, while the whole epoch (30 over
20) passes through at both thresholds.
This extends an [existing
discussion](https://github.com/apache/spark/pull/59823#discussion_r4389483460).
**Verification:**
- **Behavior:** On both the codegen and interpreted paths, a task whose
recent rows compact exactly at the spill threshold keeps aggregating and
spills, and one held to a slightly higher threshold passes its remaining rows
through, with exact passed-through and fallback counts.
- **Regression:** If the spill-check window started one periodic interval
earlier than the rows since the check before last, the boundary coverage fails
on at least one execution path.
##########
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:
**Nit (P3):** Right after the sentence about the recent ratio climbing as
the map fills, 'If the check passes' reads as the ratio meeting
`minCompactionToSpill`. But that is the case where the map spills and keeps
aggregating. The code at L256-259 passes the rest through when
`ineffective(...)` holds, i.e. when the recent rows compact by less than the
threshold. The codegen comment and the tuning guide both state it that way
round. Merge-base wording was unambiguous. A same-length rewording:
```suggestion
// climbs above the epoch's as the map fills. Below the threshold, the
map does not spill
```
**Verification:**
- **Inspection:** The interpreted path's comment states that the map is not
spilled and the rest of the input is passed through exactly when the recent
rows compact by less than the spill threshold, matching the predicate that sets
pass-through.
--
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]
