From 4c1f54d4f1abe8a5573b59b8cc736f27401ae6d0 Mon Sep 17 00:00:00 2001 From: prrao87 Date: Fri, 6 Nov 2026 20:06:53 -0500 Subject: [PATCH] perf(query): narrow each variable through an OR of ANDs in WHERE A WHERE such as (a.id = $x AND b.id = $y) OR (a.id = $y AND b.id = $x) reads both ends in every branch, so no part of it could be pushed below the expansion. Correlated pattern comprehensions then expanded from every node for each outer row: the official LDBC IC14 text took about two minutes on SF1. When planning a WHERE, also add, per variable, the OR of each branch without that variable's conjuncts (here a.id = $x OR a.id = $y). It is implied by the original, which stays in the filter, so results are unchanged; DataFusion's existing pushdown moves it below the expansion. IC14's official text now runs in 0.31 s with identical rows. Refs #123 --- crates/lance-csr/src/logical_plan.rs | 6 +- crates/lance-csr/src/pattern_order.rs | 133 +++++++++++++++++++++++ docs/internals/index.md | 2 +- docs/internals/optimizer.md | 24 ++++ python/python/tests/test_or_factoring.py | 63 +++++++++++ 5 files changed, 226 insertions(+), 2 deletions(-) create mode 100644 python/python/tests/test_or_factoring.py diff --git a/crates/lance-csr/src/logical_plan.rs b/crates/lance-csr/src/logical_plan.rs index 2988c17..06b7490 100644 --- a/crates/lance-csr/src/logical_plan.rs +++ b/crates/lance-csr/src/logical_plan.rs @@ -1172,7 +1172,11 @@ impl<'a> LogicalPlanner<'a> { input: LogicalOperator, predicate: &BooleanExpression, ) -> Result { - 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; diff --git a/crates/lance-csr/src/pattern_order.rs b/crates/lance-csr/src/pattern_order.rs index 6049602..c8f8018 100644 --- a/crates/lance-csr/src/pattern_order.rs +++ b/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 { + 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 { + 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 { + 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()); + } } diff --git a/docs/internals/index.md b/docs/internals/index.md index 4546892..dba91d8 100644 --- a/docs/internals/index.md +++ b/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 | diff --git a/docs/internals/optimizer.md b/docs/internals/optimizer.md index e9b9ead..439f52d 100644 --- a/docs/internals/optimizer.md +++ b/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 diff --git a/python/python/tests/test_or_factoring.py b/python/python/tests/test_or_factoring.py new file mode 100644 index 0000000..93b7ccf --- /dev/null +++ b/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")