You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by ka...@apache.org on 2026/07/05 13:34:08 UTC
(datafusion) branch final-aggr-with-lazily-materialize-poc-candidate1 updated: Inline final partition replay transitions
This is an automated email from the ASF dual-hosted git repository.
Rachelint pushed a commit to branch final-aggr-with-lazily-materialize-poc-candidate1
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/final-aggr-with-lazily-materialize-poc-candidate1 by this push:
new 59affe49d3 Inline final partition replay transitions
59affe49d3 is described below
commit 59affe49d33a537abe03775eb851916cfec46875
Author: kamille <ka...@apache.org>
AuthorDate: Sun Jul 5 21:34:02 2026 +0800
Inline final partition replay transitions
---
.../physical-plan/src/aggregates/hash_aggregate.rs | 50 ++++++++--------------
1 file changed, 18 insertions(+), 32 deletions(-)
diff --git a/datafusion/physical-plan/src/aggregates/hash_aggregate.rs b/datafusion/physical-plan/src/aggregates/hash_aggregate.rs
index 4aa39a3a5e..77d40ec3f0 100644
--- a/datafusion/physical-plan/src/aggregates/hash_aggregate.rs
+++ b/datafusion/physical-plan/src/aggregates/hash_aggregate.rs
@@ -1013,28 +1013,6 @@ impl FinalHashAggregateStream {
Ok(())
}
- fn begin_partition_run_replay(
- &mut self,
- hash_table: AggregateHashTable<FinalMarker>,
- ) -> Result<FinalHashAggregateState> {
- self.load_next_partition_run(hash_table)
- }
-
- fn next_state_after_partition_output(
- &mut self,
- hash_table: AggregateHashTable<FinalMarker>,
- ) -> Result<FinalHashAggregateState> {
- if self
- .partition_run_state
- .as_ref()
- .is_some_and(FinalPartitionRunState::has_buffered_runs)
- {
- self.load_next_partition_run(hash_table)
- } else {
- Ok(FinalHashAggregateState::Done)
- }
- }
-
fn load_next_partition_run(
&mut self,
mut hash_table: AggregateHashTable<FinalMarker>,
@@ -1243,7 +1221,7 @@ impl FinalHashAggregateStream {
let timer = elapsed_compute.timer();
let result = match self.partition_run_state.as_ref() {
Some(state) if !state.run_batch_indices.is_empty() => {
- self.begin_partition_run_replay(original_state.into_hash_table())
+ self.load_next_partition_run(original_state.into_hash_table())
}
_ => self
.start_output(original_state.hash_table_mut())
@@ -1288,16 +1266,24 @@ impl FinalHashAggregateStream {
let next_state = if original_state.hash_table().is_done()
|| original_state.hash_table().is_building()
{
- match self.next_state_after_partition_output(
- original_state.into_hash_table(),
- ) {
- Ok(next_state) => next_state,
- Err(e) => {
- return ControlFlow::Break((
- Poll::Ready(Some(Err(e))),
- FinalHashAggregateState::Done,
- ));
+ if self
+ .partition_run_state
+ .as_ref()
+ .is_some_and(FinalPartitionRunState::has_buffered_runs)
+ {
+ match self
+ .load_next_partition_run(original_state.into_hash_table())
+ {
+ Ok(next_state) => next_state,
+ Err(e) => {
+ return ControlFlow::Break((
+ Poll::Ready(Some(Err(e))),
+ FinalHashAggregateState::Done,
+ ));
+ }
}
+ } else {
+ FinalHashAggregateState::Done
}
} else {
let _ =
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org