fix: mark avg, bit_and/or/xor, stddev and variance as order-insensitive

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<Vec<FieldRef>> {
         if args.is_distinct {
             // Decimal accumulator actually uses a different precision during accumulation,

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<Vec<FieldRef>> {
         if args.input_fields[0].data_type().is_null() {
             Ok(vec![

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<Vec<FieldRef>> {
         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<Vec<FieldRef>> {
         Ok(vec![
             Field::new(

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<Vec<FieldRef>> {
         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<Vec<FieldRef>> {
         match args.is_distinct {
             false => {

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.