From 2bf19fb48ac3ecefdecb1e859c64194673cdef0f Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Mon, 10 Aug 2026 12:29:59 -0400 Subject: [PATCH 1/3] refactor: let a hash join carry more than one self filter `handle_child_pushdown_result` reached for the single self filter it had pushed and `dynamic_expressions_produced` cloned the single field behind it. Route both through one `HashJoinExecDynamicFilter::produced_expressions` helper and pop the driven filter off the pushed list instead of indexing position zero, so a second pushed filter is a matter of adding a field rather than of rewriting the plumbing. No behaviour change: exactly one self filter is still pushed. Co-Authored-By: Claude Opus 5 --- .../physical-plan/src/joins/hash_join/exec.rs | 58 ++++++++++--------- 1 file changed, 32 insertions(+), 26 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 4e68d871b81a..99984d82813f 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -787,6 +787,13 @@ struct HashJoinExecDynamicFilter { build_accumulator: OnceLock>, } +impl HashJoinExecDynamicFilter { + /// Every dynamic expression this join produces. + fn produced_expressions(&self) -> Vec> { + vec![Arc::::clone(&self.filter) as _] + } +} + impl fmt::Debug for HashJoinExec { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("HashJoinExec") @@ -1338,20 +1345,17 @@ impl ExecutionPlan for HashJoinExec { .filter .iter() .map(|filter| Arc::clone(filter.expression())); - let dynamic_filter = self.dynamic_filter.iter().map(|dynamic_filter| { - Arc::::clone(&dynamic_filter.filter) - as Arc - }); + let dynamic_filter = self + .dynamic_filter + .iter() + .flat_map(HashJoinExecDynamicFilter::produced_expressions); crate::apply_expression_roots(join_keys.chain(filter).chain(dynamic_filter), f) } fn dynamic_expressions_produced(&self) -> Vec> { self.dynamic_filter .iter() - .map(|dynamic_filter| { - Arc::::clone(&dynamic_filter.filter) - as Arc - }) + .flat_map(HashJoinExecDynamicFilter::produced_expressions) .collect() } @@ -1758,24 +1762,26 @@ impl ExecutionPlan for HashJoinExec { let mut result = FilterPushdownPropagation::if_any(child_pushdown_result.clone()); assert_eq!(child_pushdown_result.self_filters.len(), 2); // Should always be 2, we have 2 children let right_child_self_filters = &child_pushdown_result.self_filters[1]; // We only push down filters to the right child - // We expect 0 or 1 self filters - if let Some(filter) = right_child_self_filters.first() { - // Note that we don't check PushdDownPredicate::discrimnant because even if nothing said - // "yes, I can fully evaluate this filter" things might still use it for statistics -> it's worth updating - let predicate = Arc::clone(&filter.predicate); - if let Ok(dynamic_filter) = - Arc::downcast::(predicate) - { - // We successfully pushed down our self filter - we need to make a new node with the dynamic filter - let new_node = self - .builder() - .with_dynamic_filter(Some(HashJoinExecDynamicFilter { - filter: dynamic_filter, - build_accumulator: OnceLock::new(), - })) - .build_exec()?; - result = result.with_updated_node(new_node); - } + // Note that we don't check PushdDownPredicate::discrimnant because even if nothing said + // "yes, I can fully evaluate this filter" things might still use it for statistics -> it's worth updating + let mut pushed = right_child_self_filters + .iter() + .filter_map(|filter| { + let predicate = Arc::clone(&filter.predicate); + Arc::downcast::(predicate).ok() + }) + .collect::>(); + // The filter this join drives is always the last one pushed. + if let Some(filter) = pushed.pop() { + // We successfully pushed down our self filter - we need to make a new node with the dynamic filter + let new_node = self + .builder() + .with_dynamic_filter(Some(HashJoinExecDynamicFilter { + filter, + build_accumulator: OnceLock::new(), + })) + .build_exec()?; + result = result.with_updated_node(new_node); } Ok(result) } From 2020202df947504ba297167ac50a2aedded27e7e Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Mon, 10 Aug 2026 12:32:42 -0400 Subject: [PATCH 2/3] Split the partitioned hash join dynamic filter into bounds AND membership MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A partitioned join used to push one `DynamicFilter[CASE hash % n WHEN i THEN (bounds_i AND membership_i) ... END]`. Everything the join knows was sealed inside a single opaque conjunct: `split_conjunction` splits only on a top-level `BinaryExpr(And)` and does not descend into a `DynamicFilterPhysicalExpr`, so the Parquet reader saw one `ArrowPredicate` that had to compute the routing hash for every probe row before it could reject any of them, and row-group pruning could make nothing of a `CASE`. Push two filters instead: DynamicFilter[ merged bounds ] AND DynamicFilter[ routing CASE ] Two *wrappers* is the load-bearing part, not two expressions inside one wrapper. Split in two, `row_filter.rs` turns each into its own `ArrowPredicate` and arrow-rs applies them in sequence against an accumulating `RowSelection`, building a fresh array reader per predicate over only the still-selected rows. The cheap vectorized range check therefore runs first and the routing hash and hash-table lookup only see the survivors. The range half is also plain enough for `PruningPredicate` to use against row-group statistics, which a `CASE` never was. The merged bounds are the union of the per-partition ranges (see `bounds_union`), which is a relaxation: it admits keys that route to a partition that does not hold them. That is sound because the membership half behind it is exact. The bounds stay inside the `CASE` — exactly as before — whenever the union cannot describe the build side (a canceled partition), the merge is degenerate, or the second filter did not survive pushdown, so no plan loses selectivity it used to have. `CollectLeft` joins have no routing to hoist out of and are left untouched. Co-Authored-By: Claude Opus 5 --- .../src/joins/hash_join/bounds_union.rs | 447 ++++++++++++++++++ .../physical-plan/src/joins/hash_join/exec.rs | 64 ++- .../physical-plan/src/joins/hash_join/mod.rs | 1 + .../src/joins/hash_join/shared_bounds.rs | 363 +++++++++++--- .../test_files/preserve_file_partitioning.slt | 2 +- .../test_files/push_down_filter_parquet.slt | 2 +- .../test_files/statistics_registry.slt | 6 +- 7 files changed, 808 insertions(+), 77 deletions(-) create mode 100644 datafusion/physical-plan/src/joins/hash_join/bounds_union.rs diff --git a/datafusion/physical-plan/src/joins/hash_join/bounds_union.rs b/datafusion/physical-plan/src/joins/hash_join/bounds_union.rs new file mode 100644 index 000000000000..80fca5ccda6d --- /dev/null +++ b/datafusion/physical-plan/src/joins/hash_join/bounds_union.rs @@ -0,0 +1,447 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Merging the per-partition build-side ranges of a partitioned hash join into +//! a single, routing-free conjunct. +//! +//! A partitioned hash join knows `[min, max]` per join key column *per build +//! partition*. Evaluating those bounds exactly requires knowing which partition +//! a probe row routes to, which is why they live inside the routing `CASE`. +//! +//! This module computes the set-theoretic **union** of those ranges instead. +//! The union does not need routing, so it can be pushed as its own top-level +//! conjunct where [`split_conjunction`] can see it — which is what lets the +//! Parquet reader prune row groups with it and evaluate it as a separate, +//! cheap `ArrowPredicate` before the expensive membership check. +//! +//! [`split_conjunction`]: datafusion_physical_expr::split_conjunction +//! +//! # This is a relaxation +//! +//! The union is a *superset* of what the `CASE` accepts: a probe key that falls +//! inside partition 1's range but routes to partition 0 passes the union and is +//! rejected by the `CASE`. That is sound — the membership half behind it is +//! exact — but slightly less selective. [`MergedBounds::relaxation`] estimates +//! how much, assuming keys are uniformly scattered across partitions. +//! +//! # Multi-column keys +//! +//! With more than one join key the per-partition bounds describe a *box*, and +//! the union of boxes is not a box. This module merges **each column +//! independently** and emits the product of the merged per-column ranges, which +//! is a superset of the true union (it also admits the "corners" that no single +//! partition covers). That is still sound, and the corner loss is reflected in +//! the reported relaxation. + +use std::sync::Arc; + +use super::shared_bounds::PartitionBounds; + +use arrow::datatypes::DataType; +use datafusion_common::ScalarValue; +use datafusion_expr::Operator; +use datafusion_physical_expr::expressions::{BinaryExpr, lit}; +use datafusion_physical_expr::{PhysicalExpr, PhysicalExprRef}; + +/// Upper bound on how many `OR`'d ranges a single join key column may +/// contribute. Past this the ranges are collapsed to their convex hull: more +/// terms cost evaluation time on every probe batch, and a long `OR` chain stops +/// being useful for row-group pruning long before it stops being correct. +const MAX_RANGES_PER_COLUMN: usize = 8; + +/// A closed `[min, max]` range over one join key column. +type Range = (ScalarValue, ScalarValue); + +/// The union of every build partition's ranges, merged per column. +#[derive(Debug, Default, PartialEq)] +pub(super) struct MergedBounds { + /// One entry per join key column, positionally matching the join's + /// `on_right` expressions. An empty entry means the column has no usable + /// bounds (some partition did not report any, or the values are `NULL`) and + /// therefore contributes no term — constraining it would drop valid rows. + per_column: Vec>, + /// Expected fraction of probe keys that the merged bounds admit but the + /// per-partition `CASE` would have rejected, assuming keys scatter + /// uniformly over partitions. + /// + /// `None` when the key types have no numeric measure (e.g. strings), so no + /// estimate can be formed. + relaxation: Option, +} + +impl MergedBounds { + /// Expected fraction of probe keys wrongly admitted by the merged bounds. + /// See [`MergedBounds::relaxation`](#structfield.relaxation). + pub(super) fn relaxation(&self) -> Option { + self.relaxation + } + + /// True when the merge produced nothing to filter on, so the conjunct would + /// be a pure cost with no pruning power. + pub(super) fn is_degenerate(&self) -> bool { + self.per_column.iter().all(|ranges| ranges.is_empty()) + } +} + +/// Merges the per-partition ranges of every reported, non-empty build partition +/// into the minimal set of disjoint ranges covering their union. +/// +/// `partitions` must contain **every** partition that can accept a probe row. +/// Omitting one (e.g. a partition whose content is unknown because it was +/// canceled) would produce bounds that reject rows that partition could match. +pub(super) fn merge_partition_bounds( + num_columns: usize, + partitions: &[&PartitionBounds], +) -> MergedBounds { + if partitions.is_empty() { + return MergedBounds::default(); + } + + let mut per_column = Vec::with_capacity(num_columns); + for column in 0..num_columns { + per_column.push(merge_column(column, partitions)); + } + + let relaxation = estimate_relaxation(&per_column, partitions); + MergedBounds { + per_column, + relaxation, + } +} + +/// Merges one column's ranges across partitions, or returns an empty set when +/// any partition failed to report a usable range for it. +fn merge_column(column: usize, partitions: &[&PartitionBounds]) -> Vec { + let mut ranges = Vec::with_capacity(partitions.len()); + for bounds in partitions { + // A partition with no usable range for this column could match any + // value, so the union over this column is unbounded and the column + // must not be constrained at all. + let Some(column_bounds) = bounds.get_column_bounds(column) else { + return Vec::new(); + }; + if column_bounds.min.is_null() || column_bounds.max.is_null() { + return Vec::new(); + } + ranges.push((column_bounds.min.clone(), column_bounds.max.clone())); + } + + // `ScalarValue`'s ordering is the same one the emitted `>=` / `<=` + // comparisons use, so sorting by lower bound and sweeping merges exactly + // the ranges that overlap. Non-comparable values (a type mismatch between + // partitions, which would be a construction bug) sort as equal and are + // conservatively merged into their neighbour, which only ever widens. + ranges + .sort_by(|(a, _), (b, _)| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)); + + let mut merged: Vec = Vec::with_capacity(ranges.len()); + for (min, max) in ranges { + match merged.last_mut() { + // Overlapping (or touching): extend the open range in place. + Some((_, current_max)) if min <= *current_max => { + if max > *current_max { + *current_max = max; + } + } + _ => merged.push((min, max)), + } + } + + if merged.len() > MAX_RANGES_PER_COLUMN { + let min = merged.first().expect("non-empty").0.clone(); + let max = merged + .iter() + .map(|(_, max)| max) + .max_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)) + .expect("non-empty") + .clone(); + merged = vec![(min, max)]; + } + + merged +} + +/// Estimates the fraction of probe keys the merged bounds admit but the +/// per-partition `CASE` rejects. +/// +/// A probe key routes to one partition uniformly at random, so the probability +/// it is *correctly* admitted is the average, over partitions, of the share of +/// the merged region that partition covers. With independent per-column merging +/// that share is the product of the per-column width ratios. +fn estimate_relaxation( + per_column: &[Vec], + partitions: &[&PartitionBounds], +) -> Option { + let mut union_widths = Vec::with_capacity(per_column.len()); + for (column, ranges) in per_column.iter().enumerate() { + if ranges.is_empty() { + continue; + } + let mut total = 0.0; + for (min, max) in ranges { + total += width(min, max)?; + } + union_widths.push((column, total)); + } + if union_widths.is_empty() { + return None; + } + + let mut covered = 0.0; + for bounds in partitions { + let mut share = 1.0; + for (column, union_width) in &union_widths { + // A zero-width union means every partition pins the column to the + // same single value, so this column separates nothing. + if *union_width == 0.0 { + continue; + } + let column_bounds = bounds.get_column_bounds(*column)?; + share *= width(&column_bounds.min, &column_bounds.max)? / union_width; + } + covered += share; + } + + Some(1.0 - covered / partitions.len() as f64) +} + +/// Numeric width of a range, or `None` for types with no numeric measure. +fn width(min: &ScalarValue, max: &ScalarValue) -> Option { + let to_f64 = |value: &ScalarValue| match value.cast_to(&DataType::Float64) { + Ok(ScalarValue::Float64(Some(value))) if value.is_finite() => Some(value), + _ => None, + }; + Some(to_f64(max)? - to_f64(min)?) +} + +/// Builds `(col >= min AND col <= max) OR …` per column, `AND`'d across columns. +/// +/// Returns `None` when the merge is degenerate, so the caller can emit a +/// constant instead of paying to evaluate a predicate that prunes nothing. +pub(super) fn create_merged_bounds_predicate( + on_right: &[PhysicalExprRef], + merged: &MergedBounds, +) -> Option> { + let mut column_predicates: Vec> = Vec::new(); + + for (right_expr, ranges) in on_right.iter().zip(merged.per_column.iter()) { + let Some(column_predicate) = ranges + .iter() + .map(|(min, max)| range_predicate(right_expr, min, max)) + .reduce(|acc, range| { + Arc::new(BinaryExpr::new(acc, Operator::Or, range)) + as Arc + }) + else { + continue; + }; + column_predicates.push(column_predicate); + } + + column_predicates.into_iter().reduce(|acc, predicate| { + Arc::new(BinaryExpr::new(acc, Operator::And, predicate)) as Arc + }) +} + +fn range_predicate( + right_expr: &PhysicalExprRef, + min: &ScalarValue, + max: &ScalarValue, +) -> Arc { + let min_expr = Arc::new(BinaryExpr::new( + Arc::clone(right_expr), + Operator::GtEq, + lit(min.clone()), + )) as Arc; + let max_expr = Arc::new(BinaryExpr::new( + Arc::clone(right_expr), + Operator::LtEq, + lit(max.clone()), + )) as Arc; + Arc::new(BinaryExpr::new(min_expr, Operator::And, max_expr)) as Arc +} + +#[cfg(test)] +mod tests { + use super::*; + + use crate::joins::hash_join::shared_bounds::ColumnBounds; + + use datafusion_physical_expr::expressions::Column; + + fn partition(ranges: &[(i32, i32)]) -> PartitionBounds { + PartitionBounds::new( + ranges + .iter() + .map(|(min, max)| { + ColumnBounds::new( + ScalarValue::Int32(Some(*min)), + ScalarValue::Int32(Some(*max)), + ) + }) + .collect(), + ) + } + + fn merge(num_columns: usize, partitions: &[PartitionBounds]) -> MergedBounds { + let refs = partitions.iter().collect::>(); + merge_partition_bounds(num_columns, &refs) + } + + fn ranges(merged: &MergedBounds, column: usize) -> Vec<(i32, i32)> { + merged.per_column[column] + .iter() + .map(|(min, max)| match (min, max) { + (ScalarValue::Int32(Some(min)), ScalarValue::Int32(Some(max))) => { + (*min, *max) + } + other => panic!("expected Int32 range, got {other:?}"), + }) + .collect() + } + + #[test] + fn overlapping_ranges_collapse_to_one() { + let merged = merge(1, &[partition(&[(0, 10)]), partition(&[(5, 20)])]); + assert_eq!(ranges(&merged, 0), vec![(0, 20)]); + } + + #[test] + fn disjoint_ranges_stay_separate() { + let merged = merge(1, &[partition(&[(100, 110)]), partition(&[(0, 10)])]); + assert_eq!(ranges(&merged, 0), vec![(0, 10), (100, 110)]); + } + + #[test] + fn contained_range_is_absorbed() { + let merged = merge(1, &[partition(&[(0, 100)]), partition(&[(10, 20)])]); + assert_eq!(ranges(&merged, 0), vec![(0, 100)]); + } + + #[test] + fn too_many_disjoint_ranges_collapse_to_convex_hull() { + let partitions = (0..MAX_RANGES_PER_COLUMN + 1) + .map(|i| partition(&[(i as i32 * 100, i as i32 * 100 + 1)])) + .collect::>(); + let merged = merge(1, &partitions); + assert_eq!( + ranges(&merged, 0), + vec![(0, MAX_RANGES_PER_COLUMN as i32 * 100 + 1)] + ); + } + + #[test] + fn a_partition_without_bounds_disables_the_column() { + // The second partition reports nothing for column 0, so any value could + // route there and the column must stay unconstrained. + let merged = merge(1, &[partition(&[(0, 10)]), partition(&[])]); + assert!(merged.per_column[0].is_empty()); + assert!(merged.is_degenerate()); + } + + #[test] + fn null_bounds_disable_the_column() { + let with_null = PartitionBounds::new(vec![ColumnBounds::new( + ScalarValue::Int32(None), + ScalarValue::Int32(None), + )]); + let merged = merge(1, &[partition(&[(0, 10)]), with_null]); + assert!(merged.per_column[0].is_empty()); + } + + #[test] + fn columns_are_merged_independently() { + let merged = merge( + 2, + &[ + partition(&[(0, 10), (100, 110)]), + partition(&[(20, 30), (0, 5)]), + ], + ); + assert_eq!(ranges(&merged, 0), vec![(0, 10), (20, 30)]); + assert_eq!(ranges(&merged, 1), vec![(0, 5), (100, 110)]); + } + + #[test] + fn relaxation_is_zero_when_every_partition_spans_the_union() { + let merged = merge(1, &[partition(&[(0, 100)]), partition(&[(0, 100)])]); + assert_eq!(merged.relaxation(), Some(0.0)); + } + + #[test] + fn relaxation_is_small_when_partitions_nearly_span_the_union() { + // The hash-scattered case: every partition sees nearly the whole key + // range, so the union is barely wider than any single partition. + let merged = merge(1, &[partition(&[(0, 99)]), partition(&[(1, 100)])]); + let relaxation = merged.relaxation().expect("numeric bounds"); + assert!( + (relaxation - 0.01).abs() < 1e-9, + "expected ~1% relaxation, got {relaxation}" + ); + } + + #[test] + fn relaxation_counts_only_the_disjoint_ranges_kept() { + // Two non-overlapping partitions of equal width: a key routes to one of + // them, so half of the two-range union is wrongly admitted. Keeping the + // ranges disjoint is what keeps this at 0.5 instead of ~0.98, which is + // what collapsing to the convex hull [0, 100] would cost. + let merged = merge(1, &[partition(&[(0, 1)]), partition(&[(99, 100)])]); + assert_eq!(ranges(&merged, 0), vec![(0, 1), (99, 100)]); + assert_eq!(merged.relaxation(), Some(0.5)); + } + + #[test] + fn relaxation_is_unknown_for_non_numeric_keys() { + let utf8 = |min: &str, max: &str| { + PartitionBounds::new(vec![ColumnBounds::new( + ScalarValue::from(min), + ScalarValue::from(max), + )]) + }; + let merged = merge(1, &[utf8("a", "m"), utf8("n", "z")]); + assert_eq!(merged.relaxation(), None); + // The ranges themselves are still usable: only the estimate is missing. + assert!(!merged.is_degenerate()); + } + + #[test] + fn predicate_ors_disjoint_ranges_and_ands_columns() { + let merged = merge( + 2, + &[ + partition(&[(0, 10), (0, 5)]), + partition(&[(100, 110), (0, 5)]), + ], + ); + let on_right: Vec = + vec![Arc::new(Column::new("a", 0)), Arc::new(Column::new("b", 1))]; + let predicate = create_merged_bounds_predicate(&on_right, &merged) + .expect("expected a bounds predicate"); + assert_eq!( + format!("{predicate}"), + "(a@0 >= 0 AND a@0 <= 10 OR a@0 >= 100 AND a@0 <= 110) AND b@1 >= 0 AND b@1 <= 5" + ); + } + + #[test] + fn degenerate_merge_has_no_predicate() { + let merged = merge(1, &[partition(&[(0, 10)]), partition(&[])]); + let on_right: Vec = vec![Arc::new(Column::new("a", 0))]; + assert!(create_merged_bounds_predicate(&on_right, &merged).is_none()); + } +} diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 99984d82813f..0be6587ff807 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -782,15 +782,30 @@ pub struct HashJoinExec { struct HashJoinExecDynamicFilter { /// Dynamic filter that we'll update with the results of the build side once that is done. filter: Arc, + /// Second dynamic filter, pushed alongside [`Self::filter`] so the probe + /// side sees `DynamicFilter[bounds] AND DynamicFilter[membership]` rather + /// than one opaque conjunct. Only created for `PartitionMode::Partitioned`, + /// where the membership check is otherwise buried inside a routing `CASE`. + /// See `SharedBuildAccumulator::bounds_dynamic_filter` for why the split + /// has to happen at the wrapper level. + bounds_filter: Option>, /// Build accumulator to collect build-side information (hash maps and/or bounds) from each partition. /// It is lazily initialized during execution to make sure we use the actual execution time partition counts. build_accumulator: OnceLock>, } impl HashJoinExecDynamicFilter { - /// Every dynamic expression this join produces. + /// Every dynamic expression this join produces, membership first. + /// + /// The membership filter leads because it is the self-sufficient one: it + /// still carries the build-side bounds inside its routing `CASE` whenever + /// the bounds filter is absent, so consumers that only look at the first + /// produced expression (plan serialization, for one) stay correct. fn produced_expressions(&self) -> Vec> { - vec![Arc::::clone(&self.filter) as _] + std::iter::once(&self.filter) + .chain(self.bounds_filter.as_ref()) + .map(|filter| Arc::::clone(filter) as _) + .collect() } } @@ -989,6 +1004,10 @@ impl HashJoinExec { } self.dynamic_filter = Some(HashJoinExecDynamicFilter { filter, + // The bounds half is created by filter pushdown; a filter restored + // from a serialized plan carries only the membership half, which is + // self-sufficient (it keeps the bounds inside its routing `CASE`). + bounds_filter: None, // Initialize with an empty accumulator which will be lazily populated // during execution. build_accumulator: OnceLock::new(), @@ -1429,6 +1448,20 @@ impl ExecutionPlan for HashJoinExec { .then(|| { self.dynamic_filter.as_ref().map(|df| { let filter = Arc::clone(&df.filter); + // Only drive the bounds filter when the probe subtree + // actually consumes it; an orphaned one would never be + // read, and the accumulator keeps the bounds inside the + // routing `CASE` when it is absent. + let bounds_filter = df + .bounds_filter + .as_ref() + .filter(|bounds_filter| { + bounds_filter.expression_id().is_some_and(|id| { + plan_contains_expression_id(&self.right, id) + .unwrap_or(false) + }) + }) + .map(Arc::clone); let on_right = self .on .iter() @@ -1440,6 +1473,7 @@ impl ExecutionPlan for HashJoinExec { self.left.as_ref(), self.right.as_ref(), filter, + bounds_filter, on_right, repartition_random_state, self.null_equality, @@ -1743,9 +1777,22 @@ impl ExecutionPlan for HashJoinExec { && self.dynamic_filter.is_none() && self.allow_join_dynamic_filter_pushdown(config) { - // Add actual dynamic filter to right side (probe side) - let dynamic_filter = Self::create_dynamic_filter(&self.on); - right_child = right_child.with_self_filter(dynamic_filter); + // Add actual dynamic filter to right side (probe side). + // + // A partitioned join pushes *two* filters: the build-side value + // ranges and the membership check. They stay separate wrappers on + // purpose — `split_conjunction` does not look inside a + // `DynamicFilterPhysicalExpr`, so only two wrappers reach the + // Parquet reader as two predicates, letting the cheap range check + // run (and prune row groups) before the expensive membership + // lookup. The bounds filter is listed first so it is the first + // conjunct the reader sees. + let mut self_filters: Vec> = Vec::with_capacity(2); + if self.mode == PartitionMode::Partitioned { + self_filters.push(Self::create_dynamic_filter(&self.on)); + } + self_filters.push(Self::create_dynamic_filter(&self.on)); + right_child = right_child.with_self_filters(self_filters); } Ok(FilterDescription::new() @@ -1762,6 +1809,8 @@ impl ExecutionPlan for HashJoinExec { let mut result = FilterPushdownPropagation::if_any(child_pushdown_result.clone()); assert_eq!(child_pushdown_result.self_filters.len(), 2); // Should always be 2, we have 2 children let right_child_self_filters = &child_pushdown_result.self_filters[1]; // We only push down filters to the right child + // We expect 0, 1 or 2 self filters: a partitioned join also pushes a + // separate bounds filter ahead of the membership one. // Note that we don't check PushdDownPredicate::discrimnant because even if nothing said // "yes, I can fully evaluate this filter" things might still use it for statistics -> it's worth updating let mut pushed = right_child_self_filters @@ -1771,13 +1820,15 @@ impl ExecutionPlan for HashJoinExec { Arc::downcast::(predicate).ok() }) .collect::>(); - // The filter this join drives is always the last one pushed. + // The membership filter is always the last one pushed, so pop it first; + // whatever remains is the bounds filter. if let Some(filter) = pushed.pop() { // We successfully pushed down our self filter - we need to make a new node with the dynamic filter let new_node = self .builder() .with_dynamic_filter(Some(HashJoinExecDynamicFilter { filter, + bounds_filter: pushed.pop(), build_accumulator: OnceLock::new(), })) .build_exec()?; @@ -2762,6 +2813,7 @@ mod tests { )?; join.dynamic_filter = Some(HashJoinExecDynamicFilter { filter: Arc::clone(&dynamic_filter), + bounds_filter: None, build_accumulator: OnceLock::new(), }); diff --git a/datafusion/physical-plan/src/joins/hash_join/mod.rs b/datafusion/physical-plan/src/joins/hash_join/mod.rs index b915802ea401..90fce85b39e1 100644 --- a/datafusion/physical-plan/src/joins/hash_join/mod.rs +++ b/datafusion/physical-plan/src/joins/hash_join/mod.rs @@ -20,6 +20,7 @@ pub use exec::{HashJoinExec, HashJoinExecBuilder}; pub use partitioned_hash_eval::{HashExpr, HashTableLookupExpr, SeededRandomState}; +mod bounds_union; mod exec; mod inlist_builder; mod partitioned_hash_eval; 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 7b58107e93c3..1f3972fa0050 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -25,6 +25,9 @@ use crate::ExecutionPlan; use crate::ExecutionPlanProperties; use crate::joins::Map; use crate::joins::PartitionMode; +use crate::joins::hash_join::bounds_union::{ + create_merged_bounds_predicate, merge_partition_bounds, +}; use crate::joins::hash_join::exec::HASH_JOIN_SEED; use crate::joins::hash_join::inlist_builder::build_struct_fields; use crate::joins::hash_join::partitioned_hash_eval::{ @@ -248,8 +251,24 @@ pub(crate) struct SharedBuildAccumulator { /// result, then broadcasts), so late subscribers simply re-check the /// state under the mutex and return immediately. completion_notify: Notify, - /// Dynamic filter for pushdown to probe side + /// Dynamic filter carrying the membership half of the pushed predicate + /// (and, when [`Self::bounds_dynamic_filter`] is absent, the whole thing). dynamic_filter: Arc, + /// Second dynamic filter carrying the build-side value ranges, `AND`'d with + /// [`Self::dynamic_filter`] on the probe side. + /// + /// Two wrappers rather than one `DynamicFilter[bounds AND membership]` is + /// deliberate: `split_conjunction` splits only on a top-level `AND` and + /// does not look inside a `DynamicFilterPhysicalExpr`, so a single wrapper + /// reaches the Parquet reader as one opaque conjunct. Split in two, the + /// reader builds two `ArrowPredicate`s and applies them in sequence against + /// an accumulating `RowSelection` — the cheap vectorized range check runs + /// first and the expensive membership lookup only sees the survivors — and + /// the range half additionally becomes visible to row-group pruning. + /// + /// `None` for `CollectLeft` joins, which have no routing to hoist bounds + /// out of, and whenever the second filter did not survive pushdown. + bounds_dynamic_filter: Option>, /// Right side join expressions needed for creating filter expressions on_right: Vec, /// Random state for partitioning (RepartitionExec's hash function with 0,0,0,0 seeds) @@ -370,6 +389,7 @@ impl SharedBuildAccumulator { left_child: &dyn ExecutionPlan, right_child: &dyn ExecutionPlan, dynamic_filter: Arc, + bounds_dynamic_filter: Option>, on_right: Vec, repartition_random_state: SeededRandomState, null_equality: NullEquality, @@ -417,6 +437,11 @@ impl SharedBuildAccumulator { }), completion_notify: Notify::new(), dynamic_filter, + // Only a partitioned join routes through a `CASE`, so only it has + // bounds to hoist out into a second conjunct. + bounds_dynamic_filter: matches!(partition_mode, PartitionMode::Partitioned) + .then_some(bounds_dynamic_filter) + .flatten(), on_right, repartition_random_state, probe_schema: right_child.schema(), @@ -568,6 +593,9 @@ impl SharedBuildAccumulator { fn finish(&self, finalize_input: FinalizeInput) { let result = self.build_filter(finalize_input).map_err(Arc::new); self.dynamic_filter.mark_complete(); + if let Some(bounds_filter) = &self.bounds_dynamic_filter { + bounds_filter.mark_complete(); + } let mut guard = self.inner.lock(); guard.completion = CompletionState::Ready(result); @@ -627,20 +655,7 @@ impl SharedBuildAccumulator { } }, FinalizeInput::Partitioned(partitions) => { - let num_partitions = partitions.len(); - let routing_hash_expr = Arc::new(HashExpr::new( - self.on_right.clone(), - self.repartition_random_state.clone(), - "hash_repartition".to_string(), - )) as Arc; - - let modulo_expr = Arc::new(BinaryExpr::new( - routing_hash_expr, - Operator::Modulo, - lit(ScalarValue::UInt64(Some(num_partitions as u64))), - )) as Arc; - - let mut real_branches = Vec::new(); + let mut real_partitions: Vec<(usize, &PartitionData)> = Vec::new(); let mut empty_partition_ids = Vec::new(); let mut has_canceled_unknown = false; let mut keys_have_null = false; @@ -654,25 +669,7 @@ impl SharedBuildAccumulator { } PartitionStatus::Reported(partition) => { keys_have_null |= partition.keys_have_null; - let membership_expr = create_membership_predicate( - &self.on_right, - partition.pushdown.clone(), - &HASH_JOIN_SEED, - self.probe_schema.as_ref(), - )?; - let bounds_expr = create_bounds_predicate( - &self.on_right, - &partition.bounds, - ); - let then_expr = combine_membership_and_bounds( - membership_expr, - bounds_expr, - ) - .unwrap_or_else(|| lit(true)); - real_branches.push(( - lit(ScalarValue::UInt64(Some(partition_id as u64))), - then_expr, - )); + real_partitions.push((partition_id, partition)); } PartitionStatus::CanceledUnknown => { has_canceled_unknown = true; @@ -688,49 +685,154 @@ impl SharedBuildAccumulator { } } - let filter_expr = if has_canceled_unknown { - let mut when_then_branches = empty_partition_ids - .into_iter() - .map(|partition_id| { - ( - lit(ScalarValue::UInt64(Some(partition_id as u64))), - lit(false), - ) - }) - .collect::>(); - when_then_branches.extend(real_branches); - - if when_then_branches.is_empty() { - lit(true) + // A canceled partition can hold any value, so its route must stay + // permissive and no set-theoretic union over the *reported* partitions + // describes the build side. Keep the bounds inside the `CASE` in that + // case, exactly as before. + let split_bounds = + self.bounds_dynamic_filter.is_some() && !has_canceled_unknown; + + let bounds_expr = if split_bounds { + let merged = merge_partition_bounds( + self.on_right.len(), + &real_partitions + .iter() + .map(|(_, partition)| &partition.bounds) + .collect::>(), + ); + log::debug!( + "hash join merged build-side bounds over {} partitions, \ + degenerate {}, estimated relaxation {:?}", + real_partitions.len(), + merged.is_degenerate(), + merged.relaxation() + ); + // A degenerate merge constrains nothing, so emitting it + // would cost an evaluation per probe batch and prune + // nothing. Leave the conjunct at `true` instead. + if merged.is_degenerate() { + None } else { - Arc::new(CaseExpr::try_new( - Some(modulo_expr), - when_then_branches, - Some(lit(true)), - )?) as Arc + create_merged_bounds_predicate(&self.on_right, &merged) } - } else if real_branches.is_empty() { - lit(false) - } else if real_branches.len() == 1 - && empty_partition_ids.len() + 1 == num_partitions - { - Arc::clone(&real_branches[0].1) } else { - Arc::new(CaseExpr::try_new( - Some(modulo_expr), - real_branches, - Some(lit(false)), - )?) as Arc + None }; - self.dynamic_filter - .update(self.preserve_probe_nulls(filter_expr, keys_have_null)?)?; + let membership_expr = self.build_partitioned_membership( + partitions.len(), + &real_partitions, + &empty_partition_ids, + has_canceled_unknown, + // When the bounds cannot be hoisted into their own conjunct they + // must stay in the routed branches, or they are lost entirely. + !split_bounds, + )?; + + match (&self.bounds_dynamic_filter, bounds_expr) { + (Some(bounds_filter), Some(bounds_expr)) => { + // Both halves keep the NULL widening: the probe side sees + // `(nulls OR bounds) AND (nulls OR membership)`, which is + // `nulls OR (bounds AND membership)`. + bounds_filter.update( + self.preserve_probe_nulls(bounds_expr, keys_have_null)?, + )?; + } + // Nothing usable to hoist; leave the conjunct as the no-op it + // was initialized with so it folds away. + (Some(bounds_filter), None) => bounds_filter.update(lit(true))?, + (None, _) => {} + } + + self.dynamic_filter.update( + self.preserve_probe_nulls(membership_expr, keys_have_null)?, + )?; } } Ok(()) } + /// Builds the routing half of a partitioned join's pushed predicate. + /// + /// Normally this is a `CASE hash(keys) % n WHEN i THEN `, + /// but when every non-empty partition pushes an `InList` the routing is + /// redundant and the whole thing collapses to a single `InList` over the + /// union — see [`Self::union_inlist_membership`]. + fn build_partitioned_membership( + &self, + num_partitions: usize, + real_partitions: &[(usize, &PartitionData)], + empty_partition_ids: &[usize], + has_canceled_unknown: bool, + include_bounds: bool, + ) -> Result> { + let mut real_branches = Vec::with_capacity(real_partitions.len()); + for (partition_id, partition) in real_partitions { + let membership_expr = create_membership_predicate( + &self.on_right, + partition.pushdown.clone(), + &HASH_JOIN_SEED, + self.probe_schema.as_ref(), + )?; + let bounds_expr = include_bounds + .then(|| create_bounds_predicate(&self.on_right, &partition.bounds)) + .flatten(); + let then_expr = combine_membership_and_bounds(membership_expr, bounds_expr) + .unwrap_or_else(|| lit(true)); + real_branches.push(( + lit(ScalarValue::UInt64(Some(*partition_id as u64))), + then_expr, + )); + } + + let routing_hash_expr = Arc::new(HashExpr::new( + self.on_right.clone(), + self.repartition_random_state.clone(), + "hash_repartition".to_string(), + )) as Arc; + let modulo_expr = Arc::new(BinaryExpr::new( + routing_hash_expr, + Operator::Modulo, + lit(ScalarValue::UInt64(Some(num_partitions as u64))), + )) as Arc; + + if has_canceled_unknown { + let mut when_then_branches = empty_partition_ids + .iter() + .map(|partition_id| { + ( + lit(ScalarValue::UInt64(Some(*partition_id as u64))), + lit(false), + ) + }) + .collect::>(); + when_then_branches.extend(real_branches); + + if when_then_branches.is_empty() { + Ok(lit(true)) + } else { + Ok(Arc::new(CaseExpr::try_new( + Some(modulo_expr), + when_then_branches, + Some(lit(true)), + )?) as Arc) + } + } else if real_branches.is_empty() { + Ok(lit(false)) + } else if real_branches.len() == 1 + && empty_partition_ids.len() + 1 == num_partitions + { + Ok(Arc::clone(&real_branches[0].1)) + } else { + Ok(Arc::new(CaseExpr::try_new( + Some(modulo_expr), + real_branches, + Some(lit(false)), + )?) as Arc) + } + } + /// Keeps probe rows with a NULL key when the join semantics need them. /// /// The build-side predicate drops probe rows whose key is NULL. A null-aware anti join @@ -806,6 +908,7 @@ pub(super) fn make_partitioned_accumulator_for_test( }), completion_notify: Notify::new(), dynamic_filter, + bounds_dynamic_filter: None, on_right: vec![], repartition_random_state: SeededRandomState::with_seed(1), probe_schema, @@ -831,6 +934,8 @@ pub(super) fn completed_partitions_for_test(acc: &SharedBuildAccumulator) -> usi mod tests { use super::*; + use crate::joins::join_hash_map::JoinHashMapU32; + use arrow::array::{ArrayRef, Int32Array}; use datafusion_physical_expr::expressions::{Column, Literal}; @@ -855,8 +960,17 @@ mod tests { fn make_accumulator_for_test( data: AccumulatedBuildData, on_right: Vec, + ) -> SharedBuildAccumulator { + make_accumulator_with_bounds_filter_for_test(data, on_right, false) + } + + fn make_accumulator_with_bounds_filter_for_test( + data: AccumulatedBuildData, + on_right: Vec, + split_bounds: bool, ) -> SharedBuildAccumulator { let dynamic_filter = test_dynamic_filter(&on_right); + let bounds_dynamic_filter = split_bounds.then(|| test_dynamic_filter(&on_right)); SharedBuildAccumulator { inner: Mutex::new(AccumulatorState { data, @@ -864,6 +978,7 @@ mod tests { }), completion_notify: Notify::new(), dynamic_filter, + bounds_dynamic_filter, on_right, repartition_random_state: SeededRandomState::with_seed(1), probe_schema: test_probe_schema(), @@ -895,10 +1010,41 @@ mod tests { ) } + /// A partitioned accumulator wired the way filter pushdown wires it in + /// production: a separate dynamic filter for the bounds conjunct. + fn make_split_partitioned_accumulator_for_test( + num_partitions: usize, + ) -> SharedBuildAccumulator { + make_accumulator_with_bounds_filter_for_test( + AccumulatedBuildData::Partitioned { + partitions: vec![PartitionStatus::Pending; num_partitions], + completed_partitions: 0, + }, + test_on_right(), + true, + ) + } + + fn current_bounds_expr(acc: &SharedBuildAccumulator) -> PhysicalExprRef { + acc.bounds_dynamic_filter + .as_ref() + .expect("expected a bounds dynamic filter") + .current() + .expect("bounds dynamic filter current expression should be available") + } + fn in_list(values: &[i32]) -> PushdownStrategy { PushdownStrategy::InList(Arc::new(Int32Array::from(values.to_vec())) as ArrayRef) } + /// A build side too large for an `InList`, which is what forces the routed + /// `CASE` to survive. + fn map_pushdown() -> PushdownStrategy { + PushdownStrategy::Map(Arc::new(Map::HashMap(Box::new( + JoinHashMapU32::with_capacity(1), + )))) + } + fn bounds(min: i32, max: i32) -> PartitionBounds { PartitionBounds::new(vec![ColumnBounds::new( ScalarValue::Int32(Some(min)), @@ -1056,6 +1202,90 @@ mod tests { assert!(expr.downcast_ref::().is_none()); } + #[test] + fn partitioned_split_hoists_merged_bounds_out_of_the_case() { + let acc = make_split_partitioned_accumulator_for_test(2); + + // Two large (map-backed) partitions whose ranges overlap: the union is + // a single range and the routing CASE keeps only the membership checks. + acc.build_filter(FinalizeInput::Partitioned(vec![ + reported(map_pushdown(), bounds(0, 10)), + reported(map_pushdown(), bounds(5, 20)), + ])) + .unwrap(); + + let bounds_expr = current_bounds_expr(&acc); + assert_eq!( + format!("{bounds_expr}"), + "probe_key@0 >= 0 AND probe_key@0 <= 20" + ); + + // The membership half is a routing CASE with no bounds left in it. + let membership_expr = current_expr(&acc); + let case = case_expr(&membership_expr); + assert_eq!(case.when_then_expr().len(), 2); + for (_, then_expr) in case.when_then_expr() { + assert!( + then_expr.downcast_ref::().is_some(), + "expected a bare membership check, got {then_expr}" + ); + } + } + + #[test] + fn partitioned_split_ors_disjoint_partition_ranges() { + let acc = make_split_partitioned_accumulator_for_test(2); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + reported(map_pushdown(), bounds(0, 10)), + reported(map_pushdown(), bounds(100, 110)), + ])) + .unwrap(); + + // Disjoint ranges are kept apart instead of being widened to the hull: + // `[0, 110]` would admit everything in between for nothing. + let bounds_expr = current_bounds_expr(&acc); + assert_eq!( + format!("{bounds_expr}"), + "probe_key@0 >= 0 AND probe_key@0 <= 10 OR probe_key@0 >= 100 AND probe_key@0 <= 110" + ); + } + + #[test] + fn partitioned_split_leaves_bounds_in_the_case_when_a_partition_is_canceled() { + let acc = make_split_partitioned_accumulator_for_test(2); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + PartitionStatus::CanceledUnknown, + reported(map_pushdown(), bounds(0, 10)), + ])) + .unwrap(); + + // A canceled partition can hold any value, so no union over the + // reported partitions describes the build side: the hoisted conjunct + // must stay a no-op and the bounds stay inside the routed branch. + assert_literal_bool(¤t_bounds_expr(&acc), true); + let membership_expr = current_expr(&acc); + let case = case_expr(&membership_expr); + assert_eq!(case.when_then_expr().len(), 1); + assert_top_binary_op(&case.when_then_expr()[0].1, Operator::And); + } + + #[test] + fn partitioned_split_without_usable_bounds_emits_a_no_op_conjunct() { + let acc = make_split_partitioned_accumulator_for_test(2); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + reported(map_pushdown(), bounds(0, 10)), + // No bounds reported here, so any value could route to partition 1 + // and the column cannot be constrained at all. + reported(map_pushdown(), no_bounds()), + ])) + .unwrap(); + + assert_literal_bool(¤t_bounds_expr(&acc), true); + } + #[test] fn partitioned_canceled_unknown_partitions_keep_unknown_routes_permissive() { let acc = make_partitioned_expr_accumulator_for_test(2); @@ -1150,6 +1380,7 @@ mod tests { }), completion_notify: Notify::new(), dynamic_filter: Arc::new(DynamicFilterPhysicalExpr::new(vec![], lit(true))), + bounds_dynamic_filter: None, on_right, repartition_random_state: SeededRandomState::with_seed(1), probe_schema, diff --git a/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt b/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt index e2dd22cc82bb..f0916d6cf313 100644 --- a/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt +++ b/datafusion/sqllogictest/test_files/preserve_file_partitioning.slt @@ -643,7 +643,7 @@ physical_plan 05)--------RepartitionExec: partitioning=Hash([d_dkey@1], 3), input_partitions=3 06)----------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension_partitioned/d_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension_partitioned/d_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/dimension_partitioned/d_dkey=C/data.parquet]]}, projection=[env, d_dkey], file_type=parquet 07)--------RepartitionExec: partitioning=Hash([f_dkey@1], 3), input_partitions=3 -08)----------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible +08)----------DataSourceExec: file_groups={3 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=A/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=B/data.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/preserve_file_partitioning/fact/f_dkey=C/data.parquet]]}, projection=[value, f_dkey], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible query TTR rowsort SELECT f.f_dkey, d.env, sum(f.value) diff --git a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt index 72d034067663..5d25c1ae48fd 100644 --- a/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt +++ b/datafusion/sqllogictest/test_files/push_down_filter_parquet.slt @@ -1237,7 +1237,7 @@ physical_plan 02)--RepartitionExec: partitioning=Hash([id@0], 4), input_partitions=2 03)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/pnej_build/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/pnej_build/2.parquet]]}, projection=[id], file_type=parquet 04)--RepartitionExec: partitioning=Hash([id@0], 4), input_partitions=2 -05)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/pnej_probe/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/pnej_probe/2.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible +05)----DataSourceExec: file_groups={2 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/pnej_probe/1.parquet], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/push_down_filter_parquet/pnej_probe/2.parquet]]}, projection=[id], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible query II rowsort SELECT pnej_build.id, pnej_probe.id FROM pnej_build JOIN pnej_probe ON pnej_build.id IS NOT DISTINCT FROM pnej_probe.id diff --git a/datafusion/sqllogictest/test_files/statistics_registry.slt b/datafusion/sqllogictest/test_files/statistics_registry.slt index 89258bec299c..bccfd13f4ba3 100644 --- a/datafusion/sqllogictest/test_files/statistics_registry.slt +++ b/datafusion/sqllogictest/test_files/statistics_registry.slt @@ -104,9 +104,9 @@ physical_plan 02)--RepartitionExec: partitioning=Hash([small_id@2], 4), input_partitions=1, maintains_sort_order=true 03)----HashJoinExec: mode=Partitioned, join_type=Inner, on=[(customer_id@0, customer_id@1)], projection=[region_id@1, order_id@2, small_id@4] 04)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/customers.parquet]]}, projection=[customer_id, region_id], output_ordering=[region_id@1 ASC NULLS LAST], file_type=parquet -05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/orders.parquet]]}, projection=[order_id, customer_id, small_id], output_ordering=[order_id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible +05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/orders.parquet]]}, projection=[order_id, customer_id, small_id], output_ordering=[order_id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible 06)--RepartitionExec: partitioning=Hash([small_id@0], 4), input_partitions=1, maintains_sort_order=true -07)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/dim_small.parquet]]}, projection=[small_id, label], output_ordering=[small_id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible +07)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/dim_small.parquet]]}, projection=[small_id, label], output_ordering=[small_id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible # -- With registry ----------------------------------------------------------- # Conservative estimate 100 > 50: dim_small correctly swapped to build side @@ -127,7 +127,7 @@ physical_plan 04)--RepartitionExec: partitioning=Hash([small_id@2], 4), input_partitions=1, maintains_sort_order=true 05)----HashJoinExec: mode=Partitioned, join_type=Inner, on=[(customer_id@0, customer_id@1)], projection=[region_id@1, order_id@2, small_id@4] 06)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/customers.parquet]]}, projection=[customer_id, region_id], output_ordering=[region_id@1 ASC NULLS LAST], file_type=parquet -07)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/orders.parquet]]}, projection=[order_id, customer_id, small_id], output_ordering=[order_id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible +07)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/statistics_registry/orders.parquet]]}, projection=[order_id, customer_id, small_id], output_ordering=[order_id@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible # -- Verify results are identical regardless of join order -------------------- From 8768c312fe0ee4f50791e2139d79c97766effa8f Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Mon, 10 Aug 2026 12:35:09 -0400 Subject: [PATCH 3/3] Collapse an all-InList partitioned membership check into one InList MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When every non-empty build partition pushes an `InList`, the routing `CASE` is redundant. Routing is a deterministic function of the key columns, so every build row holding key `K` lands in the same partition a probe row holding `K` routes to; testing `K` against the union of all the lists therefore accepts and rejects *precisely* what the `CASE` does. This is an equality, not the relaxation the bounds merge is. Collapsing drops the routing hash from the probe path entirely and leaves an `InListExpr`, which unlike a `CASE` is a shape `datafusion/pruning` understands (up to `max_in_list_size`). The union is capped at 1 MiB: each partition's list is independently limited by `hash_join_inlist_pushdown_max_size`, so the concatenation grows with the partition count, and past some size the routed `CASE` — where a probe row only ever probes one list — is the cheaper shape. Partitions that push a hash map, and builds where a canceled partition makes the union incomplete, keep the `CASE`. Co-Authored-By: Claude Opus 5 --- .../src/joins/hash_join/shared_bounds.rs | 105 ++++++++++++++++++ 1 file changed, 105 insertions(+) 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 1f3972fa0050..218e78ef14f5 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -34,6 +34,7 @@ use crate::joins::hash_join::partitioned_hash_eval::{ HashExpr, HashTableLookupExpr, SeededRandomState, }; use arrow::array::ArrayRef; +use arrow::compute::concat; use arrow::datatypes::{DataType, Field, Schema}; use datafusion_common::config::ConfigOptions; use datafusion_common::{ @@ -284,6 +285,15 @@ pub(crate) struct SharedBuildAccumulator { null_aware: bool, } +/// Ceiling on the combined size of the per-partition `InList` arrays that +/// [`SharedBuildAccumulator::union_inlist_membership`] will concatenate. +/// +/// Each partition's list is independently capped by +/// `hash_join_inlist_pushdown_max_size`, so without a combined cap the union +/// grows with the partition count. Past this size, keeping the routed `CASE` — +/// where each probe row only ever probes one list — is the cheaper shape. +const MAX_UNIONED_INLIST_BYTES: usize = 1024 * 1024; + /// Strategy for filter pushdown (decided at collection time) #[derive(Clone)] pub(crate) enum PushdownStrategy { @@ -767,6 +777,13 @@ impl SharedBuildAccumulator { has_canceled_unknown: bool, include_bounds: bool, ) -> Result> { + if !has_canceled_unknown + && !include_bounds + && let Some(union) = self.union_inlist_membership(real_partitions)? + { + return Ok(union); + } + let mut real_branches = Vec::with_capacity(real_partitions.len()); for (partition_id, partition) in real_partitions { let membership_expr = create_membership_predicate( @@ -833,6 +850,60 @@ impl SharedBuildAccumulator { } } + /// Collapses an all-`InList` partitioned membership check into one `InList` + /// over the union of the per-partition lists. + /// + /// This is **exact**, not a relaxation: routing is a deterministic function + /// of the key columns, so every build row holding key `K` lands in the same + /// partition a probe row holding `K` routes to. Testing `K` against the + /// union therefore accepts and rejects precisely what the `CASE` does, and + /// the result no longer needs the routing hash at all — nor is it opaque to + /// pruning the way a `CASE` is. + /// + /// Returns `None` when the collapse does not apply (some partition pushes a + /// hash map instead of a list, the lists disagree on type, or the union + /// would be too large to be worth materializing). + fn union_inlist_membership( + &self, + real_partitions: &[(usize, &PartitionData)], + ) -> Result>> { + if real_partitions.is_empty() { + return Ok(None); + } + + let mut arrays = Vec::with_capacity(real_partitions.len()); + let mut total_bytes = 0; + for (_, partition) in real_partitions { + let PushdownStrategy::InList(values) = &partition.pushdown else { + return Ok(None); + }; + total_bytes += values.get_array_memory_size(); + if total_bytes > MAX_UNIONED_INLIST_BYTES { + return Ok(None); + } + if arrays + .first() + .is_some_and(|first: &&ArrayRef| first.data_type() != values.data_type()) + { + return Ok(None); + } + arrays.push(values); + } + + let union: ArrayRef = if arrays.len() == 1 { + Arc::clone(arrays[0]) + } else { + concat(&arrays.iter().map(|a| a.as_ref()).collect::>())? + }; + + create_membership_predicate( + &self.on_right, + PushdownStrategy::InList(union), + &HASH_JOIN_SEED, + self.probe_schema.as_ref(), + ) + } + /// Keeps probe rows with a NULL key when the join semantics need them. /// /// The build-side predicate drops probe rows whose key is NULL. A null-aware anti join @@ -1251,6 +1322,40 @@ mod tests { ); } + #[test] + fn partitioned_all_inlist_collapses_to_a_single_union_inlist() { + let acc = make_split_partitioned_accumulator_for_test(3); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + reported(in_list(&[1, 4]), bounds(1, 4)), + reported(in_list(&[2, 5]), bounds(2, 5)), + reported(PushdownStrategy::Empty, no_bounds()), + ])) + .unwrap(); + + // Routing is a function of the key, so a probe key only ever matches + // the list of the partition it routes to: the union is exact and the + // `CASE` (and its routing hash) disappears entirely. + let membership_expr = current_expr(&acc); + assert!(membership_expr.downcast_ref::().is_none()); + assert_in_list_column_values(&membership_expr, "probe_key", 0, &[1, 4, 2, 5]); + } + + #[test] + fn partitioned_mixed_strategies_keep_the_routing_case() { + let acc = make_split_partitioned_accumulator_for_test(2); + + acc.build_filter(FinalizeInput::Partitioned(vec![ + reported(in_list(&[1, 2]), bounds(1, 2)), + reported(map_pushdown(), bounds(3, 4)), + ])) + .unwrap(); + + // One partition still needs a hash-table lookup, so routing is required. + let membership_expr = current_expr(&acc); + assert_eq!(case_expr(&membership_expr).when_then_expr().len(), 2); + } + #[test] fn partitioned_split_leaves_bounds_in_the_case_when_a_partition_is_canceled() { let acc = make_split_partitioned_accumulator_for_test(2);