Derive WindowUDFImpl equality, hash from Eq, Hash traits (#17081)

* Move `DynEq`, `DynHash` to `expr-common`

Use `DynEq` and `DynHash` traits from physical expressions crate to a
common crate for physical and logical expressions. This allows them to
be used by logical expressions.

* 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.

* upgrade guide

* fix typo

* Fix PartialEq for WindowUDF impl

Wrong Any was compared

* Seal DynEq, DynHash

* Link to epic in upgrade guide

Co-authored-by: Andrew Lamb <[email protected]>

---------

Co-authored-by: Andrew Lamb <[email protected]>

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

@@ -0,0 +1,64 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::any::Any;
+use std::hash::{Hash, Hasher};
+
+/// A dyn-compatible version of [`Eq`] trait.
+/// The implementation constraints for this trait are the same as for [`Eq`]:
+/// the implementation must be reflexive, symmetric, and transitive.
+/// Additionally, if two values can be compared with [`DynEq`] and [`PartialEq`] then
+/// 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.
+#[allow(private_bounds)]
+pub trait DynEq: private::EqSealed {
+    fn dyn_eq(&self, other: &dyn Any) -> bool;
+}
+
+impl<T: Eq + Any> private::EqSealed for T {}
+impl<T: Eq + Any> DynEq for T {
+    fn dyn_eq(&self, other: &dyn Any) -> bool {
+        other.downcast_ref::<Self>() == Some(self)
+    }
+}
+
+/// 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.
+#[allow(private_bounds)]
+pub trait DynHash: private::HashSealed {
+    fn dyn_hash(&self, _state: &mut dyn Hasher);
+}
+
+impl<T: Hash + Any> private::HashSealed for T {}
+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)
+    }
+}
+
+mod private {
+    pub(super) trait EqSealed {}
+    pub(super) trait HashSealed {}
+}

datafusion/expr-common/src/lib.rs

@@ -35,6 +35,7 @@
 pub mod accumulator;
 pub mod casts;
 pub mod columnar_value;
+pub mod dyn_eq;
 pub mod groups_accumulator;
 pub mod interval_arithmetic;
 pub mod operator;

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.as_any())
     }
 }
 
 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)
     }
 }
 
@@ -229,6 +229,10 @@ where
 /// This trait exposes the full API for implementing user defined window functions and
 /// can be used to implement any function.
 ///
+/// While the trait depends on [`DynEq`] and [`DynHash`] traits, these should not be
+/// implemented directly. Instead, implement [`Eq`] and [`Hash`] and leverage the
+/// blanket implementations of [`DynEq`] and [`DynHash`].
+///
 /// See [`advanced_udwf.rs`] for a full example with complete implementation and
 /// [`WindowUDF`] for other available options.
 ///
@@ -246,7 +250,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 +309,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 +362,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 +430,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 +502,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 +565,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 +604,7 @@ mod test {
         }
     }
 
-    #[derive(Debug, Clone)]
+    #[derive(Debug, Clone, PartialEq, Eq, Hash)]
     struct BWindowUDF {
         signature: Signature,
     }
@@ -676,6 +643,15 @@ mod test {
         }
     }
 
+    #[test]
+    fn test_partial_eq() {
+        let a1 = WindowUDF::from(AWindowUDF::new());
+        let a2 = WindowUDF::from(AWindowUDF::new());
+        let eq = a1 == a2;
+        assert!(eq);
+        assert_eq!(a1, a2);
+    }
+
     #[test]
     fn test_partial_ord() {
         let a1 = WindowUDF::from(AWindowUDF::new());