fix(planner): plan optional match pattern predicates that read earlier variables

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"

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"

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"

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<String> = names
                 .into_iter()

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();

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"

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<i64>| {
+        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::<Int64Array>();
+            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}");
+    }
+}