fix: SortPreservingMerge sanity check rejects valid ORDER BY with CASE expression (#18342)

## Which issue does this PR close?

- Closes #18327

## Rationale for this change

ORDER BY with a CASE statement didn't always work, raising a sanity
check error in SortPreservingMergeExec. The plan showed that the
partitions all had the same ordering, but for whatever reason they were
not detected as being equal. Using a single partition succeeded always.

## What changes are included in this PR?

The changes are non-obvious and I spent a lot of time bisecting/debug
printing and landed on a failure in bounds checking with boolean
interval arithmetic. Returning UNCERTAIN if either leg of the interval
is NULL resolves the upstream issue where CASE statements end up being
deemed Unordered.

My rust-fu is hobbyist at best, so while this appears to resolve my
issue I cannot for-certain exclaim that I've solved it all (Claude 4.5
agrees with my fix, but that's not an indication its any good). I'm also
reasonably certain my unit tests are more ham fisted than necessary.

## Are these changes tested?

1. Yes, unit tests have been added.

## Are there any user-facing changes?

This does not change any behavior beyond resolving a bug with a valid
SQL statement.

---------

Co-authored-by: Andrew Lamb <[email protected]>

Author: watford-ep

datafusion/expr-common/src/interval_arithmetic.rs

@@ -583,7 +583,9 @@ impl Interval {
                     upper: ScalarValue::Boolean(Some(upper)),
                 })
             }
-            _ => internal_err!("Incompatible data types for logical conjunction"),
+
+            // Return UNCERTAIN when intervals don't have concrete boolean bounds
+            _ => Ok(Self::UNCERTAIN),
         }
     }
 
@@ -606,7 +608,9 @@ impl Interval {
                     upper: ScalarValue::Boolean(Some(upper)),
                 })
             }
-            _ => internal_err!("Incompatible data types for logical disjunction"),
+
+            // Return UNCERTAIN when intervals don't have concrete boolean bounds
+            _ => Ok(Self::UNCERTAIN),
         }
     }
 
@@ -2517,6 +2521,64 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_and_or_with_normalized_boolean_intervals() -> Result<()> {
+        // Verify that NULL boolean bounds are normalized and don't cause errors
+        let from_nulls =
+            Interval::try_new(ScalarValue::Boolean(None), ScalarValue::Boolean(None))?;
+
+        assert!(from_nulls.or(&Interval::CERTAINLY_TRUE).is_ok());
+        assert!(from_nulls.and(&Interval::CERTAINLY_FALSE).is_ok());
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_and_null_boolean_intervals() -> Result<()> {
+        let null_interval =
+            Interval::try_new(ScalarValue::Boolean(None), ScalarValue::Boolean(None))?;
+
+        let and_result = null_interval.and(&Interval::CERTAINLY_FALSE)?;
+        assert_eq!(and_result, Interval::CERTAINLY_FALSE);
+
+        let and_result = Interval::CERTAINLY_FALSE.and(&null_interval)?;
+        assert_eq!(and_result, Interval::CERTAINLY_FALSE);
+
+        let and_result = null_interval.and(&Interval::CERTAINLY_TRUE)?;
+        assert_eq!(and_result, Interval::UNCERTAIN);
+
+        let and_result = Interval::CERTAINLY_TRUE.and(&null_interval)?;
+        assert_eq!(and_result, Interval::UNCERTAIN);
+
+        let and_result = null_interval.and(&null_interval)?;
+        assert_eq!(and_result, Interval::UNCERTAIN);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_or_null_boolean_intervals() -> Result<()> {
+        let null_interval =
+            Interval::try_new(ScalarValue::Boolean(None), ScalarValue::Boolean(None))?;
+
+        let or_result = null_interval.or(&Interval::CERTAINLY_FALSE)?;
+        assert_eq!(or_result, Interval::UNCERTAIN);
+
+        let or_result = Interval::CERTAINLY_FALSE.or(&null_interval)?;
+        assert_eq!(or_result, Interval::UNCERTAIN);
+
+        let or_result = null_interval.or(&Interval::CERTAINLY_TRUE)?;
+        assert_eq!(or_result, Interval::CERTAINLY_TRUE);
+
+        let or_result = Interval::CERTAINLY_TRUE.or(&null_interval)?;
+        assert_eq!(or_result, Interval::CERTAINLY_TRUE);
+
+        let or_result = null_interval.or(&null_interval)?;
+        assert_eq!(or_result, Interval::UNCERTAIN);
+
+        Ok(())
+    }
+
     #[test]
     fn intersect_test() -> Result<()> {
         let possible_cases = vec![

datafusion/sqllogictest/test_files/union.slt

@@ -953,3 +953,40 @@ drop table u1;
 
 statement count 0
 drop table u2;
+
+# repro for https://github.com/apache/datafusion/issues/18327
+# should not error
+query TITT
+  WITH typ(oid, typnamespace, typname, typtype) AS (
+                      SELECT * FROM (VALUES (1, 10, 't1', 'b'))
+            UNION ALL SELECT * FROM (VALUES (2, NULL, 't2', 'b'))
+            UNION ALL SELECT * FROM (VALUES (3, 12, 't3', NULL))
+       )
+       , ns(oid, nspname) AS (VALUES (1, 'ns1'), (2, 'ns2'))
+    SELECT ns.nspname, typ.oid, typ.typname, typ.typtype
+      FROM typ JOIN ns ON (ns.oid = typ.typnamespace)
+     WHERE typ.typtype IN ('b','r','m','e','d')
+     ORDER BY CASE WHEN typ.typtype IN ('b','e','p') THEN 0
+                   WHEN typ.typtype = 'r' THEN 1
+              END
+----
+
+# Add another row with a non-NULL value `m` which is retained by the
+# filter but not matching any WHEN branch m?
+query TITT
+  WITH typ(oid, typnamespace, typname, typtype) AS (
+                      SELECT * FROM (VALUES (1, 10, 't1', 'b'))
+            UNION ALL SELECT * FROM (VALUES (2, NULL, 't2', 'b'))
+            UNION ALL SELECT * FROM (VALUES (3, 12, 't3', NULL))
+            UNION ALL SELECT * FROM (VALUES (4, 40, 't3', 'm'))
+       ), ns(oid, nspname) AS (
+         VALUES (1, 'ns1'), (2, 'ns2'), (40, 'ns3')
+       )
+    SELECT ns.nspname, typ.oid, typ.typname, typ.typtype
+      FROM typ JOIN ns ON (ns.oid = typ.typnamespace)
+     WHERE typ.typtype IN ('b','r','m','e','d')
+     ORDER BY CASE WHEN typ.typtype IN ('b','e','p') THEN 0
+                   WHEN typ.typtype = 'r' THEN 1
+              END
+----
+ns3 4 t3 m