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/mod.rs

@@ -18,7 +18,11 @@
 //! [`HashJoinExec`] Partitioned Hash Join Operator
 
 pub use exec::{HashJoinExec, HashJoinExecBuilder};
-pub use partitioned_hash_eval::{HashExpr, HashTableLookupExpr, SeededRandomState};
+#[expect(deprecated)]
+pub use partitioned_hash_eval::HashExpr;
+pub use partitioned_hash_eval::{
+    HashTableLookupExpr, MultiMapLookupExpr, SeededRandomState,
+};
 
 mod exec;
 mod inlist_builder;