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/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/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" 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}"); + } +}