diff --git a/datafusion/functions-aggregate/src/average.rs b/datafusion/functions-aggregate/src/average.rs index f59ac816a930..b0966c4af334 100644 --- a/datafusion/functions-aggregate/src/average.rs +++ b/datafusion/functions-aggregate/src/average.rs @@ -36,7 +36,7 @@ use datafusion_common::{ Result, ScalarValue, exec_datafusion_err, exec_err, internal_err, not_impl_err, }; use datafusion_expr::function::{AccumulatorArgs, StateFieldsArgs}; -use datafusion_expr::utils::format_state_name; +use datafusion_expr::utils::{AggregateOrderSensitivity, format_state_name}; use datafusion_expr::{ Accumulator, AggregateUDFImpl, Coercion, Documentation, EmitTo, Expr, GroupSelection, GroupsAccumulator, ReversedUDAF, Signature, TypeSignature, TypeSignatureClass, @@ -350,6 +350,10 @@ impl AggregateUDFImpl for Avg { } } + fn order_sensitivity(&self) -> AggregateOrderSensitivity { + AggregateOrderSensitivity::Insensitive + } + fn state_fields(&self, args: StateFieldsArgs) -> Result> { if args.is_distinct { // Decimal accumulator actually uses a different precision during accumulation, diff --git a/datafusion/functions-aggregate/src/bit_and_or_xor.rs b/datafusion/functions-aggregate/src/bit_and_or_xor.rs index 02d77c84ba0b..9d5f7bfaac23 100644 --- a/datafusion/functions-aggregate/src/bit_and_or_xor.rs +++ b/datafusion/functions-aggregate/src/bit_and_or_xor.rs @@ -33,7 +33,7 @@ use datafusion_common::cast::as_list_array; use datafusion_common::{Result, ScalarValue, not_impl_err}; use datafusion_expr::DistinctHandling; use datafusion_expr::function::{AccumulatorArgs, StateFieldsArgs}; -use datafusion_expr::utils::format_state_name; +use datafusion_expr::utils::{AggregateOrderSensitivity, format_state_name}; use datafusion_expr::{ Accumulator, AggregateUDFImpl, Coercion, Documentation, GroupsAccumulator, ReversedUDAF, Signature, TypeSignatureClass, Volatility, @@ -256,6 +256,10 @@ impl AggregateUDFImpl for BitwiseOperation { downcast_bitwise_accumulator!(acc_args, self.operation, acc_args.is_distinct) } + fn order_sensitivity(&self) -> AggregateOrderSensitivity { + AggregateOrderSensitivity::Insensitive + } + fn state_fields(&self, args: StateFieldsArgs) -> Result> { if args.input_fields[0].data_type().is_null() { Ok(vec![ diff --git a/datafusion/functions-aggregate/src/stddev.rs b/datafusion/functions-aggregate/src/stddev.rs index 14a28fa80997..25a5a703e9c5 100644 --- a/datafusion/functions-aggregate/src/stddev.rs +++ b/datafusion/functions-aggregate/src/stddev.rs @@ -28,7 +28,7 @@ use arrow::{array::ArrayRef, datatypes::DataType, datatypes::Field}; use datafusion_common::ScalarValue; use datafusion_common::{Result, internal_err, not_impl_err}; use datafusion_expr::function::{AccumulatorArgs, StateFieldsArgs}; -use datafusion_expr::utils::format_state_name; +use datafusion_expr::utils::{AggregateOrderSensitivity, format_state_name}; use datafusion_expr::{ Accumulator, AggregateUDFImpl, Documentation, GroupSelection, GroupsAccumulator, Signature, Volatility, @@ -96,6 +96,10 @@ impl AggregateUDFImpl for Stddev { Ok(DataType::Float64) } + fn order_sensitivity(&self) -> AggregateOrderSensitivity { + AggregateOrderSensitivity::Insensitive + } + fn state_fields(&self, args: StateFieldsArgs) -> Result> { Ok(vec![ Field::new( @@ -200,6 +204,10 @@ impl AggregateUDFImpl for StddevPop { &self.signature } + fn order_sensitivity(&self) -> AggregateOrderSensitivity { + AggregateOrderSensitivity::Insensitive + } + fn state_fields(&self, args: StateFieldsArgs) -> Result> { Ok(vec![ Field::new( diff --git a/datafusion/functions-aggregate/src/variance.rs b/datafusion/functions-aggregate/src/variance.rs index 154a677cb28f..a9f746b16ed4 100644 --- a/datafusion/functions-aggregate/src/variance.rs +++ b/datafusion/functions-aggregate/src/variance.rs @@ -30,7 +30,7 @@ use datafusion_expr::{ Accumulator, AggregateUDFImpl, Documentation, GroupSelection, GroupsAccumulator, Signature, Volatility, function::{AccumulatorArgs, StateFieldsArgs}, - utils::format_state_name, + utils::{AggregateOrderSensitivity, format_state_name}, }; use datafusion_functions_aggregate_common::utils::GenericDistinctBuffer; use datafusion_functions_aggregate_common::{ @@ -96,6 +96,10 @@ impl AggregateUDFImpl for VarianceSample { Ok(DataType::Float64) } + fn order_sensitivity(&self) -> AggregateOrderSensitivity { + AggregateOrderSensitivity::Insensitive + } + fn state_fields(&self, args: StateFieldsArgs) -> Result> { let name = args.name; match args.is_distinct { @@ -195,6 +199,10 @@ impl AggregateUDFImpl for VariancePopulation { Ok(DataType::Float64) } + fn order_sensitivity(&self) -> AggregateOrderSensitivity { + AggregateOrderSensitivity::Insensitive + } + fn state_fields(&self, args: StateFieldsArgs) -> Result> { match args.is_distinct { false => { diff --git a/datafusion/sqllogictest/test_files/group_by.slt b/datafusion/sqllogictest/test_files/group_by.slt index 942a6f3cc898..f23915966626 100644 --- a/datafusion/sqllogictest/test_files/group_by.slt +++ b/datafusion/sqllogictest/test_files/group_by.slt @@ -2927,6 +2927,31 @@ GRC 110 [30.0, 80.0] FRA 250 [50.0, 200.0] TUR 175 [75.0, 100.0] +# ORDER BY is ignored by order-insensitive aggregators. These aggregators used +# to panic because their ORDER BY expressions were passed to the accumulator as +# extra input columns (issue #25401). +statement ok +CREATE TABLE insensitive_order_by (g INT, k INT, v INT) AS VALUES + (1, 2, 6), (1, 1, 3), (2, 4, 12), (2, 3, 10); + +query IRIIIRRRR +SELECT g, + avg(v ORDER BY k), + bit_and(v ORDER BY k), + bit_or(v ORDER BY k), + bit_xor(v ORDER BY k), + var_samp(v ORDER BY k), + var_pop(v ORDER BY k), + stddev_pop(v ORDER BY k), + round(stddev(v ORDER BY k), 4) +FROM insensitive_order_by GROUP BY g ORDER BY g; +---- +1 4.5 2 7 5 4.5 2.25 1.5 2.1213 +2 11 8 14 6 2 1 1 1.4142 + +statement ok +DROP TABLE insensitive_order_by; + # test_reverse_aggregate_expr5 # If all of the ordering sensitive aggregation functions are reversible # we should be able to reverse requirements, if this helps to remove a SortExec.