Implement/fix Eq and Hash for Expr and LogicalPlan

datafusion/common/src/dfschema.rs

@@ -20,6 +20,7 @@
 
 use std::collections::{HashMap, HashSet};
 use std::convert::TryFrom;
+use std::hash::Hash;
 use std::sync::Arc;
 
 use crate::error::{DataFusionError, Result, SchemaError};
@@ -496,6 +497,15 @@ impl From<DFSchema> for SchemaRef {
     }
 }
 
+// Hashing refers to a subset of fields considered in PartialEq.
+#[allow(clippy::derive_hash_xor_eq)]
+impl Hash for DFSchema {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.fields.hash(state);
+        self.metadata.len().hash(state); // HashMap is not hashable
+    }
+}
+
 /// Convenience trait to convert Schema like things to DFSchema and DFSchemaRef with fewer keystrokes
 pub trait ToDFSchema
 where
@@ -587,7 +597,7 @@ impl ExprSchema for DFSchema {
 }
 
 /// DFField wraps an Arrow field and adds an optional qualifier
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct DFField {
     /// Optional qualifier (usually a table or relation name)
     qualifier: Option<String>,

datafusion/common/src/parsers.rs

@@ -26,7 +26,7 @@ const SECONDS_PER_HOUR: f64 = 3_600_f64;
 const NANOS_PER_SECOND: f64 = 1_000_000_000_f64;
 
 /// Readable file compression type
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum CompressionTypeVariant {
     /// Gzip-ed file
     GZIP,

datafusion/common/src/table_reference.rs

@@ -69,7 +69,7 @@ pub enum TableReference<'a> {
 
 /// Represents a path to a table that may require further resolution
 /// that owns the underlying names
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub enum OwnedTableReference {
     /// An unqualified table reference, e.g. "table"
     Bare {

datafusion/core/src/physical_plan/planner.rs

@@ -2373,6 +2373,7 @@ Internal error: Optimizer rule 'type_coercion' failed due to unexpected error: E
         }
     }
     /// An example extension node that doesn't do anything
+    #[derive(PartialEq, Eq, Hash)]
     struct NoOpExtensionNode {
         schema: DFSchemaRef,
     }
@@ -2425,6 +2426,19 @@ Internal error: Optimizer rule 'type_coercion' failed due to unexpected error: E
         ) -> Arc<dyn UserDefinedLogicalNode> {
             unimplemented!("NoOp");
         }
+
+        fn dyn_eq(&self, other: &dyn UserDefinedLogicalNode) -> bool {
+            match other.as_any().downcast_ref::<Self>() {
+                Some(o) => self == o,
+                None => false,
+            }
+        }
+
+        fn dyn_hash(&self, state: &mut dyn std::hash::Hasher) {
+            use std::hash::Hash;
+            let mut s = state;
+            self.hash(&mut s);
+        }
     }
 
     #[derive(Debug)]

datafusion/core/tests/user_defined_plan.rs

@@ -324,6 +324,7 @@ impl OptimizerRule for TopKOptimizerRule {
     }
 }
 
+#[derive(PartialEq, Eq, Hash)]
 struct TopKPlanNode {
     k: usize,
     input: LogicalPlan,
@@ -376,6 +377,19 @@ impl UserDefinedLogicalNode for TopKPlanNode {
             expr: exprs[0].clone(),
         })
     }
+
+    fn dyn_eq(&self, other: &dyn UserDefinedLogicalNode) -> bool {
+        match other.as_any().downcast_ref::<Self>() {
+            Some(o) => self == o,
+            None => false,
+        }
+    }
+
+    fn dyn_hash(&self, state: &mut dyn std::hash::Hasher) {
+        use std::hash::Hash;
+        let mut s = state;
+        self.hash(&mut s);
+    }
 }
 
 /// Physical planner for TopK nodes

datafusion/expr/src/logical_plan/extension.rs

@@ -18,14 +18,15 @@
 //! This module defines the interface for logical nodes
 use crate::{Expr, LogicalPlan};
 use datafusion_common::DFSchemaRef;
-use std::{any::Any, collections::HashSet, fmt, sync::Arc};
+use std::hash::{Hash, Hasher};
+use std::{any::Any, cmp::Eq, collections::HashSet, fmt, sync::Arc};
 
 /// This defines the interface for `LogicalPlan` nodes that can be
 /// used to extend DataFusion with custom relational operators.
 ///
 /// See the example in
 /// [user_defined_plan.rs](../../tests/user_defined_plan.rs) for an
-/// example of how to use this extension API
+/// example of how to use this extension API.
 pub trait UserDefinedLogicalNode: fmt::Debug + Send + Sync {
     /// Return a reference to self as Any, to support dynamic downcasting
     fn as_any(&self) -> &dyn Any;
@@ -77,4 +78,26 @@ pub trait UserDefinedLogicalNode: fmt::Debug + Send + Sync {
         exprs: &[Expr],
         inputs: &[LogicalPlan],
     ) -> Arc<dyn UserDefinedLogicalNode>;
+
+    /// Hashing respecting requirements from [std::hash::Hash].
+    fn dyn_hash(&self, state: &mut dyn Hasher);
+
+    /// Comparison respecting requirements from [std::cmp::Eq].
+    ///
+    /// When `other` has an another type than `self`, then the values are *not* equal.
+    fn dyn_eq(&self, other: &dyn UserDefinedLogicalNode) -> bool;
+}
+
+impl Hash for dyn UserDefinedLogicalNode {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.dyn_hash(state);
+    }
 }
+
+impl std::cmp::PartialEq for dyn UserDefinedLogicalNode {
+    fn eq(&self, other: &Self) -> bool {
+        self.dyn_eq(other)
+    }
+}
+
+impl Eq for dyn UserDefinedLogicalNode {}