From db584a8d16f2fa79f4d9decb58e72388f2db042f Mon Sep 17 00:00:00 2001 From: siddu Date: Thu, 8 Oct 2026 00:51:13 +0530 Subject: [PATCH 1/6] fix: skip HashJoinExec nodes carrying dynamic filters in JoinSelection (#26106) ## Which issue does this PR close? - Closes #26106. ## Rationale for this change `JoinSelection` is not safe to run on plans that already carry dynamic filters, which occurs during re-optimization passes (e.g. downstream pipelines that wrap an already-optimized plan into a writer sink and re-run physical optimization). Once `FilterPushdown` has wired up a dynamic filter between the build side and probe side of a `HashJoinExec`, swapping the inputs via `swap_inputs` panics with: `Internal error: Cannot swap HashJoinExec inputs after dynamic filter has been constructed` The join's build side is already committed at that stage, and swapping inputs would invalidate the dynamic filter expressions that reference probe-side columns. `JoinSelection` should detect this and leave the `HashJoinExec` unchanged instead of failing the query. ## What changes are included in this PR? - In `JoinSelection::statistical_join_selection_subrule`, check if `!hash_join.dynamic_expressions_produced().is_empty()` and return `None` (leaving the plan unchanged). - In `can_swap_hash_join`, guard against swapping when dynamic expressions are produced. - In `hash_join_swap_subrule`, guard against swapping unbounded left inputs when dynamic expressions are produced. - Added tests in `join_selection.rs` verifying that `JoinSelection` skips `HashJoinExec` carrying dynamic filters in both `CollectLeft` and `Partitioned` modes. ## What is the testing strategy for this PR? Added `test_join_selection_skips_hash_join_with_dynamic_filter` in `datafusion/core/tests/physical_optimizer/join_selection.rs` verifying that `JoinSelection` leaves the plan unchanged for both `CollectLeft` and `Partitioned` modes without error. ## Are there any user-facing changes? No API changes. Fixes an internal error when re-optimizing plans that contain dynamic filters. --- .../physical_optimizer/join_selection.rs | 57 ++++++++++++++++- .../physical-optimizer/src/join_selection.rs | 63 +++++++++++-------- 2 files changed, 93 insertions(+), 27 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/join_selection.rs b/datafusion/core/tests/physical_optimizer/join_selection.rs index d4561decc6331..d82f290051b1e 100644 --- a/datafusion/core/tests/physical_optimizer/join_selection.rs +++ b/datafusion/core/tests/physical_optimizer/join_selection.rs @@ -33,7 +33,9 @@ use datafusion_execution::{RecordBatchStream, SendableRecordBatchStream, TaskCon use datafusion_expr::Operator; use datafusion_physical_expr::PhysicalExprRef; use datafusion_physical_expr::expressions::col; -use datafusion_physical_expr::expressions::{BinaryExpr, Column, NegativeExpr}; +use datafusion_physical_expr::expressions::{ + BinaryExpr, Column, DynamicFilterPhysicalExpr, NegativeExpr, lit, +}; use datafusion_physical_expr::intervals::utils::check_support; use datafusion_physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr}; use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr; @@ -1969,3 +1971,56 @@ fn test_join_with_maybe_swap_unbounded_case(t: TestCase) -> Result<()> { } Ok(()) } + +#[rstest] +#[case(PartitionMode::CollectLeft)] +#[case(PartitionMode::Partitioned)] +#[tokio::test] +async fn test_join_selection_skips_hash_join_with_dynamic_filter( + #[case] partition_mode: PartitionMode, +) -> Result<()> { + // Left has larger statistics than right, which would normally trigger swap + let (big, small) = create_big_and_small(); + let on = vec![( + Arc::new(Column::new_with_schema("big_col", &big.schema())?) as PhysicalExprRef, + Arc::new(Column::new_with_schema("small_col", &small.schema())?) as PhysicalExprRef, + )]; + + let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new( + vec![Arc::clone(&on[0].1)], + lit(true), + )); + + #[allow(deprecated)] + let join = Arc::new( + HashJoinExec::try_new( + Arc::clone(&big), + Arc::clone(&small), + on, + None, + &JoinType::Inner, + None, + partition_mode, + NullEquality::NullEqualsNothing, + false, + )? + .with_dynamic_filter_expr(dynamic_filter)?, + ); + + let original_schema = join.schema(); + + // JoinSelection must not fail and must leave the join unchanged + let optimized = JoinSelection::new().optimize(join, &ConfigOptions::new())?; + let optimized_join = optimized + .downcast_ref::() + .expect("join should remain HashJoinExec without wrapping projection"); + + assert_eq!(optimized_join.partition_mode(), partition_mode); + assert_eq!(*optimized_join.join_type(), JoinType::Inner); + assert!(Arc::ptr_eq(optimized_join.left(), &big)); + assert!(Arc::ptr_eq(optimized_join.right(), &small)); + assert_eq!(optimized_join.schema(), original_schema); + assert_eq!(optimized_join.dynamic_expressions_produced().len(), 1); + + Ok(()) +} diff --git a/datafusion/physical-optimizer/src/join_selection.rs b/datafusion/physical-optimizer/src/join_selection.rs index 2b342bece7040..68cfe8abef62a 100644 --- a/datafusion/physical-optimizer/src/join_selection.rs +++ b/datafusion/physical-optimizer/src/join_selection.rs @@ -173,9 +173,11 @@ impl PhysicalOptimizerRule for JoinSelection { } /// Determines whether it is possible to swap inputs of a hash join - for null-aware joins, we can only swap an uncorrelated `LeftAnti` -/// (a single join key and no filter), because the swapped `RightAnti` has no per-row NULL handling +/// (a single join key and no filter), because the swapped `RightAnti` has no per-row NULL handling. +/// Joins carrying a dynamic filter cannot be swapped. fn can_swap_hash_join(hash_join: &HashJoinExec) -> bool { hash_join.join_type().supports_swap() + && hash_join.dynamic_expressions_produced().is_empty() && (!hash_join.null_aware || (*hash_join.join_type() == JoinType::LeftAnti && hash_join.on().len() == 1 @@ -301,32 +303,40 @@ fn statistical_join_selection_subrule( context: &dyn PhysicalOptimizerContext, ) -> Result>> { let transformed = if let Some(hash_join) = plan.downcast_ref::() { - match hash_join.partition_mode() { - PartitionMode::Auto => try_collect_left(hash_join, false, context)? - .map_or_else( - || partitioned_hash_join(hash_join, context).map(Some), - |v| Ok(Some(v)), - )?, - PartitionMode::CollectLeft => try_collect_left(hash_join, true, context)? - .map_or_else( - || partitioned_hash_join(hash_join, context).map(Some), - |v| Ok(Some(v)), - )?, - PartitionMode::Partitioned => { - let left = hash_join.left(); - let right = hash_join.right(); - if can_swap_hash_join(hash_join) - && should_swap_join_order(&**left, &**right, context)? - { - // Null-aware RightAnti only supports CollectLeft - let partition_mode = if hash_join.null_aware { - PartitionMode::CollectLeft + if !hash_join.dynamic_expressions_produced().is_empty() { + // Once a HashJoinExec carries a dynamic filter, its build side + // has already been determined and the dynamic filter has been wired + // to the probe side. Reordering inputs would invalidate the dynamic + // filter, so skip this join and leave it unchanged. + None + } else { + match hash_join.partition_mode() { + PartitionMode::Auto => try_collect_left(hash_join, false, context)? + .map_or_else( + || partitioned_hash_join(hash_join, context).map(Some), + |v| Ok(Some(v)), + )?, + PartitionMode::CollectLeft => try_collect_left(hash_join, true, context)? + .map_or_else( + || partitioned_hash_join(hash_join, context).map(Some), + |v| Ok(Some(v)), + )?, + PartitionMode::Partitioned => { + let left = hash_join.left(); + let right = hash_join.right(); + if can_swap_hash_join(hash_join) + && should_swap_join_order(&**left, &**right, context)? + { + // Null-aware RightAnti only supports CollectLeft + let partition_mode = if hash_join.null_aware { + PartitionMode::CollectLeft + } else { + PartitionMode::Partitioned + }; + hash_join.swap_inputs(partition_mode).map(Some)? } else { - PartitionMode::Partitioned - }; - hash_join.swap_inputs(partition_mode).map(Some)? - } else { - None + None + } } } } @@ -524,6 +534,7 @@ pub fn hash_join_swap_subrule( _config_options: &ConfigOptions, ) -> Result> { if let Some(hash_join) = input.downcast_ref::() + && hash_join.dynamic_expressions_produced().is_empty() && hash_join.left.boundedness().is_unbounded() && !hash_join.right.boundedness().is_unbounded() && !hash_join.null_aware // Don't swap null-aware anti joins From dd8e781db7b5f389630400ac931e5610eace06b9 Mon Sep 17 00:00:00 2001 From: siddu Date: Thu, 8 Oct 2026 16:15:32 +0530 Subject: [PATCH 2/6] fix: address cargo fmt and PartitionMode comparison in tests --- datafusion/core/tests/physical_optimizer/join_selection.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/join_selection.rs b/datafusion/core/tests/physical_optimizer/join_selection.rs index d82f290051b1e..26513d20552a2 100644 --- a/datafusion/core/tests/physical_optimizer/join_selection.rs +++ b/datafusion/core/tests/physical_optimizer/join_selection.rs @@ -1983,7 +1983,8 @@ async fn test_join_selection_skips_hash_join_with_dynamic_filter( let (big, small) = create_big_and_small(); let on = vec![( Arc::new(Column::new_with_schema("big_col", &big.schema())?) as PhysicalExprRef, - Arc::new(Column::new_with_schema("small_col", &small.schema())?) as PhysicalExprRef, + Arc::new(Column::new_with_schema("small_col", &small.schema())?) + as PhysicalExprRef, )]; let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new( @@ -2015,7 +2016,7 @@ async fn test_join_selection_skips_hash_join_with_dynamic_filter( .downcast_ref::() .expect("join should remain HashJoinExec without wrapping projection"); - assert_eq!(optimized_join.partition_mode(), partition_mode); + assert_eq!(*optimized_join.partition_mode(), partition_mode); assert_eq!(*optimized_join.join_type(), JoinType::Inner); assert!(Arc::ptr_eq(optimized_join.left(), &big)); assert!(Arc::ptr_eq(optimized_join.right(), &small)); From 5fb39d7ce7f47d137401f9cd6ea954e7d11a8527 Mon Sep 17 00:00:00 2001 From: siddu Date: Thu, 8 Oct 2026 23:35:43 +0530 Subject: [PATCH 3/6] Fix clippy and codecov issues in join selection tests --- .../physical_optimizer/join_selection.rs | 68 ++++++++++++++++++- .../physical-optimizer/src/join_selection.rs | 4 +- 2 files changed, 68 insertions(+), 4 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/join_selection.rs b/datafusion/core/tests/physical_optimizer/join_selection.rs index 26513d20552a2..b259bd866ae5f 100644 --- a/datafusion/core/tests/physical_optimizer/join_selection.rs +++ b/datafusion/core/tests/physical_optimizer/join_selection.rs @@ -1992,7 +1992,7 @@ async fn test_join_selection_skips_hash_join_with_dynamic_filter( lit(true), )); - #[allow(deprecated)] + #[expect(deprecated)] let join = Arc::new( HashJoinExec::try_new( Arc::clone(&big), @@ -2025,3 +2025,69 @@ async fn test_join_selection_skips_hash_join_with_dynamic_filter( Ok(()) } + +#[tokio::test] +async fn test_join_selection_skips_unbounded_hash_join_with_dynamic_filter() -> Result<()> { + let left_exec: Arc = Arc::new(UnboundedExec::new( + None, + RecordBatch::new_empty(Arc::new(Schema::new(vec![Field::new( + "a", + DataType::Int32, + false, + )]))), + 2, + )); + let right_exec: Arc = Arc::new(UnboundedExec::new( + Some(1), + RecordBatch::new_empty(Arc::new(Schema::new(vec![Field::new( + "b", + DataType::Int32, + false, + )]))), + 2, + )); + + let on = vec![( + col("a", &left_exec.schema())?, + col("b", &right_exec.schema())?, + )]; + + let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new( + vec![Arc::clone(&on[0].1)], + lit(true), + )); + + #[expect(deprecated)] + let join = Arc::new( + HashJoinExec::try_new( + Arc::clone(&left_exec), + Arc::clone(&right_exec), + on, + None, + &JoinType::Inner, + None, + PartitionMode::Partitioned, + NullEquality::NullEqualsNothing, + false, + )? + .with_dynamic_filter_expr(dynamic_filter)?, + ); + + let original_schema = join.schema(); + + // hash_join_swap_subrule would normally swap unbounded left with bounded right, + // but must skip this join because it has a dynamic filter. + let optimized = JoinSelection::new().optimize(join, &ConfigOptions::new())?; + let optimized_join = optimized + .downcast_ref::() + .expect("join should remain HashJoinExec without wrapping projection"); + + assert_eq!(*optimized_join.partition_mode(), PartitionMode::Partitioned); + assert_eq!(*optimized_join.join_type(), JoinType::Inner); + assert!(Arc::ptr_eq(optimized_join.left(), &left_exec)); + assert!(Arc::ptr_eq(optimized_join.right(), &right_exec)); + assert_eq!(optimized_join.schema(), original_schema); + assert_eq!(optimized_join.dynamic_expressions_produced().len(), 1); + + Ok(()) +} diff --git a/datafusion/physical-optimizer/src/join_selection.rs b/datafusion/physical-optimizer/src/join_selection.rs index 68cfe8abef62a..a0eb6997a4c6b 100644 --- a/datafusion/physical-optimizer/src/join_selection.rs +++ b/datafusion/physical-optimizer/src/join_selection.rs @@ -173,11 +173,9 @@ impl PhysicalOptimizerRule for JoinSelection { } /// Determines whether it is possible to swap inputs of a hash join - for null-aware joins, we can only swap an uncorrelated `LeftAnti` -/// (a single join key and no filter), because the swapped `RightAnti` has no per-row NULL handling. -/// Joins carrying a dynamic filter cannot be swapped. +/// (a single join key and no filter), because the swapped `RightAnti` has no per-row NULL handling fn can_swap_hash_join(hash_join: &HashJoinExec) -> bool { hash_join.join_type().supports_swap() - && hash_join.dynamic_expressions_produced().is_empty() && (!hash_join.null_aware || (*hash_join.join_type() == JoinType::LeftAnti && hash_join.on().len() == 1 From 7192dc769751b1f423373ec025781f76474e3753 Mon Sep 17 00:00:00 2001 From: siddu Date: Sun, 11 Oct 2026 15:13:34 +0530 Subject: [PATCH 4/6] test: address review on join selection dynamic filter guards - run cargo fmt to fix test_join_selection_skips_unbounded_hash_join_with_dynamic_filter signature - add end-to-end regression that lets FilterPushdown::new_post_optimization create the dynamic filter, then verifies JoinSelection leaves it unchanged and the plan executes with correct results --- .../physical_optimizer/join_selection.rs | 123 +++++++++++++++++- 1 file changed, 122 insertions(+), 1 deletion(-) diff --git a/datafusion/core/tests/physical_optimizer/join_selection.rs b/datafusion/core/tests/physical_optimizer/join_selection.rs index b259bd866ae5f..4208c3d49c9ea 100644 --- a/datafusion/core/tests/physical_optimizer/join_selection.rs +++ b/datafusion/core/tests/physical_optimizer/join_selection.rs @@ -22,8 +22,11 @@ use std::{ task::{Context, Poll}, }; +use arrow::array::record_batch; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; use arrow::record_batch::RecordBatch; +use datafusion::datasource::object_store::ObjectStoreUrl; +use datafusion::prelude::{SessionConfig, SessionContext}; use datafusion_common::config::ConfigOptions; use datafusion_common::tree_node::TreeNodeRecursion; use datafusion_common::{ColumnStatistics, JoinType, ScalarValue, stats::Precision}; @@ -41,7 +44,9 @@ use datafusion_physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr; use datafusion_physical_optimizer::PhysicalOptimizerContext; use datafusion_physical_optimizer::PhysicalOptimizerRule; +use datafusion_physical_optimizer::filter_pushdown::FilterPushdown; use datafusion_physical_optimizer::join_selection::JoinSelection; +use datafusion_physical_plan::collect; use datafusion_physical_plan::displayable; use datafusion_physical_plan::joins::utils::ColumnIndex; use datafusion_physical_plan::joins::utils::JoinFilter; @@ -50,6 +55,7 @@ use datafusion_physical_plan::operator_statistics::{ ClosureStatisticsProvider, StatisticsRegistry, StatisticsResult, }; use datafusion_physical_plan::projection::ProjectionExec; +use datafusion_physical_plan::repartition::RepartitionExec; use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec; use datafusion_physical_plan::{ ChildrenPropertiesMode, ExecutionPlanProperties, ReplaceChildrenOptions, @@ -61,8 +67,11 @@ use datafusion_physical_plan::{ }; use futures::Stream; +use object_store::memory::InMemory; use rstest::rstest; +use super::pushdown_utils::TestScanBuilder; + /// Return statistics for empty table fn empty_statistics() -> Statistics { Statistics { @@ -2027,7 +2036,8 @@ async fn test_join_selection_skips_hash_join_with_dynamic_filter( } #[tokio::test] -async fn test_join_selection_skips_unbounded_hash_join_with_dynamic_filter() -> Result<()> { +async fn test_join_selection_skips_unbounded_hash_join_with_dynamic_filter() -> Result<()> +{ let left_exec: Arc = Arc::new(UnboundedExec::new( None, RecordBatch::new_empty(Arc::new(Schema::new(vec![Field::new( @@ -2091,3 +2101,114 @@ async fn test_join_selection_skips_unbounded_hash_join_with_dynamic_filter() -> Ok(()) } + +/// End-to-end regression for #26106: let `FilterPushdown::new_post_optimization()` +/// wire up a real dynamic filter, then ensure `JoinSelection` leaves the join +/// untouched and the plan still executes with correct results. +#[tokio::test] +async fn test_join_selection_skips_real_dynamic_filter_pushdown() -> Result<()> { + let build_schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Utf8, false), + Field::new("b", DataType::Utf8, false), + ])); + let build_scan = TestScanBuilder::new(Arc::clone(&build_schema)) + .with_support(true) + .with_batches(vec![ + record_batch!(("a", Utf8, ["aa", "ab"]), ("b", Utf8, ["ba", "bb"])).unwrap(), + ]) + .build(); + + let probe_schema = Arc::new(Schema::new(vec![ + Field::new("a", DataType::Utf8, false), + Field::new("b", DataType::Utf8, false), + ])); + let probe_scan = TestScanBuilder::new(Arc::clone(&probe_schema)) + .with_support(true) + .with_batches(vec![ + record_batch!( + ("a", Utf8, ["aa", "ab", "ac"]), + ("b", Utf8, ["ba", "bb", "bc"]) + ) + .unwrap(), + ]) + .build(); + + let partition_count = 4; + let build_repartition = Arc::new( + RepartitionExec::try_new( + build_scan, + Partitioning::Hash( + vec![col("a", &build_schema)?, col("b", &build_schema)?], + partition_count, + ), + ) + .unwrap(), + ); + let probe_repartition = Arc::new( + RepartitionExec::try_new( + probe_scan, + Partitioning::Hash( + vec![col("a", &probe_schema)?, col("b", &probe_schema)?], + partition_count, + ), + ) + .unwrap(), + ); + + let on = vec![ + (col("a", &build_schema)?, col("a", &probe_schema)?), + (col("b", &build_schema)?, col("b", &probe_schema)?), + ]; + let join = Arc::new( + HashJoinExec::try_new( + build_repartition, + probe_repartition, + on, + None, + &JoinType::Inner, + None, + PartitionMode::Partitioned, + NullEquality::NullEqualsNothing, + false, + ) + .unwrap(), + ) as Arc; + + let mut config = ConfigOptions::new(); + config.execution.parquet.pushdown_filters = true; + + // Real optimizer wiring (not manual `with_dynamic_filter_expr`) + let with_filter = FilterPushdown::new_post_optimization().optimize(join, &config)?; + let join_with_filter = with_filter + .downcast_ref::() + .expect("plan should still be a HashJoinExec after pushdown"); + assert!( + !join_with_filter.dynamic_expressions_produced().is_empty(), + "FilterPushdown should have created a dynamic filter" + ); + + // JoinSelection must not panic in `swap_inputs` and must leave the join as-is + let after_selection = JoinSelection::new().optimize(with_filter, &config)?; + let final_join = after_selection + .downcast_ref::() + .expect("JoinSelection must leave dynamic-filter join as HashJoinExec"); + assert!(!final_join.dynamic_expressions_produced().is_empty()); + assert_eq!(*final_join.partition_mode(), PartitionMode::Partitioned); + + // The wired plan must still execute and return the 2 matching rows + let session_config = SessionConfig::new(); + let session_ctx = SessionContext::new_with_config(session_config); + session_ctx.register_object_store( + ObjectStoreUrl::parse("test://").unwrap().as_ref(), + Arc::new(InMemory::new()), + ); + let task_ctx = session_ctx.task_ctx(); + let batches = collect(after_selection, task_ctx).await?; + let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum(); + assert_eq!( + total_rows, 2, + "expected 2 inner-join matches, got {batches:?}" + ); + + Ok(()) +} From 1643c176dfa6eecac07cf1667127d7620bee0b49 Mon Sep 17 00:00:00 2001 From: siddu Date: Mon, 12 Oct 2026 00:25:14 +0530 Subject: [PATCH 5/6] test: force swap statistics in dynamic filter regression so it fails without guard --- .../physical_optimizer/join_selection.rs | 91 ++++++++++++++++--- 1 file changed, 78 insertions(+), 13 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/join_selection.rs b/datafusion/core/tests/physical_optimizer/join_selection.rs index 4208c3d49c9ea..44772360694d1 100644 --- a/datafusion/core/tests/physical_optimizer/join_selection.rs +++ b/datafusion/core/tests/physical_optimizer/join_selection.rs @@ -2105,29 +2105,37 @@ async fn test_join_selection_skips_unbounded_hash_join_with_dynamic_filter() -> /// End-to-end regression for #26106: let `FilterPushdown::new_post_optimization()` /// wire up a real dynamic filter, then ensure `JoinSelection` leaves the join /// untouched and the plan still executes with correct results. +/// +/// The build side reports larger statistics than the probe side so that +/// `should_swap_join_order` returns true: without the dynamic-filter guard, +/// `JoinSelection` would attempt `swap_inputs` and fail. #[tokio::test] async fn test_join_selection_skips_real_dynamic_filter_pushdown() -> Result<()> { let build_schema = Arc::new(Schema::new(vec![ - Field::new("a", DataType::Utf8, false), - Field::new("b", DataType::Utf8, false), + Field::new("build_a", DataType::Utf8, false), + Field::new("build_b", DataType::Utf8, false), ])); let build_scan = TestScanBuilder::new(Arc::clone(&build_schema)) .with_support(true) .with_batches(vec![ - record_batch!(("a", Utf8, ["aa", "ab"]), ("b", Utf8, ["ba", "bb"])).unwrap(), + record_batch!( + ("build_a", Utf8, ["aa", "ab"]), + ("build_b", Utf8, ["ba", "bb"]) + ) + .unwrap(), ]) .build(); let probe_schema = Arc::new(Schema::new(vec![ - Field::new("a", DataType::Utf8, false), - Field::new("b", DataType::Utf8, false), + Field::new("probe_a", DataType::Utf8, false), + Field::new("probe_b", DataType::Utf8, false), ])); let probe_scan = TestScanBuilder::new(Arc::clone(&probe_schema)) .with_support(true) .with_batches(vec![ record_batch!( - ("a", Utf8, ["aa", "ab", "ac"]), - ("b", Utf8, ["ba", "bb", "bc"]) + ("probe_a", Utf8, ["aa", "ab", "ac"]), + ("probe_b", Utf8, ["ba", "bb", "bc"]) ) .unwrap(), ]) @@ -2138,7 +2146,10 @@ async fn test_join_selection_skips_real_dynamic_filter_pushdown() -> Result<()> RepartitionExec::try_new( build_scan, Partitioning::Hash( - vec![col("a", &build_schema)?, col("b", &build_schema)?], + vec![ + col("build_a", &build_schema)?, + col("build_b", &build_schema)?, + ], partition_count, ), ) @@ -2148,7 +2159,10 @@ async fn test_join_selection_skips_real_dynamic_filter_pushdown() -> Result<()> RepartitionExec::try_new( probe_scan, Partitioning::Hash( - vec![col("a", &probe_schema)?, col("b", &probe_schema)?], + vec![ + col("probe_a", &probe_schema)?, + col("probe_b", &probe_schema)?, + ], partition_count, ), ) @@ -2156,8 +2170,14 @@ async fn test_join_selection_skips_real_dynamic_filter_pushdown() -> Result<()> ); let on = vec![ - (col("a", &build_schema)?, col("a", &probe_schema)?), - (col("b", &build_schema)?, col("b", &probe_schema)?), + ( + col("build_a", &build_schema)?, + col("probe_a", &probe_schema)?, + ), + ( + col("build_b", &build_schema)?, + col("probe_b", &probe_schema)?, + ), ]; let join = Arc::new( HashJoinExec::try_new( @@ -2186,14 +2206,59 @@ async fn test_join_selection_skips_real_dynamic_filter_pushdown() -> Result<()> !join_with_filter.dynamic_expressions_produced().is_empty(), "FilterPushdown should have created a dynamic filter" ); + let orig_left = Arc::clone(join_with_filter.left()); + let orig_right = Arc::clone(join_with_filter.right()); + + // Both `TestScanBuilder` inputs report the same 123-byte size, so force the + // build side to look larger. Without the dynamic-filter guard this makes + // `JoinSelection` attempt the swap and fail in `swap_inputs`. + let mut registry = StatisticsRegistry::new(); + registry.register(Arc::new(ClosureStatisticsProvider::with_matches( + |plan| { + plan.schema() + .fields() + .iter() + .any(|f| f.name().starts_with("build_")) + }, + |_plan, _child_stats| Ok(StatisticsResult::Computed(big_statistics().into())), + ))); + + struct ContextWithRegistry { + config: ConfigOptions, + registry: StatisticsRegistry, + } + + impl PhysicalOptimizerContext for ContextWithRegistry { + fn config_options(&self) -> &ConfigOptions { + &self.config + } + + fn statistics_registry(&self) -> Option<&StatisticsRegistry> { + Some(&self.registry) + } + } - // JoinSelection must not panic in `swap_inputs` and must leave the join as-is - let after_selection = JoinSelection::new().optimize(with_filter, &config)?; + let context = ContextWithRegistry { + config: config.clone(), + registry, + }; + + // JoinSelection must not fail in `swap_inputs` and must leave the join as-is + let after_selection = + JoinSelection::new().optimize_with_context(with_filter, &context)?; let final_join = after_selection .downcast_ref::() .expect("JoinSelection must leave dynamic-filter join as HashJoinExec"); assert!(!final_join.dynamic_expressions_produced().is_empty()); assert_eq!(*final_join.partition_mode(), PartitionMode::Partitioned); + assert!( + Arc::ptr_eq(final_join.left(), &orig_left), + "build side must not be swapped" + ); + assert!( + Arc::ptr_eq(final_join.right(), &orig_right), + "probe side must not be swapped" + ); // The wired plan must still execute and return the 2 matching rows let session_config = SessionConfig::new(); From d26b57e1382f70f65ac54baafcb1523da2720926 Mon Sep 17 00:00:00 2001 From: siddu Date: Mon, 12 Oct 2026 22:04:24 +0530 Subject: [PATCH 6/6] test: add probe-side small stats so swap is forced without guard --- .../tests/physical_optimizer/join_selection.rs | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/join_selection.rs b/datafusion/core/tests/physical_optimizer/join_selection.rs index 44772360694d1..1840568f1bf66 100644 --- a/datafusion/core/tests/physical_optimizer/join_selection.rs +++ b/datafusion/core/tests/physical_optimizer/join_selection.rs @@ -2209,9 +2209,11 @@ async fn test_join_selection_skips_real_dynamic_filter_pushdown() -> Result<()> let orig_left = Arc::clone(join_with_filter.left()); let orig_right = Arc::clone(join_with_filter.right()); - // Both `TestScanBuilder` inputs report the same 123-byte size, so force the - // build side to look larger. Without the dynamic-filter guard this makes - // `JoinSelection` attempt the swap and fail in `swap_inputs`. + // Both `TestScanBuilder` inputs report the same 123-byte size, and after + // `FilterPushdown` wires the dynamic filter the probe side reports Absent + // stats, so force both sides: build looks larger than probe. Without the + // dynamic-filter guard this makes `JoinSelection` attempt the swap and + // fail in `swap_inputs`. let mut registry = StatisticsRegistry::new(); registry.register(Arc::new(ClosureStatisticsProvider::with_matches( |plan| { @@ -2222,6 +2224,15 @@ async fn test_join_selection_skips_real_dynamic_filter_pushdown() -> Result<()> }, |_plan, _child_stats| Ok(StatisticsResult::Computed(big_statistics().into())), ))); + registry.register(Arc::new(ClosureStatisticsProvider::with_matches( + |plan| { + plan.schema() + .fields() + .iter() + .any(|f| f.name().starts_with("probe_")) + }, + |_plan, _child_stats| Ok(StatisticsResult::Computed(small_statistics().into())), + ))); struct ContextWithRegistry { config: ConfigOptions,