Removed Arc wrapping for AggregateFunctionExpr (#12353)

Author: athultr1997

datafusion/core/src/physical_optimizer/update_aggr_exprs.rs

@@ -118,7 +118,7 @@ impl PhysicalOptimizerRule for OptimizeAggregateOrder {
 ///
 /// # Parameters
 ///
-/// * `aggr_exprs` - A vector of `Arc<AggregateFunctionExpr>` representing the
+/// * `aggr_exprs` - A vector of `AggregateFunctionExpr` representing the
 ///   aggregate expressions to be optimized.
 /// * `prefix_requirement` - An array slice representing the ordering
 ///   requirements preceding the aggregate expressions.
@@ -131,10 +131,10 @@ impl PhysicalOptimizerRule for OptimizeAggregateOrder {
 /// successfully. Any errors occurring during the conversion process are
 /// passed through.
 fn try_convert_aggregate_if_better(
-    aggr_exprs: Vec<Arc<AggregateFunctionExpr>>,
+    aggr_exprs: Vec<AggregateFunctionExpr>,
     prefix_requirement: &[PhysicalSortRequirement],
     eq_properties: &EquivalenceProperties,
-) -> Result<Vec<Arc<AggregateFunctionExpr>>> {
+) -> Result<Vec<AggregateFunctionExpr>> {
     aggr_exprs
         .into_iter()
         .map(|aggr_expr| {

datafusion/core/src/physical_planner.rs

@@ -1541,7 +1541,7 @@ pub fn create_window_expr(
 }
 
 type AggregateExprWithOptionalArgs = (
-    Arc<AggregateFunctionExpr>,
+    AggregateFunctionExpr,
     // The filter clause, if any
     Option<Arc<dyn PhysicalExpr>>,
     // Ordering requirements, if any

datafusion/core/src/test_util/mod.rs

@@ -427,7 +427,7 @@ impl TestAggregate {
     }
 
     /// Return appropriate expr depending if COUNT is for col or table (*)
-    pub fn count_expr(&self, schema: &Schema) -> Arc<AggregateFunctionExpr> {
+    pub fn count_expr(&self, schema: &Schema) -> AggregateFunctionExpr {
         AggregateExprBuilder::new(count_udaf(), vec![self.column()])
             .schema(Arc::new(schema.clone()))
             .alias(self.column_name())

datafusion/core/tests/physical_optimizer/combine_partial_final_agg.rs

@@ -84,7 +84,7 @@ fn parquet_exec(schema: &SchemaRef) -> Arc<ParquetExec> {
 fn partial_aggregate_exec(
     input: Arc<dyn ExecutionPlan>,
     group_by: PhysicalGroupBy,
-    aggr_expr: Vec<Arc<AggregateFunctionExpr>>,
+    aggr_expr: Vec<AggregateFunctionExpr>,
 ) -> Arc<dyn ExecutionPlan> {
     let schema = input.schema();
     let n_aggr = aggr_expr.len();
@@ -104,7 +104,7 @@ fn partial_aggregate_exec(
 fn final_aggregate_exec(
     input: Arc<dyn ExecutionPlan>,
     group_by: PhysicalGroupBy,
-    aggr_expr: Vec<Arc<AggregateFunctionExpr>>,
+    aggr_expr: Vec<AggregateFunctionExpr>,
 ) -> Arc<dyn ExecutionPlan> {
     let schema = input.schema();
     let n_aggr = aggr_expr.len();
@@ -130,7 +130,7 @@ fn count_expr(
     expr: Arc<dyn PhysicalExpr>,
     name: &str,
     schema: &Schema,
-) -> Arc<AggregateFunctionExpr> {
+) -> AggregateFunctionExpr {
     AggregateExprBuilder::new(count_udaf(), vec![expr])
         .schema(Arc::new(schema.clone()))
         .alias(name)

datafusion/physical-expr/src/aggregate.rs

@@ -88,7 +88,7 @@ impl AggregateExprBuilder {
         }
     }
 
-    pub fn build(self) -> Result<Arc<AggregateFunctionExpr>> {
+    pub fn build(self) -> Result<AggregateFunctionExpr> {
         let Self {
             fun,
             args,
@@ -132,7 +132,7 @@ impl AggregateExprBuilder {
             Some(alias) => alias,
         };
 
-        Ok(Arc::new(AggregateFunctionExpr {
+        Ok(AggregateFunctionExpr {
             fun: Arc::unwrap_or_clone(fun),
             args,
             data_type,
@@ -145,7 +145,7 @@ impl AggregateExprBuilder {
             input_types: input_exprs_types,
             is_reversed,
             is_nullable,
-        }))
+        })
     }
 
     pub fn alias(mut self, alias: impl Into<String>) -> Self {
@@ -328,9 +328,9 @@ impl AggregateFunctionExpr {
     /// not implement the method, returns an error. Order insensitive and hard
     /// requirement aggregators return `Ok(None)`.
     pub fn with_beneficial_ordering(
-        self: Arc<Self>,
+        self,
         beneficial_ordering: bool,
-    ) -> Result<Option<Arc<AggregateFunctionExpr>>> {
+    ) -> Result<Option<AggregateFunctionExpr>> {
         let Some(updated_fn) = self
             .fun
             .clone()
@@ -457,10 +457,10 @@ impl AggregateFunctionExpr {
     /// Typically the "reverse" expression is itself (e.g. SUM, COUNT).
     /// For aggregates that do not support calculation in reverse,
     /// returns None (which is the default value).
-    pub fn reverse_expr(&self) -> Option<Arc<AggregateFunctionExpr>> {
+    pub fn reverse_expr(&self) -> Option<AggregateFunctionExpr> {
         match self.fun.reverse_udf() {
             ReversedUDAF::NotSupported => None,
-            ReversedUDAF::Identical => Some(Arc::new(self.clone())),
+            ReversedUDAF::Identical => Some(self.clone()),
             ReversedUDAF::Reversed(reverse_udf) => {
                 let reverse_ordering_req = reverse_order_bys(&self.ordering_req);
                 let mut name = self.name().to_string();
@@ -507,7 +507,7 @@ impl AggregateFunctionExpr {
         &self,
         _args: Vec<Arc<dyn PhysicalExpr>>,
         _order_by_exprs: Vec<Arc<dyn PhysicalExpr>>,
-    ) -> Option<Arc<AggregateFunctionExpr>> {
+    ) -> Option<AggregateFunctionExpr> {
         None
     }
 

datafusion/physical-expr/src/window/aggregate.rs

@@ -41,7 +41,7 @@ use crate::{expressions::PhysicalSortExpr, reverse_order_bys, PhysicalExpr};
 /// See comments on [`WindowExpr`] for more details.
 #[derive(Debug)]
 pub struct PlainAggregateWindowExpr {
-    aggregate: Arc<AggregateFunctionExpr>,
+    aggregate: AggregateFunctionExpr,
     partition_by: Vec<Arc<dyn PhysicalExpr>>,
     order_by: Vec<PhysicalSortExpr>,
     window_frame: Arc<WindowFrame>,
@@ -50,7 +50,7 @@ pub struct PlainAggregateWindowExpr {
 impl PlainAggregateWindowExpr {
     /// Create a new aggregate window function expression
     pub fn new(
-        aggregate: Arc<AggregateFunctionExpr>,
+        aggregate: AggregateFunctionExpr,
         partition_by: &[Arc<dyn PhysicalExpr>],
         order_by: &[PhysicalSortExpr],
         window_frame: Arc<WindowFrame>,
@@ -64,7 +64,7 @@ impl PlainAggregateWindowExpr {
     }
 
     /// Get aggregate expr of AggregateWindowExpr
-    pub fn get_aggregate_expr(&self) -> &Arc<AggregateFunctionExpr> {
+    pub fn get_aggregate_expr(&self) -> &AggregateFunctionExpr {
         &self.aggregate
     }
 }

datafusion/physical-expr/src/window/sliding_aggregate.rs

@@ -41,7 +41,7 @@ use crate::{expressions::PhysicalSortExpr, reverse_order_bys, PhysicalExpr};
 /// See comments on [`WindowExpr`] for more details.
 #[derive(Debug)]
 pub struct SlidingAggregateWindowExpr {
-    aggregate: Arc<AggregateFunctionExpr>,
+    aggregate: AggregateFunctionExpr,
     partition_by: Vec<Arc<dyn PhysicalExpr>>,
     order_by: Vec<PhysicalSortExpr>,
     window_frame: Arc<WindowFrame>,
@@ -50,7 +50,7 @@ pub struct SlidingAggregateWindowExpr {
 impl SlidingAggregateWindowExpr {
     /// Create a new (sliding) aggregate window function expression.
     pub fn new(
-        aggregate: Arc<AggregateFunctionExpr>,
+        aggregate: AggregateFunctionExpr,
         partition_by: &[Arc<dyn PhysicalExpr>],
         order_by: &[PhysicalSortExpr],
         window_frame: Arc<WindowFrame>,
@@ -64,7 +64,7 @@ impl SlidingAggregateWindowExpr {
     }
 
     /// Get the [AggregateFunctionExpr] of this object.
-    pub fn get_aggregate_expr(&self) -> &Arc<AggregateFunctionExpr> {
+    pub fn get_aggregate_expr(&self) -> &AggregateFunctionExpr {
         &self.aggregate
     }
 }

datafusion/physical-optimizer/src/combine_partial_final_agg.rs

@@ -125,7 +125,7 @@ impl PhysicalOptimizerRule for CombinePartialFinalAggregate {
 
 type GroupExprsRef<'a> = (
     &'a PhysicalGroupBy,
-    &'a [Arc<AggregateFunctionExpr>],
+    &'a [AggregateFunctionExpr],
     &'a [Option<Arc<dyn PhysicalExpr>>],
 );