This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 35c56b020e chore: mark partial reduce stream as can handle back
pressure (spillable) (#25075)
35c56b020e is described below
commit 35c56b020e190126f45395a217d47475f5f6f020
Author: Raz Luvaton <[email protected]>
AuthorDate: Wed Sep 9 04:01:58 2026 +0000
chore: mark partial reduce stream as can handle back pressure (spillable)
(#25075)
## Which issue does this PR close?
N/A
## Rationale for this change
While adding fuzz tests for aggregate streams I encounter OOMs that
partial reduce takes up all the memory and leave nothing to partial
Same as:
https://github.com/apache/datafusion/blob/a5c809f98dedcf4b22f5d317ea8efe0720834925/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs#L544-L546
## What changes are included in this PR?
mark reduce hash aggregate stream as spillable
## What is the testing strategy for this PR?
This was found as part of:
- #24881
## Are there any user-facing changes?
No
---
datafusion/physical-plan/src/aggregates/hash_stream.rs | 3 +++
datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs | 4 ++++
2 files changed, 7 insertions(+)
diff --git a/datafusion/physical-plan/src/aggregates/hash_stream.rs
b/datafusion/physical-plan/src/aggregates/hash_stream.rs
index 2195b5eb60..796a2b2885 100644
--- a/datafusion/physical-plan/src/aggregates/hash_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/hash_stream.rs
@@ -420,6 +420,9 @@ impl PartialHashAggregateStream {
let reservation =
MemoryConsumer::new(format!("PartialHashAggregateStream[{partition}]"))
+ // We interpret 'can spill' as 'can handle memory back
pressure'.
+ // This value needs to be set to true for the default memory
pool implementations
+ // to ensure fair application of back pressure amongst the
memory consumers.
.with_can_spill(true)
.register(context.memory_pool());
diff --git a/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs
b/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs
index c9a6cb675d..286a324e96 100644
--- a/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs
@@ -198,6 +198,10 @@ impl PartialReduceHashAggregateStream {
let reservation =
MemoryConsumer::new(format!("PartialReduceHashAggregateStream[{partition}]"))
+ // We interpret 'can spill' as 'can handle memory back
pressure'.
+ // This value needs to be set to true for the default memory
pool implementations
+ // to ensure fair application of back pressure amongst the
memory consumers.
+ .with_can_spill(true)
.register(context.memory_pool());
Ok(Self {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
