Full Factorized Query Execution Engine

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,
     }
 }

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<String>) {
+    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<String>) {
+    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<String>) {
+    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<String>) -> 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());

src/model/csv_loader.rs

@@ -1051,8 +1051,8 @@ mod ldbc_tests {
         let place = g
             .node_id_by_name("Place:1")
             .expect("place 1 should be loaded");
-        match g.node_props(place).get("id") {
-            Some(Value::Int(v)) => assert_eq!(*v, 1),
+        match g.node_prop(place, "id") {
+            Some(Value::Int(v)) => assert_eq!(v, 1),
             other => panic!("expected id=Int(1) on place 1, got {:?}", other),
         }
 

src/model/graph.rs

@@ -460,6 +460,36 @@ impl super::graph_access::GraphAccess for MemoryGraphStore {
         apply_prop_mods(&mut base, overlay.mod_edge_props.get(&id));
         base
     }
+    fn node_prop(&self, id: Id, prop: &str) -> Option<Value> {
+        let overlay = self.overlay.borrow();
+        if let Some(n) = overlay.get_new_node(id) {
+            return n.props.get(prop).cloned();
+        }
+        if let Some(mods) = overlay.mod_node_props.get(&id) {
+            if let Some(op) = mods.set.get(prop) {
+                return op.clone();
+            }
+            if mods.cleared {
+                return None;
+            }
+        }
+        self.node_props[id as usize].get(prop).cloned()
+    }
+    fn edge_prop(&self, id: Id, prop: &str) -> Option<Value> {
+        let overlay = self.overlay.borrow();
+        if let Some(e) = overlay.get_new_edge(id) {
+            return e.props.get(prop).cloned();
+        }
+        if let Some(mods) = overlay.mod_edge_props.get(&id) {
+            if let Some(op) = mods.set.get(prop) {
+                return op.clone();
+            }
+            if mods.cleared {
+                return None;
+            }
+        }
+        self.edge_props[id as usize].get(prop).cloned()
+    }
     fn src(&self, edge_id: Id) -> Id {
         if let Some(e) = self.overlay.borrow().get_new_edge(edge_id) {
             return e.src;

src/model/graph_access.rs

@@ -19,6 +19,10 @@ pub trait GraphAccess {
     fn node_props(&self, id: Id) -> Props;
     /// Properties for an edge by internal ID.
     fn edge_props(&self, id: Id) -> Props;
+    /// Single property lookup for a node by internal ID and key name.
+    fn node_prop(&self, id: Id, prop: &str) -> Option<Value>;
+    /// Single property lookup for an edge by internal ID and key name.
+    fn edge_prop(&self, id: Id, prop: &str) -> Option<Value>;
     /// Source node ID of a directed edge.
     fn src(&self, edge_id: Id) -> Id;
     /// Target node ID of a directed edge.

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<PathValue>);
 
 impl Path {

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(_) => {}
     }
 

src/optimizer/pushdown.rs

@@ -363,15 +363,16 @@ fn merge_constraints(p: PathPattern, c: &Constraints) -> PathPattern {
 fn boundary_node_vars(p: &PathPattern) -> [Option<String>; 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<String>) {
+fn collect_top_node_vars(p: &PathPattern, out: &mut Vec<Option<String>>) {
     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);