datafusion_physical_plan/joins/hash_join/
partitioned_hash_eval.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! Hash computation and hash table lookup expressions for dynamic filtering
19
20use std::{fmt::Display, hash::Hash, sync::Arc};
21
22use arrow::{
23    array::{ArrayRef, UInt64Array},
24    datatypes::{DataType, Schema},
25    record_batch::RecordBatch,
26};
27use datafusion_common::Result;
28use datafusion_common::hash_utils::RandomState;
29use datafusion_common::hash_utils::{create_hashes, with_hashes};
30#[cfg(feature = "proto")]
31use datafusion_common::internal_err;
32use datafusion_expr::ColumnarValue;
33use datafusion_physical_expr_common::physical_expr::{
34    DynHash, PhysicalExpr, PhysicalExprRef,
35};
36
37use crate::joins::Map;
38
39/// RandomState wrapper that preserves the seed used to create it.
40///
41/// This is needed because `RandomState` doesn't expose its seed after creation,
42/// but we need them for serialization (e.g., protobuf serde).
43#[derive(Clone, Debug)]
44pub struct SeededRandomState {
45    random_state: RandomState,
46    seed: u64,
47}
48
49impl SeededRandomState {
50    /// Create a new SeededRandomState with the given seed.
51    pub const fn with_seed(k: u64) -> Self {
52        Self {
53            random_state: RandomState::with_seed(k),
54            seed: k,
55        }
56    }
57
58    /// Get the inner RandomState.
59    pub fn random_state(&self) -> &RandomState {
60        &self.random_state
61    }
62
63    /// Get the seed used to create this RandomState.
64    pub fn seed(&self) -> u64 {
65        self.seed
66    }
67}
68
69/// Physical expression that computes hash values for a set of columns
70///
71/// This expression computes the hash of join key columns using a specific RandomState.
72/// It returns a UInt64Array containing the hash values.
73///
74/// This is used for:
75/// - Computing routing hashes (with RepartitionExec's 0,0,0,0 seeds)
76/// - Computing lookup hashes (with HashJoin's 'J','O','I','N' seeds)
77pub struct HashExpr {
78    /// Columns to hash
79    on_columns: Vec<PhysicalExprRef>,
80    /// Random state for hashing (with seeds preserved for serialization)
81    random_state: SeededRandomState,
82    /// Description for display
83    description: String,
84}
85
86impl HashExpr {
87    /// Create a new HashExpr
88    ///
89    /// # Arguments
90    /// * `on_columns` - Columns to hash
91    /// * `random_state` - SeededRandomState for hashing
92    /// * `description` - Description for debugging (e.g., "hash_repartition", "hash_join")
93    pub fn new(
94        on_columns: Vec<PhysicalExprRef>,
95        random_state: SeededRandomState,
96        description: String,
97    ) -> Self {
98        Self {
99            on_columns,
100            random_state,
101            description,
102        }
103    }
104
105    /// Get the columns being hashed.
106    pub fn on_columns(&self) -> &[PhysicalExprRef] {
107        &self.on_columns
108    }
109
110    /// Get the seed used for hashing.
111    pub fn seed(&self) -> u64 {
112        self.random_state.seed()
113    }
114
115    /// Get the description.
116    pub fn description(&self) -> &str {
117        &self.description
118    }
119}
120
121impl std::fmt::Debug for HashExpr {
122    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
123        let cols = self
124            .on_columns
125            .iter()
126            .map(|e| e.to_string())
127            .collect::<Vec<_>>()
128            .join(", ");
129        let seed = self.seed();
130        write!(f, "{}({cols}, [{seed}])", self.description)
131    }
132}
133
134impl Hash for HashExpr {
135    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
136        self.on_columns.dyn_hash(state);
137        self.description.hash(state);
138        self.seed().hash(state);
139    }
140}
141
142impl PartialEq for HashExpr {
143    fn eq(&self, other: &Self) -> bool {
144        self.on_columns == other.on_columns
145            && self.description == other.description
146            && self.seed() == other.seed()
147    }
148}
149
150impl Eq for HashExpr {}
151
152impl Display for HashExpr {
153    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
154        write!(f, "{}", self.description)
155    }
156}
157
158impl PhysicalExpr for HashExpr {
159    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
160        self.on_columns.iter().collect()
161    }
162
163    fn with_new_children(
164        self: Arc<Self>,
165        children: Vec<Arc<dyn PhysicalExpr>>,
166    ) -> Result<Arc<dyn PhysicalExpr>> {
167        Ok(Arc::new(HashExpr::new(
168            children,
169            self.random_state.clone(),
170            self.description.clone(),
171        )))
172    }
173
174    fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
175        Ok(DataType::UInt64)
176    }
177
178    fn nullable(&self, _input_schema: &Schema) -> Result<bool> {
179        Ok(false)
180    }
181
182    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
183        let num_rows = batch.num_rows();
184
185        // Evaluate columns
186        let keys_values = evaluate_columns(&self.on_columns, batch)?;
187
188        // Compute hashes
189        let mut hashes_buffer = vec![0; num_rows];
190        create_hashes(
191            &keys_values,
192            self.random_state.random_state(),
193            &mut hashes_buffer,
194        )?;
195
196        Ok(ColumnarValue::Array(Arc::new(UInt64Array::from(
197            hashes_buffer,
198        ))))
199    }
200
201    fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
202        write!(f, "{}", self.description)
203    }
204
205    #[cfg(feature = "proto")]
206    fn try_to_proto(
207        &self,
208        ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
209    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
210        use datafusion_proto_models::protobuf;
211        let on_columns = ctx.encode_children_expressions(&self.on_columns)?;
212        Ok(Some(protobuf::PhysicalExprNode {
213            expr_id: None,
214            expr_type: Some(protobuf::physical_expr_node::ExprType::HashExpr(
215                protobuf::PhysicalHashExprNode {
216                    on_columns,
217                    seed0: self.seed(),
218                    description: self.description.clone(),
219                },
220            )),
221        }))
222    }
223}
224
225#[cfg(feature = "proto")]
226impl HashExpr {
227    /// Reconstruct a [`HashExpr`] from its protobuf representation.
228    ///
229    /// Takes the whole [`PhysicalExprNode`], the exact inverse of what
230    /// [`PhysicalExpr::try_to_proto`] produces, so every expression's
231    /// `try_from_proto` shares one signature. Child sub-expressions are
232    /// decoded recursively via [`PhysicalExprDecodeCtx::decode`].
233    ///
234    /// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode
235    /// [`PhysicalExpr::try_to_proto`]: datafusion_physical_expr_common::physical_expr::PhysicalExpr::try_to_proto
236    /// [`PhysicalExprDecodeCtx::decode`]: datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx::decode
237    pub fn try_from_proto(
238        node: &datafusion_proto_models::protobuf::PhysicalExprNode,
239        ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
240    ) -> Result<Arc<dyn PhysicalExpr>> {
241        use datafusion_proto_models::protobuf;
242        let hash_expr = match &node.expr_type {
243            Some(protobuf::physical_expr_node::ExprType::HashExpr(h)) => h,
244            _ => return internal_err!("PhysicalExprNode is not a HashExpr"),
245        };
246        let on_columns = ctx.decode_children_expressions(&hash_expr.on_columns)?;
247        Ok(Arc::new(HashExpr::new(
248            on_columns,
249            SeededRandomState::with_seed(hash_expr.seed0),
250            hash_expr.description.clone(),
251        )))
252    }
253}
254
255/// Physical expression that checks join keys in a [`Map`] (hash table or array map).
256///
257/// Returns a [`BooleanArray`](arrow::array::BooleanArray) indicating if join keys (from `on_columns`) exist in the map.
258// TODO: rename to MapLookupExpr
259pub struct HashTableLookupExpr {
260    /// Columns in the ON clause used to compute the join key for lookups
261    on_columns: Vec<PhysicalExprRef>,
262    /// Random state for hashing (with seeds preserved for serialization)
263    random_state: SeededRandomState,
264    /// Map to check against (hash table or array map)
265    map: Arc<Map>,
266    /// Description for display
267    description: String,
268}
269impl HashTableLookupExpr {
270    /// Create a new HashTableLookupExpr
271    ///
272    /// # Arguments
273    /// * `on_columns` - Columns in the ON clause used to compute the join key
274    /// * `random_state` - SeededRandomState for hashing
275    /// * `map` - Map to check membership (hash table or array map)
276    /// * `description` - Description for debugging
277    /// # Note
278    /// This is public for internal testing purposes only and is not
279    /// guaranteed to be stable across versions.
280    pub fn new(
281        on_columns: Vec<PhysicalExprRef>,
282        random_state: SeededRandomState,
283        map: Arc<Map>,
284        description: String,
285    ) -> Self {
286        Self {
287            on_columns,
288            random_state,
289            map,
290            description,
291        }
292    }
293}
294impl std::fmt::Debug for HashTableLookupExpr {
295    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
296        let cols = self
297            .on_columns
298            .iter()
299            .map(|e| e.to_string())
300            .collect::<Vec<_>>()
301            .join(", ");
302        let seed = self.random_state.seed();
303        write!(f, "{}({cols}, [{seed}])", self.description)
304    }
305}
306
307impl Hash for HashTableLookupExpr {
308    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
309        self.on_columns.dyn_hash(state);
310        self.description.hash(state);
311        self.random_state.seed().hash(state);
312        // Note that we compare hash_map by pointer equality.
313        // Actually comparing the contents of the hash maps would be expensive.
314        // The way these hash maps are used in actuality is that HashJoinExec creates
315        // one per partition per query execution, thus it is never possible for two different
316        // hash maps to have the same content in practice.
317        // Theoretically this is a public API and users could create identical hash maps,
318        // but that seems unlikely and not worth paying the cost of deep comparison all the time.
319        Arc::as_ptr(&self.map).hash(state);
320    }
321}
322
323impl PartialEq for HashTableLookupExpr {
324    fn eq(&self, other: &Self) -> bool {
325        // Note that we compare hash_map by pointer equality.
326        // Actually comparing the contents of the hash maps would be expensive.
327        // The way these hash maps are used in actuality is that HashJoinExec creates
328        // one per partition per query execution, thus it is never possible for two different
329        // hash maps to have the same content in practice.
330        // Theoretically this is a public API and users could create identical hash maps,
331        // but that seems unlikely and not worth paying the cost of deep comparison all the time.
332        self.on_columns == other.on_columns
333            && self.description == other.description
334            && self.random_state.seed() == other.random_state.seed()
335            && Arc::ptr_eq(&self.map, &other.map)
336    }
337}
338
339impl Eq for HashTableLookupExpr {}
340
341impl Display for HashTableLookupExpr {
342    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
343        write!(f, "{}", self.description)
344    }
345}
346
347impl PhysicalExpr for HashTableLookupExpr {
348    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
349        self.on_columns.iter().collect()
350    }
351
352    fn with_new_children(
353        self: Arc<Self>,
354        children: Vec<Arc<dyn PhysicalExpr>>,
355    ) -> Result<Arc<dyn PhysicalExpr>> {
356        Ok(Arc::new(HashTableLookupExpr::new(
357            children,
358            self.random_state.clone(),
359            Arc::clone(&self.map),
360            self.description.clone(),
361        )))
362    }
363
364    fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
365        Ok(DataType::Boolean)
366    }
367
368    fn nullable(&self, _input_schema: &Schema) -> Result<bool> {
369        Ok(false)
370    }
371
372    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
373        // Evaluate columns
374        let join_keys = evaluate_columns(&self.on_columns, batch)?;
375
376        match self.map.as_ref() {
377            Map::HashMap(map) => {
378                with_hashes(&join_keys, self.random_state.random_state(), |hashes| {
379                    let array = map.contain_hashes(hashes);
380                    Ok(ColumnarValue::Array(Arc::new(array)))
381                })
382            }
383            Map::ArrayMap(map) => {
384                let array = map.contain_keys(&join_keys)?;
385                Ok(ColumnarValue::Array(Arc::new(array)))
386            }
387        }
388    }
389    #[cfg(feature = "proto")]
390    fn try_to_proto(
391        &self,
392        _ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
393    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
394        use datafusion_proto_models::protobuf;
395        use datafusion_proto_models::protobuf::physical_expr_node::ExprType;
396
397        // HashTableLookupExpr holds a runtime Arc<Map> (the build-side hash
398        // table) that cannot be serialized, so it is replaced with lit(true).
399        //
400        // Dynamic filtering is a performance optimisation only — replacing the
401        // lookup with lit(true) preserves correctness by allowing all rows
402        // through.
403        //
404        // If a plan is serialized before execution, HashTableLookupExpr is not
405        // yet present in the dynamic filter expression.
406        //
407        // If a plan is serialized after execution, any runtime-created
408        // HashTableLookupExpr is replaced during serialization. Re-executing
409        // the plan requires reset_state(), after which HashJoinExec rebuilds
410        // fresh dynamic filters at runtime.
411        let value = datafusion_proto_common::ScalarValue {
412            value: Some(datafusion_proto_common::scalar_value::Value::BoolValue(
413                true,
414            )),
415        };
416        Ok(Some(protobuf::PhysicalExprNode {
417            expr_id: None,
418            expr_type: Some(ExprType::Literal(value)),
419        }))
420    }
421    fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
422        write!(f, "{}", self.description)
423    }
424}
425
426fn evaluate_columns(
427    columns: &[PhysicalExprRef],
428    batch: &RecordBatch,
429) -> Result<Vec<ArrayRef>> {
430    let num_rows = batch.num_rows();
431    columns
432        .iter()
433        .map(|c| c.evaluate(batch)?.into_array(num_rows))
434        .collect()
435}
436
437#[cfg(test)]
438mod tests {
439    use super::*;
440    use crate::joins::join_hash_map::JoinHashMapU32;
441    use datafusion_physical_expr::expressions::Column;
442    use std::collections::hash_map::DefaultHasher;
443    use std::hash::Hasher;
444
445    fn compute_hash<T: Hash>(value: &T) -> u64 {
446        let mut hasher = DefaultHasher::new();
447        value.hash(&mut hasher);
448        hasher.finish()
449    }
450
451    #[test]
452    fn test_hash_expr_eq_same() {
453        let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0));
454        let col_b: PhysicalExprRef = Arc::new(Column::new("b", 1));
455
456        let expr1 = HashExpr::new(
457            vec![Arc::clone(&col_a), Arc::clone(&col_b)],
458            SeededRandomState::with_seed(1),
459            "test_hash".to_string(),
460        );
461
462        let expr2 = HashExpr::new(
463            vec![Arc::clone(&col_a), Arc::clone(&col_b)],
464            SeededRandomState::with_seed(1),
465            "test_hash".to_string(),
466        );
467
468        assert_eq!(expr1, expr2);
469    }
470
471    #[test]
472    fn test_hash_expr_eq_different_columns() {
473        let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0));
474        let col_b: PhysicalExprRef = Arc::new(Column::new("b", 1));
475        let col_c: PhysicalExprRef = Arc::new(Column::new("c", 2));
476
477        let expr1 = HashExpr::new(
478            vec![Arc::clone(&col_a), Arc::clone(&col_b)],
479            SeededRandomState::with_seed(1),
480            "test_hash".to_string(),
481        );
482
483        let expr2 = HashExpr::new(
484            vec![Arc::clone(&col_a), Arc::clone(&col_c)],
485            SeededRandomState::with_seed(1),
486            "test_hash".to_string(),
487        );
488
489        assert_ne!(expr1, expr2);
490    }
491
492    #[test]
493    fn test_hash_expr_eq_different_description() {
494        let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0));
495
496        let expr1 = HashExpr::new(
497            vec![Arc::clone(&col_a)],
498            SeededRandomState::with_seed(1),
499            "hash_one".to_string(),
500        );
501
502        let expr2 = HashExpr::new(
503            vec![Arc::clone(&col_a)],
504            SeededRandomState::with_seed(1),
505            "hash_two".to_string(),
506        );
507
508        assert_ne!(expr1, expr2);
509    }
510
511    #[test]
512    fn test_hash_expr_eq_different_seeds() {
513        let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0));
514
515        let expr1 = HashExpr::new(
516            vec![Arc::clone(&col_a)],
517            SeededRandomState::with_seed(1),
518            "test_hash".to_string(),
519        );
520
521        let expr2 = HashExpr::new(
522            vec![Arc::clone(&col_a)],
523            SeededRandomState::with_seed(5),
524            "test_hash".to_string(),
525        );
526
527        assert_ne!(expr1, expr2);
528    }
529
530    #[test]
531    fn test_hash_expr_hash_consistency() {
532        let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0));
533        let col_b: PhysicalExprRef = Arc::new(Column::new("b", 1));
534
535        let expr1 = HashExpr::new(
536            vec![Arc::clone(&col_a), Arc::clone(&col_b)],
537            SeededRandomState::with_seed(1),
538            "test_hash".to_string(),
539        );
540
541        let expr2 = HashExpr::new(
542            vec![Arc::clone(&col_a), Arc::clone(&col_b)],
543            SeededRandomState::with_seed(1),
544            "test_hash".to_string(),
545        );
546
547        // Equal expressions should have equal hashes
548        assert_eq!(expr1, expr2);
549        assert_eq!(compute_hash(&expr1), compute_hash(&expr2));
550    }
551
552    #[cfg(feature = "proto")]
553    mod proto_tests {
554        use super::*;
555        use arrow::datatypes::{DataType, Field};
556        use datafusion_common::internal_datafusion_err;
557        use datafusion_physical_expr_common::physical_expr::proto_decode::{
558            PhysicalExprDecode, PhysicalExprDecodeCtx,
559        };
560        use datafusion_physical_expr_common::physical_expr::proto_encode::{
561            PhysicalExprEncode, PhysicalExprEncodeCtx,
562        };
563        use datafusion_proto_models::protobuf;
564
565        struct TestEncoder;
566
567        impl PhysicalExprEncode for TestEncoder {
568            fn encode(
569                &self,
570                expr: &Arc<dyn PhysicalExpr>,
571            ) -> Result<protobuf::PhysicalExprNode> {
572                let ctx = PhysicalExprEncodeCtx::new(self);
573                expr.try_to_proto(&ctx)?.ok_or_else(|| {
574                    internal_datafusion_err!("test encoder cannot encode {expr:?}")
575                })
576            }
577        }
578
579        struct TestDecoder;
580
581        impl PhysicalExprDecode for TestDecoder {
582            fn decode(
583                &self,
584                node: &protobuf::PhysicalExprNode,
585                schema: &Schema,
586            ) -> Result<Arc<dyn PhysicalExpr>> {
587                let ctx = PhysicalExprDecodeCtx::new(schema, self);
588                match &node.expr_type {
589                    Some(protobuf::physical_expr_node::ExprType::Column(_)) => {
590                        Column::try_from_proto(node, &ctx)
591                    }
592                    _ => internal_err!("test decoder cannot decode {node:?}"),
593                }
594            }
595        }
596
597        fn test_decode_ctx<'a>(
598            schema: &'a Schema,
599            decoder: &'a TestDecoder,
600        ) -> PhysicalExprDecodeCtx<'a> {
601            PhysicalExprDecodeCtx::new(schema, decoder)
602        }
603
604        #[test]
605        fn hash_expr_try_to_proto() {
606            let expr = HashExpr::new(
607                vec![Arc::new(Column::new("a", 0)), Arc::new(Column::new("b", 1))],
608                SeededRandomState::with_seed(42),
609                "hash_join".to_string(),
610            );
611            let encoder = TestEncoder;
612            let ctx = PhysicalExprEncodeCtx::new(&encoder);
613
614            let proto = expr.try_to_proto(&ctx).unwrap().unwrap();
615
616            assert_eq!(proto.expr_id, None);
617            let hash_expr = match proto.expr_type.unwrap() {
618                protobuf::physical_expr_node::ExprType::HashExpr(hash_expr) => hash_expr,
619                other => panic!("expected HashExpr, got {other:?}"),
620            };
621            assert_eq!(hash_expr.seed0, 42);
622            assert_eq!(hash_expr.description, "hash_join");
623            assert_eq!(hash_expr.on_columns.len(), 2);
624            assert!(
625                hash_expr
626                    .on_columns
627                    .iter()
628                    .all(|expr| expr.expr_id.is_none())
629            );
630        }
631
632        #[test]
633        fn hash_expr_try_from_proto() {
634            let schema = Schema::new(vec![
635                Field::new("a", DataType::Int32, false),
636                Field::new("b", DataType::Utf8, true),
637            ]);
638            let decoder = TestDecoder;
639            let ctx = test_decode_ctx(&schema, &decoder);
640            let proto = protobuf::PhysicalExprNode {
641                expr_id: None,
642                expr_type: Some(protobuf::physical_expr_node::ExprType::HashExpr(
643                    protobuf::PhysicalHashExprNode {
644                        on_columns: vec![
645                            protobuf::PhysicalExprNode {
646                                expr_id: None,
647                                expr_type: Some(
648                                    protobuf::physical_expr_node::ExprType::Column(
649                                        protobuf::PhysicalColumn {
650                                            name: "a".to_string(),
651                                            index: 0,
652                                        },
653                                    ),
654                                ),
655                            },
656                            protobuf::PhysicalExprNode {
657                                expr_id: None,
658                                expr_type: Some(
659                                    protobuf::physical_expr_node::ExprType::Column(
660                                        protobuf::PhysicalColumn {
661                                            name: "b".to_string(),
662                                            index: 1,
663                                        },
664                                    ),
665                                ),
666                            },
667                        ],
668                        seed0: 42,
669                        description: "hash_join".to_string(),
670                    },
671                )),
672            };
673
674            let expr = HashExpr::try_from_proto(&proto, &ctx).unwrap();
675            let expr = expr.downcast_ref::<HashExpr>().unwrap();
676
677            assert_eq!(expr.seed(), 42);
678            assert_eq!(expr.description(), "hash_join");
679            assert_eq!(expr.on_columns().len(), 2);
680            assert_eq!(
681                expr.on_columns()[0]
682                    .downcast_ref::<Column>()
683                    .map(|col| (col.name(), col.index())),
684                Some(("a", 0))
685            );
686            assert_eq!(
687                expr.on_columns()[1]
688                    .downcast_ref::<Column>()
689                    .map(|col| (col.name(), col.index())),
690                Some(("b", 1))
691            );
692        }
693
694        #[test]
695        fn hash_expr_try_from_proto_rejects_wrong_node_type() {
696            let schema = Schema::empty();
697            let decoder = TestDecoder;
698            let ctx = test_decode_ctx(&schema, &decoder);
699            let proto = protobuf::PhysicalExprNode {
700                expr_id: None,
701                expr_type: Some(protobuf::physical_expr_node::ExprType::Column(
702                    protobuf::PhysicalColumn {
703                        name: "a".to_string(),
704                        index: 0,
705                    },
706                )),
707            };
708
709            let err = HashExpr::try_from_proto(&proto, &ctx).unwrap_err();
710            assert!(
711                err.to_string()
712                    .contains("PhysicalExprNode is not a HashExpr"),
713                "{err}"
714            );
715        }
716    }
717
718    #[test]
719    fn test_hash_table_lookup_expr_eq_same() {
720        let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0));
721        let hash_map =
722            Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(10))));
723
724        let expr1 = HashTableLookupExpr::new(
725            vec![Arc::clone(&col_a)],
726            SeededRandomState::with_seed(1),
727            Arc::clone(&hash_map),
728            "lookup".to_string(),
729        );
730
731        let expr2 = HashTableLookupExpr::new(
732            vec![Arc::clone(&col_a)],
733            SeededRandomState::with_seed(1),
734            Arc::clone(&hash_map),
735            "lookup".to_string(),
736        );
737
738        assert_eq!(expr1, expr2);
739    }
740
741    #[test]
742    fn test_hash_table_lookup_expr_eq_different_columns() {
743        let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0));
744        let col_b: PhysicalExprRef = Arc::new(Column::new("b", 1));
745
746        let hash_map =
747            Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(10))));
748
749        let expr1 = HashTableLookupExpr::new(
750            vec![Arc::clone(&col_a)],
751            SeededRandomState::with_seed(1),
752            Arc::clone(&hash_map),
753            "lookup".to_string(),
754        );
755
756        let expr2 = HashTableLookupExpr::new(
757            vec![Arc::clone(&col_b)],
758            SeededRandomState::with_seed(1),
759            Arc::clone(&hash_map),
760            "lookup".to_string(),
761        );
762
763        assert_ne!(expr1, expr2);
764    }
765
766    #[test]
767    fn test_hash_table_lookup_expr_eq_different_description() {
768        let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0));
769        let hash_map =
770            Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(10))));
771
772        let expr1 = HashTableLookupExpr::new(
773            vec![Arc::clone(&col_a)],
774            SeededRandomState::with_seed(1),
775            Arc::clone(&hash_map),
776            "lookup_one".to_string(),
777        );
778
779        let expr2 = HashTableLookupExpr::new(
780            vec![Arc::clone(&col_a)],
781            SeededRandomState::with_seed(1),
782            Arc::clone(&hash_map),
783            "lookup_two".to_string(),
784        );
785
786        assert_ne!(expr1, expr2);
787    }
788
789    #[test]
790    fn test_hash_table_lookup_expr_eq_different_hash_map() {
791        let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0));
792
793        // Two different Arc pointers (even with same content) should not be equal
794        let hash_map1 =
795            Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(10))));
796        let hash_map2 =
797            Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(10))));
798        let expr1 = HashTableLookupExpr::new(
799            vec![Arc::clone(&col_a)],
800            SeededRandomState::with_seed(1),
801            hash_map1,
802            "lookup".to_string(),
803        );
804
805        let expr2 = HashTableLookupExpr::new(
806            vec![Arc::clone(&col_a)],
807            SeededRandomState::with_seed(1),
808            hash_map2,
809            "lookup".to_string(),
810        );
811
812        // Different Arc pointers means not equal (uses Arc::ptr_eq)
813        assert_ne!(expr1, expr2);
814    }
815
816    #[test]
817    fn test_hash_table_lookup_expr_hash_consistency() {
818        let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0));
819        let hash_map =
820            Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(10))));
821
822        let expr1 = HashTableLookupExpr::new(
823            vec![Arc::clone(&col_a)],
824            SeededRandomState::with_seed(1),
825            Arc::clone(&hash_map),
826            "lookup".to_string(),
827        );
828
829        let expr2 = HashTableLookupExpr::new(
830            vec![Arc::clone(&col_a)],
831            SeededRandomState::with_seed(1),
832            Arc::clone(&hash_map),
833            "lookup".to_string(),
834        );
835
836        // Equal expressions should have equal hashes
837        assert_eq!(expr1, expr2);
838        assert_eq!(compute_hash(&expr1), compute_hash(&expr2));
839    }
840}
datafusion_physical_plan/joins/hash_join/partitioned_hash_eval.rs

datafusion_physical_plan/joins/hash_join/
partitioned_hash_eval.rs
