This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25075-bf01e288eeb5cbfb2c8a4cf77f82f97f81c109ba in repository https://gitbox.apache.org/repos/asf/datafusion.git
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]
