diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index f10725fd30803..783c216f20140 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -24,7 +24,7 @@ use arrow::{ }; use arrow_schema::SortOptions; use datafusion::{ - assert_batches_eq, + assert_batches_eq, assert_batches_sorted_eq, logical_expr::Operator, physical_plan::{ PhysicalExpr, @@ -51,7 +51,7 @@ use datafusion_functions_aggregate::{ }; use datafusion_physical_expr::{ LexOrdering, PhysicalSortExpr, - expressions::{DynamicFilterPhysicalExpr, col}, + expressions::{DynamicFilterPhysicalExpr, IsNullExpr, cast, col}, utils::conjunction, }; use datafusion_physical_expr::{ @@ -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 " ); } @@ -1752,8 +1752,10 @@ fn test_aggregate_pushdown_preserves_duplicate_grouping_columns() { assert_eq!(filters[0][1].predicate.to_string(), "id@0 = x"); } -/// A semi join key that is not a plain column cannot be mapped to the other -/// side, so a filter on that output key only reaches the emitted side. +/// A semi join whose emitted-side key is an expression has no output column +/// that is a join key, so a filter on the emitted column stays on that side. +/// When the emitted-side key is a plain column, the filter is transferred to +/// the other side, rewritten over that side's key expression. #[test] fn test_hashjoin_parent_filter_pushdown_semi_join_expression_key() { use datafusion_physical_expr::expressions::CastExpr; @@ -1763,7 +1765,10 @@ fn test_hashjoin_parent_filter_pushdown_semi_join_expression_key() { let key = || Arc::new(Column::new("id", 0)) as Arc; let cast_key = || Arc::new(CastExpr::new(key(), DataType::Utf8, None)) as Arc; - for on in [(cast_key(), key()), (key(), cast_key())] { + for (on, transferred) in [ + ((cast_key(), key()), None), + ((key(), cast_key()), Some("CAST(id@0 AS Utf8) = x")), + ] { let join = HashJoinExec::try_new( TestScanBuilder::new(Arc::clone(&schema)).build(), TestScanBuilder::new(Arc::clone(&schema)).build(), @@ -1785,8 +1790,14 @@ fn test_hashjoin_parent_filter_pushdown_semi_join_expression_key() { .unwrap() .parent_filters(); assert!(matches!(filters[0][0].discriminant, PushedDown::Yes)); - assert!(matches!(filters[1][0].discriminant, PushedDown::No)); assert_eq!(filters[0][0].predicate.to_string(), "id@0 = x"); + match transferred { + Some(expected) => { + assert!(matches!(filters[1][0].discriminant, PushedDown::Yes)); + assert_eq!(filters[1][0].predicate.to_string(), expected); + } + None => assert!(matches!(filters[1][0].discriminant, PushedDown::No)), + } } } @@ -1895,7 +1906,9 @@ fn test_from_child_with_allowed_indices_rejects_unresolvable_name() { } /// A join's output projection must map to child positions even when a child -/// contains multiple columns with the same name. +/// contains multiple columns with the same name. Position 0 on each side is +/// the join key, so a filter on it is also transferred to the other side's +/// key; the same-named non-key column at position 1 stays on its own side. #[test] fn test_hashjoin_parent_filter_pushdown_duplicate_child_columns() { use datafusion_physical_plan::filter_pushdown::{FilterPushdownPhase, PushedDown}; @@ -1937,11 +1950,16 @@ fn test_hashjoin_parent_filter_pushdown_duplicate_child_columns() { .parent_filters(); let side = input_index / 2; assert!(matches!(filters[side][0].discriminant, PushedDown::Yes)); - assert!(matches!(filters[1 - side][0].discriminant, PushedDown::No)); assert_eq!( filters[side][0].predicate.to_string(), format!("id@{} = x", input_index % 2) ); + if input_index % 2 == 0 { + assert!(matches!(filters[1 - side][0].discriminant, PushedDown::Yes)); + assert_eq!(filters[1 - side][0].predicate.to_string(), "id@0 = x"); + } else { + assert!(matches!(filters[1 - side][0].discriminant, PushedDown::No)); + } } } } @@ -2151,6 +2169,698 @@ fn test_hashjoin_parent_filter_pushdown_semi_anti_join() { assert_parent_filter_remains(plan); } +/// Under `NullEqualsNull` an inner join also emits rows whose keys are NULL +/// on both sides, so `id` and `pid` are still identical in every output row +/// (both NULL or equal). A parent filter over one side's key, including an +/// `IS NULL` check, therefore transfers to the other side exactly as under +/// `NullEqualsNothing`, and the NULL-keyed matches survive the transfer. +#[tokio::test] +async fn test_hashjoin_parent_filter_transfer_null_equals_null_inner_join() { + let build_side_schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Utf8, true), + Field::new("build_val", DataType::Utf8, false), + ])); + let build_batches = vec![ + record_batch!( + ("id", Utf8, [Some("aa"), None, Some("bb")]), + ("build_val", Utf8, ["b1", "b2", "b3"]) + ) + .unwrap(), + ]; + let build_scan = TestScanBuilder::new(Arc::clone(&build_side_schema)) + .with_support(true) + .with_batches(build_batches) + .build(); + + let probe_side_schema = Arc::new(Schema::new(vec![ + Field::new("pid", DataType::Utf8, true), + Field::new("probe_val", DataType::Utf8, false), + ])); + let probe_batches = vec![ + record_batch!( + ("pid", Utf8, [Some("aa"), None, Some("bb"), None]), + ("probe_val", Utf8, ["p1", "p2", "p3", "p4"]) + ) + .unwrap(), + ]; + let probe_scan = TestScanBuilder::new(Arc::clone(&probe_side_schema)) + .with_support(true) + .with_batches(probe_batches) + .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::CollectLeft, + datafusion_common::NullEquality::NullEqualsNull, + false, + ) + .unwrap(), + ); + let join_schema = join.schema(); + + // id = 'aa' OR id IS NULL: keeps the 'aa' match and the NULL/NULL matches + let predicate = Arc::new(BinaryExpr::new( + col_lit_predicate("id", "aa", &join_schema), + Operator::Or, + Arc::new(IsNullExpr::new(col("id", &join_schema).unwrap())), + )); + let plan = + Arc::new(FilterExec::try_new(predicate, join).unwrap()) as Arc; + + insta::assert_snapshot!( + OptimizationTest::new(Arc::clone(&plan), FilterPushdown::new(), true), + @r" + OptimizationTest: + input: + - FilterExec: id@0 = aa OR id@0 IS NULL + - HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, pid@0)], NullsEqual: true + - 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=CollectLeft, join_type=Inner, on=[(id@0, pid@0)], NullsEqual: true + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[id, build_val], file_type=test, pushdown_supported=true, predicate=id@0 = aa OR id@0 IS NULL + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[pid, probe_val], file_type=test, pushdown_supported=true, predicate=pid@0 = aa OR pid@0 IS NULL + " + ); + + 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(optimized, session_ctx.task_ctx()).await.unwrap(); + // The NULL build key matches both NULL probe keys; 'bb' is filtered out + // on both sides and the parent filter is gone. + #[rustfmt::skip] + let expected = [ + "+----+-----------+-----+-----------+", + "| id | build_val | pid | probe_val |", + "+----+-----------+-----+-----------+", + "| | b2 | | p2 |", + "| | b2 | | p4 |", + "| aa | b1 | aa | p1 |", + "+----+-----------+-----+-----------+", + ]; + assert_batches_sorted_eq!(expected, &batches); +} + +/// 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 + " + ); +} + +/// `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. +/// +/// 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_plan_for_test(&plan).to_string(), + @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 088376bc349c1..c5fae3a65c04f 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, + FilterPushdownPropagation, 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,73 @@ 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) { + // 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() { + 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), + ), + // 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 { + 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 { @@ -1849,58 +1916,40 @@ impl ExecutionPlan for HashJoinExec { } } - // Semi joins also allow filters on output join keys to reach the - // non-output side. Map to the paired key's position, which need not - // have the same name. Only direct column pairs can be remapped this way. - // Anti joins cannot do this: their emitted rows have no matching key. - if matches!(self.join_type, JoinType::LeftSemi | JoinType::RightSemi) { - let key_mapping: HashMap = self - .on - .iter() - .filter_map(|(left_key, right_key)| { - let left = left_key.downcast_ref::()?; - let right = right_key.downcast_ref::()?; - Some(match self.join_type { - JoinType::LeftSemi => (left.index(), right.index()), - _ => (right.index(), left.index()), - }) - }) - .collect(); - let (output_side, other_mapping) = match self.join_type { - JoinType::LeftSemi => (JoinSide::Left, &mut right_mapping), - _ => (JoinSide::Right, &mut left_mapping), - }; - let emitted = column_indices - .iter() - .enumerate() - .filter(|(_, ci)| ci.side == output_side); - for (output_idx, ci) in emitted { - if let Some(&input_idx) = key_mapping.get(&ci.index) { - other_mapping.insert(output_idx, input_idx); - } - } - } - - let left_child = if left_preserved { - ChildFilterDescription::from_child_with_column_mapping( - &parent_filters, - left_mapping, - self.left(), - )? + // 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. 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 { - ChildFilterDescription::all_unsupported(&parent_filters) + Default::default() }; - - let mut right_child = if right_preserved { - ChildFilterDescription::from_child_with_column_mapping( + 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, - right_mapping, - self.right(), - )? - } else { - ChildFilterDescription::all_unsupported(&parent_filters) + column_mapping, + child, + )?; + transfer_key_filters(&parent_filters, key_map, &mut description)?; + Ok(description) }; + let left_child = + describe_child(left_preserved, left_mapping, &to_left, self.left())?; + let mut right_child = + describe_child(right_preserved, right_mapping, &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 // `Arc` is still wired into the probe-side @@ -2510,6 +2559,73 @@ 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. +/// +/// `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, + 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 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. +/// +/// 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 transformed = Arc::clone(filter).transform_down(|expr| { + let Some(column) = expr.downcast_ref::() else { + return Ok(Transformed::no(expr)); + }; + 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. + // 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, + TreeNodeRecursion::Jump, + )), + None => { + all_keys = false; + Ok(Transformed::new(expr, false, TreeNodeRecursion::Stop)) + } + } + })?; + Ok(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 @@ -2521,7 +2637,11 @@ 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. + // 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), 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;