From db584a8d16f2fa79f4d9decb58e72388f2db042f Mon Sep 17 00:00:00 2001 From: siddu Date: Thu, 8 Oct 2026 00:51:13 +0530 Subject: [PATCH 1/7] 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/7] 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/7] 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/7] 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/7] 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/7] 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, From 00a93d732741bf02fa02fbb4cdd7d0ca3a85a45c Mon Sep 17 00:00:00 2001 From: siddu Date: Tue, 13 Oct 2026 22:23:40 +0530 Subject: [PATCH 7/7] Remove accidentally committed local helper files --- .agents/rules/datafusion-guidelines.md | 30 ---------------------- ci-check.ps1 | 34 ------------------------- test_26162.slt | Bin 752 -> 0 bytes 3 files changed, 64 deletions(-) delete mode 100644 .agents/rules/datafusion-guidelines.md delete mode 100644 ci-check.ps1 delete mode 100644 test_26162.slt diff --git a/.agents/rules/datafusion-guidelines.md b/.agents/rules/datafusion-guidelines.md deleted file mode 100644 index 9b4a212df6af0..0000000000000 --- a/.agents/rules/datafusion-guidelines.md +++ /dev/null @@ -1,30 +0,0 @@ ---- -name: Apache DataFusion Contribution Guidelines -description: Instructions and context for AI agents working in the Apache DataFusion repository to ensure high quality, CI-compliant PRs. ---- - -# Apache DataFusion Guidelines - -When working in the Apache DataFusion repository, you MUST follow these guidelines to ensure the user's PR passes CI checks and Codecov. - -## 1. Local CI Pre-checks -Before telling the user a task is complete, always ensure you have verified the following or advise the user to run them: -1. **Clippy:** DataFusion has strict clippy rules. Run `cargo clippy --all-targets --workspace -- -D warnings`. - - *Note:* DataFusion uses `#[expect(...)]` instead of `#[allow(...)]` for lint allowances to prevent stale attributes. -2. **Formatting:** Always run `cargo fmt --all`. -3. **Tests:** Ensure core tests pass. Run `cargo test -p datafusion --workspace`. - -## 2. Code Coverage (Codecov) -DataFusion strictly enforces Codecov. -- Never leave dead or unreachable code. If you add a guard/check that can never be hit, remove it, as Codecov will flag the untested `false` branch. -- When changing core logic, always add or update a test in the corresponding `tests/` module to cover the newly added logic. -- Pay attention to `unwrap()`, `panic!()`, or early returns, as they often require explicit test coverage. - -## 3. Architecture & Style -- DataFusion makes heavy use of `Arc`, `Arc`, and standard Arrow types. -- Be careful with `as _` type coercions; prefer explicit typing like `let exec: Arc = ...` to avoid type inference errors in tests. -- When working on the physical optimizer, keep in mind that rules are applied sequentially and can mutate or swap plans. Verify invariants early in the rules (e.g., presence of dynamic filters). - -## 4. Git Etiquette -- Keep commits focused and provide clear, descriptive commit messages. -- If you push changes directly for the user using their PAT, always verify tests compile first to avoid polluting the PR with broken commits. diff --git a/ci-check.ps1 b/ci-check.ps1 deleted file mode 100644 index 92f97e2146d14..0000000000000 --- a/ci-check.ps1 +++ /dev/null @@ -1,34 +0,0 @@ -param ( - [switch]$Help -) - -if ($Help) { - Write-Host "Usage: .\ci-check.ps1" - Write-Host "Runs formatting, clippy, and cargo tests to ensure the PR is ready for GitHub." - exit -} - -Write-Host "Running DataFusion CI Checks locally..." -ForegroundColor Cyan - -Write-Host "`n[1/3] Running formatting (cargo fmt)..." -ForegroundColor Yellow -cargo fmt --all -if ($LASTEXITCODE -ne 0) { - Write-Host "Formatting failed!" -ForegroundColor Red - exit 1 -} - -Write-Host "`n[2/3] Running clippy (cargo clippy)..." -ForegroundColor Yellow -cargo clippy --all-targets --workspace -- -D warnings -if ($LASTEXITCODE -ne 0) { - Write-Host "Clippy found warnings/errors!" -ForegroundColor Red - exit 1 -} - -Write-Host "`n[3/3] Running tests (cargo test)..." -ForegroundColor Yellow -cargo test --workspace -if ($LASTEXITCODE -ne 0) { - Write-Host "Tests failed!" -ForegroundColor Red - exit 1 -} - -Write-Host "`nAll checks passed! You are ready to push to GitHub." -ForegroundColor Green diff --git a/test_26162.slt b/test_26162.slt deleted file mode 100644 index f8fb6b9cc4071e6436dc8fa43f09f593ae63b742..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 752 zcmb7?OH0E*6ot>a;D4B9sv;(-4-gk=RVbt_v{rB>u|9C4Vk-FS)$fjEN{1SBG52}y z<(|WQeb&0yQcicO@Rc6)u3Are(X;mGeef34mZ4^-H$D2f8@>9>x(%Xc+bOT76LmDu zROc#miFMdQeU(10*|MOPu)%4lG2LCF+mLmM3eB)r_)>G`EK8JnWUj9Fi8Algkmo+n zLf(t;2NC`-!XHKW;AJmXg5Ie?XeVZ-bKJti7aYxvJ$KBPAGvn5jwmh00_ zJ5*Ec|GK9`-=WZ7cV#WR{<9}<*<)I*Guh_Gl>$lB-K3H!Izh8D6qyqDb&a(@Qli?3 TXZo5l_FFa~P5FwyxWzT!vcq9V