diff --git a/datafusion/core/tests/issue_25978.rs b/datafusion/core/tests/issue_25978.rs new file mode 100644 index 0000000000000..2e5005e37dcac --- /dev/null +++ b/datafusion/core/tests/issue_25978.rs @@ -0,0 +1,27 @@ +use datafusion::prelude::*; +use datafusion::error::Result; + +#[tokio::test] +async fn test_issue_25978() -> Result<()> { + let ctx = SessionContext::new(); + + // CREATE TABLE a (k VARCHAR NOT NULL, x VARCHAR) AS VALUES ('k1', 'y'), ('k2', NULL); + // CREATE TABLE b (k VARCHAR NOT NULL) AS VALUES ('k3'); + ctx.sql("CREATE TABLE a (k VARCHAR NOT NULL, x VARCHAR) AS VALUES ('k1', 'y'), ('k2', NULL)").await?.collect().await?; + ctx.sql("CREATE TABLE b (k VARCHAR NOT NULL) AS VALUES ('k3')").await?.collect().await?; + + // SELECT k, bool_or(f) AS f FROM ( + // SELECT k, coalesce(x = 'y', FALSE) AS f FROM a + // UNION ALL + // SELECT k, TRUE AS f FROM b + // ) u GROUP BY k; + let df = ctx.sql("SELECT k, bool_or(f) AS f FROM ( + SELECT k, coalesce(x = 'y', FALSE) AS f FROM a + UNION ALL + SELECT k, TRUE AS f FROM b + ) u GROUP BY k").await?; + + df.collect().await?; + + Ok(()) +} diff --git a/datafusion/core/tests/physical_optimizer/join_selection.rs b/datafusion/core/tests/physical_optimizer/join_selection.rs index d4561decc6331..b259bd866ae5f 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,123 @@ 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), + )); + + #[expect(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(()) +} + +#[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/expr/src/expr_schema.rs b/datafusion/expr/src/expr_schema.rs index 8aaca2b72b2c6..31031e6ff0647 100644 --- a/datafusion/expr/src/expr_schema.rs +++ b/datafusion/expr/src/expr_schema.rs @@ -435,6 +435,24 @@ impl ExprSchemable for Expr { Expr::ScalarSubquery(subquery) => Ok(scalar_subquery_nullable(subquery)), Expr::BinaryExpr(BinaryExpr { left, right, op }) => match op { Operator::IsDistinctFrom | Operator::IsNotDistinctFrom => Ok(false), + Operator::And => { + if contains_is_not_null(left.as_ref(), right.as_ref()) { + return Ok(false); + } + if contains_is_not_null(right.as_ref(), left.as_ref()) { + return Ok(false); + } + Ok(left.nullable(input_schema)? || right.nullable(input_schema)?) + } + Operator::Or => { + if contains_is_null(left.as_ref(), right.as_ref()) { + return Ok(false); + } + if contains_is_null(right.as_ref(), left.as_ref()) { + return Ok(false); + } + Ok(left.nullable(input_schema)? || right.nullable(input_schema)?) + } _ => Ok(left.nullable(input_schema)? || right.nullable(input_schema)?), }, Expr::Like(Like { expr, pattern, .. }) @@ -1623,3 +1641,23 @@ mod tests { } } } + +fn contains_is_not_null(expr: &Expr, target: &Expr) -> bool { + if let Expr::IsNotNull(inner) = expr { + return inner.as_ref() == target; + } + if let Expr::BinaryExpr(BinaryExpr { left, right, op: Operator::And }) = expr { + return contains_is_not_null(left, target) || contains_is_not_null(right, target); + } + false +} + +fn contains_is_null(expr: &Expr, target: &Expr) -> bool { + if let Expr::IsNull(inner) = expr { + return inner.as_ref() == target; + } + if let Expr::BinaryExpr(BinaryExpr { left, right, op: Operator::Or }) = expr { + return contains_is_null(left, target) || contains_is_null(right, target); + } + false +} diff --git a/datafusion/physical-optimizer/src/join_selection.rs b/datafusion/physical-optimizer/src/join_selection.rs index 2b342bece7040..a0eb6997a4c6b 100644 --- a/datafusion/physical-optimizer/src/join_selection.rs +++ b/datafusion/physical-optimizer/src/join_selection.rs @@ -301,32 +301,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 +532,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