Support Interval analysis for OR expressions

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

@@ -344,7 +344,7 @@ impl PhysicalExpr for BinaryExpr {
         let right_interval = children[1];
 
         let (left, right) = if self.op.is_logic_operator() {
-            // TODO: Currently, this implementation only supports the AND operator
+            // TODO: Currently, this implementation only supports the AND/OR operator
             //       and does not require any further propagation. In the future,
             //       upon adding support for additional logical operators, this
             //       method will require modification to support propagating the

datafusion/physical-expr/src/intervals/interval_aritmetic.rs

@@ -432,6 +432,27 @@ impl Interval {
         }
     }
 
+    /// Compute the logical disjunction of this (boolean) interval with the given boolean interval.
+    pub(crate) fn or<T: Borrow<Interval>>(&self, other: T) -> Result<Interval> {
+        let rhs = other.borrow();
+        if self.get_datatype()? != DataType::Boolean
+            || rhs.get_datatype()? != DataType::Boolean
+        {
+            return internal_err!(
+                "Incompatible types for logical disjunction: {self:?}, {rhs:?}"
+            );
+        }
+
+        // if either bound is always true, the result is always true.
+        if self.is_certainly_true() || rhs.is_certainly_true() {
+            Ok(Interval::CERTAINLY_TRUE)
+        } else if self.is_certainly_false() && rhs.is_certainly_false() {
+            Ok(Interval::CERTAINLY_FALSE)
+        } else {
+            Ok(Interval::UNCERTAIN)
+        }
+    }
+
     /// Compute the logical negation of this (boolean) interval.
     pub(crate) fn not(&self) -> Result<Self> {
         if !matches!(self.get_datatype()?, DataType::Boolean) {
@@ -519,6 +540,20 @@ impl Interval {
         ))
     }
 
+    /// return true if this is a Boolean interval that is known to be `true`
+    pub fn is_certainly_true(&self) -> bool {
+        self == &Self::CERTAINLY_TRUE
+    }
+
+    /// return true if this is a Boolean interval that may be either `true` or `false`
+    pub fn is_uncertain(&self) -> bool {
+        self == &Self::UNCERTAIN
+    }
+
+    /// return true if this is a Boolean interval that is known to be `false`
+    pub fn is_certainly_false(&self) -> bool {
+        self == &Self::CERTAINLY_FALSE
+    }
     pub const CERTAINLY_FALSE: Interval = Interval {
         lower: IntervalBound::new_closed(ScalarValue::Boolean(Some(false))),
         upper: IntervalBound::new_closed(ScalarValue::Boolean(Some(false))),
@@ -738,6 +773,7 @@ pub fn apply_operator(op: &Operator, lhs: &Interval, rhs: &Interval) -> Result<I
         Operator::Lt => Ok(lhs.lt(rhs)),
         Operator::LtEq => Ok(lhs.lt_eq(rhs)),
         Operator::And => lhs.and(rhs),
+        Operator::Or => lhs.or(rhs),
         Operator::Plus => lhs.add(rhs),
         Operator::Minus => lhs.sub(rhs),
         _ => Ok(Interval::default()),
@@ -1221,6 +1257,8 @@ mod tests {
 
     #[test]
     fn and_test() -> Result<()> {
+        // check that (lhs AND rhs) is equal to result
+        // (lhs_lower, lhs_upper, rhs_lower, rhs_upper, lower_result, upper_result)
         let cases = vec![
             (false, true, false, false, false, false),
             (false, false, false, true, false, false),
@@ -1240,6 +1278,35 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn or_test() -> Result<()> {
+        // check that (lhs OR rhs) is equal to result
+        // ((lhs_lower, lhs_upper), (rhs_lower, rhs_upper), (lower_result, upper_result))
+        let cases = vec![
+            // Note there are only three valid boolean intervals: [false, false], [false, true] and [true, true],
+            ((false, false), (false, false), (false, false)),
+            ((false, false), (false, true), (false, true)),
+            ((false, false), (true, true), (true, true)),
+            ((false, true), (false, false), (false, true)),
+            ((false, true), (false, true), (false, true)),
+            ((false, true), (true, true), (true, true)),
+            ((true, true), (false, false), (true, true)),
+            ((true, true), (false, true), (true, true)),
+            ((true, true), (true, true), (true, true)),
+        ];
+
+        for case in cases {
+            let (lhs, rhs, expected) = case;
+            println!("case: {:?}", case);
+            let lhs = closed_closed(Some(lhs.0), Some(lhs.1));
+            let rhs = closed_closed(Some(rhs.0), Some(rhs.1));
+            let expected = closed_closed(Some(expected.0), Some(expected.1));
+            println!("lhs: {:?}, rhs: {:?}, expected: {:?}", lhs, rhs, expected);
+            assert_eq!(lhs.or(rhs)?, expected,);
+        }
+        Ok(())
+    }
+
     #[test]
     fn add_test() -> Result<()> {
         let cases = vec![