fix: correct nullability of AND/OR expressions to resolve plan mismatch in CSE

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(())
+}

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::<HashJoinExec>()
+        .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<dyn ExecutionPlan> = Arc::new(UnboundedExec::new(
+        None,
+        RecordBatch::new_empty(Arc::new(Schema::new(vec![Field::new(
+            "a",
+            DataType::Int32,
+            false,
+        )]))),
+        2,
+    ));
+    let right_exec: Arc<dyn ExecutionPlan> = 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::<HashJoinExec>()
+        .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(())
+}

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
+}

datafusion/physical-optimizer/src/join_selection.rs

@@ -301,32 +301,40 @@ fn statistical_join_selection_subrule(
     context: &dyn PhysicalOptimizerContext,
 ) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
     let transformed = if let Some(hash_join) = plan.downcast_ref::<HashJoinExec>() {
-        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<Arc<dyn ExecutionPlan>> {
     if let Some(hash_join) = input.downcast_ref::<HashJoinExec>()
+        && 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