Remove GroupsAccumulator::supports_convert_to_state and require convert_to_state

datafusion-examples/examples/udf/advanced_udaf.rs

@@ -393,11 +393,6 @@ impl GroupsAccumulator for GeometricMeanGroupsAccumulator {
             Arc::new(counts) as ArrayRef,
         ])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         self.counts.capacity() * size_of::<u32>()
             + self.prods.capacity() * size_of::<Float64Type>()

datafusion/core/tests/user_defined/user_defined_aggregates.rs

@@ -888,11 +888,6 @@ impl GroupsAccumulator for TestGroupsAccumulator {
                 as ArrayRef,
         ])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         size_of::<u64>()
     }

datafusion/expr-common/src/groups_accumulator.rs

@@ -18,7 +18,7 @@
 //! Vectorized [`GroupsAccumulator`]
 
 use arrow::array::{ArrayRef, BooleanArray};
-use datafusion_common::{Result, not_impl_err, utils::split_vec_min_alloc};
+use datafusion_common::{Result, utils::split_vec_min_alloc};
 
 /// Describes how many rows should be emitted during grouping.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -231,17 +231,9 @@ pub trait GroupsAccumulator: Send + std::any::Any {
     /// [`Accumulator::state`]: crate::accumulator::Accumulator::state
     fn convert_to_state(
         &self,
-        _values: &[ArrayRef],
-        _opt_filter: Option<&BooleanArray>,
-    ) -> Result<Vec<ArrayRef>> {
-        not_impl_err!("Input batch conversion to state not implemented")
-    }
-
-    /// Returns `true` if [`Self::convert_to_state`] is implemented to support
-    /// intermediate aggregate state conversion.
-    fn supports_convert_to_state(&self) -> bool {
-        false
-    }
+        values: &[ArrayRef],
+        opt_filter: Option<&BooleanArray>,
+    ) -> Result<Vec<ArrayRef>>;
 
     /// Amount of memory used to store the state of this accumulator,
     /// in bytes.

datafusion/ffi/src/udaf/groups_accumulator.rs

@@ -73,8 +73,6 @@ pub struct FFI_GroupsAccumulator {
         opt_filter: FFI_Option<WrappedArray>,
     ) -> FFI_Result<SVec<WrappedArray>>,
 
-    pub supports_convert_to_state: bool,
-
     /// Release the memory of the private data when it is no longer being used.
     pub release: unsafe extern "C" fn(accumulator: &mut Self),
 
@@ -247,7 +245,6 @@ impl From<Box<dyn GroupsAccumulator>> for FFI_GroupsAccumulator {
             return accumulator.accumulator;
         }
 
-        let supports_convert_to_state = accumulator.supports_convert_to_state();
         let private_data = GroupsAccumulatorPrivateData { accumulator };
 
         Self {
@@ -257,7 +254,6 @@ impl From<Box<dyn GroupsAccumulator>> for FFI_GroupsAccumulator {
             state: state_fn_wrapper,
             merge_batch: merge_batch_fn_wrapper,
             convert_to_state: convert_to_state_fn_wrapper,
-            supports_convert_to_state,
 
             release: release_fn_wrapper,
             private_data: Box::into_raw(Box::new(private_data)) as *mut c_void,
@@ -421,10 +417,6 @@ impl GroupsAccumulator for ForeignGroupsAccumulator {
                 .collect()
         }
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        self.accumulator.supports_convert_to_state
-    }
 }
 
 #[repr(C)]

datafusion/functions-aggregate-common/src/aggregate/count_distinct/groups.rs

@@ -207,11 +207,6 @@ where
 
         Ok(vec![Arc::new(builder.finish())])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         size_of::<Self>()
             + self.seen.capacity() * (size_of::<(usize, T::Native)>() + size_of::<u64>())

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

@@ -441,10 +441,6 @@ impl GroupsAccumulator for GroupsAccumulatorAdapter {
 
         Ok(arrays)
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
 }
 
 /// Extension trait for [`Vec`] to account for allocations.

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

@@ -156,8 +156,4 @@ where
 
         Ok(vec![Arc::new(values_filtered)])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
 }

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

@@ -189,11 +189,6 @@ where
 
         Ok(vec![Arc::new(state_values)])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         self.values.capacity() * size_of::<T::Native>() + self.null_state.size()
     }

datafusion/functions-aggregate/src/approx_distinct.rs

@@ -615,11 +615,6 @@ impl GroupsAccumulator for HllGroupsAccumulator {
 
         Ok(vec![Arc::new(builder.finish())])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         self.groups.capacity() * size_of::<GroupHll>()
             + self.allocated_bytes

datafusion/functions-aggregate/src/array_agg.rs

@@ -793,11 +793,6 @@ impl GroupsAccumulator for ArrayAggGroupsAccumulator {
 
         Ok(vec![Arc::new(list_array)])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         self.batches
             .iter()

datafusion/functions-aggregate/src/average.rs

@@ -964,11 +964,6 @@ where
 
         Ok(vec![Arc::new(counts) as ArrayRef, Arc::new(sums)])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         // Heap buffers
         self.counts.capacity() * size_of::<u64>()

datafusion/functions-aggregate/src/correlation.rs

@@ -539,11 +539,6 @@ impl GroupsAccumulator for CorrelationGroupsAccumulator {
             Arc::new(Float64Array::from(sum_yy)),
         ])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn merge_batch(
         &mut self,
         values: &[ArrayRef],

datafusion/functions-aggregate/src/count.rs

@@ -763,11 +763,6 @@ impl GroupsAccumulator for CountGroupsAccumulator {
 
         Ok(vec![state_array])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         self.counts.capacity() * size_of::<usize>()
     }

datafusion/functions-aggregate/src/first_last.rs

@@ -712,11 +712,6 @@ impl<S: ValueState + 'static> GroupsAccumulator for FirstLastGroupsAccumulator<S
             + self.extreme_of_each_group_buf.0.capacity() * size_of::<usize>()
             + self.extreme_of_each_group_buf.1.capacity() / 8
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn convert_to_state(
         &self,
         values: &[ArrayRef],

datafusion/functions-aggregate/src/median.rs

@@ -535,11 +535,6 @@ impl<T: ArrowNumericType + Send> GroupsAccumulator for MedianGroupsAccumulator<T
 
         Ok(vec![Arc::new(converted_list_array)])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         self.group_values
             .iter()

datafusion/functions-aggregate/src/min_max/min_max_bytes.rs

@@ -325,11 +325,6 @@ impl GroupsAccumulator for MinMaxBytesAccumulator {
         let output = apply_filter_as_nulls(&values[0], opt_filter)?;
         Ok(vec![output])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         self.inner.size()
     }

datafusion/functions-aggregate/src/min_max/min_max_struct.rs

@@ -150,11 +150,6 @@ impl GroupsAccumulator for MinMaxStructAccumulator {
         let output = apply_filter_as_nulls(&values[0], opt_filter)?;
         Ok(vec![output])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         self.inner.size()
     }

datafusion/functions-aggregate/src/percentile_cont.rs

@@ -652,11 +652,6 @@ where
 
         Ok(vec![Arc::new(converted_list_array)])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         self.group_values
             .iter()

datafusion/functions-aggregate/src/stddev.rs

@@ -352,11 +352,6 @@ impl GroupsAccumulator for StddevGroupsAccumulator {
     ) -> Result<Vec<ArrayRef>> {
         self.variance.convert_to_state(values, opt_filter)
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         self.variance.size()
     }

datafusion/functions-aggregate/src/string_agg.rs

@@ -432,11 +432,6 @@ impl GroupsAccumulator for StringAggGroupsAccumulator {
         };
         Ok(vec![result])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         self.total_data_bytes
             + self.values.capacity() * size_of::<Option<String>>()

datafusion/functions-aggregate/src/variance.rs

@@ -613,11 +613,6 @@ impl GroupsAccumulator for VarianceGroupsAccumulator {
             Arc::new(Float64Array::new(m2s.into(), None)),
         ])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         self.m2s.capacity() * size_of::<f64>()
             + self.means.capacity() * size_of::<f64>()

datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs

@@ -567,10 +567,6 @@ impl HashAggregateAccumulator {
         self.accumulator.state(emit_to)
     }
 
-    pub(super) fn supports_convert_to_state(&self) -> bool {
-        self.accumulator.supports_convert_to_state()
-    }
-
     pub(super) fn convert_to_state(
         &mut self,
         values: &EvaluatedAccumulatorArgs,

datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs

@@ -68,14 +68,6 @@ impl AggregateHashTable<PartialMarker> {
         self.next_output_batch_inner(HashAggregateAccumulator::state)
     }
 
-    pub(in crate::aggregates) fn can_skip_aggregation(&self) -> bool {
-        self.state
-            .building()
-            .accumulators
-            .iter()
-            .all(|acc| acc.supports_convert_to_state())
-    }
-
     /// In skip-partial-aggregation optimization, when a decision has been made to skip
     /// partial stage, build a typed hash table only for aggregation state conversion
     /// row-by-row.

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

@@ -217,8 +217,7 @@ enum OutOfMemoryMode {
 /// aggregator must store the intermediate state for each group.
 ///
 /// If the ratio of the number of groups to the number of input rows exceeds a
-/// threshold, and [`GroupsAccumulator::supports_convert_to_state`] is
-/// supported, this operator will stop applying Partial aggregation and directly
+/// threshold, this operator will stop applying Partial aggregation and directly
 /// pass the input rows to the next aggregation phase.
 ///
 /// [`Accumulator::state`]: datafusion_expr::Accumulator::state
@@ -545,14 +544,9 @@ impl GroupedHashAggregateStream {
         // - aggregation mode is Partial
         // - input is not ordered by GROUP BY expressions,
         //   since Final mode expects unique group values as its input
-        // - all accumulators support input batch to intermediate
-        //   aggregate state conversion
         // - there is only one GROUP BY expressions set
         let skip_aggregation_probe = if agg.mode == AggregateMode::Partial
             && matches!(group_ordering, GroupOrdering::None)
-            && accumulators
-                .iter()
-                .all(|acc| acc.supports_convert_to_state())
             && agg_group_by.is_single()
         {
             let options = &context.session_config().options().execution;

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

@@ -293,9 +293,7 @@ impl PartialHashAggregateStream {
             Arc::clone(&schema),
             batch_size,
         )?;
-        let can_skip_aggregation =
-            agg.group_by.is_single() && hash_table.can_skip_aggregation();
-        let skip_aggregation_probe = if can_skip_aggregation {
+        let skip_aggregation_probe = if agg.group_by.is_single() {
             let options = &context.session_config().options().execution;
             let probe_ratio_threshold =
                 options.skip_partial_aggregation_probe_ratio_threshold;