feat(proto): Add protobuf serialization for HashExpr

datafusion/physical-plan/src/joins/hash_join/exec.rs

@@ -86,9 +86,11 @@ use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays;
 use futures::TryStreamExt;
 use parking_lot::Mutex;
 
+use super::partitioned_hash_eval::SeededRandomState;
+
 /// Hard-coded seed to ensure hash values from the hash join differ from `RepartitionExec`, avoiding collisions.
-pub(crate) const HASH_JOIN_SEED: RandomState =
-    RandomState::with_seeds('J' as u64, 'O' as u64, 'I' as u64, 'N' as u64);
+pub(crate) const HASH_JOIN_SEED: SeededRandomState =
+    SeededRandomState::with_seeds('J' as u64, 'O' as u64, 'I' as u64, 'N' as u64);
 
 /// HashTable and input data for the left (build side) of a join
 pub(super) struct JoinLeftData {
@@ -334,8 +336,8 @@ pub struct HashJoinExec {
     /// Each output stream waits on the `OnceAsync` to signal the completion of
     /// the hash table creation.
     left_fut: Arc<OnceAsync<JoinLeftData>>,
-    /// Shared the `RandomState` for the hashing algorithm
-    random_state: RandomState,
+    /// Shared the `SeededRandomState` for the hashing algorithm (seeds preserved for serialization)
+    random_state: SeededRandomState,
     /// Partitioning mode to use
     pub mode: PartitionMode,
     /// Execution metrics
@@ -930,7 +932,7 @@ impl ExecutionPlan for HashJoinExec {
                     MemoryConsumer::new("HashJoinInput").register(context.memory_pool());
 
                 Ok(collect_left_input(
-                    self.random_state.clone(),
+                    self.random_state.random_state().clone(),
                     left_stream,
                     on_left.clone(),
                     join_metrics.clone(),
@@ -958,7 +960,7 @@ impl ExecutionPlan for HashJoinExec {
                         .register(context.memory_pool());
 
                 OnceFut::new(collect_left_input(
-                    self.random_state.clone(),
+                    self.random_state.random_state().clone(),
                     left_stream,
                     on_left.clone(),
                     join_metrics.clone(),
@@ -1041,7 +1043,7 @@ impl ExecutionPlan for HashJoinExec {
             self.filter.clone(),
             self.join_type,
             right_stream,
-            self.random_state.clone(),
+            self.random_state.random_state().clone(),
             join_metrics,
             column_indices_after_projection,
             self.null_equality,

datafusion/physical-plan/src/joins/hash_join/mod.rs

@@ -18,7 +18,7 @@
 //! [`HashJoinExec`] Partitioned Hash Join Operator
 
 pub use exec::HashJoinExec;
-pub use partitioned_hash_eval::HashTableLookupExpr;
+pub use partitioned_hash_eval::{HashExpr, HashTableLookupExpr, SeededRandomState};
 
 mod exec;
 mod inlist_builder;

datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs

@@ -34,6 +34,36 @@ use datafusion_physical_expr_common::physical_expr::{
 
 use crate::{hash_utils::create_hashes, joins::utils::JoinHashMapType};
 
+/// RandomState wrapper that preserves the seeds used to create it.
+///
+/// This is needed because ahash's `RandomState` doesn't expose its seeds after creation,
+/// but we need them for serialization (e.g., protobuf serde).
+#[derive(Clone, Debug)]
+pub struct SeededRandomState {
+    random_state: RandomState,
+    seeds: (u64, u64, u64, u64),
+}
+
+impl SeededRandomState {
+    /// Create a new SeededRandomState with the given seeds.
+    pub const fn with_seeds(k0: u64, k1: u64, k2: u64, k3: u64) -> Self {
+        Self {
+            random_state: RandomState::with_seeds(k0, k1, k2, k3),
+            seeds: (k0, k1, k2, k3),
+        }
+    }
+
+    /// Get the inner RandomState.
+    pub fn random_state(&self) -> &RandomState {
+        &self.random_state
+    }
+
+    /// Get the seeds used to create this RandomState.
+    pub fn seeds(&self) -> (u64, u64, u64, u64) {
+        self.seeds
+    }
+}
+
 /// Physical expression that computes hash values for a set of columns
 ///
 /// This expression computes the hash of join key columns using a specific RandomState.
@@ -45,8 +75,8 @@ use crate::{hash_utils::create_hashes, joins::utils::JoinHashMapType};
 pub struct HashExpr {
     /// Columns to hash
     on_columns: Vec<PhysicalExprRef>,
-    /// Random state for hashing
-    random_state: RandomState,
+    /// Random state for hashing (with seeds preserved for serialization)
+    random_state: SeededRandomState,
     /// Description for display
     description: String,
 }
@@ -56,11 +86,11 @@ impl HashExpr {
     ///
     /// # Arguments
     /// * `on_columns` - Columns to hash
-    /// * `random_state` - RandomState for hashing
+    /// * `random_state` - SeededRandomState for hashing
     /// * `description` - Description for debugging (e.g., "hash_repartition", "hash_join")
-    pub(super) fn new(
+    pub fn new(
         on_columns: Vec<PhysicalExprRef>,
-        random_state: RandomState,
+        random_state: SeededRandomState,
         description: String,
     ) -> Self {
         Self {
@@ -69,6 +99,21 @@ impl HashExpr {
             description,
         }
     }
+
+    /// Get the columns being hashed.
+    pub fn on_columns(&self) -> &[PhysicalExprRef] {
+        &self.on_columns
+    }
+
+    /// Get the seeds used for hashing.
+    pub fn seeds(&self) -> (u64, u64, u64, u64) {
+        self.random_state.seeds()
+    }
+
+    /// Get the description.
+    pub fn description(&self) -> &str {
+        &self.description
+    }
 }
 
 impl std::fmt::Debug for HashExpr {
@@ -87,12 +132,15 @@ impl Hash for HashExpr {
     fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
         self.on_columns.dyn_hash(state);
         self.description.hash(state);
+        self.seeds().hash(state);
     }
 }
 
 impl PartialEq for HashExpr {
     fn eq(&self, other: &Self) -> bool {
-        self.on_columns == other.on_columns && self.description == other.description
+        self.on_columns == other.on_columns
+            && self.description == other.description
+            && self.seeds() == other.seeds()
     }
 }
 
@@ -147,7 +195,11 @@ impl PhysicalExpr for HashExpr {
 
         // Compute hashes
         let mut hashes_buffer = vec![0; num_rows];
-        create_hashes(&keys_values, &self.random_state, &mut hashes_buffer)?;
+        create_hashes(
+            &keys_values,
+            self.random_state.random_state(),
+            &mut hashes_buffer,
+        )?;
 
         Ok(ColumnarValue::Array(Arc::new(UInt64Array::from(
             hashes_buffer,
@@ -211,8 +263,7 @@ impl Hash for HashTableLookupExpr {
 
 impl PartialEq for HashTableLookupExpr {
     fn eq(&self, other: &Self) -> bool {
-        Arc::ptr_eq(&self.hash_expr, &other.hash_expr)
-            && self.description == other.description
+        self.hash_expr.dyn_eq(&other.hash_expr) && self.description == other.description
     }
 }
 

datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs

@@ -26,10 +26,10 @@ use crate::ExecutionPlanProperties;
 use crate::joins::PartitionMode;
 use crate::joins::hash_join::exec::HASH_JOIN_SEED;
 use crate::joins::hash_join::inlist_builder::build_struct_fields;
-use crate::joins::hash_join::partitioned_hash_eval::{HashExpr, HashTableLookupExpr};
+use crate::joins::hash_join::partitioned_hash_eval::{
+    HashExpr, HashTableLookupExpr, SeededRandomState,
+};
 use crate::joins::utils::JoinHashMapType;
-
-use ahash::RandomState;
 use arrow::array::ArrayRef;
 use arrow::datatypes::{DataType, Field, Schema};
 use datafusion_common::config::ConfigOptions;
@@ -88,7 +88,7 @@ impl PartitionBounds {
 fn create_membership_predicate(
     on_right: &[PhysicalExprRef],
     pushdown: PushdownStrategy,
-    random_state: &RandomState,
+    random_state: &SeededRandomState,
     schema: &Schema,
 ) -> Result<Option<Arc<dyn PhysicalExpr>>> {
     match pushdown {
@@ -230,7 +230,7 @@ pub(crate) struct SharedBuildAccumulator {
     on_right: Vec<PhysicalExprRef>,
     /// Random state for partitioning (RepartitionExec's hash function with 0,0,0,0 seeds)
     /// Used for PartitionedHashLookupPhysicalExpr
-    repartition_random_state: RandomState,
+    repartition_random_state: SeededRandomState,
     /// Schema of the probe (right) side for evaluating filter expressions
     probe_schema: Arc<Schema>,
 }
@@ -308,7 +308,7 @@ impl SharedBuildAccumulator {
         right_child: &dyn ExecutionPlan,
         dynamic_filter: Arc<DynamicFilterPhysicalExpr>,
         on_right: Vec<PhysicalExprRef>,
-        repartition_random_state: RandomState,
+        repartition_random_state: SeededRandomState,
     ) -> Self {
         // Troubleshooting: If partition counts are incorrect, verify this logic matches
         // the actual execution pattern in collect_build_side()

datafusion/physical-plan/src/joins/mod.rs

@@ -20,7 +20,7 @@
 use arrow::array::BooleanBufferBuilder;
 pub use cross_join::CrossJoinExec;
 use datafusion_physical_expr::PhysicalExprRef;
-pub use hash_join::{HashJoinExec, HashTableLookupExpr};
+pub use hash_join::{HashExpr, HashJoinExec, HashTableLookupExpr, SeededRandomState};
 pub use nested_loop_join::NestedLoopJoinExec;
 use parking_lot::Mutex;
 // Note: SortMergeJoin is not used in plans yet

datafusion/physical-plan/src/repartition/mod.rs

@@ -61,6 +61,7 @@ use crate::filter_pushdown::{
     ChildPushdownResult, FilterDescription, FilterPushdownPhase,
     FilterPushdownPropagation,
 };
+use crate::joins::SeededRandomState;
 use crate::sort_pushdown::SortOrderPushdownResult;
 use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
 use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays;
@@ -429,8 +430,8 @@ enum BatchPartitionerState {
 
 /// Fixed RandomState used for hash repartitioning to ensure consistent behavior across
 /// executions and runs.
-pub const REPARTITION_RANDOM_STATE: ahash::RandomState =
-    ahash::RandomState::with_seeds(0, 0, 0, 0);
+pub const REPARTITION_RANDOM_STATE: SeededRandomState =
+    SeededRandomState::with_seeds(0, 0, 0, 0);
 
 impl BatchPartitioner {
     /// Create a new [`BatchPartitioner`] with the provided [`Partitioning`]
@@ -514,7 +515,11 @@ impl BatchPartitioner {
                     hash_buffer.clear();
                     hash_buffer.resize(batch.num_rows(), 0);
 
-                    create_hashes(&arrays, &REPARTITION_RANDOM_STATE, hash_buffer)?;
+                    create_hashes(
+                        &arrays,
+                        REPARTITION_RANDOM_STATE.random_state(),
+                        hash_buffer,
+                    )?;
 
                     let mut indices: Vec<_> = (0..*partitions)
                         .map(|_| Vec::with_capacity(batch.num_rows()))

datafusion/proto/proto/datafusion.proto

@@ -872,6 +872,8 @@ message PhysicalExprNode {
     PhysicalExtensionExprNode extension = 19;
 
     UnknownColumn unknown_column = 20;
+
+    PhysicalHashExprNode hash_expr = 21;
   }
 }
 
@@ -990,6 +992,15 @@ message PhysicalExtensionExprNode {
   repeated PhysicalExprNode inputs = 2;
 }
 
+message PhysicalHashExprNode {
+  repeated PhysicalExprNode on_columns = 1;
+  uint64 seed0 = 2;
+  uint64 seed1 = 3;
+  uint64 seed2 = 4;
+  uint64 seed3 = 5;
+  string description = 6;
+}
+
 message FilterExecNode {
   PhysicalPlanNode input = 1;
   PhysicalExprNode expr = 2;