From 30a9139a834825cacf59c48e2c3b2800769af19b Mon Sep 17 00:00:00 2001 From: Jay Zhan Date: Sun, 20 Sep 2026 21:24:34 +0800 Subject: [PATCH] feat: prune-only transfer of parent filters across HashJoinExec keys for left, right and mark joins --- .../physical_optimizer/filter_pushdown.rs | 621 +++++++++++++++++- .../physical-plan/src/joins/hash_join/exec.rs | 141 ++-- .../join_dynamic_filter_transfer.slt | 38 ++ 3 files changed, 761 insertions(+), 39 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index ddd4bb4d271aa..2c40345c5101b 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -316,7 +316,8 @@ async fn test_static_filter_pushdown_through_hash_join() { ); let join_schema = join.schema(); - // Filter on build side column (preserved): should be pushed down + // Filter on the build side key (preserved): pushed down, and transferred + // across the join keys to prune the probe side too let left_filter = col_lit_predicate("a", "aa", &join_schema); // Filter on probe side column (not preserved): should NOT be pushed down let right_filter = col_lit_predicate("e", "ba", &join_schema); @@ -340,7 +341,7 @@ async fn test_static_filter_pushdown_through_hash_join() { - FilterExec: e@4 = ba - HashJoinExec: mode=Partitioned, join_type=Left, on=[(a@0, d@0)] - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=a@0 = aa - - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[d, e, f], file_type=test, pushdown_supported=true + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[d, e, f], file_type=test, pushdown_supported=true, predicate=d@0 = aa " ); } @@ -2861,6 +2862,622 @@ async fn test_hashjoin_dynamic_filter_transferred_through_nested_join() { ); } +// ==== Prune-only key transfer: Left, Right, LeftMark and RightMark joins ==== +// +// These joins also emit unmatched rows of their preserved side, so a filter +// over the preserved side's key, transferred to the other side, may only prune +// that side's input. It never makes the filter hold for the join's output. + +/// The side of a [`prune_only_join`] whose rows the join type preserves. +fn prune_only_preserved_side(join_type: JoinType) -> datafusion_common::JoinSide { + match join_type { + JoinType::Left | JoinType::LeftMark | JoinType::LeftAnti => { + datafusion_common::JoinSide::Left + } + JoinType::Right | JoinType::RightMark | JoinType::RightAnti => { + datafusion_common::JoinSide::Right + } + other => panic!("{other} has no single preserved side"), + } +} + +/// Options of a [`prune_only_join`]. +#[derive(Clone, Copy)] +struct PruneOnlyJoin { + join_type: JoinType, + null_equality: datafusion_common::NullEquality, + null_aware: bool, + /// Whether the left / right scan accepts pushed down filters. + left_support: bool, + right_support: bool, + /// Matching pairs must also satisfy `rv != 'r2'`. + join_filter: bool, +} + +impl PruneOnlyJoin { + fn new(join_type: JoinType) -> Self { + Self { + join_type, + null_equality: datafusion_common::NullEquality::NullEqualsNothing, + null_aware: false, + left_support: false, + right_support: false, + join_filter: false, + } + } + + fn with_support(mut self, left_support: bool, right_support: bool) -> Self { + self.left_support = left_support; + self.right_support = right_support; + self + } +} + +/// `left(lk, lv) JOIN right(rk, rv) ON lk = rk` over NULL, duplicate and +/// one-sided keys. +fn prune_only_join(options: PruneOnlyJoin) -> Arc { + let PruneOnlyJoin { + join_type, + null_equality, + null_aware, + left_support, + right_support, + join_filter, + } = options; + use datafusion_common::JoinSide; + use datafusion_physical_plan::joins::utils::{ColumnIndex, JoinFilter}; + + let left_schema = Arc::new(Schema::new(vec![ + Field::new("lk", DataType::Utf8, true), + Field::new("lv", DataType::Utf8, false), + ])); + let left_scan = TestScanBuilder::new(Arc::clone(&left_schema)) + .with_support(left_support) + .with_batches(vec![ + record_batch!( + ( + "lk", + Utf8, + [Some("aa"), Some("aa"), Some("bb"), None, Some("cc")] + ), + ("lv", Utf8, ["l1", "l2", "l3", "l4", "l5"]) + ) + .unwrap(), + ]) + .build(); + + let right_schema = Arc::new(Schema::new(vec![ + Field::new("rk", DataType::Utf8, true), + Field::new("rv", DataType::Utf8, false), + ])); + let right_scan = TestScanBuilder::new(Arc::clone(&right_schema)) + .with_support(right_support) + .with_batches(vec![ + record_batch!( + ( + "rk", + Utf8, + [Some("aa"), Some("bb"), Some("bb"), None, Some("dd"), None] + ), + ("rv", Utf8, ["r1", "r2", "r3", "r4", "r5", "r6"]) + ) + .unwrap(), + ]) + .build(); + + let filter = join_filter.then(|| { + let intermediate = + Arc::new(Schema::new(vec![Field::new("rv", DataType::Utf8, false)])); + JoinFilter::new( + Arc::new(BinaryExpr::new( + col("rv", &intermediate).unwrap(), + Operator::NotEq, + Arc::new(Literal::new(ScalarValue::from("r2"))), + )), + vec![ColumnIndex { + index: 1, + side: JoinSide::Right, + }], + intermediate, + ) + }); + + Arc::new( + HashJoinExec::try_new( + left_scan, + right_scan, + vec![( + col("lk", &left_schema).unwrap(), + col("rk", &right_schema).unwrap(), + )], + filter, + &join_type, + None, + PartitionMode::CollectLeft, + null_equality, + null_aware, + ) + .unwrap(), + ) +} + +/// Runs `FilterPushdown` with row-level pushdown on and collects the rows, +/// one formatted string per row, sorted. +async fn prune_only_run( + plan: Arc, +) -> (Arc, Vec) { + let mut config = ConfigOptions::default(); + config.execution.parquet.pushdown_filters = true; + let optimized = FilterPushdown::new().optimize(plan, &config).unwrap(); + let session_ctx = SessionContext::new(); + session_ctx.register_object_store( + ObjectStoreUrl::parse("test://").unwrap().as_ref(), + Arc::new(InMemory::new()), + ); + let batches = collect(Arc::clone(&optimized), session_ctx.task_ctx()) + .await + .unwrap(); + let formatted = pretty_format_batches(&batches).unwrap().to_string(); + let mut rows: Vec = formatted + .lines() + .filter(|line| line.starts_with('|')) + .skip(1) // header + .map(str::to_string) + .collect(); + rows.sort(); + (optimized, rows) +} + +/// The `predicate=` of each scan in `plan`, left scan first. +fn scan_predicates(plan: &Arc) -> Vec> { + format_plan_for_test(plan) + .lines() + .filter(|line| line.contains("DataSourceExec")) + .map(|line| { + line.split_once("predicate=") + .map(|(_, predicate)| predicate.to_string()) + }) + .collect() +} + +/// Predicates over the preserved side's key `k` that behave differently on +/// NULL, which is where a transferred copy could go wrong. +fn prune_only_predicates( + key: &str, + schema: &Schema, +) -> Vec<(&'static str, Arc)> { + use datafusion_physical_expr::expressions::NotExpr; + let eq = || col_lit_predicate(key, "aa", schema); + let is_null = + || Arc::new(IsNullExpr::new(col(key, schema).unwrap())) as Arc; + vec![ + ("k = aa", eq()), + ("k IS NULL", is_null()), + ("NOT (k = aa)", Arc::new(NotExpr::new(eq()))), + ( + "k = aa OR k IS NULL", + Arc::new(BinaryExpr::new(eq(), Operator::Or, is_null())), + ), + ] +} + +/// Whatever the scans accept, the rows are those of the plan in which no scan +/// accepts anything: a transferred copy only prunes rows that cannot pair with +/// a row passing the filter, and never stands in for the filter itself. +#[tokio::test] +async fn test_hashjoin_prune_only_transfer_differential() { + use datafusion_common::{JoinSide, NullEquality}; + + for join_type in [ + JoinType::Left, + JoinType::Right, + JoinType::LeftMark, + JoinType::RightMark, + ] { + let key = match prune_only_preserved_side(join_type) { + JoinSide::Left => "lk", + _ => "rk", + }; + for null_equality in [ + NullEquality::NullEqualsNothing, + NullEquality::NullEqualsNull, + ] { + for join_filter in [false, true] { + let options = PruneOnlyJoin { + null_equality, + join_filter, + ..PruneOnlyJoin::new(join_type) + }; + let schema = prune_only_join(options).schema(); + for (name, _) in prune_only_predicates(key, &schema) { + let run = |left_support: bool, right_support: bool| { + let join = prune_only_join( + options.with_support(left_support, right_support), + ); + let predicate = prune_only_predicates(key, &join.schema()) + .into_iter() + .find(|(n, _)| *n == name) + .unwrap() + .1; + prune_only_run(Arc::new( + FilterExec::try_new(predicate, join).unwrap(), + )) + }; + let (_, expected) = run(false, false).await; + for (left_support, right_support) in + [(true, false), (false, true), (true, true)] + { + let (plan, rows) = run(left_support, right_support).await; + assert_eq!( + rows, + expected, + "{join_type} {null_equality:?} join_filter={join_filter} \ + `{name}` left_support={left_support} \ + right_support={right_support}\n{}", + format_plan_for_test(&plan) + ); + } + } + } + } + } +} + +/// The copy reaches the non-preserved scan, and it alone does not remove the +/// filter: only the preserved side's scan accepting it does. +#[tokio::test] +async fn test_hashjoin_prune_only_transfer_keeps_parent_filter() { + use datafusion_common::JoinSide; + + for join_type in [ + JoinType::Left, + JoinType::Right, + JoinType::LeftMark, + JoinType::RightMark, + ] { + let preserved = prune_only_preserved_side(join_type); + let (key, other_key, preserved_idx) = match preserved { + JoinSide::Left => ("lk", "rk", 0), + _ => ("rk", "lk", 1), + }; + let plan = |preserved_support: bool| { + let (left_support, right_support) = match preserved { + JoinSide::Left => (preserved_support, true), + _ => (true, preserved_support), + }; + let join = prune_only_join( + PruneOnlyJoin::new(join_type).with_support(left_support, right_support), + ); + let predicate = col_lit_predicate(key, "aa", &join.schema()); + Arc::new(FilterExec::try_new(predicate, join).unwrap()) + as Arc + }; + + // Only the non-preserved scan accepts filters: it gets the copy, the + // filter stays. + let (optimized, _) = prune_only_run(plan(false)).await; + assert!( + optimized.downcast_ref::().is_some(), + "{join_type}: the transferred copy must not remove the filter\n{}", + format_plan_for_test(&optimized) + ); + let predicates = scan_predicates(&optimized); + assert_eq!(predicates[preserved_idx], None, "{join_type}"); + assert_eq!( + predicates[1 - preserved_idx], + Some(format!("{other_key}@0 = aa")), + "{join_type}" + ); + + // Both scans accept: the preserved side applies the filter exactly, + // so it is gone, and the other side is pruned too. + let (optimized, _) = prune_only_run(plan(true)).await; + assert!( + optimized.downcast_ref::().is_some(), + "{join_type}: the preserved side accepted the filter\n{}", + format_plan_for_test(&optimized) + ); + let predicates = scan_predicates(&optimized); + assert_eq!( + predicates[preserved_idx], + Some(format!("{key}@0 = aa")), + "{join_type}" + ); + assert_eq!( + predicates[1 - preserved_idx], + Some(format!("{other_key}@0 = aa")), + "{join_type}" + ); + } +} + +#[test] +fn test_hashjoin_prune_only_transfer_left_join_plan() { + let join = + prune_only_join(PruneOnlyJoin::new(JoinType::Left).with_support(false, true)); + let predicate = col_lit_predicate("lk", "aa", &join.schema()); + let plan = Arc::new(FilterExec::try_new(predicate, join).unwrap()); + insta::assert_snapshot!( + OptimizationTest::new(plan, FilterPushdown::new(), true), + @r" + OptimizationTest: + input: + - FilterExec: lk@0 = aa + - HashJoinExec: mode=CollectLeft, join_type=Left, on=[(lk@0, rk@0)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[lk, lv], file_type=test, pushdown_supported=false + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[rk, rv], file_type=test, pushdown_supported=true + output: + Ok: + - FilterExec: lk@0 = aa + - HashJoinExec: mode=CollectLeft, join_type=Left, on=[(lk@0, rk@0)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[lk, lv], file_type=test, pushdown_supported=false + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[rk, rv], file_type=test, pushdown_supported=true, predicate=rk@0 = aa + " + ); +} + +/// Nothing is transferred where it would be wrong: from the non-preserved +/// side (its key is also NULL for unmatched rows, so `rk IS NULL` above a left +/// join says nothing about `lk`), for non-key and mark columns, and for full, +/// anti and null-aware joins. +#[tokio::test] +async fn test_hashjoin_prune_only_transfer_negative_cases() { + let is_null = |name: &str, schema: &Schema| { + Arc::new(IsNullExpr::new(col(name, schema).unwrap())) as Arc + }; + type Predicate = Box Arc>; + let cases: Vec<(&str, JoinType, bool, Predicate)> = vec![ + ( + "left join, non-preserved key", + JoinType::Left, + false, + Box::new(move |s| is_null("rk", s)), + ), + ( + "right join, non-preserved key", + JoinType::Right, + false, + Box::new(move |s| is_null("lk", s)), + ), + ( + "left join, non-key column", + JoinType::Left, + false, + Box::new(|s| col_lit_predicate("lv", "l1", s)), + ), + ( + "left mark join, mark column", + JoinType::LeftMark, + false, + Box::new(|s| col_lit_predicate("mark", true, s)), + ), + ( + "full join", + JoinType::Full, + false, + Box::new(|s| col_lit_predicate("lk", "aa", s)), + ), + ( + "left anti join", + JoinType::LeftAnti, + false, + Box::new(|s| col_lit_predicate("lk", "aa", s)), + ), + ( + "right anti join", + JoinType::RightAnti, + false, + Box::new(|s| col_lit_predicate("rk", "aa", s)), + ), + ( + "null-aware left mark join", + JoinType::LeftMark, + true, + Box::new(|s| col_lit_predicate("lk", "aa", s)), + ), + ( + "null-aware left anti join", + JoinType::LeftAnti, + true, + Box::new(|s| col_lit_predicate("lk", "aa", s)), + ), + ]; + + for (name, join_type, null_aware, predicate) in cases { + let run = |support: bool| { + let join = prune_only_join( + PruneOnlyJoin { + null_aware, + ..PruneOnlyJoin::new(join_type) + } + .with_support(support, support), + ); + let predicate = predicate(&join.schema()); + prune_only_run(Arc::new(FilterExec::try_new(predicate, join).unwrap())) + }; + let (_, expected) = run(false).await; + let (optimized, rows) = run(true).await; + assert_eq!( + rows, + expected, + "{name}\n{}", + format_plan_for_test(&optimized) + ); + + // At most one scan holds the predicate: the side that owns its + // columns, never a transferred copy on the other side. + let holders = scan_predicates(&optimized) + .iter() + .filter(|predicate| predicate.is_some()) + .count(); + assert!( + holders <= 1, + "{name}: unexpected transfer\n{}", + format_plan_for_test(&optimized) + ); + } +} + +/// With a `fetch` on the join the rows are not unique, but the transferred +/// copy never adds output rows, so every row still comes from the join +/// without a fetch. +#[tokio::test] +async fn test_hashjoin_prune_only_transfer_with_fetch() { + use datafusion_common::JoinSide; + + for join_type in [JoinType::Left, JoinType::Right] { + let key = match prune_only_preserved_side(join_type) { + JoinSide::Left => "lk", + _ => "rk", + }; + let plan = |support: bool, fetch: Option| { + // Only the side that is not preserved accepts filters. + let (left_support, right_support) = match prune_only_preserved_side(join_type) + { + JoinSide::Left => (false, support), + _ => (support, false), + }; + let join = prune_only_join( + PruneOnlyJoin::new(join_type).with_support(left_support, right_support), + ); + let predicate = col_lit_predicate(key, "aa", &join.schema()); + let join = match fetch { + Some(_) => join.with_fetch(fetch).unwrap(), + None => join as Arc, + }; + Arc::new(FilterExec::try_new(predicate, join).unwrap()) + as Arc + }; + let (_, all_rows) = prune_only_run(plan(false, None)).await; + for fetch in 1..=4 { + let (optimized, rows) = prune_only_run(plan(true, Some(fetch))).await; + let mut remaining = all_rows.clone(); + for row in &rows { + let position = remaining.iter().position(|r| r == row); + assert!( + position.is_some(), + "{join_type} fetch={fetch}: `{row}` is not a row of the join\n{}", + format_plan_for_test(&optimized) + ); + remaining.remove(position.unwrap()); + } + } + } +} + +/// An upper join's dynamic filter over the preserved key of a left join below +/// it prunes that join's other input as well. +/// +/// The preserved scan does not accept filters, so the left join's own dynamic +/// filter still holds every `mid` key: the rows the bottom scan drops are +/// dropped by the transferred filter alone. `ab` has no match below, so its +/// row is NULL-extended with or without the pruning. +#[tokio::test] +async fn test_hashjoin_dynamic_filter_prune_only_through_left_join() { + let top_schema = Arc::new(Schema::new(vec![Field::new("t", DataType::Utf8, false)])); + let top_scan = TestScanBuilder::new(Arc::clone(&top_schema)) + .with_support(true) + .with_batches(vec![record_batch!(("t", Utf8, ["aa", "ab"])).unwrap()]) + .build(); + + let mid_schema = Arc::new(Schema::new(vec![ + Field::new("m", DataType::Utf8, false), + Field::new("c", DataType::Float64, false), + ])); + let mid_scan = TestScanBuilder::new(Arc::clone(&mid_schema)) + .with_support(false) + .with_batches(vec![ + record_batch!( + ("m", Utf8, ["aa", "ab", "ac", "ad"]), + ("c", Float64, [1.0, 2.0, 3.0, 4.0]) + ) + .unwrap(), + ]) + .build(); + + let bottom_schema = Arc::new(Schema::new(vec![ + Field::new("x", DataType::Utf8, false), + Field::new("e", DataType::Float64, false), + ])); + let bottom_scan = TestScanBuilder::new(Arc::clone(&bottom_schema)) + .with_support(true) + .with_batches(vec![ + record_batch!( + ("x", Utf8, ["aa", "ac", "ad"]), + ("e", Float64, [1.0, 3.0, 4.0]) + ) + .unwrap(), + ]) + .build(); + + let lower_join = Arc::new( + HashJoinExec::try_new( + mid_scan, + Arc::clone(&bottom_scan), + vec![( + col("m", &mid_schema).unwrap(), + col("x", &bottom_schema).unwrap(), + )], + None, + &JoinType::Left, + None, + PartitionMode::CollectLeft, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(), + ); + let lower_schema = lower_join.schema(); + let upper_join = Arc::new( + HashJoinExec::try_new( + top_scan, + lower_join, + vec![( + col("t", &top_schema).unwrap(), + col("m", &lower_schema).unwrap(), + )], + None, + &JoinType::Inner, + None, + PartitionMode::CollectLeft, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(), + ) as Arc; + + let mut config = ConfigOptions::default(); + config.execution.parquet.pushdown_filters = true; + config.optimizer.enable_dynamic_filter_pushdown = true; + let (plan, batches) = optimize_and_collect_pushdown_plan(upper_join, config).await; + + insta::assert_snapshot!( + format_plan_for_test(&plan), + @r" + - HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(t@0, m@0)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[t], file_type=test, pushdown_supported=true + - HashJoinExec: mode=CollectLeft, join_type=Left, on=[(m@0, x@0)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[m, c], file_type=test, pushdown_supported=false + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[x, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ x@0 >= aa AND x@0 <= ad AND x@0 IN (SET) ([aa, ab, ac, ad]) ] AND DynamicFilter [ x@0 >= aa AND x@0 <= ab AND x@0 IN (SET) ([aa, ab]) ] + " + ); + + // The left join's own filter lets all three `bottom` rows through; the + // transferred filter from `top` keeps only `aa`. + assert_eq!(bottom_scan.metrics().unwrap().output_rows().unwrap(), 1); + + #[rustfmt::skip] + let expected = [ + "+----+----+-----+----+-----+", + "| t | m | c | x | e |", + "+----+----+-----+----+-----+", + "| aa | aa | 1.0 | aa | 1.0 |", + "| ab | ab | 2.0 | | |", + "+----+----+-----+----+-----+", + ]; + assert_batches_sorted_eq!(expected, &batches); +} + #[test] fn test_filter_pushdown_through_union() { let scan1 = TestScanBuilder::new(schema()).with_support(true).build(); diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 9858a4c06cd4a..3aa2afc6f321d 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -980,20 +980,44 @@ impl HashJoinExec { Arc::new(DynamicFilterPhysicalExpr::new(right_keys, lit(true))) } - /// Join types whose output rows all carry a matching key on both sides. + /// How this join may transfer a parent filter over one side's join keys to + /// the other side's input, see [`KeyTransfer`]. /// - /// For these a parent filter over one side's join keys can be transferred - /// to the other side's input: an input row that fails the transferred - /// filter can only pair with rows that fail the original, so pruning it - /// changes nothing, and once the transferred filter is applied exactly on - /// one side every output row satisfies the original. Outer, anti and mark - /// joins also emit unmatched rows, whose key on the other side is absent, - /// so the transferred filter is not exact for them. - fn supports_key_transfer(join_type: JoinType) -> bool { - matches!( - join_type, - JoinType::Inner | JoinType::LeftSemi | JoinType::RightSemi - ) + /// The common ground: an input row that fails the transferred filter can + /// only pair with rows that fail the original (their keys are equal, also + /// under [`NullEquality::NullEqualsNull`], and a join filter only removes + /// pairs), so every row that passes the original keeps the same matches. + fn key_transfer(&self) -> KeyTransfer { + match self.join_type { + // Every output row carries a matching key on both sides, so once + // the transferred filter is applied exactly on one side every + // output row satisfies the original. + JoinType::Inner | JoinType::LeftSemi | JoinType::RightSemi => { + KeyTransfer::Exact + } + // A NULL probe key decides a null-aware join's whole result, and a + // transferred filter is not true for NULL, so it would prune it. + _ if self.null_aware => KeyTransfer::None, + // The preserved side also emits unmatched rows. Pruning the other + // side can turn a row that fails the filter from matched into + // unmatched (NULL-extended, or mark = false), but every row + // derived from it still fails the filter, which stays above the + // join. It never adds output rows, so a `fetch` on this join or + // above it cannot lose rows that pass the filter. + JoinType::Left | JoinType::LeftMark => { + KeyTransfer::PruneOnly { preserved_child: 0 } + } + JoinType::Right | JoinType::RightMark => { + KeyTransfer::PruneOnly { preserved_child: 1 } + } + // Neither side of a full join is preserved. An anti join would + // emit *more* rows: every preserved row whose matches were pruned + // fails the filter, yet it can fill a `fetch` between this join + // and the filter and displace rows that pass. + JoinType::Full | JoinType::LeftAnti | JoinType::RightAnti => { + KeyTransfer::None + } + } } /// Maps each output column that is a plain `Column` join key on one side @@ -1035,9 +1059,7 @@ impl HashJoinExec { .find(|(_, right_key)| is_column_at(right_key, ci.index)) .map(|(left_key, _)| left_key), ), - // Only mark joins produce mark columns, and - // `supports_key_transfer` excludes them; this arm is here for - // exhaustiveness. + // A mark column is not a key of either side. JoinSide::None => continue, }; if let Some(other_key) = other_key { @@ -1922,26 +1944,37 @@ impl ExecutionPlan for HashJoinExec { // side too, so it is also pushed there, rewritten over that side's key // expressions. This is how a dynamic filter from a join above reaches // the scans on both sides of this join, and how a semi join prunes its - // non-output side. Like the plain column routing, a transfer only - // targets a side that `lr_is_preserved` permits. - let (to_right, to_left) = if Self::supports_key_transfer(self.join_type) { - self.key_transfer_maps(&column_indices) - } else { - Default::default() + // non-output side. A side that `lr_is_preserved` does not permit + // receives nothing but such transferred filters, and only from the + // preserved side: a filter over the non-preserved side's key also + // sees the NULLs of unmatched rows (`r.k IS NULL` above a left join), + // which the preserved side's key does not have. + let (to_right, to_left) = match self.key_transfer() { + KeyTransfer::Exact => self.key_transfer_maps(&column_indices), + KeyTransfer::PruneOnly { preserved_child } => { + let (to_right, to_left) = self.key_transfer_maps(&column_indices); + if preserved_child == 0 { + (to_right, HashMap::new()) + } else { + (HashMap::new(), to_left) + } + } + KeyTransfer::None => Default::default(), }; let describe_child = |preserved: bool, column_mapping: HashMap, key_map: &KeyTransferMap, child: &Arc| -> Result { - if !preserved { - return Ok(ChildFilterDescription::all_unsupported(&parent_filters)); - } - let mut description = ChildFilterDescription::from_child_with_column_mapping( - &parent_filters, - column_mapping, - child, - )?; + let mut description = if preserved { + ChildFilterDescription::from_child_with_column_mapping( + &parent_filters, + column_mapping, + child, + )? + } else { + ChildFilterDescription::all_unsupported(&parent_filters) + }; transfer_key_filters(&parent_filters, key_map, &mut description)?; Ok(description) }; @@ -1977,7 +2010,23 @@ impl ExecutionPlan for HashJoinExec { child_pushdown_result: ChildPushdownResult, _config: &ConfigOptions, ) -> Result>> { - let mut result = FilterPushdownPropagation::if_any(child_pushdown_result.clone()); + let mut result = match self.key_transfer() { + // A copy transferred to the non-preserved side only prunes that + // side's input. It never makes the filter hold for this join's + // output, so only the preserved child's answer counts. + KeyTransfer::PruneOnly { preserved_child } => { + FilterPushdownPropagation::with_parent_pushdown_result( + child_pushdown_result + .parent_filters + .iter() + .map(|filter| filter.child_results[preserved_child]) + .collect(), + ) + } + KeyTransfer::Exact | KeyTransfer::None => { + 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 @@ -2560,6 +2609,21 @@ mod proto_tests { } } +/// How a [`HashJoinExec`] may transfer a parent filter over one side's join +/// keys to the other side, see [`HashJoinExec::key_transfer`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum KeyTransfer { + /// The transferred filter is as good as the original: if either child + /// applies its copy exactly, the filter holds for every output row. + Exact, + /// Only from the preserved child to the other one, and only to prune that + /// child's input: whether the filter holds for the output is the preserved + /// child's answer alone. + PruneOnly { preserved_child: usize }, + /// No transfer. + None, +} + /// Output column index of a join, mapped to the equivalent join-key expression /// on the other side of the join (in that side's input schema). type KeyTransferMap = HashMap; @@ -2574,9 +2638,8 @@ fn is_column_at(expr: &PhysicalExprRef, index: usize) -> bool { /// /// `key_map` only holds columns of the other side, so a filter it rewrites is /// one the plain column analysis marked unsupported for `child`. A filter that -/// references any other column is left as that analysis routed it. A filter -/// with no columns comes back unchanged and was already accepted, so -/// rewriting it is a no-op. +/// references any other column, or no column at all, is left as that analysis +/// routed it. fn transfer_key_filters( parent_filters: &[Arc], key_map: &KeyTransferMap, @@ -2594,7 +2657,7 @@ fn transfer_key_filters( } /// Rewrites `filter` over the other side's join keys, or returns `None` when -/// it references a column that is not a transferable key. +/// it references a column that is not a transferable key, or no column. /// /// A [`DynamicFilterPhysicalExpr`] comes out as a view sharing the original's /// state with its key columns remapped, so it keeps tracking the build side. @@ -2603,10 +2666,12 @@ fn transfer_filter_across_keys( key_map: &KeyTransferMap, ) -> Result>> { let mut all_keys = true; + let mut any_key = false; let transformed = Arc::clone(filter).transform_down(|expr| { let Some(column) = expr.downcast_ref::() else { return Ok(Transformed::no(expr)); }; + any_key = true; match key_map.get(&column.index()) { // The replacement is in the other side's input schema, so its // columns are not output indices of this join: `Jump` over it. @@ -2624,14 +2689,16 @@ fn transfer_filter_across_keys( } } })?; - Ok(all_keys.then_some(transformed.data)) + Ok((all_keys && any_key).then_some(transformed.data)) } /// Determines which sides of a join are "preserved" for filter pushdown. /// /// A preserved side means filters on that side's columns can be safely pushed /// below the join. This mostly mirrors the logical optimizer's `lr_is_preserved`; -/// semi joins additionally allow join-key filters on the non-output side. +/// semi joins additionally allow join-key filters on the non-output side. A +/// side that is not preserved can still receive a filter transferred from the +/// preserved side's join keys, see [`HashJoinExec::key_transfer`]. fn lr_is_preserved(join_type: JoinType) -> (bool, bool) { match join_type { JoinType::Inner => (true, true), diff --git a/datafusion/sqllogictest/test_files/join_dynamic_filter_transfer.slt b/datafusion/sqllogictest/test_files/join_dynamic_filter_transfer.slt index 04d3e4374f86a..a823a3eb4c770 100644 --- a/datafusion/sqllogictest/test_files/join_dynamic_filter_transfer.slt +++ b/datafusion/sqllogictest/test_files/join_dynamic_filter_transfer.slt @@ -116,6 +116,44 @@ JOIN dim d ON d.d_key = m.m_key; one 10 100 three 30 300 +# With a left join below, `mid` is the preserved side and `fact` is not. The +# `dim` filter over `m_key` still reaches the `fact` scan: there the copy only +# prunes `fact` rows that cannot match a `mid` row the filter keeps, it never +# stands in for the filter itself. +query TT +EXPLAIN SELECT d.d_val, m.m_c, f.f_e +FROM mid m +LEFT JOIN fact f ON m.m_key = f.f_key +JOIN dim d ON d.d_key = m.m_key; +---- +logical_plan +01)Projection: d.d_val, m.m_c, f.f_e +02)--Inner Join: m.m_key = d.d_key +03)----Projection: m.m_key, m.m_c, f.f_e +04)------Left Join: m.m_key = f.f_key +05)--------SubqueryAlias: m +06)----------TableScan: mid projection=[m_key, m_c] +07)--------SubqueryAlias: f +08)----------TableScan: fact projection=[f_key, f_e] +09)----SubqueryAlias: d +10)------TableScan: dim projection=[d_key, d_val] +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_key@0, m_key@0)], projection=[d_val@1, m_c@3, f_e@4] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/join_dynamic_filter_transfer/dim.parquet]]}, projection=[d_key, d_val], file_type=parquet +03)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +04)----HashJoinExec: mode=CollectLeft, join_type=Left, on=[(m_key@0, f_key@0)], projection=[m_key@0, m_c@1, f_e@3] +05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/join_dynamic_filter_transfer/mid.parquet]]}, projection=[m_key, m_c], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible +06)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/join_dynamic_filter_transfer/fact.parquet]]}, projection=[f_key, f_e], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible + +query TII rowsort +SELECT d.d_val, m.m_c, f.f_e +FROM mid m +LEFT JOIN fact f ON m.m_key = f.f_key +JOIN dim d ON d.d_key = m.m_key; +---- +one 10 100 +three 30 300 + statement ok DROP TABLE dim;