chore: Minor cleanups

datafusion/src/cube_ext/joinagg.rs

@@ -259,7 +259,6 @@ impl ExecutionPlan for CrossJoinAggExec {
                         accumulators = hash_aggregate::group_aggregate_batch(
                             &AggregateMode::Full,
                             &group_expr,
-                            &self.agg_expr,
                             joined,
                             std::mem::take(&mut accumulators),
                             &aggs,

datafusion/src/physical_plan/distinct_expressions.rs

@@ -27,16 +27,15 @@ use arrow::datatypes::{DataType, Field};
 
 use crate::error::{DataFusionError, Result};
 use crate::physical_plan::group_scalar::GroupByScalar;
+use crate::physical_plan::groups_accumulator::GroupsAccumulator;
+use crate::physical_plan::groups_accumulator_flat_adapter::GroupsAccumulatorFlatAdapter;
 use crate::physical_plan::{Accumulator, AggregateExpr, PhysicalExpr};
 use crate::scalar::ScalarValue;
 use itertools::Itertools;
 use smallvec::SmallVec;
 use std::collections::hash_map::RandomState;
 use std::collections::HashSet;
 
-use super::groups_accumulator::GroupsAccumulator;
-use super::groups_accumulator_flat_adapter::GroupsAccumulatorFlatAdapter;
-
 #[derive(Debug, PartialEq, Eq, Hash, Clone)]
 struct DistinctScalarValues(Vec<GroupByScalar>);
 

datafusion/src/physical_plan/expressions/min_max.rs

@@ -105,8 +105,6 @@ impl AggregateExpr for Max {
         return true;
     }
 
-    /// the groups accumulator used to accumulate values from the expression.  If this returns None,
-    /// create_accumulator must be used.
     fn create_groups_accumulator(
         &self,
     ) -> arrow::error::Result<Option<Box<dyn GroupsAccumulator>>> {

datafusion/src/physical_plan/hash_aggregate.rs

@@ -408,16 +408,17 @@ pin_project! {
     }
 }
 
-// TODO: _aggr_expr is currently unused; it's kept for perhaps, debugging usage, but probably just remove it.
 pub(crate) fn group_aggregate_batch(
     mode: &AggregateMode,
     group_expr: &[Arc<dyn PhysicalExpr>],
-    _aggr_expr: &[Arc<dyn AggregateExpr>],
     batch: RecordBatch,
     mut accumulation_state: AccumulationState,
     aggregate_expressions: &[Vec<Arc<dyn PhysicalExpr>>],
     skip_row: impl Fn(&RecordBatch, /*row_index*/ usize) -> bool,
 ) -> Result<AccumulationState> {
+    // Note: There is some parallel array &[Arc<dyn AggregateExpr>] that simply isn't passed to this
+    // function, but which exists and might be useful.
+
     // evaluate the grouping expressions
     let group_values = evaluate(group_expr, &batch)?;
 
@@ -813,7 +814,6 @@ async fn compute_grouped_hash_aggregate(
         accumulators = group_aggregate_batch(
             &mode,
             &group_expr,
-            &aggr_expr,
             batch,
             accumulators,
             &aggregate_expressions,