perf(query): narrow each variable through an or of ands in where

crates/lance-csr/src/logical_plan.rs

@@ -1172,7 +1172,11 @@ impl<'a> LogicalPlanner<'a> {
         input: LogicalOperator,
         predicate: &BooleanExpression,
     ) -> Result<LogicalOperator> {
-        let mut predicate = predicate.clone();
+        let mut predicate = crate::pattern_order::implied_by_disjunctions(predicate)
+            .into_iter()
+            .fold(predicate.clone(), |p, implied| {
+                BooleanExpression::And(Box::new(p), Box::new(implied))
+            });
         let saved = self.seed_source.replace(input.clone());
         let input = self.rewrite_exists(input, &mut predicate);
         self.seed_source = saved;

crates/lance-csr/src/pattern_order.rs

@@ -111,6 +111,89 @@ fn conjuncts(expr: &BooleanExpression) -> Vec<&BooleanExpression> {
     }
 }
 
+/// The top-level `OR` disjuncts of a predicate.
+fn disjuncts(expr: &BooleanExpression) -> Vec<&BooleanExpression> {
+    match expr {
+        BooleanExpression::Or(left, right) => {
+            let mut out = disjuncts(left);
+            out.extend(disjuncts(right));
+            out
+        }
+        other => vec![other],
+    }
+}
+
+/// Predicates implied by `predicate`'s `OR`-of-`AND` conjuncts, one per variable they
+/// read: each disjunct without its conjuncts that read that variable, OR-ed back together.
+///
+/// `(a.id = $x AND b.id = $y) OR (a.id = $y AND b.id = $x)` implies `a.id = $x OR a.id = $y`
+/// (and the same for `b`). Adding that to the filter lets planning narrow `a` before
+/// expanding to `b`, where the whole `OR` can only run once both are bound. The original
+/// stays in the filter, so results are unchanged: a true disjunct makes each of its
+/// conjuncts, hence the implied disjunct, true. Only comparisons and `IN` without
+/// subqueries are copied, so no pattern or comprehension is planned twice.
+pub(crate) fn implied_by_disjunctions(predicate: &BooleanExpression) -> Vec<BooleanExpression> {
+    let copyable = |c: &BooleanExpression| match c {
+        BooleanExpression::Comparison { left, right, .. } => {
+            !crate::ast::needs_relational_evaluation(left)
+                && !crate::ast::needs_relational_evaluation(right)
+        }
+        BooleanExpression::In { expression, list } => {
+            !crate::ast::needs_relational_evaluation(expression)
+                && !list.iter().any(crate::ast::needs_relational_evaluation)
+        }
+        _ => false,
+    };
+    let mut implied = Vec::new();
+    for conjunct in conjuncts(predicate) {
+        let branches: Vec<_> = disjuncts(conjunct).into_iter().map(conjuncts).collect();
+        if branches.len() < 2 {
+            continue;
+        }
+        let mut variables = HashSet::new();
+        boolean_variables(conjunct, &mut variables);
+        let mut variables: Vec<_> = variables.into_iter().collect();
+        variables.sort_unstable();
+        for dropped in variables {
+            let reads = |c: &BooleanExpression| {
+                let mut names = HashSet::new();
+                boolean_variables(c, &mut names);
+                names.contains(dropped)
+            };
+            // Each branch keeps its copyable conjuncts that don't read `dropped`; a branch
+            // left with none implies nothing, and neither does a branch that lost none.
+            let mut weakened = false;
+            let mut kept_branches = Vec::with_capacity(branches.len());
+            for branch in &branches {
+                let kept: Vec<_> = branch
+                    .iter()
+                    .filter(|c| copyable(c) && !reads(c))
+                    .map(|c| (*c).clone())
+                    .collect();
+                weakened |= kept.len() < branch.len();
+                let Some(kept) = kept
+                    .into_iter()
+                    .reduce(|l, r| BooleanExpression::And(Box::new(l), Box::new(r)))
+                else {
+                    kept_branches.clear();
+                    break;
+                };
+                kept_branches.push(kept);
+            }
+            let Some(or) = kept_branches
+                .into_iter()
+                .reduce(|l, r| BooleanExpression::Or(Box::new(l), Box::new(r)))
+            else {
+                continue;
+            };
+            if weakened && !implied.contains(&or) {
+                implied.push(or);
+            }
+        }
+    }
+    implied
+}
+
 /// Selectivity of `conjunct` for `variable`, if it constrains only that variable.
 fn score_conjunct(
     conjunct: &BooleanExpression,
@@ -626,4 +709,54 @@ mod tests {
             None
         );
     }
+
+    fn implied(predicate: &str) -> Vec<BooleanExpression> {
+        let ast =
+            parse_cypher_query(&format!("MATCH (a), (b) WHERE {predicate} RETURN a")).unwrap();
+        implied_by_disjunctions(&ast.where_clause.unwrap().expression)
+    }
+
+    fn parsed(predicates: &[&str]) -> Vec<BooleanExpression> {
+        predicates
+            .iter()
+            .map(|p| {
+                let ast =
+                    parse_cypher_query(&format!("MATCH (a), (b) WHERE {p} RETURN a")).unwrap();
+                ast.where_clause.unwrap().expression
+            })
+            .collect()
+    }
+
+    #[test]
+    fn disjunctions_imply_one_predicate_per_dropped_variable() {
+        // Either direction of an edge between a and b: each end narrows on its own.
+        assert_eq!(
+            implied("(a.id = $x AND b.id = $y) OR (a.id = $y AND b.id = $x)"),
+            parsed(&["b.id = $y OR b.id = $x", "a.id = $x OR a.id = $y"]),
+        );
+        // A top-level conjunct beside the OR is untouched; only the OR is weakened.
+        assert_eq!(
+            implied("a.age > 1 AND ((a.id = 1 AND b.id = 2) OR (a.id IN [3, 4] AND b.id = 5))"),
+            parsed(&["b.id = 2 OR b.id = 5", "a.id = 1 OR a.id IN [3, 4]"]),
+        );
+        // Dropping a leaves every branch a conjunct on b; dropping b empties one branch.
+        assert_eq!(
+            implied("(a.id = 1 AND b.id = 2) OR b.id = 3"),
+            parsed(&["b.id = 2 OR b.id = 3"]),
+        );
+        // A subquery conjunct is never copied, but the rest of its branch still implies.
+        assert_eq!(
+            implied("(a.id = 1 AND EXISTS { (a)-[:R]->(b) }) OR (a.id = 2 AND b.id = 3)"),
+            parsed(&["a.id = 1 OR a.id = 2"]),
+        );
+    }
+
+    #[test]
+    fn disjunctions_that_imply_nothing_new() {
+        // A branch left with no conjunct implies nothing.
+        assert!(implied("a.id = 1 OR b.id = 2").is_empty());
+        // Nothing to drop: the implied predicate would be the OR itself.
+        assert!(implied("(a.id = 1 AND a.age = 2) OR a.id = 3").is_empty());
+        assert!(implied("a.id = 1 AND b.id = 2").is_empty());
+    }
 }

docs/internals/index.md

@@ -117,7 +117,7 @@ All paths are under
 | `path_match.rs`, `path_inventory.rs`, `path_value.rs` | Named paths, the edge inventory, Arrow path values | lance-csr |
 | `apply.rs` | `ApplyNode` and `ApplyExec` for CALL subqueries | lance-csr |
 | `key_statistics.rs` | `KeyStatisticsProvider` and `KeyStatisticsExec` | lance-csr |
-| `pattern_order.rs`, `pattern_alternatives.rs` | Re-rooting, alternation and multi-label branching | lance-csr |
+| `pattern_order.rs`, `pattern_alternatives.rs` | Re-rooting, predicates implied by an `OR`, alternation and multi-label branching | lance-csr |
 | `logical_plan.rs` | Graph logical operators | [lance-graph](https://github.com/lance-format/lance-graph), extended in lance-csr |
 | `datafusion_planner/` | Graph plan to DataFusion plan; `builder/expand_ops.rs` chooses CSR or join for each hop | [lance-graph](https://github.com/lance-format/lance-graph), extended in lance-csr |
 | `engine.rs` | `CypherEngine`: snapshots, session setup, refresh, plan template cache, SQL entry points | lance-csr |

docs/internals/optimizer.md

@@ -77,6 +77,30 @@ flowchart LR
 The same edges, endpoints, labels and variables are matched, and the full `WHERE` clause still applies above the
 match. Only the order of work changes.
 
+### Predicates implied by an `OR`
+
+A condition that accepts either direction of an edge between `a` and `b` is an `OR` whose branches each read both
+variables:
+
+```cypher
+WHERE (a.id = $x AND b.id = $y) OR (a.id = $y AND b.id = $x)
+```
+
+Neither re-rooting nor DataFusion's filter pushdown can use it, because no single conjunct reads only `a`. The filter
+waits until both ends are bound, so the match expands from every `a`. When the pattern comprehension or `OPTIONAL
+MATCH` around it is correlated with outer rows through this condition, that cost is paid for every distinct outer
+binding: the official LDBC IC14 text spent minutes this way on SF1.
+
+So when a `WHERE` is planned, `implied_by_disjunctions` in `pattern_order.rs` adds, for each variable the `OR` reads,
+the `OR` of each branch without the conjuncts that read that variable. Dropping `b` above leaves
+`a.id = $x OR a.id = $y`, which reads only `a` and outer values. DataFusion moves it below the expansion as an
+ordinary filter or join condition, so the expansion starts from at most two people per outer row.
+
+The added predicate is implied by the original, which stays in the filter, so the rows don't change: a true branch
+makes each of its conjuncts true, so the shortened branch is true too. Nothing is added when a branch would be left
+with no conjunct (`a.x = 1 OR b.y = 2` says nothing about `a` alone) or when no conjunct was dropped. Only
+comparisons and `IN` lists without subqueries are copied, so no pattern or comprehension is planned twice.
+
 ### Alternation and multi-label nodes become branches
 
 `-[:A|B]->` and `(m:Post:Comment)` are rewritten into branches with exactly one type per relationship and at most one

python/python/tests/test_or_factoring.py

@@ -0,0 +1,63 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The lance-csr Authors
+
+"""An OR of ANDs in WHERE also filters on what each branch says about one variable.
+
+`(a.x = 1 AND b.y = 2) OR (a.x = 2 AND b.y = 1)` implies `a.x = 1 OR a.x = 2`, which
+can narrow `a` before expanding to `b`. The original predicate still decides the rows.
+"""
+
+import pyarrow as pa
+import pytest
+from lance_csr import CypherEngine, GraphConfig
+
+
+@pytest.fixture(params=[False, True], ids=["joins", "csr"])
+def engine(request):
+    config = (
+        GraphConfig.builder()
+        .node_table("Person", "id")
+        .rel_table("KNOWS", "Person", "Person", "src", "dst")
+        .rel_table("FOLLOWS", "Person", "Person", "src", "dst")
+        .build()
+    )
+    return CypherEngine(
+        config,
+        {
+            "Person": pa.table({"id": [1, 2, 3, 4], "age": [30, 40, None, 30]}),
+            "KNOWS": pa.table({"src": [1, 2, 3], "dst": [2, 3, 4]}),
+            # A duplicate edge must still count twice.
+            "FOLLOWS": pa.table({"src": [1, 2, 2, 3, 4], "dst": [2, 1, 1, 2, 1]}),
+        },
+        use_csr=request.param,
+    )
+
+
+def test_match_with_either_direction_predicate(engine):
+    query = (
+        "MATCH (a:Person)-[:FOLLOWS]->(b:Person) "
+        "WHERE (a.age = 30 AND b.age = 40) OR (a.age = 40 AND b.age = 30) "
+        "RETURN a.id AS a, b.id AS b ORDER BY a, b"
+    )
+    # Person 3's null age matches neither branch.
+    assert engine.execute(query).to_pylist() == [
+        {"a": 1, "b": 2},
+        {"a": 2, "b": 1},
+        {"a": 2, "b": 1},
+    ]
+
+
+def test_comprehension_correlated_through_an_or(engine):
+    # Per path relationship, the FOLLOWS edges between its ends in either direction.
+    query = (
+        "MATCH path = (:Person {id: 1})-[:KNOWS*1..3]->(:Person {id: 4}) "
+        "RETURN [r IN relationships(path) | size([(a:Person)-[:FOLLOWS]->(b:Person) "
+        "WHERE (a.id = startNode(r).id AND b.id = endNode(r).id) "
+        "OR (a.id = endNode(r).id AND b.id = startNode(r).id) | 1])] AS w"
+    )
+    assert engine.execute(query).to_pylist() == [{"w": [3, 1, 0]}]
+    if "CsrExpandExec" in (plan := engine.explain(query)):
+        # The implied `a.id = ... OR a.id = ...` joins below the expansion to b,
+        # instead of expanding from every person and filtering afterwards.
+        physical = plan.split("physical_plan")[1]
+        assert physical.index("CsrExpandExec") < physical.index("NestedLoopJoinExec")