Derive AggregateUDFImpl equality, hash from Eq, Hash traits (#17130)

Follows similar change for `WindowUDFImpl`, i.e. the
8494a39 commit.

Previously, the `AggregateUDFImpl` trait contained `equals` and
`hash_value` methods with contracts following the `Eq` and `Hash`
traits.  However, the existence of default implementations of these
methods made it error-prone, with many functions (scalar, aggregate,
window) missing to customize the equals even though they ought to.
There is no fix to this that's not an API breaking change, so a breaking
change is warranted.

Removing the default implementations would be enough of a solution, but
at the cost of a lot of boilerplate needed in implementations.

Instead, this removes the methods from the trait, and reuses `DynEq`,
`DynHash` traits used previously only for physical expressions. This
allows for functions to provide their implementations using no more than
`#[derive(PartialEq, Eq, Hash)]` in a typical case.

Author: findepi

datafusion-examples/examples/advanced_udaf.rs

@@ -41,7 +41,7 @@ use datafusion::prelude::*;
 /// a function `accumulator` that returns the `Accumulator` instance.
 ///
 /// To do so, we must implement the `AggregateUDFImpl` trait.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct GeoMeanUdaf {
     signature: Signature,
 }
@@ -368,7 +368,7 @@ impl GroupsAccumulator for GeometricMeanGroupsAccumulator {
 
 /// This example shows how to use the AggregateUDFImpl::simplify API to simplify/replace user
 /// defined aggregate function with a different expression which is defined in the `simplify` method.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct SimplifiedGeoMeanUdaf {
     signature: Signature,
 }

datafusion/core/tests/user_defined/user_defined_aggregates.rs

@@ -55,7 +55,7 @@ use datafusion_common::{assert_contains, exec_datafusion_err};
 use datafusion_common::{cast::as_primitive_array, exec_err};
 use datafusion_expr::expr::WindowFunction;
 use datafusion_expr::{
-    col, create_udaf, function::AccumulatorArgs, udf_equals_hash, AggregateUDFImpl, Expr,
+    col, create_udaf, function::AccumulatorArgs, AggregateUDFImpl, Expr,
     GroupsAccumulator, LogicalPlanBuilder, SimpleAggregateUDF, WindowFunctionDefinition,
 };
 use datafusion_functions_aggregate::average::AvgAccumulator;
@@ -816,8 +816,6 @@ impl AggregateUDFImpl for TestGroupsAccumulator {
     ) -> Result<Box<dyn GroupsAccumulator>> {
         Ok(Box::new(self.clone()))
     }
-
-    udf_equals_hash!(AggregateUDFImpl);
 }
 
 impl Accumulator for TestGroupsAccumulator {
@@ -970,8 +968,6 @@ impl AggregateUDFImpl for MetadataBasedAggregateUdf {
             curr_sum: 0,
         }))
     }
-
-    udf_equals_hash!(AggregateUDFImpl);
 }
 
 #[derive(Debug)]

datafusion/expr/src/expr_fn.rs

@@ -600,8 +600,6 @@ impl AggregateUDFImpl for SimpleAggregateUDF {
     fn state_fields(&self, _args: StateFieldsArgs) -> Result<Vec<FieldRef>> {
         Ok(self.state_fields.clone())
     }
-
-    udf_equals_hash!(AggregateUDFImpl);
 }
 
 /// Creates a new UDWF with a specific signature, state type and return type.

datafusion/expr/src/test/function_stub.rs

@@ -92,7 +92,7 @@ pub fn avg(expr: Expr) -> Expr {
 }
 
 /// Stub `sum` used for optimizer testing
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Eq, Hash)]
 pub struct Sum {
     signature: Signature,
 }
@@ -200,6 +200,7 @@ impl AggregateUDFImpl for Sum {
 }
 
 /// Testing stub implementation of COUNT aggregate
+#[derive(PartialEq, Eq, Hash)]
 pub struct Count {
     signature: Signature,
     aliases: Vec<String>,
@@ -288,6 +289,7 @@ pub fn min(expr: Expr) -> Expr {
 }
 
 /// Testing stub implementation of Min aggregate
+#[derive(PartialEq, Eq, Hash)]
 pub struct Min {
     signature: Signature,
 }
@@ -369,6 +371,7 @@ pub fn max(expr: Expr) -> Expr {
 }
 
 /// Testing stub implementation of MAX aggregate
+#[derive(PartialEq, Eq, Hash)]
 pub struct Max {
     signature: Signature,
 }
@@ -437,7 +440,7 @@ impl AggregateUDFImpl for Max {
 }
 
 /// Testing stub implementation of avg aggregate
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Eq, Hash)]
 pub struct Avg {
     signature: Signature,
     aliases: Vec<String>,

datafusion/expr/src/udaf.rs

@@ -20,13 +20,14 @@
 use std::any::Any;
 use std::cmp::Ordering;
 use std::fmt::{self, Debug, Formatter, Write};
-use std::hash::{DefaultHasher, Hash, Hasher};
+use std::hash::{Hash, Hasher};
 use std::sync::Arc;
 use std::vec;
 
 use arrow::datatypes::{DataType, Field, FieldRef};
 
 use datafusion_common::{exec_err, not_impl_err, Result, ScalarValue, Statistics};
+use datafusion_expr_common::dyn_eq::{DynEq, DynHash};
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 
 use crate::expr::{
@@ -41,7 +42,7 @@ use crate::groups_accumulator::GroupsAccumulator;
 use crate::udf_eq::UdfEq;
 use crate::utils::format_state_name;
 use crate::utils::AggregateOrderSensitivity;
-use crate::{expr_vec_fmt, udf_equals_hash, Accumulator, Expr};
+use crate::{expr_vec_fmt, Accumulator, Expr};
 use crate::{Documentation, Signature};
 
 /// Logical representation of a user-defined [aggregate function] (UDAF).
@@ -82,15 +83,15 @@ pub struct AggregateUDF {
 
 impl PartialEq for AggregateUDF {
     fn eq(&self, other: &Self) -> bool {
-        self.inner.equals(other.inner.as_ref())
+        self.inner.dyn_eq(other.inner.as_any())
     }
 }
 
 impl Eq for AggregateUDF {}
 
 impl Hash for AggregateUDF {
     fn hash<H: Hasher>(&self, state: &mut H) {
-        self.inner.hash_value().hash(state)
+        self.inner.dyn_hash(state)
     }
 }
 
@@ -373,7 +374,7 @@ where
 /// # use arrow::datatypes::Schema;
 /// # use arrow::datatypes::Field;
 ///
-/// #[derive(Debug, Clone)]
+/// #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 /// struct GeoMeanUdf {
 ///   signature: Signature,
 /// }
@@ -426,7 +427,7 @@ where
 /// // Call the function `geo_mean(col)`
 /// let expr = geometric_mean.call(vec![col("a")]);
 /// ```
-pub trait AggregateUDFImpl: Debug + Send + Sync {
+pub trait AggregateUDFImpl: Debug + DynEq + DynHash + Send + Sync {
     /// Returns this object as an [`Any`] trait object
     fn as_any(&self) -> &dyn Any;
 
@@ -914,41 +915,6 @@ pub trait AggregateUDFImpl: Debug + Send + Sync {
         not_impl_err!("Function {} does not implement coerce_types", self.name())
     }
 
-    /// Return true if this aggregate UDF is equal to the other.
-    ///
-    /// Allows customizing the equality of aggregate UDFs.
-    /// *Must* be implemented explicitly if the UDF type has internal state.
-    /// Must be consistent with [`Self::hash_value`] and follow the same rules as [`Eq`]:
-    ///
-    /// - reflexive: `a.equals(a)`;
-    /// - symmetric: `a.equals(b)` implies `b.equals(a)`;
-    /// - transitive: `a.equals(b)` and `b.equals(c)` implies `a.equals(c)`.
-    ///
-    /// By default, compares type, [`Self::name`], [`Self::aliases`] and [`Self::signature`].
-    fn equals(&self, other: &dyn AggregateUDFImpl) -> bool {
-        self.as_any().type_id() == other.as_any().type_id()
-            && self.name() == other.name()
-            && self.aliases() == other.aliases()
-            && self.signature() == other.signature()
-    }
-
-    /// Returns a hash value for this aggregate UDF.
-    ///
-    /// Allows customizing the hash code of aggregate UDFs.
-    /// *Must* be implemented explicitly whenever [`Self::equals`] is implemented.
-    ///
-    /// Similarly to [`Hash`] and [`Eq`], if [`Self::equals`] returns true for two UDFs,
-    /// their `hash_value`s must be the same.
-    ///
-    /// By default, it only hashes the type. The other fields are not hashed, as usually the
-    /// name, signature, and aliases are implied by the UDF type. Recall that UDFs with state
-    /// (and thus possibly changing fields) must override [`Self::equals`] and [`Self::hash_value`].
-    fn hash_value(&self) -> u64 {
-        let hasher = &mut DefaultHasher::new();
-        self.as_any().type_id().hash(hasher);
-        hasher.finish()
-    }
-
     /// If this function is max, return true
     /// If the function is min, return false
     /// Otherwise return None (the default)
@@ -1008,10 +974,11 @@ pub trait AggregateUDFImpl: Debug + Send + Sync {
 
 impl PartialEq for dyn AggregateUDFImpl {
     fn eq(&self, other: &Self) -> bool {
-        self.equals(other)
+        self.dyn_eq(other.as_any())
     }
 }
 
+// TODO (https://github.com/apache/datafusion/issues/17064) PartialOrd is not consistent with PartialEq for `dyn AggregateUDFImpl` and it should be
 // Manual implementation of `PartialOrd`
 // There might be some wackiness with it, but this is based on the impl of eq for AggregateUDFImpl
 // https://users.rust-lang.org/t/how-to-compare-two-trait-objects-for-equality/88063/5
@@ -1194,8 +1161,6 @@ impl AggregateUDFImpl for AliasedAggregateUDFImpl {
         self.inner.set_monotonicity(data_type)
     }
 
-    udf_equals_hash!(AggregateUDFImpl);
-
     fn documentation(&self) -> Option<&Documentation> {
         self.inner.documentation()
     }
@@ -1266,7 +1231,7 @@ mod test {
     use std::any::Any;
     use std::cmp::Ordering;
 
-    #[derive(Debug, Clone)]
+    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
     struct AMeanUdf {
         signature: Signature,
     }
@@ -1307,7 +1272,7 @@ mod test {
         }
     }
 
-    #[derive(Debug, Clone)]
+    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
     struct BMeanUdf {
         signature: Signature,
     }
@@ -1347,6 +1312,15 @@ mod test {
         }
     }
 
+    #[test]
+    fn test_partial_eq() {
+        let a1 = AggregateUDF::from(AMeanUdf::new());
+        let a2 = AggregateUDF::from(AMeanUdf::new());
+        let eq = a1 == a2;
+        assert!(eq);
+        assert_eq!(a1, a2);
+    }
+
     #[test]
     fn test_partial_ord() {
         // Test validates that partial ord is defined for AggregateUDF using the name and signature,

datafusion/expr/src/udf.rs

@@ -66,6 +66,7 @@ impl PartialEq for ScalarUDF {
     }
 }
 
+// TODO (https://github.com/apache/datafusion/issues/17064) PartialOrd is not consistent with PartialEq for `ScalarUDF` and it should be
 // Manual implementation based on `ScalarUDFImpl::equals`
 impl PartialOrd for ScalarUDF {
     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {

datafusion/expr/src/udf_eq.rs

@@ -95,9 +95,20 @@ macro_rules! impl_for_udf_eq {
     };
 }
 
-impl_for_udf_eq!(dyn AggregateUDFImpl + '_);
 impl_for_udf_eq!(dyn ScalarUDFImpl + '_);
 
+impl UdfPointer for Arc<dyn AggregateUDFImpl + '_> {
+    fn equals(&self, other: &(dyn AggregateUDFImpl + '_)) -> bool {
+        self.as_ref().dyn_eq(other.as_any())
+    }
+
+    fn hash_value(&self) -> u64 {
+        let hasher = &mut DefaultHasher::new();
+        self.as_ref().dyn_hash(hasher);
+        hasher.finish()
+    }
+}
+
 impl UdfPointer for Arc<dyn WindowUDFImpl + '_> {
     fn equals(&self, other: &(dyn WindowUDFImpl + '_)) -> bool {
         self.as_ref().dyn_eq(other.as_any())

datafusion/expr/src/udwf.rs

@@ -434,6 +434,7 @@ impl PartialEq for dyn WindowUDFImpl {
     }
 }
 
+// TODO (https://github.com/apache/datafusion/issues/17064) PartialOrd is not consistent with PartialEq for `dyn WindowUDFImpl` and it should be
 impl PartialOrd for dyn WindowUDFImpl {
     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
         match self.name().partial_cmp(other.name()) {

datafusion/ffi/src/udaf/mod.rs

@@ -49,7 +49,6 @@ use crate::{
     util::{rvec_wrapped_to_vec_datatype, vec_datatype_to_rvec_wrapped},
     volatility::FFI_Volatility,
 };
-use datafusion::logical_expr::udf_equals_hash;
 use prost::{DecodeError, Message};
 
 mod accumulator;
@@ -567,8 +566,6 @@ impl AggregateUDFImpl for ForeignAggregateUDF {
             Ok(rvec_wrapped_to_vec_datatype(&result_types)?)
         }
     }
-
-    udf_equals_hash!(AggregateUDFImpl);
 }
 
 #[repr(C)]

datafusion/functions-aggregate/src/approx_distinct.rs

@@ -293,6 +293,7 @@ impl Default for ApproxDistinct {
 ```"#,
     standard_argument(name = "expression",)
 )]
+#[derive(PartialEq, Eq, Hash)]
 pub struct ApproxDistinct {
     signature: Signature,
 }