1use std::collections::{BTreeMap, HashMap, HashSet};
2
3use marsdb_graph::{AdjEntry, Direction, EdgeId, GraphStore, NodeId, PropertyValue, Txn, WriteTransaction};
4
5use crate::aggregate::{property_value_hash_key, value_hash_key, AggAcc, HashKey};
6use crate::ast::{
7 is_aggregate_name, CompareOp, Expr, Literal, MergeClause, NodePattern, Pattern, PropAccess, QueryClause,
8 QueryPart, RelDirection, ReturnExpr, ReturnItem, SortDir, Statement, Tail, UnwindClause, UnwindSource,
9 WithClause, WithExpr,
10};
11use crate::error::QueryError;
12use crate::ir::{ExpandDirection, LogicalPlan};
13use crate::planner::{build_match_plan, pattern_all_vars, pattern_new_vars};
14use crate::result::QueryResult;
15use crate::value::{PathElem, Value};
16
17const OPTIONAL_SEED_IDX_KEY: &str = "__seed_idx";
21
22const MERGE_CREATED_KEY: &str = "__merge_created";
26
27#[derive(Debug, Clone)]
28enum Binding {
29 Node(NodeId),
30 Edge(EdgeId),
31 Value(PropertyValue),
35 List(Vec<Value>),
44 Path(Vec<PathBinding>),
51}
52
53#[derive(Debug, Clone)]
60enum PathBinding {
61 Node(NodeId),
62 Edge(EdgeId),
63}
64
65type BindingRow = HashMap<String, Binding>;
66
67const VAR_EXPAND_DEPTH_CAP: u32 = 30;
75
76pub struct Executor<'a> {
77 store: &'a GraphStore,
78}
79
80impl<'a> Executor<'a> {
81 pub fn new(store: &'a GraphStore) -> Self {
82 Self { store }
83 }
84
85 pub fn execute(&self, stmt: &Statement) -> Result<QueryResult, QueryError> {
97 if is_read_only(stmt) {
98 let read_txn = self.store.begin_read()?;
99 let Statement::Match {
100 clauses,
101 tail,
102 order_by,
103 limit,
104 } = stmt
105 else {
106 unreachable!("is_read_only only returns true for Statement::Match")
107 };
108 return self.execute_match(Txn::Read(&read_txn), clauses, tail, order_by, *limit);
111 }
112 let write_txn = self.store.begin_write()?;
113 let outcome = match stmt {
114 Statement::Create(patterns) => self.execute_create(&write_txn, patterns),
115 Statement::Match {
116 clauses,
117 tail,
118 order_by,
119 limit,
120 } => self.execute_match(Txn::Write(&write_txn), clauses, tail, order_by, *limit),
121 };
122 match outcome {
123 Ok(result) => {
124 GraphStore::commit(write_txn)?;
125 Ok(result)
126 }
127 Err(e) => {
128 let _ = GraphStore::abort(write_txn);
130 Err(e)
131 }
132 }
133 }
134
135 fn execute_create(&self, write_txn: &WriteTransaction, patterns: &[Pattern]) -> Result<QueryResult, QueryError> {
136 self.materialize_create(write_txn, patterns, &[BindingRow::new()])
141 }
142
143 fn materialize_create(
152 &self,
153 write_txn: &WriteTransaction,
154 patterns: &[Pattern],
155 rows: &[BindingRow],
156 ) -> Result<QueryResult, QueryError> {
157 for row in rows {
158 for pattern in patterns {
159 let mut prev_id = self.resolve_or_create_node(write_txn, &pattern.start, row)?;
160 for (rel, node) in &pattern.hops {
161 if rel.hop_range.is_some() {
162 return Err(QueryError::Parse(
163 "CREATE doesn't support variable-length relationship patterns (e.g. [:TYPE*1..3])".into(),
164 ));
165 }
166 let node_id = self.resolve_or_create_node(write_txn, node, row)?;
167
168 let rel_label = rel.rel_type.clone().unwrap_or_else(|| "REL".to_string());
169 let rel_props = literal_props_to_values(&rel.props);
170 let (src, dst) = match rel.direction {
171 RelDirection::Right => (prev_id, node_id),
172 RelDirection::Left => (node_id, prev_id),
173 RelDirection::Either => {
174 return Err(QueryError::Parse(
175 "CREATE requires a directed relationship (-> or <-), not an undirected pattern".into(),
176 ))
177 }
178 };
179 GraphStore::create_edge_in_txn(write_txn, &rel_label, src, dst, rel_props)?;
180 prev_id = node_id;
181 }
182 }
183 }
184 Ok(QueryResult {
185 columns: vec![],
186 rows: vec![],
187 })
188 }
189
190 fn resolve_or_create_node(
199 &self,
200 write_txn: &WriteTransaction,
201 node: &NodePattern,
202 row: &BindingRow,
203 ) -> Result<NodeId, QueryError> {
204 if let Some(var) = &node.var {
205 if let Some(binding) = row.get(var) {
206 let Binding::Node(id) = binding else {
207 return Err(QueryError::Parse(format!(
208 "'{var}' is not a node — can't use it as a CREATE pattern endpoint"
209 )));
210 };
211 if !node.labels.is_empty() || !node.props.is_empty() {
212 return Err(QueryError::Parse(format!(
213 "'{var}' is already bound — CREATE can't add labels/properties to an existing node"
214 )));
215 }
216 return Ok(*id);
217 }
218 }
219 let labels = pattern_labels(&node.labels);
220 let props = literal_props_to_values(&node.props);
221 Ok(GraphStore::create_node_in_txn(write_txn, &labels, props)?)
222 }
223
224 fn eval_merge(
230 &self,
231 write_txn: &WriteTransaction,
232 clause: &MergeClause,
233 rows: &[BindingRow],
234 ) -> Result<Vec<BindingRow>, QueryError> {
235 let mut out = Vec::new();
236 for row in rows {
237 out.extend(self.merge_one_row(write_txn, clause, row)?);
238 }
239 self.apply_merge_set(write_txn, clause, &mut out)?;
240 Ok(out)
241 }
242
243 fn merge_one_row(
244 &self,
245 write_txn: &WriteTransaction,
246 clause: &MergeClause,
247 row: &BindingRow,
248 ) -> Result<Vec<BindingRow>, QueryError> {
249 require_mergeable(&clause.pattern.start, row)?;
255 for (rel, node) in &clause.pattern.hops {
256 if rel.hop_range.is_some() {
257 return Err(QueryError::Parse(
258 "MERGE doesn't support variable-length relationship patterns (e.g. [:TYPE*1..3])".into(),
259 ));
260 }
261 require_mergeable(node, row)?;
262 }
263
264 let carried_vars: HashSet<String> = row.keys().cloned().collect();
274 let plan = build_match_plan(&clause.pattern, &None, &carried_vars)?;
275 let found = self.eval_plan(Txn::Write(write_txn), &plan, std::slice::from_ref(row))?;
276 if !found.is_empty() {
277 return Ok(found.into_iter().map(|r| tag_merge_created(r, false)).collect());
278 }
279
280 let mut new_row = row.clone();
285 let start_id = self.resolve_or_create_node(write_txn, &clause.pattern.start, &new_row)?;
286 if let Some(var) = &clause.pattern.start.var {
287 new_row.insert(var.clone(), Binding::Node(start_id));
288 }
289 if let Some((rel, node)) = clause.pattern.hops.first() {
293 let node_id = self.resolve_or_create_node(write_txn, node, &new_row)?;
294 if let Some(var) = &node.var {
295 new_row.insert(var.clone(), Binding::Node(node_id));
296 }
297 let rel_label = rel.rel_type.clone().unwrap_or_else(|| "REL".to_string());
298 let rel_props = literal_props_to_values(&rel.props);
299 let (src, dst) = match rel.direction {
300 RelDirection::Right => (start_id, node_id),
301 RelDirection::Left => (node_id, start_id),
302 RelDirection::Either => {
303 return Err(QueryError::Parse(
304 "MERGE requires a directed relationship (-> or <-), not an undirected pattern".into(),
305 ))
306 }
307 };
308 let edge_id = GraphStore::create_edge_in_txn(write_txn, &rel_label, src, dst, rel_props)?;
309 if let Some(var) = &rel.var {
310 new_row.insert(var.clone(), Binding::Edge(edge_id));
311 }
312 }
313 Ok(vec![tag_merge_created(new_row, true)])
314 }
315
316 fn apply_merge_set(
326 &self,
327 write_txn: &WriteTransaction,
328 clause: &MergeClause,
329 rows: &mut Vec<BindingRow>,
330 ) -> Result<(), QueryError> {
331 for row in rows.iter_mut() {
332 let created = match row.remove(MERGE_CREATED_KEY) {
333 Some(Binding::Value(PropertyValue::Bool(b))) => b,
334 other => unreachable!("{MERGE_CREATED_KEY} tagged internally as Binding::Value(Bool), got {other:?}"),
335 };
336 let items = if created { &clause.on_create } else { &clause.on_match };
337 for (pa, lit) in items {
338 let binding = row.get(&pa.var).ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
339 let value = literal_to_value(lit);
340 match binding {
341 Binding::Node(id) => {
342 GraphStore::set_node_prop_in_txn(write_txn, *id, &pa.prop, value)?;
343 }
344 Binding::Edge(id) => {
345 GraphStore::set_edge_prop_in_txn(write_txn, *id, &pa.prop, value)?;
346 }
347 Binding::Value(_) | Binding::List(_) | Binding::Path(_) => {
348 return Err(QueryError::UnboundVariable(format!(
349 "'{}' is a WITH-projected scalar, not a node/edge — SET needs a graph binding",
350 pa.var
351 )))
352 }
353 }
354 }
355 }
356 Ok(())
357 }
358
359 fn execute_match(
360 &self,
361 txn: Txn,
362 clauses: &[QueryClause],
363 tail: &Option<Tail>,
364 order_by: &Option<Vec<(ReturnExpr, SortDir)>>,
365 limit: Option<i64>,
366 ) -> Result<QueryResult, QueryError> {
367 let mut carried_vars: HashSet<String> = HashSet::new();
373 let mut current_rows: Vec<BindingRow> = vec![BindingRow::new()];
374 for clause in clauses {
375 match clause {
376 QueryClause::Match(part) => {
377 current_rows = if part.shortest_path {
378 self.eval_shortest_path(txn, part, ¤t_rows)?
381 } else if let Some(path_var) = &part.path_var {
382 let (named_pattern, synthesized) = name_pattern_for_path(&part.pattern);
383 let plan = build_match_plan(&named_pattern, &part.where_clause, &carried_vars)?;
384 let mut rows = if part.optional {
385 let new_vars = pattern_new_vars(&named_pattern, &carried_vars);
386 self.eval_optional_part(txn, &plan, ¤t_rows, &new_vars)?
387 } else {
388 self.eval_plan(txn, &plan, ¤t_rows)?
389 };
390 for row in &mut rows {
391 let path_binding = assemble_path(&named_pattern, row);
392 for key in &synthesized {
393 row.remove(key);
394 }
395 row.insert(path_var.clone(), path_binding);
396 }
397 rows
398 } else {
399 let plan = build_match_plan(&part.pattern, &part.where_clause, &carried_vars)?;
400 if part.optional {
401 let new_vars = pattern_new_vars(&part.pattern, &carried_vars);
402 self.eval_optional_part(txn, &plan, ¤t_rows, &new_vars)?
403 } else {
404 self.eval_plan(txn, &plan, ¤t_rows)?
405 }
406 };
407 let mut new_vars = pattern_all_vars(&part.pattern);
408 if let Some(path_var) = &part.path_var {
409 new_vars.insert(path_var.clone());
410 }
411 current_rows = self.apply_with_or_carry(txn, &part.with, current_rows, new_vars, &mut carried_vars)?;
412 }
413 QueryClause::Unwind(u) => {
414 current_rows = self.eval_unwind(txn, u, ¤t_rows)?;
415 current_rows = self.apply_with_or_carry(
416 txn,
417 &u.with,
418 current_rows,
419 HashSet::from([u.var.clone()]),
420 &mut carried_vars,
421 )?;
422 }
423 QueryClause::Merge(m) => {
424 let write_txn = require_write_txn(txn);
431 current_rows = self.eval_merge(write_txn, m, ¤t_rows)?;
432 current_rows = self.apply_with_or_carry(
433 txn,
434 &m.with,
435 current_rows,
436 pattern_all_vars(&m.pattern),
437 &mut carried_vars,
438 )?;
439 }
440 }
441 }
442 if order_by.is_none() {
449 if let Some(count) = limit {
450 current_rows.truncate(count.max(0) as usize);
451 }
452 }
453 let mut result = match tail {
459 None => QueryResult { columns: vec![], rows: vec![] },
465 Some(Tail::Return(items)) => self.materialize_return(txn, items, ¤t_rows)?,
466 Some(Tail::Delete(vars)) => {
467 self.materialize_delete(require_write_txn(txn), vars, ¤t_rows, false)?
468 }
469 Some(Tail::DetachDelete(vars)) => {
470 self.materialize_delete(require_write_txn(txn), vars, ¤t_rows, true)?
471 }
472 Some(Tail::Set(items)) => self.materialize_set(require_write_txn(txn), items, ¤t_rows)?,
473 Some(Tail::Create(patterns)) => {
474 self.materialize_create(require_write_txn(txn), patterns, ¤t_rows)?
475 }
476 };
477 if let Some(order_by) = order_by {
478 result.rows = apply_order_by(result.rows, &result.columns, order_by)?;
479 if let Some(count) = limit {
480 result.rows.truncate(count.max(0) as usize);
481 }
482 }
483 Ok(result)
484 }
485
486 fn apply_with_or_carry(
493 &self,
494 txn: Txn,
495 with: &Option<WithClause>,
496 rows: Vec<BindingRow>,
497 new_vars: HashSet<String>,
498 carried_vars: &mut HashSet<String>,
499 ) -> Result<Vec<BindingRow>, QueryError> {
500 let Some(with) = with else {
501 carried_vars.extend(new_vars);
502 return Ok(rows);
503 };
504 let mut rows = self.materialize_with(txn, with, &rows)?;
505 if let Some(with_order_by) = &with.order_by {
506 rows = self.apply_order_by_bindings(txn, rows, with_order_by)?;
507 }
508 if let Some(with_limit) = with.limit {
509 rows.truncate(with_limit.max(0) as usize);
510 }
511 *carried_vars = with.items.iter().enumerate().map(with_item_output_name).collect();
512 Ok(rows)
513 }
514
515 fn eval_unwind(&self, txn: Txn, clause: &UnwindClause, rows: &[BindingRow]) -> Result<Vec<BindingRow>, QueryError> {
521 let mut out = Vec::new();
522 for row in rows {
523 let elements: Vec<Binding> = match &clause.source {
524 UnwindSource::Var(name) => {
525 let binding = row.get(name).ok_or_else(|| QueryError::UnboundVariable(name.clone()))?;
526 let Binding::List(items) = binding else {
527 return Err(QueryError::Parse(format!(
528 "'{name}' isn't a list — UNWIND needs a list (e.g. from collect())"
529 )));
530 };
531 items.iter().map(value_to_binding_restore).collect()
532 }
533 UnwindSource::List(literals) => {
534 literals.iter().map(|lit| Binding::Value(literal_to_value(lit))).collect()
535 }
536 };
537 for element in elements {
538 let mut new_row = row.clone();
539 new_row.insert(clause.var.clone(), element);
540 out.push(new_row);
541 }
542 }
543 if let Some(where_clause) = &clause.where_clause {
544 let mut filtered = Vec::with_capacity(out.len());
545 for row in out {
546 if self.eval_with_expr(txn, where_clause, &row)? {
547 filtered.push(row);
548 }
549 }
550 out = filtered;
551 }
552 Ok(out)
553 }
554
555 fn eval_shortest_path(
585 &self,
586 txn: Txn,
587 part: &QueryPart,
588 rows: &[BindingRow],
589 ) -> Result<Vec<BindingRow>, QueryError> {
590 let Some(path_var) = &part.path_var else {
591 return Ok(rows.to_vec());
594 };
595 let start_var = part.pattern.start.var.as_deref().expect(
596 "shortestPath()'s start node always has a var — validated at parse time by \
597 validate_shortest_path_pattern",
598 );
599 let (rel, end_node) = &part.pattern.hops[0];
600 let end_var = end_node.var.as_deref().expect(
601 "shortestPath()'s end node always has a var — validated at parse time by \
602 validate_shortest_path_pattern",
603 );
604 let (min_hops, max_hops) = rel.hop_range.expect(
605 "shortestPath()'s relationship is always variable-length — validated at parse time by \
606 validate_shortest_path_pattern",
607 );
608 let direction = match rel.direction {
609 RelDirection::Right => ExpandDirection::Out,
610 RelDirection::Left => ExpandDirection::In,
611 RelDirection::Either => ExpandDirection::Either,
612 };
613 let rel_label = rel.rel_type.as_deref();
614
615 let mut out = Vec::with_capacity(rows.len());
616 for row in rows {
617 let start_id = require_bound_node(row, start_var)?;
618 let end_id = require_bound_node(row, end_var)?;
619 let path = self.shortest_path_between(txn, start_id, end_id, direction, rel_label, min_hops, max_hops)?;
620 let mut new_row = row.clone();
621 let binding = match path {
622 Some(elems) => Binding::Path(elems),
623 None => Binding::Value(PropertyValue::Null),
624 };
625 new_row.insert(path_var.clone(), binding);
626 out.push(new_row);
627 }
628 if let Some(where_clause) = &part.where_clause {
629 let mut filtered = Vec::with_capacity(out.len());
630 for row in out {
631 if self.eval_expr(txn, where_clause, &row)? {
632 filtered.push(row);
633 }
634 }
635 out = filtered;
636 }
637 Ok(out)
638 }
639
640 fn shortest_path_between(
649 &self,
650 txn: Txn,
651 start: NodeId,
652 end: NodeId,
653 direction: ExpandDirection,
654 rel_label: Option<&str>,
655 min_hops: u32,
656 max_hops: Option<u32>,
657 ) -> Result<Option<Vec<PathBinding>>, QueryError> {
658 if start == end && min_hops == 0 {
659 return Ok(Some(vec![PathBinding::Node(start)]));
660 }
661 let cap = max_hops.unwrap_or(VAR_EXPAND_DEPTH_CAP);
662 let mut parent: HashMap<NodeId, (NodeId, EdgeId)> = HashMap::new();
663 let mut visited: HashSet<NodeId> = HashSet::new();
664 visited.insert(start);
665 let mut frontier = vec![start];
666 let mut depth = 0u32;
667 while depth < cap && !frontier.is_empty() {
668 depth += 1;
669 let mut next_frontier = Vec::new();
670 for node in frontier {
671 for entry in neighbors_for_direction(txn, node, direction, rel_label)? {
672 if entry.other == end {
673 parent.insert(entry.other, (node, entry.edge_id));
674 return Ok(Some(reconstruct_path(&parent, start, end)));
675 }
676 if visited.insert(entry.other) {
677 parent.insert(entry.other, (node, entry.edge_id));
678 next_frontier.push(entry.other);
679 }
680 }
681 }
682 frontier = next_frontier;
683 }
684 Ok(None)
685 }
686
687 fn materialize_with(
694 &self,
695 txn: Txn,
696 with: &WithClause,
697 rows: &[BindingRow],
698 ) -> Result<Vec<BindingRow>, QueryError> {
699 let mut out = if !has_aggregate(&with.items) {
700 let mut out = Vec::with_capacity(rows.len());
701 for row in rows {
702 let mut new_row = BindingRow::new();
703 for (i, item) in with.items.iter().enumerate() {
704 let name = with_item_output_name((i, item));
705 let binding = self.item_binding(txn, &item.expr, row)?;
706 new_row.insert(name, binding);
707 }
708 out.push(new_row);
709 }
710 out
711 } else {
712 validate_return_items(&with.items)?;
713 let grouped = self.resolve_grouped_rows(txn, &with.items, rows)?;
714 grouped
715 .into_iter()
716 .map(|bindings| {
717 with.items
718 .iter()
719 .enumerate()
720 .zip(bindings)
721 .map(|((i, item), b)| (with_item_output_name((i, item)), b))
722 .collect()
723 })
724 .collect()
725 };
726 if let Some(where_clause) = &with.where_clause {
727 let mut filtered = Vec::with_capacity(out.len());
728 for row in out {
729 if self.eval_with_expr(txn, where_clause, &row)? {
730 filtered.push(row);
731 }
732 }
733 out = filtered;
734 }
735 Ok(out)
736 }
737
738 fn item_binding(&self, txn: Txn, expr: &ReturnExpr, row: &BindingRow) -> Result<Binding, QueryError> {
744 match expr {
745 ReturnExpr::Var(v) => row.get(v).cloned().ok_or_else(|| QueryError::UnboundVariable(v.clone())),
746 other => {
747 let value = self.eval_return_expr(txn, other, row)?;
748 Ok(Binding::Value(value_to_property_value(&value)))
749 }
750 }
751 }
752
753 fn apply_order_by_bindings(
758 &self,
759 txn: Txn,
760 rows: Vec<BindingRow>,
761 order_by: &[(ReturnExpr, SortDir)],
762 ) -> Result<Vec<BindingRow>, QueryError> {
763 let mut keyed: Vec<(Vec<Value>, BindingRow)> = Vec::with_capacity(rows.len());
764 for row in rows {
765 let value_map = self.binding_row_to_value_map(txn, &row)?;
766 let keys = order_by
767 .iter()
768 .map(|(expr, _)| eval_projected_expr(expr, &value_map))
769 .collect::<Result<Vec<_>, _>>()?;
770 keyed.push((keys, row));
771 }
772 keyed.sort_by(|(ka, _), (kb, _)| {
773 for (i, (_, dir)) in order_by.iter().enumerate() {
774 let ord = compare_with_dir(&ka[i], &kb[i], *dir);
775 if ord != std::cmp::Ordering::Equal {
776 return ord;
777 }
778 }
779 std::cmp::Ordering::Equal
780 });
781 Ok(keyed.into_iter().map(|(_, row)| row).collect())
782 }
783
784 fn binding_row_to_value_map(
785 &self,
786 txn: Txn,
787 row: &BindingRow,
788 ) -> Result<HashMap<String, Value>, QueryError> {
789 let mut map = HashMap::with_capacity(row.len());
790 for (k, binding) in row {
791 map.insert(k.clone(), self.binding_to_value(txn, binding)?);
792 }
793 Ok(map)
794 }
795
796 fn binding_to_value(&self, txn: Txn, b: &Binding) -> Result<Value, QueryError> {
801 Ok(match b {
802 Binding::Node(id) => Value::Node(
803 GraphStore::get_node_in_txn(txn, *id)?
804 .expect("bound node exists within this statement's transaction"),
805 ),
806 Binding::Edge(id) => Value::Edge(
807 GraphStore::get_edge_in_txn(txn, *id)?
808 .expect("bound edge exists within this statement's transaction"),
809 ),
810 Binding::Value(PropertyValue::Null) => Value::Null,
811 Binding::Value(pv) => Value::Property(pv.clone()),
812 Binding::List(items) => Value::List(items.clone()),
813 Binding::Path(elems) => Value::Path(self.resolve_path_elems(txn, elems)?),
814 })
815 }
816
817 fn resolve_path_elems(&self, txn: Txn, elems: &[PathBinding]) -> Result<Vec<PathElem>, QueryError> {
822 elems
823 .iter()
824 .map(|e| {
825 Ok(match e {
826 PathBinding::Node(id) => PathElem::Node(
827 GraphStore::get_node_in_txn(txn, *id)?
828 .expect("bound node exists within this statement's transaction"),
829 ),
830 PathBinding::Edge(id) => PathElem::Edge(
831 GraphStore::get_edge_in_txn(txn, *id)?
832 .expect("bound edge exists within this statement's transaction"),
833 ),
834 })
835 })
836 .collect()
837 }
838
839 fn resolve_grouped_rows(
861 &self,
862 txn: Txn,
863 items: &[ReturnItem],
864 rows: &[BindingRow],
865 ) -> Result<Vec<Vec<Binding>>, QueryError> {
866 struct Group {
867 key_bindings: Vec<Option<Binding>>,
872 accs: Vec<Option<AggAcc>>,
873 row_count: i64,
874 }
875 fn fresh_accs(items: &[ReturnItem]) -> Vec<Option<AggAcc>> {
876 items
877 .iter()
878 .map(|item| match &item.expr {
879 ReturnExpr::Call { name, distinct, .. } if is_aggregate_name(name) => {
880 Some(AggAcc::identity(name, *distinct))
881 }
882 _ => None,
883 })
884 .collect()
885 }
886
887 let mut groups: Vec<Group> = Vec::new();
895 let mut group_index: HashMap<Vec<Option<HashKey>>, usize> = HashMap::new();
896 for row in rows {
897 let mut key_bindings = Vec::with_capacity(items.len());
898 for item in items {
899 key_bindings.push(if is_top_level_aggregate(&item.expr) {
900 None
901 } else {
902 Some(self.item_binding(txn, &item.expr, row)?)
903 });
904 }
905 let hash_key: Vec<Option<HashKey>> = key_bindings
906 .iter()
907 .map(|b| b.as_ref().map(binding_hash_key).transpose())
908 .collect::<Result<Vec<_>, _>>()?;
909 let group_idx = *group_index.entry(hash_key).or_insert_with(|| {
910 groups.push(Group {
911 key_bindings: key_bindings.clone(),
912 accs: fresh_accs(items),
913 row_count: 0,
914 });
915 groups.len() - 1
916 });
917 let group = &mut groups[group_idx];
918 group.row_count += 1;
919 for (i, item) in items.iter().enumerate() {
920 let ReturnExpr::Call { args, .. } = &item.expr else { continue };
921 if !is_top_level_aggregate(&item.expr) {
922 continue;
923 }
924 let value = self.eval_return_expr(txn, &args[0], row)?;
931 if !matches!(value, Value::Null) {
932 if let Some(acc) = &mut group.accs[i] {
933 acc.fold(&value)?;
934 }
935 }
936 }
937 }
938
939 let no_key_items = items.iter().all(|item| is_top_level_aggregate(&item.expr));
946 if groups.is_empty() && no_key_items {
947 groups.push(Group {
948 key_bindings: vec![None; items.len()],
949 accs: fresh_accs(items),
950 row_count: 0,
951 });
952 }
953
954 let mut out = Vec::with_capacity(groups.len());
955 for mut group in groups {
956 let mut row_out = Vec::with_capacity(items.len());
957 for (i, item) in items.iter().enumerate() {
958 let binding = if matches!(item.expr, ReturnExpr::CountStar) {
959 Binding::Value(PropertyValue::Int(group.row_count))
960 } else if is_top_level_aggregate(&item.expr) {
961 let value = group.accs[i]
962 .take()
963 .expect("aggregate item must have an accumulator")
964 .finish();
965 value_to_binding(value)
966 } else {
967 group.key_bindings[i].clone().expect("non-aggregate item must have a key binding")
968 };
969 row_out.push(binding);
970 }
971 out.push(row_out);
972 }
973 Ok(out)
974 }
975
976 fn eval_with_expr(&self, txn: Txn, expr: &WithExpr, row: &BindingRow) -> Result<bool, QueryError> {
980 Ok(match expr {
981 WithExpr::And(l, r) => self.eval_with_expr(txn, l, row)? && self.eval_with_expr(txn, r, row)?,
982 WithExpr::Or(l, r) => self.eval_with_expr(txn, l, row)? || self.eval_with_expr(txn, r, row)?,
983 WithExpr::Not(e) => !self.eval_with_expr(txn, e, row)?,
984 WithExpr::Compare(lhs, op, lit) => {
985 let value = self.eval_return_expr(txn, lhs, row)?;
986 compare_value(&value, *op, lit)
987 }
988 })
989 }
990
991 fn eval_optional_part(
1010 &self,
1011 txn: Txn,
1012 plan: &LogicalPlan,
1013 outer_rows: &[BindingRow],
1014 new_vars: &HashSet<String>,
1015 ) -> Result<Vec<BindingRow>, QueryError> {
1016 let tagged: Vec<BindingRow> = outer_rows
1017 .iter()
1018 .enumerate()
1019 .map(|(i, row)| {
1020 let mut r = row.clone();
1021 r.insert(OPTIONAL_SEED_IDX_KEY.to_string(), Binding::Value(PropertyValue::Int(i as i64)));
1022 r
1023 })
1024 .collect();
1025 let results = self.eval_plan(txn, plan, &tagged)?;
1026 let mut by_idx: HashMap<i64, Vec<BindingRow>> = HashMap::new();
1027 for mut row in results {
1028 let idx = match row.remove(OPTIONAL_SEED_IDX_KEY) {
1029 Some(Binding::Value(PropertyValue::Int(i))) => i,
1030 other => unreachable!("__seed_idx tagged internally as Binding::Value(Int), got {other:?}"),
1031 };
1032 by_idx.entry(idx).or_default().push(row);
1033 }
1034 let mut out = Vec::with_capacity(outer_rows.len());
1035 for (i, outer_row) in outer_rows.iter().enumerate() {
1036 match by_idx.remove(&(i as i64)) {
1037 Some(matches) => out.extend(matches),
1038 None => {
1039 let mut padded = outer_row.clone();
1040 for var in new_vars {
1041 padded.insert(var.clone(), Binding::Value(PropertyValue::Null));
1042 }
1043 out.push(padded);
1044 }
1045 }
1046 }
1047 Ok(out)
1048 }
1049
1050 fn eval_plan(
1051 &self,
1052 txn: Txn,
1053 plan: &LogicalPlan,
1054 seed: &[BindingRow],
1055 ) -> Result<Vec<BindingRow>, QueryError> {
1056 match plan {
1057 LogicalPlan::Seed { var } => {
1058 debug_assert!(
1059 seed.first().is_none_or(|row| row.contains_key(var)),
1060 "Seed{{var: {var:?}}} planned for a var not present in the carried-forward rows"
1061 );
1062 Ok(seed.to_vec())
1063 }
1064 LogicalPlan::AllNodesScan { var } => self.scan(txn, var, None, seed),
1065 LogicalPlan::NodeByLabelScan { var, label } => self.scan(txn, var, Some(label), seed),
1066 LogicalPlan::Expand {
1067 input,
1068 from_var,
1069 to_var,
1070 rel_var,
1071 rel_label,
1072 direction,
1073 } => {
1074 let base_rows = self.eval_plan(txn, input, seed)?;
1075 let mut out = Vec::new();
1076 for row in base_rows {
1077 let Some(Binding::Node(from_id)) = row.get(from_var).cloned() else {
1078 return Err(QueryError::UnboundVariable(from_var.clone()));
1079 };
1080 let entries = neighbors_for_direction(txn, from_id, *direction, rel_label.as_deref())?;
1081 for entry in entries {
1082 let mut new_row = row.clone();
1083 new_row.insert(to_var.clone(), Binding::Node(entry.other));
1084 if let Some(rv) = rel_var {
1085 new_row.insert(rv.clone(), Binding::Edge(entry.edge_id));
1086 }
1087 out.push(new_row);
1088 }
1089 }
1090 Ok(out)
1091 }
1092 LogicalPlan::VarExpand {
1093 input,
1094 from_var,
1095 to_var,
1096 rel_label,
1097 direction,
1098 min_hops,
1099 max_hops,
1100 } => {
1101 let base_rows = self.eval_plan(txn, input, seed)?;
1102 let mut out = Vec::new();
1103 let unbounded = max_hops.is_none();
1104 let effective_max = max_hops.unwrap_or(VAR_EXPAND_DEPTH_CAP);
1105 for row in base_rows {
1106 let Some(Binding::Node(start_id)) = row.get(from_var).cloned() else {
1107 return Err(QueryError::UnboundVariable(from_var.clone()));
1108 };
1109 let mut visited = HashSet::new();
1110 visited.insert(start_id);
1111 if *min_hops == 0 {
1112 let mut new_row = row.clone();
1113 new_row.insert(to_var.clone(), Binding::Node(start_id));
1114 out.push(new_row);
1115 }
1116 let mut frontier = vec![start_id];
1117 let mut depth = 0u32;
1118 while depth < effective_max && !frontier.is_empty() {
1119 depth += 1;
1120 let mut next_frontier = Vec::new();
1121 for node in frontier {
1122 let entries = neighbors_for_direction(txn, node, *direction, rel_label.as_deref())?;
1123 for entry in entries {
1124 if visited.insert(entry.other) {
1125 next_frontier.push(entry.other);
1126 if depth >= *min_hops {
1127 let mut new_row = row.clone();
1128 new_row.insert(to_var.clone(), Binding::Node(entry.other));
1129 out.push(new_row);
1130 }
1131 }
1132 }
1133 }
1134 frontier = next_frontier;
1135 if depth == effective_max && unbounded && !frontier.is_empty() {
1136 return Err(QueryError::Parse(format!(
1142 "variable-length traversal exceeded the safety depth cap ({VAR_EXPAND_DEPTH_CAP} \
1143 hops) — likely a cyclic graph or unexpectedly large fanout; narrow the pattern or \
1144 add an explicit upper bound (e.g. *0..10)"
1145 )));
1146 }
1147 }
1148 }
1149 Ok(out)
1150 }
1151 LogicalPlan::Filter { input, predicate } => {
1152 let rows = self.eval_plan(txn, input, seed)?;
1153 let mut out = Vec::with_capacity(rows.len());
1154 for row in rows {
1155 if self.eval_expr(txn, predicate, &row)? {
1156 out.push(row);
1157 }
1158 }
1159 Ok(out)
1160 }
1161 }
1162 }
1163
1164 fn scan(&self, txn: Txn, var: &str, label: Option<&str>, seed: &[BindingRow]) -> Result<Vec<BindingRow>, QueryError> {
1177 let nodes = GraphStore::all_nodes_in_txn(txn, label)?;
1178 let mut out = Vec::with_capacity(seed.len() * nodes.len());
1179 for base_row in seed {
1180 for n in &nodes {
1181 let mut row = base_row.clone();
1182 row.insert(var.to_string(), Binding::Node(n.id));
1183 out.push(row);
1184 }
1185 }
1186 Ok(out)
1187 }
1188
1189 fn eval_expr(&self, txn: Txn, expr: &Expr, row: &BindingRow) -> Result<bool, QueryError> {
1190 Ok(match expr {
1191 Expr::And(l, r) => self.eval_expr(txn, l, row)? && self.eval_expr(txn, r, row)?,
1192 Expr::Or(l, r) => self.eval_expr(txn, l, row)? || self.eval_expr(txn, r, row)?,
1193 Expr::Not(e) => !self.eval_expr(txn, e, row)?,
1194 Expr::Compare(pa, op, lit) => {
1195 let prop_value = self.lookup_prop(txn, pa, row)?;
1196 compare(&prop_value, *op, lit)
1197 }
1198 Expr::HasLabel(var, label) => {
1199 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
1200 let Binding::Node(id) = binding else {
1201 return Err(QueryError::UnboundVariable(var.clone()));
1202 };
1203 let node = GraphStore::get_node_in_txn(txn, *id)?;
1204 node.is_some_and(|n| n.labels.iter().any(|l| l == label))
1205 }
1206 Expr::VarEq(a, b) => {
1207 let ba = row.get(a).ok_or_else(|| QueryError::UnboundVariable(a.clone()))?;
1208 let bb = row.get(b).ok_or_else(|| QueryError::UnboundVariable(b.clone()))?;
1209 match (ba, bb) {
1210 (Binding::Node(x), Binding::Node(y)) => x == y,
1211 (Binding::Edge(x), Binding::Edge(y)) => x == y,
1212 _ => false,
1220 }
1221 }
1222 })
1223 }
1224
1225 fn lookup_prop(
1226 &self,
1227 txn: Txn,
1228 pa: &PropAccess,
1229 row: &BindingRow,
1230 ) -> Result<Option<PropertyValue>, QueryError> {
1231 let binding = row
1232 .get(&pa.var)
1233 .ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
1234 match binding {
1235 Binding::Node(id) => {
1236 let node = GraphStore::get_node_in_txn(txn, *id)?;
1237 Ok(node.and_then(|n| n.props.get(&pa.prop).cloned()))
1238 }
1239 Binding::Edge(id) => {
1240 let edge = GraphStore::get_edge_in_txn(txn, *id)?;
1241 Ok(edge.and_then(|e| e.props.get(&pa.prop).cloned()))
1242 }
1243 Binding::Value(_) | Binding::List(_) | Binding::Path(_) => Ok(None),
1249 }
1250 }
1251
1252 fn materialize_return(
1253 &self,
1254 txn: Txn,
1255 items: &[ReturnItem],
1256 rows: &[BindingRow],
1257 ) -> Result<QueryResult, QueryError> {
1258 let columns = items
1259 .iter()
1260 .enumerate()
1261 .map(|(i, item)| item.alias.clone().unwrap_or_else(|| default_column_name(&item.expr, i)))
1262 .collect();
1263 let out_rows = if !has_aggregate(items) {
1264 let mut out_rows = Vec::with_capacity(rows.len());
1265 for row in rows {
1266 let mut out_row = Vec::with_capacity(items.len());
1267 for item in items {
1268 out_row.push(self.eval_return_expr(txn, &item.expr, row)?);
1269 }
1270 out_rows.push(out_row);
1271 }
1272 out_rows
1273 } else {
1274 validate_return_items(items)?;
1275 let grouped = self.resolve_grouped_rows(txn, items, rows)?;
1276 grouped
1277 .into_iter()
1278 .map(|bindings| {
1279 bindings
1280 .iter()
1281 .map(|b| self.binding_to_value(txn, b))
1282 .collect::<Result<Vec<_>, _>>()
1283 })
1284 .collect::<Result<Vec<_>, _>>()?
1285 };
1286 Ok(QueryResult {
1287 columns,
1288 rows: out_rows,
1289 })
1290 }
1291
1292 fn eval_return_expr(
1293 &self,
1294 txn: Txn,
1295 expr: &ReturnExpr,
1296 row: &BindingRow,
1297 ) -> Result<Value, QueryError> {
1298 match expr {
1299 ReturnExpr::Var(var) => {
1300 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
1301 self.binding_to_value(txn, binding)
1302 }
1303 ReturnExpr::Prop(pa) => {
1304 let value = self.lookup_prop(txn, pa, row)?;
1305 Ok(match value {
1306 Some(PropertyValue::Null) | None => Value::Null,
1309 Some(pv) => Value::Property(pv),
1310 })
1311 }
1312 ReturnExpr::Lit(lit) => Ok(match lit {
1313 Literal::Null => Value::Null,
1314 other => Value::Literal(other.clone()),
1315 }),
1316 ReturnExpr::Call { name, args, .. } => {
1317 if is_aggregate_name(name) {
1325 return Err(QueryError::Parse(format!(
1326 "aggregate function '{name}' can only be used as a return item's top-level expression"
1327 )));
1328 }
1329 let arg_values = args
1330 .iter()
1331 .map(|a| self.eval_return_expr(txn, a, row))
1332 .collect::<Result<Vec<_>, _>>()?;
1333 call_builtin(name, &arg_values)
1334 }
1335 ReturnExpr::CountStar => Err(QueryError::Parse(
1336 "count(*) can only be used as a return item's top-level expression".into(),
1337 )),
1338 ReturnExpr::Case { test, whens, else_ } => {
1339 let test_value = match test {
1340 Some(t) => Some(self.eval_return_expr(txn, t, row)?),
1341 None => None,
1342 };
1343 for (when, then) in whens {
1344 let when_value = self.eval_return_expr(txn, when, row)?;
1345 let matched = match &test_value {
1351 Some(tv) => value_eq(tv, &when_value),
1352 None => matches!(when_value, Value::Literal(Literal::Bool(true))),
1353 };
1354 if matched {
1355 return self.eval_return_expr(txn, then, row);
1356 }
1357 }
1358 match else_ {
1359 Some(e) => self.eval_return_expr(txn, e, row),
1360 None => Ok(Value::Null),
1361 }
1362 }
1363 }
1364 }
1365
1366 fn materialize_delete(
1367 &self,
1368 write_txn: &WriteTransaction,
1369 vars: &[String],
1370 rows: &[BindingRow],
1371 detach: bool,
1372 ) -> Result<QueryResult, QueryError> {
1373 let mut deleted_nodes = HashSet::new();
1374 let mut deleted_edges = HashSet::new();
1375 for row in rows {
1376 for var in vars {
1377 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
1378 match binding {
1379 Binding::Node(id) => {
1380 if deleted_nodes.insert(*id) {
1381 GraphStore::delete_node_in_txn(write_txn, *id, detach)?;
1382 }
1383 }
1384 Binding::Edge(id) => {
1385 if deleted_edges.insert(*id) {
1386 GraphStore::delete_edge_in_txn(write_txn, *id)?;
1387 }
1388 }
1389 Binding::Value(_) | Binding::List(_) | Binding::Path(_) => {
1390 return Err(QueryError::UnboundVariable(format!(
1391 "'{var}' is a WITH-projected scalar, not a node/edge — DELETE needs a graph binding"
1392 )))
1393 }
1394 }
1395 }
1396 }
1397 Ok(QueryResult {
1398 columns: vec![],
1399 rows: vec![],
1400 })
1401 }
1402
1403 fn materialize_set(
1404 &self,
1405 write_txn: &WriteTransaction,
1406 items: &[(PropAccess, Literal)],
1407 rows: &[BindingRow],
1408 ) -> Result<QueryResult, QueryError> {
1409 for row in rows {
1410 for (pa, lit) in items {
1411 let binding = row.get(&pa.var).ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
1412 let value = literal_to_value(lit);
1413 match binding {
1414 Binding::Node(id) => {
1415 GraphStore::set_node_prop_in_txn(write_txn, *id, &pa.prop, value)?;
1416 }
1417 Binding::Edge(id) => {
1418 GraphStore::set_edge_prop_in_txn(write_txn, *id, &pa.prop, value)?;
1419 }
1420 Binding::Value(_) | Binding::List(_) | Binding::Path(_) => {
1421 return Err(QueryError::UnboundVariable(format!(
1422 "'{}' is a WITH-projected scalar, not a node/edge — SET needs a graph binding",
1423 pa.var
1424 )))
1425 }
1426 }
1427 }
1428 }
1429 Ok(QueryResult {
1430 columns: vec![],
1431 rows: vec![],
1432 })
1433 }
1434}
1435
1436fn is_read_only(stmt: &Statement) -> bool {
1451 let Statement::Match { tail: Some(Tail::Return(_)), clauses, .. } = stmt else {
1452 return false;
1453 };
1454 !clauses.iter().any(|c| matches!(c, QueryClause::Merge(_)))
1455}
1456
1457fn require_write_txn(txn: Txn<'_>) -> &WriteTransaction {
1464 let Txn::Write(write_txn) = txn else {
1465 unreachable!(
1466 "materialize_delete/materialize_set only reached via the write-dispatch path in \
1467 Executor::execute — is_read_only(stmt) is false for any statement with a Delete/ \
1468 DetachDelete/Set tail, so execute always opens a WriteTransaction for these"
1469 )
1470 };
1471 write_txn
1472}
1473
1474fn default_column_name(expr: &ReturnExpr, idx: usize) -> String {
1475 match expr {
1476 ReturnExpr::Var(v) => v.clone(),
1477 ReturnExpr::Prop(pa) => format!("{}.{}", pa.var, pa.prop),
1478 ReturnExpr::Lit(_) => format!("col{idx}"),
1479 ReturnExpr::Call { name, .. } => format!("{name}(...)"),
1480 ReturnExpr::CountStar => "count(*)".to_string(),
1481 ReturnExpr::Case { .. } => format!("case{idx}"),
1482 }
1483}
1484
1485fn with_item_output_name((i, item): (usize, &ReturnItem)) -> String {
1488 item.alias.clone().unwrap_or_else(|| default_column_name(&item.expr, i))
1489}
1490
1491fn is_top_level_aggregate(expr: &ReturnExpr) -> bool {
1495 match expr {
1496 ReturnExpr::CountStar => true,
1497 ReturnExpr::Call { name, .. } => is_aggregate_name(name),
1498 _ => false,
1499 }
1500}
1501
1502fn contains_aggregate(expr: &ReturnExpr) -> bool {
1508 match expr {
1509 ReturnExpr::CountStar => true,
1510 ReturnExpr::Call { name, args, .. } => is_aggregate_name(name) || args.iter().any(contains_aggregate),
1511 ReturnExpr::Case { test, whens, else_ } => {
1512 test.as_deref().is_some_and(contains_aggregate)
1513 || whens.iter().any(|(w, t)| contains_aggregate(w) || contains_aggregate(t))
1514 || else_.as_deref().is_some_and(contains_aggregate)
1515 }
1516 ReturnExpr::Var(_) | ReturnExpr::Prop(_) | ReturnExpr::Lit(_) => false,
1517 }
1518}
1519
1520fn has_aggregate(items: &[ReturnItem]) -> bool {
1526 items.iter().any(|item| is_top_level_aggregate(&item.expr))
1527}
1528
1529fn validate_return_items(items: &[ReturnItem]) -> Result<(), QueryError> {
1540 for item in items {
1541 match &item.expr {
1542 ReturnExpr::CountStar => {}
1543 ReturnExpr::Call { name, args, .. } if is_aggregate_name(name) => {
1544 if args.len() != 1 {
1545 return Err(QueryError::Parse(format!(
1546 "{name}() takes exactly one argument (use count(*) for a row count with no argument)"
1547 )));
1548 }
1549 if contains_aggregate(&args[0]) {
1550 return Err(QueryError::Parse(format!(
1551 "aggregate function '{name}' can't take another aggregate as an argument"
1552 )));
1553 }
1554 }
1555 other => {
1556 if contains_aggregate(other) {
1557 return Err(QueryError::Parse(
1558 "an aggregate function must be a return item's entire expression, not nested inside \
1559 another expression"
1560 .into(),
1561 ));
1562 }
1563 }
1564 }
1565 }
1566 Ok(())
1567}
1568
1569fn binding_hash_key(b: &Binding) -> Result<HashKey, QueryError> {
1576 Ok(match b {
1577 Binding::Node(id) => HashKey::Node(*id),
1578 Binding::Edge(id) => HashKey::Edge(*id),
1579 Binding::Value(pv) => property_value_hash_key(pv),
1580 Binding::List(items) => HashKey::List(items.iter().map(value_hash_key).collect::<Result<Vec<_>, _>>()?),
1581 Binding::Path(_) => {
1586 return Err(QueryError::Parse(
1587 "grouping or collecting by a path (e.g. a named-path/shortestPath() variable) isn't supported"
1588 .into(),
1589 ))
1590 }
1591 })
1592}
1593
1594fn value_to_binding(v: Value) -> Binding {
1600 match v {
1601 Value::List(items) => Binding::List(items),
1602 other => Binding::Value(value_to_property_value(&other)),
1603 }
1604}
1605
1606fn value_to_binding_restore(v: &Value) -> Binding {
1614 match v {
1615 Value::Node(n) => Binding::Node(n.id),
1616 Value::Edge(e) => Binding::Edge(e.id),
1617 Value::Property(pv) => Binding::Value(pv.clone()),
1618 Value::Literal(lit) => Binding::Value(literal_to_value(lit)),
1619 Value::List(items) => Binding::List(items.clone()),
1620 Value::Path(elems) => Binding::Path(elems.iter().map(path_elem_to_binding).collect()),
1621 Value::Null => Binding::Value(PropertyValue::Null),
1622 }
1623}
1624
1625fn path_elem_to_binding(elem: &PathElem) -> PathBinding {
1626 match elem {
1627 PathElem::Node(n) => PathBinding::Node(n.id),
1628 PathElem::Edge(e) => PathBinding::Edge(e.id),
1629 }
1630}
1631
1632fn name_pattern_for_path(pattern: &Pattern) -> (Pattern, HashSet<String>) {
1646 fn fresh(counter: &mut usize, synthesized: &mut HashSet<String>) -> String {
1647 *counter += 1;
1648 let name = format!("__path_elem{counter}");
1649 synthesized.insert(name.clone());
1650 name
1651 }
1652 let mut counter = 0usize;
1653 let mut synthesized = HashSet::new();
1654 let mut start = pattern.start.clone();
1655 if start.var.is_none() {
1656 start.var = Some(fresh(&mut counter, &mut synthesized));
1657 }
1658 let hops = pattern
1659 .hops
1660 .iter()
1661 .map(|(rel, node)| {
1662 let mut rel = rel.clone();
1663 if rel.var.is_none() {
1664 rel.var = Some(fresh(&mut counter, &mut synthesized));
1665 }
1666 let mut node = node.clone();
1667 if node.var.is_none() {
1668 node.var = Some(fresh(&mut counter, &mut synthesized));
1669 }
1670 (rel, node)
1671 })
1672 .collect();
1673 (Pattern { start, hops }, synthesized)
1674}
1675
1676fn assemble_path(pattern: &Pattern, row: &BindingRow) -> Binding {
1686 let Some(start_id) = path_node_id(pattern.start.var.as_deref(), row) else {
1687 return Binding::Value(PropertyValue::Null);
1688 };
1689 let mut elems = vec![PathBinding::Node(start_id)];
1690 for (rel, node) in &pattern.hops {
1691 let Some(edge_id) = path_edge_id(rel.var.as_deref(), row) else {
1692 return Binding::Value(PropertyValue::Null);
1693 };
1694 let Some(node_id) = path_node_id(node.var.as_deref(), row) else {
1695 return Binding::Value(PropertyValue::Null);
1696 };
1697 elems.push(PathBinding::Edge(edge_id));
1698 elems.push(PathBinding::Node(node_id));
1699 }
1700 Binding::Path(elems)
1701}
1702
1703fn path_node_id(var: Option<&str>, row: &BindingRow) -> Option<NodeId> {
1704 match var.and_then(|v| row.get(v)) {
1705 Some(Binding::Node(id)) => Some(*id),
1706 _ => None,
1707 }
1708}
1709
1710fn path_edge_id(var: Option<&str>, row: &BindingRow) -> Option<EdgeId> {
1711 match var.and_then(|v| row.get(v)) {
1712 Some(Binding::Edge(id)) => Some(*id),
1713 _ => None,
1714 }
1715}
1716
1717fn require_bound_node(row: &BindingRow, var: &str) -> Result<NodeId, QueryError> {
1718 match row.get(var) {
1719 Some(Binding::Node(id)) => Ok(*id),
1720 _ => Err(QueryError::UnboundVariable(format!(
1721 "'{var}' must already be bound to a node before shortestPath() — match it in a preceding MATCH"
1722 ))),
1723 }
1724}
1725
1726fn reconstruct_path(parent: &HashMap<NodeId, (NodeId, EdgeId)>, start: NodeId, end: NodeId) -> Vec<PathBinding> {
1732 let mut hops = Vec::new();
1733 let mut current = end;
1734 while current != start {
1735 let (prev, edge_id) = parent[¤t];
1736 hops.push((edge_id, current));
1737 current = prev;
1738 }
1739 hops.reverse();
1740 let mut elems = vec![PathBinding::Node(start)];
1741 for (edge_id, node) in hops {
1742 elems.push(PathBinding::Edge(edge_id));
1743 elems.push(PathBinding::Node(node));
1744 }
1745 elems
1746}
1747
1748fn compare_value(value: &Value, op: CompareOp, lit: &Literal) -> bool {
1753 let prop = match value {
1754 Value::Null => None,
1755 Value::Property(pv) => Some(pv.clone()),
1756 Value::Literal(l) => Some(literal_to_value(l)),
1757 Value::Node(_) | Value::Edge(_) | Value::List(_) | Value::Path(_) => None,
1758 };
1759 compare(&prop, op, lit)
1760}
1761
1762fn value_to_property_value(v: &Value) -> PropertyValue {
1772 match v {
1773 Value::Null => PropertyValue::Null,
1774 Value::Property(pv) => pv.clone(),
1775 Value::Literal(lit) => literal_to_value(lit),
1776 Value::Node(_) | Value::Edge(_) | Value::List(_) | Value::Path(_) => PropertyValue::Null,
1777 }
1778}
1779
1780fn literal_to_value(lit: &Literal) -> PropertyValue {
1781 match lit {
1782 Literal::Int(i) => PropertyValue::Int(*i),
1783 Literal::Float(f) => PropertyValue::Float(*f),
1784 Literal::String(s) => PropertyValue::String(s.clone()),
1785 Literal::Bool(b) => PropertyValue::Bool(*b),
1786 Literal::Null => PropertyValue::Null,
1787 Literal::Param(name) => {
1788 unreachable!("param ${name} must be substituted before execution — see params::substitute_params")
1789 }
1790 }
1791}
1792
1793fn literal_props_to_values(props: &[(String, Literal)]) -> BTreeMap<String, PropertyValue> {
1794 props.iter().map(|(k, v)| (k.clone(), literal_to_value(v))).collect()
1795}
1796
1797fn tag_merge_created(mut row: BindingRow, created: bool) -> BindingRow {
1798 row.insert(MERGE_CREATED_KEY.to_string(), Binding::Value(PropertyValue::Bool(created)));
1799 row
1800}
1801
1802fn require_mergeable(node: &NodePattern, row: &BindingRow) -> Result<(), QueryError> {
1812 let already_bound = node.var.as_ref().is_some_and(|v| row.contains_key(v));
1813 if !already_bound && node.labels.is_empty() && node.props.is_empty() {
1814 return Err(QueryError::Parse(
1815 "MERGE requires a label or property to match/create by — an unconstrained node pattern is ambiguous"
1816 .into(),
1817 ));
1818 }
1819 Ok(())
1820}
1821
1822fn pattern_labels(labels: &[String]) -> Vec<&str> {
1823 if labels.is_empty() {
1824 vec!["Node"]
1825 } else {
1826 labels.iter().map(|s| s.as_str()).collect()
1827 }
1828}
1829
1830fn neighbors_for_direction(
1834 txn: Txn,
1835 node: NodeId,
1836 direction: ExpandDirection,
1837 rel_label: Option<&str>,
1838) -> Result<Vec<AdjEntry>, QueryError> {
1839 Ok(match direction {
1840 ExpandDirection::Out => GraphStore::neighbors_in_txn(txn, node, Direction::Out, rel_label)?,
1841 ExpandDirection::In => GraphStore::neighbors_in_txn(txn, node, Direction::In, rel_label)?,
1842 ExpandDirection::Either => {
1843 let mut out = GraphStore::neighbors_in_txn(txn, node, Direction::Out, rel_label)?;
1844 let inbound = GraphStore::neighbors_in_txn(txn, node, Direction::In, rel_label)?;
1845 let seen: HashSet<EdgeId> = out.iter().map(|e| e.edge_id).collect();
1846 out.extend(inbound.into_iter().filter(|e| !seen.contains(&e.edge_id)));
1847 out
1848 }
1849 })
1850}
1851
1852fn compare(prop: &Option<PropertyValue>, op: CompareOp, lit: &Literal) -> bool {
1853 let Some(prop) = prop else { return false };
1854 match (prop, lit) {
1855 (PropertyValue::Int(a), Literal::Int(b)) => cmp_f64(op, *a as f64, *b as f64),
1856 (PropertyValue::Int(a), Literal::Float(b)) => cmp_f64(op, *a as f64, *b),
1857 (PropertyValue::Float(a), Literal::Float(b)) => cmp_f64(op, *a, *b),
1858 (PropertyValue::Float(a), Literal::Int(b)) => cmp_f64(op, *a, *b as f64),
1859 (PropertyValue::String(a), Literal::String(b)) => cmp_ord(op, a.as_str(), b.as_str()),
1860 (PropertyValue::Bool(a), Literal::Bool(b)) => match op {
1861 CompareOp::Eq => a == b,
1862 CompareOp::Ne => a != b,
1863 _ => false,
1864 },
1865 (PropertyValue::Null, Literal::Null) => matches!(op, CompareOp::Eq),
1866 _ => false,
1867 }
1868}
1869
1870fn cmp_f64(op: CompareOp, a: f64, b: f64) -> bool {
1871 match op {
1872 CompareOp::Eq => a == b,
1873 CompareOp::Ne => a != b,
1874 CompareOp::Lt => a < b,
1875 CompareOp::Le => a <= b,
1876 CompareOp::Gt => a > b,
1877 CompareOp::Ge => a >= b,
1878 }
1879}
1880
1881fn cmp_ord<T: PartialOrd>(op: CompareOp, a: T, b: T) -> bool {
1882 match op {
1883 CompareOp::Eq => a == b,
1884 CompareOp::Ne => a != b,
1885 CompareOp::Lt => a < b,
1886 CompareOp::Le => a <= b,
1887 CompareOp::Gt => a > b,
1888 CompareOp::Ge => a >= b,
1889 }
1890}
1891
1892pub(crate) fn value_eq(a: &Value, b: &Value) -> bool {
1900 match (a, b) {
1901 (Value::Null, Value::Null) => true,
1902 (Value::Null, _) | (_, Value::Null) => false,
1903 (Value::Property(pa), Value::Property(pb)) => pa == pb,
1904 (Value::Literal(la), Value::Literal(lb)) => la == lb,
1905 (Value::Property(pa), Value::Literal(lb)) => *pa == literal_to_value(lb),
1906 (Value::Literal(la), Value::Property(pb)) => literal_to_value(la) == *pb,
1907 (Value::Node(na), Value::Node(nb)) => na.id == nb.id,
1908 (Value::Edge(ea), Value::Edge(eb)) => ea.id == eb.id,
1909 (Value::List(la), Value::List(lb)) => la.len() == lb.len() && la.iter().zip(lb).all(|(x, y)| value_eq(x, y)),
1910 _ => false,
1911 }
1912}
1913
1914fn call_builtin(name: &str, args: &[Value]) -> Result<Value, QueryError> {
1915 match name.to_ascii_lowercase().as_str() {
1916 "coalesce" => Ok(args
1917 .iter()
1918 .find(|v| !matches!(v, Value::Null))
1919 .cloned()
1920 .unwrap_or(Value::Null)),
1921 "tointeger" => Ok(args.first().map(to_integer).unwrap_or(Value::Null)),
1922 "length" => Ok(match args.first() {
1927 Some(Value::Path(elems)) => Value::Property(PropertyValue::Int(((elems.len().max(1) - 1) / 2) as i64)),
1928 Some(Value::Null) | None => Value::Null,
1929 Some(other) => {
1930 return Err(QueryError::Parse(format!("length() expects a path, got {other:?}")))
1931 }
1932 }),
1933 other => Err(QueryError::Parse(format!("unknown function: {other}"))),
1934 }
1935}
1936
1937fn to_integer(v: &Value) -> Value {
1938 let as_str_parse = |s: &str| match s.trim().parse::<i64>() {
1939 Ok(i) => Value::Property(PropertyValue::Int(i)),
1940 Err(_) => Value::Null,
1941 };
1942 match v {
1943 Value::Property(PropertyValue::Int(i)) => Value::Property(PropertyValue::Int(*i)),
1944 Value::Property(PropertyValue::Float(f)) => Value::Property(PropertyValue::Int(*f as i64)),
1945 Value::Property(PropertyValue::String(s)) => as_str_parse(s),
1946 Value::Literal(Literal::Int(i)) => Value::Property(PropertyValue::Int(*i)),
1947 Value::Literal(Literal::Float(f)) => Value::Property(PropertyValue::Int(*f as i64)),
1948 Value::Literal(Literal::String(s)) => as_str_parse(s),
1949 _ => Value::Null,
1950 }
1951}
1952
1953fn apply_order_by(
1958 rows: Vec<Vec<Value>>,
1959 columns: &[String],
1960 order_by: &[(ReturnExpr, SortDir)],
1961) -> Result<Vec<Vec<Value>>, QueryError> {
1962 let mut keyed: Vec<(Vec<Value>, Vec<Value>)> = Vec::with_capacity(rows.len());
1963 for row in rows {
1964 let row_map: HashMap<String, Value> = columns.iter().cloned().zip(row.iter().cloned()).collect();
1965 let keys = order_by
1966 .iter()
1967 .map(|(expr, _)| eval_projected_expr(expr, &row_map))
1968 .collect::<Result<Vec<_>, _>>()?;
1969 keyed.push((keys, row));
1970 }
1971 keyed.sort_by(|(ka, _), (kb, _)| {
1972 for (i, (_, dir)) in order_by.iter().enumerate() {
1973 let ord = compare_with_dir(&ka[i], &kb[i], *dir);
1974 if ord != std::cmp::Ordering::Equal {
1975 return ord;
1976 }
1977 }
1978 std::cmp::Ordering::Equal
1979 });
1980 Ok(keyed.into_iter().map(|(_, row)| row).collect())
1981}
1982
1983fn eval_projected_expr(expr: &ReturnExpr, row: &HashMap<String, Value>) -> Result<Value, QueryError> {
1989 match expr {
1990 ReturnExpr::Var(name) => row
1991 .get(name)
1992 .cloned()
1993 .ok_or_else(|| QueryError::UnboundVariable(name.clone())),
1994 ReturnExpr::Prop(pa) => {
1995 let base = row
1996 .get(&pa.var)
1997 .ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
1998 let pv = match base {
1999 Value::Node(n) => n.props.get(&pa.prop).cloned(),
2000 Value::Edge(e) => e.props.get(&pa.prop).cloned(),
2001 _ => None,
2002 };
2003 Ok(match pv {
2004 Some(PropertyValue::Null) | None => Value::Null,
2005 Some(v) => Value::Property(v),
2006 })
2007 }
2008 ReturnExpr::Lit(lit) => Ok(match lit {
2009 Literal::Null => Value::Null,
2010 other => Value::Literal(other.clone()),
2011 }),
2012 ReturnExpr::Call { name, args, .. } => {
2013 if is_aggregate_name(name) {
2020 return Err(QueryError::Parse(format!(
2021 "aggregate function '{name}' can only be used as a return item's top-level expression"
2022 )));
2023 }
2024 let arg_values = args
2025 .iter()
2026 .map(|a| eval_projected_expr(a, row))
2027 .collect::<Result<Vec<_>, _>>()?;
2028 call_builtin(name, &arg_values)
2029 }
2030 ReturnExpr::CountStar => Err(QueryError::Parse(
2031 "count(*) can only be used as a return item's top-level expression".into(),
2032 )),
2033 ReturnExpr::Case { test, whens, else_ } => {
2034 let test_value = match test {
2035 Some(t) => Some(eval_projected_expr(t, row)?),
2036 None => None,
2037 };
2038 for (when, then) in whens {
2039 let when_value = eval_projected_expr(when, row)?;
2040 let matched = match &test_value {
2041 Some(tv) => value_eq(tv, &when_value),
2042 None => matches!(when_value, Value::Literal(Literal::Bool(true))),
2043 };
2044 if matched {
2045 return eval_projected_expr(then, row);
2046 }
2047 }
2048 match else_ {
2049 Some(e) => eval_projected_expr(e, row),
2050 None => Ok(Value::Null),
2051 }
2052 }
2053 }
2054}
2055
2056fn compare_with_dir(a: &Value, b: &Value, dir: SortDir) -> std::cmp::Ordering {
2059 use std::cmp::Ordering;
2060 let a_null = matches!(a, Value::Null);
2061 let b_null = matches!(b, Value::Null);
2062 match (a_null, b_null) {
2063 (true, true) => return Ordering::Equal,
2064 (true, false) => return Ordering::Greater,
2065 (false, true) => return Ordering::Less,
2066 (false, false) => {}
2067 }
2068 let ord = compare_non_null(a, b);
2069 if dir == SortDir::Desc {
2070 ord.reverse()
2071 } else {
2072 ord
2073 }
2074}
2075
2076fn compare_non_null(a: &Value, b: &Value) -> std::cmp::Ordering {
2077 use std::cmp::Ordering;
2078 let pa = value_to_comparable(a);
2079 let pb = value_to_comparable(b);
2080 match (pa, pb) {
2081 (Some(PropertyValue::Int(x)), Some(PropertyValue::Int(y))) => x.cmp(&y),
2082 (Some(PropertyValue::Int(x)), Some(PropertyValue::Float(y))) => {
2083 (x as f64).partial_cmp(&y).unwrap_or(Ordering::Equal)
2084 }
2085 (Some(PropertyValue::Float(x)), Some(PropertyValue::Int(y))) => {
2086 x.partial_cmp(&(y as f64)).unwrap_or(Ordering::Equal)
2087 }
2088 (Some(PropertyValue::Float(x)), Some(PropertyValue::Float(y))) => x.partial_cmp(&y).unwrap_or(Ordering::Equal),
2089 (Some(PropertyValue::String(x)), Some(PropertyValue::String(y))) => x.cmp(&y),
2090 (Some(PropertyValue::Bool(x)), Some(PropertyValue::Bool(y))) => x.cmp(&y),
2091 _ => Ordering::Equal,
2092 }
2093}
2094
2095fn value_to_comparable(v: &Value) -> Option<PropertyValue> {
2096 match v {
2097 Value::Property(pv) => Some(pv.clone()),
2098 Value::Literal(lit) => Some(literal_to_value(lit)),
2099 _ => None,
2100 }
2101}
2102
2103pub(crate) fn comparable_ordering(a: &Value, b: &Value) -> Option<std::cmp::Ordering> {
2113 use std::cmp::Ordering;
2114 let pa = value_to_comparable(a)?;
2115 let pb = value_to_comparable(b)?;
2116 Some(match (pa, pb) {
2117 (PropertyValue::Int(x), PropertyValue::Int(y)) => x.cmp(&y),
2118 (PropertyValue::Int(x), PropertyValue::Float(y)) => (x as f64).partial_cmp(&y).unwrap_or(Ordering::Equal),
2119 (PropertyValue::Float(x), PropertyValue::Int(y)) => x.partial_cmp(&(y as f64)).unwrap_or(Ordering::Equal),
2120 (PropertyValue::Float(x), PropertyValue::Float(y)) => x.partial_cmp(&y).unwrap_or(Ordering::Equal),
2121 (PropertyValue::String(x), PropertyValue::String(y)) => x.cmp(&y),
2122 (PropertyValue::Bool(x), PropertyValue::Bool(y)) => x.cmp(&y),
2123 _ => return None,
2124 })
2125}