From 67aca1ffc78317ef0b42f9de9567fb2a18779fbe Mon Sep 17 00:00:00 2001 From: prrao87 Date: Wed, 28 Oct 2026 08:48:31 -0700 Subject: [PATCH] fix(planner): bind each relationship once per MATCH Two single-hop relationships of the same type in one MATCH could bind the same edge. Add a filter that their endpoints differ, naming anonymous endpoints so they can be compared. Fixes #59 --- crates/lance-csr/src/logical_plan.rs | 221 +++++++++++++++--- .../tests/test_datafusion_pipeline.rs | 7 +- docs/cypher.md | 5 + .../tests/test_relationship_uniqueness.py | 44 ++++ 4 files changed, 247 insertions(+), 30 deletions(-) create mode 100644 python/python/tests/test_relationship_uniqueness.py diff --git a/crates/lance-csr/src/logical_plan.rs b/crates/lance-csr/src/logical_plan.rs index 0123ede..7346c8c 100644 --- a/crates/lance-csr/src/logical_plan.rs +++ b/crates/lance-csr/src/logical_plan.rs @@ -895,6 +895,8 @@ impl<'a> LogicalPlanner<'a> { .as_deref() .or(oriented.as_deref()) .unwrap_or(&match_clause.patterns); + let distinct = self.distinct_relationships(patterns); + let patterns = distinct.as_ref().map_or(patterns, |(named, _)| named); let mut plan = base; for pattern in patterns { @@ -939,10 +941,119 @@ impl<'a> LogicalPlanner<'a> { } } - plan.ok_or_else(|| GraphError::PlanError { + let plan = plan.ok_or_else(|| GraphError::PlanError { message: "Failed to plan MATCH clause".to_string(), location: snafu::Location::new(file!(), line!(), column!()), - }) + })?; + match distinct { + Some((_, predicate)) => self.filter(plan, &predicate), + None => Ok(plan), + } + } + + /// openCypher binds each relationship at most once per MATCH. For single-hop relationships + /// that could bind the same edge, returns the patterns with their anonymous nodes named + /// and a predicate that the edges differ, compared by their endpoint node ids. + // shortcut: variable-length and untyped hops are left out, and two distinct edges joining + // the same nodes count as one; upgrade by carrying edge row ids through the hops. + fn distinct_relationships( + &self, + patterns: &[GraphPattern], + ) -> Option<(Vec, BooleanExpression)> { + let mut named = patterns.to_vec(); + let mut labels: HashMap = self.variables.clone(); + let mut hops = Vec::new(); + let mut anonymous = 0; + for (index, pattern) in named.iter_mut().enumerate() { + let GraphPattern::Path(path) = pattern else { + continue; + }; + if path.shortest.is_some() { + continue; + } + // The path operator already keeps the edges of one path apart, by edge row. + let path_operator = path.variable.is_some() + || (self.rich_paths + && path + .segments + .iter() + .any(|s| s.relationship.variable.is_some())); + let mut name = |node: &mut NodePattern| { + let var = node + .variable + .get_or_insert_with(|| { + anonymous += 1; + format!("_rel_end_{}_{anonymous}", self.variables.len()) + }) + .clone(); + if let Some(label) = node.labels.first() { + labels.entry(var.clone()).or_insert_with(|| label.clone()); + } + var + }; + let mut start = name(&mut path.start_node); + for segment in &mut path.segments { + let end = name(&mut segment.end_node); + let rel = &segment.relationship; + if rel.length.is_none() && !rel.types.is_empty() { + let (src, dst) = match rel.direction { + RelationshipDirection::Incoming => (end.clone(), start.clone()), + _ => (start.clone(), end.clone()), + }; + let undirected = rel.direction == RelationshipDirection::Undirected; + let exact = path_operator.then_some(index); + hops.push(( + src, + dst, + undirected, + rel.types.clone(), + rel.variable.clone(), + exact, + )); + } + start = end; + } + } + + // Endpoints of different labels are never the same node. + let eq = |a: &String, b: &String| { + let (la, lb) = (labels.get(a)?, labels.get(b)?); + la.eq_ignore_ascii_case(lb) + .then(|| BooleanExpression::Comparison { + left: ValueExpression::Variable(a.clone()), + operator: ComparisonOperator::Equal, + right: ValueExpression::Variable(b.clone()), + }) + }; + let both = |a: Option, b: Option| { + Some(BooleanExpression::And(Box::new(a?), Box::new(b?))) + }; + let mut predicate: Option = None; + for (i, (s1, d1, u1, t1, v1, x1)) in hops.iter().enumerate() { + for (s2, d2, u2, t2, v2, x2) in &hops[i + 1..] { + let overlap = t1 + .iter() + .any(|a| t2.iter().any(|b| a.eq_ignore_ascii_case(b))); + let shared = v1.is_some() && v1 == v2; + if !overlap || shared || (x1.is_some() && x1 == x2) { + continue; + } + let forward = both(eq(s1, s2), eq(d1, d2)); + // An undirected hop may have walked its edge either way round. + let backward = (*u1 || *u2).then(|| both(eq(s1, d2), eq(d1, s2))).flatten(); + let same = match (forward, backward) { + (Some(f), Some(b)) => BooleanExpression::Or(Box::new(f), Box::new(b)), + (Some(e), None) | (None, Some(e)) => e, + (None, None) => continue, + }; + let differ = BooleanExpression::Not(Box::new(same)); + predicate = Some(match predicate { + Some(p) => BooleanExpression::And(Box::new(p), Box::new(differ)), + None => differ, + }); + } + } + predicate.map(|p| (named, p)) } /// `input` filtered by `predicate`, after turning any EXISTS in it into mark columns. @@ -2605,6 +2716,61 @@ mod tests { } } + #[test] + fn test_same_type_hops_bind_distinct_relationships() { + let config = GraphConfig::builder() + .node_table("P", "id") + .rel_table("R", "P", "P", "src", "dst") + .rel_table("S", "P", "P", "src", "dst") + .build() + .unwrap(); + let filtered = |q: &str| { + let logical = LogicalPlanner::new(&config) + .plan(&parse_cypher_query(q).unwrap()) + .unwrap(); + let LogicalOperator::Project { input, .. } = logical else { + panic!("Expected Project at top level"); + }; + match *input { + LogicalOperator::Filter { predicate, .. } => Some(predicate), + _ => None, + } + }; + let id = |v: &str| { + ValueExpression::Property(PropertyRef { + variable: v.into(), + property: "id".into(), + }) + }; + let eq = |a: &str, b: &str| BooleanExpression::Comparison { + left: id(a), + operator: ComparisonOperator::Equal, + right: id(b), + }; + assert_eq!( + filtered("MATCH (a:P)-[:R]->(b:P)<-[:R]-(c:P) RETURN a"), + Some(BooleanExpression::Not(Box::new(BooleanExpression::And( + Box::new(eq("a", "c")), + Box::new(eq("b", "b")), + )))) + ); + // Anonymous endpoints are named so they can be compared. + assert!(filtered("MATCH (a:P)-[:R]->(:P)<-[:R]-(c:P) RETURN a").is_some()); + // Different types, one shared variable, or separate MATCH clauses never collide. + assert_eq!( + filtered("MATCH (a:P)-[:R]->(b:P)<-[:S]-(c:P) RETURN a"), + None + ); + assert_eq!( + filtered("MATCH (a:P)-[r:R]->(b:P), (c:P)-[r:R]->(d:P) RETURN a"), + None + ); + assert_eq!( + filtered("MATCH (a:P)-[:R]->(b:P) MATCH (c:P)-[:R]->(b) RETURN a"), + None + ); + } + #[test] fn test_variable_only_node_default_label() { let q = "MATCH (x) RETURN x.name"; @@ -2724,33 +2890,36 @@ mod tests { let mut planner = LogicalPlanner::new(&config); let logical_plan = planner.plan(&ast).unwrap(); - // Expect: Project { Expand(shared->b) { Expand(a->shared) { Scan(a) } } } - match &logical_plan { - LogicalOperator::Project { input, .. } => match input.as_ref() { - LogicalOperator::Expand { - input: inner, - source_variable, - target_variable, - .. - } => { - assert_eq!(source_variable, "shared"); - assert_eq!(target_variable, "b"); + // Expect: Project { Filter(distinct KNOWS) { Expand(shared->b) { Expand(a->shared) } } } + let LogicalOperator::Project { input, .. } = &logical_plan else { + panic!("Expected Project at top level"); + }; + let LogicalOperator::Filter { input, .. } = input.as_ref() else { + panic!("Expected relationship uniqueness Filter"); + }; + match input.as_ref() { + LogicalOperator::Expand { + input: inner, + source_variable, + target_variable, + .. + } => { + assert_eq!(source_variable, "shared"); + assert_eq!(target_variable, "b"); - match inner.as_ref() { - LogicalOperator::Expand { - source_variable: first_src, - target_variable: first_dst, - .. - } => { - assert_eq!(first_src, "a"); - assert_eq!(first_dst, "shared"); - } - _ => panic!("Expected first Expand (a->shared)"), + match inner.as_ref() { + LogicalOperator::Expand { + source_variable: first_src, + target_variable: first_dst, + .. + } => { + assert_eq!(first_src, "a"); + assert_eq!(first_dst, "shared"); } + _ => panic!("Expected first Expand (a->shared)"), } - _ => panic!("Expected second Expand (shared->b)"), - }, - _ => panic!("Expected Project at top level"), + } + _ => panic!("Expected second Expand (shared->b)"), } } diff --git a/crates/lance-csr/tests/test_datafusion_pipeline.rs b/crates/lance-csr/tests/test_datafusion_pipeline.rs index d2e762a..15b3a2c 100644 --- a/crates/lance-csr/tests/test_datafusion_pipeline.rs +++ b/crates/lance-csr/tests/test_datafusion_pipeline.rs @@ -5020,10 +5020,9 @@ async fn test_datafusion_variable_reuse_with_count_distinct() { .downcast_ref::() .unwrap(); - // Distinct people who have at least 2 people knowing them: - // Bob (known by Alice), Charlie (known by Alice and Bob), - // David (known by Charlie), Eve (known by David) = 4 people - assert_eq!(counts.value(0), 4); + // The two KNOWS edges must differ, so only people known by at least 2 people count: + // Charlie (known by Alice and Bob) = 1 person + assert_eq!(counts.value(0), 1); } #[tokio::test] diff --git a/docs/cypher.md b/docs/cypher.md index 68bd896..c8a1bbd 100644 --- a/docs/cypher.md +++ b/docs/cypher.md @@ -314,6 +314,11 @@ RETURN length(p), nodes(p) Named paths also support typed variable-length alternation through their edge inventory. Unnamed undirected variable-length alternation remains unsupported. - **Undirected variable-length paths** skip immediate backtracking (`a-b-a`). +- **Relationship uniqueness**: within one `MATCH`, two single-hop relationships of the same type never + bind the same edge, whether they are named or anonymous, in one pattern or in comma-separated parts. + `(a)-[:KNOWS]->(b)<-[:KNOWS]-(c)` therefore never matches `c = a`. Separate `MATCH` clauses may bind the + same edge again. Edges are compared by their endpoints, so two parallel edges between the same nodes + count as one. Variable-length and untyped relationships are not checked, except inside a named path. - **`shortestPath` and `allShortestPaths`** need exactly one relationship type, no relationship variable, and named end nodes. Their length is unbounded unless you give a maximum. `nodes(p)` returns the node keys. With no path, `MATCH` drops the row and `OPTIONAL MATCH` returns `null`. `shortestPath` chooses one path diff --git a/python/python/tests/test_relationship_uniqueness.py b/python/python/tests/test_relationship_uniqueness.py new file mode 100644 index 0000000..66c218b --- /dev/null +++ b/python/python/tests/test_relationship_uniqueness.py @@ -0,0 +1,44 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The lance-csr Authors + +"""One MATCH binds each relationship at most once (relationship isomorphism).""" + +import pyarrow as pa +import pytest +from lance_csr import CypherEngine, GraphConfig + +CONFIG = ( + GraphConfig.builder() + .node_table("Person", "id") + .rel_table("KNOWS", "Person", "Person", "src", "dst") + .build() +) +PEOPLE = pa.table({"id": [1, 2, 3, 4], "name": ["A", "B", "C", "D"]}) +# A->B and C->B: two edges into B. +KNOWS = pa.table({"src": [1, 3], "dst": [2, 2]}) +RETURN = " RETURN a.name AS a, c.name AS c ORDER BY a, c" +DISTINCT = [{"a": "A", "c": "C"}, {"a": "C", "c": "A"}] + + +@pytest.mark.parametrize("use_csr", [False, True]) +@pytest.mark.parametrize( + ("query", "expected"), + [ + ("MATCH (a:Person)-[r1:KNOWS]->(b:Person)<-[r2:KNOWS]-(c:Person)", DISTINCT), + ("MATCH (a:Person)-[:KNOWS]->(:Person)<-[:KNOWS]-(c:Person)", DISTINCT), + ("MATCH (a:Person)-[:KNOWS]->(b:Person), (c:Person)-[:KNOWS]->(b)", DISTINCT), + ("MATCH (a:Person)-[:KNOWS]-(b:Person)-[:KNOWS]-(c:Person)", DISTINCT), + ( + "MATCH p = (a:Person)-[r1:KNOWS]->(b:Person)<-[r2:KNOWS]-(c:Person)", + DISTINCT, + ), + # Separate MATCH clauses may bind the same relationship again. + ( + "MATCH (a:Person)-[:KNOWS]->(b:Person) MATCH (c:Person)-[:KNOWS]->(b)", + [{"a": a, "c": c} for a in "AC" for c in "AC"], + ), + ], +) +def test_one_match_binds_each_relationship_once(use_csr, query, expected): + engine = CypherEngine(CONFIG, {"Person": PEOPLE, "KNOWS": KNOWS}, use_csr=use_csr) + assert engine.execute(query + RETURN).to_pylist() == expected