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 ab93e0c..03da0a9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -9,12 +9,248 @@ 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::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, .. } => { + 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/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/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/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 7aaa4ea..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() { @@ -1766,6 +1779,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 +1982,161 @@ 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 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(); + + 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 !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 { + 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 !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 { + 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( @@ -3192,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), @@ -4769,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!( 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:?}"), + } +}