From 5353ac595ed58b4360d37b5654460910c28bfbf7 Mon Sep 17 00:00:00 2001 From: Marko Grujic Date: Thu, 10 Sep 2026 15:58:33 +0200 Subject: [PATCH 1/2] Prune hash join build side via min/max, not bloom filters Exposes HashTableLookupExpr's single-column build values through the same compact sorted-domain rewrite ordinary large IN-lists already get (PrimitiveInListPruningExpr/StringInListPruningExpr), so row-group/file min/max stats alone can exclude containers - no bloom filter fetch, no LiteralGuarantee. On by default via hash_join_dynamic_pruning_max_distinct_values (default 100_000); set to 0 to disable. Co-Authored-By: Claude Sonnet 5 --- datafusion/common/src/config.rs | 20 +++ datafusion/common/src/scalar/mod.rs | 36 ++++ .../physical-plan/src/joins/hash_join/exec.rs | 36 +++- .../src/joins/hash_join/inlist_builder.rs | 45 ++++- .../joins/hash_join/partitioned_hash_eval.rs | 160 +++++++++++++++++- .../src/joins/hash_join/shared_bounds.rs | 21 ++- datafusion/proto/tests/cases/plans/exprs.rs | 1 + datafusion/pruning/src/pruning_predicate.rs | 156 +++++++++++++++++ .../test_files/dynamic_row_group_pruning.slt | 27 +++ .../test_files/information_schema.slt | 4 + .../test_files/push_down_filter_parquet.slt | 2 +- docs/source/user-guide/configs.md | 2 + 12 files changed, 488 insertions(+), 22 deletions(-) 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/common/src/scalar/mod.rs b/datafusion/common/src/scalar/mod.rs index 37ae21d90d986..82636b20a6790 100644 --- a/datafusion/common/src/scalar/mod.rs +++ b/datafusion/common/src/scalar/mod.rs @@ -4302,6 +4302,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) @@ -6039,6 +6053,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 9858a4c06cd4a..81b56a550961a 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -3094,19 +3094,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 82863c080259f..20cd75b7fd084 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` /// /// # Public Only for Internal Use: /// `datafusion-proto` tests require this constructor, but it is not part of @@ -303,14 +354,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 { @@ -379,6 +445,7 @@ impl PhysicalExpr for HashTableLookupExpr { self.random_state.clone(), Arc::clone(&self.map), self.description.clone(), + self.pruning_scalars.raw.clone(), ))) } @@ -425,6 +492,7 @@ impl PhysicalExpr for HashTableLookupExpr { random_state: _, map: _, description: _, + pruning_scalars: _, } = self; // HashTableLookupExpr holds a runtime Arc (the build-side hash @@ -470,8 +538,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; @@ -481,6 +550,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)); @@ -759,6 +905,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -766,6 +913,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); assert_eq!(expr1, expr2); @@ -784,6 +932,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -791,6 +940,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); assert_ne!(expr1, expr2); @@ -807,6 +957,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup_one".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -814,6 +965,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup_two".to_string(), + None, ); assert_ne!(expr1, expr2); @@ -833,6 +985,7 @@ mod tests { SeededRandomState::with_seed(1), hash_map1, "lookup".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -840,6 +993,7 @@ mod tests { SeededRandomState::with_seed(1), hash_map2, "lookup".to_string(), + None, ); // Different Arc pointers means not equal (uses Arc::ptr_eq) @@ -857,6 +1011,7 @@ mod tests { SeededRandomState::with_seed(1), Arc::clone(&hash_map), "lookup".to_string(), + None, ); let expr2 = HashTableLookupExpr::new( @@ -864,6 +1019,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 ee6f9819a3166..a7f6aa72000d9 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 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`. | From 39f2d8c42e28114c229a633bf895a6240a21aa32 Mon Sep 17 00:00:00 2001 From: Marko Grujic Date: Sun, 20 Sep 2026 08:55:48 +0200 Subject: [PATCH 2/2] Build the pruning domain once per build side, not per file Cache the built domain expression on the lookup and rebind it to each container's statistics, and see through the CASE that routes a partitioned join's per-partition filters. Co-Authored-By: Claude Opus 5 (1M context) --- datafusion/common/src/scalar/mod.rs | 36 -- .../src/joins/hash_join/inlist_builder.rs | 45 +-- .../joins/hash_join/partitioned_hash_eval.rs | 228 +++++++------ datafusion/pruning/src/primitive_in_list.rs | 49 +++ datafusion/pruning/src/pruning_predicate.rs | 307 ++++++++++++++---- 5 files changed, 408 insertions(+), 257 deletions(-) diff --git a/datafusion/common/src/scalar/mod.rs b/datafusion/common/src/scalar/mod.rs index 82636b20a6790..37ae21d90d986 100644 --- a/datafusion/common/src/scalar/mod.rs +++ b/datafusion/common/src/scalar/mod.rs @@ -4302,20 +4302,6 @@ 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) @@ -6053,28 +6039,6 @@ 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/inlist_builder.rs b/datafusion/physical-plan/src/joins/hash_join/inlist_builder.rs index 1f654dd22d05a..2fc3201c6363f 100644 --- a/datafusion/physical-plan/src/joins/hash_join/inlist_builder.rs +++ b/datafusion/physical-plan/src/joins/hash_join/inlist_builder.rs @@ -19,9 +19,8 @@ use std::sync::Arc; -use arrow::array::{ArrayRef, AsArray, StructArray}; -use arrow::compute::cast; -use arrow::datatypes::{Field, FieldRef, Fields, Int32Type}; +use arrow::array::{ArrayRef, StructArray}; +use arrow::datatypes::{Field, FieldRef, Fields}; use arrow_schema::DataType; use datafusion_common::Result; @@ -78,25 +77,6 @@ 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::*; @@ -162,27 +142,6 @@ 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 20cd75b7fd084..731a010204a84 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,7 +17,7 @@ //! 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::{Array, ArrayRef, UInt64Array}, @@ -25,7 +25,6 @@ use arrow::{ 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")] @@ -33,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; @@ -272,49 +272,43 @@ 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>>, +/// 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 LazyPruningScalars { +impl PruningDomain { fn new(raw: Option) -> Self { Self { - raw, - cache: std::sync::OnceLock::new(), + raw: Mutex::new(raw), + expr: 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 + /// 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.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) + let raw = self.raw.lock().take()?; + Some((data_type.clone(), build(raw.as_ref())?)) }) - .as_ref() + .as_ref()?; + (built_for == data_type).then(|| Arc::clone(expr)) } } @@ -331,9 +325,8 @@ 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, + /// Pruning-only build-side values, shared with every derived expression. + pruning_domain: Arc, } impl HashTableLookupExpr { /// Create a new HashTableLookupExpr @@ -361,21 +354,19 @@ impl HashTableLookupExpr { random_state, map, description, - pruning_scalars: LazyPruningScalars::new(raw_pruning_values), + pruning_domain: Arc::new(PruningDomain::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))) + /// 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 { @@ -440,13 +431,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(), - self.pruning_scalars.raw.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 { @@ -492,7 +483,7 @@ impl PhysicalExpr for HashTableLookupExpr { random_state: _, map: _, description: _, - pruning_scalars: _, + pruning_domain: _, } = self; // HashTableLookupExpr holds a runtime Arc (the build-side hash @@ -539,8 +530,9 @@ fn evaluate_columns( mod tests { use super::*; use crate::joins::join_hash_map::{JoinHashMapType, JoinHashMapU32}; + use arrow::array::AsArray; + use arrow::datatypes::Int32Type; use datafusion_physical_expr::expressions::Column; - use rstest::rstest; use std::collections::hash_map::DefaultHasher; use std::hash::Hasher; @@ -562,69 +554,75 @@ mod tests { 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); + 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)), + ), + )) + }) + } - let expr = HashTableLookupExpr::new( - vec![Arc::clone(&col_a)], + fn lookup_with(raw_values: Option>) -> HashTableLookupExpr { + HashTableLookupExpr::new( + vec![Arc::new(Column::new("a", 0))], SeededRandomState::with_seed(1), - hash_map, + hash_map_with_distinct_count(&[100, 200, 300]), "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)) - } - } + raw_values.map(|v| Arc::new(arrow::array::Int32Array::from(v)) as ArrayRef), + ) } #[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), + 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() ); + } - assert!(expr.cached_pruning_scalars().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] 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 21fe514b6a834..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}, }; @@ -1527,8 +1527,11 @@ impl CompactInListDomain { /// [`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. +/// 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 @@ -1541,10 +1544,11 @@ fn build_hash_lookup_pruning_expr( schema: &Schema, required_columns: &mut RequiredColumns, ) -> Option> { - let (column_expr, values) = lookup.cached_pruning_scalars()?; - if values.is_empty() { + // 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() { @@ -1554,73 +1558,147 @@ fn build_hash_lookup_pruning_expr( 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 rewritten = (|| { let min = required_columns - .min_column_expr(column, &column_expr, field) + .min_column_expr(column, column_expr, field) .ok()?; let max = required_columns - .max_column_expr(column, &column_expr, field) + .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)) + 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) })(); - let Some((min, max, non_null)) = statistics else { + if rewritten.is_none() { required_columns.columns.truncate(required_columns_len); - return None; + } + 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, }; - let may_match = match domain { - CompactInListDomain::String(values) => Arc::new(StringInListPruningExpr::new( + 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, - )) as PhysicalExprRef, - CompactInListDomain::Binary(values) => Arc::new(BinaryInListPruningExpr::new( + )) + } 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, - )) as PhysicalExprRef, - CompactInListDomain::Primitive(values) => { - values.into_expr(SetMembership::In, min, max) + )) + } else { + let domain = PrimitiveInListDomain::from_array(data_type, values)?; + if domain.is_empty() { + return None; } - }; - Some(Arc::new(phys_expr::BinaryExpr::new( - non_null, - Operator::And, - may_match, - ))) + 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 @@ -1923,6 +2001,37 @@ fn build_predicate_expression( 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::() { @@ -2514,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)] @@ -7448,25 +7561,34 @@ mod tests { 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]; + /// 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, ); - let map = Arc::new(Map::HashMap(Box::new(hash_map))); + 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)); - let values: ArrayRef = Arc::new(Int32Array::from(vec![10, 20, 30])); + // 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), @@ -7496,4 +7618,63 @@ mod tests { // 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]); + } }