Ensure existing usages in DF are calling state_mut and execute_mut

Author: mwylde

datafusion/functions-aggregate-common/src/aggregate/groups_accumulator.rs

@@ -325,9 +325,9 @@ impl GroupsAccumulator for GroupsAccumulatorAdapter {
 
         let results: Vec<ScalarValue> = states
             .into_iter()
-            .map(|state| {
+            .map(|mut state| {
                 self.free_allocation(state.size());
-                state.accumulator.evaluate()
+                state.accumulator.evaluate_mut()
             })
             .collect::<Result<_>>()?;
 
@@ -408,7 +408,7 @@ impl GroupsAccumulator for GroupsAccumulatorAdapter {
         // If there are no rows, return empty arrays
         if num_rows == 0 {
             // create empty accumulator to get the state types
-            let empty_state = (self.factory)()?.state()?;
+            let empty_state = (self.factory)()?.state_mut()?;
             let empty_arrays = empty_state
                 .into_iter()
                 .map(|state_val| new_empty_array(&state_val.data_type()))
@@ -427,7 +427,7 @@ impl GroupsAccumulator for GroupsAccumulatorAdapter {
             let values_to_accumulate =
                 slice_and_maybe_filter(values, opt_filter, &[row_idx, row_idx + 1])?;
             converted_accumulator.update_batch(&values_to_accumulate)?;
-            let states = converted_accumulator.state()?;
+            let states = converted_accumulator.state_mut()?;
 
             // Resize results to have enough columns according to the converted states
             results.resize_with(states.len(), || Vec::with_capacity(num_rows));

datafusion/functions-aggregate-common/src/utils.rs

@@ -37,7 +37,7 @@ pub fn get_accum_scalar_values_as_arrays(
     accum: &mut dyn Accumulator,
 ) -> Result<Vec<ArrayRef>> {
     accum
-        .state()?
+        .state_mut()?
         .iter()
         .map(|s| s.to_array_of_size(1))
         .collect()

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

@@ -210,7 +210,7 @@ impl AggregateWindowExpr for PlainAggregateWindowExpr {
                     .collect();
                 accumulator.update_batch(&update)?
             }
-            accumulator.evaluate()
+            accumulator.evaluate_mut()
         }
     }
 }

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

@@ -207,7 +207,7 @@ impl AggregateWindowExpr for SlidingAggregateWindowExpr {
                     .collect();
                 accumulator.retract_batch(&retract)?
             }
-            accumulator.evaluate()
+            accumulator.evaluate_mut()
         }
     }
 }

datafusion/physical-plan/src/aggregates/mod.rs

@@ -1308,7 +1308,7 @@ pub fn finalize_aggregation(
             accumulators
                 .iter_mut()
                 .map(|accumulator| {
-                    accumulator.state().and_then(|e| {
+                    accumulator.state_mut().and_then(|e| {
                         e.iter()
                             .map(|v| v.to_array())
                             .collect::<Result<Vec<ArrayRef>>>()
@@ -1324,7 +1324,7 @@ pub fn finalize_aggregation(
             // Merge the state to the final value
             accumulators
                 .iter_mut()
-                .map(|accumulator| accumulator.evaluate().and_then(|v| v.to_array()))
+                .map(|accumulator| accumulator.evaluate_mut().and_then(|v| v.to_array()))
                 .collect()
         }
     }