Implement alternative fix

Author: nuno-faria

datafusion/functions-aggregate/src/array_agg.rs

@@ -17,11 +17,10 @@
 
 //! `ARRAY_AGG` aggregate implementation: [`ArrayAgg`]
 
-use std::any::{type_name, Any};
+use std::any::Any;
 use std::cmp::Ordering;
 use std::collections::{HashSet, VecDeque};
-use std::hash::{DefaultHasher, Hash, Hasher};
-use std::mem::{size_of, size_of_val, take};
+use std::mem::{size_of, size_of_val};
 use std::sync::Arc;
 
 use arrow::array::{
@@ -33,7 +32,7 @@ use arrow::datatypes::{DataType, Field, FieldRef, Fields};
 
 use datafusion_common::cast::as_list_array;
 use datafusion_common::scalar::copy_array_data;
-use datafusion_common::utils::{compare_rows, get_row_at_idx, SingleRowListArrayBuilder};
+use datafusion_common::utils::{get_row_at_idx, SingleRowListArrayBuilder};
 use datafusion_common::{exec_err, internal_err, Result, ScalarValue};
 use datafusion_expr::function::{AccumulatorArgs, StateFieldsArgs};
 use datafusion_expr::utils::format_state_name;
@@ -76,18 +75,16 @@ This aggregation function can only mix DISTINCT and ORDER BY if the ordering exp
 "#,
     standard_argument(name = "expression",)
 )]
-#[derive(Debug, PartialEq, Eq, Hash)]
+#[derive(Debug)]
 /// ARRAY_AGG aggregate expression
 pub struct ArrayAgg {
     signature: Signature,
-    is_input_pre_ordered: bool,
 }
 
 impl Default for ArrayAgg {
     fn default() -> Self {
         Self {
             signature: Signature::any(1, Volatility::Immutable),
-            is_input_pre_ordered: false,
         }
     }
 }
@@ -148,16 +145,6 @@ impl AggregateUDFImpl for ArrayAgg {
         Ok(fields)
     }
 
-    fn with_beneficial_ordering(
-        self: Arc<Self>,
-        beneficial_ordering: bool,
-    ) -> Result<Option<Arc<dyn AggregateUDFImpl>>> {
-        Ok(Some(Arc::new(Self {
-            signature: self.signature.clone(),
-            is_input_pre_ordered: beneficial_ordering,
-        })))
-    }
-
     fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
         let data_type = acc_args.exprs[0].data_type(acc_args.schema)?;
         let ignore_nulls =
@@ -210,7 +197,6 @@ impl AggregateUDFImpl for ArrayAgg {
             &data_type,
             &ordering_dtypes,
             ordering,
-            self.is_input_pre_ordered,
             acc_args.is_reversed,
             ignore_nulls,
         )
@@ -224,23 +210,6 @@ impl AggregateUDFImpl for ArrayAgg {
     fn documentation(&self) -> Option<&Documentation> {
         self.doc()
     }
-
-    fn equals(&self, other: &dyn AggregateUDFImpl) -> bool {
-        let Some(other) = <dyn Any + 'static>::downcast_ref::<Self>(other.as_any())
-        else {
-            return false;
-        };
-        fn assert_self_impls_eq<T: Eq>() {}
-        assert_self_impls_eq::<Self>();
-        PartialEq::eq(self, other)
-    }
-
-    fn hash_value(&self) -> u64 {
-        let hasher = &mut DefaultHasher::new();
-        type_name::<Self>().hash(hasher);
-        Hash::hash(self, hasher);
-        Hasher::finish(hasher)
-    }
 }
 
 #[derive(Debug)]
@@ -550,8 +519,6 @@ pub(crate) struct OrderSensitiveArrayAggAccumulator {
     datatypes: Vec<DataType>,
     /// Stores the ordering requirement of the `Accumulator`.
     ordering_req: LexOrdering,
-    /// Whether the input is known to be pre-ordered
-    is_input_pre_ordered: bool,
     /// Whether the aggregation is running in reverse.
     reverse: bool,
     /// Whether the aggregation should ignore null values.
@@ -565,7 +532,6 @@ impl OrderSensitiveArrayAggAccumulator {
         datatype: &DataType,
         ordering_dtypes: &[DataType],
         ordering_req: LexOrdering,
-        is_input_pre_ordered: bool,
         reverse: bool,
         ignore_nulls: bool,
     ) -> Result<Self> {
@@ -576,34 +542,11 @@ impl OrderSensitiveArrayAggAccumulator {
             ordering_values: vec![],
             datatypes,
             ordering_req,
-            is_input_pre_ordered,
             reverse,
             ignore_nulls,
         })
     }
 
-    fn sort(&mut self) {
-        let sort_options = self
-            .ordering_req
-            .iter()
-            .map(|sort_expr| sort_expr.options)
-            .collect::<Vec<_>>();
-        let mut values = take(&mut self.values)
-            .into_iter()
-            .zip(take(&mut self.ordering_values))
-            .collect::<Vec<_>>();
-        let mut delayed_cmp_err = Ok(());
-        values.sort_by(|(_, left_ordering), (_, right_ordering)| {
-            compare_rows(left_ordering, right_ordering, &sort_options).unwrap_or_else(
-                |err| {
-                    delayed_cmp_err = Err(err);
-                    Ordering::Equal
-                },
-            )
-        });
-        (self.values, self.ordering_values) = values.into_iter().unzip();
-    }
-
     fn evaluate_orderings(&self) -> Result<ScalarValue> {
         let fields = ordering_fields(&self.ordering_req, &self.datatypes[1..]);
 
@@ -687,9 +630,6 @@ impl Accumulator for OrderSensitiveArrayAggAccumulator {
         let mut partition_ordering_values = vec![];
 
         // Existing values should be merged also.
-        if !self.is_input_pre_ordered {
-            self.sort();
-        }
         partition_values.push(self.values.clone().into());
         partition_ordering_values.push(self.ordering_values.clone().into());
 
@@ -740,21 +680,13 @@ impl Accumulator for OrderSensitiveArrayAggAccumulator {
     }
 
     fn state(&mut self) -> Result<Vec<ScalarValue>> {
-        if !self.is_input_pre_ordered {
-            self.sort();
-        }
-
         let mut result = vec![self.evaluate()?];
         result.push(self.evaluate_orderings()?);
 
         Ok(result)
     }
 
     fn evaluate(&mut self) -> Result<ScalarValue> {
-        if !self.is_input_pre_ordered {
-            self.sort();
-        }
-
         if self.values.is_empty() {
             return Ok(ScalarValue::new_null_list(
                 self.datatypes[0].clone(),

datafusion/functions-aggregate/src/string_agg.rs

@@ -178,6 +178,10 @@ impl AggregateUDFImpl for StringAgg {
         )))
     }
 
+    fn reverse_expr(&self) -> datafusion_expr::ReversedUDAF {
+        datafusion_expr::ReversedUDAF::Reversed(string_agg_udaf())
+    }
+
     fn documentation(&self) -> Option<&Documentation> {
         self.doc()
     }