From ae4d3cae96f0bf42f49dbbf2034cf73d6cf49b11 Mon Sep 17 00:00:00 2001 From: prrao87 Date: Sun, 1 Nov 2026 15:40:09 -0500 Subject: [PATCH 1/2] fix(planner): carry earlier variables read by an OPTIONAL MATCH pattern predicate OPTIONAL MATCH (friend)<-[:CREATED]-(post) WHERE (post)-->()<--(person) failed to plan with 'No field named person' when person was bound by an earlier clause. The optional side's seed carries the earlier variables its WHERE reads, but they were collected with boolean_variables, which skips pattern predicates. Also collect the variables named in pattern predicates, keeping only those bound earlier. --- crates/lance-csr/src/logical_plan.rs | 8 +++ crates/lance-csr/src/pattern_order.rs | 24 +++++++++ crates/lance-csr/tests/test_engine.rs | 70 +++++++++++++++++++++++++++ 3 files changed, 102 insertions(+) diff --git a/crates/lance-csr/src/logical_plan.rs b/crates/lance-csr/src/logical_plan.rs index 30eb9d3..5087a01 100644 --- a/crates/lance-csr/src/logical_plan.rs +++ b/crates/lance-csr/src/logical_plan.rs @@ -1612,6 +1612,14 @@ impl<'a> LogicalPlanner<'a> { if let Some(w) = &clause.where_clause { let mut names = std::collections::HashSet::new(); crate::pattern_order::boolean_variables(&w.expression, &mut names); + // A pattern predicate can read earlier variables too (`(post)-->()<--(person)`). + let mut in_patterns = std::collections::HashSet::new(); + crate::pattern_order::exists_variables(&w.expression, &mut in_patterns); + names.extend( + in_patterns + .into_iter() + .filter(|n| self.variables.contains_key(*n)), + ); let local = crate::pattern_order::pattern_variables(&clause.patterns); let mut scalars: Vec = names .into_iter() diff --git a/crates/lance-csr/src/pattern_order.rs b/crates/lance-csr/src/pattern_order.rs index 818b9dd..6049602 100644 --- a/crates/lance-csr/src/pattern_order.rs +++ b/crates/lance-csr/src/pattern_order.rs @@ -282,6 +282,30 @@ fn value_variables<'a>(expr: &'a ValueExpression, out: &mut HashSet<&'a str>) { } } +/// The variables written in `expr`'s pattern predicates, which `boolean_variables` skips. +// shortcut: a pattern predicate inside a value (`CASE WHEN (a)-->(b) ...`) is not +// visited; extend this if OPTIONAL MATCH ... WHERE needs it. +pub(crate) fn exists_variables<'a>(expr: &'a BooleanExpression, out: &mut HashSet<&'a str>) { + match expr { + BooleanExpression::PatternExists { + patterns, + where_clause, + } => { + out.extend(pattern_variables(patterns)); + if let Some(w) = where_clause { + boolean_variables(w, out); + exists_variables(w, out); + } + } + BooleanExpression::And(a, b) | BooleanExpression::Or(a, b) => { + exists_variables(a, out); + exists_variables(b, out); + } + BooleanExpression::Not(inner) => exists_variables(inner, out), + _ => {} + } +} + /// The node and relationship variables written in `patterns`. pub(crate) fn pattern_variables(patterns: &[GraphPattern]) -> HashSet<&str> { let mut out = HashSet::new(); diff --git a/crates/lance-csr/tests/test_engine.rs b/crates/lance-csr/tests/test_engine.rs index 9c1be2b..116d64c 100644 --- a/crates/lance-csr/tests/test_engine.rs +++ b/crates/lance-csr/tests/test_engine.rs @@ -217,3 +217,73 @@ async fn construction_errors_are_config_errors() { other => panic!("expected a ConfigError, got {:?}", other.err()), } } + +/// A pattern predicate in an OPTIONAL MATCH's WHERE may read a variable bound earlier +/// (`person`) that its own pattern doesn't mention; each row counts its own person's tags. +#[tokio::test] +async fn optional_match_where_pattern_reads_an_earlier_variable() { + let config = GraphConfig::builder() + .node_table("Person", "id") + .node_table("Post", "id") + .node_table("Tag", "id") + .rel_table("KNOWS", "Person", "Person", "src", "dst") + .rel_table("CREATED", "Post", "Person", "src", "dst") + .rel_table("HAS_TAG", "Post", "Tag", "src", "dst") + .rel_table("INTEREST", "Person", "Tag", "src", "dst") + .build() + .unwrap(); + let ids = |ids: Vec| { + let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])); + RecordBatch::try_new(schema, vec![Arc::new(Int64Array::from(ids))]).unwrap() + }; + let edges = |pairs: &[(i64, i64)]| { + let schema = Arc::new(Schema::new(vec![ + Field::new("src", DataType::Int64, false), + Field::new("dst", DataType::Int64, false), + ])); + RecordBatch::try_new( + schema, + vec![ + Arc::new(Int64Array::from_iter_values(pairs.iter().map(|p| p.0))), + Arc::new(Int64Array::from_iter_values(pairs.iter().map(|p| p.1))), + ], + ) + .unwrap() + }; + // Person 1 is interested in tags 100 and 101, person 5 in 102. Post 12 matches + // person 1 through two tags but counts once; friend 4 has no posts. + let sources = [ + ("Person", ids(vec![1, 2, 3, 4, 5])), + ("Post", ids(vec![10, 11, 12, 13])), + ("Tag", ids(vec![100, 101, 102])), + ("KNOWS", edges(&[(1, 2), (1, 3), (1, 4), (5, 2)])), + ("CREATED", edges(&[(10, 2), (11, 2), (12, 3), (13, 3)])), + ( + "HAS_TAG", + edges(&[(10, 100), (10, 102), (11, 102), (12, 100), (12, 101)]), + ), + ("INTEREST", edges(&[(1, 100), (1, 101), (5, 102)])), + ]; + let query = "MATCH (person:Person)-[:KNOWS]->(friend:Person) \ + WITH DISTINCT friend, person \ + OPTIONAL MATCH (friend)<-[:CREATED]-(post:Post) \ + WHERE (post)-[:HAS_TAG]->(:Tag)<-[:INTEREST]-(person) \ + RETURN person.id AS p, friend.id AS f, count(post) AS n ORDER BY p, f"; + for use_csr in [false, true] { + let sources = sources + .iter() + .map(|(name, batch)| (name.to_string(), EngineSource::Arrow(batch.clone()))) + .collect(); + let engine = CypherEngine::new(config.clone(), sources, use_csr) + .await + .unwrap(); + let result = engine.execute(query, HashMap::new()).await.unwrap(); + let column = |i: usize| { + let c = result.column(i).as_any().downcast_ref::(); + c.unwrap().values().to_vec() + }; + assert_eq!(column(0), [1, 1, 1, 5], "use_csr={use_csr}"); + assert_eq!(column(1), [2, 3, 4, 2], "use_csr={use_csr}"); + assert_eq!(column(2), [1, 1, 0, 2], "use_csr={use_csr}"); + } +} From e5d026fa93ee77d295d749a6489cb1bb24bdb753 Mon Sep 17 00:00:00 2001 From: prrao87 Date: Sun, 1 Nov 2026 15:40:09 -0500 Subject: [PATCH 2/2] perf(planner): format extension nodes without their input plans DataFusion's physical planner formats every extension node with Debug while planning it. The derived Debug of CsrExpandNode, NodeLookupNode, ShortestPathNode and ApplyNode printed the whole input plan, which made this quadratic in the plan's depth. Print the node's explain line instead, as PathMatchNode already does. --- crates/lance-csr/src/apply.rs | 10 +++++++++- crates/lance-csr/src/csr_expand.rs | 10 +++++++++- crates/lance-csr/src/csr_expand/node_lookup.rs | 10 +++++++++- crates/lance-csr/src/shortest_path.rs | 10 +++++++++- 4 files changed, 36 insertions(+), 4 deletions(-) diff --git a/crates/lance-csr/src/apply.rs b/crates/lance-csr/src/apply.rs index 33d0843..274dc7a 100644 --- a/crates/lance-csr/src/apply.rs +++ b/crates/lance-csr/src/apply.rs @@ -26,7 +26,7 @@ use futures::StreamExt; /// The child is deliberately unoptimized until its imports have concrete values. /// It is not an ordinary input: planning it before binding would give global top-k /// and aggregate semantics instead of evaluating them once per outer row. -#[derive(Clone, Debug, PartialEq, Eq, Hash)] +#[derive(Clone, PartialEq, Eq, Hash)] pub(crate) struct ApplyNode { input: LogicalPlan, inner: LogicalPlan, @@ -63,6 +63,14 @@ impl PartialOrd for ApplyNode { } } +// DataFusion formats every extension node while planning it; printing the input plan +// here would make that quadratic in the plan's depth. +impl std::fmt::Debug for ApplyNode { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + UserDefinedLogicalNodeCore::fmt_for_explain(self, f) + } +} + impl UserDefinedLogicalNodeCore for ApplyNode { fn name(&self) -> &str { "Apply" diff --git a/crates/lance-csr/src/csr_expand.rs b/crates/lance-csr/src/csr_expand.rs index 885ef7c..cc9c2c6 100644 --- a/crates/lance-csr/src/csr_expand.rs +++ b/crates/lance-csr/src/csr_expand.rs @@ -1398,7 +1398,7 @@ impl Neighbors { } /// Logical node: append `(key_output, neighbor_output)` for every neighbour of `key_column`. -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct CsrExpandNode { input: LogicalPlan, /// Input column holding the source node id. @@ -1616,6 +1616,14 @@ impl PartialOrd for CsrExpandNode { } } +// DataFusion formats every extension node while planning it; printing the input plan +// here would make that quadratic in the plan's depth. +impl std::fmt::Debug for CsrExpandNode { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + UserDefinedLogicalNodeCore::fmt_for_explain(self, f) + } +} + impl UserDefinedLogicalNodeCore for CsrExpandNode { fn name(&self) -> &str { "CsrExpand" diff --git a/crates/lance-csr/src/csr_expand/node_lookup.rs b/crates/lance-csr/src/csr_expand/node_lookup.rs index 019c633..7214dae 100644 --- a/crates/lance-csr/src/csr_expand/node_lookup.rs +++ b/crates/lance-csr/src/csr_expand/node_lookup.rs @@ -12,7 +12,7 @@ use super::*; /// `input` plus the columns of the node whose id is in `key_column`. A null or unknown /// id (an unmatched OPTIONAL MATCH) gets nulls. -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct NodeLookupNode { input: LogicalPlan, key_column: String, @@ -150,6 +150,14 @@ impl PartialOrd for NodeLookupNode { } } +// DataFusion formats every extension node while planning it; printing the input plan +// here would make that quadratic in the plan's depth. +impl std::fmt::Debug for NodeLookupNode { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + UserDefinedLogicalNodeCore::fmt_for_explain(self, f) + } +} + impl UserDefinedLogicalNodeCore for NodeLookupNode { fn name(&self) -> &str { "NodeLookup" diff --git a/crates/lance-csr/src/shortest_path.rs b/crates/lance-csr/src/shortest_path.rs index aff7502..19ddc9c 100644 --- a/crates/lance-csr/src/shortest_path.rs +++ b/crates/lance-csr/src/shortest_path.rs @@ -72,7 +72,7 @@ pub struct ShortestPathSpec { } /// Logical node; see the module docs. -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct ShortestPathNode { input: LogicalPlan, /// Scan of the relationship's endpoints; `None` when `adjacency` is given. @@ -154,6 +154,14 @@ impl PartialOrd for ShortestPathNode { } } +// DataFusion formats every extension node while planning it; printing the input plan +// here would make that quadratic in the plan's depth. +impl std::fmt::Debug for ShortestPathNode { + fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + UserDefinedLogicalNodeCore::fmt_for_explain(self, f) + } +} + impl UserDefinedLogicalNodeCore for ShortestPathNode { fn name(&self) -> &str { "ShortestPath"