perf: simplify HashJoinExec dynamic filter, drop CASE routing

datafusion/common/src/config.rs

@@ -1309,11 +1309,15 @@ config_namespace! {
         /// very large IN lists that might not provide much benefit over hash table lookups.
         ///
         /// This uses the deduplicated row count once the build side has been evaluated.
+        /// In `Partitioned` hash-join mode the same threshold also gates the
+        /// cross-partition merged InList: per-partition InList arrays are
+        /// concatenated and deduplicated, and the merged `IN (SET)` only fires
+        /// when the union has at most this many distinct values.
         ///
-        /// The default is 150 values per partition.
-        /// This is inspired by Trino's `max-filter-keys-per-column` setting.
-        /// See: <https://trino.io/docs/current/admin/dynamic-filtering.html#dynamic-filter-collection-thresholds>
-        pub hash_join_inlist_pushdown_max_distinct_values: usize, default = 150
+        /// The default is 20 distinct values, tuned so the resulting `IN (SET)`
+        /// stays small enough to participate in parquet stats / bloom-filter
+        /// pruning at the scan side.
+        pub hash_join_inlist_pushdown_max_distinct_values: usize, default = 20
 
         /// The default filter selectivity used by Filter Statistics
         /// when an exact selectivity cannot be determined. Valid values are

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -1072,7 +1072,11 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() {
         .await
         .unwrap();
 
-    // Now check what our filter looks like
+    // The dynamic filter for a `Partitioned` hash join with a small enough
+    // cross-partition InList collapses to a single global `struct(...) IN
+    // (SET) ([...])`. The two `#[cfg]` arms differ only in the order of the
+    // InList elements, which is determined by partition-iteration order
+    // (normal repartition vs. the `force_hash_collisions` collapse).
     #[cfg(not(feature = "force_hash_collisions"))]
     insta::assert_snapshot!(
         format!("{}", format_plan_for_test(&plan)),
@@ -1083,14 +1087,10 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() {
     -       RepartitionExec: partitioning=Hash([a@0, b@1], 12), input_partitions=1
     -         DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
     -       RepartitionExec: partitioning=Hash([a@0, b@1], 12), input_partitions=1
-    -         DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ CASE hash_repartition % 12 WHEN 5 THEN a@0 >= ab AND a@0 <= ab AND b@1 >= bb AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:ab,c1:bb}]) WHEN 8 THEN a@0 >= aa AND a@0 <= aa AND b@1 >= ba AND b@1 <= ba AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}]) ELSE false END ]
+    -         DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:ab,c1:bb}, {c0:aa,c1:ba}]) ]
     "
     );
 
-    // When hash collisions force all data into a single partition, we optimize away the CASE expression.
-    // This avoids calling create_hashes() for every row on the probe side, since hash % 1 == 0 always,
-    // meaning the WHEN 0 branch would always match. This optimization is also important for primary key
-    // joins or any scenario where all build-side data naturally lands in one partition.
     #[cfg(feature = "force_hash_collisions")]
     insta::assert_snapshot!(
         format!("{}", format_plan_for_test(&plan)),
@@ -2572,11 +2572,17 @@ async fn test_hashjoin_hash_table_pushdown_partitioned() {
         .await
         .unwrap();
 
-    // Verify that hash_lookup is used instead of IN (SET)
+    // The dynamic filter for an all-Map Partitioned hash join uses a single
+    // shared `multi_hash_lookup` over every partition's hash table. There is
+    // no per-row `hash_repartition` routing.
     let plan_str = format_plan_for_test(&plan).to_string();
     assert!(
-        plan_str.contains("hash_lookup"),
-        "Expected hash_lookup in plan but got: {plan_str}"
+        plan_str.contains("multi_hash_lookup"),
+        "Expected multi_hash_lookup in plan but got: {plan_str}"
+    );
+    assert!(
+        !plan_str.contains("hash_repartition"),
+        "Expected no routing hash_repartition in plan but got: {plan_str}"
     );
     assert!(
         !plan_str.contains("IN (SET)"),

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

@@ -51,7 +51,6 @@ use crate::projection::{
     EmbeddedProjection, JoinData, ProjectionExec, try_embed_projection,
     try_pushdown_through_join,
 };
-use crate::repartition::REPARTITION_RANDOM_STATE;
 use crate::spill::get_record_batch_memory_size;
 use crate::{
     DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, Partitioning,
@@ -1316,9 +1315,8 @@ impl ExecutionPlan for HashJoinExec {
             .with_category(MetricCategory::Rows)
             .counter(ARRAY_MAP_CREATED_COUNT_METRIC_NAME, partition);
 
-        // Initialize build_accumulator lazily with runtime partition counts (only if enabled)
-        // Use RepartitionExec's random state (seeds: 0,0,0,0) for partition routing
-        let repartition_random_state = REPARTITION_RANDOM_STATE;
+        // Initialize build_accumulator lazily with runtime partition counts
+        // (only when dynamic filter pushdown is enabled).
         let build_accumulator = enable_dynamic_filter_pushdown
             .then(|| {
                 self.dynamic_filter.as_ref().map(|df| {
@@ -1328,14 +1326,19 @@ impl ExecutionPlan for HashJoinExec {
                         .iter()
                         .map(|(_, right_expr)| Arc::clone(right_expr))
                         .collect::<Vec<_>>();
+                    let inlist_max_distinct_values = context
+                        .session_config()
+                        .options()
+                        .optimizer
+                        .hash_join_inlist_pushdown_max_distinct_values;
                     Some(Arc::clone(df.build_accumulator.get_or_init(|| {
                         Arc::new(SharedBuildAccumulator::new_from_partition_mode(
                             self.mode,
                             self.left.as_ref(),
                             self.right.as_ref(),
                             filter,
                             on_right,
-                            repartition_random_state,
+                            inlist_max_distinct_values,
                         ))
                     })))
                 })
@@ -2038,29 +2041,31 @@ async fn collect_left_input(
 
     let map = Arc::new(join_hash_map);
 
+    // The hash map is needed by the join itself, so it always travels in
+    // `PushdownStrategy::map`. The optional `inlist` array is attached when
+    // the build side fits under the per-partition InList caps; the
+    // `SharedBuildAccumulator` may then merge those arrays across partitions
+    // into a single `IN (SET)` for scan-side pruning.
     let membership = if num_rows == 0 {
-        PushdownStrategy::Empty
+        PushdownStrategy::empty()
     } else {
-        // If the build side is small enough we can use IN list pushdown.
-        // If it's too big we fall back to pushing down a reference to the hash table.
-        // See `PushdownStrategy` for more details.
         let estimated_size = left_values
             .iter()
             .map(|arr| arr.get_array_memory_size())
             .sum::<usize>();
-        if left_values.is_empty()
-            || left_values[0].is_empty()
-            || estimated_size > config.optimizer.hash_join_inlist_pushdown_max_size
-            || map.num_of_distinct_key()
-                > config
+        let fits_inlist = !left_values.is_empty()
+            && !left_values[0].is_empty()
+            && estimated_size <= config.optimizer.hash_join_inlist_pushdown_max_size
+            && map.num_of_distinct_key()
+                <= config
                     .optimizer
-                    .hash_join_inlist_pushdown_max_distinct_values
+                    .hash_join_inlist_pushdown_max_distinct_values;
+        if fits_inlist
+            && let Some(in_list_values) = build_struct_inlist_values(&left_values)?
         {
-            PushdownStrategy::Map(Arc::clone(&map))
-        } else if let Some(in_list_values) = build_struct_inlist_values(&left_values)? {
-            PushdownStrategy::InList(in_list_values)
+            PushdownStrategy::from_map_and_inlist(Arc::clone(&map), in_list_values)
         } else {
-            PushdownStrategy::Map(Arc::clone(&map))
+            PushdownStrategy::from_map(Arc::clone(&map))
         }
     };
 

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

@@ -20,7 +20,7 @@
 use std::{fmt::Display, hash::Hash, sync::Arc};
 
 use arrow::{
-    array::{ArrayRef, UInt64Array},
+    array::{ArrayRef, BooleanArray, BooleanBufferBuilder, UInt64Array},
     datatypes::{DataType, Schema},
     record_batch::RecordBatch,
 };
@@ -203,7 +203,7 @@ impl PhysicalExpr for HashExpr {
 
 /// Physical expression that checks join keys in a [`Map`] (hash table or array map).
 ///
-/// Returns a [`BooleanArray`](arrow::array::BooleanArray) indicating if join keys (from `on_columns`) exist in the map.
+/// Returns a [`BooleanArray`] indicating if join keys (from `on_columns`) exist in the map.
 // TODO: rename to MapLookupExpr
 pub struct HashTableLookupExpr {
     /// Columns in the ON clause used to compute the join key for lookups
@@ -343,6 +343,197 @@ impl PhysicalExpr for HashTableLookupExpr {
     }
 }
 
+/// Physical expression that probes the same join keys against multiple [`Map`]s
+/// and returns `true` for any row whose join keys match at least one map.
+///
+/// Equivalent to `OR`-ing several [`HashTableLookupExpr`]s, but
+/// `create_hashes` runs exactly once for the whole batch and every
+/// [`Map::HashMap`] probe shares the same hash buffer. All `HashMap`
+/// entries must therefore have been built with the same `RandomState`;
+/// [`Map::ArrayMap`] entries are queried via `contain_keys` and do not
+/// consume hashes.
+pub struct MultiMapLookupExpr {
+    /// Join-key expressions evaluated against each input batch.
+    on_columns: Vec<PhysicalExprRef>,
+    /// Hashing seed shared by every entry in `maps`.
+    random_state: SeededRandomState,
+    /// Build-side maps to OR over, one per partition.
+    maps: Vec<Arc<Map>>,
+    /// Display name used in `EXPLAIN` output (e.g. `"multi_hash_lookup"`).
+    description: String,
+}
+
+impl MultiMapLookupExpr {
+    pub fn new(
+        on_columns: Vec<PhysicalExprRef>,
+        random_state: SeededRandomState,
+        maps: Vec<Arc<Map>>,
+        description: String,
+    ) -> Self {
+        Self {
+            on_columns,
+            random_state,
+            maps,
+            description,
+        }
+    }
+}
+
+impl std::fmt::Debug for MultiMapLookupExpr {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let cols = self
+            .on_columns
+            .iter()
+            .map(|e| e.to_string())
+            .collect::<Vec<_>>()
+            .join(", ");
+        write!(
+            f,
+            "{}({cols}, [{}], maps={})",
+            self.description,
+            self.random_state.seed(),
+            self.maps.len()
+        )
+    }
+}
+
+impl Hash for MultiMapLookupExpr {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.on_columns.dyn_hash(state);
+        self.description.hash(state);
+        self.random_state.seed().hash(state);
+        // See HashTableLookupExpr — pointer identity is what we use for Maps.
+        for map in &self.maps {
+            Arc::as_ptr(map).hash(state);
+        }
+    }
+}
+
+impl PartialEq for MultiMapLookupExpr {
+    fn eq(&self, other: &Self) -> bool {
+        self.on_columns == other.on_columns
+            && self.description == other.description
+            && self.random_state.seed() == other.random_state.seed()
+            && self.maps.len() == other.maps.len()
+            && self
+                .maps
+                .iter()
+                .zip(other.maps.iter())
+                .all(|(a, b)| Arc::ptr_eq(a, b))
+    }
+}
+
+impl Eq for MultiMapLookupExpr {}
+
+impl Display for MultiMapLookupExpr {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.description)
+    }
+}
+
+impl PhysicalExpr for MultiMapLookupExpr {
+    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
+        self.on_columns.iter().collect()
+    }
+
+    fn with_new_children(
+        self: Arc<Self>,
+        children: Vec<Arc<dyn PhysicalExpr>>,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        Ok(Arc::new(MultiMapLookupExpr::new(
+            children,
+            self.random_state.clone(),
+            self.maps.clone(),
+            self.description.clone(),
+        )))
+    }
+
+    fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
+        Ok(DataType::Boolean)
+    }
+
+    fn nullable(&self, _input_schema: &Schema) -> Result<bool> {
+        Ok(false)
+    }
+
+    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
+        let num_rows = batch.num_rows();
+        let join_keys = evaluate_columns(&self.on_columns, batch)?;
+
+        if self.maps.is_empty() || num_rows == 0 {
+            // Empty `maps` would not be constructed by the dynamic-filter
+            // builder — guard anyway: an empty OR is `false` for every row.
+            let buffer = BooleanBufferBuilder::new(num_rows);
+            let mut buffer = buffer;
+            buffer.append_n(num_rows, false);
+            return Ok(ColumnarValue::Array(Arc::new(BooleanArray::new(
+                buffer.finish(),
+                None,
+            ))));
+        }
+
+        // Hashes are only needed for `HashMap` probes; `ArrayMap` queries
+        // its keys directly via `contain_keys`.
+        let needs_hashes = self
+            .maps
+            .iter()
+            .any(|m| matches!(m.as_ref(), Map::HashMap(_)));
+
+        let mut result = vec![false; num_rows];
+
+        let process_one_map =
+            |result: &mut [bool], map: &Map, hashes: Option<&[u64]>| -> Result<()> {
+                match map {
+                    Map::HashMap(hm) => {
+                        let hashes = hashes
+                            .expect("hashes computed when at least one map is a HashMap");
+                        let arr = hm.contain_hashes(hashes);
+                        // `contain_hashes` always returns a non-null
+                        // `BooleanArray`; OR its bits into `result`.
+                        for (slot, hit) in result.iter_mut().zip(arr.values().iter()) {
+                            *slot |= hit;
+                        }
+                    }
+                    Map::ArrayMap(am) => {
+                        let arr = am.contain_keys(&join_keys)?;
+                        for (slot, hit) in result.iter_mut().zip(arr.values().iter()) {
+                            *slot |= hit;
+                        }
+                    }
+                }
+                Ok(())
+            };
+
+        if needs_hashes {
+            // Hash the join keys once and reuse the buffer for every
+            // `HashMap` probe; `ArrayMap` probes pass `None` for hashes.
+            with_hashes(&join_keys, self.random_state.random_state(), |hashes| {
+                for map in &self.maps {
+                    process_one_map(&mut result, map, Some(hashes))?;
+                }
+                Ok::<(), datafusion_common::DataFusionError>(())
+            })?;
+        } else {
+            for map in &self.maps {
+                process_one_map(&mut result, map, None)?;
+            }
+        }
+
+        let mut buffer = BooleanBufferBuilder::new(num_rows);
+        for hit in &result {
+            buffer.append(*hit);
+        }
+        Ok(ColumnarValue::Array(Arc::new(BooleanArray::new(
+            buffer.finish(),
+            None,
+        ))))
+    }
+
+    fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.description)
+    }
+}
+
 fn evaluate_columns(
     columns: &[PhysicalExprRef],
     batch: &RecordBatch,