feat: conjunctive & negative normal forms (#1857)

NNF: `Not` only appears around "terminal" nodes (i.e. nodes other than
the binary expressions).

CNF: A conjunction of disjunctions of NNF expressions.

Author: danking

vortex-expr/src/forms/cnf.rs

@@ -0,0 +1,198 @@
+use vortex_error::VortexResult;
+
+use super::nnf::nnf;
+use crate::traversal::{Node as _, NodeVisitor, TraversalOrder};
+use crate::{BinaryExpr, ExprRef, Operator};
+
+/// Return an equivalent expression in Conjunctive Normal Form (CNF).
+///
+/// A CNF expression is a vector of vectors. The outer vector is a conjunction. The inner vectors
+/// are disjunctions. Neither [Operator::And] nor [Operator::Or] may appear in the
+/// disjunctions. Moreover, each disjunction in a CNF expression must be in Negative Normal Form.
+///
+/// # Examples
+///
+/// All the NNF examples also apply to CNF, for example double negation is removed entirely:
+///
+/// ```
+/// use vortex_expr::{not, col};
+/// use vortex_expr::forms::cnf::cnf;
+///
+/// let double_negation = not(not(col("a")));
+/// let cnfed = cnf(double_negation).unwrap();
+/// assert_eq!(cnfed, vec![vec![col("a")]]);
+/// ```
+///
+/// Unlike NNF, CNF, lifts conjunctions to the top-level and distributions disjunctions such that
+/// there is at most one disjunction for each conjunction operand:
+///
+/// ```
+/// use vortex_expr::{not, col, or, and};
+/// use vortex_expr::forms::cnf::cnf;
+///
+/// assert_eq!(
+///     cnf(
+///         or(
+///             or(
+///                 and(col("a"), col("b")),
+///                 col("c")
+///             ),
+///             col("d")
+///         )
+///     ).unwrap(),
+///     vec![
+///         vec![col("a"), col("c"), col("d")],
+///         vec![col("b"), col("c"), col("d")]
+///     ]
+/// );
+/// ```
+///
+/// ```
+/// use vortex_expr::{not, col, or, and};
+/// use vortex_expr::forms::cnf::cnf;
+///
+/// assert_eq!(
+///     cnf(
+///         or(
+///             and(col("a"), col("b")),
+///             col("c"),
+///         )
+///     ).unwrap(),
+///     vec![
+///         vec![col("a"), col("c")],
+///         vec![col("b"), col("c")],
+///     ]
+/// );
+/// ```
+///
+/// Vortex extends the CNF definition to any Boolean-valued expression, even ones with non-Boolean
+/// parameters:
+///
+/// ```
+/// use vortex_expr::{not, col, or, and, gt_eq, lit, not_eq, lt, eq};
+/// use vortex_expr::forms::cnf::cnf;
+/// use itertools::Itertools;
+///
+/// assert_eq!(
+///     cnf(
+///         or(
+///             and(
+///                 gt_eq(col("earnings"), lit(50_000)),
+///                 not_eq(col("role"), lit("Manager"))
+///             ),
+///             col("special_flag")
+///         ),
+///     ).unwrap(),
+///     vec![
+///         vec![
+///             gt_eq(col("earnings"), lit(50_000)),
+///             col("special_flag")
+///         ],
+///         vec![
+///             not_eq(col("role"), lit("Manager")),
+///             col("special_flag")
+///         ]
+///     ]
+/// );
+/// ```
+///
+/// ```
+/// use vortex_expr::{not, col, or, and, gt_eq, lit, not_eq, lt, eq};
+/// use vortex_expr::forms::cnf::cnf;
+/// use itertools::Itertools;
+///
+/// assert_eq!(
+///     cnf(
+///         or(
+///             or(
+///                 and(
+///                     gt_eq(col("earnings"), lit(50_000)),
+///                     not_eq(col("role"), lit("Manager"))
+///                 ),
+///                 col("special_flag")
+///             ),
+///             and(
+///                 lt(col("tenure"), lit(5)),
+///                 eq(col("role"), lit("Engineer"))
+///             ),
+///         )
+///     ).unwrap(),
+///     vec![
+///         vec![
+///             gt_eq(col("earnings"), lit(50_000)),
+///             col("special_flag"),
+///             lt(col("tenure"), lit(5)),
+///         ],
+///         vec![
+///             gt_eq(col("earnings"), lit(50_000)),
+///             col("special_flag"),
+///             eq(col("role"), lit("Engineer")),
+///         ],
+///         vec![
+///             not_eq(col("role"), lit("Manager")),
+///             col("special_flag"),
+///             lt(col("tenure"), lit(5)),
+///         ],
+///         vec![
+///             not_eq(col("role"), lit("Manager")),
+///             col("special_flag"),
+///             eq(col("role"), lit("Engineer")),
+///         ],
+///     ]
+/// );
+/// ```
+///
+pub fn cnf(expr: ExprRef) -> VortexResult<Vec<Vec<ExprRef>>> {
+    let nnf = nnf(expr)?;
+
+    let mut visitor = CNFVisitor::default();
+    nnf.accept(&mut visitor)?;
+    Ok(visitor.finish())
+}
+
+#[derive(Default)]
+struct CNFVisitor {
+    conjuncts_of_disjuncts: Vec<Vec<ExprRef>>,
+}
+
+impl CNFVisitor {
+    fn finish(self) -> Vec<Vec<ExprRef>> {
+        self.conjuncts_of_disjuncts
+    }
+}
+
+impl NodeVisitor<'_> for CNFVisitor {
+    type NodeTy = ExprRef;
+
+    fn visit_down(&mut self, node: &ExprRef) -> VortexResult<TraversalOrder> {
+        if let Some(binary_expr) = node.as_any().downcast_ref::<BinaryExpr>() {
+            match binary_expr.op() {
+                Operator::And => return Ok(TraversalOrder::Continue),
+                Operator::Or => {
+                    let mut visitor = CNFVisitor::default();
+                    binary_expr.lhs().accept(&mut visitor)?;
+                    let lhs_conjuncts = visitor.finish();
+
+                    let mut visitor = CNFVisitor::default();
+                    binary_expr.rhs().accept(&mut visitor)?;
+                    let rhs_conjuncts = visitor.finish();
+
+                    self.conjuncts_of_disjuncts
+                        .extend(lhs_conjuncts.iter().flat_map(|lhs_disjunct| {
+                            rhs_conjuncts.iter().map(|rhs_disjunct| {
+                                let mut lhs_copy = lhs_disjunct.clone();
+                                lhs_copy.extend(rhs_disjunct.iter().cloned());
+                                lhs_copy
+                            })
+                        }));
+
+                    return Ok(TraversalOrder::Skip);
+                }
+                _ => {}
+            }
+        }
+        // Anything other than And and Or are terminals from the perspective of CNF
+        self.conjuncts_of_disjuncts.push(vec![node.clone()]);
+        Ok(TraversalOrder::Skip)
+    }
+}

vortex-expr/src/forms/mod.rs

@@ -0,0 +1,2 @@
+pub mod cnf;
+pub mod nnf;

vortex-expr/src/forms/nnf.rs

@@ -0,0 +1,151 @@
+use vortex_error::VortexResult;
+
+use crate::traversal::{FoldChildren, FoldDown, FoldUp, Folder, Node as _};
+use crate::{not, BinaryExpr, ExprRef, Not, Operator};
+
+/// Return an equivalent expression in Negative Normal Form (NNF).
+///
+/// In NNF, [Not] expressions may only contain terminal nodes such as [Literal](crate::Literal) or
+/// [Column](crate::Column). They *may not* contain [BinaryExpr], [Not], etc.
+///
+/// # Examples
+///
+/// Double negation is removed entirely:
+///
+/// ```
+/// use vortex_expr::{not, col};
+/// use vortex_expr::forms::nnf::nnf;
+///
+/// let double_negation = not(not(col("a")));
+/// let nnfed = nnf(double_negation).unwrap();
+/// assert_eq!(&nnfed, &col("a"));
+/// ```
+///
+/// Triple negation becomes single negation:
+///
+/// ```
+/// use vortex_expr::{not, col};
+/// use vortex_expr::forms::nnf::nnf;
+///
+/// let triple_negation = not(not(not(col("a"))));
+/// let nnfed = nnf(triple_negation).unwrap();
+/// assert_eq!(&nnfed, &not(col("a")));
+/// ```
+///
+/// Negation at a high-level is pushed to the leaves, likely increasing the total number nodes:
+///
+/// ```
+/// use vortex_expr::{not, col, and, or};
+/// use vortex_expr::forms::nnf::nnf;
+///
+/// assert_eq!(
+///     &nnf(not(and(col("a"), col("b")))).unwrap(),
+///     &or(not(col("a")), not(col("b")))
+/// );
+/// ```
+///
+/// In Vortex, NNF is extended beyond simple Boolean operators to any Boolean-valued operator:
+///
+/// ```
+/// use vortex_expr::{not, col, and, or, lt, lit, gt_eq};
+/// use vortex_expr::forms::nnf::nnf;
+///
+/// assert_eq!(
+///     &nnf(not(and(gt_eq(col("a"), lit(3)), col("b")))).unwrap(),
+///     &or(lt(col("a"), lit(3)), not(col("b")))
+/// );
+/// ```
+pub fn nnf(expr: ExprRef) -> VortexResult<ExprRef> {
+    let mut visitor = NNFVisitor::default();
+    Ok(expr.transform_with_context(&mut visitor, false)?.result())
+}
+
+#[derive(Default)]
+struct NNFVisitor {}
+
+impl Folder for NNFVisitor {
+    type NodeTy = ExprRef;
+    type Out = ExprRef;
+    type Context = bool;
+
+    fn visit_down(
+        &mut self,
+        node: &ExprRef,
+        negating: bool,
+    ) -> VortexResult<FoldDown<ExprRef, bool>> {
+        let node_any = node.as_any();
+        if node_any.downcast_ref::<Not>().is_some() {
+            return Ok(FoldDown::Continue(!negating));
+        } else if let Some(binary_expr) = node_any.downcast_ref::<BinaryExpr>() {
+            match binary_expr.op() {
+                Operator::And | Operator::Or => {
+                    return Ok(FoldDown::Continue(negating));
+                }
+                _ => {}
+            }
+        }
+
+        Ok(FoldDown::SkipChildren)
+    }
+
+    fn visit_up(
+        &mut self,
+        node: ExprRef,
+        negating: bool,
+        new_children: FoldChildren<ExprRef>,
+    ) -> VortexResult<FoldUp<ExprRef>> {
+        let node_any = node.as_any();
+        let new_node = if node_any.downcast_ref::<Not>().is_some() {
+            let FoldChildren::Children(mut new_children) = new_children else {
+                unreachable!();
+            };
+            debug_assert_eq!(new_children.len(), 1);
+
+            new_children.remove(0)
+        } else if let Some(binary_expr) = node_any.downcast_ref::<BinaryExpr>() {
+            if !negating {
+                node
+            } else {
+                let new_op = match binary_expr.op() {
+                    Operator::Eq => Operator::NotEq,
+                    Operator::NotEq => Operator::Eq,
+                    Operator::Gt => Operator::Lte,
+                    Operator::Gte => Operator::Lt,
+                    Operator::Lt => Operator::Gte,
+                    Operator::Lte => Operator::Gt,
+                    Operator::And => Operator::Or,
+                    Operator::Or => Operator::And,
+                };
+                let (lhs, rhs) = match binary_expr.op() {
+                    Operator::Or | Operator::And => {
+                        let FoldChildren::Children(mut negated_children) = new_children else {
+                            unreachable!();
+                        };
+                        debug_assert_eq!(negated_children.len(), 2);
+                        let rhs = negated_children.remove(1);
+                        let lhs = negated_children.remove(0);
+                        (lhs, rhs)
+                    }
+                    _ => {
+                        let FoldChildren::Skipped = new_children else {
+                            unreachable!();
+                        };
+                        (binary_expr.lhs().clone(), binary_expr.rhs().clone())
+                    }
+                };
+                BinaryExpr::new_expr(lhs, new_op, rhs)
+            }
+        } else {
+            let FoldChildren::Skipped = new_children else {
+                unreachable!();
+            };
+
+            if negating {
+                not(node)
+            } else {
+                node
+            }
+        };
+        Ok(FoldUp::Continue(new_node))
+    }
+}

vortex-expr/src/lib.rs

@@ -5,6 +5,7 @@ use std::sync::Arc;
 mod binary;
 mod column;
 pub mod datafusion;
+pub mod forms;
 mod get_item;
 mod identity;
 mod like;
@@ -183,7 +184,7 @@ mod tests {
             "(($col1 < $col2) or ($col1 != $col2))"
         );
 
-        assert_eq!(Not::new_expr(col1.clone()).to_string(), "!$col1");
+        assert_eq!(not(col1.clone()).to_string(), "!$col1");
 
         assert_eq!(
             Select::include_expr(vec![Field::from("col1")], ident()).to_string(),

vortex-expr/src/like.rs

@@ -100,16 +100,14 @@ impl Eq for Like {}
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Arc;
-
     use vortex_array::array::BoolArray;
     use vortex_array::IntoArrayVariant;
 
-    use crate::{Identity, Not};
+    use crate::{ident, not};
 
     #[test]
     fn invert_booleans() {
-        let not_expr = Not::new_expr(Arc::new(Identity));
+        let not_expr = not(ident());
         let bools = BoolArray::from_iter([false, true, false, false, true, true]);
         assert_eq!(
             not_expr