diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index e36fa80f53126..bfc12eb2eb566 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/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 7b9e701119ef4..fecd383eccd24 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -3080,19 +3080,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/partitioned_hash_eval.rs b/datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs index 98e8b2d2fc42e..28d1c2efe3e2e 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 @@ -17,10 +17,10 @@ //! Hash computation and hash table lookup expressions for dynamic filtering -use std::{fmt::Display, hash::Hash, sync::Arc}; +use std::{fmt::Display, hash::Hash, sync::Arc, sync::OnceLock}; use arrow::{ - array::{ArrayRef, UInt64Array}, + array::{Array, ArrayRef, UInt64Array}, datatypes::{DataType, Schema}, record_batch::RecordBatch, }; @@ -32,6 +32,7 @@ use datafusion_common::internal_err; use datafusion_expr::ColumnarValue; use datafusion_expr_common::dyn_eq::DynHash; use datafusion_physical_expr_common::physical_expr::{PhysicalExpr, PhysicalExprRef}; +use parking_lot::Mutex; use crate::joins::Map; @@ -271,6 +272,46 @@ impl HashExpr { } } +/// A build side's pruning domain: the raw values until pruning first needs them, +/// then the expression built from them. `PruningPredicate` is rebuilt for file, +/// row-group and page-index pruning of every file in a scan, so the domain is built +/// once here and rebound to each container's statistics instead. +struct PruningDomain { + /// `None` once taken to build `expr`, or if no values were retained at all. + raw: Mutex>, + /// The expression and the type it was built for; `None` if that failed, so the + /// attempt is not repeated. + expr: OnceLock>, +} + +impl PruningDomain { + fn new(raw: Option) -> Self { + Self { + raw: Mutex::new(raw), + expr: OnceLock::new(), + } + } + + /// The pruning expression for `data_type`, which `build` constructs from the raw + /// build-side values on the first call, releasing the array right after. `None` + /// if there are no usable values, or if an earlier caller built the domain for a + /// different type - a file whose column type has evolved goes unpruned. + fn get_or_build( + &self, + data_type: &DataType, + build: impl FnOnce(&dyn Array) -> Option, + ) -> Option { + let (built_for, expr) = self + .expr + .get_or_init(|| { + let raw = self.raw.lock().take()?; + Some((data_type.clone(), build(raw.as_ref())?)) + }) + .as_ref()?; + (built_for == data_type).then(|| Arc::clone(expr)) + } +} + /// 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 +325,8 @@ pub struct HashTableLookupExpr { map: Arc, /// Description for display description: String, + /// Pruning-only build-side values, shared with every derived expression. + pruning_domain: Arc, } impl HashTableLookupExpr { /// Create a new HashTableLookupExpr @@ -293,6 +336,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 +345,27 @@ impl HashTableLookupExpr { random_state: SeededRandomState, map: Arc, description: String, + raw_pruning_values: Option, ) -> Self { Self { on_columns, random_state, map, description, + pruning_domain: Arc::new(PruningDomain::new(raw_pruning_values)), } } + + /// The pruning expression for this build side, which `build` constructs from + /// its raw values on the first call and every later call reuses. `None` if no + /// values were kept, or if one was already built for a different `data_type`. + pub fn cached_pruning_expr( + &self, + data_type: &DataType, + build: impl FnOnce(&dyn Array) -> Option, + ) -> Option { + self.pruning_domain.get_or_build(data_type, build) + } } impl std::fmt::Debug for HashTableLookupExpr { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { @@ -372,12 +429,13 @@ impl PhysicalExpr for HashTableLookupExpr { self: Arc, children: Vec>, ) -> Result> { - Ok(Arc::new(HashTableLookupExpr::new( - children, - self.random_state.clone(), - Arc::clone(&self.map), - self.description.clone(), - ))) + Ok(Arc::new(HashTableLookupExpr { + on_columns: children, + random_state: self.random_state.clone(), + map: Arc::clone(&self.map), + description: self.description.clone(), + pruning_domain: Arc::clone(&self.pruning_domain), + })) } fn data_type(&self, _input_schema: &Schema) -> Result { @@ -423,6 +481,7 @@ impl PhysicalExpr for HashTableLookupExpr { random_state: _, map: _, description: _, + pruning_domain: _, } = self; // HashTableLookupExpr holds a runtime Arc (the build-side hash @@ -468,7 +527,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 arrow::array::AsArray; + use arrow::datatypes::Int32Type; use datafusion_physical_expr::expressions::Column; use std::collections::hash_map::DefaultHasher; use std::hash::Hasher; @@ -479,6 +540,89 @@ 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))) + } + + fn build_domain( + expr: &HashTableLookupExpr, + seen: &std::cell::RefCell>>>, + ) -> Option { + expr.cached_pruning_expr(&DataType::Int32, |array| { + seen.borrow_mut() + .push(array.as_primitive::().iter().collect()); + Some(Arc::new( + datafusion_physical_expr::expressions::Literal::new( + datafusion_common::ScalarValue::Boolean(Some(true)), + ), + )) + }) + } + + fn lookup_with(raw_values: Option>) -> HashTableLookupExpr { + HashTableLookupExpr::new( + vec![Arc::new(Column::new("a", 0))], + SeededRandomState::with_seed(1), + hash_map_with_distinct_count(&[100, 200, 300]), + "hash_lookup".to_string(), + raw_values.map(|v| Arc::new(arrow::array::Int32Array::from(v)) as ArrayRef), + ) + } + + #[test] + fn test_cached_pruning_domain() { + let expr = lookup_with(Some(vec![3, 1, 3])); + let seen = std::cell::RefCell::new(Vec::new()); + let built = build_domain(&expr, &seen).expect("domain built"); + + assert_eq!(seen.borrow().as_slice(), [vec![Some(3), Some(1), Some(3)]]); + + // Second call is served from the cache: same expression, builder not re-run. + let again = build_domain(&expr, &seen).expect("domain cached"); + assert!(Arc::ptr_eq(&built, &again)); + assert_eq!(seen.borrow().len(), 1); + + // Assert domain absent for other data types + assert!( + expr.cached_pruning_expr(&DataType::Int64, |_| unreachable!()) + .is_none() + ); + } + + #[test] + fn test_cached_pruning_domain_absent_when_not_populated() { + let seen = std::cell::RefCell::new(Vec::new()); + assert!(build_domain(&lookup_with(None), &seen).is_none()); + assert!(seen.borrow().is_empty()); + } + + #[test] + fn test_cached_pruning_domain_shared_with_derived_children() { + let expr = lookup_with(Some(vec![1, 2, 3])); + let seen = std::cell::RefCell::new(Vec::new()); + let built = build_domain(&expr, &seen).expect("domain built"); + + let derived = Arc::new(expr) + .with_new_children(vec![Arc::new(Column::new("a", 7))]) + .unwrap(); + let derived = derived.downcast_ref::().unwrap(); + + assert_eq!(derived.children()[0].to_string(), "a@7"); + assert!(Arc::ptr_eq( + &built, + &build_domain(derived, &seen).expect("domain shared") + )); + assert_eq!(seen.borrow().len(), 1); + } + #[test] fn test_hash_expr_eq_same() { let col_a: PhysicalExprRef = Arc::new(Column::new("a", 0)); @@ -757,6 +901,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -764,6 +909,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); assert_eq!(expr1, expr2); @@ -782,6 +928,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -789,6 +936,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); assert_ne!(expr1, expr2); @@ -805,6 +953,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup_one".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -812,6 +961,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup_two".to_string(), + None, ); assert_ne!(expr1, expr2); @@ -831,6 +981,7 @@ mod tests { SeededRandomState::with_seed(1), hash_map1, "lookup".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -838,6 +989,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 +1007,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -862,6 +1015,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/primitive_in_list.rs b/datafusion/pruning/src/primitive_in_list.rs index 6b21a437862a6..78fe05ddac617 100644 --- a/datafusion/pruning/src/primitive_in_list.rs +++ b/datafusion/pruning/src/primitive_in_list.rs @@ -94,6 +94,27 @@ macro_rules! define_primitive_values { } } + /// Appends `array`'s non-null values, which must share this domain's + /// native representation (see [`shares_native_domain`]). + pub(crate) fn extend_from_array(&mut self, array: &dyn Array) -> Option<()> { + if !shares_native_domain(&self.data_type, array.data_type()) { + return None; + } + match &mut self.values { + $( + PrimitiveValues::$variant(values) => { + let array = array.as_primitive_opt::<$arrow_type>()?; + if array.null_count() == 0 { + values.extend_from_slice(array.values()); + } else { + values.extend(array.iter().flatten()); + } + } + )+ + } + Some(()) + } + pub(crate) fn is_empty(&self) -> bool { match &self.values { $(PrimitiveValues::$variant(values) => values.is_empty(),)+ @@ -195,6 +216,34 @@ define_primitive_values! { ScalarValue::DurationNanosecond(Some(value)), *value; } +/// Can values of `domain` and `array` be compared as the same native type? +/// +/// A timestamp's timezone and a decimal's declared precision are metadata that do +/// not change what the stored integer means; anything else has to match exactly, +/// since the domain is searched against statistics cast to `domain`. +fn shares_native_domain(domain: &DataType, array: &DataType) -> bool { + match (domain, array) { + (DataType::Timestamp(left, _), DataType::Timestamp(right, _)) => left == right, + (DataType::Decimal32(_, left), DataType::Decimal32(_, right)) + | (DataType::Decimal64(_, left), DataType::Decimal64(_, right)) + | (DataType::Decimal128(_, left), DataType::Decimal128(_, right)) + | (DataType::Decimal256(_, left), DataType::Decimal256(_, right)) => { + left == right + } + (domain, array) => domain == array, + } +} + +impl PrimitiveInListDomain { + /// Collects `array`'s non-null values into a domain over `data_type`. `None` if + /// `data_type` is not a supported primitive, or the array holds something else. + pub(crate) fn from_array(data_type: &DataType, array: &dyn Array) -> Option { + let mut domain = Self::new(data_type, array.len() - array.null_count())?; + domain.extend_from_array(array)?; + Some(domain) + } +} + /// Tests an inclusive statistics interval against a sorted primitive domain. struct PrimitiveInListPruningExpr where diff --git a/datafusion/pruning/src/pruning_predicate.rs b/datafusion/pruning/src/pruning_predicate.rs index b49b72058e0cd..666be88c54f95 100644 --- a/datafusion/pruning/src/pruning_predicate.rs +++ b/datafusion/pruning/src/pruning_predicate.rs @@ -28,7 +28,7 @@ use crate::string_in_list::{BinaryInListPruningExpr, StringInListPruningExpr}; use arrow::array::AsArray; use arrow::{ - array::{ArrayRef, BooleanArray, new_null_array}, + array::{Array, ArrayRef, BooleanArray, new_null_array}, datatypes::{DataType, Field, Schema, SchemaRef}, record_batch::{RecordBatch, RecordBatchOptions}, }; @@ -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,186 @@ 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, so this skips the negation/all-null-list handling +/// `build_compact_in_list_expr` needs. +/// +/// The domain itself is built once per build side and cached in the lookup; every +/// later `PruningPredicate` only rebinds it to that container's statistics columns. +/// +/// 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> { + // Composite (multi-column) keys have no IN-list equivalent. + let on_columns = lookup.children(); + let [column_expr] = on_columns[..] else { + 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, + }; + + // 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 rewritten = (|| { + 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)?; + // The cached expression holds the sorted domain and takes its min/max as + // children, so rebinding it to this container's statistics is a clone. + let may_match = lookup + .cached_pruning_expr(data_type, |array| { + build_in_list_domain_expr( + data_type, + array, + Arc::clone(&min), + Arc::clone(&max), + ) + })? + .with_new_children(vec![min, max]) + .ok()?; + Some(Arc::new(phys_expr::BinaryExpr::new( + non_null, + Operator::And, + may_match, + )) as Arc) + })(); + if rewritten.is_none() { + required_columns.columns.truncate(required_columns_len); + } + rewritten +} + +/// Collects `array`'s non-null values into an `IN` domain over `data_type` and +/// wraps it in the matching pruning expression, which sorts and deduplicates them. +/// `None` for a type the domain cannot hold. +/// +/// A dictionary-encoded build side contributes its dictionary values, including any +/// no longer referenced by a key. Extra values only make the domain match more +/// containers, never fewer, so this stays conservative. +fn build_in_list_domain_expr( + data_type: &DataType, + array: &dyn Array, + min: PhysicalExprRef, + max: PhysicalExprRef, +) -> Option { + let values = match array.data_type() { + DataType::Dictionary(_, _) => array.as_any_dictionary().values(), + _ => array, + }; + Some(if data_type.is_string() { + let values = string_values(values)?; + if values.is_empty() { + return None; + } + Arc::new(StringInListPruningExpr::new( + SetMembership::In, + min, + max, + values, + )) + } else if matches!( + data_type, + DataType::Binary | DataType::LargeBinary | DataType::BinaryView + ) { + let values = binary_values(values)?; + if values.is_empty() { + return None; + } + Arc::new(BinaryInListPruningExpr::new( + SetMembership::In, + min, + max, + values, + )) + } else { + let domain = PrimitiveInListDomain::from_array(data_type, values)?; + if domain.is_empty() { + return None; + } + domain.into_expr(SetMembership::In, min, max) + }) +} + +/// `array`'s non-null values, for any of the string layouts. The domain compares +/// bytes, so the layout the build side happens to use does not have to match the +/// column's. +fn string_values(array: &dyn Array) -> Option> { + let to_owned = |value: &str| value.to_owned(); + Some(match array.data_type() { + DataType::Utf8 => array + .as_string::() + .iter() + .flatten() + .map(to_owned) + .collect(), + DataType::LargeUtf8 => array + .as_string::() + .iter() + .flatten() + .map(to_owned) + .collect(), + DataType::Utf8View => array + .as_string_view() + .iter() + .flatten() + .map(to_owned) + .collect(), + _ => return None, + }) +} + +/// `array`'s non-null values, for any of the binary layouts. See [`string_values`]. +fn binary_values(array: &dyn Array) -> Option>> { + let to_owned = |value: &[u8]| Box::<[u8]>::from(value); + Some(match array.data_type() { + DataType::Binary => array + .as_binary::() + .iter() + .flatten() + .map(to_owned) + .collect(), + DataType::LargeBinary => array + .as_binary::() + .iter() + .flatten() + .map(to_owned) + .collect(), + DataType::BinaryView => array + .as_binary_view() + .iter() + .flatten() + .map(to_owned) + .collect(), + _ => return None, + }) +} + /// 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 +1997,41 @@ 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)); + } + // A partitioned hash join hides its per-partition filters under a `CASE` on the + // repartition hash. A row takes exactly one branch, so a container may match + // only if some branch may: the branches' disjunction is a sound relaxation, and + // the `WHEN`s (a hash, which no statistics describe) can be dropped. + if let Some(case) = expr.downcast_ref::() { + // Only a Boolean `CASE` is a predicate; anything else is a value for + // whatever compares it to handle. + if !matches!(case.data_type(schema), Ok(DataType::Boolean)) { + return unhandled_hook.handle(expr); + } + // A missing `ELSE` yields NULL, which never matches, so it adds nothing. + return case + .when_then_expr() + .iter() + .map(|(_, then)| then) + .chain(case.else_expr()) + .map(|branch| { + build_predicate_expression( + branch, + schema, + required_columns, + unhandled_hook, + max_in_list_size, + properties, + ) + }) + .reduce(|acc, branch| { + Arc::new(phys_expr::BinaryExpr::new(acc, Operator::Or, branch)) as _ + }) + .unwrap_or_else(|| unhandled_hook.handle(expr)); + } let (left, op, right) = { if let Some(bin_expr) = expr.downcast_ref::() { @@ -2407,6 +2623,10 @@ mod tests { self as phys_expr, DynamicFilterPhysicalExpr, }; use datafusion_physical_expr::planner::logical2physical; + use datafusion_physical_plan::joins::join_hash_map::{ + JoinHashMapType, JoinHashMapU32, + }; + use datafusion_physical_plan::joins::{Map, SeededRandomState}; use itertools::Itertools; #[derive(Debug, Default)] @@ -7340,4 +7560,121 @@ 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); } + + /// A hash map holding `distinct_count` entries - only the count matters for + /// `num_of_distinct_key()`, which decides whether the build side needs dedup. + fn hash_map_with_distinct_count(distinct_count: usize) -> Arc { + let mut hash_map = JoinHashMapU32::with_capacity(distinct_count); + let hashes: Vec = + (0..distinct_count as u64).map(|i| 100 * (i + 1)).collect(); + JoinHashMapType::update_from_iter( + &mut hash_map, + Box::new(hashes.iter().enumerate()), + 0, + ); + Arc::new(Map::HashMap(Box::new(hash_map))) + } + + #[test] + fn test_hash_lookup_pruning_via_min_max() { + let map = hash_map_with_distinct_count(3); + + let schema = Arc::new(Schema::new(vec![Field::new("b", DataType::Int32, true)])); + let column: Arc = Arc::new(phys_expr::Column::new("b", 0)); + // Duplicates and nulls are the domain's to handle, not the build side's. + let values: ArrayRef = Arc::new(Int32Array::from(vec![ + Some(10), + Some(20), + Some(10), + None, + Some(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]); + } + + #[test] + fn test_partition_routed_hash_lookup_pruning() { + let schema = Arc::new(Schema::new(vec![Field::new("b", DataType::Int32, true)])); + let column: Arc = Arc::new(phys_expr::Column::new("b", 0)); + + // One branch per build partition, each holding its own slice of the build + // side, as `build_partitioned_filter` produces them. + let branch = |values: Vec| -> Arc { + let values: ArrayRef = Arc::new(Int32Array::from(values)); + Arc::new(HashTableLookupExpr::new( + vec![Arc::clone(&column)], + SeededRandomState::with_seed(1), + hash_map_with_distinct_count(3), + "hash_lookup".to_string(), + Some(values), + )) + }; + let routing: Arc = + Arc::new(phys_expr::Literal::new(ScalarValue::UInt64(Some(0)))); + let when = |partition: u64| -> Arc { + Arc::new(phys_expr::Literal::new(ScalarValue::UInt64(Some( + partition, + )))) + }; + let case: Arc = Arc::new( + phys_expr::CaseExpr::try_new( + Some(routing), + vec![ + (when(0), branch(vec![10, 20, 30])), + (when(1), branch(vec![40, 50, 60])), + ], + // Partitions with no build rows reject everything routed to them. + Some(Arc::new(phys_expr::Literal::new(ScalarValue::Boolean( + Some(false), + )))), + ) + .unwrap(), + ); + + let predicate = PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(&schema)) + .try_build(case) + .unwrap(); + + let statistics = TestStatistics::new().with( + "b", + ContainerStats::new_i32( + vec![Some(5), Some(15), Some(45), Some(100)], + vec![Some(8), Some(25), Some(55), Some(200)], + ), + ); + + let result = predicate.prune(&statistics).unwrap(); + // Container 1 ([15,25]) holds 20 from the first branch and container 2 + // ([45,55]) holds 50 from the second, so a branch may match in each. The + // other two intersect neither branch, nor the `ELSE`. + assert_eq!(result, vec![false, true, 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 924883da42afc..280a672c54367 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 8dd763205f6c2..5751ae8282d73 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.72% (132/745)] 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.78% (234/1.03 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.86% (172/787)] +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.86% (172/787)] statement ok reset datafusion.explain.analyze_categories; diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 5922c9e196fdf..4e6348b69db26 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`. |