diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs index e6e690c4d1e08..eaf39929ced62 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs @@ -36,6 +36,8 @@ use crate::aggregates::{ /// Marker for raw rows -> partial state aggregation. pub(in crate::aggregates) struct PartialMarker; +/// Marker for raw rows -> final value aggregation. +pub(in crate::aggregates) struct SingleMarker; /// Marker for partial state -> partial state aggregation. pub(in crate::aggregates) struct PartialReduceMarker; /// Marker for raw rows -> partial state conversion without aggregation. diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/mod.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/mod.rs index 0d2495a1b556c..2c7ec01654a63 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/mod.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/mod.rs @@ -22,9 +22,10 @@ mod ordered_final_table; mod ordered_partial_table; mod partial_reduce_table; mod partial_table; +mod single_table; pub(super) use common::{ AggregateHashTable, FinalMarker, PartialMarker, PartialReduceMarker, - PartialSkipMarker, + PartialSkipMarker, SingleMarker, }; pub(super) use common_ordered::OrderedAggregateTable; diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/single_table.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/single_table.rs new file mode 100644 index 0000000000000..5dcb735d083c4 --- /dev/null +++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/single_table.rs @@ -0,0 +1,74 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use arrow::datatypes::SchemaRef; +use arrow::record_batch::RecordBatch; +use datafusion_common::Result; + +use crate::aggregates::AggregateExec; + +use super::common::{AggregateHashTable, HashAggregateAccumulator, SingleMarker}; + +/// Implementation specific to single aggregation, where the table stores final +/// aggregate values and the input rows are raw rows. +/// +/// Example: `AVG(x) GROUP BY k` +/// +/// - Aggregate table stores: `k, avg(x)` +/// - Input rows: `k, x` +impl AggregateHashTable { + pub(in crate::aggregates) fn new( + agg: &AggregateExec, + partition: usize, + output_schema: SchemaRef, + batch_size: usize, + ) -> Result { + Self::new_with_filters( + agg, + partition, + output_schema, + batch_size, + agg.filter_expr.iter().cloned().collect(), + ) + } + + /// Emits the next batch of aggregated group keys and final aggregate values. + /// + /// The output batch size is determined by `self.batch_size`. + /// + /// Returns `Some(batch)` for each emitted batch, `None` when output is + /// exhausted, and an internal error if polled in the `Building` state. + pub(in crate::aggregates) fn next_output_batch( + &mut self, + ) -> Result> { + self.next_output_batch_inner(HashAggregateAccumulator::evaluate_to_columns) + } + + /// Single aggregation consumes raw input rows and updates the table's + /// final-value accumulators. + pub(in crate::aggregates) fn aggregate_batch( + &mut self, + batch: &RecordBatch, + ) -> Result<()> { + self.aggregate_batch_inner(batch, HashAggregateAccumulator::update_batch) + } + + pub(in crate::aggregates) fn start_output(&mut self) -> Result<()> { + self.start_outputting(); + Ok(()) + } +} diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index d7c72253ecc0c..7819b13e9bc84 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -29,6 +29,7 @@ use crate::aggregates::{ ordered_final_stream::OrderedFinalAggregateStream, ordered_partial_stream::OrderedPartialAggregateStream, partial_reduce_stream::PartialReduceHashAggregateStream, + single_stream::SingleHashAggregateStream, }; use crate::execution_plan::{CardinalityEffect, EmissionType}; use crate::filter_pushdown::{ @@ -85,6 +86,7 @@ pub mod order; mod ordered_final_stream; mod ordered_partial_stream; mod partial_reduce_stream; +mod single_stream; mod skip_partial; mod topk; @@ -539,6 +541,9 @@ enum StreamType { /// Final stage of the hash aggregation /// Input output scheme: partial state -> final result FinalHash(FinalHashAggregateStream), + /// Single stage of the hash aggregation + /// Input output scheme: initial input -> final result + SingleHash(SingleHashAggregateStream), /// Partial stage of aggregation for ordered input. OrderedPartialAggregate(OrderedPartialAggregateStream), /// Final stage of aggregation for ordered input. @@ -567,6 +572,7 @@ impl From for SendableRecordBatchStream { StreamType::PartialHash(stream) => Box::pin(stream), StreamType::PartialReduceHash(stream) => Box::pin(stream), StreamType::FinalHash(stream) => Box::pin(stream), + StreamType::SingleHash(stream) => Box::pin(stream), StreamType::OrderedPartialAggregate(stream) => Box::pin(stream), StreamType::OrderedFinalAggregate(stream) => Box::pin(stream), StreamType::GroupedHash(stream) => Box::pin(stream), @@ -1071,6 +1077,12 @@ impl AggregateExec { self, context, partition, )?)); } + + if self.should_use_single_hash_stream(context) { + return Ok(StreamType::SingleHash(SingleHashAggregateStream::new( + self, context, partition, + )?)); + } } // Execution paths that have not been migrated use the fallback implementation @@ -1133,6 +1145,21 @@ impl AggregateExec { && self.group_by.is_single() } + fn should_use_single_hash_stream(&self, context: &TaskContext) -> bool { + // TODO: implement memory-limited path and remove this limitation + if matches!(context.memory_pool().memory_limit(), MemoryLimit::Finite(_)) { + return false; + } + + matches!( + self.mode, + AggregateMode::Single | AggregateMode::SinglePartitioned + ) && self.limit_options.is_none() + && self.input_order_mode == InputOrderMode::Linear + && !self.group_by.is_true_no_grouping() + && self.group_by.is_single() + } + fn should_use_ordered_final_aggregate_stream(&self, context: &TaskContext) -> bool { // TODO: implement memory-limited path and remove this limitation if matches!(context.memory_pool().memory_limit(), MemoryLimit::Finite(_)) { @@ -3506,6 +3533,80 @@ mod tests { Ok(()) } + fn single_test_aggregate() -> Result { + let schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::UInt32, false), + Field::new("b", DataType::Float64, false), + ])); + let input_batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(UInt32Array::from(vec![1, 2, 1, 3])), + Arc::new(Float64Array::from(vec![10.0, 20.0, 40.0, 30.0])), + ], + )?; + let input = TestMemoryExec::try_new_exec( + &[vec![input_batch]], + Arc::clone(&schema), + None, + )?; + let group_by = + PhysicalGroupBy::new_single(vec![(col("a", &schema)?, "a".to_string())]); + let aggregates: Vec> = vec![Arc::new( + AggregateExprBuilder::new(sum_udaf(), vec![col("b", &schema)?]) + .schema(Arc::clone(&schema)) + .alias("SUM(b)") + .build()?, + )]; + + AggregateExec::try_new( + AggregateMode::Single, + group_by, + aggregates, + vec![None], + input, + schema, + ) + } + + /// For single aggregation, ensures `SingleHashAggregateStream` is used when + /// enabled by migration config. + #[tokio::test] + async fn single_aggregate_planning() -> Result<()> { + let single = single_test_aggregate()?; + let task_ctx = new_migrated_hash_ctx(2); + + let stream = single.execute_typed(0, &task_ctx)?; + assert!(matches!(stream, StreamType::SingleHash(_))); + let stream: SendableRecordBatchStream = stream.into(); + let output = collect(stream).await?; + assert_eq!(output.iter().map(RecordBatch::num_rows).sum::(), 3); + assert_snapshot!(batches_to_sort_string(&output), @r" ++---+--------+ +| a | SUM(b) | ++---+--------+ +| 1 | 50.0 | +| 2 | 20.0 | +| 3 | 30.0 | ++---+--------+ +"); + + Ok(()) + } + + /// Spilling behavior is not implemented for single hash stream yet, so fall + /// back to the existing `GroupedHashAggregateStream`. + #[tokio::test] + async fn single_aggregate_with_memory_limit_planning() -> Result<()> { + let single = single_test_aggregate()?; + let task_ctx = new_finite_memory_migrated_hash_ctx(2, 1024 * 1024)?; + + let stream = single.execute_typed(0, &task_ctx)?; + assert!(matches!(stream, StreamType::GroupedHash(_))); + + Ok(()) + } + fn partial_reduce_test_aggregate() -> Result { let schema = Arc::new(Schema::new(vec![ Field::new("a", DataType::UInt32, false), diff --git a/datafusion/physical-plan/src/aggregates/single_stream.rs b/datafusion/physical-plan/src/aggregates/single_stream.rs new file mode 100644 index 0000000000000..886ffdd3a0b99 --- /dev/null +++ b/datafusion/physical-plan/src/aggregates/single_stream.rs @@ -0,0 +1,379 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Single-stage hash aggregation stream implementation. +//! +//! This stream is part of the incremental migration from +//! [`crate::aggregates::grouped_hash_stream::GroupedHashAggregateStream`]. +//! +//! See issue for details: + +use std::ops::ControlFlow; +use std::sync::Arc; +use std::task::{Context, Poll}; + +use arrow::datatypes::SchemaRef; +use arrow::record_batch::RecordBatch; +use datafusion_common::Result; +use datafusion_execution::TaskContext; +use datafusion_execution::memory_pool::{MemoryConsumer, MemoryReservation}; +use futures::stream::{Stream, StreamExt}; + +use super::AggregateExec; +use super::aggregate_hash_table::{AggregateHashTable, SingleMarker}; +use crate::metrics::{BaselineMetrics, RecordOutput}; +use crate::stream::EmptyRecordBatchStream; +use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream}; + +/// Hash aggregation can run the full logical aggregation in one operator. This +/// stream implements the single stage for grouped hash aggregation. +/// +/// # Example +/// +/// SELECT k, AVG(v) FROM t GROUP BY k; +/// +/// ## Plan +/// AggregateExec(stage=single) +/// +/// ## Single Stage Behavior +/// Input: raw rows +/// Output: final aggregate values for all groups (for example, `AVG(x)`) +/// +/// This stream implements the complete aggregation without a partial/final +/// split. It consumes raw input rows and emits final aggregate values. +pub(crate) struct SingleHashAggregateStream { + /// Output schema: group columns followed by final aggregate value columns. + schema: SchemaRef, + + /// Input batches containing raw rows, not partial aggregate state. + input: SendableRecordBatchStream, + + /// Execution metrics shared with the aggregate plan node. + baseline_metrics: BaselineMetrics, + + /// Memory reservation for group keys and accumulators. + reservation: MemoryReservation, + + /// Tracks the high-level stream lifecycle. The hash table owns the lower-level + /// state for emitting output batches. + state: Option, +} + +/// States for single hash aggregation processing. +// The typestate pattern mirrors the final stream and keeps the input/output +// semantics explicit for this mode. +enum SingleHashAggregateState { + ReadingInput { + hash_table: AggregateHashTable, + }, + ProducingOutput { + hash_table: AggregateHashTable, + }, + Done, +} + +type SingleHashAggregatePoll = Poll>>; +type SingleHashAggregateStateTransition = ControlFlow< + (SingleHashAggregatePoll, SingleHashAggregateState), + SingleHashAggregateState, +>; + +impl SingleHashAggregateState { + fn hash_table(&self) -> &AggregateHashTable { + match self { + Self::ReadingInput { hash_table } | Self::ProducingOutput { hash_table } => { + hash_table + } + Self::Done => unreachable!("Done state does not hold a hash table"), + } + } + + fn hash_table_mut(&mut self) -> &mut AggregateHashTable { + match self { + Self::ReadingInput { hash_table } | Self::ProducingOutput { hash_table } => { + hash_table + } + Self::Done => unreachable!("Done state does not hold a hash table"), + } + } + + fn into_hash_table(self) -> AggregateHashTable { + match self { + Self::ReadingInput { hash_table } | Self::ProducingOutput { hash_table } => { + hash_table + } + Self::Done => unreachable!("Done state does not hold a hash table"), + } + } + + fn into_producing_output(self) -> Self { + Self::ProducingOutput { + hash_table: self.into_hash_table(), + } + } + + fn into_done(self) -> Self { + Self::Done + } +} + +impl SingleHashAggregateStream { + pub fn new( + agg: &AggregateExec, + context: &Arc, + partition: usize, + ) -> Result { + debug_assert!(matches!( + agg.mode, + super::AggregateMode::Single | super::AggregateMode::SinglePartitioned + )); + debug_assert_eq!(agg.input_order_mode, InputOrderMode::Linear); + + let schema = Arc::clone(&agg.schema); + let input = agg.input.execute(partition, Arc::clone(context))?; + let batch_size = context.session_config().batch_size(); + let baseline_metrics = BaselineMetrics::new(&agg.metrics, partition); + + let hash_table = AggregateHashTable::::new( + agg, + partition, + Arc::clone(&schema), + batch_size, + )?; + + let reservation = + MemoryConsumer::new(format!("SingleHashAggregateStream[{partition}]")) + .register(context.memory_pool()); + + Ok(Self { + schema, + input, + baseline_metrics, + reservation, + state: Some(SingleHashAggregateState::ReadingInput { hash_table }), + }) + } + + /// Moves the aggregate hash table's inner state to `Outputting`. + /// + /// The caller guarantees that input is fully consumed, so this function can + /// eagerly release the input stream. + fn start_output( + &mut self, + hash_table: &mut AggregateHashTable, + ) -> Result<()> { + let input_schema = self.input.schema(); + self.input = Box::pin(EmptyRecordBatchStream::new(input_schema)); + hash_table.start_output() + } + + /// Handle ReadingInput state - aggregate input batches into the hash table. + /// + /// See comments at `poll_next()` for details. + /// + /// Returns the next operator state with control flow decision. + fn handle_reading_input( + &mut self, + cx: &mut Context<'_>, + mut original_state: SingleHashAggregateState, + ) -> SingleHashAggregateStateTransition { + debug_assert!(matches!( + &original_state, + SingleHashAggregateState::ReadingInput { .. } + )); + debug_assert!(original_state.hash_table().is_building()); + + match self.input.poll_next_unpin(cx) { + Poll::Pending => ControlFlow::Break((Poll::Pending, original_state)), + // Get a new input batch, aggregate it in the hash table + Poll::Ready(Some(Ok(batch))) => { + let elapsed_compute = self.baseline_metrics.elapsed_compute().clone(); + let timer = elapsed_compute.timer(); + let result = original_state.hash_table_mut().aggregate_batch(&batch); + timer.done(); + + if let Err(e) = result { + return ControlFlow::Break(( + Poll::Ready(Some(Err(e))), + original_state, + )); + } + + if let Err(e) = self + .reservation + .try_resize(original_state.hash_table().memory_size()) + { + return ControlFlow::Break(( + Poll::Ready(Some(Err(e))), + original_state, + )); + } + + ControlFlow::Continue(original_state) + } + Poll::Ready(Some(Err(e))) => { + ControlFlow::Break((Poll::Ready(Some(Err(e))), original_state)) + } + // Input ends, move to output state + Poll::Ready(None) => { + let elapsed_compute = self.baseline_metrics.elapsed_compute().clone(); + let timer = elapsed_compute.timer(); + let result = self.start_output(original_state.hash_table_mut()); + timer.done(); + + match result { + Ok(()) => { + ControlFlow::Continue(original_state.into_producing_output()) + } + Err(e) => { + ControlFlow::Break((Poll::Ready(Some(Err(e))), original_state)) + } + } + } + } + } + + /// Handle ProducingOutput state - emit final aggregate value batches. + /// + /// See comments at `poll_next()` for details. + /// + /// Returns the next operator state with control flow decision. + fn handle_producing_output( + &mut self, + mut original_state: SingleHashAggregateState, + ) -> SingleHashAggregateStateTransition { + debug_assert!(matches!( + &original_state, + SingleHashAggregateState::ProducingOutput { .. } + )); + debug_assert!(!original_state.hash_table().is_building()); + + let elapsed_compute = self.baseline_metrics.elapsed_compute().clone(); + let timer = elapsed_compute.timer(); + let result = original_state.hash_table_mut().next_output_batch(); + timer.done(); + + match result { + Ok(Some(batch)) => { + if let Err(e) = self + .reservation + .try_resize(original_state.hash_table().memory_size()) + { + return ControlFlow::Break(( + Poll::Ready(Some(Err(e))), + original_state, + )); + } + debug_assert!(batch.num_rows() > 0); + let next_state = if original_state.hash_table().is_done() { + original_state.into_done() + } else { + original_state + }; + + ControlFlow::Break(( + Poll::Ready(Some(Ok(batch.record_output(&self.baseline_metrics)))), + next_state, + )) + } + Ok(None) => { + let _ = self.reservation.try_resize(0); + ControlFlow::Continue(original_state.into_done()) + } + Err(e) => ControlFlow::Break((Poll::Ready(Some(Err(e))), original_state)), + } + } +} + +impl Stream for SingleHashAggregateStream { + type Item = Result; + + /// Entry point for the single hash aggregate state machine. + /// + /// See comments in [`SingleHashAggregateStream`] for high-level ideas. + /// + /// State transition graph: + /// + /// ```text + /// (start) + /// -> ReadingInput + /// The stream starts by polling raw input rows and aggregating those + /// rows into the single-stage hash table. + /// + /// ReadingInput + /// -> ReadingInput + /// Aggregate one raw input batch, update the inner aggregate hash + /// table, and continue with the next input batch. + /// + /// -> ProducingOutput + /// Input was exhausted. Move to the next state to start outputting + /// final aggregate values. + /// + /// ProducingOutput + /// -> ProducingOutput + /// One final output batch was yielded; repeat to continue producing + /// output incrementally. + /// + /// -> Done + /// All final output was emitted. + /// + /// Done + /// -> (end) + /// ``` + fn poll_next( + mut self: std::pin::Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + loop { + let cur_state = self + .state + .take() + .expect("SingleHashAggregateStream state should not be None"); + + let next_state = match cur_state { + state @ SingleHashAggregateState::ReadingInput { .. } => { + self.handle_reading_input(cx, state) + } + state @ SingleHashAggregateState::ProducingOutput { .. } => { + self.handle_producing_output(state) + } + state @ SingleHashAggregateState::Done => { + let _ = self.reservation.try_resize(0); + self.state = Some(state); + return Poll::Ready(None); + } + }; + + match next_state { + ControlFlow::Continue(next_state) => { + self.state = Some(next_state); + continue; + } + ControlFlow::Break((poll, next_state)) => { + self.state = Some(next_state); + return poll; + } + } + } + } +} + +impl RecordBatchStream for SingleHashAggregateStream { + fn schema(&self) -> SchemaRef { + Arc::clone(&self.schema) + } +}