Prune hash join probe side via dynamic discrete values from the build side

datafusion/common/src/config.rs

@@ -1844,6 +1844,26 @@ config_namespace! {
         /// 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
 
+        /// Maximum number of distinct build-side values to retain for row-group/file
+        /// min/max-based pruning once the build side is too large for `InList` pushdown
+        /// and falls back to an opaque hash-table-lookup filter. Set to 0 to disable.
+        ///
+        /// On by default: the check is footer-only (no bloom filter or extra I/O),
+        /// reusing the sorted-domain rewrite an ordinary large `IN (...)` list already
+        /// gets, so a container is kept only if its own min/max overlaps a value.
+        ///
+        /// When engaged, `EXPLAIN`'s `pruning_predicate=` gains an extra
+        /// `IN_SET_INTERSECTS(<col>_min, <col>_max, <n> values)` clause - the visible
+        /// sign this ran, distinct from the plain min/max bounds every join pushes down.
+        pub hash_join_dynamic_pruning_max_distinct_values: usize, default = 100_000
+
+        /// Companion size cap (bytes) for `hash_join_dynamic_pruning_max_distinct_values`,
+        /// mirroring `hash_join_inlist_pushdown_max_size`. Set to 0 to disable.
+        ///
+        /// Checked against the *raw*, undeduplicated build-side column, so this also
+        /// guards against few distinct values but many duplicate rows.
+        pub hash_join_dynamic_pruning_max_size: usize, default = 8 * 1024 * 1024
+
         /// The default filter selectivity used by Filter Statistics
         /// when an exact selectivity cannot be determined. Valid values are
         /// between 0 (no selectivity) and 100 (all rows are selected).

datafusion/common/src/scalar/mod.rs

@@ -4314,6 +4314,20 @@ impl ScalarValue {
         })
     }
 
+    /// Converts `array`'s values into non-null [`ScalarValue`]s, preserving duplicates,
+    /// returning `None` if any value fails to convert (e.g. an unsupported type).
+    pub fn nonnull_scalars(array: &dyn Array) -> Option<Vec<ScalarValue>> {
+        let mut values = Vec::with_capacity(array.len());
+        for i in 0..array.len() {
+            match ScalarValue::try_from_array(array, i) {
+                Ok(v) if !v.is_null() => values.push(v),
+                Ok(_) => {} // NULL never satisfies `=`; harmless to drop
+                Err(_) => return None,
+            }
+        }
+        Some(values)
+    }
+
     /// Try to parse `value` into a ScalarValue of type `target_type`
     pub fn try_from_string(value: String, target_type: &DataType) -> Result<Self> {
         ScalarValue::from(value).cast_to(target_type)
@@ -6051,6 +6065,28 @@ mod tests {
     use insta::assert_snapshot;
     use rand::Rng;
 
+    #[test]
+    fn test_nonnull_scalars() {
+        let array = Int32Array::from(vec![Some(1), Some(2), Some(1), None, Some(2)]);
+        let values = ScalarValue::nonnull_scalars(&array).expect("convertible");
+        assert_eq!(
+            values,
+            vec![
+                ScalarValue::Int32(Some(1)),
+                ScalarValue::Int32(Some(2)),
+                ScalarValue::Int32(Some(1)),
+                ScalarValue::Int32(Some(2)),
+            ]
+        );
+    }
+
+    #[test]
+    fn test_nonnull_scalars_all_null() {
+        let array = Int32Array::from(vec![None, None]);
+        let values = ScalarValue::nonnull_scalars(&array).expect("convertible");
+        assert!(values.is_empty());
+    }
+
     #[test]
     fn test_scalar_value_from_for_map() {
         let string_builder = StringBuilder::new();

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

@@ -2961,19 +2961,37 @@ async fn collect_left_input(
             .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 pushdown_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 pushdown_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)
         } else {
-            PushdownStrategy::Map(Arc::clone(&map))
+            // Past the InList threshold, retain raw values for pruning only (not row
+            // filtering) up to a separate, more generous cap; dedup happens lazily
+            // inside `HashTableLookupExpr` on first actual use, not eagerly here.
+            let pushdown_values = !left_values.is_empty()
+                && !left_values[0].is_empty()
+                && estimated_size <= config.optimizer.hash_join_dynamic_pruning_max_size
+                && map.num_of_distinct_key()
+                    <= config
+                        .optimizer
+                        .hash_join_dynamic_pruning_max_distinct_values;
+
+            if pushdown_values {
+                let pruning_literals = build_struct_inlist_values(&left_values)?;
+                PushdownStrategy::Map(Arc::clone(&map), pruning_literals)
+            } else {
+                PushdownStrategy::Map(Arc::clone(&map), None)
+            }
         }
     };
 

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

@@ -19,8 +19,9 @@
 
 use std::sync::Arc;
 
-use arrow::array::{ArrayRef, StructArray};
-use arrow::datatypes::{Field, FieldRef, Fields};
+use arrow::array::{ArrayRef, AsArray, StructArray};
+use arrow::compute::cast;
+use arrow::datatypes::{Field, FieldRef, Fields, Int32Type};
 use arrow_schema::DataType;
 use datafusion_common::Result;
 
@@ -77,6 +78,25 @@ pub(super) fn build_struct_inlist_values(
     Ok(Some(source_array))
 }
 
+/// Deduplicates `array`'s values (dropping nulls) via Arrow's dictionary-encoding cast,
+/// cheaper than converting every row to a `ScalarValue` to test uniqueness.
+///
+/// Returns `None` if the type isn't dictionary-encodable or every value is null -
+/// both safe to treat as "no pruning literals available".
+pub(super) fn dedupe_array_values(array: &ArrayRef) -> Option<ArrayRef> {
+    let dict_type = DataType::Dictionary(
+        Box::new(DataType::Int32),
+        Box::new(array.data_type().clone()),
+    );
+    let dict = cast(array.as_ref(), &dict_type).ok()?;
+    let values = Arc::clone(dict.as_dictionary::<Int32Type>().values());
+    if values.is_empty() {
+        None
+    } else {
+        Some(values)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -142,6 +162,27 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_dedupe_array_values() {
+        let array = Arc::new(Int32Array::from(vec![
+            Some(1),
+            Some(2),
+            Some(1),
+            None,
+            Some(2),
+        ])) as ArrayRef;
+        let deduped = dedupe_array_values(&array).expect("dictionary-encodable");
+        assert_eq!(deduped.len(), 2);
+        let deduped = deduped.as_ref().as_primitive::<Int32Type>();
+        assert_eq!(deduped.values(), &[1, 2]);
+    }
+
+    #[test]
+    fn test_dedupe_array_values_all_null() {
+        let array = Arc::new(Int32Array::from(vec![None, None])) as ArrayRef;
+        assert!(dedupe_array_values(&array).is_none());
+    }
+
     #[test]
     fn test_build_single_column_dictionary_inlist() {
         let keys = Int8Array::from(vec![0i8, 0, 0]);