#17801 Improve nullability reporting of case expressions

datafusion/core/src/physical_planner.rs

@@ -60,6 +60,7 @@ use crate::schema_equivalence::schema_satisfied_by;
 use arrow::array::{builder::StringBuilder, RecordBatch};
 use arrow::compute::SortOptions;
 use arrow::datatypes::Schema;
+use arrow_schema::Field;
 use datafusion_catalog::ScanArgs;
 use datafusion_common::display::ToStringifiedPlan;
 use datafusion_common::format::ExplainAnalyzeLevel;
@@ -2516,7 +2517,9 @@ impl<'a> OptimizationInvariantChecker<'a> {
         previous_schema: Arc<Schema>,
     ) -> Result<()> {
         // if the rule is not permitted to change the schema, confirm that it did not change.
-        if self.rule.schema_check() && plan.schema() != previous_schema {
+        if self.rule.schema_check()
+            && !is_allowed_schema_change(previous_schema.as_ref(), plan.schema().as_ref())
+        {
             internal_err!("PhysicalOptimizer rule '{}' failed. Schema mismatch. Expected original schema: {:?}, got new schema: {:?}",
                 self.rule.name(),
                 previous_schema,
@@ -2532,6 +2535,33 @@ impl<'a> OptimizationInvariantChecker<'a> {
     }
 }
 
+/// Checks if the change from `old` schema to `new` is allowed or not.
+/// The current implementation only allows nullability of individual fields to change
+/// from 'nullable' to 'not nullable'.
+fn is_allowed_schema_change(old: &Schema, new: &Schema) -> bool {
+    if new.metadata != old.metadata {
+        return false;
+    }
+
+    if new.fields.len() != old.fields.len() {
+        return false;
+    }
+
+    let new_fields = new.fields.iter().map(|f| f.as_ref());
+    let old_fields = old.fields.iter().map(|f| f.as_ref());
+    old_fields
+        .zip(new_fields)
+        .all(|(old, new)| is_allowed_field_change(old, new))
+}
+
+fn is_allowed_field_change(old_field: &Field, new_field: &Field) -> bool {
+    new_field.name() == old_field.name()
+        && new_field.data_type() == old_field.data_type()
+        && new_field.metadata() == old_field.metadata()
+        && (new_field.is_nullable() == old_field.is_nullable()
+            || !new_field.is_nullable())
+}
+
 impl<'n> TreeNodeVisitor<'n> for OptimizationInvariantChecker<'_> {
     type Node = Arc<dyn ExecutionPlan>;
 

datafusion/core/tests/tpcds_planning.rs

@@ -1052,9 +1052,12 @@ async fn regression_test(query_no: u8, create_physical: bool) -> Result<()> {
     for sql in &sql {
         let df = ctx.sql(sql).await?;
         let (state, plan) = df.into_parts();
-        let plan = state.optimize(&plan)?;
         if create_physical {
             let _ = state.create_physical_plan(&plan).await?;
+        } else {
+            // Run the logical optimizer even if we are not creating the physical plan
+            // to ensure it will properly succeed
+            let _ = state.optimize(&plan)?;
         }
     }
 

datafusion/expr-common/src/interval_arithmetic.rs

@@ -1762,6 +1762,16 @@ impl NullableInterval {
         }
     }
 
+    /// Return true if the value is definitely not true (either null or false).
+    pub fn is_certainly_not_true(&self) -> bool {
+        match self {
+            Self::Null { .. } => true,
+            Self::MaybeNull { values } | Self::NotNull { values } => {
+                values == &Interval::CERTAINLY_FALSE
+            }
+        }
+    }
+
     /// Return true if the value is definitely false (and not null).
     pub fn is_certainly_false(&self) -> bool {
         match self {
@@ -1967,6 +1977,7 @@ mod tests {
         operator::Operator,
     };
 
+    use crate::interval_arithmetic::NullableInterval;
     use arrow::datatypes::DataType;
     use datafusion_common::rounding::{next_down, next_up};
     use datafusion_common::{Result, ScalarValue};
@@ -4103,4 +4114,163 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_is_certainly_true() {
+        let test_cases = vec![
+            (
+                NullableInterval::Null {
+                    datatype: DataType::Boolean,
+                },
+                false,
+            ),
+            (
+                NullableInterval::MaybeNull {
+                    values: Interval::CERTAINLY_TRUE,
+                },
+                false,
+            ),
+            (
+                NullableInterval::MaybeNull {
+                    values: Interval::CERTAINLY_FALSE,
+                },
+                false,
+            ),
+            (
+                NullableInterval::MaybeNull {
+                    values: Interval::UNCERTAIN,
+                },
+                false,
+            ),
+            (
+                NullableInterval::NotNull {
+                    values: Interval::CERTAINLY_TRUE,
+                },
+                true,
+            ),
+            (
+                NullableInterval::NotNull {
+                    values: Interval::CERTAINLY_FALSE,
+                },
+                false,
+            ),
+            (
+                NullableInterval::NotNull {
+                    values: Interval::UNCERTAIN,
+                },
+                false,
+            ),
+        ];
+
+        for (interval, expected) in test_cases {
+            let result = interval.is_certainly_true();
+            assert_eq!(result, expected, "Failed for interval: {interval}",);
+        }
+    }
+
+    #[test]
+    fn test_is_certainly_not_true() {
+        let test_cases = vec![
+            (
+                NullableInterval::Null {
+                    datatype: DataType::Boolean,
+                },
+                true,
+            ),
+            (
+                NullableInterval::MaybeNull {
+                    values: Interval::CERTAINLY_TRUE,
+                },
+                false,
+            ),
+            (
+                NullableInterval::MaybeNull {
+                    values: Interval::CERTAINLY_FALSE,
+                },
+                true,
+            ),
+            (
+                NullableInterval::MaybeNull {
+                    values: Interval::UNCERTAIN,
+                },
+                false,
+            ),
+            (
+                NullableInterval::NotNull {
+                    values: Interval::CERTAINLY_TRUE,
+                },
+                false,
+            ),
+            (
+                NullableInterval::NotNull {
+                    values: Interval::CERTAINLY_FALSE,
+                },
+                true,
+            ),
+            (
+                NullableInterval::NotNull {
+                    values: Interval::UNCERTAIN,
+                },
+                false,
+            ),
+        ];
+
+        for (interval, expected) in test_cases {
+            let result = interval.is_certainly_not_true();
+            assert_eq!(result, expected, "Failed for interval: {interval}",);
+        }
+    }
+
+    #[test]
+    fn test_is_certainly_false() {
+        let test_cases = vec![
+            (
+                NullableInterval::Null {
+                    datatype: DataType::Boolean,
+                },
+                false,
+            ),
+            (
+                NullableInterval::MaybeNull {
+                    values: Interval::CERTAINLY_TRUE,
+                },
+                false,
+            ),
+            (
+                NullableInterval::MaybeNull {
+                    values: Interval::CERTAINLY_FALSE,
+                },
+                false,
+            ),
+            (
+                NullableInterval::MaybeNull {
+                    values: Interval::UNCERTAIN,
+                },
+                false,
+            ),
+            (
+                NullableInterval::NotNull {
+                    values: Interval::CERTAINLY_TRUE,
+                },
+                false,
+            ),
+            (
+                NullableInterval::NotNull {
+                    values: Interval::CERTAINLY_FALSE,
+                },
+                true,
+            ),
+            (
+                NullableInterval::NotNull {
+                    values: Interval::UNCERTAIN,
+                },
+                false,
+            ),
+        ];
+
+        for (interval, expected) in test_cases {
+            let result = interval.is_certainly_false();
+            assert_eq!(result, expected, "Failed for interval: {interval}",);
+        }
+    }
 }

datafusion/expr/Cargo.toml

@@ -45,6 +45,7 @@ sql = ["sqlparser"]
 [dependencies]
 arrow = { workspace = true }
 async-trait = { workspace = true }
+bitflags = "2.9.4"
 chrono = { workspace = true }
 datafusion-common = { workspace = true, default-features = false }
 datafusion-doc = { workspace = true }

datafusion/expr/src/expr_fn.rs

@@ -341,6 +341,11 @@ pub fn is_null(expr: Expr) -> Expr {
     Expr::IsNull(Box::new(expr))
 }
 
+/// Create is not null expression
+pub fn is_not_null(expr: Expr) -> Expr {
+    Expr::IsNotNull(Box::new(expr))
+}
+
 /// Create is true expression
 pub fn is_true(expr: Expr) -> Expr {
     Expr::IsTrue(Box::new(expr))