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, Pattern, PropAccess, QueryPart, RelDirection, ReturnExpr,
8 ReturnItem, SortDir, Statement, Tail, WithClause, WithExpr,
9};
10use crate::error::QueryError;
11use crate::ir::{ExpandDirection, LogicalPlan};
12use crate::planner::{build_match_plan, pattern_all_vars, pattern_new_vars};
13use crate::result::QueryResult;
14use crate::value::Value;
15
16const OPTIONAL_SEED_IDX_KEY: &str = "__seed_idx";
20
21#[derive(Debug, Clone)]
22enum Binding {
23 Node(NodeId),
24 Edge(EdgeId),
25 Value(PropertyValue),
29 List(Vec<Value>),
37}
38
39type BindingRow = HashMap<String, Binding>;
40
41const VAR_EXPAND_DEPTH_CAP: u32 = 30;
49
50pub struct Executor<'a> {
51 store: &'a GraphStore,
52}
53
54impl<'a> Executor<'a> {
55 pub fn new(store: &'a GraphStore) -> Self {
56 Self { store }
57 }
58
59 pub fn execute(&self, stmt: &Statement) -> Result<QueryResult, QueryError> {
71 if is_read_only(stmt) {
72 let read_txn = self.store.begin_read()?;
73 let Statement::Match {
74 parts,
75 tail,
76 order_by,
77 limit,
78 } = stmt
79 else {
80 unreachable!("is_read_only only returns true for Statement::Match")
81 };
82 return self.execute_match(Txn::Read(&read_txn), parts, tail, order_by, *limit);
85 }
86 let write_txn = self.store.begin_write()?;
87 let outcome = match stmt {
88 Statement::Create(patterns) => self.execute_create(&write_txn, patterns),
89 Statement::Match {
90 parts,
91 tail,
92 order_by,
93 limit,
94 } => self.execute_match(Txn::Write(&write_txn), parts, tail, order_by, *limit),
95 };
96 match outcome {
97 Ok(result) => {
98 GraphStore::commit(write_txn)?;
99 Ok(result)
100 }
101 Err(e) => {
102 let _ = GraphStore::abort(write_txn);
104 Err(e)
105 }
106 }
107 }
108
109 fn execute_create(&self, write_txn: &WriteTransaction, patterns: &[Pattern]) -> Result<QueryResult, QueryError> {
110 for pattern in patterns {
111 let start_labels = pattern_labels(&pattern.start.labels);
112 let start_props = literal_props_to_values(&pattern.start.props);
113 let mut prev_id = GraphStore::create_node_in_txn(write_txn, &start_labels, start_props)?;
114
115 for (rel, node) in &pattern.hops {
116 if rel.hop_range.is_some() {
117 return Err(QueryError::Parse(
118 "CREATE doesn't support variable-length relationship patterns (e.g. [:TYPE*1..3])".into(),
119 ));
120 }
121 let labels = pattern_labels(&node.labels);
122 let props = literal_props_to_values(&node.props);
123 let node_id = GraphStore::create_node_in_txn(write_txn, &labels, props)?;
124
125 let rel_label = rel.rel_type.clone().unwrap_or_else(|| "REL".to_string());
126 let rel_props = literal_props_to_values(&rel.props);
127 let (src, dst) = match rel.direction {
128 RelDirection::Right => (prev_id, node_id),
129 RelDirection::Left => (node_id, prev_id),
130 RelDirection::Either => {
131 return Err(QueryError::Parse(
132 "CREATE requires a directed relationship (-> or <-), not an undirected pattern".into(),
133 ))
134 }
135 };
136 GraphStore::create_edge_in_txn(write_txn, &rel_label, src, dst, rel_props)?;
137 prev_id = node_id;
138 }
139 }
140 Ok(QueryResult {
141 columns: vec![],
142 rows: vec![],
143 })
144 }
145
146 fn execute_match(
147 &self,
148 txn: Txn,
149 parts: &[QueryPart],
150 tail: &Tail,
151 order_by: &Option<Vec<(ReturnExpr, SortDir)>>,
152 limit: Option<i64>,
153 ) -> Result<QueryResult, QueryError> {
154 let mut carried_vars: HashSet<String> = HashSet::new();
160 let mut current_rows: Vec<BindingRow> = vec![BindingRow::new()];
161 for part in parts {
162 let plan = build_match_plan(&part.pattern, &part.where_clause, &carried_vars)?;
163 current_rows = if part.optional {
164 let new_vars = pattern_new_vars(&part.pattern, &carried_vars);
165 self.eval_optional_part(txn, &plan, ¤t_rows, &new_vars)?
166 } else {
167 self.eval_plan(txn, &plan, ¤t_rows)?
168 };
169 if let Some(with) = &part.with {
170 current_rows = self.materialize_with(txn, with, ¤t_rows)?;
171 if let Some(with_order_by) = &with.order_by {
172 current_rows = self.apply_order_by_bindings(txn, current_rows, with_order_by)?;
173 }
174 if let Some(with_limit) = with.limit {
175 current_rows.truncate(with_limit.max(0) as usize);
176 }
177 carried_vars = with.items.iter().enumerate().map(with_item_output_name).collect();
178 } else {
179 carried_vars.extend(pattern_all_vars(&part.pattern));
184 }
185 }
186 if order_by.is_none() {
193 if let Some(count) = limit {
194 current_rows.truncate(count.max(0) as usize);
195 }
196 }
197 let mut result = match tail {
203 Tail::Return(items) => self.materialize_return(txn, items, ¤t_rows)?,
204 Tail::Delete(vars) => {
205 self.materialize_delete(require_write_txn(txn), vars, ¤t_rows, false)?
206 }
207 Tail::DetachDelete(vars) => {
208 self.materialize_delete(require_write_txn(txn), vars, ¤t_rows, true)?
209 }
210 Tail::Set(items) => self.materialize_set(require_write_txn(txn), items, ¤t_rows)?,
211 };
212 if let Some(order_by) = order_by {
213 result.rows = apply_order_by(result.rows, &result.columns, order_by)?;
214 if let Some(count) = limit {
215 result.rows.truncate(count.max(0) as usize);
216 }
217 }
218 Ok(result)
219 }
220
221 fn materialize_with(
228 &self,
229 txn: Txn,
230 with: &WithClause,
231 rows: &[BindingRow],
232 ) -> Result<Vec<BindingRow>, QueryError> {
233 let mut out = if !has_aggregate(&with.items) {
234 let mut out = Vec::with_capacity(rows.len());
235 for row in rows {
236 let mut new_row = BindingRow::new();
237 for (i, item) in with.items.iter().enumerate() {
238 let name = with_item_output_name((i, item));
239 let binding = self.item_binding(txn, &item.expr, row)?;
240 new_row.insert(name, binding);
241 }
242 out.push(new_row);
243 }
244 out
245 } else {
246 validate_return_items(&with.items)?;
247 let grouped = self.resolve_grouped_rows(txn, &with.items, rows)?;
248 grouped
249 .into_iter()
250 .map(|bindings| {
251 with.items
252 .iter()
253 .enumerate()
254 .zip(bindings)
255 .map(|((i, item), b)| (with_item_output_name((i, item)), b))
256 .collect()
257 })
258 .collect()
259 };
260 if let Some(where_clause) = &with.where_clause {
261 let mut filtered = Vec::with_capacity(out.len());
262 for row in out {
263 if self.eval_with_expr(txn, where_clause, &row)? {
264 filtered.push(row);
265 }
266 }
267 out = filtered;
268 }
269 Ok(out)
270 }
271
272 fn item_binding(&self, txn: Txn, expr: &ReturnExpr, row: &BindingRow) -> Result<Binding, QueryError> {
278 match expr {
279 ReturnExpr::Var(v) => row.get(v).cloned().ok_or_else(|| QueryError::UnboundVariable(v.clone())),
280 other => {
281 let value = self.eval_return_expr(txn, other, row)?;
282 Ok(Binding::Value(value_to_property_value(&value)))
283 }
284 }
285 }
286
287 fn apply_order_by_bindings(
292 &self,
293 txn: Txn,
294 rows: Vec<BindingRow>,
295 order_by: &[(ReturnExpr, SortDir)],
296 ) -> Result<Vec<BindingRow>, QueryError> {
297 let mut keyed: Vec<(Vec<Value>, BindingRow)> = Vec::with_capacity(rows.len());
298 for row in rows {
299 let value_map = self.binding_row_to_value_map(txn, &row)?;
300 let keys = order_by
301 .iter()
302 .map(|(expr, _)| eval_projected_expr(expr, &value_map))
303 .collect::<Result<Vec<_>, _>>()?;
304 keyed.push((keys, row));
305 }
306 keyed.sort_by(|(ka, _), (kb, _)| {
307 for (i, (_, dir)) in order_by.iter().enumerate() {
308 let ord = compare_with_dir(&ka[i], &kb[i], *dir);
309 if ord != std::cmp::Ordering::Equal {
310 return ord;
311 }
312 }
313 std::cmp::Ordering::Equal
314 });
315 Ok(keyed.into_iter().map(|(_, row)| row).collect())
316 }
317
318 fn binding_row_to_value_map(
319 &self,
320 txn: Txn,
321 row: &BindingRow,
322 ) -> Result<HashMap<String, Value>, QueryError> {
323 let mut map = HashMap::with_capacity(row.len());
324 for (k, binding) in row {
325 map.insert(k.clone(), self.binding_to_value(txn, binding)?);
326 }
327 Ok(map)
328 }
329
330 fn binding_to_value(&self, txn: Txn, b: &Binding) -> Result<Value, QueryError> {
335 Ok(match b {
336 Binding::Node(id) => Value::Node(
337 GraphStore::get_node_in_txn(txn, *id)?
338 .expect("bound node exists within this statement's transaction"),
339 ),
340 Binding::Edge(id) => Value::Edge(
341 GraphStore::get_edge_in_txn(txn, *id)?
342 .expect("bound edge exists within this statement's transaction"),
343 ),
344 Binding::Value(PropertyValue::Null) => Value::Null,
345 Binding::Value(pv) => Value::Property(pv.clone()),
346 Binding::List(items) => Value::List(items.clone()),
347 })
348 }
349
350 fn resolve_grouped_rows(
372 &self,
373 txn: Txn,
374 items: &[ReturnItem],
375 rows: &[BindingRow],
376 ) -> Result<Vec<Vec<Binding>>, QueryError> {
377 struct Group {
378 key_bindings: Vec<Option<Binding>>,
383 accs: Vec<Option<AggAcc>>,
384 row_count: i64,
385 }
386 fn fresh_accs(items: &[ReturnItem]) -> Vec<Option<AggAcc>> {
387 items
388 .iter()
389 .map(|item| match &item.expr {
390 ReturnExpr::Call { name, distinct, .. } if is_aggregate_name(name) => {
391 Some(AggAcc::identity(name, *distinct))
392 }
393 _ => None,
394 })
395 .collect()
396 }
397
398 let mut groups: Vec<Group> = Vec::new();
406 let mut group_index: HashMap<Vec<Option<HashKey>>, usize> = HashMap::new();
407 for row in rows {
408 let mut key_bindings = Vec::with_capacity(items.len());
409 for item in items {
410 key_bindings.push(if is_top_level_aggregate(&item.expr) {
411 None
412 } else {
413 Some(self.item_binding(txn, &item.expr, row)?)
414 });
415 }
416 let hash_key: Vec<Option<HashKey>> =
417 key_bindings.iter().map(|b| b.as_ref().map(binding_hash_key)).collect();
418 let group_idx = *group_index.entry(hash_key).or_insert_with(|| {
419 groups.push(Group {
420 key_bindings: key_bindings.clone(),
421 accs: fresh_accs(items),
422 row_count: 0,
423 });
424 groups.len() - 1
425 });
426 let group = &mut groups[group_idx];
427 group.row_count += 1;
428 for (i, item) in items.iter().enumerate() {
429 let ReturnExpr::Call { args, .. } = &item.expr else { continue };
430 if !is_top_level_aggregate(&item.expr) {
431 continue;
432 }
433 let value = self.eval_return_expr(txn, &args[0], row)?;
440 if !matches!(value, Value::Null) {
441 if let Some(acc) = &mut group.accs[i] {
442 acc.fold(&value)?;
443 }
444 }
445 }
446 }
447
448 let no_key_items = items.iter().all(|item| is_top_level_aggregate(&item.expr));
455 if groups.is_empty() && no_key_items {
456 groups.push(Group {
457 key_bindings: vec![None; items.len()],
458 accs: fresh_accs(items),
459 row_count: 0,
460 });
461 }
462
463 let mut out = Vec::with_capacity(groups.len());
464 for mut group in groups {
465 let mut row_out = Vec::with_capacity(items.len());
466 for (i, item) in items.iter().enumerate() {
467 let binding = if matches!(item.expr, ReturnExpr::CountStar) {
468 Binding::Value(PropertyValue::Int(group.row_count))
469 } else if is_top_level_aggregate(&item.expr) {
470 let value = group.accs[i]
471 .take()
472 .expect("aggregate item must have an accumulator")
473 .finish();
474 value_to_binding(value)
475 } else {
476 group.key_bindings[i].clone().expect("non-aggregate item must have a key binding")
477 };
478 row_out.push(binding);
479 }
480 out.push(row_out);
481 }
482 Ok(out)
483 }
484
485 fn eval_with_expr(&self, txn: Txn, expr: &WithExpr, row: &BindingRow) -> Result<bool, QueryError> {
489 Ok(match expr {
490 WithExpr::And(l, r) => self.eval_with_expr(txn, l, row)? && self.eval_with_expr(txn, r, row)?,
491 WithExpr::Or(l, r) => self.eval_with_expr(txn, l, row)? || self.eval_with_expr(txn, r, row)?,
492 WithExpr::Not(e) => !self.eval_with_expr(txn, e, row)?,
493 WithExpr::Compare(lhs, op, lit) => {
494 let value = self.eval_return_expr(txn, lhs, row)?;
495 compare_value(&value, *op, lit)
496 }
497 })
498 }
499
500 fn eval_optional_part(
519 &self,
520 txn: Txn,
521 plan: &LogicalPlan,
522 outer_rows: &[BindingRow],
523 new_vars: &HashSet<String>,
524 ) -> Result<Vec<BindingRow>, QueryError> {
525 let tagged: Vec<BindingRow> = outer_rows
526 .iter()
527 .enumerate()
528 .map(|(i, row)| {
529 let mut r = row.clone();
530 r.insert(OPTIONAL_SEED_IDX_KEY.to_string(), Binding::Value(PropertyValue::Int(i as i64)));
531 r
532 })
533 .collect();
534 let results = self.eval_plan(txn, plan, &tagged)?;
535 let mut by_idx: HashMap<i64, Vec<BindingRow>> = HashMap::new();
536 for mut row in results {
537 let idx = match row.remove(OPTIONAL_SEED_IDX_KEY) {
538 Some(Binding::Value(PropertyValue::Int(i))) => i,
539 other => unreachable!("__seed_idx tagged internally as Binding::Value(Int), got {other:?}"),
540 };
541 by_idx.entry(idx).or_default().push(row);
542 }
543 let mut out = Vec::with_capacity(outer_rows.len());
544 for (i, outer_row) in outer_rows.iter().enumerate() {
545 match by_idx.remove(&(i as i64)) {
546 Some(matches) => out.extend(matches),
547 None => {
548 let mut padded = outer_row.clone();
549 for var in new_vars {
550 padded.insert(var.clone(), Binding::Value(PropertyValue::Null));
551 }
552 out.push(padded);
553 }
554 }
555 }
556 Ok(out)
557 }
558
559 fn eval_plan(
560 &self,
561 txn: Txn,
562 plan: &LogicalPlan,
563 seed: &[BindingRow],
564 ) -> Result<Vec<BindingRow>, QueryError> {
565 match plan {
566 LogicalPlan::Seed { var } => {
567 debug_assert!(
568 seed.first().is_none_or(|row| row.contains_key(var)),
569 "Seed{{var: {var:?}}} planned for a var not present in the carried-forward rows"
570 );
571 Ok(seed.to_vec())
572 }
573 LogicalPlan::AllNodesScan { var } => self.scan(txn, var, None),
574 LogicalPlan::NodeByLabelScan { var, label } => self.scan(txn, var, Some(label)),
575 LogicalPlan::Expand {
576 input,
577 from_var,
578 to_var,
579 rel_var,
580 rel_label,
581 direction,
582 } => {
583 let base_rows = self.eval_plan(txn, input, seed)?;
584 let mut out = Vec::new();
585 for row in base_rows {
586 let Some(Binding::Node(from_id)) = row.get(from_var).cloned() else {
587 return Err(QueryError::UnboundVariable(from_var.clone()));
588 };
589 let entries = neighbors_for_direction(txn, from_id, *direction, rel_label.as_deref())?;
590 for entry in entries {
591 let mut new_row = row.clone();
592 new_row.insert(to_var.clone(), Binding::Node(entry.other));
593 if let Some(rv) = rel_var {
594 new_row.insert(rv.clone(), Binding::Edge(entry.edge_id));
595 }
596 out.push(new_row);
597 }
598 }
599 Ok(out)
600 }
601 LogicalPlan::VarExpand {
602 input,
603 from_var,
604 to_var,
605 rel_label,
606 direction,
607 min_hops,
608 max_hops,
609 } => {
610 let base_rows = self.eval_plan(txn, input, seed)?;
611 let mut out = Vec::new();
612 let unbounded = max_hops.is_none();
613 let effective_max = max_hops.unwrap_or(VAR_EXPAND_DEPTH_CAP);
614 for row in base_rows {
615 let Some(Binding::Node(start_id)) = row.get(from_var).cloned() else {
616 return Err(QueryError::UnboundVariable(from_var.clone()));
617 };
618 let mut visited = HashSet::new();
619 visited.insert(start_id);
620 if *min_hops == 0 {
621 let mut new_row = row.clone();
622 new_row.insert(to_var.clone(), Binding::Node(start_id));
623 out.push(new_row);
624 }
625 let mut frontier = vec![start_id];
626 let mut depth = 0u32;
627 while depth < effective_max && !frontier.is_empty() {
628 depth += 1;
629 let mut next_frontier = Vec::new();
630 for node in frontier {
631 let entries = neighbors_for_direction(txn, node, *direction, rel_label.as_deref())?;
632 for entry in entries {
633 if visited.insert(entry.other) {
634 next_frontier.push(entry.other);
635 if depth >= *min_hops {
636 let mut new_row = row.clone();
637 new_row.insert(to_var.clone(), Binding::Node(entry.other));
638 out.push(new_row);
639 }
640 }
641 }
642 }
643 frontier = next_frontier;
644 if depth == effective_max && unbounded && !frontier.is_empty() {
645 return Err(QueryError::Parse(format!(
651 "variable-length traversal exceeded the safety depth cap ({VAR_EXPAND_DEPTH_CAP} \
652 hops) — likely a cyclic graph or unexpectedly large fanout; narrow the pattern or \
653 add an explicit upper bound (e.g. *0..10)"
654 )));
655 }
656 }
657 }
658 Ok(out)
659 }
660 LogicalPlan::Filter { input, predicate } => {
661 let rows = self.eval_plan(txn, input, seed)?;
662 let mut out = Vec::with_capacity(rows.len());
663 for row in rows {
664 if self.eval_expr(txn, predicate, &row)? {
665 out.push(row);
666 }
667 }
668 Ok(out)
669 }
670 }
671 }
672
673 fn scan(&self, txn: Txn, var: &str, label: Option<&str>) -> Result<Vec<BindingRow>, QueryError> {
674 let nodes = GraphStore::all_nodes_in_txn(txn, label)?;
675 Ok(nodes
676 .into_iter()
677 .map(|n| {
678 let mut row = BindingRow::new();
679 row.insert(var.to_string(), Binding::Node(n.id));
680 row
681 })
682 .collect())
683 }
684
685 fn eval_expr(&self, txn: Txn, expr: &Expr, row: &BindingRow) -> Result<bool, QueryError> {
686 Ok(match expr {
687 Expr::And(l, r) => self.eval_expr(txn, l, row)? && self.eval_expr(txn, r, row)?,
688 Expr::Or(l, r) => self.eval_expr(txn, l, row)? || self.eval_expr(txn, r, row)?,
689 Expr::Not(e) => !self.eval_expr(txn, e, row)?,
690 Expr::Compare(pa, op, lit) => {
691 let prop_value = self.lookup_prop(txn, pa, row)?;
692 compare(&prop_value, *op, lit)
693 }
694 Expr::HasLabel(var, label) => {
695 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
696 let Binding::Node(id) = binding else {
697 return Err(QueryError::UnboundVariable(var.clone()));
698 };
699 let node = GraphStore::get_node_in_txn(txn, *id)?;
700 node.is_some_and(|n| n.labels.iter().any(|l| l == label))
701 }
702 Expr::VarEq(a, b) => {
703 let ba = row.get(a).ok_or_else(|| QueryError::UnboundVariable(a.clone()))?;
704 let bb = row.get(b).ok_or_else(|| QueryError::UnboundVariable(b.clone()))?;
705 match (ba, bb) {
706 (Binding::Node(x), Binding::Node(y)) => x == y,
707 (Binding::Edge(x), Binding::Edge(y)) => x == y,
708 _ => false,
716 }
717 }
718 })
719 }
720
721 fn lookup_prop(
722 &self,
723 txn: Txn,
724 pa: &PropAccess,
725 row: &BindingRow,
726 ) -> Result<Option<PropertyValue>, QueryError> {
727 let binding = row
728 .get(&pa.var)
729 .ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
730 match binding {
731 Binding::Node(id) => {
732 let node = GraphStore::get_node_in_txn(txn, *id)?;
733 Ok(node.and_then(|n| n.props.get(&pa.prop).cloned()))
734 }
735 Binding::Edge(id) => {
736 let edge = GraphStore::get_edge_in_txn(txn, *id)?;
737 Ok(edge.and_then(|e| e.props.get(&pa.prop).cloned()))
738 }
739 Binding::Value(_) | Binding::List(_) => Ok(None),
744 }
745 }
746
747 fn materialize_return(
748 &self,
749 txn: Txn,
750 items: &[ReturnItem],
751 rows: &[BindingRow],
752 ) -> Result<QueryResult, QueryError> {
753 let columns = items
754 .iter()
755 .enumerate()
756 .map(|(i, item)| item.alias.clone().unwrap_or_else(|| default_column_name(&item.expr, i)))
757 .collect();
758 let out_rows = if !has_aggregate(items) {
759 let mut out_rows = Vec::with_capacity(rows.len());
760 for row in rows {
761 let mut out_row = Vec::with_capacity(items.len());
762 for item in items {
763 out_row.push(self.eval_return_expr(txn, &item.expr, row)?);
764 }
765 out_rows.push(out_row);
766 }
767 out_rows
768 } else {
769 validate_return_items(items)?;
770 let grouped = self.resolve_grouped_rows(txn, items, rows)?;
771 grouped
772 .into_iter()
773 .map(|bindings| {
774 bindings
775 .iter()
776 .map(|b| self.binding_to_value(txn, b))
777 .collect::<Result<Vec<_>, _>>()
778 })
779 .collect::<Result<Vec<_>, _>>()?
780 };
781 Ok(QueryResult {
782 columns,
783 rows: out_rows,
784 })
785 }
786
787 fn eval_return_expr(
788 &self,
789 txn: Txn,
790 expr: &ReturnExpr,
791 row: &BindingRow,
792 ) -> Result<Value, QueryError> {
793 match expr {
794 ReturnExpr::Var(var) => {
795 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
796 self.binding_to_value(txn, binding)
797 }
798 ReturnExpr::Prop(pa) => {
799 let value = self.lookup_prop(txn, pa, row)?;
800 Ok(match value {
801 Some(PropertyValue::Null) | None => Value::Null,
804 Some(pv) => Value::Property(pv),
805 })
806 }
807 ReturnExpr::Lit(lit) => Ok(match lit {
808 Literal::Null => Value::Null,
809 other => Value::Literal(other.clone()),
810 }),
811 ReturnExpr::Call { name, args, .. } => {
812 if is_aggregate_name(name) {
820 return Err(QueryError::Parse(format!(
821 "aggregate function '{name}' can only be used as a return item's top-level expression"
822 )));
823 }
824 let arg_values = args
825 .iter()
826 .map(|a| self.eval_return_expr(txn, a, row))
827 .collect::<Result<Vec<_>, _>>()?;
828 call_builtin(name, &arg_values)
829 }
830 ReturnExpr::CountStar => Err(QueryError::Parse(
831 "count(*) can only be used as a return item's top-level expression".into(),
832 )),
833 ReturnExpr::Case { test, whens, else_ } => {
834 let test_value = match test {
835 Some(t) => Some(self.eval_return_expr(txn, t, row)?),
836 None => None,
837 };
838 for (when, then) in whens {
839 let when_value = self.eval_return_expr(txn, when, row)?;
840 let matched = match &test_value {
846 Some(tv) => value_eq(tv, &when_value),
847 None => matches!(when_value, Value::Literal(Literal::Bool(true))),
848 };
849 if matched {
850 return self.eval_return_expr(txn, then, row);
851 }
852 }
853 match else_ {
854 Some(e) => self.eval_return_expr(txn, e, row),
855 None => Ok(Value::Null),
856 }
857 }
858 }
859 }
860
861 fn materialize_delete(
862 &self,
863 write_txn: &WriteTransaction,
864 vars: &[String],
865 rows: &[BindingRow],
866 detach: bool,
867 ) -> Result<QueryResult, QueryError> {
868 let mut deleted_nodes = HashSet::new();
869 let mut deleted_edges = HashSet::new();
870 for row in rows {
871 for var in vars {
872 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
873 match binding {
874 Binding::Node(id) => {
875 if deleted_nodes.insert(*id) {
876 GraphStore::delete_node_in_txn(write_txn, *id, detach)?;
877 }
878 }
879 Binding::Edge(id) => {
880 if deleted_edges.insert(*id) {
881 GraphStore::delete_edge_in_txn(write_txn, *id)?;
882 }
883 }
884 Binding::Value(_) | Binding::List(_) => {
885 return Err(QueryError::UnboundVariable(format!(
886 "'{var}' is a WITH-projected scalar, not a node/edge — DELETE needs a graph binding"
887 )))
888 }
889 }
890 }
891 }
892 Ok(QueryResult {
893 columns: vec![],
894 rows: vec![],
895 })
896 }
897
898 fn materialize_set(
899 &self,
900 write_txn: &WriteTransaction,
901 items: &[(PropAccess, Literal)],
902 rows: &[BindingRow],
903 ) -> Result<QueryResult, QueryError> {
904 for row in rows {
905 for (pa, lit) in items {
906 let binding = row.get(&pa.var).ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
907 let value = literal_to_value(lit);
908 match binding {
909 Binding::Node(id) => {
910 GraphStore::set_node_prop_in_txn(write_txn, *id, &pa.prop, value)?;
911 }
912 Binding::Edge(id) => {
913 GraphStore::set_edge_prop_in_txn(write_txn, *id, &pa.prop, value)?;
914 }
915 Binding::Value(_) | Binding::List(_) => {
916 return Err(QueryError::UnboundVariable(format!(
917 "'{}' is a WITH-projected scalar, not a node/edge — SET needs a graph binding",
918 pa.var
919 )))
920 }
921 }
922 }
923 }
924 Ok(QueryResult {
925 columns: vec![],
926 rows: vec![],
927 })
928 }
929}
930
931fn is_read_only(stmt: &Statement) -> bool {
942 matches!(stmt, Statement::Match { tail: Tail::Return(_), .. })
943}
944
945fn require_write_txn(txn: Txn<'_>) -> &WriteTransaction {
952 let Txn::Write(write_txn) = txn else {
953 unreachable!(
954 "materialize_delete/materialize_set only reached via the write-dispatch path in \
955 Executor::execute — is_read_only(stmt) is false for any statement with a Delete/ \
956 DetachDelete/Set tail, so execute always opens a WriteTransaction for these"
957 )
958 };
959 write_txn
960}
961
962fn default_column_name(expr: &ReturnExpr, idx: usize) -> String {
963 match expr {
964 ReturnExpr::Var(v) => v.clone(),
965 ReturnExpr::Prop(pa) => format!("{}.{}", pa.var, pa.prop),
966 ReturnExpr::Lit(_) => format!("col{idx}"),
967 ReturnExpr::Call { name, .. } => format!("{name}(...)"),
968 ReturnExpr::CountStar => "count(*)".to_string(),
969 ReturnExpr::Case { .. } => format!("case{idx}"),
970 }
971}
972
973fn with_item_output_name((i, item): (usize, &ReturnItem)) -> String {
976 item.alias.clone().unwrap_or_else(|| default_column_name(&item.expr, i))
977}
978
979fn is_top_level_aggregate(expr: &ReturnExpr) -> bool {
983 match expr {
984 ReturnExpr::CountStar => true,
985 ReturnExpr::Call { name, .. } => is_aggregate_name(name),
986 _ => false,
987 }
988}
989
990fn contains_aggregate(expr: &ReturnExpr) -> bool {
996 match expr {
997 ReturnExpr::CountStar => true,
998 ReturnExpr::Call { name, args, .. } => is_aggregate_name(name) || args.iter().any(contains_aggregate),
999 ReturnExpr::Case { test, whens, else_ } => {
1000 test.as_deref().is_some_and(contains_aggregate)
1001 || whens.iter().any(|(w, t)| contains_aggregate(w) || contains_aggregate(t))
1002 || else_.as_deref().is_some_and(contains_aggregate)
1003 }
1004 ReturnExpr::Var(_) | ReturnExpr::Prop(_) | ReturnExpr::Lit(_) => false,
1005 }
1006}
1007
1008fn has_aggregate(items: &[ReturnItem]) -> bool {
1014 items.iter().any(|item| is_top_level_aggregate(&item.expr))
1015}
1016
1017fn validate_return_items(items: &[ReturnItem]) -> Result<(), QueryError> {
1028 for item in items {
1029 match &item.expr {
1030 ReturnExpr::CountStar => {}
1031 ReturnExpr::Call { name, args, .. } if is_aggregate_name(name) => {
1032 if args.len() != 1 {
1033 return Err(QueryError::Parse(format!(
1034 "{name}() takes exactly one argument (use count(*) for a row count with no argument)"
1035 )));
1036 }
1037 if contains_aggregate(&args[0]) {
1038 return Err(QueryError::Parse(format!(
1039 "aggregate function '{name}' can't take another aggregate as an argument"
1040 )));
1041 }
1042 }
1043 other => {
1044 if contains_aggregate(other) {
1045 return Err(QueryError::Parse(
1046 "an aggregate function must be a return item's entire expression, not nested inside \
1047 another expression"
1048 .into(),
1049 ));
1050 }
1051 }
1052 }
1053 }
1054 Ok(())
1055}
1056
1057fn binding_hash_key(b: &Binding) -> HashKey {
1064 match b {
1065 Binding::Node(id) => HashKey::Node(*id),
1066 Binding::Edge(id) => HashKey::Edge(*id),
1067 Binding::Value(pv) => property_value_hash_key(pv),
1068 Binding::List(items) => HashKey::List(items.iter().map(value_hash_key).collect()),
1069 }
1070}
1071
1072fn value_to_binding(v: Value) -> Binding {
1078 match v {
1079 Value::List(items) => Binding::List(items),
1080 other => Binding::Value(value_to_property_value(&other)),
1081 }
1082}
1083
1084fn compare_value(value: &Value, op: CompareOp, lit: &Literal) -> bool {
1089 let prop = match value {
1090 Value::Null => None,
1091 Value::Property(pv) => Some(pv.clone()),
1092 Value::Literal(l) => Some(literal_to_value(l)),
1093 Value::Node(_) | Value::Edge(_) | Value::List(_) => None,
1094 };
1095 compare(&prop, op, lit)
1096}
1097
1098fn value_to_property_value(v: &Value) -> PropertyValue {
1108 match v {
1109 Value::Null => PropertyValue::Null,
1110 Value::Property(pv) => pv.clone(),
1111 Value::Literal(lit) => literal_to_value(lit),
1112 Value::Node(_) | Value::Edge(_) | Value::List(_) => PropertyValue::Null,
1113 }
1114}
1115
1116fn literal_to_value(lit: &Literal) -> PropertyValue {
1117 match lit {
1118 Literal::Int(i) => PropertyValue::Int(*i),
1119 Literal::Float(f) => PropertyValue::Float(*f),
1120 Literal::String(s) => PropertyValue::String(s.clone()),
1121 Literal::Bool(b) => PropertyValue::Bool(*b),
1122 Literal::Null => PropertyValue::Null,
1123 Literal::Param(name) => {
1124 unreachable!("param ${name} must be substituted before execution — see params::substitute_params")
1125 }
1126 }
1127}
1128
1129fn literal_props_to_values(props: &[(String, Literal)]) -> BTreeMap<String, PropertyValue> {
1130 props.iter().map(|(k, v)| (k.clone(), literal_to_value(v))).collect()
1131}
1132
1133fn pattern_labels(labels: &[String]) -> Vec<&str> {
1134 if labels.is_empty() {
1135 vec!["Node"]
1136 } else {
1137 labels.iter().map(|s| s.as_str()).collect()
1138 }
1139}
1140
1141fn neighbors_for_direction(
1145 txn: Txn,
1146 node: NodeId,
1147 direction: ExpandDirection,
1148 rel_label: Option<&str>,
1149) -> Result<Vec<AdjEntry>, QueryError> {
1150 Ok(match direction {
1151 ExpandDirection::Out => GraphStore::neighbors_in_txn(txn, node, Direction::Out, rel_label)?,
1152 ExpandDirection::In => GraphStore::neighbors_in_txn(txn, node, Direction::In, rel_label)?,
1153 ExpandDirection::Either => {
1154 let mut out = GraphStore::neighbors_in_txn(txn, node, Direction::Out, rel_label)?;
1155 let inbound = GraphStore::neighbors_in_txn(txn, node, Direction::In, rel_label)?;
1156 let seen: HashSet<EdgeId> = out.iter().map(|e| e.edge_id).collect();
1157 out.extend(inbound.into_iter().filter(|e| !seen.contains(&e.edge_id)));
1158 out
1159 }
1160 })
1161}
1162
1163fn compare(prop: &Option<PropertyValue>, op: CompareOp, lit: &Literal) -> bool {
1164 let Some(prop) = prop else { return false };
1165 match (prop, lit) {
1166 (PropertyValue::Int(a), Literal::Int(b)) => cmp_f64(op, *a as f64, *b as f64),
1167 (PropertyValue::Int(a), Literal::Float(b)) => cmp_f64(op, *a as f64, *b),
1168 (PropertyValue::Float(a), Literal::Float(b)) => cmp_f64(op, *a, *b),
1169 (PropertyValue::Float(a), Literal::Int(b)) => cmp_f64(op, *a, *b as f64),
1170 (PropertyValue::String(a), Literal::String(b)) => cmp_ord(op, a.as_str(), b.as_str()),
1171 (PropertyValue::Bool(a), Literal::Bool(b)) => match op {
1172 CompareOp::Eq => a == b,
1173 CompareOp::Ne => a != b,
1174 _ => false,
1175 },
1176 (PropertyValue::Null, Literal::Null) => matches!(op, CompareOp::Eq),
1177 _ => false,
1178 }
1179}
1180
1181fn cmp_f64(op: CompareOp, a: f64, b: f64) -> bool {
1182 match op {
1183 CompareOp::Eq => a == b,
1184 CompareOp::Ne => a != b,
1185 CompareOp::Lt => a < b,
1186 CompareOp::Le => a <= b,
1187 CompareOp::Gt => a > b,
1188 CompareOp::Ge => a >= b,
1189 }
1190}
1191
1192fn cmp_ord<T: PartialOrd>(op: CompareOp, a: T, b: T) -> bool {
1193 match op {
1194 CompareOp::Eq => a == b,
1195 CompareOp::Ne => a != b,
1196 CompareOp::Lt => a < b,
1197 CompareOp::Le => a <= b,
1198 CompareOp::Gt => a > b,
1199 CompareOp::Ge => a >= b,
1200 }
1201}
1202
1203pub(crate) fn value_eq(a: &Value, b: &Value) -> bool {
1211 match (a, b) {
1212 (Value::Null, Value::Null) => true,
1213 (Value::Null, _) | (_, Value::Null) => false,
1214 (Value::Property(pa), Value::Property(pb)) => pa == pb,
1215 (Value::Literal(la), Value::Literal(lb)) => la == lb,
1216 (Value::Property(pa), Value::Literal(lb)) => *pa == literal_to_value(lb),
1217 (Value::Literal(la), Value::Property(pb)) => literal_to_value(la) == *pb,
1218 (Value::Node(na), Value::Node(nb)) => na.id == nb.id,
1219 (Value::Edge(ea), Value::Edge(eb)) => ea.id == eb.id,
1220 (Value::List(la), Value::List(lb)) => la.len() == lb.len() && la.iter().zip(lb).all(|(x, y)| value_eq(x, y)),
1221 _ => false,
1222 }
1223}
1224
1225fn call_builtin(name: &str, args: &[Value]) -> Result<Value, QueryError> {
1226 match name.to_ascii_lowercase().as_str() {
1227 "coalesce" => Ok(args
1228 .iter()
1229 .find(|v| !matches!(v, Value::Null))
1230 .cloned()
1231 .unwrap_or(Value::Null)),
1232 "tointeger" => Ok(args.first().map(to_integer).unwrap_or(Value::Null)),
1233 other => Err(QueryError::Parse(format!("unknown function: {other}"))),
1234 }
1235}
1236
1237fn to_integer(v: &Value) -> Value {
1238 let as_str_parse = |s: &str| match s.trim().parse::<i64>() {
1239 Ok(i) => Value::Property(PropertyValue::Int(i)),
1240 Err(_) => Value::Null,
1241 };
1242 match v {
1243 Value::Property(PropertyValue::Int(i)) => Value::Property(PropertyValue::Int(*i)),
1244 Value::Property(PropertyValue::Float(f)) => Value::Property(PropertyValue::Int(*f as i64)),
1245 Value::Property(PropertyValue::String(s)) => as_str_parse(s),
1246 Value::Literal(Literal::Int(i)) => Value::Property(PropertyValue::Int(*i)),
1247 Value::Literal(Literal::Float(f)) => Value::Property(PropertyValue::Int(*f as i64)),
1248 Value::Literal(Literal::String(s)) => as_str_parse(s),
1249 _ => Value::Null,
1250 }
1251}
1252
1253fn apply_order_by(
1258 rows: Vec<Vec<Value>>,
1259 columns: &[String],
1260 order_by: &[(ReturnExpr, SortDir)],
1261) -> Result<Vec<Vec<Value>>, QueryError> {
1262 let mut keyed: Vec<(Vec<Value>, Vec<Value>)> = Vec::with_capacity(rows.len());
1263 for row in rows {
1264 let row_map: HashMap<String, Value> = columns.iter().cloned().zip(row.iter().cloned()).collect();
1265 let keys = order_by
1266 .iter()
1267 .map(|(expr, _)| eval_projected_expr(expr, &row_map))
1268 .collect::<Result<Vec<_>, _>>()?;
1269 keyed.push((keys, row));
1270 }
1271 keyed.sort_by(|(ka, _), (kb, _)| {
1272 for (i, (_, dir)) in order_by.iter().enumerate() {
1273 let ord = compare_with_dir(&ka[i], &kb[i], *dir);
1274 if ord != std::cmp::Ordering::Equal {
1275 return ord;
1276 }
1277 }
1278 std::cmp::Ordering::Equal
1279 });
1280 Ok(keyed.into_iter().map(|(_, row)| row).collect())
1281}
1282
1283fn eval_projected_expr(expr: &ReturnExpr, row: &HashMap<String, Value>) -> Result<Value, QueryError> {
1289 match expr {
1290 ReturnExpr::Var(name) => row
1291 .get(name)
1292 .cloned()
1293 .ok_or_else(|| QueryError::UnboundVariable(name.clone())),
1294 ReturnExpr::Prop(pa) => {
1295 let base = row
1296 .get(&pa.var)
1297 .ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
1298 let pv = match base {
1299 Value::Node(n) => n.props.get(&pa.prop).cloned(),
1300 Value::Edge(e) => e.props.get(&pa.prop).cloned(),
1301 _ => None,
1302 };
1303 Ok(match pv {
1304 Some(PropertyValue::Null) | None => Value::Null,
1305 Some(v) => Value::Property(v),
1306 })
1307 }
1308 ReturnExpr::Lit(lit) => Ok(match lit {
1309 Literal::Null => Value::Null,
1310 other => Value::Literal(other.clone()),
1311 }),
1312 ReturnExpr::Call { name, args, .. } => {
1313 if is_aggregate_name(name) {
1320 return Err(QueryError::Parse(format!(
1321 "aggregate function '{name}' can only be used as a return item's top-level expression"
1322 )));
1323 }
1324 let arg_values = args
1325 .iter()
1326 .map(|a| eval_projected_expr(a, row))
1327 .collect::<Result<Vec<_>, _>>()?;
1328 call_builtin(name, &arg_values)
1329 }
1330 ReturnExpr::CountStar => Err(QueryError::Parse(
1331 "count(*) can only be used as a return item's top-level expression".into(),
1332 )),
1333 ReturnExpr::Case { test, whens, else_ } => {
1334 let test_value = match test {
1335 Some(t) => Some(eval_projected_expr(t, row)?),
1336 None => None,
1337 };
1338 for (when, then) in whens {
1339 let when_value = eval_projected_expr(when, row)?;
1340 let matched = match &test_value {
1341 Some(tv) => value_eq(tv, &when_value),
1342 None => matches!(when_value, Value::Literal(Literal::Bool(true))),
1343 };
1344 if matched {
1345 return eval_projected_expr(then, row);
1346 }
1347 }
1348 match else_ {
1349 Some(e) => eval_projected_expr(e, row),
1350 None => Ok(Value::Null),
1351 }
1352 }
1353 }
1354}
1355
1356fn compare_with_dir(a: &Value, b: &Value, dir: SortDir) -> std::cmp::Ordering {
1359 use std::cmp::Ordering;
1360 let a_null = matches!(a, Value::Null);
1361 let b_null = matches!(b, Value::Null);
1362 match (a_null, b_null) {
1363 (true, true) => return Ordering::Equal,
1364 (true, false) => return Ordering::Greater,
1365 (false, true) => return Ordering::Less,
1366 (false, false) => {}
1367 }
1368 let ord = compare_non_null(a, b);
1369 if dir == SortDir::Desc {
1370 ord.reverse()
1371 } else {
1372 ord
1373 }
1374}
1375
1376fn compare_non_null(a: &Value, b: &Value) -> std::cmp::Ordering {
1377 use std::cmp::Ordering;
1378 let pa = value_to_comparable(a);
1379 let pb = value_to_comparable(b);
1380 match (pa, pb) {
1381 (Some(PropertyValue::Int(x)), Some(PropertyValue::Int(y))) => x.cmp(&y),
1382 (Some(PropertyValue::Int(x)), Some(PropertyValue::Float(y))) => {
1383 (x as f64).partial_cmp(&y).unwrap_or(Ordering::Equal)
1384 }
1385 (Some(PropertyValue::Float(x)), Some(PropertyValue::Int(y))) => {
1386 x.partial_cmp(&(y as f64)).unwrap_or(Ordering::Equal)
1387 }
1388 (Some(PropertyValue::Float(x)), Some(PropertyValue::Float(y))) => x.partial_cmp(&y).unwrap_or(Ordering::Equal),
1389 (Some(PropertyValue::String(x)), Some(PropertyValue::String(y))) => x.cmp(&y),
1390 (Some(PropertyValue::Bool(x)), Some(PropertyValue::Bool(y))) => x.cmp(&y),
1391 _ => Ordering::Equal,
1392 }
1393}
1394
1395fn value_to_comparable(v: &Value) -> Option<PropertyValue> {
1396 match v {
1397 Value::Property(pv) => Some(pv.clone()),
1398 Value::Literal(lit) => Some(literal_to_value(lit)),
1399 _ => None,
1400 }
1401}
1402
1403pub(crate) fn comparable_ordering(a: &Value, b: &Value) -> Option<std::cmp::Ordering> {
1413 use std::cmp::Ordering;
1414 let pa = value_to_comparable(a)?;
1415 let pb = value_to_comparable(b)?;
1416 Some(match (pa, pb) {
1417 (PropertyValue::Int(x), PropertyValue::Int(y)) => x.cmp(&y),
1418 (PropertyValue::Int(x), PropertyValue::Float(y)) => (x as f64).partial_cmp(&y).unwrap_or(Ordering::Equal),
1419 (PropertyValue::Float(x), PropertyValue::Int(y)) => x.partial_cmp(&(y as f64)).unwrap_or(Ordering::Equal),
1420 (PropertyValue::Float(x), PropertyValue::Float(y)) => x.partial_cmp(&y).unwrap_or(Ordering::Equal),
1421 (PropertyValue::String(x), PropertyValue::String(y)) => x.cmp(&y),
1422 (PropertyValue::Bool(x), PropertyValue::Bool(y)) => x.cmp(&y),
1423 _ => return None,
1424 })
1425}