From 96c4fce062ee0462e882900d84823414edb9f5de Mon Sep 17 00:00:00 2001 From: linfeng Date: Sat, 11 Jul 2026 18:12:43 +0800 Subject: [PATCH] Remove GroupsAccumulator convert-to-state capability flag --- .../examples/udf/advanced_udaf.rs | 5 - .../user_defined/user_defined_aggregates.rs | 5 - .../expr-common/src/groups_accumulator.rs | 16 +-- datafusion/ffi/src/udaf/groups_accumulator.rs | 8 -- .../src/aggregate/count_distinct/groups.rs | 5 - .../src/aggregate/groups_accumulator.rs | 4 - .../aggregate/groups_accumulator/bool_op.rs | 4 - .../aggregate/groups_accumulator/prim_op.rs | 5 - .../src/approx_distinct.rs | 5 - .../functions-aggregate/src/array_agg.rs | 5 - datafusion/functions-aggregate/src/average.rs | 5 - .../functions-aggregate/src/correlation.rs | 5 - datafusion/functions-aggregate/src/count.rs | 5 - .../functions-aggregate/src/first_last.rs | 5 - datafusion/functions-aggregate/src/median.rs | 5 - .../src/min_max/min_max_bytes.rs | 5 - .../src/min_max/min_max_struct.rs | 5 - .../src/percentile_cont.rs | 5 - datafusion/functions-aggregate/src/stddev.rs | 5 - .../functions-aggregate/src/string_agg.rs | 5 - .../functions-aggregate/src/variance.rs | 5 - .../aggregates/aggregate_hash_table/common.rs | 4 - .../aggregate_hash_table/partial_table.rs | 8 -- .../src/aggregates/grouped_hash_stream.rs | 8 +- .../src/aggregates/hash_stream.rs | 4 +- .../physical-plan/src/aggregates/mod.rs | 97 +++++++++++++++++++ .../spark/src/function/aggregate/avg.rs | 11 --- .../library-user-guide/upgrading/55.0.0.md | 33 +++++++ 28 files changed, 136 insertions(+), 146 deletions(-) diff --git a/datafusion-examples/examples/udf/advanced_udaf.rs b/datafusion-examples/examples/udf/advanced_udaf.rs index 096753d2b5d7b..bca4c7edab2c5 100644 --- 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::() + self.prods.capacity() * size_of::() diff --git a/datafusion/core/tests/user_defined/user_defined_aggregates.rs b/datafusion/core/tests/user_defined/user_defined_aggregates.rs index 1d4b22230147f..323925bcfaf82 100644 --- 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::() } diff --git a/datafusion/expr-common/src/groups_accumulator.rs b/datafusion/expr-common/src/groups_accumulator.rs index b021674cbec2c..6679bff754218 100644 --- 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> { - 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>; /// 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 index 272afdb6abfb1..4d1b0b4be0a2b 100644 --- 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, ) -> FFI_Result>, - 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> 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> 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 index 986d4ec0d71ae..10aa21c3acad2 100644 --- 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.seen.capacity() * (size_of::<(usize, T::Native)>() + size_of::()) diff --git a/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator.rs b/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator.rs index b412b4ffe09f2..b5610419166df 100644 --- 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 index afb1dec24a484..77bb7598e2747 100644 --- 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 index 474899d8f3c6a..c5d74978664c9 100644 --- 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::() + self.null_state.size() } diff --git a/datafusion/functions-aggregate/src/approx_distinct.rs b/datafusion/functions-aggregate/src/approx_distinct.rs index 74bc9ad6cbbdc..df132bb7e7b9b 100644 --- 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::() + self.allocated_bytes diff --git a/datafusion/functions-aggregate/src/array_agg.rs b/datafusion/functions-aggregate/src/array_agg.rs index 1dd111f9182c9..d84365ecfe831 100644 --- 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 index 278a861de2024..ff2cbe4afb012 100644 --- a/datafusion/functions-aggregate/src/average.rs +++ b/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::() diff --git a/datafusion/functions-aggregate/src/correlation.rs b/datafusion/functions-aggregate/src/correlation.rs index 2e90cac6d9298..b9bc57dfa989c 100644 --- 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 index f0ce8c82a1bb2..b4bf73f0ca0ac 100644 --- a/datafusion/functions-aggregate/src/count.rs +++ b/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::() } diff --git a/datafusion/functions-aggregate/src/first_last.rs b/datafusion/functions-aggregate/src/first_last.rs index cecb277cb844a..e4c0a4a4fb38f 100644 --- a/datafusion/functions-aggregate/src/first_last.rs +++ b/datafusion/functions-aggregate/src/first_last.rs @@ -712,11 +712,6 @@ impl GroupsAccumulator for FirstLastGroupsAccumulator() + 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 index 9a6ef3e7e5fc5..7a399f73ec8e2 100644 --- a/datafusion/functions-aggregate/src/median.rs +++ b/datafusion/functions-aggregate/src/median.rs @@ -535,11 +535,6 @@ impl GroupsAccumulator for MedianGroupsAccumulator 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 index 7a3c605d82e4d..efeaea314c4f5 100644 --- 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 index 15df0f1d44eff..d1bac4e2f90db 100644 --- 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 index e8e6fd127e65d..cfab1303028ad 100644 --- 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 index a31517b93e003..15511bf4a565f 100644 --- a/datafusion/functions-aggregate/src/stddev.rs +++ b/datafusion/functions-aggregate/src/stddev.rs @@ -352,11 +352,6 @@ impl GroupsAccumulator for StddevGroupsAccumulator { ) -> Result> { 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 index 6b0665f479d78..3fe2b0a186ae3 100644 --- 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::>() diff --git a/datafusion/functions-aggregate/src/variance.rs b/datafusion/functions-aggregate/src/variance.rs index 0278ce2c233e4..df652731ff4f4 100644 --- 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::() + self.means.capacity() * size_of::() diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs index e6e690c4d1e08..88d74751a939d 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs +++ b/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, 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 index ffac42feaa3b3..4bcacb49afb04 100644 --- 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 { 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 index 0d00e5c4d0d86..99c101199459f 100644 --- 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 index 62b92965030ae..e7f0f075b33a5 100644 --- 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 index e7832629b7a59..a8b704461a3c4 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -3421,6 +3421,87 @@ mod tests { Ok(()) } + #[tokio::test] + async fn partial_hash_skip_aggregation_uses_required_convert_to_state() -> Result<()> + { + let schema = Arc::new(Schema::new(vec![ + Field::new("key", DataType::Int32, false), + Field::new("value", DataType::Int32, false), + ])); + let input_batches = vec![ + RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(Int32Array::from(vec![1, 2, 3])), + Arc::new(Int32Array::from(vec![10, 20, 30])), + ], + )?, + RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(Int32Array::from(vec![4, 5, 6])), + Arc::new(Int32Array::from(vec![40, 50, 60])), + ], + )?, + ]; + let input = + TestMemoryExec::try_new_exec(&[input_batches], Arc::clone(&schema), None)?; + let group_by = + PhysicalGroupBy::new_single(vec![(col("key", &schema)?, "key".to_string())]); + let udaf = Arc::new(AggregateUDF::from(NoFirstEmitUdaf::new())); + let aggregates: Vec> = vec![Arc::new( + AggregateExprBuilder::new(udaf, vec![col("value", &schema)?]) + .schema(Arc::clone(&schema)) + .alias("no_first_emit(value)") + .build()?, + )]; + let aggregate = Arc::new(AggregateExec::try_new( + AggregateMode::Partial, + group_by, + aggregates, + vec![None], + input, + Arc::clone(&schema), + )?); + let task_ctx = Arc::new( + TaskContext::default().with_session_config( + SessionConfig::new() + .set_bool("datafusion.execution.enable_migration_aggregate", true) + .set( + "datafusion.execution.skip_partial_aggregation_probe_rows_threshold", + &ScalarValue::Int64(Some(2)), + ) + .set( + "datafusion.execution.skip_partial_aggregation_probe_ratio_threshold", + &ScalarValue::Float64(Some(0.1)), + ), + ), + ); + + let output = collect(aggregate.execute(0, Arc::clone(&task_ctx))?).await?; + + assert_snapshot!(batches_to_sort_string(&output), @r" + +-----+-----------------------------+ + | key | no_first_emit(value)[count] | + +-----+-----------------------------+ + | 1 | 1 | + | 2 | 1 | + | 3 | 1 | + | 4 | 1 | + | 5 | 1 | + | 6 | 1 | + +-----+-----------------------------+ + "); + let metrics = aggregate.metrics().unwrap(); + let skipped_rows = metrics + .sum_by_name("skipped_aggregation_rows") + .map(|m| m.as_usize()) + .unwrap_or(0); + assert_eq!(skipped_rows, 3); + + Ok(()) + } + #[tokio::test] async fn limited_distinct_aggregate_uses_migrated_hash_streams() -> Result<()> { let schema = @@ -6901,6 +6982,22 @@ mod tests { Ok(vec![self.emit_counts(emit_to)?]) } + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, + ) -> Result> { + 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::>(), + 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 index 6ca3c59309e70..46e63013dbafb 100644 --- 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::() + self.sums.capacity() * size_of::() } @@ -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 index 181e0e0b7f266..fdd4836abb357 100644 --- a/docs/source/library-user-guide/upgrading/55.0.0.md +++ b/docs/source/library-user-guide/upgrading/55.0.0.md @@ -238,6 +238,39 @@ it was `None`), that code can simply be deleted. See [issue #22775](https://github.com/apache/datafusion/issues/22775) for details. +### `GroupsAccumulator::convert_to_state` is now required + +`datafusion_expr_common::groups_accumulator::GroupsAccumulator::convert_to_state` +no longer provides a default implementation. Custom `GroupsAccumulator` +implementations must now implement it: + +```diff + fn convert_to_state( + &self, + values: &[ArrayRef], + opt_filter: Option<&BooleanArray>, +- ) -> Result> { +- not_impl_err!("Input batch conversion to state not implemented") +- } ++ ) -> Result>; +``` + +The `GroupsAccumulator::supports_convert_to_state` capability method has also +been removed. Implementations that previously provided it should delete it: + +```diff +- fn supports_convert_to_state(&self) -> bool { +- true +- } +``` + +This also removes the `supports_convert_to_state` field from +`datafusion_ffi::udaf::groups_accumulator::FFI_GroupsAccumulator`, changing its +ABI layout. FFI providers and consumers must be rebuilt against DataFusion 55 +and must not exchange this struct with older major versions. + +See [issue #23081](https://github.com/apache/datafusion/issues/23081) for details. + ### `is_dynamic_physical_expr` is deprecated `datafusion_physical_expr_common::physical_expr::is_dynamic_physical_expr` is