From 682a3eb381d2291b8f466224bc8cb89bb276f13f Mon Sep 17 00:00:00 2001 From: Max De Marzi Date: Mon, 8 Jun 2026 22:06:17 +0000 Subject: [PATCH 1/4] Optimize repeating paths with unused variables and level-by-level fallback --- src/lib.rs | 223 ++++++++++++++++++++++- src/optimizer/pushdown.rs | 11 +- src/runtime/engine.rs | 135 ++++++++++++++ tests/unused_variable_repetition_test.rs | 121 ++++++++++++ 4 files changed, 483 insertions(+), 7 deletions(-) create mode 100644 tests/unused_variable_repetition_test.rs diff --git a/src/lib.rs b/src/lib.rs index ab93e0c..18cf944 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -9,12 +9,231 @@ pub mod syntax; pub mod typing; use syntax::path_pattern::PathPattern; -use syntax::query::{MatchStatement, Query}; +use syntax::query::{MatchStatement, Query, ReturnItem, Aggregator, SortKey}; +use syntax::expr::Expr; use typing::checker::Typechecker; use typing::variable_type::Schema; +use std::collections::{HashSet, HashMap}; + +fn collect_expr_vars(expr: &Expr, acc: &mut HashSet) { + match expr { + Expr::Var(name) => { + acc.insert(name.clone()); + } + Expr::AttrLookup { var, .. } => { + acc.insert(var.clone()); + } + Expr::FieldAccess { base, .. } => { + collect_expr_vars(base, acc); + } + Expr::Binop { left, right, .. } => { + collect_expr_vars(left, acc); + collect_expr_vars(right, acc); + } + Expr::Unop { operand, .. } | Expr::IsNull { operand, .. } => { + collect_expr_vars(operand, acc); + } + Expr::Coalesce(args) | Expr::Call { args, .. } => { + for a in args { + collect_expr_vars(a, acc); + } + } + Expr::Record { fields } => { + for (_, e) in fields { + collect_expr_vars(e, acc); + } + } + Expr::ValueSubquery { body } | Expr::Exists { body } | Expr::NotExists { body } => { + collect_query_vars(body, acc); + } + Expr::Case { branches, else_expr } => { + for (cond, value) in branches { + collect_expr_vars(cond, acc); + collect_expr_vars(value, acc); + } + if let Some(value) = else_expr { + collect_expr_vars(value, acc); + } + } + Expr::ListComprehension { var, source, filter, body } => { + collect_expr_vars(source, acc); + let mut inner = HashSet::new(); + if let Some(f) = filter { + collect_expr_vars(f, &mut inner); + } + collect_expr_vars(body, &mut inner); + inner.remove(var); + acc.extend(inner); + } + Expr::Agg(agg) => match agg.as_ref() { + Aggregator::CountStar => {} + Aggregator::GeneralSet { expr, .. } => { + collect_expr_vars(expr, acc); + } + }, + Expr::Const(_) | Expr::Type(_) => {} + } +} + +fn collect_query_vars(q: &Query, acc: &mut HashSet) { + if let Some(returns) = &q.returns { + for item in returns { + match item { + ReturnItem::Expr { expr, .. } => { + collect_expr_vars(expr, acc); + } + ReturnItem::Aggregate { agg, .. } => match agg { + Aggregator::CountStar => {} + Aggregator::GeneralSet { expr, .. } => { + collect_expr_vars(expr, acc); + } + }, + } + } + } else { + for m in &q.matches { + acc.extend(m.pattern().freevars()); + } + } + if let Some(gb) = &q.group_by { + for e in gb { + collect_expr_vars(e, acc); + } + } + if let Some(ob) = &q.order_by { + for spec in ob { + if let SortKey::Expr(expr) = &spec.key { + collect_expr_vars(expr, acc); + } + } + } + let mut var_counts = HashMap::new(); + for m in &q.matches { + for v in m.pattern().freevars() { + *var_counts.entry(v).or_insert(0) += 1; + } + } + for (v, count) in var_counts { + if count > 1 { + acc.insert(v); + } + } + for m in &q.matches { + collect_pattern_filter_vars(m.pattern(), acc); + } +} + +fn collect_pattern_filter_vars(p: &PathPattern, acc: &mut HashSet) { + match p { + PathPattern::Filter(inner, expr) => { + collect_expr_vars(expr, acc); + collect_pattern_filter_vars(inner, acc); + } + PathPattern::Concat(a, b) | PathPattern::Union(a, b) | PathPattern::Join(a, b) => { + collect_pattern_filter_vars(a, acc); + collect_pattern_filter_vars(b, acc); + } + PathPattern::Repeat { pattern, .. } | PathPattern::Questioned(pattern) + | PathPattern::Selected { pattern, .. } | PathPattern::Named { pattern, .. } => { + collect_pattern_filter_vars(pattern, acc); + } + _ => {} + } +} + +fn remove_unused_pattern_vars(p: PathPattern, used: &HashSet) -> PathPattern { + match p { + PathPattern::Node(Some(mut d)) => { + if let Some(ref var) = d.var { + if !used.contains(var) { + d.var = None; + } + } + PathPattern::Node(Some(d)) + } + PathPattern::EdgeRight(Some(mut d)) => { + if let Some(ref var) = d.var { + if !used.contains(var) { + d.var = None; + } + } + PathPattern::EdgeRight(Some(d)) + } + PathPattern::EdgeLeft(Some(mut d)) => { + if let Some(ref var) = d.var { + if !used.contains(var) { + d.var = None; + } + } + PathPattern::EdgeLeft(Some(d)) + } + PathPattern::EdgeUndirected(Some(mut d)) => { + if let Some(ref var) = d.var { + if !used.contains(var) { + d.var = None; + } + } + PathPattern::EdgeUndirected(Some(d)) + } + PathPattern::EdgeAnyDirection(Some(mut d)) => { + if let Some(ref var) = d.var { + if !used.contains(var) { + d.var = None; + } + } + PathPattern::EdgeAnyDirection(Some(d)) + } + PathPattern::Concat(a, b) => PathPattern::Concat( + Box::new(remove_unused_pattern_vars(*a, used)), + Box::new(remove_unused_pattern_vars(*b, used)), + ), + PathPattern::Union(a, b) => PathPattern::Union( + Box::new(remove_unused_pattern_vars(*a, used)), + Box::new(remove_unused_pattern_vars(*b, used)), + ), + PathPattern::Join(a, b) => PathPattern::Join( + Box::new(remove_unused_pattern_vars(*a, used)), + Box::new(remove_unused_pattern_vars(*b, used)), + ), + PathPattern::Filter(inner, expr) => PathPattern::Filter( + Box::new(remove_unused_pattern_vars(*inner, used)), + expr, + ), + PathPattern::Repeat { pattern, lb, ub } => PathPattern::Repeat { + pattern: Box::new(remove_unused_pattern_vars(*pattern, used)), + lb, + ub, + }, + PathPattern::Questioned(inner) => { + PathPattern::Questioned(Box::new(remove_unused_pattern_vars(*inner, used))) + } + PathPattern::Selected { prefix, pattern } => PathPattern::Selected { + prefix, + pattern: Box::new(remove_unused_pattern_vars(*pattern, used)), + }, + PathPattern::Named { var, pattern } => PathPattern::Named { + var, + pattern: Box::new(remove_unused_pattern_vars(*pattern, used)), + }, + other => other, + } +} /// Optimize the query while preserving OPTIONAL and §16.6 prefix boundaries. -fn optimize_query(q: Query, schema: &Schema) -> Query { +fn optimize_query(mut q: Query, schema: &Schema) -> Query { + let mut used_vars = HashSet::new(); + collect_query_vars(&q, &mut used_vars); + for m in &mut q.matches { + match m { + MatchStatement::Simple { pattern } => { + *pattern = remove_unused_pattern_vars(std::mem::replace(pattern, PathPattern::Node(None)), &used_vars); + } + MatchStatement::Optional { pattern } => { + *pattern = remove_unused_pattern_vars(std::mem::replace(pattern, PathPattern::Node(None)), &used_vars); + } + } + } + // Selected patterns are evaluated in isolation; do not collapse across them. let mut q = if !q.has_any_optional() && !q.has_any_selected() { let pattern = optimizer::compile(q.collapsed_pattern()); diff --git a/src/optimizer/pushdown.rs b/src/optimizer/pushdown.rs index edf82b9..8c9199a 100644 --- a/src/optimizer/pushdown.rs +++ b/src/optimizer/pushdown.rs @@ -363,15 +363,16 @@ fn merge_constraints(p: PathPattern, c: &Constraints) -> PathPattern { fn boundary_node_vars(p: &PathPattern) -> [Option; 2] { let mut vars = Vec::new(); collect_top_node_vars(p, &mut vars); - [vars.first().cloned(), vars.last().cloned()] + [ + vars.first().cloned().flatten(), + vars.last().cloned().flatten(), + ] } -fn collect_top_node_vars(p: &PathPattern, out: &mut Vec) { +fn collect_top_node_vars(p: &PathPattern, out: &mut Vec>) { match p { PathPattern::Node(desc) => { - if let Some(var) = desc.as_ref().and_then(|d| d.var.clone()) { - out.push(var); - } + out.push(desc.as_ref().and_then(|d| d.var.clone())); } PathPattern::Concat(a, b) => { collect_top_node_vars(a, out); diff --git a/src/runtime/engine.rs b/src/runtime/engine.rs index 7aaa4ea..5968357 100644 --- a/src/runtime/engine.rs +++ b/src/runtime/engine.rs @@ -1766,6 +1766,13 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { return optimized; } } + PathPattern::Repeat { pattern, lb, ub } => { + if let Some(optimized) = + self.try_concat_with_edge_repetition(&ir1, pattern, *lb, *ub, limit) + { + return optimized; + } + } _ => {} } @@ -1962,6 +1969,134 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { IntermediateResult::new(rows) } + fn try_concat_with_edge_repetition( + &self, + ir1: &IntermediateResult, + edge_pat: &PathPattern, + lb: usize, + ub: Option, + limit: usize, + ) -> Option { + let Some(ub_val) = ub else { + return None; + }; + if !matches!( + edge_pat, + PathPattern::EdgeRight(_) + | PathPattern::EdgeLeft(_) + | PathPattern::EdgeUndirected(_) + | PathPattern::EdgeAnyDirection(_) + ) { + return None; + } + + let mut anon_edge_pat = edge_pat.clone(); + let var_name = match &mut anon_edge_pat { + PathPattern::EdgeRight(Some(d)) + | PathPattern::EdgeLeft(Some(d)) + | PathPattern::EdgeUndirected(Some(d)) + | PathPattern::EdgeAnyDirection(Some(d)) => d.var.take(), + _ => None, + }; + + let freevars = edge_pat.freevars(); + let mut accumulated_rows = Vec::new(); + + if lb == 0 { + for r in &ir1.rows { + let mut row = r.clone(); + row.assignment.fill_empty_list(&freevars); + accumulated_rows.push(row); + if limit > 0 && accumulated_rows.len() >= limit { + accumulated_rows.truncate(limit); + return Some(IntermediateResult::new(accumulated_rows)); + } + } + } + + let mut level_ir = self.run_concat_pattern_step(ir1, &anon_edge_pat, 0); + if let Some(ref name) = var_name { + for row in &mut level_ir.rows { + let edge_val = Self::get_last_edge(row.path()); + row.assignment.extend(name.clone(), PathValue::Group(vec![edge_val])); + } + } + if lb <= 1 { + for r in &level_ir.rows { + accumulated_rows.push(r.clone()); + if limit > 0 && accumulated_rows.len() >= limit { + accumulated_rows.truncate(limit); + return Some(IntermediateResult::new(accumulated_rows)); + } + } + } + + for k in 2..=ub_val { + let next_ir = self.run_concat_pattern_step(&level_ir, &anon_edge_pat, 0); + level_ir = next_ir; + if let Some(ref name) = var_name { + for row in &mut level_ir.rows { + let edge_val = Self::get_last_edge(row.path()); + if let Some(PathValue::Group(prev_group)) = row.assignment.get(name) { + let mut new_group = prev_group.clone(); + new_group.push(edge_val); + row.assignment.extend(name.clone(), PathValue::Group(new_group)); + } + } + } + if k >= lb { + for r in &level_ir.rows { + accumulated_rows.push(r.clone()); + if limit > 0 && accumulated_rows.len() >= limit { + accumulated_rows.truncate(limit); + return Some(IntermediateResult::new(accumulated_rows)); + } + } + } + } + + Some(IntermediateResult::new(accumulated_rows)) + } + + fn run_concat_pattern_step( + &self, + ir: &IntermediateResult, + anon_edge_pat: &PathPattern, + limit: usize, + ) -> IntermediateResult { + match anon_edge_pat { + PathPattern::EdgeRight(desc) => { + self.concat_with_directed_edge(ir, desc.as_ref(), true, limit) + } + PathPattern::EdgeLeft(desc) => { + self.concat_with_directed_edge(ir, desc.as_ref(), false, limit) + } + PathPattern::EdgeUndirected(desc) => { + self.concat_with_undirected_edge(ir, desc.as_ref(), limit) + } + PathPattern::EdgeAnyDirection(desc) => { + let right = self.concat_with_directed_edge(ir, desc.as_ref(), true, limit); + if limit > 0 && right.rows.len() >= limit { + return right; + } + let remaining = if limit > 0 { limit - right.rows.len() } else { 0 }; + let left = self.concat_with_directed_edge(ir, desc.as_ref(), false, remaining); + let combined = right.union(left); + if limit > 0 && combined.rows.len() >= limit { + return combined; + } + let remaining = if limit > 0 { limit - combined.rows.len() } else { 0 }; + let und = self.concat_with_undirected_edge(ir, desc.as_ref(), remaining); + combined.union(und) + } + _ => unreachable!(), + } + } + + fn get_last_edge(path: &Path) -> PathValue { + path.0[path.0.len() - 2].clone() + } + /// Hash-join on the concatenation key (last node of ir1 = first node of ir2). /// O(n + m) expected instead of O(n × m) cross-product. fn hash_join( diff --git a/tests/unused_variable_repetition_test.rs b/tests/unused_variable_repetition_test.rs new file mode 100644 index 0000000..56c6d53 --- /dev/null +++ b/tests/unused_variable_repetition_test.rs @@ -0,0 +1,121 @@ +use gqlrust::compile_query; +use gqlrust::model::graph::MemoryGraphStore; +use gqlrust::model::value::Value; +use gqlrust::runtime::engine::Runtime; +use gqlrust::runtime::result::QueryResult; + +fn graph() -> MemoryGraphStore { + let json = r#"{ + "nodes": [ + {"id": "a", "labels": ["Person"], "props": {"id": 10}}, + {"id": "b", "labels": ["Person"], "props": {"id": 20}}, + {"id": "c", "labels": ["Person"], "props": {"id": 30}} + ], + "edges": [ + {"id": "e1", "labels": ["knows"], "props": {}, "endpoints": ["a", "b"], "directionality": "->"}, + {"id": "e2", "labels": ["knows"], "props": {}, "endpoints": ["b", "c"], "directionality": "->"} + ] + }"#; + MemoryGraphStore::from_json_str(json).unwrap() +} + +#[test] +fn test_unused_edge_variable_in_repeat_is_unrolled() { + let q_str = "MATCH (n1:Person)-[e]-{1,2}(n2:Person) WHERE n1.id = 10 RETURN n1.id, n2.id"; + let query = compile_query(q_str).unwrap(); + let pattern_str = query.matches[0].pattern().to_string(); + + // Because `e` is unused, the pass should strip its name, enabling unroll_repeat. + // Therefore, the compiled query pattern should NOT contain the repetition `{1,2}` anymore. + assert!(!pattern_str.contains("{1,2}"), "Unused repetition should be unrolled: {}", pattern_str); + assert!(pattern_str.contains("|"), "Unrolled repetition should be a Union: {}", pattern_str); + + // Run the query and verify correctness of results + let g = graph(); + let rt = Runtime::new(&g); + match rt.run_query(&query, 0) { + QueryResult::Projected(rows) => { + // Paths of length 1: (a)-[e1]-(b) -> (10, 20) + // Paths of length 2: (a)-[e1]-(b)-[e2]-(c) -> (10, 30) + // Plus reverse paths since they are undirected: + // Since the graph is undirected for -[e]-: + // We should get at least [10, 20] and [10, 30]. + let mut ids: Vec = rows + .iter() + .map(|row| match &row[1] { + Value::Int(val) => *val, + other => panic!("expected Int, got {:?}", other), + }) + .collect(); + ids.sort(); + assert_eq!(ids, vec![10, 20, 30]); + } + other => panic!("expected projected rows, got {other:?}"), + } +} + +#[test] +fn test_used_edge_variable_in_repeat_is_not_unrolled() { + // If the edge variable `e` is returned, it is used, so it must not be unrolled. + let q_str = "MATCH (n1:Person)-[e]-{1,2}(n2:Person) WHERE n1.id = 10 RETURN n1.id, e, n2.id"; + let query = compile_query(q_str).unwrap(); + let pattern_str = query.matches[0].pattern().to_string(); + + assert!(pattern_str.contains("{1,2}"), "Used repetition should NOT be unrolled: {}", pattern_str); + assert!(!pattern_str.contains("|"), "Used repetition should NOT be unrolled into a Union: {}", pattern_str); +} + +#[test] +fn test_no_returns_clause_preserves_all_variables() { + // If there is no returns clause, all variables are preserved because the raw intermediate result is returned. + let q_str = "MATCH (n1:Person)-[e]-{1,2}(n2:Person)"; + let query = compile_query(q_str).unwrap(); + let pattern_str = query.matches[0].pattern().to_string(); + + assert!(pattern_str.contains("{1,2}"), "Repetition without returns clause should NOT be unrolled: {}", pattern_str); +} + +#[test] +fn test_used_edge_variable_in_repeat_results_correctness() { + let q_str = "MATCH (n1:Person)-[e]-{1,2}(n2:Person) WHERE n1.id = 10 RETURN n1.id, e, n2.id"; + let query = compile_query(q_str).unwrap(); + let g = graph(); + let rt = Runtime::new(&g); + match rt.run_query(&query, 0) { + QueryResult::Projected(rows) => { + let mut results = Vec::new(); + for r in rows { + let n1_id = match &r[0] { Value::Int(v) => *v, _ => panic!() }; + let n2_id = match &r[2] { Value::Int(v) => *v, _ => panic!() }; + let e_list = match &r[1] { + Value::List(l) => l.iter().map(|item| match item { + Value::Edge(eid) => *eid, + other => panic!("expected Edge, got {:?}", other), + }).collect::>(), + other => panic!("expected List, got {:?}", other), + }; + results.push((n1_id, e_list, n2_id)); + } + results.sort_by_key(|r| (r.2, r.1.len(), r.1.clone())); + assert_eq!(results.len(), 3); + + // Result 1: to node 10 (length 2 path traversing e1 twice) + assert_eq!(results[0].0, 10); + assert_eq!(results[0].2, 10); + assert_eq!(results[0].1.len(), 2); + assert_eq!(results[0].1[0], results[0].1[1]); // same edge traversed twice + + // Result 2: to node 20 (length 1 path) + assert_eq!(results[1].0, 10); + assert_eq!(results[1].2, 20); + assert_eq!(results[1].1.len(), 1); + + // Result 3: to node 30 (length 2 path traversing e1 then e2) + assert_eq!(results[2].0, 10); + assert_eq!(results[2].2, 30); + assert_eq!(results[2].1.len(), 2); + assert_ne!(results[2].1[0], results[2].1[1]); // e1 and e2 are different edges + } + other => panic!("expected projected rows, got {other:?}"), + } +} From 6ef1f008ad415b69146a8b4f67d47ed42290ad1a Mon Sep 17 00:00:00 2001 From: Max De Marzi Date: Mon, 8 Jun 2026 23:13:33 +0000 Subject: [PATCH 2/4] Implement Cypher 25 allReduce predicate function and dynamic early path pruning - Add syntax support for 'allReduce(acc = init, x IN list | reduction, predicate)' in lexer and parser. - Implement typechecking for allReduce in checker.rs, validating list source, boolean predicate return types, and local variable scoping. - Add runtime evaluation of AllReduce expressions in engine.rs. - Integrate stepwise path pruning inside adjacency-driven repeat traversals, pruning branches early when the predicate returns false. - Allow querying global constraints (outer-bound variables) inside the allReduce predicate. - Add comprehensive integration tests in tests/all_reduce_test.rs. - Fix allocator subtraction overflow issues in tests/bench_test.rs. --- src/elaborate/mod.rs | 15 ++++ src/lib.rs | 17 ++++ src/optimizer/existential.rs | 12 +++ src/parser/grammar.rs | 34 ++++++++ src/parser/lexer.rs | 3 + src/runtime/engine.rs | 157 ++++++++++++++++++++++++++++++++++- src/syntax/expr.rs | 62 ++++++++++++++ src/typing/checker.rs | 30 +++++++ tests/all_reduce_test.rs | 132 +++++++++++++++++++++++++++++ tests/bench_test.rs | 12 +-- 10 files changed, 466 insertions(+), 8 deletions(-) create mode 100644 tests/all_reduce_test.rs diff --git a/src/elaborate/mod.rs b/src/elaborate/mod.rs index d011b97..2da9177 100644 --- a/src/elaborate/mod.rs +++ b/src/elaborate/mod.rs @@ -174,6 +174,21 @@ pub fn elaborate_expr(e: Expr) -> Expr { filter: filter.map(|f| Box::new(elaborate_expr(*f))), body: Box::new(elaborate_expr(*body)), }, + Expr::AllReduce { + acc_name, + initial, + step_var, + list_expr, + reduction, + predicate, + } => Expr::AllReduce { + acc_name, + initial: Box::new(elaborate_expr(*initial)), + step_var, + list_expr: Box::new(elaborate_expr(*list_expr)), + reduction: Box::new(elaborate_expr(*reduction)), + predicate: Box::new(elaborate_expr(*predicate)), + }, other => other, } } diff --git a/src/lib.rs b/src/lib.rs index 18cf944..03da0a9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -65,6 +65,23 @@ fn collect_expr_vars(expr: &Expr, acc: &mut HashSet) { inner.remove(var); acc.extend(inner); } + Expr::AllReduce { + acc_name, + initial, + step_var, + list_expr, + reduction, + predicate, + } => { + collect_expr_vars(initial, acc); + collect_expr_vars(list_expr, acc); + let mut inner = HashSet::new(); + collect_expr_vars(reduction, &mut inner); + collect_expr_vars(predicate, &mut inner); + inner.remove(acc_name); + inner.remove(step_var); + acc.extend(inner); + } Expr::Agg(agg) => match agg.as_ref() { Aggregator::CountStar => {} Aggregator::GeneralSet { expr, .. } => { diff --git a/src/optimizer/existential.rs b/src/optimizer/existential.rs index 73c99a2..23135e8 100644 --- a/src/optimizer/existential.rs +++ b/src/optimizer/existential.rs @@ -143,6 +143,18 @@ fn fold_in_expr(e: &mut Expr, schema: &Schema) { } fold_in_expr(body, schema); } + Expr::AllReduce { + initial, + list_expr, + reduction, + predicate, + .. + } => { + fold_in_expr(initial, schema); + fold_in_expr(list_expr, schema); + fold_in_expr(reduction, schema); + fold_in_expr(predicate, schema); + } Expr::Const(_) | Expr::Var(_) | Expr::AttrLookup { .. } | Expr::Type(_) => {} } diff --git a/src/parser/grammar.rs b/src/parser/grammar.rs index 8b915b1..0b30d4a 100644 --- a/src/parser/grammar.rs +++ b/src/parser/grammar.rs @@ -1954,6 +1954,10 @@ impl Parser { body: Box::new(body), }) } + Token::AllReduce => { + self.advance(); + self.parse_all_reduce() + } _ => Err(format!("expected expression, got {:?}", self.peek())), } } @@ -2039,6 +2043,36 @@ impl Parser { }) } + fn parse_all_reduce(&mut self) -> Result { + self.expect(&Token::LParen)?; + let acc_name = match self.advance() { + Token::Name(n) => n, + t => return Err(format!("allReduce: expected accumulator variable name, got {t:?}")), + }; + self.expect(&Token::Eq)?; + let initial = self.expr()?; + self.expect(&Token::Comma)?; + let step_var = match self.advance() { + Token::Name(n) => n, + t => return Err(format!("allReduce: expected step variable name, got {t:?}")), + }; + self.expect(&Token::In)?; + let list_expr = self.expr()?; + self.expect(&Token::Pipe)?; + let reduction = self.expr()?; + self.expect(&Token::Comma)?; + let predicate = self.expr()?; + self.expect(&Token::RParen)?; + Ok(Expr::AllReduce { + acc_name, + initial: Box::new(initial), + step_var, + list_expr: Box::new(list_expr), + reduction: Box::new(reduction), + predicate: Box::new(predicate), + }) + } + // ===== Top-level statement: query or DDL ===== fn statement(&mut self) -> Result { diff --git a/src/parser/lexer.rs b/src/parser/lexer.rs index e03678b..07e6265 100644 --- a/src/parser/lexer.rs +++ b/src/parser/lexer.rs @@ -112,6 +112,8 @@ pub enum Token { /// Soft keyword (only before `{`) so `value` stays usable as an /// identifier. Runs a correlated subquery and projects one value. Value, + /// Cypher 25 `allReduce` function. Soft keyword (only before `(`). + AllReduce, // Symbols LParen, // ( @@ -616,6 +618,7 @@ impl Lexer { } "FLOOR" | "floor" if self.peek_non_space() == Some('(') => Token::Floor, "CAST" | "cast" if self.peek_non_space() == Some('(') => Token::Cast, + "allReduce" | "ALLREDUCE" if self.peek_non_space() == Some('(') => Token::AllReduce, // RECORD / VALUE are gated on a following `{` so the // lowercase forms stay usable as identifiers. "RECORD" | "record" if self.peek_non_space() == Some('{') => Token::Record, diff --git a/src/runtime/engine.rs b/src/runtime/engine.rs index 5968357..e422b87 100644 --- a/src/runtime/engine.rs +++ b/src/runtime/engine.rs @@ -334,6 +334,9 @@ pub struct Runtime<'g, G: GraphAccess> { /// comprehension element (a `Value`, which `Assignment` cannot hold) /// resolves correctly. Pushed/popped per element during evaluation. comprehension_scope: RefCell>, + /// Active query Filters at the parent level, used to extract allReduce + /// predicates for early path pruning during repetition traversals. + active_filters: RefCell>, /// Outer bindings made visible inside a *parameter-correlated* subquery /// body — outer variables referenced in the body's WHERE/RETURN but not /// bound by the body's own pattern. `Expr::Var` / `Expr::AttrLookup` @@ -397,6 +400,7 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { value_subquery_cache: RefCell::new(HashMap::new()), unbounded_policy: Cell::new(UnboundedPolicy::Forbidden), comprehension_scope: RefCell::new(Vec::new()), + active_filters: RefCell::new(Vec::new()), correlation_scope: RefCell::new(Vec::new()), } } @@ -413,6 +417,7 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { value_subquery_cache: RefCell::new(HashMap::new()), unbounded_policy: Cell::new(UnboundedPolicy::Forbidden), comprehension_scope: RefCell::new(Vec::new()), + active_filters: RefCell::new(Vec::new()), correlation_scope: RefCell::new(Vec::new()), } } @@ -853,7 +858,10 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { Some(IntermediateResult::new(rows)) } PathPattern::Filter(inner, expr) => { - let ir = self.pinned_run(inner, index, var, id)?; + self.active_filters.borrow_mut().push(expr.clone()); + let ir = self.pinned_run(inner, index, var, id); + self.active_filters.borrow_mut().pop(); + let ir = ir?; let filtered: Vec = ir .rows .into_iter() @@ -889,7 +897,10 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { Some(IntermediateResult::new(rows)) } PathPattern::Filter(inner, expr) => { - let ir = self.pinned_run_multi(inner, index, pins, 0)?; + self.active_filters.borrow_mut().push(expr.clone()); + let ir = self.pinned_run_multi(inner, index, pins, 0); + self.active_filters.borrow_mut().pop(); + let ir = ir?; let filtered: Vec = ir .rows .into_iter() @@ -1425,7 +1436,9 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { PathPattern::Filter(inner, expr) => { // For filters, we can't know how many pre-filter results we need, // so pass 0 (unlimited) to inner and filter+truncate after. + self.active_filters.borrow_mut().push(expr.clone()); let ir = self.run_path_pattern(inner, 0); + self.active_filters.borrow_mut().pop(); let mut rows = Vec::new(); for r in ir.rows { if self.run_expr(&r.assignment, expr).get_bool() { @@ -1999,6 +2012,13 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { _ => None, }; + let mut active_reduces = Vec::new(); + if let Some(ref name) = var_name { + for filter_expr in self.active_filters.borrow().iter() { + active_reduces.extend(find_all_reduce_for_var(filter_expr, name)); + } + } + let freevars = edge_pat.freevars(); let mut accumulated_rows = Vec::new(); @@ -2020,6 +2040,16 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { let edge_val = Self::get_last_edge(row.path()); row.assignment.extend(name.clone(), PathValue::Group(vec![edge_val])); } + if !active_reduces.is_empty() { + level_ir.rows.retain(|row| { + active_reduces.iter().all(|expr| { + matches!( + self.run_expr(&row.assignment, expr), + ExprResult::Success(Value::Bool(true)) + ) + }) + }); + } } if lb <= 1 { for r in &level_ir.rows { @@ -2043,6 +2073,16 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { row.assignment.extend(name.clone(), PathValue::Group(new_group)); } } + if !active_reduces.is_empty() { + level_ir.rows.retain(|row| { + active_reduces.iter().all(|expr| { + matches!( + self.run_expr(&row.assignment, expr), + ExprResult::Success(Value::Bool(true)) + ) + }) + }); + } } if k >= lb { for r in &level_ir.rows { @@ -3327,6 +3367,62 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { ExprResult::Success(Value::List(out)) } + Expr::AllReduce { + acc_name, + initial, + step_var, + list_expr, + reduction, + predicate, + } => { + let items = match self.run_expr(mu, list_expr) { + ExprResult::Success(Value::List(items)) => items, + ExprResult::Success(Value::Null) | ExprResult::Failure(_) => { + return ExprResult::Success(Value::Bool(true)); + } + ExprResult::Success(other) => { + return ExprResult::Failure(format!( + "allReduce list expression is not a list: {other}" + )); + } + }; + let mut acc_value = match self.run_expr(mu, initial) { + ExprResult::Success(v) => v, + ExprResult::Failure(err) => return ExprResult::Failure(err), + }; + for item in items { + self.comprehension_scope + .borrow_mut() + .push((acc_name.clone(), acc_value.clone())); + self.comprehension_scope + .borrow_mut() + .push((step_var.clone(), item.clone())); + let next_acc = self.run_expr(mu, reduction); + self.comprehension_scope.borrow_mut().pop(); + self.comprehension_scope.borrow_mut().pop(); + acc_value = match next_acc { + ExprResult::Success(v) => v, + ExprResult::Failure(err) => return ExprResult::Failure(err), + }; + self.comprehension_scope + .borrow_mut() + .push((acc_name.clone(), acc_value.clone())); + self.comprehension_scope + .borrow_mut() + .push((step_var.clone(), item.clone())); + let pred_res = self.run_expr(mu, predicate); + self.comprehension_scope.borrow_mut().pop(); + self.comprehension_scope.borrow_mut().pop(); + match pred_res { + ExprResult::Success(Value::Bool(true)) => {} + _ => { + return ExprResult::Success(Value::Bool(false)); + } + } + } + ExprResult::Success(Value::Bool(true)) + } + Expr::Type(_) => ExprResult::Failure("bare type in expression".into()), Expr::Call { name, args } => self.eval_call(mu, name, args), @@ -4904,6 +5000,63 @@ fn value_cmp(a: &Value, b: &Value) -> Option { } } +fn find_all_reduce_for_var(expr: &Expr, var_name: &str) -> Vec { + let mut results = Vec::new(); + fn walk(e: &Expr, var_name: &str, results: &mut Vec) { + match e { + Expr::AllReduce { list_expr, .. } => { + if let Expr::Var(ref name) = **list_expr { + if name == var_name { + results.push(e.clone()); + } + } + } + Expr::Binop { left, right, .. } => { + walk(left, var_name, results); + walk(right, var_name, results); + } + Expr::Unop { operand, .. } => { + walk(operand, var_name, results); + } + Expr::IsNull { operand, .. } => { + walk(operand, var_name, results); + } + Expr::FieldAccess { base, .. } => { + walk(base, var_name, results); + } + Expr::Coalesce(args) | Expr::Call { args, .. } => { + for a in args { + walk(a, var_name, results); + } + } + Expr::Record { fields } => { + for (_, ex) in fields { + walk(ex, var_name, results); + } + } + Expr::Case { branches, else_expr } => { + for (cond, value) in branches { + walk(cond, var_name, results); + walk(value, var_name, results); + } + if let Some(value) = else_expr { + walk(value, var_name, results); + } + } + Expr::ListComprehension { source, filter, body, .. } => { + walk(source, var_name, results); + if let Some(f) = filter { + walk(f, var_name, results); + } + walk(body, var_name, results); + } + _ => {} + } + } + walk(expr, var_name, &mut results); + results +} + #[cfg(test)] mod group_key_tests { //! Tests for `GroupKey` — the Hash + Eq wrapper used for grouping diff --git a/src/syntax/expr.rs b/src/syntax/expr.rs index 8793b50..91c0aca 100644 --- a/src/syntax/expr.rs +++ b/src/syntax/expr.rs @@ -210,6 +210,14 @@ pub enum Expr { filter: Option>, body: Box, }, + AllReduce { + acc_name: String, + initial: Box, + step_var: String, + list_expr: Box, + reduction: Box, + predicate: Box, + }, /// Right-hand side of `is`/`as` operators — a type, not a value. Type(SimpleType), /// `EXISTS { }` — Boolean predicate over a subquery body. @@ -268,6 +276,18 @@ impl Expr { || filter.as_deref().is_some_and(Expr::contains_agg) || body.contains_agg() } + Expr::AllReduce { + initial, + list_expr, + reduction, + predicate, + .. + } => { + initial.contains_agg() + || list_expr.contains_agg() + || reduction.contains_agg() + || predicate.contains_agg() + } Expr::Const(_) | Expr::Var(_) | Expr::AttrLookup { .. } @@ -317,6 +337,18 @@ impl Expr { || filter.as_deref().is_some_and(Expr::contains_subquery) || body.contains_subquery() } + Expr::AllReduce { + initial, + list_expr, + reduction, + predicate, + .. + } => { + initial.contains_subquery() + || list_expr.contains_subquery() + || reduction.contains_subquery() + || predicate.contains_subquery() + } Expr::Const(_) | Expr::Var(_) | Expr::AttrLookup { .. } @@ -388,6 +420,23 @@ impl Expr { inner.remove(var); acc.extend(inner); } + Expr::AllReduce { + acc_name, + initial, + step_var, + list_expr, + reduction, + predicate, + } => { + initial.referenced_vars(acc); + list_expr.referenced_vars(acc); + let mut inner = std::collections::BTreeSet::new(); + reduction.referenced_vars(&mut inner); + predicate.referenced_vars(&mut inner); + inner.remove(acc_name); + inner.remove(step_var); + acc.extend(inner); + } // Subqueries are handled via `contains_subquery`; their // internal references are out of scope here. Expr::Const(_) @@ -490,6 +539,19 @@ impl fmt::Display for Expr { Some(cond) => write!(f, "[{var} IN {source} WHERE {cond} | {body}]"), None => write!(f, "[{var} IN {source} | {body}]"), }, + Expr::AllReduce { + acc_name, + initial, + step_var, + list_expr, + reduction, + predicate, + } => { + write!( + f, + "allReduce({acc_name} = {initial}, {step_var} IN {list_expr} | {reduction}, {predicate})" + ) + } Expr::Type(t) => write!(f, "{t}"), Expr::Exists { body } => write!(f, "EXISTS {{ {} }}", display_subquery(body)), Expr::NotExists { body } => write!(f, "NOT EXISTS {{ {} }}", display_subquery(body)), diff --git a/src/typing/checker.rs b/src/typing/checker.rs index 22ff420..0766f39 100644 --- a/src/typing/checker.rs +++ b/src/typing/checker.rs @@ -893,6 +893,36 @@ impl Typechecker { SimpleType::List(Box::new(body_t)) } + Expr::AllReduce { + acc_name, + initial, + step_var, + list_expr, + reduction, + predicate, + } => { + let _init_t = self.check_expr(initial, env); + let list_t = self.check_expr(list_expr, env); + if !matches!(list_t, SimpleType::List(_) | SimpleType::Star) + && SimpleType::meet(&list_t, &SimpleType::Star) != list_t + { + self.warnings.push(format!( + "allReduce source list has non-list type {list_t}" + )); + } + self.comprehension_scope.push(acc_name.clone()); + self.comprehension_scope.push(step_var.clone()); + let _red_t = self.check_expr(reduction, env); + let pred_t = self.check_expr(predicate, env); + if SimpleType::meet(&pred_t, &SimpleType::B) == SimpleType::Zero { + self.warnings + .push(format!("allReduce predicate has non-bool type {pred_t}")); + } + self.comprehension_scope.pop(); + self.comprehension_scope.pop(); + SimpleType::B + } + Expr::Record { fields } => { // A record constructor types as a closed record over the // field expression types. diff --git a/tests/all_reduce_test.rs b/tests/all_reduce_test.rs new file mode 100644 index 0000000..a64dee3 --- /dev/null +++ b/tests/all_reduce_test.rs @@ -0,0 +1,132 @@ +//! Tests for Cypher 25 `allReduce` function and early path pruning. + +use gqlrust::compile_query; +use gqlrust::model::graph::MemoryGraphStore; +use gqlrust::model::value::Value; +use gqlrust::runtime::engine::Runtime; +use gqlrust::runtime::result::QueryResult; + +fn graph() -> MemoryGraphStore { + let json = r#"{ + "nodes": [ + {"id": "a", "labels": ["Station"], "props": {"id": "a", "cost_limit": 6}}, + {"id": "b", "labels": ["Station"], "props": {"id": "b"}}, + {"id": "c", "labels": ["Station"], "props": {"id": "c"}}, + {"id": "d", "labels": ["Station"], "props": {"id": "d"}}, + {"id": "x", "labels": ["Station"], "props": {"id": "x"}}, + {"id": "y", "labels": ["Station"], "props": {"id": "y"}} + ], + "edges": [ + {"id": "e1", "labels": ["LINK"], "props": {"cost": 3}, "endpoints": ["a", "b"], "directionality": "->"}, + {"id": "e2", "labels": ["LINK"], "props": {"cost": 4}, "endpoints": ["b", "c"], "directionality": "->"}, + {"id": "e3", "labels": ["LINK"], "props": {"cost": 5}, "endpoints": ["c", "d"], "directionality": "->"}, + {"id": "e4", "labels": ["LINK"], "props": {"cost": 10}, "endpoints": ["a", "x"], "directionality": "->"}, + {"id": "e5", "labels": ["LINK"], "props": {"cost": 10}, "endpoints": ["x", "y"], "directionality": "->"} + ] + }"#; + MemoryGraphStore::from_json_str(json).unwrap() +} + +fn proj(g: &MemoryGraphStore, q: &str) -> Vec> { + let query = compile_query(q).unwrap(); + let rt = Runtime::new(g); + match rt.run_query(&query, 0) { + QueryResult::Projected(r) => r, + other => panic!("expected projected rows, got {other:?}"), + } +} + +#[test] +fn all_reduce_as_pure_expression() { + let g = graph(); + let rows = proj( + &g, + "MATCH (p:Station {id: 'a'}) RETURN allReduce(sum = 0, x IN [1, 2, 3] | sum + x, sum < 10) AS r", + ); + // 0 + 1 = 1 (<10, true) + // 1 + 2 = 3 (<10, true) + // 3 + 3 = 6 (<10, true) + // Entire list passes. + assert_eq!(rows, vec![vec![Value::Bool(true)]]); + + let rows_fail = proj( + &g, + "MATCH (p:Station {id: 'a'}) RETURN allReduce(sum = 0, x IN [1, 5, 5] | sum + x, sum < 10) AS r", + ); + // 0 + 1 = 1 (<10, true) + // 1 + 5 = 6 (<10, true) + // 6 + 5 = 11 (not <10, false) + // List fails at the last step. + assert_eq!(rows_fail, vec![vec![Value::Bool(false)]]); +} + +#[test] +fn all_reduce_path_pruning_under_6() { + let g = graph(); + // With threshold < 6: + // a -> b (cost 3, valid) + // a -> b -> c (cost 7, pruned) + // a -> x (cost 10, pruned) + let rows = proj( + &g, + "MATCH (a:Station {id: 'a'})-[e:LINK]->{1,3}(n) \ + WHERE allReduce(dist = 0, edge IN e | dist + edge.cost, dist < 6) \ + RETURN n.id AS nid ORDER BY nid ASC", + ); + assert_eq!(rows, vec![vec![Value::Str("b".into())]]); +} + +#[test] +fn all_reduce_path_pruning_under_8() { + let g = graph(); + // With threshold < 8: + // a -> b (cost 3, valid) + // a -> b -> c (cost 7, valid) + // a -> b -> c -> d (cost 12, pruned) + // a -> x (cost 10, pruned) + let rows = proj( + &g, + "MATCH (a:Station {id: 'a'})-[e:LINK]->{1,3}(n) \ + WHERE allReduce(dist = 0, edge IN e | dist + edge.cost, dist < 8) \ + RETURN n.id AS nid ORDER BY nid ASC", + ); + assert_eq!( + rows, + vec![vec![Value::Str("b".into())], vec![Value::Str("c".into())]] + ); +} + +#[test] +fn all_reduce_path_pruning_with_global_constraint() { + let g = graph(); + // With threshold < a.cost_limit (which is 6): + // a -> b (cost 3, valid) + // a -> b -> c (cost 7, pruned) + // a -> x (cost 10, pruned) + let rows = proj( + &g, + "MATCH (a:Station {id: 'a'})-[e:LINK]->{1,3}(n) \ + WHERE allReduce(dist = 0, edge IN e | dist + edge.cost, dist < a.cost_limit) \ + RETURN n.id AS nid ORDER BY nid ASC", + ); + assert_eq!(rows, vec![vec![Value::Str("b".into())]]); +} + +#[test] +fn typecheck_all_reduce() { + // Valid query compiles + assert!(compile_query( + "MATCH (a:Station)-[e:LINK]->{1,3}(n) \ + WHERE allReduce(dist = 0, edge IN e | dist + edge.cost, dist < 100) \ + RETURN n" + ) + .is_ok()); + + // Invalid list source should warn/fail typecheck depending on exact compiler strictness + // (We put a warning in check_expr if source is not a list/star). + let q = compile_query( + "MATCH (a:Station) \ + RETURN allReduce(dist = 0, edge IN 123 | dist + edge, dist < 100) AS r" + ); + assert!(q.is_ok()); // Warnings are not errors, query still compiles. +} diff --git a/tests/bench_test.rs b/tests/bench_test.rs index faf533d..fc33eab 100644 --- a/tests/bench_test.rs +++ b/tests/bench_test.rs @@ -530,15 +530,15 @@ fn bench_graph_vs_lazy() { // --- Measure memory for all three --- let before = allocated_bytes(); let g_mem = MemoryGraphStore::open(&db_path).unwrap(); - let mem_graph = allocated_bytes() - before; + let mem_graph = allocated_bytes().saturating_sub(before); let before = allocated_bytes(); let g_lazy = LazyGraphStore::open(&db_path).unwrap(); - let mem_lazy = allocated_bytes() - before; + let mem_lazy = allocated_bytes().saturating_sub(before); let before = allocated_bytes(); let g_disk = DiskGraphStore::open(&db_path).unwrap(); - let mem_disk = allocated_bytes() - before; + let mem_disk = allocated_bytes().saturating_sub(before); print_graph_stats(&g_mem); println!("\n Memory:"); @@ -650,19 +650,19 @@ fn bench_memory_scaling() { // Measure MemoryGraphStore memory let before = allocated_bytes(); let g = MemoryGraphStore::open(&db_path).unwrap(); - let graph_mem = allocated_bytes() - before; + let graph_mem = allocated_bytes().saturating_sub(before); drop(g); // Measure Lazy memory let before = allocated_bytes(); let l = LazyGraphStore::open(&db_path).unwrap(); - let lazy_mem = allocated_bytes() - before; + let lazy_mem = allocated_bytes().saturating_sub(before); drop(l); // Measure Disk memory let before = allocated_bytes(); let d = DiskGraphStore::open(&db_path).unwrap(); - let disk_mem = allocated_bytes() - before; + let disk_mem = allocated_bytes().saturating_sub(before); drop(d); println!( From 358abf76e687585c45b7433dfbc759a451f41fea Mon Sep 17 00:00:00 2001 From: Max De Marzi Date: Mon, 8 Jun 2026 23:45:09 +0000 Subject: [PATCH 3/4] Optimize allReduce traversal with incremental accumulator caching and add deep repetition tests --- src/model/value.rs | 2 +- src/runtime/engine.rs | 135 ++++++++++++++++++++++++++++++++++----- tests/all_reduce_test.rs | 26 ++++++++ 3 files changed, 147 insertions(+), 16 deletions(-) diff --git a/src/model/value.rs b/src/model/value.rs index d42b3f6..c43f132 100644 --- a/src/model/value.rs +++ b/src/model/value.rs @@ -130,7 +130,7 @@ impl fmt::Display for PathValue { } /// A path is a sequence of alternating nodes and edges: [n1, e1, n2, e2, n3, ...]. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct Path(pub Vec); impl Path { diff --git a/src/runtime/engine.rs b/src/runtime/engine.rs index e422b87..02d0cf3 100644 --- a/src/runtime/engine.rs +++ b/src/runtime/engine.rs @@ -1990,9 +1990,7 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { ub: Option, limit: usize, ) -> Option { - let Some(ub_val) = ub else { - return None; - }; + let ub_val = ub?; if !matches!( edge_pat, PathPattern::EdgeRight(_) @@ -2034,6 +2032,8 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { } } + let mut acc_cache: HashMap<(usize, Path), Value> = HashMap::new(); + let mut level_ir = self.run_concat_pattern_step(ir1, &anon_edge_pat, 0); if let Some(ref name) = var_name { for row in &mut level_ir.rows { @@ -2041,14 +2041,33 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { row.assignment.extend(name.clone(), PathValue::Group(vec![edge_val])); } if !active_reduces.is_empty() { + let mut cache_inserts = Vec::new(); level_ir.rows.retain(|row| { - active_reduces.iter().all(|expr| { - matches!( - self.run_expr(&row.assignment, expr), - ExprResult::Success(Value::Bool(true)) - ) - }) + let mut ok = true; + let edge_val = Self::get_last_edge(row.path()); + for (idx, expr) in active_reduces.iter().enumerate() { + if let Some((next_acc, match_ok)) = self.eval_all_reduce_incremental( + &row.assignment, + expr, + &edge_val, + None, + ) { + if match_ok { + cache_inserts.push(((idx, row.path().clone()), next_acc)); + } else { + ok = false; + break; + } + } else { + ok = false; + break; + } + } + ok }); + for (key, val) in cache_inserts { + acc_cache.insert(key, val); + } } } if lb <= 1 { @@ -2074,14 +2093,36 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { } } if !active_reduces.is_empty() { + let mut cache_inserts = Vec::new(); level_ir.rows.retain(|row| { - active_reduces.iter().all(|expr| { - matches!( - self.run_expr(&row.assignment, expr), - ExprResult::Success(Value::Bool(true)) - ) - }) + let mut ok = true; + let path_len = row.path().0.len(); + let parent_path = Path(row.path().0[..path_len - 2].to_vec()); + let edge_val = Self::get_last_edge(row.path()); + for (idx, expr) in active_reduces.iter().enumerate() { + let prev_acc = acc_cache.get(&(idx, parent_path.clone())).cloned(); + if let Some((next_acc, match_ok)) = self.eval_all_reduce_incremental( + &row.assignment, + expr, + &edge_val, + prev_acc, + ) { + if match_ok { + cache_inserts.push(((idx, row.path().clone()), next_acc)); + } else { + ok = false; + break; + } + } else { + ok = false; + break; + } + } + ok }); + for (key, val) in cache_inserts { + acc_cache.insert(key, val); + } } } if k >= lb { @@ -2098,6 +2139,70 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { Some(IntermediateResult::new(accumulated_rows)) } + fn eval_all_reduce_incremental( + &self, + mu: &Assignment, + expr: &Expr, + new_edge: &PathValue, + prev_acc: Option, + ) -> Option<(Value, bool)> { + let Expr::AllReduce { + acc_name, + initial, + step_var, + list_expr: _, + reduction, + predicate, + } = expr else { + return None; + }; + + // 1. Get previous/initial accumulator value + let acc_value = match prev_acc { + Some(v) => v, + None => match self.run_expr(mu, initial) { + ExprResult::Success(v) => v, + ExprResult::Failure(_) => return None, + }, + }; + + let step_value = path_value_to_value(new_edge); + + // 2. Compute next accumulator value: acc_value = reduction(acc_value, step_value) + self.comprehension_scope + .borrow_mut() + .push((acc_name.clone(), acc_value)); + self.comprehension_scope + .borrow_mut() + .push((step_var.clone(), step_value.clone())); + let next_acc_res = self.run_expr(mu, reduction); + self.comprehension_scope.borrow_mut().pop(); + self.comprehension_scope.borrow_mut().pop(); + + let next_acc = match next_acc_res { + ExprResult::Success(v) => v, + ExprResult::Failure(_) => return None, + }; + + // 3. Evaluate predicate on the next accumulator: predicate(next_acc, step_value) + self.comprehension_scope + .borrow_mut() + .push((acc_name.clone(), next_acc.clone())); + self.comprehension_scope + .borrow_mut() + .push((step_var.clone(), step_value)); + let pred_res = self.run_expr(mu, predicate); + self.comprehension_scope.borrow_mut().pop(); + self.comprehension_scope.borrow_mut().pop(); + + let ok = match pred_res { + ExprResult::Success(Value::Bool(b)) => b, + _ => false, + }; + + Some((next_acc, ok)) + } + fn run_concat_pattern_step( &self, ir: &IntermediateResult, diff --git a/tests/all_reduce_test.rs b/tests/all_reduce_test.rs index a64dee3..2b8828b 100644 --- a/tests/all_reduce_test.rs +++ b/tests/all_reduce_test.rs @@ -130,3 +130,29 @@ fn typecheck_all_reduce() { ); assert!(q.is_ok()); // Warnings are not errors, query still compiles. } + +#[test] +fn all_reduce_deep_repetition_pruning() { + let g = graph(); + // With threshold < 15 and repetition {1,8} (which exceeds MAX_UNROLL (4) and triggers try_concat_with_edge_repetition): + // a -> b (cost 3, valid) + // a -> x (cost 10, valid) + // a -> b -> c (cost 7, valid) + // a -> x -> y (cost 20, pruned at step 2) + // a -> b -> c -> d (cost 12, valid) + let rows = proj( + &g, + "MATCH (a:Station {id: 'a'})-[e:LINK]->{1,8}(n) \ + WHERE allReduce(dist = 0, edge IN e | dist + edge.cost, dist < 15) \ + RETURN n.id AS nid ORDER BY nid ASC", + ); + assert_eq!( + rows, + vec![ + vec![Value::Str("b".into())], + vec![Value::Str("c".into())], + vec![Value::Str("d".into())], + vec![Value::Str("x".into())], + ] + ); +} From 5593e820a22f5193d4d632635db8389c9abab239 Mon Sep 17 00:00:00 2001 From: Max De Marzi Date: Mon, 8 Jun 2026 23:52:42 +0000 Subject: [PATCH 4/4] Implement edge variable binding in BFS/SHORTEST path patterns --- src/runtime/engine.rs | 49 ++++++++++++++++++++++++++++++++------ tests/shortest_bfs_test.rs | 20 ++++++++++++++++ 2 files changed, 62 insertions(+), 7 deletions(-) diff --git a/src/runtime/engine.rs b/src/runtime/engine.rs index 02d0cf3..7df17ba 100644 --- a/src/runtime/engine.rs +++ b/src/runtime/engine.rs @@ -2720,9 +2720,7 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { return None; } let (edge_desc, dir) = single_edge_dir(inner)?; - if edge_desc.and_then(|d| d.var.as_deref()).is_some() { - return None; - } + let edge_var = edge_desc.and_then(|d| d.var.clone()); // Validate both endpoints are node patterns; evaluate the full // (Filter-wrapped) sub-patterns so endpoint value predicates run. node_endpoint(src)?; @@ -2764,9 +2762,17 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { if !a.assignment.can_unify(&b.assignment) { continue; } + let mut mu = a.assignment.unify(&b.assignment); + if let Some(ref name) = edge_var { + let group_pv = PathValue::Group(vec![]); + if !mu.can_unify(&Assignment::from_optional(Some(name), group_pv.clone())) { + continue; + } + mu.extend(name.clone(), group_pv); + } rows.push(ResultRow::with_paths( vec![Path(vec![PathValue::Node(id)])], - a.assignment.unify(&b.assignment), + mu, )); } } else if !self_ids.is_empty() { @@ -2778,7 +2784,7 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { if dir == BfsEdgeDir::Undirected { for &id in &self_ids { self.emit_undirected_self_cycles( - id, edge_desc, ub, groups, &src_end, &tgt_end, &mut rows, + id, edge_desc, &edge_var, ub, groups, &src_end, &tgt_end, &mut rows, ); } } else { @@ -2814,6 +2820,7 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { self.bfs_from_source( s, edge_desc, + &edge_var, dir, reverse, ub, @@ -2848,6 +2855,7 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { &self, n: Id, edge_desc: Option<&Descriptor>, + edge_var: &Option, ub: Option, groups: bool, src_end: &BfsEndpoint, @@ -2914,7 +2922,20 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { Vec::new() }; for p in paths { - rows.push(ResultRow::with_paths(vec![p], mu.clone())); + let mut row_mu = mu.clone(); + if let Some(ref name) = edge_var { + let edges: Vec = p.0.iter() + .enumerate() + .filter(|(idx, _)| idx % 2 == 1) + .map(|(_, pv)| pv.clone()) + .collect(); + let group_pv = PathValue::Group(edges); + if !row_mu.can_unify(&Assignment::from_optional(Some(name), group_pv.clone())) { + continue; + } + row_mu.extend(name.clone(), group_pv); + } + rows.push(ResultRow::with_paths(vec![p], row_mu)); } } @@ -2927,6 +2948,7 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { &self, s: Id, edge_desc: Option<&Descriptor>, + edge_var: &Option, dir: BfsEdgeDir, reverse: bool, ub: Option, @@ -3012,9 +3034,22 @@ impl<'g, G: GraphAccess> Runtime<'g, G> { if !a.assignment.can_unify(&b.assignment) { continue; } + let mut mu = a.assignment.unify(&b.assignment); + if let Some(ref name) = edge_var { + let edges: Vec = oriented.iter() + .enumerate() + .filter(|(idx, _)| idx % 2 == 1) + .map(|(_, pv)| pv.clone()) + .collect(); + let group_pv = PathValue::Group(edges); + if !mu.can_unify(&Assignment::from_optional(Some(name), group_pv.clone())) { + continue; + } + mu.extend(name.clone(), group_pv); + } rows.push(ResultRow::with_paths( vec![Path(oriented)], - a.assignment.unify(&b.assignment), + mu, )); if limit > 0 && rows.len() >= limit { return; diff --git a/tests/shortest_bfs_test.rs b/tests/shortest_bfs_test.rs index c6762aa..6bdd634 100644 --- a/tests/shortest_bfs_test.rs +++ b/tests/shortest_bfs_test.rs @@ -219,3 +219,23 @@ fn differential_unreachable() { // 3 has no outgoing R edges, so 3→0 is unreachable forward. assert_same(&g, "MATCH path = ANY SHORTEST (s:N {id:3})-[:R]->*(t:N {id:0}) RETURN s.id, t.id, PATH_LENGTH(path) AS len"); } + +#[test] +fn differential_edge_variables() { + let g = g_undirected(); + // Test ANY SHORTEST with edge variable on undirected graph + assert_same(&g, "MATCH path = ANY SHORTEST (s:N {id:0})~[e:R]~*(t:N {id:3}) RETURN s.id, t.id, e, PATH_LENGTH(path) AS len"); + // Test ALL SHORTEST with edge variable on undirected graph + assert_same(&g, "MATCH path = ALL SHORTEST (s:N {id:0})~[e:R]~*(t:N {id:3}) RETURN s.id, t.id, e, PATH_LENGTH(path) AS len"); + // Test length-0 self-match with edge variable + assert_same(&g, "MATCH path = ANY SHORTEST (s:N {id:0})~[e:R]~*(t:N {id:0}) RETURN s.id, t.id, e, PATH_LENGTH(path) AS len"); + // Test coincident endpoint with `+` undirected (which does self cycles) + assert_same(&g, "MATCH path = ANY SHORTEST (s:N {id:0})~[e:R]~+(t:N {id:0}) RETURN s.id, t.id, e, PATH_LENGTH(path) AS len"); + assert_same(&g, "MATCH path = ALL SHORTEST (s:N {id:0})~[e:R]~+(t:N {id:0}) RETURN s.id, t.id, e, PATH_LENGTH(path) AS len"); + + let g2 = g_directed(); + // Test ANY SHORTEST with edge variable on directed graph + assert_same(&g2, "MATCH path = ANY SHORTEST (s:N {id:0})-[e:R]->*(t:N {id:3}) RETURN s.id, t.id, e, PATH_LENGTH(path) AS len"); + // Test ALL SHORTEST with edge variable on directed graph + assert_same(&g2, "MATCH path = ALL SHORTEST (s:N {id:0})-[e:R]->*(t:N {id:3}) RETURN s.id, t.id, e, PATH_LENGTH(path) AS len"); +}