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 4e68d871b81a..0be6587ff807 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -782,11 +782,33 @@ 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, 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> { + std::iter::once(&self.filter) + .chain(self.bounds_filter.as_ref()) + .map(|filter| Arc::::clone(filter) as _) + .collect() + } +} + impl fmt::Debug for HashJoinExec { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("HashJoinExec") @@ -982,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(), @@ -1338,20 +1364,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() } @@ -1425,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() @@ -1436,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, @@ -1739,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() @@ -1758,24 +1809,30 @@ 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); - } + // 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 + .iter() + .filter_map(|filter| { + let predicate = Arc::clone(&filter.predicate); + Arc::downcast::(predicate).ok() + }) + .collect::>(); + // 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()?; + result = result.with_updated_node(new_node); } Ok(result) } @@ -2756,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..218e78ef14f5 100644 --- a/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs +++ b/datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs @@ -25,12 +25,16 @@ 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::{ 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::{ @@ -248,8 +252,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) @@ -265,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 { @@ -370,6 +399,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 +447,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 +603,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 +665,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 +679,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 +695,215 @@ 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> { + 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( + &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) + } + } + + /// 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 @@ -806,6 +979,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 +1005,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 +1031,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 +1049,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 +1081,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 +1273,124 @@ 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_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); + + 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 +1485,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 --------------------