Derive WindowUDFImpl equality, hash from Eq, Hash traits

Previously, the `WindowUDFImpl` 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_udwf.rs

@@ -43,7 +43,7 @@ use datafusion::prelude::*;
 /// a function `partition_evaluator` that returns the `MyPartitionEvaluator` instance.
 ///
 /// To do so, we must implement the `WindowUDFImpl` trait.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct SmoothItUdf {
     signature: Signature,
 }
@@ -149,7 +149,7 @@ impl PartitionEvaluator for MyPartitionEvaluator {
 }
 
 /// This UDWF will show how to use the WindowUDFImpl::simplify() API
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 struct SimplifySmoothItUdf {
     signature: Signature,
 }

datafusion/core/tests/user_defined/user_defined_window_functions.rs

@@ -30,8 +30,7 @@ use datafusion::prelude::SessionContext;
 use datafusion_common::exec_datafusion_err;
 use datafusion_expr::ptr_eq::PtrEq;
 use datafusion_expr::{
-    udf_equals_hash, PartitionEvaluator, Signature, TypeSignature, Volatility, WindowUDF,
-    WindowUDFImpl,
+    PartitionEvaluator, Signature, TypeSignature, Volatility, WindowUDF, WindowUDFImpl,
 };
 use datafusion_functions_window_common::partition::PartitionEvaluatorArgs;
 use datafusion_functions_window_common::{
@@ -42,7 +41,7 @@ use datafusion_physical_expr::{
     PhysicalExpr,
 };
 use std::collections::HashMap;
-use std::hash::{DefaultHasher, Hash, Hasher};
+use std::hash::{Hash, Hasher};
 use std::{
     any::Any,
     ops::Range,
@@ -571,8 +570,6 @@ impl OddCounter {
             fn field(&self, field_args: WindowUDFFieldArgs) -> Result<FieldRef> {
                 Ok(Field::new(field_args.name(), DataType::Int64, true).into())
             }
-
-            udf_equals_hash!(WindowUDFImpl);
         }
 
         ctx.register_udwf(WindowUDF::from(SimpleWindowUDF::new(test_state)))
@@ -648,7 +645,7 @@ fn odd_count_arr(arr: &Int64Array, num_rows: usize) -> ArrayRef {
     Arc::new(array)
 }
 
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Eq, Hash)]
 struct VariadicWindowUDF {
     signature: Signature,
 }
@@ -770,6 +767,31 @@ struct MetadataBasedWindowUdf {
     metadata: HashMap<String, String>,
 }
 
+impl PartialEq for MetadataBasedWindowUdf {
+    fn eq(&self, other: &Self) -> bool {
+        let Self {
+            name,
+            signature,
+            metadata,
+        } = self;
+        name == &other.name
+            && signature == &other.signature
+            && metadata == &other.metadata
+    }
+}
+impl Eq for MetadataBasedWindowUdf {}
+impl Hash for MetadataBasedWindowUdf {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        let Self {
+            name,
+            signature,
+            metadata: _, // unhashable
+        } = self;
+        name.hash(state);
+        signature.hash(state);
+    }
+}
+
 impl MetadataBasedWindowUdf {
     fn new(metadata: HashMap<String, String>) -> Self {
         // The name we return must be unique. Otherwise we will not call distinct
@@ -820,33 +842,6 @@ impl WindowUDFImpl for MetadataBasedWindowUdf {
             .with_metadata(self.metadata.clone())
             .into())
     }
-
-    fn equals(&self, other: &dyn WindowUDFImpl) -> bool {
-        let Some(other) = other.as_any().downcast_ref::<Self>() else {
-            return false;
-        };
-        let Self {
-            name,
-            signature,
-            metadata,
-        } = self;
-        name == &other.name
-            && signature == &other.signature
-            && metadata == &other.metadata
-    }
-
-    fn hash_value(&self) -> u64 {
-        let Self {
-            name,
-            signature,
-            metadata: _, // unhashable
-        } = self;
-        let mut hasher = DefaultHasher::new();
-        std::any::type_name::<Self>().hash(&mut hasher);
-        name.hash(&mut hasher);
-        signature.hash(&mut hasher);
-        hasher.finish()
-    }
 }
 
 #[derive(Debug)]

datafusion/expr-common/src/dyn_eq.rs

@@ -25,25 +25,43 @@ use std::hash::{Hash, Hasher};
 /// they must be [`DynEq`]-equal if and only if they are [`PartialEq`]-equal.
 /// It is therefore strongly discouraged to implement this trait for types
 /// that implement `PartialEq<Other>` or `Eq<Other>` for any type `Other` other than `Self`.
+///
+/// Note: This trait should not be implemented directly. Implement `Eq` and `Any` and use
+/// the blanket implementation.
 pub trait DynEq {
     fn dyn_eq(&self, other: &dyn Any) -> bool;
+
+    fn i_did_not_implement_the_trait_directly_but_using_the_blanked_impl_instead()
+    where
+        Self: Sized;
 }
 
 impl<T: Eq + Any> DynEq for T {
     fn dyn_eq(&self, other: &dyn Any) -> bool {
         other.downcast_ref::<Self>() == Some(self)
     }
+
+    fn i_did_not_implement_the_trait_directly_but_using_the_blanked_impl_instead() {}
 }
 
 /// A dyn-compatible version of [`Hash`] trait.
 /// If two values are equal according to [`DynEq`], they must produce the same hash value.
+///
+/// Note: This trait should not be implemented directly. Implement `Hash` and `Any` and use
+/// the blanket implementation.
 pub trait DynHash {
     fn dyn_hash(&self, _state: &mut dyn Hasher);
+
+    fn i_did_not_implement_the_trait_directly_but_using_the_blanked_impl_instead()
+    where
+        Self: Sized;
 }
 
 impl<T: Hash + Any> DynHash for T {
     fn dyn_hash(&self, mut state: &mut dyn Hasher) {
         self.type_id().hash(&mut state);
         self.hash(&mut state)
     }
+
+    fn i_did_not_implement_the_trait_directly_but_using_the_blanked_impl_instead() {}
 }

datafusion/expr/src/expr_fn.rs

@@ -695,8 +695,6 @@ impl WindowUDFImpl for SimpleWindowUDF {
             true,
         )))
     }
-
-    udf_equals_hash!(WindowUDFImpl);
 }
 
 pub fn interval_year_month_lit(value: &str) -> Expr {

datafusion/expr/src/udf_eq.rs

@@ -17,7 +17,7 @@
 
 use crate::{AggregateUDFImpl, ScalarUDFImpl, WindowUDFImpl};
 use std::fmt::Debug;
-use std::hash::{Hash, Hasher};
+use std::hash::{DefaultHasher, Hash, Hasher};
 use std::ops::Deref;
 use std::sync::Arc;
 
@@ -97,7 +97,18 @@ macro_rules! impl_for_udf_eq {
 
 impl_for_udf_eq!(dyn AggregateUDFImpl + '_);
 impl_for_udf_eq!(dyn ScalarUDFImpl + '_);
-impl_for_udf_eq!(dyn WindowUDFImpl + '_);
+
+impl UdfPointer for Arc<dyn WindowUDFImpl + '_> {
+    fn equals(&self, other: &(dyn WindowUDFImpl + '_)) -> 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()
+    }
+}
 
 #[cfg(test)]
 mod tests {

datafusion/expr/src/udwf.rs

@@ -19,7 +19,7 @@
 
 use arrow::compute::SortOptions;
 use std::cmp::Ordering;
-use std::hash::{DefaultHasher, Hash, Hasher};
+use std::hash::{Hash, Hasher};
 use std::{
     any::Any,
     fmt::{self, Debug, Display, Formatter},
@@ -31,11 +31,11 @@ use arrow::datatypes::{DataType, FieldRef};
 use crate::expr::WindowFunction;
 use crate::udf_eq::UdfEq;
 use crate::{
-    function::WindowFunctionSimplification, udf_equals_hash, Expr, PartitionEvaluator,
-    Signature,
+    function::WindowFunctionSimplification, Expr, PartitionEvaluator, Signature,
 };
 use datafusion_common::{not_impl_err, Result};
 use datafusion_doc::Documentation;
+use datafusion_expr_common::dyn_eq::{DynEq, DynHash};
 use datafusion_functions_window_common::expr::ExpressionArgs;
 use datafusion_functions_window_common::field::WindowUDFFieldArgs;
 use datafusion_functions_window_common::partition::PartitionEvaluatorArgs;
@@ -82,15 +82,15 @@ impl Display for WindowUDF {
 
 impl PartialEq for WindowUDF {
     fn eq(&self, other: &Self) -> bool {
-        self.inner.equals(other.inner.as_ref())
+        self.inner.dyn_eq(&other.inner)
     }
 }
 
 impl Eq for WindowUDF {}
 
 impl Hash for WindowUDF {
     fn hash<H: Hasher>(&self, state: &mut H) {
-        self.inner.hash_value().hash(state)
+        self.inner.dyn_hash(state)
     }
 }
 
@@ -246,7 +246,7 @@ where
 /// # use datafusion_functions_window_common::partition::PartitionEvaluatorArgs;
 /// # use datafusion_expr::window_doc_sections::DOC_SECTION_ANALYTICAL;
 ///
-/// #[derive(Debug, Clone)]
+/// #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 /// struct SmoothIt {
 ///   signature: Signature,
 /// }
@@ -305,7 +305,7 @@ where
 ///     .build()
 ///     .unwrap();
 /// ```
-pub trait WindowUDFImpl: Debug + Send + Sync {
+pub trait WindowUDFImpl: Debug + DynEq + DynHash + Send + Sync {
     /// Returns this object as an [`Any`] trait object
     fn as_any(&self) -> &dyn Any;
 
@@ -358,41 +358,6 @@ pub trait WindowUDFImpl: Debug + Send + Sync {
         None
     }
 
-    /// Return true if this window UDF is equal to the other.
-    ///
-    /// Allows customizing the equality of window 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 WindowUDFImpl) -> 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 window UDF.
-    ///
-    /// Allows customizing the hash code of window 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()
-    }
-
     /// The [`FieldRef`] of the final result of evaluating this window function.
     ///
     /// Call `field_args.name()` to get the fully qualified name for defining
@@ -461,7 +426,7 @@ pub enum ReversedUDWF {
 
 impl PartialEq for dyn WindowUDFImpl {
     fn eq(&self, other: &Self) -> bool {
-        self.equals(other)
+        self.dyn_eq(other.as_any())
     }
 }
 
@@ -533,8 +498,6 @@ impl WindowUDFImpl for AliasedWindowUDFImpl {
         self.inner.simplify()
     }
 
-    udf_equals_hash!(WindowUDFImpl);
-
     fn field(&self, field_args: WindowUDFFieldArgs) -> Result<FieldRef> {
         self.inner.field(field_args)
     }
@@ -598,7 +561,7 @@ mod test {
     use std::any::Any;
     use std::cmp::Ordering;
 
-    #[derive(Debug, Clone)]
+    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
     struct AWindowUDF {
         signature: Signature,
     }
@@ -637,7 +600,7 @@ mod test {
         }
     }
 
-    #[derive(Debug, Clone)]
+    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
     struct BWindowUDF {
         signature: Signature,
     }

datafusion/ffi/src/udwf/mod.rs

@@ -25,7 +25,6 @@ use arrow::{
     datatypes::{DataType, SchemaRef},
 };
 use arrow_schema::{Field, FieldRef};
-use datafusion::logical_expr::udf_equals_hash;
 use datafusion::{
     error::DataFusionError,
     logical_expr::{
@@ -349,8 +348,6 @@ impl WindowUDFImpl for ForeignWindowUDF {
         let options: Option<&FFI_SortOptions> = self.udf.sort_options.as_ref().into();
         options.map(|s| s.into())
     }
-
-    udf_equals_hash!(WindowUDFImpl);
 }
 
 #[repr(C)]

datafusion/functions-window/src/cume_dist.rs

@@ -63,7 +63,7 @@ FROM employees;
 ```
 "#
 )]
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Eq, Hash)]
 pub struct CumeDist {
     signature: Signature,
 }

datafusion/functions-window/src/lead_lag.rs

@@ -25,8 +25,8 @@ use datafusion_common::arrow::datatypes::Field;
 use datafusion_common::{arrow_datafusion_err, DataFusionError, Result, ScalarValue};
 use datafusion_expr::window_doc_sections::DOC_SECTION_ANALYTICAL;
 use datafusion_expr::{
-    udf_equals_hash, Documentation, Literal, PartitionEvaluator, ReversedUDWF, Signature,
-    TypeSignature, Volatility, WindowUDFImpl,
+    Documentation, Literal, PartitionEvaluator, ReversedUDWF, Signature, TypeSignature,
+    Volatility, WindowUDFImpl,
 };
 use datafusion_functions_window_common::expr::ExpressionArgs;
 use datafusion_functions_window_common::field::WindowUDFFieldArgs;
@@ -299,8 +299,6 @@ impl WindowUDFImpl for WindowShift {
             WindowShiftKind::Lead => Some(get_lead_doc()),
         }
     }
-
-    udf_equals_hash!(WindowUDFImpl);
 }
 
 /// When `lead`/`lag` is evaluated on a `NULL` expression we attempt to