Remove GroupsAccumulator::supports_convert_to_state and require convert_to_state (#23489)

lyne7-sc · web-flow · commit 3f8161373061 · 2026-07-24T12:22:36.000Z
## Which issue does this PR close?  - Closes #23081. ## Rationale for this change  Following #23275, all `GroupsAccumulator` implementations now provide `convert_to_state`. The `supports_convert_to_state` capability flag is therefore no longer needed. ## What changes are included in this PR?  - Make `GroupsAccumulator::convert_to_state` a required trait method. - Remove `GroupsAccumulator::supports_convert_to_state` and its implementations. - Remove the corresponding capability checks from hash aggregation. - Simplify skip-partial aggregation to use the required `convert_to_state` implementation directly. - Add a regression test covering the partial hash aggregation skip path. - Document the breaking trait change in the 55.0.0 upgrading guide. - Remove `FFI_GroupsAccumulator::supports_convert_to_state`. This changes the FFI ABI layout, so providers and consumers must be rebuilt against DataFusion 55. ## Are these changes tested?  Yes. Added a regression test verifying that skip-partial aggregation uses the required `convert_to_state` implementation without a capability flag. Existing physical-plan and FFI tests continue to pass. ## Are there any user-facing changes?   Yes. This is a breaking Rust API change for external `GroupsAccumulator` implementations: - `convert_to_state` must now be implemented. - `supports_convert_to_state` should be removed. The migration is documented in the 55.0.0 upgrading guide. The `FFI_GroupsAccumulator` layout has changed. FFI providers and consumers must be rebuilt against DataFusion 55 and must not exchange this struct with older major versions.
diff --git a/datafusion-examples/examples/udf/advanced_udaf.rs b/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>()
diff --git a/datafusion/core/tests/user_defined/user_defined_aggregates.rs b/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>()
     }
diff --git a/datafusion/expr-common/src/groups_accumulator.rs b/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.
diff --git a/datafusion/ffi/src/udaf/groups_accumulator.rs b/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)]
diff --git a/datafusion/functions-aggregate-common/src/aggregate/count_distinct/groups.rs b/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>())
diff --git a/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator.rs b/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.
diff --git a/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/bool_op.rs b/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
-    }
 }
diff --git a/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/prim_op.rs b/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()
     }
diff --git a/datafusion/functions-aggregate/src/approx_distinct.rs b/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
diff --git a/datafusion/functions-aggregate/src/array_agg.rs b/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()
diff --git a/datafusion/functions-aggregate/src/average.rs b/datafusion/functions-aggregate/src/average.rs
@@ -1133,11 +1133,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>()
diff --git a/datafusion/functions-aggregate/src/correlation.rs b/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],
diff --git a/datafusion/functions-aggregate/src/count.rs b/datafusion/functions-aggregate/src/count.rs
@@ -773,11 +773,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>()
     }
diff --git a/datafusion/functions-aggregate/src/first_last.rs b/datafusion/functions-aggregate/src/first_last.rs
@@ -719,11 +719,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],
diff --git a/datafusion/functions-aggregate/src/median.rs b/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()
diff --git a/datafusion/functions-aggregate/src/min_max/min_max_bytes.rs b/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()
     }
diff --git a/datafusion/functions-aggregate/src/min_max/min_max_struct.rs b/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()
     }
diff --git a/datafusion/functions-aggregate/src/percentile_cont.rs b/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()
diff --git a/datafusion/functions-aggregate/src/stddev.rs b/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()
     }
diff --git a/datafusion/functions-aggregate/src/string_agg.rs b/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>>()
diff --git a/datafusion/functions-aggregate/src/variance.rs b/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>()
diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs
@@ -569,10 +569,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,
diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs b/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.
diff --git a/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs b/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;
diff --git a/datafusion/physical-plan/src/aggregates/hash_stream.rs b/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;
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -7132,6 +7132,22 @@ mod tests {
             Ok(vec![self.emit_counts(emit_to)?])
         }
 
+        fn convert_to_state(
+            &self,
+            values: &[ArrayRef],
+            opt_filter: Option<&BooleanArray>,
+        ) -> Result<Vec<ArrayRef>> {
+            assert_eq!(values.len(), 1, "one argument to convert_to_state");
+            let counts = match opt_filter {
+                Some(filter) => filter
+                    .iter()
+                    .map(|value| i64::from(value.unwrap_or(false)))
+                    .collect::<Vec<_>>(),
+                None => vec![1; values[0].len()],
+            };
+            Ok(vec![Arc::new(Int64Array::from(counts))])
+        }
+
         fn merge_batch(
             &mut self,
             _values: &[ArrayRef],
diff --git a/datafusion/spark/src/function/aggregate/avg.rs b/datafusion/spark/src/function/aggregate/avg.rs
@@ -367,11 +367,6 @@ where
             Arc::new(counts) as ArrayRef,
         ])
     }
-
-    fn supports_convert_to_state(&self) -> bool {
-        true
-    }
-
     fn size(&self) -> usize {
         self.counts.capacity() * size_of::<i64>() + self.sums.capacity() * size_of::<T>()
     }
@@ -387,12 +382,6 @@ mod tests {
             Ok(sum / count as f64)
         })
     }
-
-    #[test]
-    fn supports_convert_to_state() {
-        assert!(make_acc().supports_convert_to_state());
-    }
-
     #[test]
     fn convert_to_state_basic() {
         let acc = make_acc();
diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md

Original file line number	Diff line number	Diff line change
`@@ -888,11 +888,6 @@ impl GroupsAccumulator for TestGroupsAccumulator {`
`888`	`888`	`as ArrayRef,`
`889`	`889`	`])`
`890`	`890`	`}`
`891`		`-`
`892`		`- fn supports_convert_to_state(&self) -> bool {`
`893`		`- true`
`894`		`- }`
`895`		`-`
`896`	`891`	`fn size(&self) -> usize {`
`897`	`892`	`size_of::<u64>()`
`898`	`893`	`}`
Original file line number	Diff line number	Diff line change
`@@ -441,10 +441,6 @@ impl GroupsAccumulator for GroupsAccumulatorAdapter {`
`441`	`441`
`442`	`442`	`Ok(arrays)`
`443`	`443`	`}`
`444`		`-`
`445`		`- fn supports_convert_to_state(&self) -> bool {`
`446`		`- true`
`447`		`- }`
`448`	`444`	`}`
`449`	`445`
`450`	`446`	/// Extension trait for [`Vec`] to account for allocations.
Original file line number	Diff line number	Diff line change
`@@ -156,8 +156,4 @@ where`
`156`	`156`
`157`	`157`	`Ok(vec![Arc::new(values_filtered)])`
`158`	`158`	`}`
`159`		`-`
`160`		`- fn supports_convert_to_state(&self) -> bool {`
`161`		`- true`
`162`		`- }`
`163`	`159`	`}`
Original file line number	Diff line number	Diff line change
`@@ -189,11 +189,6 @@ where`
`189`	`189`
`190`	`190`	`Ok(vec![Arc::new(state_values)])`
`191`	`191`	`}`
`192`		`-`
`193`		`- fn supports_convert_to_state(&self) -> bool {`
`194`		`- true`
`195`		`- }`
`196`		`-`
`197`	`192`	`fn size(&self) -> usize {`
`198`	`193`	`self.values.capacity() * size_of::<T::Native>() + self.null_state.size()`
`199`	`194`	`}`