Fix physical distinctness expression nullability

datafusion/physical-expr/src/expressions/binary.rs

@@ -528,7 +528,13 @@ impl PhysicalExpr for BinaryExpr {
     }
 
     fn nullable(&self, input_schema: &Schema) -> Result<bool> {
-        Ok(self.left.nullable(input_schema)? || self.right.nullable(input_schema)?)
+        match self.op {
+            Operator::IsDistinctFrom | Operator::IsNotDistinctFrom => Ok(false),
+            _ => {
+                Ok(self.left.nullable(input_schema)?
+                    || self.right.nullable(input_schema)?)
+            }
+        }
     }
 
     fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
@@ -4179,6 +4185,27 @@ mod tests {
         apply_logic_op(&schema, &a, &b, Operator::IsDistinctFrom, expected).unwrap();
     }
 
+    #[test]
+    fn distinct_from_op_nullability() -> Result<()> {
+        let schema = Schema::new(vec![
+            Field::new("nullable", DataType::Boolean, true),
+            Field::new("non_nullable", DataType::Boolean, false),
+        ]);
+        let cases = [
+            (Operator::IsDistinctFrom, "nullable", false),
+            (Operator::IsNotDistinctFrom, "nullable", false),
+            (Operator::Eq, "nullable", true),
+            (Operator::Eq, "non_nullable", false),
+        ];
+
+        for (op, column, expected) in cases {
+            let expr = BinaryExpr::new(col(column, &schema)?, op, lit(true));
+            assert_eq!(expr.nullable(&schema)?, expected, "{op} with {column}");
+        }
+
+        Ok(())
+    }
+
     #[test]
     fn is_not_distinct_from_op_bool() {
         let (schema, a, b) = bool_test_arrays();

datafusion/sqllogictest/test_files/scalar.slt

@@ -1853,6 +1853,24 @@ true
 false
 true
 
+# Aggregate CSE preserves the non-nullability of truth predicates
+# issue: https://github.com/apache/datafusion/issues/24096
+query IIIII
+SELECT
+  SUM(CASE WHEN b IS TRUE THEN 1 ELSE 0 END),
+  COUNT(CASE WHEN b IS TRUE THEN 1 END),
+  SUM(CASE WHEN b IS FALSE THEN 1 ELSE 0 END),
+  SUM(CASE WHEN b IS NOT TRUE THEN 1 ELSE 0 END),
+  SUM(CASE WHEN b IS NOT FALSE THEN 1 ELSE 0 END)
+FROM (
+  VALUES
+    (TRUE),
+    (FALSE),
+    (CAST(NULL AS BOOLEAN))
+) AS t(b);
+----
+1 1 1 2 2
+
 
 # query_without_from()