fix: charge accumulators' construction-time state to the memory pool in ungrouped aggregation - #24692
Open
ranflarion wants to merge 1 commit into
Open
fix: charge accumulators' construction-time state to the memory pool in ungrouped aggregation#24692ranflarion wants to merge 1 commit into
ranflarion wants to merge 1 commit into
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Rationale for this change
An ungrouped aggregation can hold arbitrarily large accumulator state that the memory pool never sees.
AggregateStreamgrows its reservation only by per-batchsize()deltas (size_post.saturating_sub(size_pre)inaggregate_batch), so an accumulator that allocates its retained state in its constructor and never resizes it is charged nothing for the stream's lifetime. The grouped path already handles this:GroupsAccumulatorAdapterchargesstate.size()when it creates each accumulator. Embedding DataFusion 54.1.0 in a Spark accelerator we measured Spark's runtime-filterbloom_filter_agg(an 8 MiB bit array zero-filled at construction, one per concurrent task) running entirely unaccounted, so a fair-spill pool could neither fail admission nor pressure other consumers to spill.What changes are included in this PR?
One charge in
AggregateStream::new: after registering the reservation,try_growthe sum of the accumulators' initialsize(). Per-batch deltas compose on top of it, so there is no double counting.test_oom's ungrouped arm is updated for the earlier failure point: with its 1-byte limit the median accumulator's initial 48 bytes now fail admission atexecute_typed, so that arm asserts the construction-timeResourcesExhaustedinstead of collecting first; the grouped arms are unchanged.Are these changes tested?
The updated
test_oomis the regression pin: onmainwithout the fix the ungrouped stream constructs successfully and the arm fails, with the fix it passes. The fulldatafusion-physical-planlib suite passes (1784 tests).Are there any user-facing changes?
Yes, behavioral: an ungrouped aggregation whose accumulators' construction-time state exceeds the memory limit now fails with
ResourcesExhaustedatexecuteinstead of silently running past the configured limit, and memory-limited setups that relied on that overshoot may need limits raised to what the query actually uses. No API changes.