From cc0c910182f49ae8659c759142f20b0a1484f2d7 Mon Sep 17 00:00:00 2001 From: Jay Zhan Date: Sun, 13 Sep 2026 11:54:58 +0800 Subject: [PATCH 1/2] feat: transfer parent filters across HashJoinExec equi-join keys for inner and semi joins --- .../physical_optimizer/filter_pushdown.rs | 291 +++++++++++++++++- .../physical-plan/src/joins/hash_join/exec.rs | 179 ++++++++--- .../join_dynamic_filter_transfer.slt | 135 ++++++++ 3 files changed, 562 insertions(+), 43 deletions(-) create mode 100644 datafusion/sqllogictest/test_files/join_dynamic_filter_transfer.slt diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index 26e6e0c74c49d..6cb96d3c5b40a 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -287,7 +287,7 @@ async fn test_static_filter_pushdown_through_hash_join() { - FilterExec: a@0 = d@3 - HashJoinExec: mode=Partitioned, join_type=Inner, 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, predicate=e@1 = ba + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[d, e, f], file_type=test, pushdown_supported=true, predicate=d@0 = aa AND e@1 = ba " ); @@ -1596,7 +1596,7 @@ fn test_hashjoin_parent_filter_pushdown_same_column_names() { Ok: - HashJoinExec: mode=Partitioned, join_type=Inner, on=[(id@0, id@0)] - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[id, build_val], file_type=test, pushdown_supported=true, predicate=id@0 = aa - - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[id, probe_val], file_type=test, pushdown_supported=true, predicate=probe_val@1 = x + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[id, probe_val], file_type=test, pushdown_supported=true, predicate=id@0 = aa AND probe_val@1 = x " ); } @@ -1754,6 +1754,293 @@ fn test_hashjoin_parent_filter_pushdown_semi_anti_join() { assert_parent_filter_remains(plan); } +/// A parent filter over one side's join keys is transferred to the other side, +/// rewritten over that side's key expressions, even when the key names differ. +/// Filters over non-key columns stay on their own side. +#[test] +fn test_hashjoin_parent_filter_transferred_across_join_keys() { + let build_side_schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Utf8, false), + Field::new("build_val", DataType::Utf8, false), + ])); + let build_scan = TestScanBuilder::new(Arc::clone(&build_side_schema)) + .with_support(true) + .build(); + + let probe_side_schema = Arc::new(Schema::new(vec![ + Field::new("pid", DataType::Utf8, false), + Field::new("probe_val", DataType::Utf8, false), + ])); + let probe_scan = TestScanBuilder::new(Arc::clone(&probe_side_schema)) + .with_support(true) + .build(); + + let on = vec![( + col("id", &build_side_schema).unwrap(), + col("pid", &probe_side_schema).unwrap(), + )]; + let join = Arc::new( + HashJoinExec::try_new( + build_scan, + probe_scan, + on, + None, + &JoinType::Inner, + None, + PartitionMode::Partitioned, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(), + ); + + let join_schema = join.schema(); + + let build_key_filter = col_lit_predicate("id", "aa", &join_schema); + let probe_key_filter = col_lit_predicate("pid", "ab", &join_schema); + let build_val_filter = col_lit_predicate("build_val", "x", &join_schema); + + let filter = + Arc::new(FilterExec::try_new(build_key_filter, Arc::clone(&join) as _).unwrap()); + let filter = Arc::new(FilterExec::try_new(probe_key_filter, filter).unwrap()); + let plan = Arc::new(FilterExec::try_new(build_val_filter, filter).unwrap()) + as Arc; + + insta::assert_snapshot!( + OptimizationTest::new(Arc::clone(&plan), FilterPushdown::new(), true), + @r" + OptimizationTest: + input: + - FilterExec: build_val@1 = x + - FilterExec: pid@2 = ab + - FilterExec: id@0 = aa + - HashJoinExec: mode=Partitioned, join_type=Inner, on=[(id@0, pid@0)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[id, build_val], file_type=test, pushdown_supported=true + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[pid, probe_val], file_type=test, pushdown_supported=true + output: + Ok: + - HashJoinExec: mode=Partitioned, join_type=Inner, on=[(id@0, pid@0)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[id, build_val], file_type=test, pushdown_supported=true, predicate=id@0 = aa AND id@0 = ab AND build_val@1 = x + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[pid, probe_val], file_type=test, pushdown_supported=true, predicate=pid@0 = aa AND pid@0 = ab + " + ); +} + +/// The non-output side of a semi join receives key filters through the same +/// transfer, so differently named keys work too. +#[test] +fn test_hashjoin_parent_filter_transfer_semi_join_different_key_names() { + let left_schema = Arc::new(Schema::new(vec![ + Field::new("k", DataType::Utf8, false), + Field::new("v", DataType::Utf8, false), + ])); + let left_scan = TestScanBuilder::new(Arc::clone(&left_schema)) + .with_support(true) + .build(); + + let right_schema = Arc::new(Schema::new(vec![ + Field::new("w", DataType::Utf8, false), + Field::new("rk", DataType::Utf8, false), + ])); + let right_scan = TestScanBuilder::new(Arc::clone(&right_schema)) + .with_support(true) + .build(); + + let on = vec![( + col("k", &left_schema).unwrap(), + col("rk", &right_schema).unwrap(), + )]; + let join = Arc::new( + HashJoinExec::try_new( + left_scan, + right_scan, + on, + None, + &JoinType::LeftSemi, + None, + PartitionMode::Partitioned, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(), + ); + + let join_schema = join.schema(); + let key_filter = col_lit_predicate("k", "x", &join_schema); + let plan = Arc::new(FilterExec::try_new(key_filter, join).unwrap()) + as Arc; + + insta::assert_snapshot!( + OptimizationTest::new(Arc::clone(&plan), FilterPushdown::new(), true), + @r" + OptimizationTest: + input: + - FilterExec: k@0 = x + - HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(k@0, rk@1)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[k, v], file_type=test, pushdown_supported=true + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[w, rk], file_type=test, pushdown_supported=true + output: + Ok: + - HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(k@0, rk@1)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[k, v], file_type=test, pushdown_supported=true, predicate=k@0 = x + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[w, rk], file_type=test, pushdown_supported=true, predicate=rk@1 = x + " + ); +} + +/// A join's dynamic filter over the build-side key of the join below it is +/// transferred across that join's keys onto its probe-side scan: the probe +/// scan is pruned by the keys of a table it is not joined with. +/// +/// The lower build-side scan does not accept filters, so the lower join's own +/// dynamic filter still holds every `mid` key and prunes nothing: the rows the +/// bottom scan drops are dropped by the transferred filter alone. +#[tokio::test] +async fn test_hashjoin_dynamic_filter_transferred_through_nested_join() { + // Upper build side: the two keys that survive. + 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(); + + // Lower build side: joined with `top` on `m = t`. Rejects pushed filters, + // so all four keys reach the lower join's hash table. + 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(); + + // Lower probe side: joined with `mid` on `x = m`, never directly with `top`. + 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", "ab", "ac", "ad"]), + ("e", Float64, [1.0, 2.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::Inner, + 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(), + ); + let upper_schema = upper_join.schema(); + let plan = Arc::new(SortExec::new( + LexOrdering::new(vec![PhysicalSortExpr::new( + col("x", &upper_schema).unwrap(), + SortOptions::new(false, false), + )]) + .unwrap(), + upper_join, + )) as Arc; + + insta::assert_snapshot!( + OptimizationTest::new(Arc::clone(&plan), FilterPushdown::new_post_optimization(), true), + @r" + OptimizationTest: + input: + - SortExec: expr=[x@3 ASC NULLS LAST], preserve_partitioning=[false] + - 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=Inner, 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 + output: + Ok: + - SortExec: expr=[x@3 ASC NULLS LAST], preserve_partitioning=[false] + - 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=Inner, 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 [ empty ] AND DynamicFilter [ empty ] + " + ); + + 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(plan, config).await; + + // The bottom scan carries the lower join's own filter, which still lists + // all four `mid` keys, and the upper join's filter rewritten over `x`. + insta::assert_snapshot!( + format!("{}", format_plan_for_test(&plan)), + @r" + - SortExec: expr=[x@3 ASC NULLS LAST], preserve_partitioning=[false] + - 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=Inner, 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 lower join's own filter lets all four `bottom` rows through; the + // transferred filter from `top` prunes them to two before the join. + let bottom_scan_metrics = bottom_scan.metrics().unwrap(); + assert_eq!(bottom_scan_metrics.output_rows().unwrap(), 2); + + insta::assert_snapshot!( + format!("{}", pretty_format_batches(&batches).unwrap()), + @r" + +----+----+-----+----+-----+ + | t | m | c | x | e | + +----+----+-----+----+-----+ + | aa | aa | 1.0 | aa | 1.0 | + | ab | ab | 2.0 | ab | 2.0 | + +----+----+-----+----+-----+ + ", + ); +} + #[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 b72e180543f9a..e25b10ca58d6a 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::fmt; use std::mem::size_of; use std::sync::{Arc, OnceLock}; @@ -27,7 +27,7 @@ use crate::execution_plan::{ }; use crate::filter_pushdown::{ ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase, - FilterPushdownPropagation, + FilterPushdownPropagation, PushedDown, PushedDownPredicate, }; use crate::joins::Map; use crate::joins::array_map::ArrayMap; @@ -76,7 +76,7 @@ use arrow::record_batch::RecordBatch; use arrow::util::bit_util; use arrow_schema::{DataType, Schema}; use datafusion_common::config::ConfigOptions; -use datafusion_common::tree_node::TreeNodeRecursion; +use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion}; use datafusion_common::utils::memory::{RecordBatchMemoryCounter, estimate_memory_size}; use datafusion_common::{ JoinSide, JoinType, NullEquality, Result, assert_or_internal_err, internal_err, @@ -979,6 +979,58 @@ impl HashJoinExec { Arc::new(DynamicFilterPhysicalExpr::new(right_keys, lit(true))) } + /// Join types whose output rows all carry a matching key on both sides. + /// + /// 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 + ) + } + + /// Maps each output column that is a plain `Column` join key on one side + /// to the key expression on the other side, as `(to_right, to_left)`. + /// + /// `column_indices` are the (projected) output columns of this join. A key + /// column that appears in several `on` pairs maps to the first of them. + fn key_transfer_maps( + &self, + column_indices: &[ColumnIndex], + ) -> (KeyTransferMap, KeyTransferMap) { + let mut to_right = HashMap::new(); + let mut to_left = HashMap::new(); + for (output_idx, ci) in column_indices.iter().enumerate() { + let (map, other_key) = match ci.side { + JoinSide::Left => ( + &mut to_right, + self.on + .iter() + .find(|(left_key, _)| is_column_at(left_key, ci.index)) + .map(|(_, right_key)| right_key), + ), + JoinSide::Right => ( + &mut to_left, + self.on + .iter() + .find(|(_, right_key)| is_column_at(right_key, ci.index)) + .map(|(left_key, _)| left_key), + ), + JoinSide::None => continue, + }; + if let Some(other_key) = other_key { + map.insert(output_idx, Arc::clone(other_key)); + } + } + (to_right, to_left) + } + fn allow_join_dynamic_filter_pushdown(&self, config: &ConfigOptions) -> bool { let (_, probe_preserved) = self.join_type.on_lr_is_preserved(); if !probe_preserved || !config.optimizer.enable_join_dynamic_filter_pushdown { @@ -1847,43 +1899,7 @@ impl ExecutionPlan for HashJoinExec { }; }); - // For semi joins, filters on output join keys can also be pushed to the - // non-output side: every emitted row has an equal key there. This is not - // true for anti joins, whose emitted rows have no match. - match self.join_type { - JoinType::LeftSemi => { - let left_key_indices: HashSet = self - .on - .iter() - .filter_map(|(left_key, _)| { - left_key.downcast_ref::().map(|c| c.index()) - }) - .collect(); - for (output_idx, ci) in column_indices.iter().enumerate() { - if ci.side == JoinSide::Left && left_key_indices.contains(&ci.index) { - right_allowed.insert(output_idx); - } - } - } - JoinType::RightSemi => { - let right_key_indices: HashSet = self - .on - .iter() - .filter_map(|(_, right_key)| { - right_key.downcast_ref::().map(|c| c.index()) - }) - .collect(); - for (output_idx, ci) in column_indices.iter().enumerate() { - if ci.side == JoinSide::Right && right_key_indices.contains(&ci.index) - { - left_allowed.insert(output_idx); - } - } - } - _ => {} - } - - let left_child = if left_preserved { + let mut left_child = if left_preserved { ChildFilterDescription::from_child_with_allowed_indices( &parent_filters, left_allowed, @@ -1903,6 +1919,18 @@ impl ExecutionPlan for HashJoinExec { ChildFilterDescription::all_unsupported(&parent_filters) }; + // Transfer filters across the equi-join keys: a parent filter over one + // side's join-key columns holds for every matching row of the other + // 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. + if Self::supports_key_transfer(self.join_type) { + let (to_right, to_left) = self.key_transfer_maps(&column_indices); + transfer_key_filters(&parent_filters, &to_right, &mut right_child)?; + transfer_key_filters(&parent_filters, &to_left, &mut left_child)?; + } + // Add dynamic filters in Post phase if enabled. Skip when this join // already carries a dynamic filter from a previous pass — the shared // `Arc` is still wired into the probe-side @@ -2512,6 +2540,74 @@ mod proto_tests { } } +/// 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; + +fn is_column_at(expr: &PhysicalExprRef, index: usize) -> bool { + expr.downcast_ref::() + .is_some_and(|column| column.index() == index) +} + +/// Marks every parent filter whose columns are all join keys in `key_map` as +/// supported for `child`, rewritten over the other side's key expressions. +/// +/// Filters `child` already accepts directly are left alone, as are filters +/// that reference a non-key column or no column at all: the former cannot be +/// expressed on the other side, the latter were already routed by the plain +/// column analysis. +fn transfer_key_filters( + parent_filters: &[Arc], + key_map: &KeyTransferMap, + child: &mut ChildFilterDescription, +) -> Result<()> { + if key_map.is_empty() { + return Ok(()); + } + for (filter, pushed) in parent_filters.iter().zip(child.parent_filters.iter_mut()) { + if matches!(pushed.discriminant, PushedDown::Yes) { + continue; + } + if let Some(transferred) = transfer_filter_across_keys(filter, key_map)? { + *pushed = PushedDownPredicate::supported(transferred); + } + } + Ok(()) +} + +/// Rewrites `filter` over the other side's join keys, or returns `None` when +/// 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. +fn transfer_filter_across_keys( + filter: &Arc, + key_map: &KeyTransferMap, +) -> Result>> { + let mut all_keys = true; + let mut any_column = false; + let transformed = Arc::clone(filter).transform_down(|expr| { + let Some(column) = expr.downcast_ref::() else { + return Ok(Transformed::no(expr)); + }; + any_column = true; + match key_map.get(&column.index()) { + // The replacement is already in the other side's schema: do not + // descend into it, its columns are not indices of this join. + Some(other_key) => Ok(Transformed::new( + Arc::clone(other_key), + true, + TreeNodeRecursion::Jump, + )), + None => { + all_keys = false; + Ok(Transformed::new(expr, false, TreeNodeRecursion::Stop)) + } + } + })?; + Ok((any_column && all_keys).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 @@ -2523,7 +2619,8 @@ fn lr_is_preserved(join_type: JoinType) -> (bool, bool) { JoinType::Left => (true, false), JoinType::Right => (false, true), JoinType::Full => (false, false), - // Callers restrict the non-output side of semi joins to join-key columns. + // The non-output side of a semi join only receives filters transferred + // across the join keys, see `HashJoinExec::supports_key_transfer`. JoinType::LeftSemi | JoinType::RightSemi => (true, true), JoinType::LeftAnti | JoinType::LeftMark => (true, false), JoinType::RightAnti | JoinType::RightMark => (false, true), diff --git a/datafusion/sqllogictest/test_files/join_dynamic_filter_transfer.slt b/datafusion/sqllogictest/test_files/join_dynamic_filter_transfer.slt new file mode 100644 index 0000000000000..04d3e4374f86a --- /dev/null +++ b/datafusion/sqllogictest/test_files/join_dynamic_filter_transfer.slt @@ -0,0 +1,135 @@ +# 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. + +# A hash join's dynamic filter is transferred across the equi-join keys of the +# joins below it: a filter over one side's join key also holds for the other +# side's key, so it reaches the scans on both sides of that join. +# +# Here `dim` is the build side of the upper join and its keys filter `mid` +# directly. `fact` is only joined with `mid`, but through `mid.m_key = fact.f_key` +# the `dim` filter is rewritten over `f_key` and reaches the `fact` scan too. + +statement ok +CREATE TABLE dim_src(d_key INT, d_val VARCHAR) AS VALUES +(1, 'one'), +(3, 'three'); + +statement ok +CREATE TABLE mid_src(m_key INT, m_c INT) AS VALUES +(1, 10), +(2, 20), +(3, 30), +(4, 40), +(5, 50); + +statement ok +CREATE TABLE fact_src(f_key INT, f_e INT) AS VALUES +(1, 100), +(2, 200), +(3, 300), +(4, 400), +(5, 500), +(6, 600), +(7, 700), +(8, 800); + +query I +COPY dim_src TO 'test_files/scratch/join_dynamic_filter_transfer/dim.parquet' STORED AS PARQUET; +---- +2 + +query I +COPY mid_src TO 'test_files/scratch/join_dynamic_filter_transfer/mid.parquet' STORED AS PARQUET; +---- +5 + +query I +COPY fact_src TO 'test_files/scratch/join_dynamic_filter_transfer/fact.parquet' STORED AS PARQUET; +---- +8 + +statement ok +CREATE EXTERNAL TABLE dim(d_key INT, d_val VARCHAR) +STORED AS PARQUET +LOCATION 'test_files/scratch/join_dynamic_filter_transfer/dim.parquet'; + +statement ok +CREATE EXTERNAL TABLE mid(m_key INT, m_c INT) +STORED AS PARQUET +LOCATION 'test_files/scratch/join_dynamic_filter_transfer/mid.parquet'; + +statement ok +CREATE EXTERNAL TABLE fact(f_key INT, f_e INT) +STORED AS PARQUET +LOCATION 'test_files/scratch/join_dynamic_filter_transfer/fact.parquet'; + +# The `fact` scan carries two dynamic filters: the lower join's own filter, +# built from `mid`, and the upper join's filter from `dim` transferred over +# `f_key`. The transferred one arrives with `dim`'s bounds and IN list even +# when `mid` is large enough that its own filter is a hash-table lookup, which +# cannot prune row groups. +query TT +EXPLAIN SELECT d.d_val, m.m_c, f.f_e +FROM mid m +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)------Inner 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=Inner, 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 +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; + +statement ok +DROP TABLE mid; + +statement ok +DROP TABLE fact; + +statement ok +DROP TABLE dim_src; + +statement ok +DROP TABLE mid_src; + +statement ok +DROP TABLE fact_src; From e62f53f99757f19c8cde8fd201c9feeeacf34245 Mon Sep 17 00:00:00 2001 From: Jay Zhan Date: Mon, 14 Sep 2026 21:21:04 +0800 Subject: [PATCH 2/2] test: cover RightSemi, first on pair, CAST key and shadowed key name transfers; gate transfer on lr_is_preserved --- .../physical_optimizer/filter_pushdown.rs | 298 +++++++++++++++++- .../physical-plan/src/joins/hash_join/exec.rs | 104 +++--- 2 files changed, 359 insertions(+), 43 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index 6cb96d3c5b40a..6cdf17df501d6 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -51,7 +51,7 @@ use datafusion_functions_aggregate::{ }; use datafusion_physical_expr::{ LexOrdering, PhysicalSortExpr, - expressions::{DynamicFilterPhysicalExpr, col}, + expressions::{DynamicFilterPhysicalExpr, cast, col}, utils::conjunction, }; use datafusion_physical_expr::{ @@ -1888,6 +1888,302 @@ fn test_hashjoin_parent_filter_transfer_semi_join_different_key_names() { ); } +/// `RightSemi` variant of the test above: the join outputs only the right +/// side, so a filter over its key reaches the left scan only by transfer. +#[test] +fn test_hashjoin_parent_filter_transfer_right_semi_join_different_key_names() { + let left_schema = Arc::new(Schema::new(vec![ + Field::new("k", DataType::Utf8, false), + Field::new("v", DataType::Utf8, false), + ])); + let left_scan = TestScanBuilder::new(Arc::clone(&left_schema)) + .with_support(true) + .build(); + + let right_schema = Arc::new(Schema::new(vec![ + Field::new("w", DataType::Utf8, false), + Field::new("rk", DataType::Utf8, false), + ])); + let right_scan = TestScanBuilder::new(Arc::clone(&right_schema)) + .with_support(true) + .build(); + + let on = vec![( + col("k", &left_schema).unwrap(), + col("rk", &right_schema).unwrap(), + )]; + let join = Arc::new( + HashJoinExec::try_new( + left_scan, + right_scan, + on, + None, + &JoinType::RightSemi, + None, + PartitionMode::Partitioned, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(), + ); + + let join_schema = join.schema(); + let key_filter = col_lit_predicate("rk", "x", &join_schema); + let plan = Arc::new(FilterExec::try_new(key_filter, join).unwrap()) + as Arc; + + insta::assert_snapshot!( + OptimizationTest::new(Arc::clone(&plan), FilterPushdown::new(), true), + @r" + OptimizationTest: + input: + - FilterExec: rk@1 = x + - HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(k@0, rk@1)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[k, v], file_type=test, pushdown_supported=true + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[w, rk], file_type=test, pushdown_supported=true + output: + Ok: + - HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(k@0, rk@1)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[k, v], file_type=test, pushdown_supported=true, predicate=k@0 = x + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[w, rk], file_type=test, pushdown_supported=true, predicate=rk@1 = x + " + ); +} + +/// Regression test for the name-based semi-join routing this transfer +/// replaced: the non-output side has a column with the key's *name* that is +/// not the key. The old code pushed the key filter to that column, so the +/// scan was pruned by the wrong column and the parent filter dropped. The +/// transfer rewrites the filter over the actual key instead. +#[test] +fn test_hashjoin_parent_filter_transfer_semi_join_key_name_shadowed_by_non_key() { + // LeftSemi: the right side has a non-key `k` at index 2, the key is `j`. + let left_schema = Arc::new(Schema::new(vec![ + Field::new("k", DataType::Utf8, false), + Field::new("v", DataType::Utf8, false), + ])); + let right_schema = Arc::new(Schema::new(vec![ + Field::new("j", DataType::Utf8, false), + Field::new("w", DataType::Utf8, false), + Field::new("k", DataType::Utf8, false), + ])); + let on = vec![( + col("k", &left_schema).unwrap(), + col("j", &right_schema).unwrap(), + )]; + let join = Arc::new( + HashJoinExec::try_new( + TestScanBuilder::new(Arc::clone(&left_schema)) + .with_support(true) + .build(), + TestScanBuilder::new(Arc::clone(&right_schema)) + .with_support(true) + .build(), + on, + None, + &JoinType::LeftSemi, + None, + PartitionMode::Partitioned, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(), + ); + let key_filter = col_lit_predicate("k", "x", &join.schema()); + let plan = Arc::new(FilterExec::try_new(key_filter, join).unwrap()) + as Arc; + + insta::assert_snapshot!( + OptimizationTest::new(Arc::clone(&plan), FilterPushdown::new(), true), + @r" + OptimizationTest: + input: + - FilterExec: k@0 = x + - HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(k@0, j@0)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[k, v], file_type=test, pushdown_supported=true + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[j, w, k], file_type=test, pushdown_supported=true + output: + Ok: + - HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(k@0, j@0)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[k, v], file_type=test, pushdown_supported=true, predicate=k@0 = x + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[j, w, k], file_type=test, pushdown_supported=true, predicate=j@0 = x + " + ); + + // RightSemi mirror image: the left side has a non-key `k` at index 2. + let on = vec![( + col("j", &right_schema).unwrap(), + col("k", &left_schema).unwrap(), + )]; + let join = Arc::new( + HashJoinExec::try_new( + TestScanBuilder::new(Arc::clone(&right_schema)) + .with_support(true) + .build(), + TestScanBuilder::new(Arc::clone(&left_schema)) + .with_support(true) + .build(), + on, + None, + &JoinType::RightSemi, + None, + PartitionMode::Partitioned, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(), + ); + let key_filter = col_lit_predicate("k", "x", &join.schema()); + let plan = Arc::new(FilterExec::try_new(key_filter, join).unwrap()) + as Arc; + + insta::assert_snapshot!( + OptimizationTest::new(Arc::clone(&plan), FilterPushdown::new(), true), + @r" + OptimizationTest: + input: + - FilterExec: k@0 = x + - HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(j@0, k@0)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[j, w, k], file_type=test, pushdown_supported=true + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[k, v], file_type=test, pushdown_supported=true + output: + Ok: + - HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(j@0, k@0)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[j, w, k], file_type=test, pushdown_supported=true, predicate=j@0 = x + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[k, v], file_type=test, pushdown_supported=true, predicate=k@0 = x + " + ); +} + +/// A key column that appears in several `on` pairs is transferred over the +/// first pair. Any pair would be correct, since all of them are equal for a +/// matching row; this pins the documented choice. +#[test] +fn test_hashjoin_parent_filter_transfer_uses_first_on_pair() { + let left_schema = Arc::new(Schema::new(vec![ + Field::new("k", DataType::Utf8, false), + Field::new("v", DataType::Utf8, false), + ])); + let right_schema = Arc::new(Schema::new(vec![ + Field::new("x", DataType::Utf8, false), + Field::new("y", DataType::Utf8, false), + ])); + // ON k = x AND k = y + let on = vec![ + ( + col("k", &left_schema).unwrap(), + col("x", &right_schema).unwrap(), + ), + ( + col("k", &left_schema).unwrap(), + col("y", &right_schema).unwrap(), + ), + ]; + let join = Arc::new( + HashJoinExec::try_new( + TestScanBuilder::new(Arc::clone(&left_schema)) + .with_support(true) + .build(), + TestScanBuilder::new(Arc::clone(&right_schema)) + .with_support(true) + .build(), + on, + None, + &JoinType::Inner, + None, + PartitionMode::Partitioned, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(), + ); + let key_filter = col_lit_predicate("k", "a", &join.schema()); + let plan = Arc::new(FilterExec::try_new(key_filter, join).unwrap()) + as Arc; + + insta::assert_snapshot!( + OptimizationTest::new(Arc::clone(&plan), FilterPushdown::new(), true), + @r" + OptimizationTest: + input: + - FilterExec: k@0 = a + - HashJoinExec: mode=Partitioned, join_type=Inner, on=[(k@0, x@0), (k@0, y@1)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[k, v], file_type=test, pushdown_supported=true + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[x, y], file_type=test, pushdown_supported=true + output: + Ok: + - HashJoinExec: mode=Partitioned, join_type=Inner, on=[(k@0, x@0), (k@0, y@1)] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[k, v], file_type=test, pushdown_supported=true, predicate=k@0 = a + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[x, y], file_type=test, pushdown_supported=true, predicate=x@0 = a + " + ); +} + +/// A transferred filter is rewritten over the other side's key *expression*, +/// here a `CAST`. The projection puts the right key at output index 0, the +/// same index as the left column inside the cast: the rewrite must not +/// descend into the substituted expression, or it would substitute again +/// without end. +#[test] +fn test_hashjoin_parent_filter_transfer_cast_key_with_projection() { + let left_schema = Arc::new(Schema::new(vec![ + Field::new("k", DataType::Int32, false), + Field::new("v", DataType::Utf8, false), + ])); + let right_schema = Arc::new(Schema::new(vec![ + Field::new("j", DataType::Int64, false), + Field::new("w", DataType::Utf8, false), + ])); + let on = vec![( + cast( + col("k", &left_schema).unwrap(), + &left_schema, + DataType::Int64, + ) + .unwrap(), + col("j", &right_schema).unwrap(), + )]; + let join = Arc::new( + HashJoinExec::try_new( + TestScanBuilder::new(Arc::clone(&left_schema)) + .with_support(true) + .build(), + TestScanBuilder::new(Arc::clone(&right_schema)) + .with_support(true) + .build(), + on, + None, + &JoinType::Inner, + // Output only `j`, at index 0. + Some(vec![2]), + PartitionMode::Partitioned, + datafusion_common::NullEquality::NullEqualsNothing, + false, + ) + .unwrap(), + ); + let key_filter = col_lit_predicate("j", 5i64, &join.schema()); + let plan = Arc::new(FilterExec::try_new(key_filter, join).unwrap()) + as Arc; + + insta::assert_snapshot!( + OptimizationTest::new(Arc::clone(&plan), FilterPushdown::new(), true), + @r" + OptimizationTest: + input: + - FilterExec: j@0 = 5 + - HashJoinExec: mode=Partitioned, join_type=Inner, on=[(CAST(k@0 AS Int64), j@0)], projection=[j@2] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[k, v], file_type=test, pushdown_supported=true + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[j, w], file_type=test, pushdown_supported=true + output: + Ok: + - HashJoinExec: mode=Partitioned, join_type=Inner, on=[(CAST(k@0 AS Int64), j@0)], projection=[j@2] + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[k, v], file_type=test, pushdown_supported=true, predicate=CAST(k@0 AS Int64) = 5 + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[j, w], file_type=test, pushdown_supported=true, predicate=j@0 = 5 + " + ); +} + /// A join's dynamic filter over the build-side key of the join below it is /// transferred across that join's keys onto its probe-side scan: the probe /// scan is pruned by the keys of a table it is not joined with. diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index e25b10ca58d6a..6a9dff49b89b2 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -27,7 +27,7 @@ use crate::execution_plan::{ }; use crate::filter_pushdown::{ ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase, - FilterPushdownPropagation, PushedDown, PushedDownPredicate, + FilterPushdownPropagation, PushedDownPredicate, }; use crate::joins::Map; use crate::joins::array_map::ArrayMap; @@ -1004,6 +1004,18 @@ impl HashJoinExec { &self, column_indices: &[ColumnIndex], ) -> (KeyTransferMap, KeyTransferMap) { + // A transferred filter compares the other side's key with literals + // typed for this side's key. The planner coerces both keys to one + // type, and `try_new` does not check it, so make the assumption + // explicit here. + debug_assert!( + self.on.iter().all(|(left_key, right_key)| { + left_key.data_type(&self.left.schema()).ok() + == right_key.data_type(&self.right.schema()).ok() + }), + "join key data types differ: {:?}", + self.on + ); let mut to_right = HashMap::new(); let mut to_left = HashMap::new(); for (output_idx, ci) in column_indices.iter().enumerate() { @@ -1022,6 +1034,9 @@ 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. JoinSide::None => continue, }; if let Some(other_key) = other_key { @@ -1899,37 +1914,40 @@ impl ExecutionPlan for HashJoinExec { }; }); - let mut left_child = if left_preserved { - ChildFilterDescription::from_child_with_allowed_indices( - &parent_filters, - left_allowed, - self.left(), - )? - } else { - ChildFilterDescription::all_unsupported(&parent_filters) - }; - - let mut right_child = if right_preserved { - ChildFilterDescription::from_child_with_allowed_indices( - &parent_filters, - right_allowed, - self.right(), - )? - } else { - ChildFilterDescription::all_unsupported(&parent_filters) - }; - // Transfer filters across the equi-join keys: a parent filter over one // side's join-key columns holds for every matching row of the other // 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. - if Self::supports_key_transfer(self.join_type) { - let (to_right, to_left) = self.key_transfer_maps(&column_indices); - transfer_key_filters(&parent_filters, &to_right, &mut right_child)?; - transfer_key_filters(&parent_filters, &to_left, &mut left_child)?; - } + // 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() + }; + let describe_child = |preserved: bool, + allowed: HashSet, + key_map: &KeyTransferMap, + child: &Arc| + -> Result { + if !preserved { + return Ok(ChildFilterDescription::all_unsupported(&parent_filters)); + } + let mut description = + ChildFilterDescription::from_child_with_allowed_indices( + &parent_filters, + allowed, + child, + )?; + transfer_key_filters(&parent_filters, key_map, &mut description)?; + Ok(description) + }; + + let left_child = + describe_child(left_preserved, left_allowed, &to_left, self.left())?; + let mut right_child = + describe_child(right_preserved, right_allowed, &to_right, self.right())?; // Add dynamic filters in Post phase if enabled. Skip when this join // already carries a dynamic filter from a previous pass — the shared @@ -2552,10 +2570,11 @@ fn is_column_at(expr: &PhysicalExprRef, index: usize) -> bool { /// Marks every parent filter whose columns are all join keys in `key_map` as /// supported for `child`, rewritten over the other side's key expressions. /// -/// Filters `child` already accepts directly are left alone, as are filters -/// that reference a non-key column or no column at all: the former cannot be -/// expressed on the other side, the latter were already routed by the plain -/// column analysis. +/// `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. fn transfer_key_filters( parent_filters: &[Arc], key_map: &KeyTransferMap, @@ -2565,9 +2584,6 @@ fn transfer_key_filters( return Ok(()); } for (filter, pushed) in parent_filters.iter().zip(child.parent_filters.iter_mut()) { - if matches!(pushed.discriminant, PushedDown::Yes) { - continue; - } if let Some(transferred) = transfer_filter_across_keys(filter, key_map)? { *pushed = PushedDownPredicate::supported(transferred); } @@ -2576,7 +2592,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, or no column. +/// it references a column that is not a transferable key. /// /// A [`DynamicFilterPhysicalExpr`] comes out as a view sharing the original's /// state with its key columns remapped, so it keeps tracking the build side. @@ -2585,15 +2601,16 @@ fn transfer_filter_across_keys( key_map: &KeyTransferMap, ) -> Result>> { let mut all_keys = true; - let mut any_column = false; let transformed = Arc::clone(filter).transform_down(|expr| { let Some(column) = expr.downcast_ref::() else { return Ok(Transformed::no(expr)); }; - any_column = true; match key_map.get(&column.index()) { - // The replacement is already in the other side's schema: do not - // descend into it, its columns are not indices of this join. + // The replacement is in the other side's input schema, so its + // columns are not output indices of this join: `Jump` over it. + // Descending would substitute again whenever the key column's + // index is also an output index, e.g. `CAST(k@0 AS Int64)` for + // output column 0, and never terminate. Some(other_key) => Ok(Transformed::new( Arc::clone(other_key), true, @@ -2605,7 +2622,7 @@ fn transfer_filter_across_keys( } } })?; - Ok((any_column && all_keys).then_some(transformed.data)) + Ok(all_keys.then_some(transformed.data)) } /// Determines which sides of a join are "preserved" for filter pushdown. @@ -2619,8 +2636,11 @@ fn lr_is_preserved(join_type: JoinType) -> (bool, bool) { JoinType::Left => (true, false), JoinType::Right => (false, true), JoinType::Full => (false, false), - // The non-output side of a semi join only receives filters transferred - // across the join keys, see `HashJoinExec::supports_key_transfer`. + // A semi join emits only matched rows, so pruning either input by a + // filter its output satisfies is exact. The non-output side has no + // output columns, so the column routing sends it nothing but + // column-free filters; key filters reach it through the transfer in + // `HashJoinExec::gather_filters_for_pushdown`. JoinType::LeftSemi | JoinType::RightSemi => (true, true), JoinType::LeftAnti | JoinType::LeftMark => (true, false), JoinType::RightAnti | JoinType::RightMark => (false, true),