feat: add UnboundPredicate::negate()

Issue: apache#150

crates/iceberg/src/expr/mod.rs

@@ -64,3 +64,26 @@ impl Display for PredicateOperator {
         }
     }
 }
+
+impl PredicateOperator {
+
+    /// Returns the predicate that is the inverse of self
+    pub fn negate(self) -> PredicateOperator {
+        match self {
+            PredicateOperator::IsNull => PredicateOperator::NotNull,
+            PredicateOperator::NotNull => PredicateOperator::IsNull,
+            PredicateOperator::IsNan => PredicateOperator::NotNan,
+            PredicateOperator::NotNan => PredicateOperator::IsNan,
+            PredicateOperator::LessThan => PredicateOperator::GreaterThanOrEq,
+            PredicateOperator::LessThanOrEq => PredicateOperator::GreaterThan,
+            PredicateOperator::GreaterThan => PredicateOperator::LessThanOrEq,
+            PredicateOperator::GreaterThanOrEq => PredicateOperator::LessThan,
+            PredicateOperator::Eq => PredicateOperator::NotEq,
+            PredicateOperator::NotEq => PredicateOperator::Eq,
+            PredicateOperator::In => PredicateOperator::NotIn,
+            PredicateOperator::NotIn => PredicateOperator::In,
+            PredicateOperator::StartsWith => PredicateOperator::NotStartsWith,
+            PredicateOperator::NotStartsWith => PredicateOperator::StartsWith,
+        }
+    }
+}

crates/iceberg/src/expr/predicate.rs

@@ -21,6 +21,7 @@
 
 use crate::expr::{BoundReference, PredicateOperator, UnboundReference};
 use crate::spec::Datum;
+use itertools::Itertools;
 use std::collections::HashSet;
 use std::fmt::{Debug, Display, Formatter};
 use std::ops::Not;
@@ -101,6 +102,24 @@ pub struct SetExpression<T: Debug> {
     literals: HashSet<Datum>,
 }
 
+impl<T: Debug> SetExpression<T> {
+    pub(crate) fn new(op: PredicateOperator, term: T, literals: HashSet<Datum>) -> Self {
+        Self { op, term, literals }
+    }
+}
+
+impl<T: Display + Debug> Display for SetExpression<T> {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{} {} ({})",
+            self.term,
+            self.op,
+            self.literals.iter().join(", ")
+        )
+    }
+}
+
 /// Unbound predicate expression before binding to a schema.
 #[derive(Debug)]
 pub enum UnboundPredicate {
@@ -195,6 +214,68 @@ impl UnboundPredicate {
     pub fn or(self, other: UnboundPredicate) -> UnboundPredicate {
         UnboundPredicate::Or(LogicalExpression::new([Box::new(self), Box::new(other)]))
     }
+
+    /// Returns a predicate representing the negation ('NOT') of this one,
+    /// by using inverse predicates rather than wrapping in a `NOT`.
+    /// Used for `NOT` elimination.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use std::ops::Bound::Unbounded;
+    /// use iceberg::expr::BoundPredicate::Unary;
+    /// use iceberg::expr::{LogicalExpression, UnboundPredicate, UnboundReference};
+    /// use iceberg::spec::Datum;
+    /// let expr1 = UnboundReference::new("a").less_than(Datum::long(10));
+    /// let expr2 = UnboundReference::new("b").less_than(Datum::long(5)).and(UnboundReference::new("c").less_than(Datum::long(10)));
+    ///
+    /// let result = expr1.negate();
+    /// assert_eq!(&format!("{result}"), "a >= 10");
+    ///
+    /// let result = expr2.negate();
+    /// assert_eq!(&format!("{result}"), "(b >= 5) OR (c >= 10)");
+    /// ```
+    /// Returns a predicate representing the negation ('NOT') of this one,
+    /// by using inverse predicates rather than wrapping in a `NOT`.
+    /// Used for `NOT` elimination.
+    ///
+    /// # Example
+    ///
+    /// ```rust
+    /// use std::ops::Bound::Unbounded;
+    /// use iceberg::expr::BoundPredicate::Unary;
+    /// use iceberg::expr::{LogicalExpression, UnboundPredicate, UnboundReference};
+    /// use iceberg::spec::Datum;
+    /// let expr1 = UnboundReference::new("a").less_than(Datum::long(10));
+    /// let expr2 = UnboundReference::new("b").less_than(Datum::long(5)).and(UnboundReference::new("c").less_than(Datum::long(10)));
+    ///
+    /// let result = expr1.negate();
+    /// assert_eq!(&format!("{result}"), "a >= 10");
+    ///
+    /// let result = expr2.negate();
+    /// assert_eq!(&format!("{result}"), "(b >= 5) OR (c >= 10)");
+    /// ```
+    pub fn negate(self) -> UnboundPredicate {
+        match self {
+            UnboundPredicate::And(expr) => UnboundPredicate::Or(LogicalExpression::new(
+                expr.inputs.map(|expr| Box::new(expr.negate())),
+            )),
+            UnboundPredicate::Or(expr) => UnboundPredicate::And(LogicalExpression::new(
+                expr.inputs.map(|expr| Box::new(expr.negate())),
+            )),
+            UnboundPredicate::Not(expr) => {
+                let LogicalExpression { inputs: [input_0] } = expr;
+                input_0.negate()
+            }
+            UnboundPredicate::Unary(expr) => UnboundPredicate::Unary(UnaryExpression::new(expr.op.negate(), expr.term)),
+            UnboundPredicate::Binary(expr) => {
+                UnboundPredicate::Binary(BinaryExpression::new(expr.op.negate(), expr.term, expr.literal))
+            }
+            UnboundPredicate::Set(expr) => {
+                UnboundPredicate::Set(SetExpression::new(expr.op.negate(), expr.term, expr.literals))
+            }
+        }
+    }
 }
 
 impl Not for UnboundPredicate {