(datafusion) branch main updated: minor: move `create_group_accumulator` out of the legacy aggregate stream (#25539)

[gi...@apache.org]

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 9bffbea2a4 minor: move `create_group_accumulator` out of the legacy aggregate stream (#25539)
9bffbea2a4 is described below

commit 9bffbea2a4ccaa6a57aee303da01a72dc91f4f8e
Author: Jay Zhan <ja...@gmail.com>
AuthorDate: Sun Sep 20 13:35:59 2026 +0000

    minor: move `create_group_accumulator` out of the legacy aggregate stream (#25539)
    
    ## Which issue does this PR close?
    
    - Part of #25537. Follow-up to #22710.
    
    ## Rationale for this change
    
    `create_group_accumulator` is defined in `grouped_hash_stream.rs`, the
    legacy `GroupedHashAggregateStream` that is only reachable with
    `datafusion.execution.enable_migration_aggregate = false` and is
    scheduled for removal. It is, however, the one piece of that file the
    new code depends on: both `AggregateHashTable`
    (`aggregate_hash_table/common.rs`) and `OrderedAggregateTable`
    (`aggregate_hash_table/common_ordered.rs`) import it from there.
    
    Moving it next to its real users removes the only dependency of the new
    aggregation code on the legacy file, so the legacy stream can later be
    deleted without touching the new streams.
    
    ## What changes are included in this PR?
    
    Pure code move, no behaviour change.
    
    - Move `create_group_accumulator` to `aggregate_hash_table/common.rs`
    and re-export it from `aggregate_hash_table`.
    - `grouped_hash_stream.rs` now imports it from there (dependency
    direction reversed: legacy → new).
    
    ## What is the testing strategy for this PR?
    
    Existing tests; no new behaviour.
    
    ## Are there any user-facing changes?
    
    No.
---
 .../src/aggregates/aggregate_hash_table/common.rs  | 26 +++++++++++++++++-
 .../aggregate_hash_table/common_ordered.rs         |  3 +-
 .../src/aggregates/aggregate_hash_table/mod.rs     |  2 +-
 .../src/aggregates/grouped_hash_stream.rs          | 32 +++-------------------
 4 files changed, 31 insertions(+), 32 deletions(-)

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 f517f6a998..0c5c4185e2 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs
@@ -29,14 +29,15 @@ use arrow::record_batch::RecordBatch;
 use datafusion_common::{Result, internal_err};
 use datafusion_execution::memory_pool::proxy::VecAllocExt;
 use datafusion_expr::{AggregateMetrics, EmitTo, GroupsAccumulator};
+use datafusion_physical_expr::GroupsAccumulatorAdapter;
 use datafusion_physical_expr::aggregate::AggregateFunctionExpr;
+use log::debug;
 
 use crate::PhysicalExpr;
 use crate::aggregates::group_values::{
     AccumulatorPhase, AggregateAccumulatorMetrics, AggregateArgumentMetrics,
     GroupByMetrics, GroupValues, new_group_values,
 };
-use crate::aggregates::grouped_hash_stream::create_group_accumulator;
 use crate::aggregates::order::GroupOrdering;
 use crate::aggregates::{
     AggregateExec, PhysicalGroupBy, aggregate_expressions, evaluate_group_by,
@@ -56,6 +57,29 @@ pub(in crate::aggregates) struct PartialSkipMarker;
 /// Marker for partial state -> final value aggregation.
 pub(in crate::aggregates) struct FinalMarker;
 
+/// Create an accumulator for `agg_expr` -- a [`GroupsAccumulator`] if
+/// that is supported by the aggregate, or a
+/// [`GroupsAccumulatorAdapter`] if not.
+pub(in crate::aggregates) fn create_group_accumulator(
+    agg_expr: &Arc<AggregateFunctionExpr>,
+    metrics: Arc<dyn AggregateMetrics>,
+) -> Result<Box<dyn GroupsAccumulator>> {
+    if agg_expr.groups_accumulator_supported() {
+        agg_expr.create_groups_accumulator_with_metrics(metrics)
+    } else {
+        // Note in the log when the slow path is used
+        debug!(
+            "Creating GroupsAccumulatorAdapter for {}: {agg_expr:?}",
+            agg_expr.name()
+        );
+        let agg_expr = Arc::clone(agg_expr);
+        let mut adapter =
+            GroupsAccumulatorAdapter::new(move || agg_expr.create_accumulator());
+        adapter.set_metrics(metrics);
+        Ok(Box::new(adapter))
+    }
+}
+
 /// Grouped hash table shared by the partial and final paths.
 ///
 /// While building, it consumes input batches and updates group / accumulator
diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs
index 6446c2ecc1..26a1beb0a2 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs
@@ -34,7 +34,6 @@ use crate::aggregates::group_values::{
     AccumulatorPhase, AggregateAccumulatorMetrics, AggregateArgumentMetrics,
     GroupByMetrics, GroupValues, new_group_values,
 };
-use crate::aggregates::grouped_hash_stream::create_group_accumulator;
 use crate::aggregates::order::GroupOrdering;
 use crate::aggregates::{
     AggregateExec, AggregateMode, PhysicalGroupBy, aggregate_expressions,
@@ -44,7 +43,7 @@ use crate::aggregates::{
 use super::AggregateTableMetrics;
 use super::common::{
     AggregateAccumulator, AggregateBatchFn, AggregateHashTable, EvaluatedAggregateBatch,
-    MaterializeAccumulatorFn,
+    MaterializeAccumulatorFn, create_group_accumulator,
 };
 
 #[derive(Clone)]
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 29ef4a662f..005b2bac09 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/mod.rs
@@ -99,7 +99,7 @@ impl AggregateTableMetrics {
 
 pub(super) use common::{
     AggregateHashTable, FinalMarker, PartialMarker, PartialReduceMarker,
-    PartialSkipMarker, SingleMarker,
+    PartialSkipMarker, SingleMarker, create_group_accumulator,
 };
 pub(super) use common_ordered::{OrderedAggregateTable, OrderedAggregateTableMetrics};
 
diff --git a/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs b/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
index 092a1b26b3..042f4a9044 100644
--- a/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
@@ -34,7 +34,7 @@ use std::sync::Arc;
 use std::task::{Context, Poll};
 use std::vec;
 
-use super::aggregate_hash_table::accumulator_phases;
+use super::aggregate_hash_table::{accumulator_phases, create_group_accumulator};
 use super::order::GroupOrdering;
 use super::skip_partial::SkipAggregationProbe;
 use super::{AggregateExec, format_human_display};
@@ -64,10 +64,9 @@ use datafusion_common::{
 use datafusion_execution::TaskContext;
 use datafusion_execution::memory_pool::proxy::VecAllocExt;
 use datafusion_execution::memory_pool::{MemoryConsumer, MemoryReservation};
-use datafusion_expr::{AggregateMetrics, EmitTo, GroupsAccumulator};
-use datafusion_physical_expr::aggregate::AggregateFunctionExpr;
+use datafusion_expr::{EmitTo, GroupsAccumulator};
+use datafusion_physical_expr::PhysicalSortExpr;
 use datafusion_physical_expr::expressions::Column;
-use datafusion_physical_expr::{GroupsAccumulatorAdapter, PhysicalSortExpr};
 use datafusion_physical_expr_common::sort_expr::LexOrdering;
 use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays;
 
@@ -652,29 +651,6 @@ impl GroupedHashAggregateStream {
     }
 }
 
-/// Create an accumulator for `agg_expr` -- a [`GroupsAccumulator`] if
-/// that is supported by the aggregate, or a
-/// [`GroupsAccumulatorAdapter`] if not.
-pub(crate) fn create_group_accumulator(
-    agg_expr: &Arc<AggregateFunctionExpr>,
-    metrics: Arc<dyn AggregateMetrics>,
-) -> Result<Box<dyn GroupsAccumulator>> {
-    if agg_expr.groups_accumulator_supported() {
-        agg_expr.create_groups_accumulator_with_metrics(metrics)
-    } else {
-        // Note in the log when the slow path is used
-        debug!(
-            "Creating GroupsAccumulatorAdapter for {}: {agg_expr:?}",
-            agg_expr.name()
-        );
-        let agg_expr = Arc::clone(agg_expr);
-        let mut adapter =
-            GroupsAccumulatorAdapter::new(move || agg_expr.create_accumulator());
-        adapter.set_metrics(metrics);
-        Ok(Box::new(adapter))
-    }
-}
-
 impl Stream for GroupedHashAggregateStream {
     type Item = Result<RecordBatch>;
 
@@ -1512,7 +1488,7 @@ mod tests {
     use arrow::array::{Int32Array, Int64Array, UInt32Array};
     use arrow::datatypes::{DataType, Field, Schema};
     use datafusion_execution::runtime_env::RuntimeEnvBuilder;
-    use datafusion_expr::AggregateMetric;
+    use datafusion_expr::{AggregateMetric, AggregateMetrics};
     use datafusion_functions_aggregate::{array_agg::array_agg_udaf, count::count_udaf};
     use datafusion_physical_expr::aggregate::AggregateExprBuilder;
     use datafusion_physical_expr::expressions::col;


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org