diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 360586b0e9bae..97512cfcb727f 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1844,6 +1844,26 @@ config_namespace! { /// See: 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(_min, _max, 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). diff --git a/datafusion/common/src/scalar/mod.rs b/datafusion/common/src/scalar/mod.rs index bad526a3a2227..92045d5bf6711 100644 --- a/datafusion/common/src/scalar/mod.rs +++ b/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> { + 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 { 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(); diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index b72e180543f9a..28f0314d60440 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/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::(); - 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) + } } }; diff --git a/datafusion/physical-plan/src/joins/hash_join/inlist_builder.rs b/datafusion/physical-plan/src/joins/hash_join/inlist_builder.rs index 2fc3201c6363f..1f654dd22d05a 100644 --- a/datafusion/physical-plan/src/joins/hash_join/inlist_builder.rs +++ b/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 { + 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::().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::(); + 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]); diff --git a/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs b/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs index 98e8b2d2fc42e..e196d47b1b095 100644 --- a/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs +++ b/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs @@ -20,11 +20,12 @@ use std::{fmt::Display, hash::Hash, sync::Arc}; use arrow::{ - array::{ArrayRef, UInt64Array}, + array::{Array, ArrayRef, UInt64Array}, datatypes::{DataType, Schema}, record_batch::RecordBatch, }; use datafusion_common::Result; +use datafusion_common::ScalarValue; use datafusion_common::hash_utils::RandomState; use datafusion_common::hash_utils::{create_hashes, with_hashes}; #[cfg(feature = "proto")] @@ -271,6 +272,52 @@ impl HashExpr { } } +/// A single-column dynamic-pruning literal set: the tested column and its non-null, +/// deduplicated build-side values. Returned by [`HashTableLookupExpr::cached_pruning_scalars`]. +pub type PruningScalars = (Arc, Arc<[ScalarValue]>); + +/// A build side's raw values, deduplicated and converted to non-null [`ScalarValue`]s +/// lazily on first use and cached from then on. +/// +/// `PruningPredicate` is rebuilt independently for file, row-group, and page-index +/// pruning, and again per partition, so without this the conversion would re-run on +/// every rebuild - this dominated wall time for large build sides. +struct LazyPruningScalars { + /// Raw (undeduplicated) build-side values, or `None` if unavailable. Kept around + /// after `cache` is populated so [`HashTableLookupExpr::with_new_children`] can + /// seed a fresh instance without forcing a recompute here. + raw: Option, + cache: std::sync::OnceLock>>, +} + +impl LazyPruningScalars { + fn new(raw: Option) -> Self { + Self { + raw, + cache: std::sync::OnceLock::new(), + } + } + + /// Returns the deduplicated, non-null scalars, computing and caching them on + /// the first call. `distinct_count` (the map's, covering non-null build rows + /// only) is only consulted then, to decide whether `raw` is already unique + /// (common star-schema case) and dedup can be skipped. + fn get_or_init(&self, distinct_count: usize) -> Option<&Arc<[ScalarValue]>> { + self.cache + .get_or_init(|| { + let raw = self.raw.as_ref()?; + // Compare against the non-null row count, to match `distinct_count`. + let deduped = if distinct_count == raw.len() - raw.null_count() { + Arc::clone(raw) + } else { + super::inlist_builder::dedupe_array_values(raw)? + }; + ScalarValue::nonnull_scalars(deduped.as_ref()).map(Arc::from) + }) + .as_ref() + } +} + /// 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. @@ -284,6 +331,9 @@ pub struct HashTableLookupExpr { map: Arc, /// Description for display description: String, + /// Build-side values for dynamic pruning only (see + /// [`Self::cached_pruning_scalars`]) - `evaluate` always uses `map` instead. + pruning_scalars: LazyPruningScalars, } impl HashTableLookupExpr { /// Create a new HashTableLookupExpr @@ -293,6 +343,7 @@ impl HashTableLookupExpr { /// * `random_state` - SeededRandomState for hashing /// * `map` - Map to check membership (hash table or array map) /// * `description` - Description for debugging + /// * `raw_pruning_values` - undeduplicated build-side values for pruning only, or `None` /// # Note /// This is public for internal testing purposes only and is not /// guaranteed to be stable across versions. @@ -301,14 +352,29 @@ impl HashTableLookupExpr { random_state: SeededRandomState, map: Arc, description: String, + raw_pruning_values: Option, ) -> Self { Self { on_columns, random_state, map, description, + pruning_scalars: LazyPruningScalars::new(raw_pruning_values), } } + + /// If this lookup is on a single column, returns the tested column and its + /// deduplicated, non-null build-side values as [`ScalarValue`]s, so pruning code + /// can treat it like an IN-list. `None` for composite (multi-column) keys. + pub fn cached_pruning_scalars(&self) -> Option { + if self.on_columns.len() != 1 { + return None; + } + let scalars = self + .pruning_scalars + .get_or_init(self.map.num_of_distinct_key())?; + Some((Arc::clone(&self.on_columns[0]), Arc::clone(scalars))) + } } impl std::fmt::Debug for HashTableLookupExpr { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { @@ -377,6 +443,7 @@ impl PhysicalExpr for HashTableLookupExpr { self.random_state.clone(), Arc::clone(&self.map), self.description.clone(), + self.pruning_scalars.raw.clone(), ))) } @@ -423,6 +490,7 @@ impl PhysicalExpr for HashTableLookupExpr { random_state: _, map: _, description: _, + pruning_scalars: _, } = self; // HashTableLookupExpr holds a runtime Arc (the build-side hash @@ -468,8 +536,9 @@ fn evaluate_columns( #[cfg(test)] mod tests { use super::*; - use crate::joins::join_hash_map::JoinHashMapU32; + use crate::joins::join_hash_map::{JoinHashMapType, JoinHashMapU32}; use datafusion_physical_expr::expressions::Column; + use rstest::rstest; use std::collections::hash_map::DefaultHasher; use std::hash::Hasher; @@ -479,6 +548,83 @@ mod tests { hasher.finish() } + /// Builds a `JoinHashMapU32` containing exactly `distinct_hashes.len()` entries - + /// only the count matters for `num_of_distinct_key()`, not the hash content. + fn hash_map_with_distinct_count(distinct_hashes: &[u64]) -> Arc { + let mut map = JoinHashMapU32::with_capacity(distinct_hashes.len()); + JoinHashMapType::update_from_iter( + &mut map, + Box::new(distinct_hashes.iter().enumerate()), + 0, + ); + Arc::new(Map::HashMap(Box::new(map))) + } + + /// Covers three shapes of `cached_pruning_scalars`: deduplicating a build side + /// with real duplicates, passing one through unchanged when already unique + /// (also checking the result is cached, not recomputed on a second call), and + /// returning `None` when no raw values were populated in the first place. + #[rstest] + #[case::dedups_when_duplicates_present(Some(vec![1, 2, 1, 3]), Some(vec![1, 2, 3]))] + #[case::matches_when_already_unique(Some(vec![1, 2, 3]), Some(vec![1, 2, 3]))] + #[case::absent_when_not_populated(None, None)] + fn test_cached_pruning_scalars( + #[case] raw_values: Option>, + #[case] expected: Option>, + ) { + let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0)); + // 3 distinct keys: dedups the 4-value case, is a no-op for the 3-value one. + let hash_map = hash_map_with_distinct_count(&[100, 200, 300]); + let values: Option = + raw_values.map(|v| Arc::new(arrow::array::Int32Array::from(v)) as ArrayRef); + + let expr = HashTableLookupExpr::new( + vec![Arc::clone(&col_a)], + SeededRandomState::with_seed(1), + hash_map, + "hash_lookup".to_string(), + values, + ); + + match (expr.cached_pruning_scalars(), expected) { + (None, None) => {} + (Some((col, scalars)), Some(expected)) => { + assert_eq!(col.to_string(), col_a.to_string()); + let expected: Vec = expected + .into_iter() + .map(|v| ScalarValue::Int32(Some(v))) + .collect(); + assert_eq!(scalars.as_ref(), expected.as_slice()); + + // Second call hits the cache: same allocation, not reconverted. + let (_, scalars_again) = expr.cached_pruning_scalars().unwrap(); + assert!(Arc::ptr_eq(&scalars, &scalars_again)); + } + (actual, expected) => { + panic!("expected {expected:?}, got {:?}", actual.map(|(_, s)| s)) + } + } + } + + #[test] + fn test_cached_pruning_scalars_absent_for_multi_column_keys() { + let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0)); + let col_b: PhysicalExprRef = Arc::new(Column::new("b", 1)); + let hash_map = + Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(10)))); + let values: ArrayRef = Arc::new(arrow::array::Int32Array::from(vec![1, 2, 3])); + + let expr = HashTableLookupExpr::new( + vec![col_a, col_b], + SeededRandomState::with_seed(1), + hash_map, + "hash_lookup".to_string(), + Some(values), + ); + + assert!(expr.cached_pruning_scalars().is_none()); + } + #[test] fn test_hash_expr_eq_same() { let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0)); @@ -757,6 +903,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -764,6 +911,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); assert_eq!(expr1, expr2); @@ -782,6 +930,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -789,6 +938,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); assert_ne!(expr1, expr2); @@ -805,6 +955,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup_one".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -812,6 +963,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup_two".to_string(), + None, ); assert_ne!(expr1, expr2); @@ -831,6 +983,7 @@ mod tests { SeededRandomState::with_seed(1), hash_map1, "lookup".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -838,6 +991,7 @@ mod tests { SeededRandomState::with_seed(1), hash_map2, "lookup".to_string(), + None, ); // Different Arc pointers means not equal (uses Arc::ptr_eq) @@ -855,6 +1009,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -862,6 +1017,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); // Equal expressions should have equal hashes diff --git a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs index 62087c14c5179..c1f124acfb7e7 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -136,12 +136,15 @@ fn create_membership_predicate( )?))) } // Use hash table lookup for large build sides - PushdownStrategy::Map(hash_map) => Ok(Some(Arc::new(HashTableLookupExpr::new( - on_right.to_vec(), - random_state.clone(), - hash_map, - "hash_lookup".to_string(), - )) as Arc)), + PushdownStrategy::Map(hash_map, pruning_literals) => { + Ok(Some(Arc::new(HashTableLookupExpr::new( + on_right.to_vec(), + random_state.clone(), + hash_map, + "hash_lookup".to_string(), + pruning_literals, + )) as Arc)) + } // Empty partition - should not create a filter for this PushdownStrategy::Empty => Ok(None), } @@ -277,8 +280,10 @@ pub(crate) struct SharedBuildAccumulator { pub(crate) enum PushdownStrategy { /// Use InList for small build sides (< 128MB) InList(ArrayRef), - /// Use map lookup for large build sides - Map(Arc), + /// Use map lookup for large build sides. The second field is the distinct + /// build-side values for pruning only (see `hash_join_dynamic_pruning_max_distinct_values`), + /// `None` if unavailable. + Map(Arc, Option), /// There was no data in this partition, do not build a dynamic filter for it Empty, } diff --git a/datafusion/proto/tests/cases/plans/exprs.rs b/datafusion/proto/tests/cases/plans/exprs.rs index f2b14b043959f..29ae980442810 100644 --- a/datafusion/proto/tests/cases/plans/exprs.rs +++ b/datafusion/proto/tests/cases/plans/exprs.rs @@ -117,6 +117,7 @@ fn roundtrip_hash_table_lookup_expr_to_lit() -> Result<()> { datafusion::physical_plan::joins::SeededRandomState::with_seed(0), hash_map, "test_lookup".to_string(), + None, )); // Create a filter with the lookup expression diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs index b49b72058e0cd..21fe514b6a834 100644 --- a/datafusion/pruning/src/pruning_predicate.rs +++ b/datafusion/pruning/src/pruning_predicate.rs @@ -52,6 +52,7 @@ use datafusion_expr_common::operator::Operator; use datafusion_physical_expr::utils::{Guarantee, LiteralGuarantee}; use datafusion_physical_expr::{PhysicalExprRef, expressions as phys_expr}; use datafusion_physical_expr_common::physical_expr::snapshot_physical_expr_opt; +use datafusion_physical_plan::joins::HashTableLookupExpr; use datafusion_physical_plan::{ColumnarValue, PhysicalExpr}; /// Used to prove that arbitrary predicates (boolean expression) can not @@ -1520,6 +1521,108 @@ impl CompactInListDomain { } } +/// Builds the same kind of compact, sorted-domain "may match" expression as +/// [`build_compact_in_list_expr`], but for a [`HashTableLookupExpr`] (a large join +/// build side pushed down as an opaque hash-table lookup, not exposed via +/// [`InListExpr`]): tests container min/max stats against the build-side values. +/// +/// Always `IN` semantics (`lookup` never represents `NOT IN`) with nulls already +/// stripped by [`HashTableLookupExpr::cached_pruning_scalars`], so this skips the +/// negation/all-null-list handling `build_compact_in_list_expr` needs. +/// +/// Deliberately **not** gated by `max_in_list_size`: that cap is for literal SQL +/// `IN (...)` lists, and its small default (20) would defeat this for every build +/// side big enough to reach here. `hash_join_dynamic_pruning_max_distinct_values` +/// and `_max_size` bound this instead. +/// +/// [`InListExpr`]: datafusion_physical_expr::expressions::InListExpr +fn build_hash_lookup_pruning_expr( + lookup: &HashTableLookupExpr, + schema: &Schema, + required_columns: &mut RequiredColumns, +) -> Option> { + let (column_expr, values) = lookup.cached_pruning_scalars()?; + if values.is_empty() { + return None; + } + let column = column_expr.downcast_ref::()?; + let field = schema.fields().get(column.index())?; + if field.name() != column.name() { + return None; + } + let data_type = match field.data_type() { + DataType::Dictionary(_, value) => value.as_ref(), + data_type => data_type, + }; + let mut domain = if data_type.is_string() { + CompactInListDomain::String(Vec::with_capacity(values.len())) + } else if matches!( + data_type, + DataType::Binary | DataType::LargeBinary | DataType::BinaryView + ) { + CompactInListDomain::Binary(Vec::with_capacity(values.len())) + } else { + CompactInListDomain::Primitive(PrimitiveInListDomain::new( + data_type, + values.len(), + )?) + }; + for value in values.iter() { + let value = unwrap_scalar(value); + match &mut domain { + CompactInListDomain::String(vals) => { + vals.push(unpack_string(value)?.to_owned()) + } + CompactInListDomain::Binary(vals) => vals.push(extract_binary(value)?.into()), + CompactInListDomain::Primitive(vals) => vals.push(value)?, + } + } + if domain.is_empty() { + return None; + } + + // Roll back appended statistics columns if the rewrite cannot be completed. + // `RequiredColumns::stat_column_expr` only appends entries. + let required_columns_len = required_columns.columns.len(); + let statistics = (|| { + let min = required_columns + .min_column_expr(column, &column_expr, field) + .ok()?; + let max = required_columns + .max_column_expr(column, &column_expr, field) + .ok()?; + let non_null = + build_is_null_column_expr(&column_expr, schema, required_columns, true)?; + Some((min, max, non_null)) + })(); + let Some((min, max, non_null)) = statistics else { + required_columns.columns.truncate(required_columns_len); + return None; + }; + let may_match = match domain { + CompactInListDomain::String(values) => Arc::new(StringInListPruningExpr::new( + SetMembership::In, + min, + max, + values, + )) as PhysicalExprRef, + CompactInListDomain::Binary(values) => Arc::new(BinaryInListPruningExpr::new( + SetMembership::In, + min, + max, + values, + )) as PhysicalExprRef, + CompactInListDomain::Primitive(values) => { + values.into_expr(SetMembership::In, min, max) + } + }; + Some(Arc::new(phys_expr::BinaryExpr::new( + non_null, + Operator::And, + may_match, + ))) +} + /// Keep large literal lists of supported ordered types compact instead of /// building a per-value tree: an OR tree for `IN`, an AND chain for `NOT IN`. /// @@ -1816,6 +1919,10 @@ fn build_predicate_expression( return unhandled_hook.handle(expr); } } + if let Some(lookup) = expr.downcast_ref::() { + return build_hash_lookup_pruning_expr(lookup, schema, required_columns) + .unwrap_or_else(|| unhandled_hook.handle(expr)); + } let (left, op, right) = { if let Some(bin_expr) = expr.downcast_ref::() { @@ -7340,4 +7447,53 @@ mod tests { "c1_null_count@2 != row_count@3 AND c1_min@0 <= a AND a <= c1_max@1"; assert_eq!(res.to_string(), expected); } + + #[test] + fn test_hash_lookup_pruning_via_min_max() { + use datafusion_physical_plan::joins::join_hash_map::{ + JoinHashMapType, JoinHashMapU32, + }; + use datafusion_physical_plan::joins::{Map, SeededRandomState}; + + let mut hash_map = JoinHashMapU32::with_capacity(3); + let hashes = [100u64, 200, 300]; + JoinHashMapType::update_from_iter( + &mut hash_map, + Box::new(hashes.iter().enumerate()), + 0, + ); + let map = Arc::new(Map::HashMap(Box::new(hash_map))); + + let schema = Arc::new(Schema::new(vec![Field::new("b", DataType::Int32, true)])); + let column: Arc = Arc::new(phys_expr::Column::new("b", 0)); + let values: ArrayRef = Arc::new(Int32Array::from(vec![10, 20, 30])); + let lookup: Arc = Arc::new(HashTableLookupExpr::new( + vec![Arc::clone(&column)], + SeededRandomState::with_seed(1), + map, + "hash_lookup".to_string(), + Some(values), + )); + + let predicate = PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(&schema)) + .try_build(lookup) + .unwrap(); + + // No `with_contained` call anywhere: `contained()` always returns `None`, + // matching real row-group/file statistics (no bloom filter). Exclusion here + // can only come from the min/max-only rewrite, not from `LiteralGuarantee`. + let statistics = TestStatistics::new().with( + "b", + ContainerStats::new_i32( + vec![Some(5), Some(15), Some(100)], + vec![Some(8), Some(25), Some(200)], + ), + ); + + let result = predicate.prune(&statistics).unwrap(); + // Container 0 ([5,8]) and container 2 ([100,200]) contain none of {10,20,30}; + // container 1 ([15,25]) contains 20 - kept. + assert_eq!(result, vec![false, true, false]); + } } diff --git a/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt b/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt index c674dede75706..f32299aee401c 100644 --- a/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt +++ b/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt @@ -318,6 +318,33 @@ SELECT b FROM rgsel WHERE a >= 50 ORDER BY b ASC LIMIT 5; statement ok set datafusion.execution.parquet.pushdown_filters = true; +# Regression test: a hash join's build side must get the same compact, +# sorted-domain min/max pruning an ordinary large `IN (...)` list already +# gets - and do better than the plain min/max *bounds* check every join +# already gets for free. `dim`'s overall envelope [1000, 3050] spans RG 1 +# (b=2000..2099) entirely, so bounds alone cannot exclude it; only the +# discrete check can prove none of {1000, 1050, 3050} falls inside it. +# Force the hash-table-lookup path (rather than `InList`) regardless of size. +statement ok +CREATE TABLE dim AS VALUES (1000), (1050), (3050); + +statement ok +set datafusion.optimizer.hash_join_inlist_pushdown_max_distinct_values = 0; + +query TT +explain analyze select rgsel.b from dim join rgsel on dim.column1 = rgsel.b; +---- +Plan with Metrics +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(column1@0, b@0)], projection=[b@1], metrics=[output_rows=3, elapsed_compute=, output_bytes=, output_batches=1, build_mem_used=, array_map_created_count=0, build_input_batches=1, build_input_rows=3, input_batches=2, input_rows=3, build_time=, join_time=, avg_fanout=100% (3/3), probe_hit_rate=100% (3/3)] +02)--DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] +03)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_row_group_pruning/rgsel.parquet]]}, projection=[b], file_type=parquet, predicate=DynamicFilter [ b@1 >= 1000 AND b@1 <= 3050 AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 >= 1000 AND b_null_count@1 != row_count@2 AND b_min@3 <= 3050 AND b_null_count@1 != row_count@2 AND IN_SET_INTERSECTS(b_min@3, b_max@0, 3 values), required_guarantees=[], metrics=[output_rows=3, elapsed_compute=, output_bytes=, output_batches=2, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 2 matched, row_groups_pruned_bloom_filter=2 total → 2 matched, page_index_pages_pruned=20 total → 3 matched, page_index_rows_pruned=200 total → 30 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, bytes_processed=, bytes_scanned=, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=3, pushdown_rows_pruned=27, row_groups_pruned_dynamic_filter=0, predicate_cache_inner_records=200, predicate_cache_records=13, bloom_filter_eval_time=, metadata_load_time=, page_index_eval_time=, row_pushdown_eval_time=, statistics_eval_time=, time_elapsed_opening=, time_elapsed_processing=, time_elapsed_scanning_total=, time_elapsed_scanning_until_data=, output_rows_skew=, scan_efficiency_ratio=] + +statement ok +RESET datafusion.optimizer.hash_join_inlist_pushdown_max_distinct_values; + +statement ok +drop table dim; + statement ok drop table rgsel; diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index b270eba99d7b0..498b528aa9d33 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -322,6 +322,8 @@ datafusion.optimizer.enable_window_limits true datafusion.optimizer.enable_window_topn false datafusion.optimizer.expand_views_at_output false datafusion.optimizer.filter_null_join_keys false +datafusion.optimizer.hash_join_dynamic_pruning_max_distinct_values 100000 +datafusion.optimizer.hash_join_dynamic_pruning_max_size 8388608 datafusion.optimizer.hash_join_inlist_pushdown_max_distinct_values 150 datafusion.optimizer.hash_join_inlist_pushdown_max_size 131072 datafusion.optimizer.hash_join_single_partition_threshold 4194304 @@ -483,6 +485,8 @@ datafusion.optimizer.enable_window_limits true When set to true, the optimizer w datafusion.optimizer.enable_window_topn false When set to true, the optimizer will replace Filter(rn<=K) → Window(ROW_NUMBER) → Sort patterns with a PartitionedTopKExec that maintains per-partition heaps, avoiding a full sort of the input. When the window partition key has low cardinality, enabling this optimization can improve performance. However, for high cardinality keys, it may cause regressions in both memory usage and runtime. datafusion.optimizer.expand_views_at_output false When set to true, if the returned type is a view type then the output will be coerced to a non-view. Coerces `Utf8View` to `LargeUtf8`, and `BinaryView` to `LargeBinary`. datafusion.optimizer.filter_null_join_keys false When set to true, the optimizer will insert filters before a join between a nullable and non-nullable column to filter out nulls on the nullable side. This filter can add additional overhead when the file format does not fully support predicate push down. +datafusion.optimizer.hash_join_dynamic_pruning_max_distinct_values 100000 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(_min, _max, values)` clause - the visible sign this ran, distinct from the plain min/max bounds every join pushes down. +datafusion.optimizer.hash_join_dynamic_pruning_max_size 8388608 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. datafusion.optimizer.hash_join_inlist_pushdown_max_distinct_values 150 Maximum number of distinct values (rows) in the build side of a hash join to be pushed down as an InList expression for dynamic filtering. Build sides with more rows than this will use hash table lookups instead. Set to 0 to always use hash table lookups. This provides an additional limit beyond `hash_join_inlist_pushdown_max_size` to prevent 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. The default is 150 values per partition. This is inspired by Trino's `max-filter-keys-per-column` setting. See: datafusion.optimizer.hash_join_inlist_pushdown_max_size 131072 Maximum size in bytes for the build side of a hash join to be pushed down as an InList expression for dynamic filtering. Build sides larger than this will use hash table lookups instead. Set to 0 to always use hash table lookups. InList pushdown can be more efficient for small build sides because it can result in better statistics pruning as well as use any bloom filters present on the scan side. InList expressions are also more transparent and easier to serialize over the network in distributed uses of DataFusion. On the other hand InList pushdown requires making a copy of the data and thus adds some overhead to the build side and uses more memory. This setting is per-partition, so we may end up using `hash_join_inlist_pushdown_max_size` * `target_partitions` memory. The default is 128kB per partition. This should allow point lookup joins (e.g. joining on a unique primary key) to use InList pushdown in most cases but avoids excessive memory usage or overhead for larger joins. datafusion.optimizer.hash_join_single_partition_threshold 4194304 The maximum estimated size in bytes for one input side of a HashJoin will be collected into a single partition diff --git a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt index 6774d4f3a01db..449fc22acf0d6 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt @@ -476,7 +476,7 @@ Plan with Metrics 02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, b@0)], metrics=[output_rows=2, output_batches=1, array_map_created_count=0, build_input_batches=1, build_input_rows=2, input_batches=1, input_rows=2, avg_fanout=100% (2/2), probe_hit_rate=100% (2/2)] 03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t1.parquet]]}, projection=[a, x], file_type=parquet, metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, predicate_cache_inner_records=0, predicate_cache_records=0, scan_efficiency_ratio=17.37% (132/760)] 04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t2.parquet]]}, projection=[b, c, y], file_type=parquet, predicate=DynamicFilter [ b@0 >= aa AND b@0 <= ab AND b@0 IN (SET) ([aa, ab]) ], dynamic_rg_pruning=eligible, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 >= aa AND b_null_count@1 != row_count@2 AND b_min@3 <= ab AND (b_null_count@1 != row_count@2 AND b_min@3 <= aa AND aa <= b_max@0 OR b_null_count@1 != row_count@2 AND b_min@3 <= ab AND ab <= b_max@0), required_guarantees=[b in (aa, ab)], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=1 total → 1 matched, page_index_rows_pruned=5 total → 5 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=2, pushdown_rows_pruned=3, predicate_cache_inner_records=5, predicate_cache_records=2, scan_efficiency_ratio=22.46% (234/1.04 K)] -05)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t3.parquet]]}, projection=[d, z], file_type=parquet, predicate=DynamicFilter [ d@0 >= ca AND d@0 <= cb AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=d_null_count@1 != row_count@2 AND d_max@0 >= ca AND d_null_count@1 != row_count@2 AND d_min@3 <= cb, required_guarantees=[], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=1 total → 1 matched, page_index_rows_pruned=8 total → 8 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=2, pushdown_rows_pruned=6, predicate_cache_inner_records=8, predicate_cache_records=2, scan_efficiency_ratio=21.45% (172/802)] +05)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/nested_t3.parquet]]}, projection=[d, z], file_type=parquet, predicate=DynamicFilter [ d@0 >= ca AND d@0 <= cb AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=d_null_count@1 != row_count@2 AND d_max@0 >= ca AND d_null_count@1 != row_count@2 AND d_min@3 <= cb AND d_null_count@1 != row_count@2 AND IN_SET_INTERSECTS(d_min@3, d_max@0, 2 values), required_guarantees=[], metrics=[output_rows=2, output_batches=1, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=1 total → 1 matched, page_index_rows_pruned=8 total → 8 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=2, pushdown_rows_pruned=6, predicate_cache_inner_records=8, predicate_cache_records=2, scan_efficiency_ratio=21.45% (172/802)] statement ok reset datafusion.explain.analyze_categories; diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 0085d4ac7c1fa..7433044c03e8a 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -178,6 +178,8 @@ The following configuration settings are available: | datafusion.optimizer.hash_join_single_partition_threshold_rows | 131072 | The maximum estimated size in rows for one input side of a HashJoin will be collected into a single partition | | datafusion.optimizer.hash_join_inlist_pushdown_max_size | 131072 | Maximum size in bytes for the build side of a hash join to be pushed down as an InList expression for dynamic filtering. Build sides larger than this will use hash table lookups instead. Set to 0 to always use hash table lookups. InList pushdown can be more efficient for small build sides because it can result in better statistics pruning as well as use any bloom filters present on the scan side. InList expressions are also more transparent and easier to serialize over the network in distributed uses of DataFusion. On the other hand InList pushdown requires making a copy of the data and thus adds some overhead to the build side and uses more memory. This setting is per-partition, so we may end up using `hash_join_inlist_pushdown_max_size` \* `target_partitions` memory. The default is 128kB per partition. This should allow point lookup joins (e.g. joining on a unique primary key) to use InList pushdown in most cases but avoids excessive memory usage or overhead for larger joins. | | datafusion.optimizer.hash_join_inlist_pushdown_max_distinct_values | 150 | Maximum number of distinct values (rows) in the build side of a hash join to be pushed down as an InList expression for dynamic filtering. Build sides with more rows than this will use hash table lookups instead. Set to 0 to always use hash table lookups. This provides an additional limit beyond `hash_join_inlist_pushdown_max_size` to prevent 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. The default is 150 values per partition. This is inspired by Trino's `max-filter-keys-per-column` setting. See: | +| datafusion.optimizer.hash_join_dynamic_pruning_max_distinct_values | 100000 | 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(_min, _max, values)` clause - the visible sign this ran, distinct from the plain min/max bounds every join pushes down. | +| datafusion.optimizer.hash_join_dynamic_pruning_max_size | 8388608 | 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. | | datafusion.optimizer.default_filter_selectivity | 20 | 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.optimizer.prefer_existing_union | false | When set to true, the optimizer will not attempt to convert Union to Interleave | | datafusion.optimizer.expand_views_at_output | false | When set to true, if the returned type is a view type then the output will be coerced to a non-view. Coerces `Utf8View` to `LargeUtf8`, and `BinaryView` to `LargeBinary`. |