This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 3f81613730 Remove `GroupsAccumulator::supports_convert_to_state` and
require `convert_to_state` (#23489)
3f81613730 is described below
commit 3f816137306112f75b412e2806abd24fed0b916f
Author: linfeng <[email protected]>
AuthorDate: Fri Jul 24 20:22:36 2026 +0800
Remove `GroupsAccumulator::supports_convert_to_state` and require
`convert_to_state` (#23489)
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
- Closes #23081.
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
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?
<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes 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?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
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?
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api
change` label.
-->
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.
---
datafusion-examples/examples/udf/advanced_udaf.rs | 5 ----
.../tests/user_defined/user_defined_aggregates.rs | 5 ----
datafusion/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 ---
.../src/aggregate/groups_accumulator/bool_op.rs | 4 ---
.../src/aggregate/groups_accumulator/prim_op.rs | 5 ----
.../functions-aggregate/src/approx_distinct.rs | 5 ----
datafusion/functions-aggregate/src/array_agg.rs | 5 ----
datafusion/functions-aggregate/src/average.rs | 5 ----
datafusion/functions-aggregate/src/correlation.rs | 5 ----
datafusion/functions-aggregate/src/count.rs | 5 ----
datafusion/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 ----
.../functions-aggregate/src/percentile_cont.rs | 5 ----
datafusion/functions-aggregate/src/stddev.rs | 5 ----
datafusion/functions-aggregate/src/string_agg.rs | 5 ----
datafusion/functions-aggregate/src/variance.rs | 5 ----
.../src/aggregates/aggregate_hash_table/common.rs | 4 ---
.../aggregate_hash_table/partial_table.rs | 8 -----
.../src/aggregates/grouped_hash_stream.rs | 8 +----
.../physical-plan/src/aggregates/hash_stream.rs | 4 +--
datafusion/physical-plan/src/aggregates/mod.rs | 16 ++++++++++
datafusion/spark/src/function/aggregate/avg.rs | 11 -------
docs/source/library-user-guide/upgrading/55.0.0.md | 34 ++++++++++++++++++++++
28 files changed, 56 insertions(+), 146 deletions(-)
diff --git a/datafusion-examples/examples/udf/advanced_udaf.rs
b/datafusion-examples/examples/udf/advanced_udaf.rs
index 096753d2b5..bca4c7edab 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::<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
index 1d4b222301..323925bcfa 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::<u64>()
}
diff --git a/datafusion/expr-common/src/groups_accumulator.rs
b/datafusion/expr-common/src/groups_accumulator.rs
index 13b2f853c9..5c01418e04 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<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
index 272afdb6ab..4d1b0b4be0 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<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
index 986d4ec0d7..10aa21c3ac 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>()
+ 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
index b412b4ffe0..b561041916 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 afb1dec24a..77bb7598e2 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 474899d8f3..c5d7497866 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::<T::Native>() +
self.null_state.size()
}
diff --git a/datafusion/functions-aggregate/src/approx_distinct.rs
b/datafusion/functions-aggregate/src/approx_distinct.rs
index f36c658e7f..1746edd823 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::<GroupHll>()
+ self.allocated_bytes
diff --git a/datafusion/functions-aggregate/src/array_agg.rs
b/datafusion/functions-aggregate/src/array_agg.rs
index 1937f17973..b563a6389e 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 f1159f22b2..e5030bf39e 100644
--- 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
index 2e90cac6d9..b9bc57dfa9 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 983828ea90..f0de9d9848 100644
--- 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
index 7c2540aa03..e2da7ec753 100644
--- 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
index 9a6ef3e7e5..7a399f73ec 100644
--- 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
index 7a3c605d82..efeaea314c 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 15df0f1d44..d1bac4e2f9 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 e8e6fd127e..cfab130302 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 a31517b93e..15511bf4a5 100644
--- 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
index 6b0665f479..3fe2b0a186 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::<Option<String>>()
diff --git a/datafusion/functions-aggregate/src/variance.rs
b/datafusion/functions-aggregate/src/variance.rs
index 0278ce2c23..df652731ff 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::<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
index eaf39929ce..42014f336f 100644
--- 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
index ffac42feaa..4bcacb49af 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<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
index 0d00e5c4d0..99c1011994 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 62b9296503..e7f0f075b3 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 7d93cd7398..e3cf1c4568 100644
--- 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
index 6ca3c59309..46e63013db 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::<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
index 57da7b7dac..5f1609af30 100644
--- a/docs/source/library-user-guide/upgrading/55.0.0.md
+++ b/docs/source/library-user-guide/upgrading/55.0.0.md
@@ -276,6 +276,40 @@ 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, and the
+`GroupsAccumulator::supports_convert_to_state` capability method has been
+removed. All `GroupsAccumulator` implementations must now support converting
+input batches directly to intermediate aggregate state.
+
+**Who is affected:**
+
+- Users with custom `GroupsAccumulator` implementations.
+- FFI providers and consumers that use `FFI_GroupsAccumulator`.
+
+**Migration guide:**
+
+Custom `GroupsAccumulator` implementations must now provide their own
+`convert_to_state` implementation.
+
+Delete `supports_convert_to_state` implementations because `convert_to_state`
+is now required:
+
+```diff
+- fn supports_convert_to_state(&self) -> bool {
+- true
+- }
+```
+
+The `supports_convert_to_state` field has also been removed from
+`datafusion_ffi::udaf::groups_accumulator::FFI_GroupsAccumulator`, changing its
+ABI layout. Rebuild both FFI providers and consumers against DataFusion 55, and
+do not exchange this struct with libraries built against 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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
