1use std::collections::{BTreeMap, HashMap, HashSet};
2
3use marsdb_graph::{AdjEntry, Direction, EdgeId, GraphStore, NodeId, PropertyValue, WriteTransaction};
4
5use crate::ast::{
6 CompareOp, Expr, Literal, Pattern, PropAccess, QueryPart, RelDirection, ReturnExpr, ReturnItem, SortDir,
7 Statement, Tail, WithClause,
8};
9use crate::error::QueryError;
10use crate::ir::{ExpandDirection, LogicalPlan};
11use crate::planner::{build_match_plan, pattern_all_vars, pattern_new_vars};
12use crate::result::QueryResult;
13use crate::value::Value;
14
15const OPTIONAL_SEED_IDX_KEY: &str = "__seed_idx";
19
20#[derive(Debug, Clone)]
21enum Binding {
22 Node(NodeId),
23 Edge(EdgeId),
24 Value(PropertyValue),
28}
29
30type BindingRow = HashMap<String, Binding>;
31
32const VAR_EXPAND_DEPTH_CAP: u32 = 30;
40
41pub struct Executor<'a> {
42 store: &'a GraphStore,
43}
44
45impl<'a> Executor<'a> {
46 pub fn new(store: &'a GraphStore) -> Self {
47 Self { store }
48 }
49
50 pub fn execute(&self, stmt: &Statement) -> Result<QueryResult, QueryError> {
57 let write_txn = self.store.begin_write()?;
58 let outcome = match stmt {
59 Statement::Create(patterns) => self.execute_create(&write_txn, patterns),
60 Statement::Match {
61 parts,
62 tail,
63 order_by,
64 limit,
65 } => self.execute_match(&write_txn, parts, tail, order_by, *limit),
66 };
67 match outcome {
68 Ok(result) => {
69 GraphStore::commit(write_txn)?;
70 Ok(result)
71 }
72 Err(e) => {
73 let _ = GraphStore::abort(write_txn);
75 Err(e)
76 }
77 }
78 }
79
80 fn execute_create(&self, write_txn: &WriteTransaction, patterns: &[Pattern]) -> Result<QueryResult, QueryError> {
81 for pattern in patterns {
82 let start_labels = pattern_labels(&pattern.start.labels);
83 let start_props = literal_props_to_values(&pattern.start.props);
84 let mut prev_id = GraphStore::create_node_in_txn(write_txn, &start_labels, start_props)?;
85
86 for (rel, node) in &pattern.hops {
87 if rel.hop_range.is_some() {
88 return Err(QueryError::Parse(
89 "CREATE doesn't support variable-length relationship patterns (e.g. [:TYPE*1..3])".into(),
90 ));
91 }
92 let labels = pattern_labels(&node.labels);
93 let props = literal_props_to_values(&node.props);
94 let node_id = GraphStore::create_node_in_txn(write_txn, &labels, props)?;
95
96 let rel_label = rel.rel_type.clone().unwrap_or_else(|| "REL".to_string());
97 let rel_props = literal_props_to_values(&rel.props);
98 let (src, dst) = match rel.direction {
99 RelDirection::Right => (prev_id, node_id),
100 RelDirection::Left => (node_id, prev_id),
101 RelDirection::Either => {
102 return Err(QueryError::Parse(
103 "CREATE requires a directed relationship (-> or <-), not an undirected pattern".into(),
104 ))
105 }
106 };
107 GraphStore::create_edge_in_txn(write_txn, &rel_label, src, dst, rel_props)?;
108 prev_id = node_id;
109 }
110 }
111 Ok(QueryResult {
112 columns: vec![],
113 rows: vec![],
114 })
115 }
116
117 fn execute_match(
118 &self,
119 write_txn: &WriteTransaction,
120 parts: &[QueryPart],
121 tail: &Tail,
122 order_by: &Option<Vec<(ReturnExpr, SortDir)>>,
123 limit: Option<i64>,
124 ) -> Result<QueryResult, QueryError> {
125 let mut carried_vars: HashSet<String> = HashSet::new();
131 let mut current_rows: Vec<BindingRow> = vec![BindingRow::new()];
132 for part in parts {
133 let plan = build_match_plan(&part.pattern, &part.where_clause, &carried_vars)?;
134 current_rows = if part.optional {
135 let new_vars = pattern_new_vars(&part.pattern, &carried_vars);
136 self.eval_optional_part(write_txn, &plan, ¤t_rows, &new_vars)?
137 } else {
138 self.eval_plan(write_txn, &plan, ¤t_rows)?
139 };
140 if let Some(with) = &part.with {
141 current_rows = self.materialize_with(write_txn, with, ¤t_rows)?;
142 if let Some(with_order_by) = &with.order_by {
143 current_rows = self.apply_order_by_bindings(write_txn, current_rows, with_order_by)?;
144 }
145 if let Some(with_limit) = with.limit {
146 current_rows.truncate(with_limit.max(0) as usize);
147 }
148 carried_vars = with.items.iter().enumerate().map(with_item_output_name).collect();
149 } else {
150 carried_vars.extend(pattern_all_vars(&part.pattern));
155 }
156 }
157 if order_by.is_none() {
164 if let Some(count) = limit {
165 current_rows.truncate(count.max(0) as usize);
166 }
167 }
168 let mut result = match tail {
169 Tail::Return(items) => self.materialize_return(write_txn, items, ¤t_rows)?,
170 Tail::Delete(vars) => self.materialize_delete(write_txn, vars, ¤t_rows, false)?,
171 Tail::DetachDelete(vars) => self.materialize_delete(write_txn, vars, ¤t_rows, true)?,
172 Tail::Set(items) => self.materialize_set(write_txn, items, ¤t_rows)?,
173 };
174 if let Some(order_by) = order_by {
175 result.rows = apply_order_by(result.rows, &result.columns, order_by)?;
176 if let Some(count) = limit {
177 result.rows.truncate(count.max(0) as usize);
178 }
179 }
180 Ok(result)
181 }
182
183 fn materialize_with(
190 &self,
191 write_txn: &WriteTransaction,
192 with: &WithClause,
193 rows: &[BindingRow],
194 ) -> Result<Vec<BindingRow>, QueryError> {
195 let mut out = Vec::with_capacity(rows.len());
196 for row in rows {
197 let mut new_row = BindingRow::new();
198 for (i, item) in with.items.iter().enumerate() {
199 let name = with_item_output_name((i, item));
200 let binding = match &item.expr {
201 ReturnExpr::Var(v) => row
202 .get(v)
203 .cloned()
204 .ok_or_else(|| QueryError::UnboundVariable(v.clone()))?,
205 other => {
206 let value = self.eval_return_expr(write_txn, other, row)?;
207 Binding::Value(value_to_property_value(&value))
208 }
209 };
210 new_row.insert(name, binding);
211 }
212 out.push(new_row);
213 }
214 Ok(out)
215 }
216
217 fn apply_order_by_bindings(
222 &self,
223 write_txn: &WriteTransaction,
224 rows: Vec<BindingRow>,
225 order_by: &[(ReturnExpr, SortDir)],
226 ) -> Result<Vec<BindingRow>, QueryError> {
227 let mut keyed: Vec<(Vec<Value>, BindingRow)> = Vec::with_capacity(rows.len());
228 for row in rows {
229 let value_map = self.binding_row_to_value_map(write_txn, &row)?;
230 let keys = order_by
231 .iter()
232 .map(|(expr, _)| eval_projected_expr(expr, &value_map))
233 .collect::<Result<Vec<_>, _>>()?;
234 keyed.push((keys, row));
235 }
236 keyed.sort_by(|(ka, _), (kb, _)| {
237 for (i, (_, dir)) in order_by.iter().enumerate() {
238 let ord = compare_with_dir(&ka[i], &kb[i], *dir);
239 if ord != std::cmp::Ordering::Equal {
240 return ord;
241 }
242 }
243 std::cmp::Ordering::Equal
244 });
245 Ok(keyed.into_iter().map(|(_, row)| row).collect())
246 }
247
248 fn binding_row_to_value_map(
249 &self,
250 write_txn: &WriteTransaction,
251 row: &BindingRow,
252 ) -> Result<HashMap<String, Value>, QueryError> {
253 let mut map = HashMap::with_capacity(row.len());
254 for (k, binding) in row {
255 let value = match binding {
256 Binding::Node(id) => Value::Node(
257 GraphStore::get_node_in_txn(write_txn, *id)?
258 .expect("bound node exists within this statement's transaction"),
259 ),
260 Binding::Edge(id) => Value::Edge(
261 GraphStore::get_edge_in_txn(write_txn, *id)?
262 .expect("bound edge exists within this statement's transaction"),
263 ),
264 Binding::Value(PropertyValue::Null) => Value::Null,
265 Binding::Value(pv) => Value::Property(pv.clone()),
266 };
267 map.insert(k.clone(), value);
268 }
269 Ok(map)
270 }
271
272 fn eval_optional_part(
291 &self,
292 write_txn: &WriteTransaction,
293 plan: &LogicalPlan,
294 outer_rows: &[BindingRow],
295 new_vars: &HashSet<String>,
296 ) -> Result<Vec<BindingRow>, QueryError> {
297 let tagged: Vec<BindingRow> = outer_rows
298 .iter()
299 .enumerate()
300 .map(|(i, row)| {
301 let mut r = row.clone();
302 r.insert(OPTIONAL_SEED_IDX_KEY.to_string(), Binding::Value(PropertyValue::Int(i as i64)));
303 r
304 })
305 .collect();
306 let results = self.eval_plan(write_txn, plan, &tagged)?;
307 let mut by_idx: HashMap<i64, Vec<BindingRow>> = HashMap::new();
308 for mut row in results {
309 let idx = match row.remove(OPTIONAL_SEED_IDX_KEY) {
310 Some(Binding::Value(PropertyValue::Int(i))) => i,
311 other => unreachable!("__seed_idx tagged internally as Binding::Value(Int), got {other:?}"),
312 };
313 by_idx.entry(idx).or_default().push(row);
314 }
315 let mut out = Vec::with_capacity(outer_rows.len());
316 for (i, outer_row) in outer_rows.iter().enumerate() {
317 match by_idx.remove(&(i as i64)) {
318 Some(matches) => out.extend(matches),
319 None => {
320 let mut padded = outer_row.clone();
321 for var in new_vars {
322 padded.insert(var.clone(), Binding::Value(PropertyValue::Null));
323 }
324 out.push(padded);
325 }
326 }
327 }
328 Ok(out)
329 }
330
331 fn eval_plan(
332 &self,
333 write_txn: &WriteTransaction,
334 plan: &LogicalPlan,
335 seed: &[BindingRow],
336 ) -> Result<Vec<BindingRow>, QueryError> {
337 match plan {
338 LogicalPlan::Seed { var } => {
339 debug_assert!(
340 seed.first().is_none_or(|row| row.contains_key(var)),
341 "Seed{{var: {var:?}}} planned for a var not present in the carried-forward rows"
342 );
343 Ok(seed.to_vec())
344 }
345 LogicalPlan::AllNodesScan { var } => self.scan(write_txn, var, None),
346 LogicalPlan::NodeByLabelScan { var, label } => self.scan(write_txn, var, Some(label)),
347 LogicalPlan::Expand {
348 input,
349 from_var,
350 to_var,
351 rel_var,
352 rel_label,
353 direction,
354 } => {
355 let base_rows = self.eval_plan(write_txn, input, seed)?;
356 let mut out = Vec::new();
357 for row in base_rows {
358 let Some(Binding::Node(from_id)) = row.get(from_var).cloned() else {
359 return Err(QueryError::UnboundVariable(from_var.clone()));
360 };
361 let entries = neighbors_for_direction(write_txn, from_id, *direction, rel_label.as_deref())?;
362 for entry in entries {
363 let mut new_row = row.clone();
364 new_row.insert(to_var.clone(), Binding::Node(entry.other));
365 if let Some(rv) = rel_var {
366 new_row.insert(rv.clone(), Binding::Edge(entry.edge_id));
367 }
368 out.push(new_row);
369 }
370 }
371 Ok(out)
372 }
373 LogicalPlan::VarExpand {
374 input,
375 from_var,
376 to_var,
377 rel_label,
378 direction,
379 min_hops,
380 max_hops,
381 } => {
382 let base_rows = self.eval_plan(write_txn, input, seed)?;
383 let mut out = Vec::new();
384 let unbounded = max_hops.is_none();
385 let effective_max = max_hops.unwrap_or(VAR_EXPAND_DEPTH_CAP);
386 for row in base_rows {
387 let Some(Binding::Node(start_id)) = row.get(from_var).cloned() else {
388 return Err(QueryError::UnboundVariable(from_var.clone()));
389 };
390 let mut visited = HashSet::new();
391 visited.insert(start_id);
392 if *min_hops == 0 {
393 let mut new_row = row.clone();
394 new_row.insert(to_var.clone(), Binding::Node(start_id));
395 out.push(new_row);
396 }
397 let mut frontier = vec![start_id];
398 let mut depth = 0u32;
399 while depth < effective_max && !frontier.is_empty() {
400 depth += 1;
401 let mut next_frontier = Vec::new();
402 for node in frontier {
403 let entries = neighbors_for_direction(write_txn, node, *direction, rel_label.as_deref())?;
404 for entry in entries {
405 if visited.insert(entry.other) {
406 next_frontier.push(entry.other);
407 if depth >= *min_hops {
408 let mut new_row = row.clone();
409 new_row.insert(to_var.clone(), Binding::Node(entry.other));
410 out.push(new_row);
411 }
412 }
413 }
414 }
415 frontier = next_frontier;
416 if depth == effective_max && unbounded && !frontier.is_empty() {
417 return Err(QueryError::Parse(format!(
423 "variable-length traversal exceeded the safety depth cap ({VAR_EXPAND_DEPTH_CAP} \
424 hops) — likely a cyclic graph or unexpectedly large fanout; narrow the pattern or \
425 add an explicit upper bound (e.g. *0..10)"
426 )));
427 }
428 }
429 }
430 Ok(out)
431 }
432 LogicalPlan::Filter { input, predicate } => {
433 let rows = self.eval_plan(write_txn, input, seed)?;
434 let mut out = Vec::with_capacity(rows.len());
435 for row in rows {
436 if self.eval_expr(write_txn, predicate, &row)? {
437 out.push(row);
438 }
439 }
440 Ok(out)
441 }
442 }
443 }
444
445 fn scan(&self, write_txn: &WriteTransaction, var: &str, label: Option<&str>) -> Result<Vec<BindingRow>, QueryError> {
446 let nodes = GraphStore::all_nodes_in_txn(write_txn, label)?;
447 Ok(nodes
448 .into_iter()
449 .map(|n| {
450 let mut row = BindingRow::new();
451 row.insert(var.to_string(), Binding::Node(n.id));
452 row
453 })
454 .collect())
455 }
456
457 fn eval_expr(&self, write_txn: &WriteTransaction, expr: &Expr, row: &BindingRow) -> Result<bool, QueryError> {
458 Ok(match expr {
459 Expr::And(l, r) => self.eval_expr(write_txn, l, row)? && self.eval_expr(write_txn, r, row)?,
460 Expr::Or(l, r) => self.eval_expr(write_txn, l, row)? || self.eval_expr(write_txn, r, row)?,
461 Expr::Not(e) => !self.eval_expr(write_txn, e, row)?,
462 Expr::Compare(pa, op, lit) => {
463 let prop_value = self.lookup_prop(write_txn, pa, row)?;
464 compare(&prop_value, *op, lit)
465 }
466 Expr::HasLabel(var, label) => {
467 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
468 let Binding::Node(id) = binding else {
469 return Err(QueryError::UnboundVariable(var.clone()));
470 };
471 let node = GraphStore::get_node_in_txn(write_txn, *id)?;
472 node.is_some_and(|n| n.labels.iter().any(|l| l == label))
473 }
474 Expr::VarEq(a, b) => {
475 let ba = row.get(a).ok_or_else(|| QueryError::UnboundVariable(a.clone()))?;
476 let bb = row.get(b).ok_or_else(|| QueryError::UnboundVariable(b.clone()))?;
477 match (ba, bb) {
478 (Binding::Node(x), Binding::Node(y)) => x == y,
479 (Binding::Edge(x), Binding::Edge(y)) => x == y,
480 _ => false,
488 }
489 }
490 })
491 }
492
493 fn lookup_prop(
494 &self,
495 write_txn: &WriteTransaction,
496 pa: &PropAccess,
497 row: &BindingRow,
498 ) -> Result<Option<PropertyValue>, QueryError> {
499 let binding = row
500 .get(&pa.var)
501 .ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
502 match binding {
503 Binding::Node(id) => {
504 let node = GraphStore::get_node_in_txn(write_txn, *id)?;
505 Ok(node.and_then(|n| n.props.get(&pa.prop).cloned()))
506 }
507 Binding::Edge(id) => {
508 let edge = GraphStore::get_edge_in_txn(write_txn, *id)?;
509 Ok(edge.and_then(|e| e.props.get(&pa.prop).cloned()))
510 }
511 Binding::Value(_) => Ok(None),
516 }
517 }
518
519 fn materialize_return(
520 &self,
521 write_txn: &WriteTransaction,
522 items: &[ReturnItem],
523 rows: &[BindingRow],
524 ) -> Result<QueryResult, QueryError> {
525 let columns = items
526 .iter()
527 .enumerate()
528 .map(|(i, item)| item.alias.clone().unwrap_or_else(|| default_column_name(&item.expr, i)))
529 .collect();
530 let mut out_rows = Vec::with_capacity(rows.len());
531 for row in rows {
532 let mut out_row = Vec::with_capacity(items.len());
533 for item in items {
534 out_row.push(self.eval_return_expr(write_txn, &item.expr, row)?);
535 }
536 out_rows.push(out_row);
537 }
538 Ok(QueryResult {
539 columns,
540 rows: out_rows,
541 })
542 }
543
544 fn eval_return_expr(
545 &self,
546 write_txn: &WriteTransaction,
547 expr: &ReturnExpr,
548 row: &BindingRow,
549 ) -> Result<Value, QueryError> {
550 match expr {
551 ReturnExpr::Var(var) => {
552 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
553 match binding {
554 Binding::Node(id) => {
555 let node = GraphStore::get_node_in_txn(write_txn, *id)?
556 .expect("bound node exists within this statement's transaction");
557 Ok(Value::Node(node))
558 }
559 Binding::Edge(id) => {
560 let edge = GraphStore::get_edge_in_txn(write_txn, *id)?
561 .expect("bound edge exists within this statement's transaction");
562 Ok(Value::Edge(edge))
563 }
564 Binding::Value(PropertyValue::Null) => Ok(Value::Null),
565 Binding::Value(pv) => Ok(Value::Property(pv.clone())),
566 }
567 }
568 ReturnExpr::Prop(pa) => {
569 let value = self.lookup_prop(write_txn, pa, row)?;
570 Ok(match value {
571 Some(PropertyValue::Null) | None => Value::Null,
574 Some(pv) => Value::Property(pv),
575 })
576 }
577 ReturnExpr::Lit(lit) => Ok(match lit {
578 Literal::Null => Value::Null,
579 other => Value::Literal(other.clone()),
580 }),
581 ReturnExpr::Call(name, args) => {
582 let arg_values = args
583 .iter()
584 .map(|a| self.eval_return_expr(write_txn, a, row))
585 .collect::<Result<Vec<_>, _>>()?;
586 call_builtin(name, &arg_values)
587 }
588 ReturnExpr::Case { test, whens, else_ } => {
589 let test_value = match test {
590 Some(t) => Some(self.eval_return_expr(write_txn, t, row)?),
591 None => None,
592 };
593 for (when, then) in whens {
594 let when_value = self.eval_return_expr(write_txn, when, row)?;
595 let matched = match &test_value {
601 Some(tv) => value_eq(tv, &when_value),
602 None => matches!(when_value, Value::Literal(Literal::Bool(true))),
603 };
604 if matched {
605 return self.eval_return_expr(write_txn, then, row);
606 }
607 }
608 match else_ {
609 Some(e) => self.eval_return_expr(write_txn, e, row),
610 None => Ok(Value::Null),
611 }
612 }
613 }
614 }
615
616 fn materialize_delete(
617 &self,
618 write_txn: &WriteTransaction,
619 vars: &[String],
620 rows: &[BindingRow],
621 detach: bool,
622 ) -> Result<QueryResult, QueryError> {
623 let mut deleted_nodes = HashSet::new();
624 let mut deleted_edges = HashSet::new();
625 for row in rows {
626 for var in vars {
627 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
628 match binding {
629 Binding::Node(id) => {
630 if deleted_nodes.insert(*id) {
631 GraphStore::delete_node_in_txn(write_txn, *id, detach)?;
632 }
633 }
634 Binding::Edge(id) => {
635 if deleted_edges.insert(*id) {
636 GraphStore::delete_edge_in_txn(write_txn, *id)?;
637 }
638 }
639 Binding::Value(_) => {
640 return Err(QueryError::UnboundVariable(format!(
641 "'{var}' is a WITH-projected scalar, not a node/edge — DELETE needs a graph binding"
642 )))
643 }
644 }
645 }
646 }
647 Ok(QueryResult {
648 columns: vec![],
649 rows: vec![],
650 })
651 }
652
653 fn materialize_set(
654 &self,
655 write_txn: &WriteTransaction,
656 items: &[(PropAccess, Literal)],
657 rows: &[BindingRow],
658 ) -> Result<QueryResult, QueryError> {
659 for row in rows {
660 for (pa, lit) in items {
661 let binding = row.get(&pa.var).ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
662 let value = literal_to_value(lit);
663 match binding {
664 Binding::Node(id) => {
665 GraphStore::set_node_prop_in_txn(write_txn, *id, &pa.prop, value)?;
666 }
667 Binding::Edge(id) => {
668 GraphStore::set_edge_prop_in_txn(write_txn, *id, &pa.prop, value)?;
669 }
670 Binding::Value(_) => {
671 return Err(QueryError::UnboundVariable(format!(
672 "'{}' is a WITH-projected scalar, not a node/edge — SET needs a graph binding",
673 pa.var
674 )))
675 }
676 }
677 }
678 }
679 Ok(QueryResult {
680 columns: vec![],
681 rows: vec![],
682 })
683 }
684}
685
686fn default_column_name(expr: &ReturnExpr, idx: usize) -> String {
687 match expr {
688 ReturnExpr::Var(v) => v.clone(),
689 ReturnExpr::Prop(pa) => format!("{}.{}", pa.var, pa.prop),
690 ReturnExpr::Lit(_) => format!("col{idx}"),
691 ReturnExpr::Call(name, _) => format!("{name}(...)"),
692 ReturnExpr::Case { .. } => format!("case{idx}"),
693 }
694}
695
696fn with_item_output_name((i, item): (usize, &ReturnItem)) -> String {
699 item.alias.clone().unwrap_or_else(|| default_column_name(&item.expr, i))
700}
701
702fn value_to_property_value(v: &Value) -> PropertyValue {
710 match v {
711 Value::Null => PropertyValue::Null,
712 Value::Property(pv) => pv.clone(),
713 Value::Literal(lit) => literal_to_value(lit),
714 Value::Node(_) | Value::Edge(_) => PropertyValue::Null,
715 }
716}
717
718fn literal_to_value(lit: &Literal) -> PropertyValue {
719 match lit {
720 Literal::Int(i) => PropertyValue::Int(*i),
721 Literal::Float(f) => PropertyValue::Float(*f),
722 Literal::String(s) => PropertyValue::String(s.clone()),
723 Literal::Bool(b) => PropertyValue::Bool(*b),
724 Literal::Null => PropertyValue::Null,
725 Literal::Param(name) => {
726 unreachable!("param ${name} must be substituted before execution — see params::substitute_params")
727 }
728 }
729}
730
731fn literal_props_to_values(props: &[(String, Literal)]) -> BTreeMap<String, PropertyValue> {
732 props.iter().map(|(k, v)| (k.clone(), literal_to_value(v))).collect()
733}
734
735fn pattern_labels(labels: &[String]) -> Vec<&str> {
736 if labels.is_empty() {
737 vec!["Node"]
738 } else {
739 labels.iter().map(|s| s.as_str()).collect()
740 }
741}
742
743fn neighbors_for_direction(
747 write_txn: &WriteTransaction,
748 node: NodeId,
749 direction: ExpandDirection,
750 rel_label: Option<&str>,
751) -> Result<Vec<AdjEntry>, QueryError> {
752 Ok(match direction {
753 ExpandDirection::Out => GraphStore::neighbors_in_txn(write_txn, node, Direction::Out, rel_label)?,
754 ExpandDirection::In => GraphStore::neighbors_in_txn(write_txn, node, Direction::In, rel_label)?,
755 ExpandDirection::Either => {
756 let mut out = GraphStore::neighbors_in_txn(write_txn, node, Direction::Out, rel_label)?;
757 let inbound = GraphStore::neighbors_in_txn(write_txn, node, Direction::In, rel_label)?;
758 let seen: HashSet<EdgeId> = out.iter().map(|e| e.edge_id).collect();
759 out.extend(inbound.into_iter().filter(|e| !seen.contains(&e.edge_id)));
760 out
761 }
762 })
763}
764
765fn compare(prop: &Option<PropertyValue>, op: CompareOp, lit: &Literal) -> bool {
766 let Some(prop) = prop else { return false };
767 match (prop, lit) {
768 (PropertyValue::Int(a), Literal::Int(b)) => cmp_f64(op, *a as f64, *b as f64),
769 (PropertyValue::Int(a), Literal::Float(b)) => cmp_f64(op, *a as f64, *b),
770 (PropertyValue::Float(a), Literal::Float(b)) => cmp_f64(op, *a, *b),
771 (PropertyValue::Float(a), Literal::Int(b)) => cmp_f64(op, *a, *b as f64),
772 (PropertyValue::String(a), Literal::String(b)) => cmp_ord(op, a.as_str(), b.as_str()),
773 (PropertyValue::Bool(a), Literal::Bool(b)) => match op {
774 CompareOp::Eq => a == b,
775 CompareOp::Ne => a != b,
776 _ => false,
777 },
778 (PropertyValue::Null, Literal::Null) => matches!(op, CompareOp::Eq),
779 _ => false,
780 }
781}
782
783fn cmp_f64(op: CompareOp, a: f64, b: f64) -> bool {
784 match op {
785 CompareOp::Eq => a == b,
786 CompareOp::Ne => a != b,
787 CompareOp::Lt => a < b,
788 CompareOp::Le => a <= b,
789 CompareOp::Gt => a > b,
790 CompareOp::Ge => a >= b,
791 }
792}
793
794fn cmp_ord<T: PartialOrd>(op: CompareOp, a: T, b: T) -> bool {
795 match op {
796 CompareOp::Eq => a == b,
797 CompareOp::Ne => a != b,
798 CompareOp::Lt => a < b,
799 CompareOp::Le => a <= b,
800 CompareOp::Gt => a > b,
801 CompareOp::Ge => a >= b,
802 }
803}
804
805fn value_eq(a: &Value, b: &Value) -> bool {
809 match (a, b) {
810 (Value::Null, Value::Null) => true,
811 (Value::Null, _) | (_, Value::Null) => false,
812 (Value::Property(pa), Value::Property(pb)) => pa == pb,
813 (Value::Literal(la), Value::Literal(lb)) => la == lb,
814 (Value::Property(pa), Value::Literal(lb)) => *pa == literal_to_value(lb),
815 (Value::Literal(la), Value::Property(pb)) => literal_to_value(la) == *pb,
816 _ => false,
817 }
818}
819
820fn call_builtin(name: &str, args: &[Value]) -> Result<Value, QueryError> {
821 match name.to_ascii_lowercase().as_str() {
822 "coalesce" => Ok(args
823 .iter()
824 .find(|v| !matches!(v, Value::Null))
825 .cloned()
826 .unwrap_or(Value::Null)),
827 "tointeger" => Ok(args.first().map(to_integer).unwrap_or(Value::Null)),
828 other => Err(QueryError::Parse(format!("unknown function: {other}"))),
829 }
830}
831
832fn to_integer(v: &Value) -> Value {
833 let as_str_parse = |s: &str| match s.trim().parse::<i64>() {
834 Ok(i) => Value::Property(PropertyValue::Int(i)),
835 Err(_) => Value::Null,
836 };
837 match v {
838 Value::Property(PropertyValue::Int(i)) => Value::Property(PropertyValue::Int(*i)),
839 Value::Property(PropertyValue::Float(f)) => Value::Property(PropertyValue::Int(*f as i64)),
840 Value::Property(PropertyValue::String(s)) => as_str_parse(s),
841 Value::Literal(Literal::Int(i)) => Value::Property(PropertyValue::Int(*i)),
842 Value::Literal(Literal::Float(f)) => Value::Property(PropertyValue::Int(*f as i64)),
843 Value::Literal(Literal::String(s)) => as_str_parse(s),
844 _ => Value::Null,
845 }
846}
847
848fn apply_order_by(
853 rows: Vec<Vec<Value>>,
854 columns: &[String],
855 order_by: &[(ReturnExpr, SortDir)],
856) -> Result<Vec<Vec<Value>>, QueryError> {
857 let mut keyed: Vec<(Vec<Value>, Vec<Value>)> = Vec::with_capacity(rows.len());
858 for row in rows {
859 let row_map: HashMap<String, Value> = columns.iter().cloned().zip(row.iter().cloned()).collect();
860 let keys = order_by
861 .iter()
862 .map(|(expr, _)| eval_projected_expr(expr, &row_map))
863 .collect::<Result<Vec<_>, _>>()?;
864 keyed.push((keys, row));
865 }
866 keyed.sort_by(|(ka, _), (kb, _)| {
867 for (i, (_, dir)) in order_by.iter().enumerate() {
868 let ord = compare_with_dir(&ka[i], &kb[i], *dir);
869 if ord != std::cmp::Ordering::Equal {
870 return ord;
871 }
872 }
873 std::cmp::Ordering::Equal
874 });
875 Ok(keyed.into_iter().map(|(_, row)| row).collect())
876}
877
878fn eval_projected_expr(expr: &ReturnExpr, row: &HashMap<String, Value>) -> Result<Value, QueryError> {
884 match expr {
885 ReturnExpr::Var(name) => row
886 .get(name)
887 .cloned()
888 .ok_or_else(|| QueryError::UnboundVariable(name.clone())),
889 ReturnExpr::Prop(pa) => {
890 let base = row
891 .get(&pa.var)
892 .ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
893 let pv = match base {
894 Value::Node(n) => n.props.get(&pa.prop).cloned(),
895 Value::Edge(e) => e.props.get(&pa.prop).cloned(),
896 _ => None,
897 };
898 Ok(match pv {
899 Some(PropertyValue::Null) | None => Value::Null,
900 Some(v) => Value::Property(v),
901 })
902 }
903 ReturnExpr::Lit(lit) => Ok(match lit {
904 Literal::Null => Value::Null,
905 other => Value::Literal(other.clone()),
906 }),
907 ReturnExpr::Call(name, args) => {
908 let arg_values = args
909 .iter()
910 .map(|a| eval_projected_expr(a, row))
911 .collect::<Result<Vec<_>, _>>()?;
912 call_builtin(name, &arg_values)
913 }
914 ReturnExpr::Case { test, whens, else_ } => {
915 let test_value = match test {
916 Some(t) => Some(eval_projected_expr(t, row)?),
917 None => None,
918 };
919 for (when, then) in whens {
920 let when_value = eval_projected_expr(when, row)?;
921 let matched = match &test_value {
922 Some(tv) => value_eq(tv, &when_value),
923 None => matches!(when_value, Value::Literal(Literal::Bool(true))),
924 };
925 if matched {
926 return eval_projected_expr(then, row);
927 }
928 }
929 match else_ {
930 Some(e) => eval_projected_expr(e, row),
931 None => Ok(Value::Null),
932 }
933 }
934 }
935}
936
937fn compare_with_dir(a: &Value, b: &Value, dir: SortDir) -> std::cmp::Ordering {
940 use std::cmp::Ordering;
941 let a_null = matches!(a, Value::Null);
942 let b_null = matches!(b, Value::Null);
943 match (a_null, b_null) {
944 (true, true) => return Ordering::Equal,
945 (true, false) => return Ordering::Greater,
946 (false, true) => return Ordering::Less,
947 (false, false) => {}
948 }
949 let ord = compare_non_null(a, b);
950 if dir == SortDir::Desc {
951 ord.reverse()
952 } else {
953 ord
954 }
955}
956
957fn compare_non_null(a: &Value, b: &Value) -> std::cmp::Ordering {
958 use std::cmp::Ordering;
959 let pa = value_to_comparable(a);
960 let pb = value_to_comparable(b);
961 match (pa, pb) {
962 (Some(PropertyValue::Int(x)), Some(PropertyValue::Int(y))) => x.cmp(&y),
963 (Some(PropertyValue::Int(x)), Some(PropertyValue::Float(y))) => {
964 (x as f64).partial_cmp(&y).unwrap_or(Ordering::Equal)
965 }
966 (Some(PropertyValue::Float(x)), Some(PropertyValue::Int(y))) => {
967 x.partial_cmp(&(y as f64)).unwrap_or(Ordering::Equal)
968 }
969 (Some(PropertyValue::Float(x)), Some(PropertyValue::Float(y))) => x.partial_cmp(&y).unwrap_or(Ordering::Equal),
970 (Some(PropertyValue::String(x)), Some(PropertyValue::String(y))) => x.cmp(&y),
971 (Some(PropertyValue::Bool(x)), Some(PropertyValue::Bool(y))) => x.cmp(&y),
972 _ => Ordering::Equal,
973 }
974}
975
976fn value_to_comparable(v: &Value) -> Option<PropertyValue> {
977 match v {
978 Value::Property(pv) => Some(pv.clone()),
979 Value::Literal(lit) => Some(literal_to_value(lit)),
980 _ => None,
981 }
982}