1use crate::cypher::ast::{
4 AggArg, AggFunc, Expr, LimitSkip, Operand, OrderItem, OrderTarget, RetItem, RetVal, UnwindExpr,
5};
6use crate::cypher::plan::PlanOp;
7use crate::cypher::RelDir;
8use crate::filter::eval_cmp;
9use crate::result::ResultSet;
10use crate::traverse::{expand, Dir, EdgeRef};
11use crate::value_ops::{cmp_optional, values_equal};
12use crate::view::GraphView;
13use core_storage::{Value, ValueKey};
14use std::collections::{BTreeMap, BTreeSet, HashMap};
15
16#[cfg(test)]
20static FUSED_SCAN_FIRES: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
21
22#[cfg(test)]
26static SCAN_KEY_FIRES: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
27
28pub struct Params<'a>(pub &'a BTreeMap<String, Value>);
31
32#[derive(Clone, Debug)]
33enum Cell {
34 Node(u32),
35 Rel(EdgeRef),
36 Path(u8),
39 Scalar(Value),
41}
42
43type Row = Vec<Option<Cell>>;
46
47struct VarTable {
48 names: Vec<String>,
49}
50
51impl VarTable {
52 fn intern(&mut self, name: &str) -> usize {
53 if let Some(i) = self.names.iter().position(|n| n == name) {
54 return i;
55 }
56 self.names.push(name.to_string());
57 self.names.len() - 1
58 }
59
60 fn slot(&self, name: &str) -> Option<usize> {
61 self.names.iter().position(|n| n == name)
62 }
63}
64
65struct Projected {
66 columns: Vec<String>,
67 rows: Vec<Vec<Option<Value>>>,
68}
69
70const MAX_INTERMEDIATE_ROWS: usize = 1_000_000;
73
74const MAX_GROUPS: usize = 1_000_000;
76
77type GroupKey = Vec<Option<ValueKey>>;
80type GroupEntry = (Vec<Option<Value>>, Vec<AggAcc>);
85
86#[cfg(test)]
87thread_local! {
88 static TEST_MAX_INTERMEDIATE_ROWS: std::cell::Cell<Option<usize>> =
89 const { std::cell::Cell::new(None) };
90 static TEST_EXPAND_PRODUCED: std::cell::Cell<Option<usize>> =
93 const { std::cell::Cell::new(None) };
94 static TEST_MAX_GROUPS: std::cell::Cell<Option<usize>> =
96 const { std::cell::Cell::new(None) };
97}
98
99fn max_intermediate_rows() -> usize {
100 #[cfg(test)]
101 {
102 TEST_MAX_INTERMEDIATE_ROWS
103 .with(|c| c.get())
104 .unwrap_or(MAX_INTERMEDIATE_ROWS)
105 }
106 #[cfg(not(test))]
107 {
108 MAX_INTERMEDIATE_ROWS
109 }
110}
111
112fn max_groups() -> usize {
113 #[cfg(test)]
114 {
115 TEST_MAX_GROUPS.with(|c| c.get()).unwrap_or(MAX_GROUPS)
116 }
117 #[cfg(not(test))]
118 {
119 MAX_GROUPS
120 }
121}
122
123#[cfg(test)]
126pub(crate) fn with_max_groups<R>(cap: usize, f: impl FnOnce() -> R) -> R {
127 TEST_MAX_GROUPS.with(|c| {
128 let prev = c.replace(Some(cap));
129 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
130 c.set(prev);
131 match result {
132 Ok(v) => v,
133 Err(p) => std::panic::resume_unwind(p),
134 }
135 })
136}
137
138#[cfg(test)]
142pub(crate) fn with_max_intermediate_rows<R>(cap: usize, f: impl FnOnce() -> R) -> R {
143 TEST_MAX_INTERMEDIATE_ROWS.with(|c| {
144 let prev = c.replace(Some(cap));
145 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
146 c.set(prev);
147 match result {
148 Ok(v) => v,
149 Err(p) => std::panic::resume_unwind(p),
150 }
151 })
152}
153
154#[cfg(test)]
156fn record_expand_row() {
157 TEST_EXPAND_PRODUCED.with(|c| {
158 if let Some(prev) = c.get() {
159 c.set(Some(prev + 1));
160 }
161 });
162}
163
164#[cfg(test)]
170pub(crate) fn with_expand_counter<R>(f: impl FnOnce() -> R) -> (R, usize) {
171 TEST_EXPAND_PRODUCED.with(|c| c.set(Some(0)));
172 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
173 let count = TEST_EXPAND_PRODUCED.with(|c| c.get().unwrap_or(0));
174 TEST_EXPAND_PRODUCED.with(|c| c.set(None));
175 match result {
176 Ok(v) => (v, count),
177 Err(p) => std::panic::resume_unwind(p),
178 }
179}
180
181fn row_cap_err(cap: usize) -> String {
182 format!(
183 "intermediate result exceeds {cap} rows; add a LIMIT or constrain patterns with shared variables"
184 )
185}
186
187fn group_cap_err() -> String {
188 format!(
189 "group count exceeds {} distinct keys; add a WHERE clause or constrain the grouping key",
190 max_groups()
191 )
192}
193
194fn group_key_normalize(v: &Value) -> Option<ValueKey> {
205 match v {
206 Value::Int(n) => Some(ValueKey::FloatBits((*n as f64).to_bits())),
207 Value::Float(f) => Some(ValueKey::FloatBits(f.to_bits())),
208 _ => ValueKey::from_value(v),
209 }
210}
211
212pub fn execute(view: &GraphView, plan: &[PlanOp], params: &Params) -> Result<ResultSet, String> {
224 use crate::cypher::plan::row_bound;
225 execute_inner(view, plan, params, row_bound(plan))
226}
227
228#[cfg(test)]
233pub(crate) fn execute_unbounded(
234 view: &GraphView,
235 plan: &[PlanOp],
236 params: &Params,
237) -> Result<ResultSet, String> {
238 execute_inner(view, plan, params, None)
239}
240
241fn execute_inner(
242 view: &GraphView,
243 plan: &[PlanOp],
244 params: &Params,
245 row_bound: Option<usize>,
246) -> Result<ResultSet, String> {
247 check_params(plan, params)?;
248
249 let has_var_expand = plan
255 .iter()
256 .any(|op| matches!(op, PlanOp::VarExpand { .. } | PlanOp::ShortestPath { .. }));
257
258 let is_pipeline = plan.iter().any(|op| {
264 matches!(
265 op,
266 PlanOp::With { .. } | PlanOp::Unwind { .. } | PlanOp::LeftOuterApply { .. }
267 )
268 }) || {
269 let mut saw_gagg = false;
270 plan.iter().any(|op| {
271 if matches!(op, PlanOp::GroupAggregate { .. }) {
272 saw_gagg = true;
273 false
274 } else {
275 saw_gagg && matches!(op, PlanOp::Filter { .. })
276 }
277 })
278 };
279
280 if !has_var_expand
284 && !is_pipeline
285 && plan
286 .iter()
287 .any(|op| matches!(op, PlanOp::GroupAggregate { .. }))
288 {
289 return execute_group_aggregate(view, plan, params);
290 }
291
292 if !has_var_expand
296 && !is_pipeline
297 && plan.iter().any(|op| matches!(op, PlanOp::Aggregate { .. }))
298 {
299 return execute_aggregate(view, plan, params);
300 }
301
302 if let Some(bound) = row_bound {
306 return execute_pull(view, plan, params, bound);
307 }
308
309 let vars = collect_vars(plan);
311 let mut rows: Vec<Row> = vec![vec![None; vars.names.len()]];
312 let mut projected: Option<Projected> = None;
313 let mut pipeline_group_columns: Option<Vec<String>> = None;
316
317 for op in plan {
318 match op {
319 PlanOp::ScanLabel { var, label } => {
320 rows = scan_label(view, &vars, &rows, var, label.as_deref())?;
321 }
322 PlanOp::ScanKey { var, key, label } => {
323 rows = scan_key(view, &vars, &rows, var, key, label.as_deref(), params)?;
324 }
325 PlanOp::LookupProps { var, props } => {
326 rows = retain_node(view, &vars, &rows, var, None, props, params)?;
327 }
328 PlanOp::JoinBound { var, label, props } => {
329 rows = retain_node(view, &vars, &rows, var, label.as_deref(), props, params)?;
330 }
331 PlanOp::Expand { .. } => {
332 rows = exec_expand(view, &vars, &rows, op, params)?;
333 }
334 PlanOp::VarExpand {
335 from,
336 rel_var,
337 etype,
338 dir,
339 to,
340 min,
341 max,
342 } => {
343 rows = exec_var_expand(
344 view, &vars, &rows, from, rel_var, etype, *dir, to, *min, *max,
345 )?;
346 }
347 PlanOp::ShortestPath {
348 from,
349 rel_var,
350 etype,
351 dir,
352 to,
353 max_hops,
354 } => {
355 rows = exec_shortest_path(
356 view, &vars, &rows, from, rel_var, etype, *dir, to, *max_hops,
357 )?;
358 }
359 PlanOp::Filter { expr } => {
360 rows = exec_filter(view, &vars, &rows, expr, params)?;
361 }
362 PlanOp::Project { items } => {
363 projected = Some(exec_project(view, &vars, &rows, items, params)?);
364 }
365 PlanOp::Distinct => {
366 if let Some(table) = projected.as_mut() {
367 exec_distinct(table)?;
368 } else {
369 return Err("DISTINCT requires a Project".into());
370 }
371 }
372 PlanOp::OrderBy { items } => {
373 if let Some(table) = projected.as_mut() {
374 exec_order_by(table, items)?;
375 } else {
376 exec_order_by_rows(&vars, &mut rows, items, view);
378 }
379 }
380 PlanOp::Skip(ls) => {
381 let n = resolve_ls(ls, params)?;
382 if let Some(table) = projected.as_mut() {
383 apply_skip(&mut table.rows, n);
384 } else {
385 apply_skip(&mut rows, n);
386 }
387 }
388 PlanOp::Limit(ls) => {
389 let n = resolve_ls(ls, params)?;
390 if let Some(table) = projected.as_mut() {
391 apply_limit(&mut table.rows, n);
392 } else {
393 apply_limit(&mut rows, n);
394 }
395 }
396 PlanOp::GroupAggregate { keys, aggs } => {
401 let mut grp_groups: HashMap<GroupKey, GroupEntry> = HashMap::new();
402 let mut grp_key_order: Vec<GroupKey> = Vec::new();
403 for row in &rows {
404 let mut gk: GroupKey = Vec::with_capacity(keys.len());
405 let mut display_vals: Vec<Option<Value>> = Vec::with_capacity(keys.len());
406 for (_, item) in keys {
407 let val = project_item(view, &vars, row, item, params)?;
408 gk.push(val.as_ref().and_then(group_key_normalize));
409 display_vals.push(val);
410 }
411 if !grp_groups.contains_key(&gk) {
412 if grp_groups.len() >= max_groups() {
413 return Err(group_cap_err());
414 }
415 grp_key_order.push(gk.clone());
416 grp_groups.insert(
417 gk.clone(),
418 (
419 display_vals,
420 aggs.iter().map(|(f, _, _)| AggAcc::new(f)).collect(),
421 ),
422 );
423 }
424 let (_, accs) = grp_groups.get_mut(&gk).unwrap();
425 for (acc, (func, arg, _)) in accs.iter_mut().zip(aggs.iter()) {
426 update_acc(view, &vars, row, func, arg, acc)?;
427 }
428 }
429 if keys.is_empty() && grp_key_order.is_empty() {
432 let empty_key: GroupKey = vec![];
433 grp_key_order.push(empty_key.clone());
434 grp_groups.insert(
435 empty_key,
436 (
437 vec![],
438 aggs.iter().map(|(f, _, _)| AggAcc::new(f)).collect(),
439 ),
440 );
441 }
442 if is_pipeline {
443 rows = group_result_to_rows(keys, aggs, grp_key_order, &mut grp_groups, &vars);
446 projected = None;
447 let mut cols: Vec<String> = keys.iter().map(|(c, _)| c.clone()).collect();
450 cols.extend(aggs.iter().map(|(_, _, c)| c.clone()));
451 pipeline_group_columns = Some(cols);
452 } else {
453 projected = Some(build_group_projected(
454 keys,
455 aggs,
456 grp_key_order,
457 &mut grp_groups,
458 ));
459 pipeline_group_columns = None;
460 }
461 }
462 PlanOp::With {
465 items,
466 where_expr,
467 order_by,
468 skip,
469 limit,
470 } => {
471 let row_len = vars.names.len();
475 let mut new_rows: Vec<Row> = Vec::with_capacity(rows.len());
476 for row in &rows {
477 let mut new_row: Row = vec![None; row_len];
478 for item in items {
479 let col = column_name(item);
480 let Some(dst_slot) = vars.slot(&col) else {
481 continue;
482 };
483 match &item.value {
484 RetVal::Var(v) => {
485 if let Some(src_slot) = vars.slot(v) {
487 new_row[dst_slot] = row.get(src_slot).cloned().flatten();
488 }
489 }
490 RetVal::Prop { var, field } => {
491 let val = resolve_prop(view, &vars, row, var, field)?;
492 new_row[dst_slot] = val.map(Cell::Scalar);
493 }
494 RetVal::Agg { .. } => {} RetVal::FuncCall { name, args } => {
496 let val = eval_func(name, args, view, &vars, row, params)?;
497 new_row[dst_slot] = val.map(Cell::Scalar);
498 }
499 RetVal::ScalarExpr(op) => {
500 let val = resolve_operand(view, &vars, row, op, params)?;
501 new_row[dst_slot] = val.map(Cell::Scalar);
502 }
503 }
504 }
505 new_rows.push(new_row);
506 }
507 rows = new_rows;
508 if let Some(expr) = where_expr {
510 rows = exec_filter(view, &vars, &rows, expr, params)?;
511 }
512 if !order_by.is_empty() {
514 exec_order_by_rows(&vars, &mut rows, order_by, view);
515 }
516 if let Some(ls) = skip {
517 let n = resolve_ls(ls, params)?;
518 apply_skip(&mut rows, n);
519 }
520 if let Some(ls) = limit {
521 let n = resolve_ls(ls, params)?;
522 apply_limit(&mut rows, n);
523 }
524 }
525 PlanOp::Unwind { expr, alias } => {
527 let alias_slot = vars
528 .slot(alias)
529 .ok_or_else(|| format!("UNWIND alias `{alias}` not in VarTable"))?;
530 let cap = max_intermediate_rows();
531 let mut new_rows: Vec<Row> = Vec::new();
532 for row in &rows {
533 let list_val: Option<Value> = match expr {
534 UnwindExpr::Lit(vals) => Some(Value::List(vals.clone())),
535 UnwindExpr::Prop { var, field } => {
536 resolve_prop(view, &vars, row, var, field)?
537 }
538 UnwindExpr::Var(name) => {
539 let slot = vars
540 .slot(name)
541 .ok_or_else(|| format!("UNWIND variable `{name}` is not bound"))?;
542 match row.get(slot).and_then(|c| c.as_ref()) {
543 Some(Cell::Scalar(v)) => Some(v.clone()),
544 Some(Cell::Node(_) | Cell::Rel(_) | Cell::Path(_)) => {
545 return Err(format!(
546 "UNWIND requires a list; `{name}` is bound to a graph element"
547 ));
548 }
549 None => None,
550 }
551 }
552 };
553 match list_val {
554 None => {} Some(Value::List(items_list)) => {
556 for item_val in items_list {
558 if new_rows.len() >= cap {
559 return Err(row_cap_err(cap));
560 }
561 let mut new_row = row.clone();
562 new_row[alias_slot] = Some(Cell::Scalar(item_val));
563 new_rows.push(new_row);
564 }
565 }
566 Some(other) => {
567 let type_name = match &other {
568 Value::Int(_) => "Int",
569 Value::Float(_) => "Float",
570 Value::Str(_) => "Str",
571 Value::Bool(_) => "Bool",
572 Value::List(_) => unreachable!(),
573 Value::Map(_) => "Map",
574 };
575 return Err(format!(
576 "UNWIND requires a list; got {type_name} value for `{alias}`"
577 ));
578 }
579 }
580 }
581 rows = new_rows;
582 }
583 PlanOp::Aggregate { func, arg, column } => {
588 let ctx = AggStreamCtx {
589 view,
590 vars: &vars,
591 params,
592 func,
593 arg,
594 };
595 let mut acc = AggAcc::new(func);
596 for row in &rows {
597 agg_stream(&ctx, &[], row, &mut acc)?;
599 }
600 let value = acc.finish();
601 let mut rs = ResultSet::new(vec![column.clone()]);
602 rs.push_row(vec![value]);
603 return Ok(rs);
604 }
605 PlanOp::LeftOuterApply {
611 inner,
612 optional_vars,
613 } => {
614 let cap = max_intermediate_rows();
615 let mut new_rows: Vec<Row> = Vec::new();
616 for outer_row in &rows {
617 let inner_seed: Vec<Row> = vec![outer_row.clone()];
619 let inner_result =
622 exec_left_outer_inner(view, &vars, inner_seed, inner, params)?;
623 if inner_result.is_empty() {
624 let mut null_row = outer_row.clone();
626 for v in optional_vars {
627 if let Some(slot) = vars.slot(v) {
628 null_row[slot] = None;
629 }
630 }
631 if new_rows.len() >= cap {
632 return Err(row_cap_err(cap));
633 }
634 new_rows.push(null_row);
635 } else {
636 for r in inner_result {
637 if new_rows.len() >= cap {
638 return Err(row_cap_err(cap));
639 }
640 new_rows.push(r);
641 }
642 }
643 }
644 rows = new_rows;
645 }
646 }
647 }
648
649 Ok(match projected {
650 Some(table) => finish(table),
651 None => {
652 if let Some(cols) = pipeline_group_columns {
655 let mut rs = ResultSet::new(cols.clone());
656 for row in rows {
657 let vals: Vec<Option<Value>> = cols
658 .iter()
659 .map(|col| {
660 vars.slot(col)
661 .and_then(|s| row.get(s))
662 .and_then(|c| c.as_ref())
663 .and_then(|c| match c {
664 Cell::Scalar(v) => Some(v.clone()),
665 Cell::Node(id) => {
666 view.ids.key_of(*id).map(|k| Value::Str(k.to_owned()))
667 }
668 Cell::Path(h) => Some(Value::Int(*h as i64)),
669 Cell::Rel(_) => None,
670 })
671 })
672 .collect();
673 rs.push_row(vals);
674 }
675 rs
676 } else {
677 ResultSet::new(vec![])
678 }
679 }
680 })
681}
682
683fn exec_left_outer_inner(
692 view: &GraphView,
693 vars: &VarTable,
694 mut rows: Vec<Row>,
695 inner: &[PlanOp],
696 params: &Params,
697) -> Result<Vec<Row>, String> {
698 for op in inner {
699 match op {
700 PlanOp::ScanLabel { var, label } => {
701 rows = scan_label(view, vars, &rows, var, label.as_deref())?;
702 }
703 PlanOp::ScanKey { var, key, label } => {
704 rows = scan_key(view, vars, &rows, var, key, label.as_deref(), params)?;
705 }
706 PlanOp::LookupProps { var, props } => {
707 rows = retain_node(view, vars, &rows, var, None, props, params)?;
708 }
709 PlanOp::JoinBound { var, label, props } => {
710 rows = retain_node(view, vars, &rows, var, label.as_deref(), props, params)?;
711 }
712 PlanOp::Expand { .. } => {
713 rows = exec_expand(view, vars, &rows, op, params)?;
714 }
715 PlanOp::Filter { expr } => {
716 rows = exec_filter(view, vars, &rows, expr, params)?;
717 }
718 other => {
719 return Err(format!(
720 "unsupported op inside OPTIONAL MATCH inner plan: {other:?}"
721 ));
722 }
723 }
724 }
725 Ok(rows)
726}
727
728fn finish(table: Projected) -> ResultSet {
729 let mut rs = ResultSet::new(table.columns);
730 for row in table.rows {
731 rs.push_row(row);
732 }
733 rs
734}
735
736fn check_params(plan: &[PlanOp], params: &Params) -> Result<(), String> {
737 let mut names = Vec::new();
738 let mut seen = BTreeSet::new();
739 collect_params_from_ops(plan, &mut names, &mut seen)?;
740 for name in names {
741 if !params.0.contains_key(&name) {
742 return Err(format!("missing parameter `{name}`"));
743 }
744 }
745 Ok(())
746}
747
748fn collect_params_from_ops(
749 plan: &[PlanOp],
750 names: &mut Vec<String>,
751 seen: &mut BTreeSet<String>,
752) -> Result<(), String> {
753 for op in plan {
754 match op {
755 PlanOp::ScanKey { key, .. } => collect_operand(key, names, seen),
756 PlanOp::LookupProps { props, .. }
757 | PlanOp::JoinBound { props, .. }
758 | PlanOp::Expand {
759 to_props: props, ..
760 } => {
761 for (_, operand) in props {
762 collect_operand(operand, names, seen);
763 }
764 }
765 PlanOp::Filter { expr } => collect_expr(expr, names, seen, 0)?,
766 PlanOp::LeftOuterApply { inner, .. } => {
767 collect_params_from_ops(inner, names, seen)?;
768 }
769 PlanOp::Project { items } => {
772 for item in items {
773 collect_ret_item_params(item, names, seen);
774 }
775 }
776 PlanOp::With {
777 items, where_expr, ..
778 } => {
779 for item in items {
780 collect_ret_item_params(item, names, seen);
781 }
782 if let Some(expr) = where_expr {
783 let _ = collect_expr(expr, names, seen, 0);
784 }
785 }
786 PlanOp::GroupAggregate { keys, aggs } => {
787 for (_, item) in keys {
788 collect_ret_item_params(item, names, seen);
789 }
790 for (_, arg, _) in aggs {
791 match arg {
792 AggArg::Star => {}
793 AggArg::Var(_) => {}
794 AggArg::Prop { .. } => {}
795 }
796 }
797 }
798 PlanOp::Skip(LimitSkip::Param(n)) | PlanOp::Limit(LimitSkip::Param(n))
801 if seen.insert(n.clone()) =>
802 {
803 names.push(n.clone());
804 }
805 _ => {}
806 }
807 }
808 Ok(())
809}
810
811fn collect_ret_item_params(item: &RetItem, names: &mut Vec<String>, seen: &mut BTreeSet<String>) {
813 match &item.value {
814 RetVal::FuncCall { args, .. } => {
815 for arg in args {
816 collect_operand(arg, names, seen);
817 }
818 }
819 RetVal::ScalarExpr(op) => {
820 collect_operand(op, names, seen);
821 }
822 RetVal::Prop { .. } | RetVal::Var(_) | RetVal::Agg { .. } => {}
823 }
824}
825
826fn collect_operand(op: &Operand, names: &mut Vec<String>, seen: &mut BTreeSet<String>) {
827 match op {
828 Operand::Param(n) => {
829 if seen.insert(n.clone()) {
830 names.push(n.clone());
831 }
832 }
833 Operand::FuncCall { args, .. } => {
834 for arg in args {
835 collect_operand(arg, names, seen);
836 }
837 }
838 Operand::BinArith { left, right, .. } => {
839 collect_operand(left, names, seen);
840 collect_operand(right, names, seen);
841 }
842 _ => {}
843 }
844}
845
846fn collect_expr(
847 expr: &Expr,
848 names: &mut Vec<String>,
849 seen: &mut BTreeSet<String>,
850 depth: u32,
851) -> Result<(), String> {
852 if depth > 256 {
853 return Err("expression nesting too deep".into());
854 }
855 match expr {
856 Expr::And(lhs, rhs) | Expr::Or(lhs, rhs) => {
857 collect_expr(lhs, names, seen, depth + 1)?;
858 collect_expr(rhs, names, seen, depth + 1)
859 }
860 Expr::Not(inner) => collect_expr(inner, names, seen, depth + 1),
861 Expr::Cmp { lhs, rhs, .. } => {
862 collect_operand(lhs, names, seen);
863 collect_operand(rhs, names, seen);
864 Ok(())
865 }
866 Expr::Truthy(op) => {
867 collect_operand(op, names, seen);
868 Ok(())
869 }
870 Expr::IsNull(op) | Expr::IsNotNull(op) => {
871 collect_operand(op, names, seen);
872 Ok(())
873 }
874 Expr::In { expr, list } => {
875 collect_operand(expr, names, seen);
876 for item in list {
877 collect_operand(item, names, seen);
878 }
879 Ok(())
880 }
881 }
882}
883
884fn collect_vars(plan: &[PlanOp]) -> VarTable {
885 let mut vars = VarTable { names: Vec::new() };
886 for op in plan {
887 match op {
888 PlanOp::ScanLabel { var, .. } => {
889 vars.intern(var);
890 }
891 PlanOp::ScanKey { var, key, .. } => {
892 vars.intern(var);
893 intern_operand(&mut vars, key);
894 }
895 PlanOp::LookupProps { var, props } | PlanOp::JoinBound { var, props, .. } => {
896 vars.intern(var);
897 for (_, operand) in props {
898 intern_operand(&mut vars, operand);
899 }
900 }
901 PlanOp::Expand {
902 from,
903 rel_var,
904 to,
905 to_props,
906 ..
907 } => {
908 vars.intern(from);
909 vars.intern(to);
910 if let Some(r) = rel_var {
911 vars.intern(r);
912 }
913 for (_, operand) in to_props {
914 intern_operand(&mut vars, operand);
915 }
916 }
917 PlanOp::VarExpand {
918 from, rel_var, to, ..
919 } => {
920 vars.intern(from);
921 vars.intern(to);
922 if let Some(r) = rel_var {
923 vars.intern(r);
924 }
925 }
926 PlanOp::ShortestPath {
927 from, rel_var, to, ..
928 } => {
929 vars.intern(from);
930 vars.intern(to);
931 if let Some(r) = rel_var {
932 vars.intern(r);
933 }
934 }
935 PlanOp::Filter { expr } => intern_expr(&mut vars, expr),
936 PlanOp::Project { items } => {
937 for item in items {
938 match &item.value {
939 RetVal::Var(name) | RetVal::Prop { var: name, .. } => {
940 vars.intern(name);
941 }
942 RetVal::Agg { .. } => {} RetVal::FuncCall { args, .. } => {
944 for arg in args {
945 intern_operand(&mut vars, arg);
946 }
947 }
948 RetVal::ScalarExpr(op) => {
949 intern_operand(&mut vars, op);
950 }
951 }
952 }
953 }
954 PlanOp::Aggregate { arg, .. } => match arg {
955 AggArg::Star => {}
956 AggArg::Var(v) => {
957 vars.intern(v);
958 }
959 AggArg::Prop { var, .. } => {
960 vars.intern(var);
961 }
962 },
963 PlanOp::GroupAggregate { keys, aggs } => {
964 for (col, item) in keys {
966 match &item.value {
967 RetVal::Var(name) | RetVal::Prop { var: name, .. } => {
968 vars.intern(name);
969 }
970 RetVal::Agg { .. } => {}
971 RetVal::FuncCall { args, .. } => {
972 for arg in args {
973 intern_operand(&mut vars, arg);
974 }
975 }
976 RetVal::ScalarExpr(op) => {
977 intern_operand(&mut vars, op);
978 }
979 }
980 vars.intern(col);
982 }
983 for (_, arg, col) in aggs {
984 match arg {
985 AggArg::Star => {}
986 AggArg::Var(v) => {
987 vars.intern(v);
988 }
989 AggArg::Prop { var, .. } => {
990 vars.intern(var);
991 }
992 }
993 vars.intern(col);
995 }
996 }
997 PlanOp::With {
998 items,
999 where_expr,
1000 order_by,
1001 ..
1002 } => {
1003 for item in items {
1004 match &item.value {
1005 RetVal::Var(name) | RetVal::Prop { var: name, .. } => {
1006 vars.intern(name);
1007 }
1008 RetVal::Agg { .. } => {}
1009 RetVal::FuncCall { args, .. } => {
1010 for arg in args {
1011 intern_operand(&mut vars, arg);
1012 }
1013 }
1014 RetVal::ScalarExpr(op) => {
1015 intern_operand(&mut vars, op);
1016 }
1017 }
1018 if let Some(alias) = &item.alias {
1020 vars.intern(alias);
1021 } else {
1022 match &item.value {
1023 RetVal::Var(name) => {
1024 vars.intern(name);
1025 }
1026 RetVal::Prop { var, field } => {
1027 let col = format!("{var}.{field}");
1028 vars.intern(&col);
1029 }
1030 RetVal::Agg { .. } => {}
1031 RetVal::ScalarExpr(_) => {
1032 vars.intern("<expr>");
1033 }
1034 RetVal::FuncCall { name, args } => {
1035 let arg_strs: Vec<String> = args
1037 .iter()
1038 .map(|a| match a {
1039 Operand::Var(v) => v.clone(),
1040 Operand::Prop { var, field } => format!("{var}.{field}"),
1041 Operand::Lit(_) => "<lit>".to_string(),
1042 Operand::Param(p) => format!("${p}"),
1043 Operand::FuncCall { name: n, .. } => format!("{n}(...)"),
1044 Operand::BinArith { .. } => "<arith>".to_string(),
1045 })
1046 .collect();
1047 let col = format!("{name}({})", arg_strs.join(", "));
1048 vars.intern(&col);
1049 }
1050 }
1051 }
1052 }
1053 if let Some(expr) = where_expr {
1054 intern_expr(&mut vars, expr);
1055 }
1056 for oi in order_by {
1057 match &oi.target {
1058 OrderTarget::Alias(name) | OrderTarget::Var(name) => {
1059 vars.intern(name);
1060 }
1061 OrderTarget::Prop { var, .. } => {
1062 vars.intern(var);
1063 }
1064 }
1065 }
1066 }
1067 PlanOp::Unwind { expr, alias } => {
1068 vars.intern(alias);
1069 match expr {
1070 UnwindExpr::Prop { var, .. } => {
1071 vars.intern(var);
1072 }
1073 UnwindExpr::Var(name) => {
1074 vars.intern(name);
1075 }
1076 UnwindExpr::Lit(_) => {}
1077 }
1078 }
1079 PlanOp::LeftOuterApply {
1080 inner,
1081 optional_vars,
1082 } => {
1083 for op in inner {
1085 match op {
1089 PlanOp::ScanLabel { var, .. } => {
1090 vars.intern(var);
1091 }
1092 PlanOp::ScanKey { var, key, .. } => {
1093 vars.intern(var);
1094 intern_operand(&mut vars, key);
1095 }
1096 PlanOp::Expand {
1097 from, rel_var, to, ..
1098 } => {
1099 vars.intern(from);
1100 vars.intern(to);
1101 if let Some(r) = rel_var {
1102 vars.intern(r);
1103 }
1104 }
1105 PlanOp::JoinBound { var, .. } | PlanOp::LookupProps { var, .. } => {
1106 vars.intern(var);
1107 }
1108 PlanOp::Filter { expr } => intern_expr(&mut vars, expr),
1109 _ => {}
1110 }
1111 }
1112 for v in optional_vars {
1113 vars.intern(v);
1114 }
1115 }
1116 _ => {}
1117 }
1118 }
1119 vars
1120}
1121
1122fn intern_operand(vars: &mut VarTable, operand: &Operand) {
1123 match operand {
1124 Operand::Prop { var, .. } | Operand::Var(var) => {
1125 vars.intern(var);
1126 }
1127 Operand::Lit(_) | Operand::Param(_) => {}
1128 Operand::BinArith { left, right, .. } => {
1129 intern_operand(vars, left);
1130 intern_operand(vars, right);
1131 }
1132 Operand::FuncCall { args, .. } => {
1133 for arg in args {
1134 intern_operand(vars, arg);
1135 }
1136 }
1137 }
1138}
1139
1140fn intern_expr(vars: &mut VarTable, expr: &Expr) {
1141 match expr {
1142 Expr::And(lhs, rhs) | Expr::Or(lhs, rhs) => {
1143 intern_expr(vars, lhs);
1144 intern_expr(vars, rhs);
1145 }
1146 Expr::Not(inner) => intern_expr(vars, inner),
1147 Expr::Cmp { lhs, rhs, .. } => {
1148 intern_operand(vars, lhs);
1149 intern_operand(vars, rhs);
1150 }
1151 Expr::Truthy(op) => intern_operand(vars, op),
1152 Expr::IsNull(op) | Expr::IsNotNull(op) => intern_operand(vars, op),
1153 Expr::In { expr, list } => {
1154 intern_operand(vars, expr);
1155 for item in list {
1156 intern_operand(vars, item);
1157 }
1158 }
1159 }
1160}
1161
1162fn scan_ids(view: &GraphView, label: Option<&str>) -> Vec<u32> {
1163 let ids = match label {
1164 Some(label) => view.nodes_with_label(label),
1165 None => (0..view.ids.len() as u32)
1167 .filter(|&id| view.label_of(id).is_some())
1168 .collect(),
1169 };
1170 if view.mask.is_some() {
1171 ids.into_iter().filter(|&id| view.visible(id)).collect()
1172 } else {
1173 ids
1174 }
1175}
1176
1177fn resolve_scan_key_id(
1180 view: &GraphView,
1181 vars: &VarTable,
1182 row: &Row,
1183 key: &Operand,
1184 label: Option<&str>,
1185 params: &Params,
1186) -> Result<Option<u32>, String> {
1187 #[cfg(test)]
1188 SCAN_KEY_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1189
1190 let Some(val) = resolve_operand(view, vars, row, key, params)? else {
1191 return Ok(None);
1192 };
1193 let Value::Str(s) = val else {
1194 return Ok(None);
1195 };
1196 let Some(id) = view.node_id(&s) else {
1197 return Ok(None);
1198 };
1199 if !view.visible(id) {
1200 return Ok(None);
1201 }
1202 if let Some(want) = label {
1203 match view.label_of(id) {
1204 Some(got) if got == want => {}
1205 _ => return Ok(None),
1206 }
1207 }
1208 Ok(Some(id))
1209}
1210
1211fn scan_key(
1212 view: &GraphView,
1213 vars: &VarTable,
1214 rows: &[Row],
1215 var: &str,
1216 key: &Operand,
1217 label: Option<&str>,
1218 params: &Params,
1219) -> Result<Vec<Row>, String> {
1220 let slot = vars
1221 .slot(var)
1222 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1223 let cap = max_intermediate_rows();
1224 let mut out = Vec::new();
1225 for row in rows {
1226 let Some(id) = resolve_scan_key_id(view, vars, row, key, label, params)? else {
1227 continue;
1228 };
1229 if out.len() >= cap {
1230 return Err(row_cap_err(cap));
1231 }
1232 let mut next = row.clone();
1233 next[slot] = Some(Cell::Node(id));
1234 out.push(next);
1235 }
1236 Ok(out)
1237}
1238
1239fn scan_label(
1240 view: &GraphView,
1241 vars: &VarTable,
1242 rows: &[Row],
1243 var: &str,
1244 label: Option<&str>,
1245) -> Result<Vec<Row>, String> {
1246 let ids = scan_ids(view, label);
1247 let slot = vars
1248 .slot(var)
1249 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1250 let cap = max_intermediate_rows();
1251 let mut out = Vec::with_capacity(rows.len().saturating_mul(ids.len()).min(cap));
1252 for row in rows {
1253 for &id in &ids {
1254 if out.len() >= cap {
1255 return Err(row_cap_err(cap));
1256 }
1257 let mut next = row.clone();
1258 next[slot] = Some(Cell::Node(id));
1259 out.push(next);
1260 }
1261 }
1262 Ok(out)
1263}
1264
1265fn require_cell<'a>(row: &'a Row, vars: &VarTable, var: &str) -> Result<&'a Cell, String> {
1266 let slot = vars
1267 .slot(var)
1268 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1269 row.get(slot)
1270 .and_then(|c| c.as_ref())
1271 .ok_or_else(|| format!("unbound variable `{var}`"))
1272}
1273
1274fn require_node(row: &Row, vars: &VarTable, var: &str) -> Result<u32, String> {
1275 match require_cell(row, vars, var)? {
1276 Cell::Node(id) => Ok(*id),
1277 Cell::Rel(_) => Err(format!("variable `{var}` is not a node")),
1278 Cell::Path(_) => Err(format!("variable `{var}` is a path, not a node")),
1279 Cell::Scalar(_) => Err(format!("variable `{var}` is a scalar value, not a node")),
1280 }
1281}
1282
1283const SCALAR_FUNCS: &[&str] = &[
1285 "toLower",
1286 "toUpper",
1287 "size",
1288 "coalesce",
1289 "type",
1290 "abs",
1291 "round",
1292 "textMatches",
1293];
1294
1295fn eval_func(
1299 name: &str,
1300 args: &[Operand],
1301 view: &GraphView,
1302 vars: &VarTable,
1303 row: &Row,
1304 params: &Params,
1305) -> Result<Option<Value>, String> {
1306 let norm = name.to_ascii_lowercase();
1307 match norm.as_str() {
1308 "tolower" => {
1309 if args.len() != 1 {
1310 return Err(format!(
1311 "toLower() requires exactly 1 argument, got {}",
1312 args.len()
1313 ));
1314 }
1315 let v = resolve_operand(view, vars, row, &args[0], params)?;
1316 Ok(v.map(|val| match val {
1317 Value::Str(s) => Value::Str(s.to_ascii_lowercase()),
1318 other => other, }))
1320 }
1321 "toupper" => {
1322 if args.len() != 1 {
1323 return Err(format!(
1324 "toUpper() requires exactly 1 argument, got {}",
1325 args.len()
1326 ));
1327 }
1328 let v = resolve_operand(view, vars, row, &args[0], params)?;
1329 Ok(v.map(|val| match val {
1330 Value::Str(s) => Value::Str(s.to_ascii_uppercase()),
1331 other => other,
1332 }))
1333 }
1334 "size" => {
1335 if args.len() != 1 {
1336 return Err(format!(
1337 "size() requires exactly 1 argument, got {}",
1338 args.len()
1339 ));
1340 }
1341 let v = resolve_operand(view, vars, row, &args[0], params)?;
1342 match v {
1343 None => Ok(None), Some(Value::Str(s)) => Ok(Some(Value::Int(s.len() as i64))),
1345 Some(Value::List(items)) => Ok(Some(Value::Int(items.len() as i64))),
1346 Some(_) => Ok(None), }
1348 }
1349 "coalesce" => {
1350 for arg in args {
1352 if let Some(v) = resolve_operand(view, vars, row, arg, params)? {
1353 return Ok(Some(v));
1354 }
1355 }
1356 Ok(None)
1357 }
1358 "type" => {
1359 if args.len() != 1 {
1360 return Err(format!(
1361 "type() requires exactly 1 argument, got {}",
1362 args.len()
1363 ));
1364 }
1365 let Operand::Var(var_name) = &args[0] else {
1367 return Err(
1368 "type() argument must be a relationship variable (e.g. type(r))".to_string(),
1369 );
1370 };
1371 let slot = vars
1372 .slot(var_name)
1373 .ok_or_else(|| format!("unbound variable `{var_name}` in type()"))?;
1374 match row.get(slot).and_then(|c| c.as_ref()) {
1375 Some(Cell::Rel(e)) => {
1376 let etype = view.syms.resolve(e.etype).unwrap_or("").to_owned();
1378 Ok(Some(Value::Str(etype)))
1379 }
1380 Some(Cell::Node(_)) => Err(format!(
1381 "type() argument `{var_name}` is a node, not a relationship"
1382 )),
1383 Some(Cell::Scalar(_) | Cell::Path(_)) => Err(format!(
1384 "type() argument `{var_name}` is not a relationship"
1385 )),
1386 None => Ok(None), }
1388 }
1389 "abs" => {
1390 if args.len() != 1 {
1391 return Err(format!(
1392 "abs() requires exactly 1 argument, got {}",
1393 args.len()
1394 ));
1395 }
1396 let v = resolve_operand(view, vars, row, &args[0], params)?;
1397 match v {
1398 None => Ok(None),
1399 Some(Value::Int(n)) => Ok(Some(Value::Int(n.abs()))),
1400 Some(Value::Float(f)) => Ok(Some(Value::Float(f.abs()))),
1401 Some(_) => Ok(None), }
1403 }
1404 "round" => {
1405 if args.len() != 1 {
1406 return Err(format!(
1407 "round() requires exactly 1 argument, got {}",
1408 args.len()
1409 ));
1410 }
1411 let v = resolve_operand(view, vars, row, &args[0], params)?;
1412 match v {
1413 None => Ok(None),
1414 Some(Value::Int(n)) => Ok(Some(Value::Int(n))), Some(Value::Float(f)) => Ok(Some(Value::Float(f.round()))),
1416 Some(_) => Ok(None), }
1418 }
1419 "textmatches" => {
1420 if args.len() != 2 {
1432 return Err(format!(
1433 "textMatches() requires exactly 2 arguments (field_value, query), got {}",
1434 args.len()
1435 ));
1436 }
1437 let field_val = resolve_operand(view, vars, row, &args[0], params)?;
1438 let query_val = resolve_operand(view, vars, row, &args[1], params)?;
1439 match (field_val, query_val) {
1440 (None, _) | (_, None) => Ok(None),
1441 (Some(Value::Str(s)), Some(Value::Str(q))) => {
1442 use std::collections::BTreeSet;
1443 let node_tokens: BTreeSet<String> =
1444 core_storage::fulltext::tokenize(&s).into_iter().collect();
1445 Ok(Some(Value::Bool(core_storage::fulltext::eval_query(
1446 &node_tokens,
1447 &q,
1448 ))))
1449 }
1450 (Some(Value::List(items)), Some(Value::Str(q))) => {
1451 use std::collections::BTreeSet;
1452 let node_tokens: BTreeSet<String> = items
1453 .iter()
1454 .flat_map(|v| {
1455 if let Value::Str(s) = v {
1456 core_storage::fulltext::tokenize(s)
1457 } else {
1458 vec![]
1459 }
1460 })
1461 .collect();
1462 Ok(Some(Value::Bool(core_storage::fulltext::eval_query(
1463 &node_tokens,
1464 &q,
1465 ))))
1466 }
1467 _ => Ok(Some(Value::Bool(false))), }
1469 }
1470 _ => Err(format!(
1471 "unknown function `{name}`; supported: {}",
1472 SCALAR_FUNCS.join(", ")
1473 )),
1474 }
1475}
1476
1477fn resolve_operand(
1478 view: &GraphView,
1479 vars: &VarTable,
1480 row: &Row,
1481 operand: &Operand,
1482 params: &Params,
1483) -> Result<Option<Value>, String> {
1484 match operand {
1485 Operand::Lit(v) => Ok(Some(v.clone())),
1486 Operand::Param(name) => match params.0.get(name) {
1487 Some(v) => Ok(Some(v.clone())),
1488 None => Err(format!("missing parameter `{name}`")),
1489 },
1490 Operand::Prop { var, field } => resolve_prop(view, vars, row, var, field),
1491 Operand::Var(name) => {
1492 match vars
1494 .slot(name)
1495 .and_then(|s| row.get(s))
1496 .and_then(|c| c.as_ref())
1497 {
1498 Some(Cell::Scalar(v)) => Ok(Some(v.clone())),
1499 Some(Cell::Node(id)) => match view.ids.key_of(*id) {
1500 Some(key) => Ok(Some(Value::Str(key.to_owned()))),
1501 None => Ok(None),
1502 },
1503 Some(Cell::Path(hops)) => Ok(Some(Value::Int(*hops as i64))),
1504 Some(Cell::Rel(_)) => Err(format!("variable `{name}` is a relationship")),
1505 None => Ok(None),
1506 }
1507 }
1508 Operand::FuncCall { name, args } => eval_func(name, args, view, vars, row, params),
1509 Operand::BinArith { op, left, right } => {
1510 use super::ast::ArithOp;
1511 let lv = resolve_operand(view, vars, row, left, params)?;
1512 let rv = resolve_operand(view, vars, row, right, params)?;
1513 match (lv, rv) {
1514 (None, _) | (_, None) => Ok(None), (Some(Value::Int(a)), Some(Value::Int(b))) => {
1516 let result = match op {
1517 ArithOp::Sub => a.saturating_sub(b),
1518 ArithOp::Mul => a.saturating_mul(b),
1519 ArithOp::Add => a.saturating_add(b),
1520 ArithOp::Div => {
1521 if b == 0 {
1522 return Err("division by zero".into());
1523 }
1524 a.checked_div(b).unwrap_or(i64::MAX)
1525 }
1526 };
1527 Ok(Some(Value::Int(result)))
1528 }
1529 (Some(lv), Some(rv)) => {
1530 let a = match &lv {
1531 Value::Float(f) => *f,
1532 Value::Int(i) => *i as f64,
1533 _ => return Err(format!("arithmetic operand must be numeric, got {lv:?}")),
1534 };
1535 let b = match &rv {
1536 Value::Float(f) => *f,
1537 Value::Int(i) => *i as f64,
1538 _ => return Err(format!("arithmetic operand must be numeric, got {rv:?}")),
1539 };
1540 let result = match op {
1541 ArithOp::Sub => a - b,
1542 ArithOp::Mul => a * b,
1543 ArithOp::Add => a + b,
1544 ArithOp::Div => {
1545 if b == 0.0 {
1546 return Err("division by zero".into());
1547 }
1548 a / b
1549 }
1550 };
1551 Ok(Some(Value::Float(result)))
1552 }
1553 }
1554 }
1555 }
1556}
1557
1558fn resolve_prop(
1559 view: &GraphView,
1560 vars: &VarTable,
1561 row: &Row,
1562 var: &str,
1563 field: &str,
1564) -> Result<Option<Value>, String> {
1565 let slot = vars
1568 .slot(var)
1569 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1570 let cell = match row.get(slot).and_then(|c| c.as_ref()) {
1571 Some(c) => c,
1572 None => return Ok(None),
1574 };
1575 match cell {
1576 Cell::Node(id) => Ok(view.prop(*id, field).cloned()),
1577 Cell::Rel(e) => Ok(view.edge_props.get(e.etype, e.src, e.dst, field).cloned()),
1578 Cell::Path(hops) => {
1580 if field == "length" {
1581 Ok(Some(Value::Int(*hops as i64)))
1582 } else {
1583 Ok(None)
1584 }
1585 }
1586 Cell::Scalar(_) => Ok(None),
1588 }
1589}
1590
1591fn node_matches(
1592 view: &GraphView,
1593 vars: &VarTable,
1594 row: &Row,
1595 id: u32,
1596 label: Option<&str>,
1597 props: &[(String, Operand)],
1598 params: &Params,
1599) -> Result<bool, String> {
1600 if let Some(want) = label {
1601 match view.label_of(id) {
1602 Some(got) if got == want => {}
1603 _ => return Ok(false),
1604 }
1605 }
1606 for (field, operand) in props {
1607 let Some(expected) = resolve_operand(view, vars, row, operand, params)? else {
1608 return Ok(false);
1609 };
1610 match view.prop(id, field) {
1611 Some(got) if values_equal(got, &expected) => {}
1612 _ => return Ok(false),
1613 }
1614 }
1615 Ok(true)
1616}
1617
1618fn retain_node(
1619 view: &GraphView,
1620 vars: &VarTable,
1621 rows: &[Row],
1622 var: &str,
1623 label: Option<&str>,
1624 props: &[(String, Operand)],
1625 params: &Params,
1626) -> Result<Vec<Row>, String> {
1627 let mut out = Vec::with_capacity(rows.len());
1628 for row in rows {
1629 let id = require_node(row, vars, var)?;
1630 if node_matches(view, vars, row, id, label, props, params)? {
1631 out.push(row.clone());
1632 }
1633 }
1634 Ok(out)
1635}
1636
1637fn map_dir(dir: RelDir) -> Dir {
1638 match dir {
1639 RelDir::Right => Dir::Out,
1640 RelDir::Left => Dir::In,
1641 RelDir::Undirected => Dir::Both,
1642 }
1643}
1644
1645fn neighbor(from: u32, e: &EdgeRef, dir: RelDir) -> u32 {
1646 match dir {
1647 RelDir::Right => e.dst,
1648 RelDir::Left => e.src,
1649 RelDir::Undirected => {
1650 if e.src == from {
1651 e.dst
1652 } else {
1653 e.src
1654 }
1655 }
1656 }
1657}
1658
1659fn row_has_edge(row: &Row, e: &EdgeRef) -> bool {
1660 row.iter()
1661 .any(|c| matches!(c, Some(Cell::Rel(existing)) if existing == e))
1662}
1663
1664fn resolve_etypes(view: &GraphView, etype: Option<&str>) -> Option<Vec<u32>> {
1665 etype.map(|name| view.syms.get(name).into_iter().collect())
1666}
1667
1668fn exec_expand(
1669 view: &GraphView,
1670 vars: &VarTable,
1671 rows: &[Row],
1672 op: &PlanOp,
1673 params: &Params,
1674) -> Result<Vec<Row>, String> {
1675 let PlanOp::Expand {
1676 from,
1677 rel_var,
1678 etype,
1679 dir,
1680 to,
1681 to_label,
1682 to_props,
1683 } = op
1684 else {
1685 return Err("internal: expected Expand".into());
1686 };
1687 let etypes = resolve_etypes(view, etype.as_deref());
1688 let exp_dir = map_dir(*dir);
1689 let to_slot = vars
1690 .slot(to)
1691 .ok_or_else(|| format!("unbound variable `{to}`"))?;
1692 let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
1693 let cap = max_intermediate_rows();
1694 let mut out = Vec::with_capacity(rows.len().saturating_mul(2).min(cap));
1695 for row in rows {
1696 let from_id = require_node(row, vars, from)?;
1697 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
1698 Some(Cell::Node(id)) => Some(*id),
1699 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
1700 return Err(format!("variable `{to}` is not a node"))
1701 }
1702 None => None,
1703 };
1704 for e in expand(view, from_id, etypes.as_deref(), exp_dir) {
1705 if row_has_edge(row, &e) {
1706 continue;
1707 }
1708 let nbr = neighbor(from_id, &e, *dir);
1709 if !view.visible(nbr) {
1710 continue;
1711 }
1712 if let Some(want) = bound_to {
1713 if nbr != want {
1714 continue;
1715 }
1716 }
1717 if !node_matches(view, vars, row, nbr, to_label.as_deref(), to_props, params)? {
1718 continue;
1719 }
1720 if out.len() >= cap {
1721 return Err(row_cap_err(cap));
1722 }
1723 let mut next = row.clone();
1724 if let Some(slot) = rel_slot {
1725 next[slot] = Some(Cell::Rel(e));
1726 }
1727 if bound_to.is_none() {
1728 next[to_slot] = Some(Cell::Node(nbr));
1729 }
1730 out.push(next);
1731 #[cfg(test)]
1732 record_expand_row();
1733 }
1734 }
1735 Ok(out)
1736}
1737
1738#[allow(clippy::too_many_arguments)]
1745fn exec_var_expand(
1746 view: &GraphView,
1747 vars: &VarTable,
1748 rows: &[Row],
1749 from: &str,
1750 rel_var: &Option<String>,
1751 etype: &Option<String>,
1752 dir: RelDir,
1753 to: &str,
1754 min: u8,
1755 max: u8,
1756) -> Result<Vec<Row>, String> {
1757 let etypes = resolve_etypes(view, etype.as_deref());
1758 let exp_dir = map_dir(dir);
1759 let to_slot = vars
1760 .slot(to)
1761 .ok_or_else(|| format!("unbound variable `{to}`"))?;
1762 let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
1763 let cap = max_intermediate_rows();
1764 let mut out: Vec<Row> = Vec::new();
1765
1766 for row in rows {
1767 let from_id = require_node(row, vars, from)?;
1768 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
1769 Some(Cell::Node(id)) => Some(*id),
1770 Some(_) => return Err(format!("variable `{to}` is not a node")),
1771 None => None,
1772 };
1773
1774 struct PathState {
1777 node: u32,
1778 edges: Vec<EdgeRef>,
1779 }
1780
1781 let mut frontier: Vec<PathState> = vec![PathState {
1782 node: from_id,
1783 edges: Vec::new(),
1784 }];
1785
1786 let mut frontier_count: usize = 0;
1791
1792 for depth in 1u8..=max {
1793 let mut next_frontier: Vec<PathState> = Vec::new();
1794 for state in &frontier {
1795 for e in expand(view, state.node, etypes.as_deref(), exp_dir) {
1796 if state.edges.contains(&e) {
1798 continue;
1799 }
1800 let nbr = neighbor(state.node, &e, dir);
1801 if !view.visible(nbr) {
1804 continue;
1805 }
1806
1807 if depth >= min {
1810 let dest_matches = match bound_to {
1811 Some(want) => nbr == want,
1812 None => true,
1813 };
1814 if dest_matches {
1815 if out.len() >= cap {
1816 return Err(row_cap_err(cap));
1817 }
1818 let mut next = row.clone();
1819 if let Some(slot) = rel_slot {
1820 next[slot] = Some(Cell::Path(depth));
1821 }
1822 next[to_slot] = Some(Cell::Node(nbr));
1823 out.push(next);
1824 }
1825 }
1826
1827 if depth < max {
1831 frontier_count += 1;
1832 if frontier_count >= cap {
1833 return Err(row_cap_err(cap));
1834 }
1835 let mut new_edges = state.edges.clone();
1836 new_edges.push(e);
1837 next_frontier.push(PathState {
1838 node: nbr,
1839 edges: new_edges,
1840 });
1841 }
1842 }
1843 }
1844 frontier = next_frontier;
1845 if frontier.is_empty() {
1846 break;
1847 }
1848 }
1849 }
1850 Ok(out)
1851}
1852
1853#[allow(clippy::too_many_arguments)]
1859fn exec_shortest_path(
1860 view: &GraphView,
1861 vars: &VarTable,
1862 rows: &[Row],
1863 from: &str,
1864 rel_var: &Option<String>,
1865 etype: &Option<String>,
1866 dir: RelDir,
1867 to: &str,
1868 max_hops: u8,
1869) -> Result<Vec<Row>, String> {
1870 let etypes = resolve_etypes(view, etype.as_deref());
1871 let exp_dir = map_dir(dir);
1872 let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
1873 let mut out: Vec<Row> = Vec::new();
1874
1875 for row in rows {
1876 let from_id = require_node(row, vars, from)?;
1877 let to_id = require_node(row, vars, to)?;
1878
1879 let mut visited = std::collections::BTreeSet::new();
1881 visited.insert(from_id);
1882 let mut frontier: Vec<u32> = vec![from_id];
1883
1884 'bfs: for depth in 1u8..=max_hops {
1885 let mut next_frontier: Vec<u32> = Vec::new();
1886 for &node in &frontier {
1887 for e in expand(view, node, etypes.as_deref(), exp_dir) {
1888 let nbr = neighbor(node, &e, dir);
1889 if !view.visible(nbr) {
1893 continue;
1894 }
1895 if nbr == to_id {
1896 let mut next = row.clone();
1898 if let Some(slot) = rel_slot {
1899 next[slot] = Some(Cell::Path(depth));
1900 }
1901 out.push(next);
1902 break 'bfs;
1903 }
1904 if !visited.contains(&nbr) {
1905 visited.insert(nbr);
1906 next_frontier.push(nbr);
1907 }
1908 }
1909 }
1910 frontier = next_frontier;
1911 if frontier.is_empty() {
1912 break;
1913 }
1914 }
1915 }
1916 Ok(out)
1917}
1918
1919fn exec_filter(
1920 view: &GraphView,
1921 vars: &VarTable,
1922 rows: &[Row],
1923 expr: &Expr,
1924 params: &Params,
1925) -> Result<Vec<Row>, String> {
1926 let mut out = Vec::with_capacity(rows.len());
1927 for row in rows {
1928 if eval_expr(view, vars, row, expr, params, 0)? {
1929 out.push(row.clone());
1930 }
1931 }
1932 Ok(out)
1933}
1934
1935struct PullCtx<'a> {
1951 view: &'a GraphView<'a>,
1952 vars: &'a VarTable,
1953 project_items: &'a [RetItem],
1954 params: &'a Params<'a>,
1955 bound: usize,
1956}
1957
1958fn execute_pull(
1959 view: &GraphView,
1960 plan: &[PlanOp],
1961 params: &Params,
1962 bound: usize,
1963) -> Result<ResultSet, String> {
1964 let proj_pos = match plan
1965 .iter()
1966 .position(|op| matches!(op, PlanOp::Project { .. }))
1967 {
1968 Some(p) => p,
1969 None => return Ok(ResultSet::new(vec![])),
1970 };
1971 let producers = &plan[..proj_pos];
1972 let project_items = match &plan[proj_pos] {
1973 PlanOp::Project { items } => items,
1974 _ => unreachable!(),
1975 };
1976 let columns: Vec<String> = project_items.iter().map(column_name).collect();
1977 let vars = collect_vars(plan);
1978 let ctx = PullCtx {
1979 view,
1980 vars: &vars,
1981 project_items,
1982 params,
1983 bound,
1984 };
1985 let mut initial_row: Row = vec![None; vars.names.len()];
1986 let mut result_rows: Vec<Vec<Option<Value>>> = Vec::with_capacity(bound);
1987 pull_rows(&ctx, producers, &mut initial_row, &mut result_rows)?;
1988 let skip_n = plan[proj_pos + 1..]
1991 .iter()
1992 .find_map(|op| match op {
1993 PlanOp::Skip(ls) => Some(resolve_ls(ls, params)),
1994 _ => None,
1995 })
1996 .transpose()?
1997 .unwrap_or(0);
1998 let skip_n = usize::try_from(skip_n).unwrap_or(usize::MAX);
1999 let mut rs = ResultSet::new(columns);
2000 for row in result_rows.into_iter().skip(skip_n) {
2001 rs.push_row(row);
2002 }
2003 Ok(rs)
2004}
2005
2006enum AggAcc {
2020 Count(u64),
2021 Sum { val: f64, has_value: bool },
2022 Avg { sum: f64, n: u64 },
2023 Min(Option<Value>),
2024 Max(Option<Value>),
2025}
2026
2027impl AggAcc {
2028 fn new(func: &AggFunc) -> Self {
2029 match func {
2030 AggFunc::Count => AggAcc::Count(0),
2031 AggFunc::Sum => AggAcc::Sum {
2032 val: 0.0,
2033 has_value: false,
2034 },
2035 AggFunc::Avg => AggAcc::Avg { sum: 0.0, n: 0 },
2036 AggFunc::Min => AggAcc::Min(None),
2037 AggFunc::Max => AggAcc::Max(None),
2038 }
2039 }
2040
2041 fn finish(self) -> Option<Value> {
2042 match self {
2043 AggAcc::Count(n) => Some(Value::Int(i64::try_from(n).unwrap_or(i64::MAX))),
2048 AggAcc::Sum { val, has_value } => {
2049 if has_value {
2050 Some(Value::Float(val))
2051 } else {
2052 None
2053 }
2054 }
2055 AggAcc::Avg { sum, n } => {
2056 if n > 0 {
2057 Some(Value::Float(sum / n as f64))
2058 } else {
2059 None
2060 }
2061 }
2062 AggAcc::Min(v) => v,
2063 AggAcc::Max(v) => v,
2064 }
2065 }
2066}
2067
2068fn numeric_val(v: &Value) -> Option<f64> {
2071 match v {
2072 Value::Int(n) => Some(*n as f64),
2073 Value::Float(f) => Some(*f),
2074 _ => None,
2075 }
2076}
2077
2078struct AggStreamCtx<'a> {
2080 view: &'a GraphView<'a>,
2081 vars: &'a VarTable,
2082 params: &'a Params<'a>,
2083 func: &'a AggFunc,
2084 arg: &'a AggArg,
2085}
2086
2087fn execute_aggregate(
2092 view: &GraphView,
2093 plan: &[PlanOp],
2094 params: &Params,
2095) -> Result<ResultSet, String> {
2096 let agg_pos = match plan
2097 .iter()
2098 .position(|op| matches!(op, PlanOp::Aggregate { .. }))
2099 {
2100 Some(p) => p,
2101 None => return Ok(ResultSet::new(vec![])),
2102 };
2103 let producers = &plan[..agg_pos];
2104 let (func, arg, column) = match &plan[agg_pos] {
2105 PlanOp::Aggregate { func, arg, column } => (func, arg, column),
2106 _ => unreachable!(),
2107 };
2108 let vars = collect_vars(plan);
2109 let ctx = AggStreamCtx {
2110 view,
2111 vars: &vars,
2112 params,
2113 func,
2114 arg,
2115 };
2116 let initial_row: Row = vec![None; vars.names.len()];
2117 let mut acc = AggAcc::new(func);
2118 agg_stream(&ctx, producers, &initial_row, &mut acc)?;
2119
2120 let value = acc.finish();
2121 let mut rs = ResultSet::new(vec![column.clone()]);
2122 rs.push_row(vec![value]);
2123 Ok(rs)
2124}
2125
2126fn update_acc(
2132 view: &GraphView,
2133 vars: &VarTable,
2134 row: &Row,
2135 func: &AggFunc,
2136 arg: &AggArg,
2137 acc: &mut AggAcc,
2138) -> Result<(), String> {
2139 match (func, arg) {
2140 (AggFunc::Count, AggArg::Star) => {
2141 if let AggAcc::Count(n) = acc {
2142 *n += 1;
2143 }
2144 }
2145 (AggFunc::Count, AggArg::Var(v)) => {
2146 let slot = vars.slot(v);
2147 let is_bound = slot
2148 .and_then(|s| row.get(s))
2149 .and_then(|c| c.as_ref())
2150 .is_some();
2151 if is_bound {
2152 if let AggAcc::Count(n) = acc {
2153 *n += 1;
2154 }
2155 }
2156 }
2157 (AggFunc::Count, AggArg::Prop { var, field }) => {
2158 let val = resolve_prop(view, vars, row, var, field)?;
2159 if val.is_some() {
2160 if let AggAcc::Count(n) = acc {
2161 *n += 1;
2162 }
2163 }
2164 }
2165 (AggFunc::Sum, AggArg::Prop { var, field }) => {
2166 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2167 if let Some(num) = numeric_val(&v) {
2168 if let AggAcc::Sum { val, has_value } = acc {
2169 *val += num;
2170 *has_value = true;
2171 }
2172 }
2173 }
2174 }
2175 (AggFunc::Avg, AggArg::Prop { var, field }) => {
2176 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2177 if let Some(num) = numeric_val(&v) {
2178 if let AggAcc::Avg { sum, n } = acc {
2179 *sum += num;
2180 *n += 1;
2181 }
2182 }
2183 }
2184 }
2185 (AggFunc::Min, AggArg::Prop { var, field }) => {
2186 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2187 if numeric_val(&v).is_some() {
2188 if let AggAcc::Min(current) = acc {
2189 *current = Some(match current.take() {
2190 None => v,
2191 Some(prev) => {
2192 if cmp_optional(Some(&prev), Some(&v), false)
2193 == std::cmp::Ordering::Greater
2194 {
2195 v
2196 } else {
2197 prev
2198 }
2199 }
2200 });
2201 }
2202 }
2203 }
2204 }
2205 (AggFunc::Max, AggArg::Prop { var, field }) => {
2206 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2207 if numeric_val(&v).is_some() {
2208 if let AggAcc::Max(current) = acc {
2209 *current = Some(match current.take() {
2210 None => v,
2211 Some(prev) => {
2212 if cmp_optional(Some(&prev), Some(&v), true)
2213 == std::cmp::Ordering::Greater
2214 {
2215 v
2216 } else {
2217 prev
2218 }
2219 }
2220 });
2221 }
2222 }
2223 }
2224 }
2225 _ => {}
2228 }
2229 Ok(())
2230}
2231
2232fn agg_stream(
2235 ctx: &AggStreamCtx<'_>,
2236 ops: &[PlanOp],
2237 row: &Row,
2238 acc: &mut AggAcc,
2239) -> Result<(), String> {
2240 let (op, rest) = match ops.split_first() {
2241 Some(pair) => pair,
2242 None => {
2243 update_acc(ctx.view, ctx.vars, row, ctx.func, ctx.arg, acc)?;
2245 return Ok(());
2246 }
2247 };
2248
2249 match op {
2250 PlanOp::ScanLabel { var, label } => {
2251 let ids = scan_ids(ctx.view, label.as_deref());
2252 let slot = ctx
2253 .vars
2254 .slot(var)
2255 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2256 for &id in &ids {
2257 let mut next = row.clone();
2258 next[slot] = Some(Cell::Node(id));
2259 agg_stream(ctx, rest, &next, acc)?;
2260 }
2261 }
2262 PlanOp::ScanKey { var, key, label } => {
2263 let slot = ctx
2264 .vars
2265 .slot(var)
2266 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2267 if let Some(id) =
2268 resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
2269 {
2270 let mut next = row.clone();
2271 next[slot] = Some(Cell::Node(id));
2272 agg_stream(ctx, rest, &next, acc)?;
2273 }
2274 }
2275 PlanOp::Expand {
2276 from,
2277 rel_var,
2278 etype,
2279 dir,
2280 to,
2281 to_label,
2282 to_props,
2283 } => {
2284 let etypes = resolve_etypes(ctx.view, etype.as_deref());
2285 let exp_dir = map_dir(*dir);
2286 let to_slot = ctx
2287 .vars
2288 .slot(to)
2289 .ok_or_else(|| format!("unbound variable `{to}`"))?;
2290 let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
2291 let from_id = require_node(row, ctx.vars, from)?;
2292 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
2293 Some(Cell::Node(id)) => Some(*id),
2294 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
2295 return Err(format!("variable `{to}` is not a node"))
2296 }
2297 None => None,
2298 };
2299 for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
2300 if row_has_edge(row, &e) {
2301 continue;
2302 }
2303 let nbr = neighbor(from_id, &e, *dir);
2304 if !ctx.view.visible(nbr) {
2305 continue;
2306 }
2307 if let Some(want) = bound_to {
2308 if nbr != want {
2309 continue;
2310 }
2311 }
2312 if !node_matches(
2313 ctx.view,
2314 ctx.vars,
2315 row,
2316 nbr,
2317 to_label.as_deref(),
2318 to_props,
2319 ctx.params,
2320 )? {
2321 continue;
2322 }
2323 let mut next = row.clone();
2324 if let Some(slot) = rel_slot {
2325 next[slot] = Some(Cell::Rel(e));
2326 }
2327 if bound_to.is_none() {
2328 next[to_slot] = Some(Cell::Node(nbr));
2329 }
2330 agg_stream(ctx, rest, &next, acc)?;
2331 }
2332 }
2333 PlanOp::Filter { expr } => {
2334 if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
2335 agg_stream(ctx, rest, row, acc)?;
2336 }
2337 }
2338 PlanOp::LookupProps { var, props } => {
2339 let id = require_node(row, ctx.vars, var)?;
2340 if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
2341 agg_stream(ctx, rest, row, acc)?;
2342 }
2343 }
2344 PlanOp::JoinBound { var, label, props } => {
2345 let id = require_node(row, ctx.vars, var)?;
2346 if node_matches(
2347 ctx.view,
2348 ctx.vars,
2349 row,
2350 id,
2351 label.as_deref(),
2352 props,
2353 ctx.params,
2354 )? {
2355 agg_stream(ctx, rest, row, acc)?;
2356 }
2357 }
2358 PlanOp::VarExpand {
2359 from,
2360 rel_var,
2361 etype,
2362 dir,
2363 to,
2364 min,
2365 max,
2366 } => {
2367 let new_rows = exec_var_expand(
2368 ctx.view,
2369 ctx.vars,
2370 std::slice::from_ref(row),
2371 from,
2372 rel_var,
2373 etype,
2374 *dir,
2375 to,
2376 *min,
2377 *max,
2378 )?;
2379 for nr in &new_rows {
2380 agg_stream(ctx, rest, nr, acc)?;
2381 }
2382 }
2383 PlanOp::ShortestPath {
2384 from,
2385 rel_var,
2386 etype,
2387 dir,
2388 to,
2389 max_hops,
2390 } => {
2391 let new_rows = exec_shortest_path(
2392 ctx.view,
2393 ctx.vars,
2394 std::slice::from_ref(row),
2395 from,
2396 rel_var,
2397 etype,
2398 *dir,
2399 to,
2400 *max_hops,
2401 )?;
2402 for nr in &new_rows {
2403 agg_stream(ctx, rest, nr, acc)?;
2404 }
2405 }
2406 PlanOp::Project { .. } => {
2408 return Err(
2409 "agg executor: Project in producer slice — plan is structurally malformed"
2410 .to_string(),
2411 );
2412 }
2413 PlanOp::Distinct => {
2414 return Err(
2415 "agg executor: Distinct in producer slice — plan is structurally malformed"
2416 .to_string(),
2417 );
2418 }
2419 PlanOp::OrderBy { .. } => {
2420 return Err(
2421 "agg executor: OrderBy in producer slice — structurally malformed".to_string(),
2422 );
2423 }
2424 PlanOp::Skip(_) => {
2425 return Err(
2426 "agg executor: Skip in producer slice — structurally malformed".to_string(),
2427 );
2428 }
2429 PlanOp::Limit(_) => {
2430 return Err(
2431 "agg executor: Limit in producer slice — structurally malformed".to_string(),
2432 );
2433 }
2434 PlanOp::Aggregate { .. } => {
2435 return Err(
2436 "agg executor: nested Aggregate in producer slice — structurally malformed"
2437 .to_string(),
2438 );
2439 }
2440 PlanOp::GroupAggregate { .. } => {
2441 return Err(
2442 "agg executor: GroupAggregate in producer slice — structurally malformed"
2443 .to_string(),
2444 );
2445 }
2446 PlanOp::With { .. } => {
2447 return Err(
2448 "agg executor: With in producer slice — structurally malformed".to_string(),
2449 );
2450 }
2451 PlanOp::Unwind { .. } => {
2452 return Err(
2453 "agg executor: Unwind in producer slice — structurally malformed".to_string(),
2454 );
2455 }
2456 PlanOp::LeftOuterApply { .. } => {
2457 return Err(
2458 "agg executor: LeftOuterApply in producer slice — structurally malformed"
2459 .to_string(),
2460 );
2461 }
2462 }
2463 Ok(())
2464}
2465
2466struct GroupStreamCtx<'a> {
2474 view: &'a GraphView<'a>,
2475 vars: &'a VarTable,
2476 params: &'a Params<'a>,
2477 keys: &'a [(String, RetItem)],
2478 aggs: &'a [(AggFunc, AggArg, String)],
2479}
2480
2481fn build_group_projected(
2483 keys: &[(String, RetItem)],
2484 aggs: &[(AggFunc, AggArg, String)],
2485 key_order: Vec<GroupKey>,
2486 groups: &mut HashMap<GroupKey, GroupEntry>,
2487) -> Projected {
2488 let columns: Vec<String> = keys
2489 .iter()
2490 .map(|(col, _)| col.clone())
2491 .chain(aggs.iter().map(|(_, _, col)| col.clone()))
2492 .collect();
2493 let mut rows: Vec<Vec<Option<Value>>> = Vec::with_capacity(key_order.len());
2494 for gk in key_order {
2495 let (display_keys, accs) = groups.remove(&gk).unwrap_or_default();
2496 let mut row: Vec<Option<Value>> = Vec::with_capacity(columns.len());
2497 for display_val in display_keys {
2500 row.push(display_val);
2501 }
2502 for acc in accs {
2503 row.push(acc.finish());
2504 }
2505 rows.push(row);
2506 }
2507 Projected { columns, rows }
2508}
2509
2510fn group_result_to_rows(
2517 keys: &[(String, RetItem)],
2518 aggs: &[(AggFunc, AggArg, String)],
2519 key_order: Vec<GroupKey>,
2520 groups: &mut HashMap<GroupKey, GroupEntry>,
2521 vars: &VarTable,
2522) -> Vec<Row> {
2523 let row_len = vars.names.len();
2524 let mut out: Vec<Row> = Vec::with_capacity(key_order.len());
2525 for gk in key_order {
2526 let (display_keys, accs) = groups.remove(&gk).unwrap_or_default();
2527 let mut row: Row = vec![None; row_len];
2528 for ((col, _), val) in keys.iter().zip(display_keys) {
2529 if let Some(slot) = vars.slot(col) {
2530 row[slot] = val.map(Cell::Scalar);
2531 }
2532 }
2533 for ((_, _, col), acc) in aggs.iter().zip(accs) {
2534 if let Some(slot) = vars.slot(col) {
2535 row[slot] = acc.finish().map(Cell::Scalar);
2536 }
2537 }
2538 out.push(row);
2539 }
2540 out
2541}
2542
2543fn exec_order_by_rows(vars: &VarTable, rows: &mut Vec<Row>, items: &[OrderItem], view: &GraphView) {
2550 let mut key_table: Vec<Vec<Option<Value>>> = Vec::with_capacity(rows.len());
2553 for row in rows.iter() {
2554 let mut row_key: Vec<Option<Value>> = Vec::with_capacity(items.len());
2555 for item in items {
2556 let val = match &item.target {
2557 OrderTarget::Alias(name) | OrderTarget::Var(name) => vars
2558 .slot(name)
2559 .and_then(|s| row.get(s))
2560 .and_then(|c| c.as_ref())
2561 .and_then(|c| match c {
2562 Cell::Scalar(v) => Some(v.clone()),
2563 Cell::Node(id) => view.ids.key_of(*id).map(|k| Value::Str(k.to_owned())),
2564 Cell::Path(hops) => Some(Value::Int(*hops as i64)),
2565 Cell::Rel(_) => None,
2566 }),
2567 OrderTarget::Prop { var, field } => vars
2568 .slot(var)
2569 .and_then(|s| row.get(s))
2570 .and_then(|c| c.as_ref())
2571 .and_then(|c| match c {
2572 Cell::Node(id) => view.prop(*id, field).cloned(),
2573 Cell::Rel(e) => view.edge_props.get(e.etype, e.src, e.dst, field).cloned(),
2574 _ => None,
2575 }),
2576 };
2577 row_key.push(val);
2578 }
2579 key_table.push(row_key);
2580 }
2581 let mut indices: Vec<usize> = (0..rows.len()).collect();
2583 indices.sort_by(|&a, &b| {
2584 for (ki, item) in items.iter().enumerate() {
2585 let c = cmp_optional(
2586 key_table[a].get(ki).and_then(|x| x.as_ref()),
2587 key_table[b].get(ki).and_then(|x| x.as_ref()),
2588 item.descending,
2589 );
2590 if c != std::cmp::Ordering::Equal {
2591 return c;
2592 }
2593 }
2594 std::cmp::Ordering::Equal
2595 });
2596 let sorted: Vec<Row> = indices.into_iter().map(|i| rows[i].clone()).collect();
2597 *rows = sorted;
2598}
2599
2600fn execute_group_aggregate(
2606 view: &GraphView,
2607 plan: &[PlanOp],
2608 params: &Params,
2609) -> Result<ResultSet, String> {
2610 let gagg_pos = plan
2611 .iter()
2612 .position(|op| matches!(op, PlanOp::GroupAggregate { .. }))
2613 .ok_or_else(|| "internal: GroupAggregate op not found in plan".to_string())?;
2614 let producers = &plan[..gagg_pos];
2615 let (keys, aggs) = match &plan[gagg_pos] {
2616 PlanOp::GroupAggregate { keys, aggs } => (keys, aggs),
2617 _ => unreachable!(),
2618 };
2619 let tail = &plan[gagg_pos + 1..];
2620 let vars = collect_vars(plan);
2621 let initial_row: Row = vec![None; vars.names.len()];
2622 let mut groups: HashMap<GroupKey, GroupEntry> = HashMap::new();
2623 let mut key_order: Vec<GroupKey> = Vec::new();
2624 let ctx = GroupStreamCtx {
2625 view,
2626 vars: &vars,
2627 params,
2628 keys,
2629 aggs,
2630 };
2631 group_stream(&ctx, producers, &initial_row, &mut groups, &mut key_order)?;
2632 if keys.is_empty() && key_order.is_empty() {
2636 let empty_key: GroupKey = vec![];
2637 key_order.push(empty_key.clone());
2638 groups.insert(
2639 empty_key,
2640 (
2641 vec![],
2642 aggs.iter().map(|(f, _, _)| AggAcc::new(f)).collect(),
2643 ),
2644 );
2645 }
2646 let mut projected = build_group_projected(keys, aggs, key_order, &mut groups);
2647 for op in tail {
2649 match op {
2650 PlanOp::OrderBy { items } => exec_order_by(&mut projected, items)?,
2651 PlanOp::Skip(ls) => {
2652 let n = resolve_ls(ls, params)?;
2653 apply_skip(&mut projected.rows, n);
2654 }
2655 PlanOp::Limit(ls) => {
2656 let n = resolve_ls(ls, params)?;
2657 apply_limit(&mut projected.rows, n);
2658 }
2659 _ => {} }
2661 }
2662 Ok(finish(projected))
2663}
2664
2665fn group_stream(
2672 ctx: &GroupStreamCtx<'_>,
2673 ops: &[PlanOp],
2674 row: &Row,
2675 groups: &mut HashMap<GroupKey, GroupEntry>,
2676 key_order: &mut Vec<GroupKey>,
2677) -> Result<(), String> {
2678 let (op, rest) = match ops.split_first() {
2679 Some(pair) => pair,
2680 None => {
2681 let mut gk: GroupKey = Vec::with_capacity(ctx.keys.len());
2684 let mut display_vals: Vec<Option<Value>> = Vec::with_capacity(ctx.keys.len());
2685 for (_, item) in ctx.keys {
2686 let val = project_item(ctx.view, ctx.vars, row, item, ctx.params)?;
2687 gk.push(val.as_ref().and_then(group_key_normalize));
2688 display_vals.push(val);
2689 }
2690 if !groups.contains_key(&gk) {
2691 if groups.len() >= max_groups() {
2692 return Err(group_cap_err());
2693 }
2694 key_order.push(gk.clone());
2695 let init: Vec<AggAcc> = ctx.aggs.iter().map(|(f, _, _)| AggAcc::new(f)).collect();
2696 groups.insert(gk.clone(), (display_vals, init));
2697 }
2698 let (_, accs) = groups.get_mut(&gk).unwrap();
2699 for (acc, (func, arg, _)) in accs.iter_mut().zip(ctx.aggs.iter()) {
2700 update_acc(ctx.view, ctx.vars, row, func, arg, acc)?;
2701 }
2702 return Ok(());
2703 }
2704 };
2705
2706 match op {
2707 PlanOp::ScanLabel { var, label } => {
2708 let ids = scan_ids(ctx.view, label.as_deref());
2709 let slot = ctx
2710 .vars
2711 .slot(var)
2712 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2713 for &id in &ids {
2714 let mut next = row.clone();
2715 next[slot] = Some(Cell::Node(id));
2716 group_stream(ctx, rest, &next, groups, key_order)?;
2717 }
2718 }
2719 PlanOp::ScanKey { var, key, label } => {
2720 let slot = ctx
2721 .vars
2722 .slot(var)
2723 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2724 if let Some(id) =
2725 resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
2726 {
2727 let mut next = row.clone();
2728 next[slot] = Some(Cell::Node(id));
2729 group_stream(ctx, rest, &next, groups, key_order)?;
2730 }
2731 }
2732 PlanOp::Expand {
2733 from,
2734 rel_var,
2735 etype,
2736 dir,
2737 to,
2738 to_label,
2739 to_props,
2740 } => {
2741 let etypes = resolve_etypes(ctx.view, etype.as_deref());
2742 let exp_dir = map_dir(*dir);
2743 let to_slot = ctx
2744 .vars
2745 .slot(to)
2746 .ok_or_else(|| format!("unbound variable `{to}`"))?;
2747 let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
2748 let from_id = require_node(row, ctx.vars, from)?;
2749 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
2750 Some(Cell::Node(id)) => Some(*id),
2751 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
2752 return Err(format!("variable `{to}` is not a node"))
2753 }
2754 None => None,
2755 };
2756 for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
2757 if row_has_edge(row, &e) {
2758 continue;
2759 }
2760 let nbr = neighbor(from_id, &e, *dir);
2761 if !ctx.view.visible(nbr) {
2762 continue;
2763 }
2764 if let Some(want) = bound_to {
2765 if nbr != want {
2766 continue;
2767 }
2768 }
2769 if !node_matches(
2770 ctx.view,
2771 ctx.vars,
2772 row,
2773 nbr,
2774 to_label.as_deref(),
2775 to_props,
2776 ctx.params,
2777 )? {
2778 continue;
2779 }
2780 let mut next = row.clone();
2781 if let Some(slot) = rel_slot {
2782 next[slot] = Some(Cell::Rel(e));
2783 }
2784 if bound_to.is_none() {
2785 next[to_slot] = Some(Cell::Node(nbr));
2786 }
2787 group_stream(ctx, rest, &next, groups, key_order)?;
2788 }
2789 }
2790 PlanOp::Filter { expr } => {
2791 if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
2792 group_stream(ctx, rest, row, groups, key_order)?;
2793 }
2794 }
2795 PlanOp::LookupProps { var, props } => {
2796 let id = require_node(row, ctx.vars, var)?;
2797 if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
2798 group_stream(ctx, rest, row, groups, key_order)?;
2799 }
2800 }
2801 PlanOp::JoinBound { var, label, props } => {
2802 let id = require_node(row, ctx.vars, var)?;
2803 if node_matches(
2804 ctx.view,
2805 ctx.vars,
2806 row,
2807 id,
2808 label.as_deref(),
2809 props,
2810 ctx.params,
2811 )? {
2812 group_stream(ctx, rest, row, groups, key_order)?;
2813 }
2814 }
2815 PlanOp::VarExpand {
2816 from,
2817 rel_var,
2818 etype,
2819 dir,
2820 to,
2821 min,
2822 max,
2823 } => {
2824 let new_rows = exec_var_expand(
2825 ctx.view,
2826 ctx.vars,
2827 std::slice::from_ref(row),
2828 from,
2829 rel_var,
2830 etype,
2831 *dir,
2832 to,
2833 *min,
2834 *max,
2835 )?;
2836 for nr in &new_rows {
2837 group_stream(ctx, rest, nr, groups, key_order)?;
2838 }
2839 }
2840 PlanOp::ShortestPath {
2841 from,
2842 rel_var,
2843 etype,
2844 dir,
2845 to,
2846 max_hops,
2847 } => {
2848 let new_rows = exec_shortest_path(
2849 ctx.view,
2850 ctx.vars,
2851 std::slice::from_ref(row),
2852 from,
2853 rel_var,
2854 etype,
2855 *dir,
2856 to,
2857 *max_hops,
2858 )?;
2859 for nr in &new_rows {
2860 group_stream(ctx, rest, nr, groups, key_order)?;
2861 }
2862 }
2863 PlanOp::Project { .. } => {
2865 return Err(
2866 "group executor: Project in producer slice — structurally malformed".to_string(),
2867 );
2868 }
2869 PlanOp::Distinct => {
2870 return Err(
2871 "group executor: Distinct in producer slice — structurally malformed".to_string(),
2872 );
2873 }
2874 PlanOp::OrderBy { .. } => {
2875 return Err(
2876 "group executor: OrderBy in producer slice — structurally malformed".to_string(),
2877 );
2878 }
2879 PlanOp::Skip(_) => {
2880 return Err(
2881 "group executor: Skip in producer slice — structurally malformed".to_string(),
2882 );
2883 }
2884 PlanOp::Limit(_) => {
2885 return Err(
2886 "group executor: Limit in producer slice — structurally malformed".to_string(),
2887 );
2888 }
2889 PlanOp::Aggregate { .. } => {
2890 return Err(
2891 "group executor: Aggregate in producer slice — structurally malformed".to_string(),
2892 );
2893 }
2894 PlanOp::GroupAggregate { .. } => {
2895 return Err(
2896 "group executor: nested GroupAggregate in producer slice — structurally malformed"
2897 .to_string(),
2898 );
2899 }
2900 PlanOp::With { .. } => {
2901 return Err(
2902 "group executor: With in producer slice — structurally malformed".to_string(),
2903 );
2904 }
2905 PlanOp::Unwind { .. } => {
2906 return Err(
2907 "group executor: Unwind in producer slice — structurally malformed".to_string(),
2908 );
2909 }
2910 PlanOp::LeftOuterApply { .. } => {
2911 return Err(
2912 "group executor: LeftOuterApply in producer slice — structurally malformed"
2913 .to_string(),
2914 );
2915 }
2916 }
2917 Ok(())
2918}
2919
2920fn pull_rows(
2934 ctx: &PullCtx<'_>,
2935 ops: &[PlanOp],
2936 row: &mut Row,
2937 result: &mut Vec<Vec<Option<Value>>>,
2938) -> Result<(), String> {
2939 if result.len() >= ctx.bound {
2940 return Ok(());
2941 }
2942 let (op, rest) = match ops.split_first() {
2943 Some(pair) => pair,
2944 None => {
2945 let mut cells = Vec::with_capacity(ctx.project_items.len());
2947 for item in ctx.project_items {
2948 cells.push(project_item(ctx.view, ctx.vars, row, item, ctx.params)?);
2949 }
2950 result.push(cells);
2951 return Ok(());
2952 }
2953 };
2954 match op {
2955 PlanOp::ScanLabel { var, label } => {
2956 let slot = ctx
2957 .vars
2958 .slot(var)
2959 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2960 let want_sym = label.as_deref().and_then(|l| ctx.view.syms.get(l));
2963 if label.is_some() && want_sym.is_none() {
2964 return Ok(());
2965 }
2966 let prev = row[slot].clone();
2968
2969 let fused_filter = rest.first().and_then(|next_op| {
2975 if let PlanOp::Filter {
2976 expr:
2977 Expr::Cmp {
2978 lhs:
2979 Operand::Prop {
2980 var: ref fv,
2981 field: ref f,
2982 },
2983 op: ref cmp_op_ref,
2984 rhs: Operand::Lit(ref lit),
2985 },
2986 } = *next_op
2987 {
2988 if fv == var {
2989 return Some((f.as_str(), cmp_op_ref, lit));
2990 }
2991 }
2992 None
2993 });
2994
2995 if let Some((field, cmp_op_ref, lit)) = fused_filter {
2996 #[cfg(test)]
2999 FUSED_SCAN_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
3000 let col = ctx.view.props.column(field);
3001 let rest_after_filter = &rest[1..];
3002 for (i, &sym) in ctx.view.labels.iter().enumerate() {
3003 if result.len() >= ctx.bound {
3004 break;
3005 }
3006 if sym == u32::MAX {
3007 continue;
3008 }
3009 if let Some(ws) = want_sym {
3010 if sym != ws {
3011 continue;
3012 }
3013 }
3014 let id = i as u32;
3015 if !ctx.view.visible(id) {
3016 continue;
3017 }
3018 if let Some(v) = col.get(id) {
3019 if eval_cmp(cmp_op_ref, v, lit) {
3020 row[slot] = Some(Cell::Node(id));
3021 pull_rows(ctx, rest_after_filter, row, result)?;
3022 }
3023 }
3024 }
3025 } else {
3026 for (i, &sym) in ctx.view.labels.iter().enumerate() {
3029 if result.len() >= ctx.bound {
3030 break;
3031 }
3032 if sym == u32::MAX {
3034 continue;
3035 }
3036 if let Some(ws) = want_sym {
3038 if sym != ws {
3039 continue;
3040 }
3041 }
3042 let id = i as u32;
3043 if !ctx.view.visible(id) {
3044 continue;
3045 }
3046 row[slot] = Some(Cell::Node(id));
3047 pull_rows(ctx, rest, row, result)?;
3048 }
3049 }
3050 row[slot] = prev;
3056 }
3057 PlanOp::ScanKey { var, key, label } => {
3058 let slot = ctx
3059 .vars
3060 .slot(var)
3061 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3062 if let Some(id) =
3063 resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
3064 {
3065 let prev = row[slot].clone();
3066 row[slot] = Some(Cell::Node(id));
3067 pull_rows(ctx, rest, row, result)?;
3068 row[slot] = prev;
3069 }
3070 }
3071 PlanOp::Expand {
3072 from,
3073 rel_var,
3074 etype,
3075 dir,
3076 to,
3077 to_label,
3078 to_props,
3079 } => {
3080 let etypes = resolve_etypes(ctx.view, etype.as_deref());
3081 let exp_dir = map_dir(*dir);
3082 let to_slot = ctx
3083 .vars
3084 .slot(to)
3085 .ok_or_else(|| format!("unbound variable `{to}`"))?;
3086 let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
3087 let from_id = require_node(row, ctx.vars, from)?;
3088 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
3089 Some(Cell::Node(id)) => Some(*id),
3090 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
3091 return Err(format!("variable `{to}` is not a node"))
3092 }
3093 None => None,
3094 };
3095 for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
3096 if result.len() >= ctx.bound {
3097 break;
3098 }
3099 if row_has_edge(row, &e) {
3100 continue;
3101 }
3102 let nbr = neighbor(from_id, &e, *dir);
3103 if !ctx.view.visible(nbr) {
3104 continue;
3105 }
3106 if let Some(want) = bound_to {
3107 if nbr != want {
3108 continue;
3109 }
3110 }
3111 if !node_matches(
3112 ctx.view,
3113 ctx.vars,
3114 row,
3115 nbr,
3116 to_label.as_deref(),
3117 to_props,
3118 ctx.params,
3119 )? {
3120 continue;
3121 }
3122 let mut next = row.clone();
3123 if let Some(slot) = rel_slot {
3124 next[slot] = Some(Cell::Rel(e));
3125 }
3126 if bound_to.is_none() {
3127 next[to_slot] = Some(Cell::Node(nbr));
3128 }
3129 #[cfg(test)]
3130 record_expand_row();
3131 pull_rows(ctx, rest, &mut next, result)?;
3132 }
3133 }
3134 PlanOp::Filter { expr } => {
3135 if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
3136 pull_rows(ctx, rest, row, result)?;
3137 }
3138 }
3139 PlanOp::LookupProps { var, props } => {
3140 let id = require_node(row, ctx.vars, var)?;
3141 if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
3142 pull_rows(ctx, rest, row, result)?;
3143 }
3144 }
3145 PlanOp::JoinBound { var, label, props } => {
3146 let id = require_node(row, ctx.vars, var)?;
3147 if node_matches(
3148 ctx.view,
3149 ctx.vars,
3150 row,
3151 id,
3152 label.as_deref(),
3153 props,
3154 ctx.params,
3155 )? {
3156 pull_rows(ctx, rest, row, result)?;
3157 }
3158 }
3159 PlanOp::Project { .. } => {
3164 return Err(
3165 "pull executor: Project reached pull_rows — plan is structurally malformed"
3166 .to_string(),
3167 );
3168 }
3169 PlanOp::Distinct => {
3170 return Err(
3171 "pull executor: Distinct reached pull_rows — DISTINCT queries must use \
3172 the staged path (row_bound returns None)"
3173 .to_string(),
3174 );
3175 }
3176 PlanOp::OrderBy { .. } => {
3177 return Err(
3178 "pull executor: OrderBy reached pull_rows — queries with ORDER BY \
3179 must use the staged path (row_bound returns None)"
3180 .to_string(),
3181 );
3182 }
3183 PlanOp::Skip(_) => {
3184 return Err(
3185 "pull executor: Skip reached pull_rows — Skip must appear after Project"
3186 .to_string(),
3187 );
3188 }
3189 PlanOp::Limit(_) => {
3190 return Err(
3191 "pull executor: Limit reached pull_rows — Limit must appear after Project"
3192 .to_string(),
3193 );
3194 }
3195 PlanOp::Aggregate { .. } => {
3196 return Err(
3197 "pull executor: Aggregate reached pull_rows — aggregate plans must use \
3198 the execute_aggregate path (routed before pull in execute_inner)"
3199 .to_string(),
3200 );
3201 }
3202 PlanOp::VarExpand { .. } => {
3207 return Err(
3208 "pull executor: VarExpand reached pull_rows — variable-length path \
3209 plans must use the staged path (row_bound returns None)"
3210 .to_string(),
3211 );
3212 }
3213 PlanOp::ShortestPath { .. } => {
3214 return Err(
3215 "pull executor: ShortestPath reached pull_rows — shortestPath plans \
3216 must use the staged path (row_bound returns None)"
3217 .to_string(),
3218 );
3219 }
3220 PlanOp::GroupAggregate { .. } => {
3221 return Err(
3222 "pull executor: GroupAggregate reached pull_rows — grouped aggregate plans \
3223 must use the execute_group_aggregate path (routed before pull in execute_inner)"
3224 .to_string(),
3225 );
3226 }
3227 PlanOp::With { .. } => {
3228 return Err(
3229 "pull executor: With reached pull_rows — pipeline plans must use the staged path \
3230 (row_bound returns None for plans containing With)"
3231 .to_string(),
3232 );
3233 }
3234 PlanOp::Unwind { .. } => {
3235 return Err(
3236 "pull executor: Unwind reached pull_rows — pipeline plans must use the staged path \
3237 (row_bound returns None for plans containing Unwind)"
3238 .to_string(),
3239 );
3240 }
3241 PlanOp::LeftOuterApply { .. } => {
3242 return Err(
3243 "pull executor: LeftOuterApply reached pull_rows — OPTIONAL MATCH plans must use \
3244 the staged path (row_bound returns None for plans containing LeftOuterApply)"
3245 .to_string(),
3246 );
3247 }
3248 }
3249 Ok(())
3250}
3251
3252fn eval_expr(
3253 view: &GraphView,
3254 vars: &VarTable,
3255 row: &Row,
3256 expr: &Expr,
3257 params: &Params,
3258 depth: u32,
3259) -> Result<bool, String> {
3260 if depth > 256 {
3261 return Err("expression nesting too deep".into());
3262 }
3263 match expr {
3264 Expr::And(lhs, rhs) => {
3265 let l = eval_expr(view, vars, row, lhs, params, depth + 1)?;
3266 let r = eval_expr(view, vars, row, rhs, params, depth + 1)?;
3267 Ok(l && r)
3268 }
3269 Expr::Or(lhs, rhs) => {
3270 let l = eval_expr(view, vars, row, lhs, params, depth + 1)?;
3271 let r = eval_expr(view, vars, row, rhs, params, depth + 1)?;
3272 Ok(l || r)
3273 }
3274 Expr::Not(inner) => Ok(!eval_expr(view, vars, row, inner, params, depth + 1)?),
3275 Expr::Cmp { lhs, op, rhs } => {
3276 let l = resolve_operand(view, vars, row, lhs, params)?;
3277 let r = resolve_operand(view, vars, row, rhs, params)?;
3278 match (l, r) {
3279 (Some(a), Some(b)) => Ok(eval_cmp(op, &a, &b)),
3280 _ => Ok(false),
3281 }
3282 }
3283 Expr::Truthy(op) => {
3284 let val = resolve_operand(view, vars, row, op, params)?;
3285 Ok(match val {
3286 None => false,
3287 Some(Value::Bool(b)) => b,
3288 Some(Value::Int(n)) => n != 0,
3289 Some(Value::Float(f)) => f != 0.0,
3290 Some(Value::Str(s)) => !s.is_empty(),
3291 Some(Value::List(v)) => !v.is_empty(),
3292 Some(Value::Map(m)) => !m.is_empty(),
3293 })
3294 }
3295 Expr::IsNull(op) => {
3296 let val = resolve_operand(view, vars, row, op, params)?;
3297 Ok(val.is_none())
3298 }
3299 Expr::IsNotNull(op) => {
3300 let val = resolve_operand(view, vars, row, op, params)?;
3301 Ok(val.is_some())
3302 }
3303 Expr::In { expr, list } => eval_in(view, vars, row, expr, list, params),
3304 }
3305}
3306
3307fn eval_in(
3308 view: &GraphView,
3309 vars: &VarTable,
3310 row: &Row,
3311 expr: &Operand,
3312 list: &[Operand],
3313 params: &Params,
3314) -> Result<bool, String> {
3315 let Some(needle) = resolve_operand(view, vars, row, expr, params)? else {
3316 return Ok(false);
3317 };
3318 for item_op in list {
3319 match resolve_operand(view, vars, row, item_op, params)? {
3320 None => {}
3321 Some(Value::List(items)) => {
3322 for item in items {
3323 if crate::filter::eval_cmp(&crate::filter::CmpOp::Eq, &needle, &item) {
3324 return Ok(true);
3325 }
3326 }
3327 }
3328 Some(item) if crate::filter::eval_cmp(&crate::filter::CmpOp::Eq, &needle, &item) => {
3329 return Ok(true);
3330 }
3331 Some(_) => {}
3332 }
3333 }
3334 Ok(false)
3335}
3336
3337fn exec_distinct(table: &mut Projected) -> Result<(), String> {
3339 let cap = max_intermediate_rows();
3340 let mut seen: BTreeSet<Vec<Option<ValueKey>>> = BTreeSet::new();
3341 let mut out = Vec::with_capacity(table.rows.len().min(cap));
3342 for row in table.rows.drain(..) {
3343 let key: Vec<Option<ValueKey>> = row
3344 .iter()
3345 .map(|cell| cell.as_ref().and_then(group_key_normalize))
3346 .collect();
3347 if seen.insert(key) {
3348 if out.len() >= cap {
3349 return Err(row_cap_err(cap));
3350 }
3351 out.push(row);
3352 }
3353 }
3354 table.rows = out;
3355 Ok(())
3356}
3357
3358fn column_name(item: &RetItem) -> String {
3359 if let Some(alias) = &item.alias {
3360 return alias.clone();
3361 }
3362 match &item.value {
3363 RetVal::Var(v) => v.clone(),
3364 RetVal::Prop { var, field } => format!("{var}.{field}"),
3365 RetVal::Agg { func, arg } => {
3368 use crate::cypher::ast::AggArg;
3369 let f = match func {
3370 AggFunc::Count => "COUNT",
3371 AggFunc::Sum => "SUM",
3372 AggFunc::Avg => "AVG",
3373 AggFunc::Min => "MIN",
3374 AggFunc::Max => "MAX",
3375 };
3376 let a = match arg {
3377 AggArg::Star => "*".to_string(),
3378 AggArg::Var(v) => v.clone(),
3379 AggArg::Prop { var, field } => format!("{var}.{field}"),
3380 };
3381 format!("{f}({a})")
3382 }
3383 RetVal::FuncCall { name, args } => {
3384 let arg_strs: Vec<String> = args
3385 .iter()
3386 .map(|a| match a {
3387 Operand::Var(v) => v.clone(),
3388 Operand::Prop { var, field } => format!("{var}.{field}"),
3389 Operand::Lit(_) => "<lit>".to_string(),
3390 Operand::Param(p) => format!("${p}"),
3391 Operand::FuncCall { name: n, .. } => format!("{n}(...)"),
3392 Operand::BinArith { .. } => "<arith>".to_string(),
3393 })
3394 .collect();
3395 format!("{name}({})", arg_strs.join(", "))
3396 }
3397 RetVal::ScalarExpr(_) => "<expr>".to_string(),
3398 }
3399}
3400
3401fn exec_project(
3402 view: &GraphView,
3403 vars: &VarTable,
3404 rows: &[Row],
3405 items: &[RetItem],
3406 params: &Params,
3407) -> Result<Projected, String> {
3408 let columns: Vec<String> = items.iter().map(column_name).collect();
3409 let mut out_rows = Vec::with_capacity(rows.len());
3410 for row in rows {
3411 let mut cells = Vec::with_capacity(items.len());
3412 for item in items {
3413 cells.push(project_item(view, vars, row, item, params)?);
3414 }
3415 out_rows.push(cells);
3416 }
3417 Ok(Projected {
3418 columns,
3419 rows: out_rows,
3420 })
3421}
3422
3423fn project_item(
3424 view: &GraphView,
3425 vars: &VarTable,
3426 row: &Row,
3427 item: &RetItem,
3428 params: &Params,
3429) -> Result<Option<Value>, String> {
3430 match &item.value {
3431 RetVal::Var(v) => {
3432 let slot = vars.slot(v).ok_or_else(|| format!("unbound variable `{v}`"))?;
3434 match row.get(slot).and_then(|c| c.as_ref()) {
3435 None => Ok(None),
3437 Some(Cell::Node(id)) => match view.ids.key_of(*id) {
3438 Some(key) => Ok(Some(Value::Str(key.to_owned()))),
3439 None => Err(format!("unknown node id {id}")),
3440 },
3441 Some(Cell::Scalar(val)) => Ok(Some(val.clone())),
3443 Some(Cell::Rel(_)) => Err(format!(
3444 "variable `{v}` is a relationship; return its properties ({v}.field) instead"
3445 )),
3446 Some(Cell::Path(hops)) => Ok(Some(Value::Int(*hops as i64))),
3447 }
3448 }
3449 RetVal::Prop { var, field } => resolve_prop(view, vars, row, var, field),
3450 RetVal::Agg { .. } => Err(
3453 "project_item: Agg variant reached exec_project — aggregate plans must not contain Project"
3454 .to_string(),
3455 ),
3456 RetVal::FuncCall { name, args } => eval_func(name, args, view, vars, row, params),
3457 RetVal::ScalarExpr(op) => resolve_operand(view, vars, row, op, params),
3458 }
3459}
3460
3461fn order_column(item: &OrderItem) -> String {
3462 match &item.target {
3463 OrderTarget::Alias(name) | OrderTarget::Var(name) => name.clone(),
3464 OrderTarget::Prop { var, field } => format!("{var}.{field}"),
3465 }
3466}
3467
3468fn exec_order_by(table: &mut Projected, items: &[OrderItem]) -> Result<(), String> {
3469 let mut keys = Vec::with_capacity(items.len());
3470 for item in items {
3471 let name = order_column(item);
3472 let idx = table
3473 .columns
3474 .iter()
3475 .position(|c| c == &name)
3476 .ok_or_else(|| format!("ORDER BY target `{name}` is not a projected column"))?;
3477 keys.push((idx, item.descending));
3478 }
3479 table.rows.sort_by(|a, b| {
3480 for &(idx, desc) in &keys {
3481 let c = cmp_optional(
3482 a.get(idx).and_then(|x| x.as_ref()),
3483 b.get(idx).and_then(|x| x.as_ref()),
3484 desc,
3485 );
3486 if c != std::cmp::Ordering::Equal {
3487 return c;
3488 }
3489 }
3490 std::cmp::Ordering::Equal
3491 });
3492 Ok(())
3493}
3494
3495fn resolve_ls(ls: &LimitSkip, params: &Params) -> Result<u64, String> {
3500 match ls {
3501 LimitSkip::Exact(n) => Ok(*n),
3502 LimitSkip::Param(name) => {
3503 let val = params
3504 .0
3505 .get(name)
3506 .ok_or_else(|| format!("missing parameter `{name}` (used in LIMIT/SKIP)"))?;
3507 match val {
3508 Value::Int(i) if *i >= 0 => Ok(*i as u64),
3509 Value::Int(i) => Err(format!(
3510 "LIMIT/SKIP parameter `{name}` must be a non-negative integer, got {i}"
3511 )),
3512 other => Err(format!(
3513 "LIMIT/SKIP parameter `{name}` must be an integer, got {other:?}"
3514 )),
3515 }
3516 }
3517 }
3518}
3519
3520fn apply_skip<T>(rows: &mut Vec<T>, n: u64) {
3521 let n = usize::try_from(n).unwrap_or(usize::MAX);
3522 if n >= rows.len() {
3523 rows.clear();
3524 } else {
3525 rows.drain(0..n);
3526 }
3527}
3528
3529fn apply_limit<T>(rows: &mut Vec<T>, n: u64) {
3530 let n = usize::try_from(n).unwrap_or(usize::MAX);
3531 rows.truncate(n);
3532}
3533
3534#[cfg(test)]
3535mod tests {
3536 use super::{execute, resolve_operand, Params, Row, VarTable};
3537 use crate::cypher::ast::{
3538 ArithOp, LimitSkip, Operand, OrderItem, OrderTarget, RetItem, RetVal,
3539 };
3540 use crate::cypher::plan::{plan, PlanOp};
3541 use crate::cypher::{lex, parse, RelDir};
3542 use crate::result::ResultSet;
3543 use crate::view::GraphView;
3544 use core_storage::{ColumnStore, EdgeProps, IdMap, Interner, Topology, Value};
3545 use proptest::prelude::*;
3546 use std::collections::BTreeMap;
3547
3548 struct Fx {
3549 ids: IdMap,
3550 syms: Interner,
3551 labels: Vec<u32>,
3552 props: ColumnStore,
3553 topo: Topology,
3554 eprops: EdgeProps,
3555 }
3556
3557 impl Fx {
3558 fn new() -> Self {
3559 Fx {
3560 ids: IdMap::new(),
3561 syms: Interner::new(),
3562 labels: vec![],
3563 props: ColumnStore::new(),
3564 topo: Topology::new(),
3565 eprops: EdgeProps::new(),
3566 }
3567 }
3568
3569 fn add(&mut self, label: &str, key: &str, props: Vec<(&str, Value)>) -> u32 {
3570 let id = self.ids.get_or_insert(key);
3571 let sym = self.syms.intern(label);
3572 self.labels.resize(id as usize + 1, u32::MAX);
3573 self.labels[id as usize] = sym;
3574 for (f, v) in props {
3575 self.props.set(id, f, v);
3576 }
3577 id
3578 }
3579
3580 fn edge(&mut self, etype: &str, src: u32, dst: u32, props: Vec<(&str, Value)>) {
3581 let et = self.syms.intern(etype);
3582 self.topo.add_edge(et, src, dst);
3583 for (f, v) in props {
3584 self.eprops.set(et, src, dst, f, v);
3585 }
3586 }
3587
3588 fn view(&self) -> GraphView<'_> {
3589 GraphView {
3590 ids: &self.ids,
3591 syms: &self.syms,
3592 labels: &self.labels,
3593 props: &self.props,
3594 topo: &self.topo,
3595 edge_props: &self.eprops,
3596 mask: None,
3597 }
3598 }
3599 }
3600
3601 fn compile(src: &str) -> Vec<PlanOp> {
3602 plan(&parse(&lex(src).expect("lex")).expect("parse")).expect("plan")
3603 }
3604
3605 fn run(
3606 view: &GraphView,
3607 src: &str,
3608 params: &BTreeMap<String, Value>,
3609 ) -> Result<ResultSet, String> {
3610 execute(view, &compile(src), &Params(params))
3611 }
3612
3613 fn s(v: &str) -> Value {
3614 Value::Str(v.into())
3615 }
3616
3617 fn f(v: f64) -> Value {
3618 Value::Float(v)
3619 }
3620
3621 fn i(v: i64) -> Value {
3622 Value::Int(v)
3623 }
3624
3625 fn rows_of(rs: &ResultSet) -> Vec<Vec<Option<Value>>> {
3626 (0..rs.len()).map(|i| rs.row(i).to_vec()).collect()
3627 }
3628
3629 fn col(rs: &ResultSet, name: &str) -> Vec<Option<Value>> {
3630 (0..rs.len()).map(|i| rs.get(i, name).cloned()).collect()
3631 }
3632
3633 fn hop_graph() -> Fx {
3634 let mut fx = Fx::new();
3635 let ada = fx.add("Person", "ada", vec![]);
3636 let bob = fx.add("Person", "bob", vec![]);
3637 let cam = fx.add("Person", "cam", vec![]);
3638 let acme = fx.add("Company", "acme", vec![]);
3639 fx.edge("KNOWS", ada, bob, vec![]);
3640 fx.edge("KNOWS", ada, cam, vec![]);
3641 fx.edge("KNOWS", bob, cam, vec![]);
3642 fx.edge("LIKES", ada, acme, vec![]);
3643 fx
3644 }
3645
3646 fn undirected_graph() -> Fx {
3647 let mut fx = Fx::new();
3648 let a = fx.add("N", "a", vec![]);
3649 let b = fx.add("N", "b", vec![]);
3650 fx.edge("T", a, b, vec![("w", i(42))]);
3651 fx
3652 }
3653
3654 fn triangle() -> Fx {
3655 let mut fx = Fx::new();
3656 let a = fx.add("N", "a", vec![]);
3657 let b = fx.add("N", "b", vec![]);
3658 let c = fx.add("N", "c", vec![]);
3659 fx.edge("T", a, b, vec![("eid", i(1))]);
3660 fx.edge("T", b, c, vec![("eid", i(2))]);
3661 fx.edge("T", c, a, vec![("eid", i(3))]);
3662 fx
3663 }
3664
3665 fn single_edge() -> (Fx, u32, u32) {
3666 let mut fx = Fx::new();
3667 let a = fx.add("N", "a", vec![]);
3668 let b = fx.add("N", "b", vec![]);
3669 fx.edge("T", a, b, vec![]);
3670 (fx, a, b)
3671 }
3672
3673 fn dogfood_graph() -> Fx {
3675 let mut fx = Fx::new();
3676 let t1 = fx.add("Talent", "t1", vec![("id", s("t1"))]);
3677 let acme = fx.add("Company", "acme", vec![]);
3678 let beta = fx.add("Company", "beta", vec![]);
3679 let gamma = fx.add("Company", "gamma", vec![]);
3680 let delta = fx.add("Company", "delta", vec![]);
3681 let echo = fx.add("Company", "echo", vec![]);
3682 let foxtrot = fx.add("Company", "foxtrot", vec![]);
3683 let zeta = fx.add("Company", "zeta", vec![]);
3684 fx.edge("INDUSTRY_ALIGNMENT", acme, t1, vec![("score", f(0.9))]);
3685 fx.edge("SPECIALTY_MATCH", acme, t1, vec![("score", f(0.8))]);
3686 fx.edge("INDUSTRY_ALIGNMENT", beta, t1, vec![("score", f(0.6))]);
3687 fx.edge("SPECIALTY_MATCH", beta, t1, vec![("score", f(0.7))]);
3688 fx.edge("INDUSTRY_ALIGNMENT", gamma, t1, vec![("score", f(0.4))]);
3689 fx.edge("SPECIALTY_MATCH", gamma, t1, vec![("score", f(0.9))]);
3690 fx.edge("INDUSTRY_ALIGNMENT", delta, t1, vec![("score", f(0.8))]);
3691 fx.edge("SPECIALTY_MATCH", delta, t1, vec![("score", f(0.3))]);
3692 fx.edge("INDUSTRY_ALIGNMENT", echo, t1, vec![("score", f(0.5))]);
3693 fx.edge("SPECIALTY_MATCH", echo, t1, vec![("score", f(0.5))]);
3694 fx.edge("INDUSTRY_ALIGNMENT", foxtrot, t1, vec![("score", f(0.95))]);
3695 fx.edge("INDUSTRY_ALIGNMENT", zeta, t1, vec![("score", f(0.9))]);
3696 fx.edge("SPECIALTY_MATCH", zeta, t1, vec![("score", f(0.6))]);
3697 fx
3698 }
3699
3700 const DOGFOOD: &str = "\
3701MATCH (t:Talent {id: $tid}) \
3702MATCH (c:Company)-[i:INDUSTRY_ALIGNMENT]->(t) \
3703MATCH (c)-[s:SPECIALTY_MATCH]->(t) \
3704WHERE i.score >= 0.5 AND s.score >= 0.5 \
3705RETURN c, i.score AS industry, s.score AS specialty \
3706ORDER BY industry DESC, specialty DESC \
3707LIMIT 10";
3708
3709 fn tid_params() -> BTreeMap<String, Value> {
3710 let mut p = BTreeMap::new();
3711 p.insert("tid".into(), s("t1"));
3712 p
3713 }
3714
3715 #[test]
3716 fn single_hop_match_label_and_etype_filters() {
3717 let fx = hop_graph();
3718 let v = fx.view();
3719 let rs = run(
3720 &v,
3721 "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b",
3722 &BTreeMap::new(),
3723 )
3724 .expect("single-hop");
3725 assert_eq!(rs.columns(), &["a".to_string(), "b".to_string()]);
3726 assert_eq!(
3727 rows_of(&rs),
3728 vec![
3729 vec![Some(s("ada")), Some(s("bob"))],
3730 vec![Some(s("ada")), Some(s("cam"))],
3731 vec![Some(s("bob")), Some(s("cam"))],
3732 ]
3733 );
3734 let likes = run(
3736 &v,
3737 "MATCH (a:Person)-[:LIKES]->(b:Company) RETURN a, b",
3738 &BTreeMap::new(),
3739 )
3740 .unwrap();
3741 assert_eq!(rows_of(&likes), vec![vec![Some(s("ada")), Some(s("acme"))]]);
3742 let no_combo = run(
3743 &v,
3744 "MATCH (a:Person)-[:KNOWS]->(b:Company) RETURN a, b",
3745 &BTreeMap::new(),
3746 )
3747 .unwrap();
3748 assert!(no_combo.is_empty());
3749 }
3750
3751 #[test]
3752 fn undirected_match_finds_both_orientations_and_binds_true_triple() {
3753 let fx = undirected_graph();
3754 let v = fx.view();
3755 let rs =
3758 run(&v, "MATCH (x)-[r:T]-(y) RETURN x, y, r.w", &BTreeMap::new()).expect("undirected");
3759 assert_eq!(
3760 rows_of(&rs),
3761 vec![
3762 vec![Some(s("a")), Some(s("b")), Some(i(42))],
3763 vec![Some(s("b")), Some(s("a")), Some(i(42))],
3764 ]
3765 );
3766
3767 let left = run(
3768 &v,
3769 "MATCH (y)<-[r:T]-(x) RETURN x, y, r.w",
3770 &BTreeMap::new(),
3771 )
3772 .unwrap();
3773 assert_eq!(
3774 rows_of(&left),
3775 vec![vec![Some(s("a")), Some(s("b")), Some(i(42))]]
3776 );
3777 }
3778
3779 #[test]
3780 fn relationship_uniqueness_triangle_and_two_hop_cycle() {
3781 let tri = triangle();
3782 let v = tri.view();
3783 let rs = run(
3784 &v,
3785 "MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z, r1.eid, r2.eid",
3786 &BTreeMap::new(),
3787 )
3788 .expect("triangle 2-hop");
3789 assert_eq!(
3790 rows_of(&rs),
3791 vec![
3792 vec![
3793 Some(s("a")),
3794 Some(s("b")),
3795 Some(s("c")),
3796 Some(i(1)),
3797 Some(i(2))
3798 ],
3799 vec![
3800 Some(s("b")),
3801 Some(s("c")),
3802 Some(s("a")),
3803 Some(i(2)),
3804 Some(i(3))
3805 ],
3806 vec![
3807 Some(s("c")),
3808 Some(s("a")),
3809 Some(s("b")),
3810 Some(i(3)),
3811 Some(i(1))
3812 ],
3813 ]
3814 );
3815 for row in rows_of(&rs) {
3816 assert_ne!(row[3], row[4], "r1 must never bind the same edge as r2");
3817 }
3818
3819 let (mut one, a, b) = single_edge();
3820 let v = one.view();
3821 let cycle = run(
3822 &v,
3823 "MATCH (x)-[:T]->(y)-[:T]->(x) RETURN x",
3824 &BTreeMap::new(),
3825 )
3826 .expect("2-hop cycle");
3827 assert!(
3828 cycle.is_empty(),
3829 "single directed edge cannot close a 2-hop cycle"
3830 );
3831
3832 let undirected_cycle = run(&v, "MATCH (x)-[:T]-(y)-[:T]-(x) RETURN x", &BTreeMap::new())
3834 .expect("undirected uniqueness");
3835 assert!(
3836 undirected_cycle.is_empty(),
3837 "relationship uniqueness must reject walking the same triple back"
3838 );
3839
3840 one.edge("T", b, a, vec![]);
3841 let v = one.view();
3842 let with_recip = run(
3843 &v,
3844 "MATCH (x)-[:T]->(y)-[:T]->(x) RETURN x",
3845 &BTreeMap::new(),
3846 )
3847 .unwrap();
3848 assert_eq!(col(&with_recip, "x"), vec![Some(s("a")), Some(s("b"))]);
3849 }
3850
3851 #[test]
3852 fn multi_match_join_bound_and_bound_destination_expand() {
3853 let fx = dogfood_graph();
3854 let v = fx.view();
3855 let rs = run(
3858 &v,
3859 "MATCH (t:Talent {id: $tid}) \
3860 MATCH (c:Company)-[i:INDUSTRY_ALIGNMENT]->(t) \
3861 MATCH (c)-[s:SPECIALTY_MATCH]->(t) \
3862 RETURN c",
3863 &tid_params(),
3864 )
3865 .expect("join + bound dest");
3866 assert_eq!(
3867 col(&rs, "c"),
3868 vec![
3869 Some(s("acme")),
3870 Some(s("beta")),
3871 Some(s("gamma")),
3872 Some(s("delta")),
3873 Some(s("echo")),
3874 Some(s("zeta")),
3875 ]
3876 );
3877
3878 let keep = run(&v, "MATCH (c:Company) MATCH (c) RETURN c", &BTreeMap::new()).unwrap();
3880 assert_eq!(keep.len(), 7);
3881 let drop = run(
3883 &v,
3884 "MATCH (c:Company) MATCH (c:Talent) RETURN c",
3885 &BTreeMap::new(),
3886 )
3887 .unwrap();
3888 assert!(drop.is_empty());
3889 }
3890
3891 #[test]
3892 fn scan_key_exec_does_not_use_label_scan() {
3893 let mut fx = Fx::new();
3894 fx.add("Person", "p1", vec![]);
3895 fx.add("Person", "p2", vec![]);
3896 fx.add("Person", "p3", vec![]);
3897 fx.add("Company", "c1", vec![]);
3898 let v = fx.view();
3899 let params = BTreeMap::new();
3900
3901 let fires_before = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
3902 let rs = run(&v, "MATCH (n:Person {id: 'p2'}) RETURN n", ¶ms).expect("scan key");
3903 let fires_after = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
3904 assert!(
3905 fires_after > fires_before,
3906 "SCAN_KEY_FIRES must increment; before={fires_before} after={fires_after}"
3907 );
3908 assert_eq!(rows_of(&rs), vec![vec![Some(s("p2"))]]);
3909
3910 let miss = run(&v, "MATCH (n:Person {id: 'nope'}) RETURN n", ¶ms).unwrap();
3911 assert!(
3912 miss.is_empty(),
3913 "missing key must be zero rows, not an error"
3914 );
3915
3916 let wrong = run(&v, "MATCH (n:Person {id: 'c1'}) RETURN n", ¶ms).unwrap();
3917 assert!(
3918 wrong.is_empty(),
3919 "wrong label must be zero rows, not an error"
3920 );
3921 }
3922
3923 #[test]
3924 fn rel_var_edge_prop_filter() {
3925 let mut fx = Fx::new();
3926 let a = fx.add("N", "a", vec![]);
3927 let b = fx.add("N", "b", vec![]);
3928 let c = fx.add("N", "c", vec![]);
3929 fx.edge("T", a, b, vec![("w", f(0.7))]);
3930 fx.edge("T", a, c, vec![("w", f(0.3))]);
3931 let v = fx.view();
3932 let rs = run(
3933 &v,
3934 "MATCH (x)-[r:T]->(y) WHERE r.w >= 0.5 RETURN y, r.w",
3935 &BTreeMap::new(),
3936 )
3937 .expect("edge-prop filter");
3938 assert_eq!(rows_of(&rs), vec![vec![Some(s("b")), Some(f(0.7))]]);
3939 let fail = run(
3940 &v,
3941 "MATCH (x)-[r:T]->(y) WHERE r.w >= 0.8 RETURN y",
3942 &BTreeMap::new(),
3943 )
3944 .unwrap();
3945 assert!(fail.is_empty());
3946 }
3947
3948 #[test]
3949 fn params_present_resolve_missing_is_err_before_rows() {
3950 let fx = dogfood_graph();
3951 let v = fx.view();
3952 let hit =
3953 run(&v, "MATCH (t:Talent {id: $tid}) RETURN t", &tid_params()).expect("present param");
3954 assert_eq!(col(&hit, "t"), vec![Some(s("t1"))]);
3955
3956 let err = run(
3958 &v,
3959 "MATCH (t:NoSuchLabel {id: $tid}) RETURN t",
3960 &BTreeMap::new(),
3961 )
3962 .expect_err("missing param must be Err, not Ok(empty)");
3963 assert!(
3964 err.contains("tid")
3965 && (err.contains("param") || err.contains("Param") || err.contains("missing")),
3966 "missing-param error must name the parameter, got: {err}"
3967 );
3968
3969 let err = run(
3970 &v,
3971 "MATCH (t:Talent) WHERE t.id = $tid RETURN t",
3972 &BTreeMap::new(),
3973 )
3974 .expect_err("missing WHERE param");
3975 assert!(err.contains("tid"), "got: {err}");
3976 }
3977
3978 #[test]
3979 fn order_by_none_last_then_skip_limit() {
3980 let mut fx = Fx::new();
3981 fx.add("Person", "ada", vec![("age", i(30))]);
3982 fx.add("Person", "bob", vec![]); fx.add("Person", "cam", vec![("age", i(10))]);
3984 fx.add("Person", "dan", vec![("age", i(20))]);
3985 let v = fx.view();
3986
3987 let asc = run(
3988 &v,
3989 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age",
3990 &BTreeMap::new(),
3991 )
3992 .unwrap();
3993 assert_eq!(
3994 rows_of(&asc),
3995 vec![
3996 vec![Some(s("cam")), Some(i(10))],
3997 vec![Some(s("dan")), Some(i(20))],
3998 vec![Some(s("ada")), Some(i(30))],
3999 vec![Some(s("bob")), None],
4000 ]
4001 );
4002
4003 let desc = run(
4004 &v,
4005 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age DESC",
4006 &BTreeMap::new(),
4007 )
4008 .unwrap();
4009 assert_eq!(
4010 rows_of(&desc),
4011 vec![
4012 vec![Some(s("ada")), Some(i(30))],
4013 vec![Some(s("dan")), Some(i(20))],
4014 vec![Some(s("cam")), Some(i(10))],
4015 vec![Some(s("bob")), None],
4016 ]
4017 );
4018
4019 let skip_lim = run(
4020 &v,
4021 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age SKIP 1 LIMIT 2",
4022 &BTreeMap::new(),
4023 )
4024 .unwrap();
4025 assert_eq!(
4026 rows_of(&skip_lim),
4027 vec![
4028 vec![Some(s("dan")), Some(i(20))],
4029 vec![Some(s("ada")), Some(i(30))],
4030 ]
4031 );
4032
4033 let desc_sl = run(
4034 &v,
4035 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age DESC SKIP 1 LIMIT 2",
4036 &BTreeMap::new(),
4037 )
4038 .unwrap();
4039 assert_eq!(
4040 rows_of(&desc_sl),
4041 vec![
4042 vec![Some(s("dan")), Some(i(20))],
4043 vec![Some(s("cam")), Some(i(10))],
4044 ]
4045 );
4046 }
4047
4048 #[test]
4049 fn unknown_label_and_etype_are_ok_empty() {
4050 let fx = hop_graph();
4051 let v = fx.view();
4052 let lab = run(&v, "MATCH (x:Nope) RETURN x", &BTreeMap::new()).expect("unknown label");
4053 assert!(lab.is_empty());
4054 let et = run(
4055 &v,
4056 "MATCH (a)-[:NO_SUCH_ETYPE]->(b) RETURN a",
4057 &BTreeMap::new(),
4058 )
4059 .expect("unknown etype");
4060 assert!(et.is_empty());
4061 }
4062
4063 #[test]
4064 fn execute_is_deterministic() {
4065 let fx = dogfood_graph();
4066 let v = fx.view();
4067 let p = tid_params();
4068 let a = run(&v, DOGFOOD, &p).expect("first");
4069 let b = run(&v, DOGFOOD, &p).expect("second");
4070 assert_eq!(a, b);
4071 let hop = hop_graph();
4072 let hv = hop.view();
4073 let q = "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b";
4074 assert_eq!(run(&hv, q, &BTreeMap::new()), run(&hv, q, &BTreeMap::new()));
4075 }
4076
4077 #[test]
4078 fn dogfood_pipeline_exact_rows() {
4079 let fx = dogfood_graph();
4080 let v = fx.view();
4081 let rs = run(&v, DOGFOOD, &tid_params()).expect("dogfood");
4082 assert_eq!(
4083 rs.columns(),
4084 &[
4085 "c".to_string(),
4086 "industry".to_string(),
4087 "specialty".to_string()
4088 ]
4089 );
4090 assert_eq!(
4092 rows_of(&rs),
4093 vec![
4094 vec![Some(s("acme")), Some(f(0.9)), Some(f(0.8))],
4095 vec![Some(s("zeta")), Some(f(0.9)), Some(f(0.6))],
4096 vec![Some(s("beta")), Some(f(0.6)), Some(f(0.7))],
4097 vec![Some(s("echo")), Some(f(0.5)), Some(f(0.5))],
4098 ]
4099 );
4100 }
4101
4102 #[test]
4103 fn unknown_var_in_op_is_err_not_panic() {
4104 let fx = hop_graph();
4105 let v = fx.view();
4106 let plan = vec![
4107 PlanOp::ScanLabel {
4108 var: "a".into(),
4109 label: None,
4110 },
4111 PlanOp::Expand {
4112 from: "zzz".into(),
4113 rel_var: Some("r".into()),
4114 etype: None,
4115 dir: RelDir::Right,
4116 to: "b".into(),
4117 to_label: None,
4118 to_props: vec![],
4119 },
4120 PlanOp::Project {
4121 items: vec![RetItem {
4122 value: RetVal::Var("a".into()),
4123 alias: None,
4124 }],
4125 },
4126 ];
4127 let params = BTreeMap::new();
4128 let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
4129 execute(&v, &plan, &Params(¶ms))
4130 }));
4131 assert!(caught.is_ok(), "execute panicked on unknown var");
4132 let err = caught.unwrap().expect_err("unknown var must be Err");
4133 assert!(
4134 err.contains("zzz") && err.to_ascii_lowercase().contains("unbound"),
4135 "got: {err}"
4136 );
4137
4138 let join = vec![
4139 PlanOp::JoinBound {
4140 var: "ghost".into(),
4141 label: None,
4142 props: vec![],
4143 },
4144 PlanOp::Project {
4145 items: vec![RetItem {
4146 value: RetVal::Var("ghost".into()),
4147 alias: None,
4148 }],
4149 },
4150 ];
4151 let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
4152 execute(&v, &join, &Params(¶ms))
4153 }));
4154 assert!(caught.is_ok(), "execute panicked on JoinBound unknown var");
4155 assert!(caught.unwrap().is_err());
4156 }
4157
4158 #[test]
4159 fn dest_props_on_bound_expand_are_applied() {
4160 let fx = dogfood_graph();
4161 let v = fx.view();
4162 let rs = run(
4163 &v,
4164 "MATCH (t:Talent {id: $tid}) \
4165 MATCH (c:Company)-[r:INDUSTRY_ALIGNMENT]->(t:Talent {id: $tid}) \
4166 RETURN c",
4167 &tid_params(),
4168 )
4169 .unwrap();
4170 assert_eq!(
4172 col(&rs, "c"),
4173 vec![
4174 Some(s("acme")),
4175 Some(s("beta")),
4176 Some(s("gamma")),
4177 Some(s("delta")),
4178 Some(s("echo")),
4179 Some(s("foxtrot")),
4180 Some(s("zeta")),
4181 ]
4182 );
4183 let miss = run(
4184 &v,
4185 "MATCH (t:Talent {id: $tid}) \
4186 MATCH (c:Company)-[r:INDUSTRY_ALIGNMENT]->(t {id: 'nope'}) \
4187 RETURN c",
4188 &tid_params(),
4189 )
4190 .unwrap();
4191 assert!(miss.is_empty());
4192 }
4193
4194 #[test]
4195 fn missing_node_prop_projects_none_and_pattern_misses() {
4196 let mut fx = Fx::new();
4197 fx.add("Person", "ada", vec![("age", i(30))]);
4198 fx.add("Person", "bob", vec![]);
4199 let v = fx.view();
4200 let rs = run(&v, "MATCH (p:Person) RETURN p.age", &BTreeMap::new()).unwrap();
4201 assert_eq!(col(&rs, "p.age"), vec![Some(i(30)), None]);
4202 let pat = run(&v, "MATCH (p:Person {age: 30}) RETURN p", &BTreeMap::new()).unwrap();
4203 assert_eq!(col(&pat, "p"), vec![Some(s("ada"))]);
4204 }
4205
4206 #[test]
4207 fn unlabeled_match_does_not_project_sentinel_ghost_key() {
4208 let mut fx = Fx::new();
4209 fx.add("Person", "ada", vec![]);
4210 fx.ids.get_or_insert("ghost");
4213 fx.labels.resize(fx.ids.len(), u32::MAX);
4214 fx.add("Person", "bob", vec![]);
4215 let v = fx.view();
4216 let rs = run(&v, "MATCH (n) RETURN n", &BTreeMap::new()).expect("unlabeled scan");
4217 assert_eq!(col(&rs, "n"), vec![Some(s("ada")), Some(s("bob"))]);
4218 assert!(
4219 !rows_of(&rs)
4220 .iter()
4221 .any(|row| row.iter().any(|c| *c == Some(s("ghost")))),
4222 "sentinel slot must not project a ghost key"
4223 );
4224 }
4225
4226 #[test]
4227 fn execute_never_panics_on_hostile_plans() {
4228 let fx = hop_graph();
4229 let v = fx.view();
4230 let params = BTreeMap::new();
4231 let hostile = vec![
4232 vec![],
4233 vec![PlanOp::Project { items: vec![] }],
4234 vec![PlanOp::OrderBy {
4235 items: vec![OrderItem {
4236 target: OrderTarget::Alias("nope".into()),
4237 descending: false,
4238 }],
4239 }],
4240 vec![PlanOp::Filter {
4241 expr: crate::cypher::ast::Expr::Cmp {
4242 lhs: Operand::Prop {
4243 var: "missing".into(),
4244 field: "x".into(),
4245 },
4246 op: crate::filter::CmpOp::Eq,
4247 rhs: Operand::Lit(i(1)),
4248 },
4249 }],
4250 vec![
4251 PlanOp::ScanLabel {
4252 var: "a".into(),
4253 label: None,
4254 },
4255 PlanOp::LookupProps {
4256 var: "zzz".into(),
4257 props: vec![("k".into(), Operand::Lit(i(1)))],
4258 },
4259 ],
4260 vec![
4261 PlanOp::Skip(LimitSkip::Exact(99)),
4262 PlanOp::Limit(LimitSkip::Exact(0)),
4263 ],
4264 ];
4265 for plan in hostile {
4266 let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
4267 execute(&v, &plan, &Params(¶ms))
4268 }));
4269 assert!(caught.is_ok(), "execute panicked on hostile plan: {plan:?}");
4270 }
4271 }
4272
4273 #[test]
4274 fn unlabeled_and_labeled_scans_skip_tombstoned_ids() {
4275 let mut fx = Fx::new();
4276 let ada = fx.add("Person", "ada", vec![]);
4277 let bob = fx.add("Person", "bob", vec![]);
4278 fx.edge("KNOWS", ada, bob, vec![]);
4279 fx.ids.delete("ada");
4281 fx.labels[ada as usize] = u32::MAX;
4282 let knows = fx.syms.get("KNOWS").unwrap();
4283 fx.topo.remove_edge(knows, ada, bob);
4284 let v = fx.view();
4285
4286 let labeled = run(&v, "MATCH (p:Person) RETURN p", &BTreeMap::new()).unwrap();
4287 assert_eq!(col(&labeled, "p"), vec![Some(s("bob"))]);
4288 let unlabeled = run(&v, "MATCH (n) RETURN n", &BTreeMap::new()).unwrap();
4289 assert_eq!(col(&unlabeled, "n"), vec![Some(s("bob"))]);
4290 let hop = run(&v, "MATCH (x)-[:KNOWS]->(y) RETURN x, y", &BTreeMap::new()).unwrap();
4291 assert!(
4292 hop.is_empty(),
4293 "expand cannot yield edges to a deleted node once topology is swept"
4294 );
4295 }
4296
4297 proptest! {
4302 #[test]
4315 fn prop_bounded_equals_unbounded_slice(
4316 n_nodes in 2u32..10u32,
4317 edge_pairs in proptest::collection::vec(
4318 (any::<u32>(), any::<u32>()), 0..20usize
4319 ),
4320 recip_pairs in proptest::collection::vec(
4322 (any::<u32>(), any::<u32>()), 0..8usize
4323 ),
4324 n_hops in 1u32..4u32, use_filter in any::<bool>(),
4326 threshold in 0i64..8i64, limit in 1u64..10u64,
4328 skip in 0u64..4u64,
4329 ) {
4330 let mut fx = Fx::new();
4331 let mut node_ids = Vec::new();
4332 for idx in 0..n_nodes {
4333 let id = fx.add("N", &format!("n{idx}"), vec![("v", i(idx as i64 % 8))]);
4335 node_ids.push(id);
4336 }
4337 let n = node_ids.len();
4338
4339 for (si, di) in &edge_pairs {
4341 let si = (*si as usize) % n;
4342 let di = (*di as usize) % n;
4343 if si != di {
4344 fx.edge("T", node_ids[si], node_ids[di], vec![]);
4345 }
4346 }
4347 for (si, di) in &recip_pairs {
4350 let si = (*si as usize) % n;
4351 let di = (*di as usize) % n;
4352 if si != di {
4353 fx.edge("T", node_ids[si], node_ids[di], vec![]);
4354 fx.edge("T", node_ids[di], node_ids[si], vec![]);
4355 }
4356 }
4357
4358 let v = fx.view();
4359 let params = BTreeMap::new();
4360
4361 let var_names = ["a", "b", "c", "d"];
4364 let hop_count = n_hops as usize;
4365 let mut pattern = format!("({}:N)", var_names[0]);
4366 for h in 0..hop_count {
4367 pattern.push_str(&format!("-[:T]->({}", var_names[h + 1]));
4368 if h + 1 == hop_count {
4370 pattern.push_str(":N)");
4371 } else {
4372 pattern.push(')');
4373 }
4374 }
4375 let last_var = var_names[hop_count];
4376 let where_clause = if use_filter {
4377 format!(" WHERE {last_var}.v > {threshold}")
4378 } else {
4379 String::new()
4380 };
4381 let ret_vars: Vec<&str> = var_names[..=hop_count].to_vec();
4382 let ret_clause = ret_vars.join(", ");
4383
4384 let full_q = format!("MATCH {pattern}{where_clause} RETURN {ret_clause}");
4385 let bounded_q = format!(
4386 "MATCH {pattern}{where_clause} RETURN {ret_clause} SKIP {skip} LIMIT {limit}"
4387 );
4388
4389 let full_plan = compile(&full_q);
4390 let unbounded = super::with_max_intermediate_rows(100_000, || {
4393 super::execute_unbounded(&v, &full_plan, &Params(¶ms))
4394 });
4395 let unbounded = match unbounded {
4398 Ok(rs) => rs,
4399 Err(_) => return Ok(()),
4400 };
4401 let total = unbounded.len();
4402 let full_rows = rows_of(&unbounded);
4403
4404 let bounded = super::with_max_intermediate_rows(100_000, || {
4405 run(&v, &bounded_q, ¶ms)
4406 }).expect("bounded must not error");
4407
4408 let s = (skip as usize).min(total);
4409 let e = (skip as usize + limit as usize).min(total);
4410 prop_assert_eq!(
4411 rows_of(&bounded),
4412 full_rows[s..e].to_vec(),
4413 "hops={} filter={} threshold={} SKIP {} LIMIT {}: \
4414 bounded != unbounded[{}..{}]",
4415 hop_count, use_filter, threshold, skip, limit, s, e
4416 );
4417 }
4418 }
4419
4420 proptest! {
4421 #[test]
4439 fn prop_scan_filter_fused_equals_unbounded_slice(
4440 n_nodes in 0u32..20u32,
4441 prop_mask in any::<u32>(),
4443 float_mask in any::<u32>(),
4445 threshold in -1i64..8i64,
4448 op_idx in 0u32..6u32,
4450 skip in 0u64..5u64,
4451 limit in 1u64..8u64,
4452 ) {
4453 let op_str = match op_idx {
4454 0 => "=",
4455 1 => "<>",
4456 2 => "<",
4457 3 => "<=",
4458 4 => ">",
4459 _ => ">=",
4460 };
4461
4462 let mut fx = Fx::new();
4463 for idx in 0..n_nodes {
4464 let has_prop = (prop_mask >> (idx % 32)) & 1 == 1;
4465 let use_float = (float_mask >> (idx % 32)) & 1 == 1;
4466 let props: Vec<(&str, Value)> = if has_prop {
4468 if use_float {
4469 vec![("v", f(idx as f64 % 7.0))]
4470 } else {
4471 vec![("v", i(idx as i64 % 7))]
4472 }
4473 } else {
4474 vec![]
4475 };
4476 fx.add("N", &format!("n{idx}"), props);
4477 }
4478
4479 let v = fx.view();
4480 let params = BTreeMap::new();
4481
4482 let full_q = format!("MATCH (n:N) WHERE n.v {op_str} {threshold} RETURN n");
4484 let bounded_q = format!(
4485 "MATCH (n:N) WHERE n.v {op_str} {threshold} RETURN n SKIP {skip} LIMIT {limit}"
4486 );
4487
4488 let full_plan = compile(&full_q);
4489 let unbounded = super::execute_unbounded(&v, &full_plan, &Params(¶ms));
4490 let unbounded = match unbounded {
4491 Ok(rs) => rs,
4492 Err(_) => return Ok(()),
4493 };
4494 let total = unbounded.len();
4495 let full_rows = rows_of(&unbounded);
4496
4497 let fires_before =
4499 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
4500 let bounded = run(&v, &bounded_q, ¶ms).expect("fused bounded must not error");
4501 let fires_after =
4502 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
4503
4504 if n_nodes > 0 {
4509 prop_assert!(
4510 fires_after > fires_before,
4511 "fused arm did NOT fire for op={} threshold={} n_nodes={}: \
4512 counter before={} after={}",
4513 op_str,
4514 threshold,
4515 n_nodes,
4516 fires_before,
4517 fires_after
4518 );
4519 }
4520
4521 let s = (skip as usize).min(total);
4522 let e = (skip as usize + limit as usize).min(total);
4523 prop_assert_eq!(
4524 rows_of(&bounded),
4525 full_rows[s..e].to_vec(),
4526 "fused path: op={} threshold={} n_nodes={} SKIP {} LIMIT {}: \
4527 bounded != unbounded[{}..{}]",
4528 op_str, threshold, n_nodes, skip, limit, s, e
4529 );
4530
4531 let compound_q = format!(
4535 "MATCH (n:N) WHERE n.v {} {} AND n.v >= -999 RETURN n SKIP {} LIMIT {}",
4536 op_str, threshold, skip, limit
4537 );
4538 let fires_before_compound =
4539 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
4540 let compound_bounded =
4541 run(&v, &compound_q, ¶ms).expect("compound-AND bounded must not error");
4542 let fires_after_compound =
4543 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
4544
4545 prop_assert_eq!(
4547 fires_after_compound,
4548 fires_before_compound,
4549 "fused arm fired for compound-AND shape (should use generic path): \
4550 op={} threshold={} n_nodes={}",
4551 op_str,
4552 threshold,
4553 n_nodes
4554 );
4555 prop_assert_eq!(
4556 rows_of(&compound_bounded),
4557 full_rows[s..e].to_vec(),
4558 "compound-AND fallback: op={} threshold={} n_nodes={} SKIP {} LIMIT {}: \
4559 result differs from fused",
4560 op_str, threshold, n_nodes, skip, limit
4561 );
4562 }
4563 }
4564
4565 #[test]
4576 fn dense_hop1_with_filter_survives_pull() {
4577 const LEAVES: usize = 120; const CAP: usize = 100;
4579
4580 let mut fx = Fx::new();
4581 let src = fx.add("Src", "src", vec![]);
4582 for idx in 0..LEAVES {
4583 let leaf = fx.add("Leaf", &format!("l{idx}"), vec![("v", i(idx as i64))]);
4584 fx.edge("T", src, leaf, vec![]);
4585 }
4586 let v = fx.view();
4587 let params = BTreeMap::new();
4588
4589 let staged_err = super::with_max_intermediate_rows(CAP, || {
4591 super::execute_unbounded(
4592 &v,
4593 &compile("MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v >= 110 RETURN l, l.v"),
4594 &Params(¶ms),
4595 )
4596 });
4597 assert!(
4598 staged_err.is_err(),
4599 "staged path must error on 120 leaves with cap={CAP}"
4600 );
4601 assert!(
4602 staged_err
4603 .unwrap_err()
4604 .contains("intermediate result exceeds"),
4605 "wrong error message"
4606 );
4607
4608 let ok = super::with_max_intermediate_rows(CAP, || {
4612 run(
4613 &v,
4614 "MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v >= 110 RETURN l, l.v LIMIT 5",
4615 ¶ms,
4616 )
4617 });
4618 let rs = ok.expect("pull-based must survive despite dense hop-1 exceeding cap");
4619 assert_eq!(rs.len(), 5, "LIMIT 5 must return exactly 5 rows");
4620
4621 let vs: Vec<i64> = (0..rs.len())
4623 .filter_map(|i| match rs.get(i, "l.v") {
4624 Some(Value::Int(n)) => Some(*n),
4625 _ => None,
4626 })
4627 .collect();
4628 assert_eq!(vs.len(), 5, "all projected rows must have v");
4629 for v_val in &vs {
4630 assert!(*v_val >= 110, "filter must hold: v={v_val} is not >= 110");
4631 }
4632
4633 let (pull_result, pull_produced) = super::with_expand_counter(|| {
4637 super::with_max_intermediate_rows(1_000_000, || {
4638 run(
4639 &v,
4640 "MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v < 10 RETURN l LIMIT 5",
4641 ¶ms,
4642 )
4643 })
4644 });
4645 pull_result.expect("pull must succeed without cap");
4646 assert!(
4648 pull_produced <= 5,
4649 "pull expand count {pull_produced} should be ≤ 5 (stops after 5 passing leaves)"
4650 );
4651
4652 let (staged_result, staged_produced) = super::with_expand_counter(|| {
4653 super::with_max_intermediate_rows(1_000_000, || {
4654 super::execute_unbounded(
4655 &v,
4656 &compile("MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v < 10 RETURN l"),
4657 &Params(¶ms),
4658 )
4659 })
4660 });
4661 staged_result.expect("staged must succeed with 1M cap");
4662 assert_eq!(
4663 staged_produced, LEAVES,
4664 "staged must expand all {LEAVES} leaves"
4665 );
4666
4667 assert!(
4668 staged_produced >= pull_produced * 10,
4669 "staged ({staged_produced}) must be ≥ 10× pull ({pull_produced})"
4670 );
4671 }
4672
4673 #[test]
4691 fn harness_shape_two_hop_dense_survives_pull() {
4692 const N_TALENT: usize = 70;
4693 const N_COMPANY: usize = 20;
4694 const N_INDUSTRY: usize = 3;
4695 const CAP: usize = 100;
4696 const LIMIT: usize = 10;
4697
4698 let mut fx = Fx::new();
4699
4700 let mut talent_ids: Vec<u32> = Vec::new();
4702 let mut talent_industry: Vec<usize> = Vec::new();
4703 for i in 0..N_TALENT {
4704 let ind = i % N_INDUSTRY;
4705 let id = fx.add(
4706 "Talent",
4707 &format!("t{i}"),
4708 vec![("industry", s(&ind.to_string()))],
4709 );
4710 talent_ids.push(id);
4711 talent_industry.push(ind);
4712 }
4713
4714 let mut company_ids: Vec<u32> = Vec::new();
4716 let mut company_industry: Vec<usize> = Vec::new();
4717 for i in 0..N_COMPANY {
4718 let ind = i % N_INDUSTRY;
4719 let id = fx.add(
4720 "Company",
4721 &format!("c{i}"),
4722 vec![("industry", s(&ind.to_string()))],
4723 );
4724 company_ids.push(id);
4725 company_industry.push(ind);
4726 }
4727
4728 for (ti, &tid) in talent_ids.iter().enumerate() {
4730 for (ci, &cid) in company_ids.iter().enumerate() {
4731 if talent_industry[ti] == company_industry[ci] {
4732 fx.edge("INDUSTRY_ALIGNMENT", tid, cid, vec![]);
4733 }
4734 }
4735 }
4736
4737 let v = fx.view();
4738 let params = BTreeMap::new();
4739 let query = format!(
4740 "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company)\
4741 <-[:INDUSTRY_ALIGNMENT]-(t2:Talent) RETURN t, c, t2 LIMIT {LIMIT}"
4742 );
4743
4744 const UNBOUNDED_Q: &str = "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company)\
4746 <-[:INDUSTRY_ALIGNMENT]-(t2:Talent) RETURN t, c, t2";
4747 let staged_err = super::with_max_intermediate_rows(CAP, || {
4748 super::execute_unbounded(&v, &compile(UNBOUNDED_Q), &Params(¶ms))
4749 });
4750 assert!(
4751 staged_err.is_err(),
4752 "staged must error with cap={CAP} on harness-shape graph"
4753 );
4754 assert!(
4755 staged_err
4756 .unwrap_err()
4757 .contains("intermediate result exceeds"),
4758 "wrong error"
4759 );
4760
4761 let ok = super::with_max_intermediate_rows(CAP, || run(&v, &query, ¶ms));
4763 let rs = ok.expect("pull-based must complete on harness-shape with LIMIT 10");
4764 assert_eq!(rs.len(), LIMIT, "must return exactly {LIMIT} rows");
4765
4766 for i in 0..rs.len() {
4771 let row = rs.row(i);
4772 assert_eq!(row.len(), 3, "each row must have 3 columns (t, c, t2)");
4773 assert!(row.iter().all(|c| c.is_some()), "all cells must be Some");
4774 }
4775 }
4776
4777 #[test]
4778 fn intermediate_row_cap_errors_on_scan_and_expand() {
4779 let cap_msg = |n: usize| {
4780 format!(
4781 "intermediate result exceeds {n} rows; add a LIMIT or constrain patterns with shared variables"
4782 )
4783 };
4784
4785 let mut scan_fx = Fx::new();
4786 scan_fx.add("N", "a", vec![]);
4787 scan_fx.add("N", "b", vec![]);
4788 scan_fx.add("N", "c", vec![]);
4789 let sv = scan_fx.view();
4790 let scan_err = super::with_max_intermediate_rows(2, || {
4791 run(&sv, "MATCH (n:N) RETURN n", &BTreeMap::new())
4792 })
4793 .expect_err("3-row scan must exceed cap 2");
4794 assert_eq!(scan_err, cap_msg(2));
4795
4796 let mut exp_fx = Fx::new();
4797 let src = exp_fx.add("Src", "s", vec![]);
4798 let d1 = exp_fx.add("Dst", "d1", vec![]);
4799 let d2 = exp_fx.add("Dst", "d2", vec![]);
4800 exp_fx.edge("T", src, d1, vec![]);
4801 exp_fx.edge("T", src, d2, vec![]);
4802 let ev = exp_fx.view();
4803 let exp_err = super::with_max_intermediate_rows(1, || {
4805 run(&ev, "MATCH (x:Src)-[:T]->(y) RETURN x, y", &BTreeMap::new())
4806 })
4807 .expect_err("2-row expand must exceed cap 1");
4808 assert_eq!(exp_err, cap_msg(1));
4809 }
4810
4811 #[test]
4821 fn bounded_matches_unbounded_slice_various_limits() {
4822 let fx = hop_graph();
4824 let v = fx.view();
4825 let params = BTreeMap::new();
4826
4827 let full_plan = compile("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b");
4829 let full_rs = super::execute_unbounded(&v, &full_plan, &Params(¶ms)).unwrap();
4830 let full_rows = rows_of(&full_rs);
4831 assert_eq!(full_rows.len(), 3, "hop_graph has exactly 3 KNOWS paths");
4832
4833 for limit in [1u64, 2, 3, 10] {
4834 let q = format!("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b LIMIT {limit}");
4835 let rs = run(&v, &q, ¶ms).unwrap();
4836 let expected_len = (limit as usize).min(full_rows.len());
4837 assert_eq!(
4838 rs.len(),
4839 expected_len,
4840 "LIMIT {limit}: expected {expected_len} rows, got {}",
4841 rs.len()
4842 );
4843 assert_eq!(
4844 rows_of(&rs),
4845 full_rows[..expected_len],
4846 "LIMIT {limit}: rows differ from reference slice"
4847 );
4848 }
4849
4850 let skip_rs = run(
4852 &v,
4853 "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b SKIP 1 LIMIT 2",
4854 ¶ms,
4855 )
4856 .unwrap();
4857 assert_eq!(rows_of(&skip_rs), full_rows[1..3]);
4858
4859 let mut fx2 = Fx::new();
4861 fx2.add("N", "a", vec![("v", i(1))]);
4862 fx2.add("N", "b", vec![("v", i(2))]);
4863 fx2.add("N", "c", vec![("v", i(3))]);
4864 let v2 = fx2.view();
4865 let filter_full = super::execute_unbounded(
4867 &v2,
4868 &compile("MATCH (n:N) WHERE n.v > 1 RETURN n"),
4869 &Params(¶ms),
4870 )
4871 .unwrap();
4872 assert_eq!(filter_full.len(), 2);
4873
4874 let filter_lim = run(&v2, "MATCH (n:N) WHERE n.v > 1 RETURN n LIMIT 1", ¶ms).unwrap();
4875 assert_eq!(filter_lim.len(), 1, "LIMIT 1 on filter query");
4876 assert_eq!(rows_of(&filter_lim), rows_of(&filter_full)[..1]);
4877
4878 let tri = triangle();
4880 let tv = tri.view();
4881 let tri_full = super::execute_unbounded(
4883 &tv,
4884 &compile("MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z"),
4885 &Params(¶ms),
4886 )
4887 .unwrap();
4888 assert_eq!(tri_full.len(), 3, "triangle has 3 unique two-hop paths");
4889 for limit in [1u64, 2, 3, 5] {
4890 let q = format!("MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z LIMIT {limit}");
4891 let rs = run(&tv, &q, ¶ms).unwrap();
4892 let expected_len = (limit as usize).min(3);
4893 assert_eq!(
4894 rs.len(),
4895 expected_len,
4896 "triangle LIMIT {limit}: got {} rows",
4897 rs.len()
4898 );
4899 assert_eq!(
4900 rows_of(&rs),
4901 rows_of(&tri_full)[..expected_len],
4902 "triangle LIMIT {limit}: rows differ"
4903 );
4904 }
4905 }
4906
4907 #[test]
4911 fn expand_terminates_early_with_row_bound() {
4912 const LEAVES: usize = 500;
4913 let mut fx = Fx::new();
4914 let hub = fx.add("Hub", "hub", vec![]);
4915 for i in 0..LEAVES {
4916 let leaf = fx.add("Leaf", &format!("leaf-{i}"), vec![]);
4917 fx.edge("T", hub, leaf, vec![]);
4918 }
4919 let v = fx.view();
4920 let params = BTreeMap::new();
4921 let full_plan = compile("MATCH (h:Hub)-[:T]->(x:Leaf) RETURN x");
4922
4923 let (bounded_result, bounded_produced) = super::with_expand_counter(|| {
4925 run(&v, "MATCH (h:Hub)-[:T]->(x:Leaf) RETURN x LIMIT 5", ¶ms)
4926 });
4927 let bounded_rs = bounded_result.unwrap();
4928 assert_eq!(bounded_rs.len(), 5, "LIMIT 5 must return exactly 5 rows");
4929 assert!(
4930 bounded_produced <= 5,
4931 "bounded: exec_expand emitted {bounded_produced} rows, expected ≤ 5"
4932 );
4933
4934 let (unbounded_result, unbounded_produced) = super::with_expand_counter(|| {
4936 super::execute_unbounded(&v, &full_plan, &Params(¶ms))
4937 });
4938 let unbounded_rs = unbounded_result.unwrap();
4939 assert_eq!(
4940 unbounded_rs.len(),
4941 LEAVES,
4942 "unbounded must return all {LEAVES} rows"
4943 );
4944 assert_eq!(
4945 unbounded_produced, LEAVES,
4946 "unbounded: exec_expand must emit all {LEAVES} rows"
4947 );
4948
4949 assert!(
4951 unbounded_produced >= bounded_produced * 100,
4952 "unbounded ({unbounded_produced}) must be ≥ 100× bounded ({bounded_produced})"
4953 );
4954 }
4955
4956 #[test]
4962 fn bounded_query_survives_low_intermediate_row_cap() {
4963 let mut fx = Fx::new();
4965 let src = fx.add("Src", "s", vec![]);
4966 for i in 0..20usize {
4967 let dst = fx.add("Dst", &format!("d{i}"), vec![]);
4968 fx.edge("T", src, dst, vec![]);
4969 }
4970 let v = fx.view();
4971 let params = BTreeMap::new();
4972 let full_plan = compile("MATCH (s:Src)-[:T]->(d:Dst) RETURN d");
4973
4974 let cap_err = super::with_max_intermediate_rows(10, || {
4976 super::execute_unbounded(&v, &full_plan, &Params(¶ms))
4977 });
4978 assert!(
4979 cap_err.is_err(),
4980 "unbounded must hit the intermediate-row cap"
4981 );
4982 assert!(
4983 cap_err.unwrap_err().contains("intermediate result exceeds"),
4984 "cap error message must be the budget message"
4985 );
4986
4987 let ok = super::with_max_intermediate_rows(10, || {
4989 run(&v, "MATCH (s:Src)-[:T]->(d:Dst) RETURN d LIMIT 5", ¶ms)
4990 });
4991 assert_eq!(
4992 ok.unwrap().len(),
4993 5,
4994 "bounded (LIMIT 5) must complete with 5 rows, not a cap error"
4995 );
4996
4997 let at_cap = super::with_max_intermediate_rows(10, || {
4999 run(&v, "MATCH (s:Src)-[:T]->(d:Dst) RETURN d LIMIT 10", ¶ms)
5000 });
5001 assert_eq!(
5002 at_cap.unwrap().len(),
5003 10,
5004 "bounded at LIMIT==cap must complete with 10 rows"
5005 );
5006 }
5007
5008 #[test]
5013 fn count_star_returns_total_node_count() {
5014 let mut fx = Fx::new();
5015 fx.add("Person", "ada", vec![]);
5016 fx.add("Person", "bob", vec![]);
5017 fx.add("Person", "cam", vec![]);
5018 let v = fx.view();
5019 let params = BTreeMap::new();
5020
5021 let rs = run(&v, "MATCH (n:Person) RETURN COUNT(*)", ¶ms).expect("COUNT(*)");
5022 assert_eq!(rs.columns(), &["COUNT(*)".to_string()]);
5023 assert_eq!(rs.len(), 1);
5024 assert_eq!(rs.row(0), &[Some(i(3))]);
5025
5026 let rs_empty = run(&v, "MATCH (n:Ghost) RETURN COUNT(*)", ¶ms).expect("COUNT(*) empty");
5028 assert_eq!(rs_empty.row(0), &[Some(i(0))]);
5029 }
5030
5031 #[test]
5032 fn count_star_alias_sets_column_name() {
5033 let mut fx = Fx::new();
5034 fx.add("N", "a", vec![]);
5035 let v = fx.view();
5036 let params = BTreeMap::new();
5037 let rs = run(&v, "MATCH (n:N) RETURN COUNT(*) AS total", ¶ms).expect("COUNT AS");
5038 assert_eq!(rs.columns(), &["total".to_string()]);
5039 assert_eq!(rs.row(0), &[Some(i(1))]);
5040 }
5041
5042 #[test]
5043 fn count_var_skips_null_node_bindings() {
5044 let mut fx = Fx::new();
5047 fx.add("N", "a", vec![]);
5048 fx.add("N", "b", vec![]);
5049 let v = fx.view();
5050 let params = BTreeMap::new();
5051 let rs = run(&v, "MATCH (n:N) RETURN COUNT(n)", ¶ms).expect("COUNT(n)");
5052 assert_eq!(rs.columns(), &["COUNT(n)".to_string()]);
5053 assert_eq!(rs.row(0), &[Some(i(2))]);
5054 }
5055
5056 #[test]
5057 fn sum_numeric_prop_ignores_null_and_non_numeric() {
5058 let mut fx = Fx::new();
5059 fx.add("N", "a", vec![("v", i(10))]);
5060 fx.add("N", "b", vec![("v", i(20))]);
5061 fx.add("N", "c", vec![]); let v = fx.view();
5063 let params = BTreeMap::new();
5064 let rs = run(&v, "MATCH (n:N) RETURN SUM(n.v)", ¶ms).expect("SUM");
5065 assert_eq!(rs.columns(), &["SUM(n.v)".to_string()]);
5066 assert_eq!(rs.row(0), &[Some(f(30.0))]);
5068
5069 let rs_null = run(&v, "MATCH (n:N) RETURN SUM(n.missing)", ¶ms).expect("SUM null");
5071 assert_eq!(rs_null.row(0), &[None]);
5072 }
5073
5074 #[test]
5075 fn avg_numeric_prop() {
5076 let mut fx = Fx::new();
5077 fx.add("N", "a", vec![("v", i(10))]);
5078 fx.add("N", "b", vec![("v", i(30))]);
5079 let v = fx.view();
5080 let params = BTreeMap::new();
5081 let rs = run(&v, "MATCH (n:N) RETURN AVG(n.v) AS avg_v", ¶ms).expect("AVG");
5082 assert_eq!(rs.columns(), &["avg_v".to_string()]);
5083 assert_eq!(rs.row(0), &[Some(f(20.0))]);
5085
5086 let rs_empty = run(&v, "MATCH (n:Ghost) RETURN AVG(n.v)", ¶ms).expect("AVG empty");
5088 assert_eq!(rs_empty.row(0), &[None]);
5089 }
5090
5091 #[test]
5092 fn min_max_numeric_prop() {
5093 let mut fx = Fx::new();
5094 fx.add("N", "a", vec![("v", i(5))]);
5095 fx.add("N", "b", vec![("v", i(1))]);
5096 fx.add("N", "c", vec![("v", i(9))]);
5097 fx.add("N", "d", vec![]); let v = fx.view();
5099 let params = BTreeMap::new();
5100
5101 let min_rs = run(&v, "MATCH (n:N) RETURN MIN(n.v)", ¶ms).expect("MIN");
5102 assert_eq!(min_rs.row(0), &[Some(i(1))]);
5103
5104 let max_rs = run(&v, "MATCH (n:N) RETURN MAX(n.v)", ¶ms).expect("MAX");
5105 assert_eq!(max_rs.row(0), &[Some(i(9))]);
5106 }
5107
5108 #[test]
5111 fn min_max_mixed_int_float_props() {
5112 let mut fx = Fx::new();
5113 fx.add("N", "a", vec![("v", i(3))]);
5115 fx.add("N", "b", vec![("v", f(1.5))]);
5116 fx.add("N", "c", vec![("v", i(7))]);
5117 fx.add("N", "d", vec![("v", f(2.0))]);
5118 let v = fx.view();
5119 let params = BTreeMap::new();
5120
5121 let min_rs = run(&v, "MATCH (n:N) RETURN MIN(n.v)", ¶ms).expect("MIN mixed");
5122 assert_eq!(min_rs.row(0), &[Some(f(1.5))]);
5124
5125 let max_rs = run(&v, "MATCH (n:N) RETURN MAX(n.v)", ¶ms).expect("MAX mixed");
5126 assert_eq!(max_rs.row(0), &[Some(i(7))]);
5128 }
5129
5130 #[test]
5133 fn aggregate_limit_skip_order_by_are_no_ops() {
5134 let mut fx = Fx::new();
5135 fx.add("N", "a", vec![]);
5136 fx.add("N", "b", vec![]);
5137 fx.add("N", "c", vec![]);
5138 let v = fx.view();
5139 let params = BTreeMap::new();
5140
5141 let rs_lim5 =
5143 run(&v, "MATCH (n:N) RETURN COUNT(*) LIMIT 5", ¶ms).expect("COUNT(*) LIMIT 5");
5144 assert_eq!(
5145 rs_lim5.len(),
5146 1,
5147 "aggregate with LIMIT 5 must still return 1 row"
5148 );
5149 assert_eq!(rs_lim5.row(0), &[Some(i(3))]);
5150
5151 let rs_lim0 =
5153 run(&v, "MATCH (n:N) RETURN COUNT(*) LIMIT 0", ¶ms).expect("COUNT(*) LIMIT 0");
5154 assert_eq!(
5155 rs_lim0.len(),
5156 1,
5157 "aggregate with LIMIT 0 must still return 1 row"
5158 );
5159 assert_eq!(rs_lim0.row(0), &[Some(i(3))]);
5160
5161 let rs_skip =
5163 run(&v, "MATCH (n:N) RETURN COUNT(*) SKIP 100", ¶ms).expect("COUNT(*) SKIP 100");
5164 assert_eq!(
5165 rs_skip.len(),
5166 1,
5167 "aggregate with large SKIP must still return 1 row"
5168 );
5169
5170 let rs_ord = plan_src("MATCH (n:N) RETURN COUNT(*) ORDER BY n");
5174 assert!(
5177 rs_ord.is_ok(),
5178 "COUNT(*) ORDER BY should plan without error (ORDER BY dropped)"
5179 );
5180 let plan_ops = rs_ord.unwrap();
5181 assert!(
5183 !plan_ops
5184 .iter()
5185 .any(|op| matches!(op, crate::cypher::plan::PlanOp::OrderBy { .. })),
5186 "aggregate plan must not contain OrderBy"
5187 );
5188 }
5189
5190 #[test]
5191 fn count_star_no_budget_cap_applies() {
5192 let mut fx = Fx::new();
5195 let src = fx.add("Src", "s", vec![]);
5196 for i in 0..30usize {
5197 let dst = fx.add("Dst", &format!("d{i}"), vec![]);
5198 fx.edge("T", src, dst, vec![]);
5199 }
5200 let v = fx.view();
5201 let params = BTreeMap::new();
5202
5203 let cap_err =
5205 super::with_max_intermediate_rows(10, || run(&v, "MATCH (n:Dst) RETURN n", ¶ms));
5206 assert!(
5207 cap_err.is_err(),
5208 "staged path must error on 30 nodes with cap=10"
5209 );
5210
5211 let agg_ok = super::with_max_intermediate_rows(10, || {
5213 run(&v, "MATCH (n:Dst) RETURN COUNT(*)", ¶ms)
5214 })
5215 .expect("aggregate must not hit the intermediate-row cap");
5216 assert_eq!(
5217 agg_ok.row(0),
5218 &[Some(i(30))],
5219 "COUNT(*) must count all 30 nodes regardless of cap"
5220 );
5221 }
5222
5223 fn plan_src(src: &str) -> Result<Vec<crate::cypher::plan::PlanOp>, String> {
5224 use crate::cypher::{lex, parse, plan};
5225 let toks = lex(src).map_err(|e| format!("lex: {e}"))?;
5226 let ast = parse(&toks).map_err(|e| format!("parse: {e}"))?;
5227 plan(&ast).map_err(|e| format!("plan: {e}"))
5228 }
5229
5230 #[test]
5233 fn grouped_aggregation_plan_routing() {
5234 use crate::cypher::plan::PlanOp;
5235
5236 let ops = plan_src("MATCH (a:N) RETURN a, COUNT(*)")
5238 .expect("grouped aggregation must now succeed");
5239 assert!(
5240 ops.iter()
5241 .any(|op| matches!(op, PlanOp::GroupAggregate { .. })),
5242 "grouped aggregation plan must contain GroupAggregate op, got: {ops:?}"
5243 );
5244
5245 let ops2 = plan_src("MATCH (a:N) RETURN COUNT(*), COUNT(a)")
5247 .expect("multi-aggregate must now succeed");
5248 assert!(
5249 ops2.iter()
5250 .any(|op| matches!(op, PlanOp::GroupAggregate { .. })),
5251 "multi-aggregate plan must contain GroupAggregate op, got: {ops2:?}"
5252 );
5253
5254 let err3 = plan_src("MATCH (a:N) RETURN SUM(*)").expect_err("SUM(*) must be plan error");
5256 assert!(
5257 err3.to_ascii_lowercase().contains("sum") || err3.to_ascii_lowercase().contains("*"),
5258 "error must mention SUM or *, got: {err3}"
5259 );
5260 }
5261
5262 #[test]
5265 fn grouped_single_key_count() {
5266 let mut fx = Fx::new();
5268 fx.add("N", "a", vec![("t", s("X"))]);
5269 fx.add("N", "b", vec![("t", s("X"))]);
5270 fx.add("N", "c", vec![("t", s("Y"))]);
5271 let v = fx.view();
5272 let params = BTreeMap::new();
5273
5274 let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
5275 .expect("single-key grouped COUNT must succeed");
5276 assert_eq!(
5277 rs.columns(),
5278 &["n.t".to_string(), "cnt".to_string()],
5279 "columns must match RETURN clause"
5280 );
5281 assert_eq!(rs.len(), 2, "must produce exactly 2 groups (X and Y)");
5282
5283 let find = |label: &Value| (0..rs.len()).find(|&i| rs.row(i)[0].as_ref() == Some(label));
5285 let xi = find(&s("X")).expect("group X must exist");
5286 let yi = find(&s("Y")).expect("group Y must exist");
5287 assert_eq!(rs.row(xi)[1], Some(i(2)), "X group count must be 2");
5288 assert_eq!(rs.row(yi)[1], Some(i(1)), "Y group count must be 1");
5289 }
5290
5291 #[test]
5292 fn grouped_two_keys_sum_avg() {
5293 let mut fx = Fx::new();
5295 fx.add(
5296 "N",
5297 "a",
5298 vec![("cat", s("A")), ("sub", s("1")), ("v", i(10))],
5299 );
5300 fx.add(
5301 "N",
5302 "b",
5303 vec![("cat", s("A")), ("sub", s("1")), ("v", i(20))],
5304 );
5305 fx.add(
5306 "N",
5307 "c",
5308 vec![("cat", s("A")), ("sub", s("2")), ("v", i(5))],
5309 );
5310 fx.add(
5311 "N",
5312 "d",
5313 vec![("cat", s("B")), ("sub", s("1")), ("v", i(100))],
5314 );
5315 let v = fx.view();
5316 let params = BTreeMap::new();
5317
5318 let rs = run(
5319 &v,
5320 "MATCH (n:N) RETURN n.cat, n.sub, SUM(n.v) AS total, AVG(n.v) AS avg_v",
5321 ¶ms,
5322 )
5323 .expect("two-key SUM + AVG must succeed");
5324 assert_eq!(
5325 rs.columns(),
5326 &[
5327 "n.cat".to_string(),
5328 "n.sub".to_string(),
5329 "total".to_string(),
5330 "avg_v".to_string()
5331 ]
5332 );
5333 assert_eq!(rs.len(), 3, "must produce 3 groups: (A,1), (A,2), (B,1)");
5334
5335 let find = |cat: &Value, sub: &Value| {
5336 (0..rs.len())
5337 .find(|&i| rs.row(i)[0].as_ref() == Some(cat) && rs.row(i)[1].as_ref() == Some(sub))
5338 };
5339 let a1 = find(&s("A"), &s("1")).expect("group (A,1) must exist");
5340 assert_eq!(rs.row(a1)[2], Some(f(30.0)), "(A,1) SUM must be 30.0");
5341 assert_eq!(rs.row(a1)[3], Some(f(15.0)), "(A,1) AVG must be 15.0");
5342
5343 let a2 = find(&s("A"), &s("2")).expect("group (A,2) must exist");
5344 assert_eq!(rs.row(a2)[2], Some(f(5.0)), "(A,2) SUM must be 5.0");
5345
5346 let b1 = find(&s("B"), &s("1")).expect("group (B,1) must exist");
5347 assert_eq!(rs.row(b1)[2], Some(f(100.0)), "(B,1) SUM must be 100.0");
5348 assert_eq!(rs.row(b1)[3], Some(f(100.0)), "(B,1) AVG must be 100.0");
5349 }
5350
5351 #[test]
5352 fn grouped_order_by_count_desc_limit() {
5353 let mut fx = Fx::new();
5355 fx.add("N", "a1", vec![("cat", s("A"))]);
5356 fx.add("N", "a2", vec![("cat", s("A"))]);
5357 fx.add("N", "a3", vec![("cat", s("A"))]);
5358 fx.add("N", "b1", vec![("cat", s("B"))]);
5359 fx.add("N", "b2", vec![("cat", s("B"))]);
5360 fx.add("N", "c1", vec![("cat", s("C"))]);
5361 fx.add("N", "d1", vec![("cat", s("D"))]);
5362 fx.add("N", "d2", vec![("cat", s("D"))]);
5363 fx.add("N", "d3", vec![("cat", s("D"))]);
5364 fx.add("N", "d4", vec![("cat", s("D"))]);
5365 fx.add("N", "e1", vec![("cat", s("E"))]);
5366 let v = fx.view();
5367 let params = BTreeMap::new();
5368
5369 let rs = run(
5370 &v,
5371 "MATCH (n:N) RETURN n.cat, COUNT(*) AS cnt ORDER BY cnt DESC LIMIT 3",
5372 ¶ms,
5373 )
5374 .expect("ORDER BY count DESC LIMIT 3 must succeed");
5375 assert_eq!(rs.len(), 3, "LIMIT 3 must return exactly 3 groups");
5376 assert_eq!(rs.row(0)[1], Some(i(4)), "row 0 must be count 4");
5378 assert_eq!(rs.row(0)[0], Some(s("D")), "row 0 must be category D");
5379 assert_eq!(rs.row(1)[1], Some(i(3)), "row 1 must be count 3");
5380 assert_eq!(rs.row(1)[0], Some(s("A")), "row 1 must be category A");
5381 assert_eq!(rs.row(2)[1], Some(i(2)), "row 2 must be count 2");
5382 assert_eq!(rs.row(2)[0], Some(s("B")), "row 2 must be category B");
5383
5384 let plan_ops =
5386 plan_src("MATCH (n:N) RETURN n.cat, COUNT(*) AS cnt ORDER BY cnt DESC LIMIT 3")
5387 .expect("plan must succeed");
5388 assert_eq!(
5389 crate::cypher::plan::row_bound(&plan_ops),
5390 None,
5391 "GroupAggregate plan with LIMIT must have row_bound = None"
5392 );
5393 }
5394
5395 #[test]
5396 fn grouped_empty_input_yields_zero_groups() {
5397 let fx = Fx::new(); let v = fx.view();
5399 let params = BTreeMap::new();
5400
5401 let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
5402 .expect("grouped aggregate on empty graph must succeed");
5403 assert_eq!(rs.len(), 0, "empty input must yield zero groups");
5404 assert_eq!(
5405 rs.columns(),
5406 &["n.t".to_string(), "cnt".to_string()],
5407 "columns must still be present even with zero rows"
5408 );
5409 }
5410
5411 #[test]
5412 fn grouped_null_key_groups_together() {
5413 let mut fx = Fx::new();
5415 fx.add("N", "a", vec![("t", s("X"))]);
5416 fx.add("N", "b", vec![]); fx.add("N", "c", vec![]); fx.add("N", "d", vec![("t", s("Y"))]);
5419 let v = fx.view();
5420 let params = BTreeMap::new();
5421
5422 let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
5423 .expect("null-key grouped aggregate must succeed");
5424 assert_eq!(rs.len(), 3, "must produce 3 groups: X, null, Y");
5425
5426 let null_row = (0..rs.len())
5428 .find(|&i| rs.row(i)[0].is_none())
5429 .expect("null group must be present");
5430 assert_eq!(
5431 rs.row(null_row)[1],
5432 Some(i(2)),
5433 "null group must count 2 rows (b and c)"
5434 );
5435
5436 let x_row = (0..rs.len())
5437 .find(|&i| rs.row(i)[0] == Some(s("X")))
5438 .expect("X group must exist");
5439 assert_eq!(rs.row(x_row)[1], Some(i(1)));
5440
5441 let y_row = (0..rs.len())
5442 .find(|&i| rs.row(i)[0] == Some(s("Y")))
5443 .expect("Y group must exist");
5444 assert_eq!(rs.row(y_row)[1], Some(i(1)));
5445 }
5446
5447 #[test]
5448 fn grouped_cap_error_on_high_cardinality() {
5449 let mut fx = Fx::new();
5452 fx.add("N", "a", vec![("t", s("A"))]);
5453 fx.add("N", "b", vec![("t", s("B"))]);
5454 fx.add("N", "c", vec![("t", s("C"))]);
5455 let v = fx.view();
5456 let params = BTreeMap::new();
5457
5458 let err = super::with_max_groups(2, || {
5459 run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
5460 })
5461 .expect_err("must error when group count exceeds cap");
5462 assert!(
5463 err.to_ascii_lowercase().contains("group count"),
5464 "error must mention group count, got: {err}"
5465 );
5466 }
5467
5468 #[test]
5469 fn multi_aggregate_no_keys() {
5470 let mut fx = Fx::new();
5473 fx.add("N", "a", vec![]);
5474 fx.add("N", "b", vec![]);
5475 let v = fx.view();
5476 let params = BTreeMap::new();
5477
5478 let rs = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
5479 .expect("multi-aggregate no keys must succeed");
5480 assert_eq!(rs.len(), 1, "must produce exactly one result row");
5481 assert_eq!(rs.row(0)[0], Some(i(2)), "COUNT(*) must be 2");
5482 assert_eq!(rs.row(0)[1], Some(i(2)), "COUNT(n) must be 2");
5483 }
5484
5485 #[test]
5486 fn aggregate_over_hop_counts_edges() {
5487 let fx = hop_graph();
5488 let v = fx.view();
5489 let params = BTreeMap::new();
5490 let rs = run(
5492 &v,
5493 "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN COUNT(*)",
5494 ¶ms,
5495 )
5496 .expect("COUNT(*) on hop graph");
5497 assert_eq!(rs.row(0), &[Some(i(3))]);
5498 }
5499
5500 #[test]
5505 fn multi_aggregate_no_keys_empty_graph() {
5506 let fx = Fx::new(); let v = fx.view();
5512 let params = BTreeMap::new();
5513
5514 let rs = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
5515 .expect("empty-graph multi-agg must succeed");
5516 assert_eq!(rs.len(), 1, "must produce exactly 1 row on empty input");
5517 assert_eq!(
5518 rs.row(0)[0],
5519 Some(i(0)),
5520 "COUNT(*) on empty graph must be 0"
5521 );
5522 assert_eq!(
5523 rs.row(0)[1],
5524 Some(i(0)),
5525 "COUNT(n) on empty graph must be 0"
5526 );
5527 }
5528
5529 #[test]
5530 fn fast_path_and_group_path_agree_on_empty_input() {
5531 let fx = Fx::new();
5537 let v = fx.view();
5538 let params = BTreeMap::new();
5539
5540 let fast = run(&v, "MATCH (n:N) RETURN COUNT(*)", ¶ms)
5541 .expect("fast-path COUNT on empty graph");
5542 assert_eq!(fast.len(), 1, "fast path: 1 row on empty input");
5543 let fast_count = fast.row(0)[0].clone();
5544
5545 let grouped = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
5546 .expect("group-path COUNT on empty graph");
5547 assert_eq!(grouped.len(), 1, "group path: 1 row on empty input");
5548 let group_count = grouped.row(0)[0].clone();
5549
5550 assert_eq!(
5551 fast_count, group_count,
5552 "fast path and group path must agree on COUNT(*) for empty input"
5553 );
5554
5555 let mut fx2 = Fx::new();
5557 fx2.add("N", "x", vec![]);
5558 fx2.add("N", "y", vec![]);
5559 let v2 = fx2.view();
5560
5561 let fast2 = run(&v2, "MATCH (n:N) RETURN COUNT(*)", ¶ms)
5562 .expect("fast-path COUNT on 2-node graph");
5563 assert_eq!(fast2.row(0)[0], Some(i(2)), "fast path: COUNT(*) = 2");
5564
5565 let grouped2 = run(&v2, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
5566 .expect("group-path COUNT on 2-node graph");
5567 assert_eq!(grouped2.row(0)[0], Some(i(2)), "group path: COUNT(*) = 2");
5568 assert_eq!(grouped2.row(0)[1], Some(i(2)), "group path: COUNT(n) = 2");
5569 }
5570
5571 #[test]
5576 fn grouped_int_float_key_unification() {
5577 let mut fx = Fx::new();
5584 fx.add("N", "a", vec![("score", i(1))]); fx.add("N", "b", vec![("score", f(1.0))]);
5586 let v = fx.view();
5587 let params = BTreeMap::new();
5588
5589 let rs = run(&v, "MATCH (n:N) RETURN n.score, COUNT(*) AS cnt", ¶ms)
5590 .expect("int/float unification must succeed");
5591 assert_eq!(
5592 rs.len(),
5593 1,
5594 "Int(1) and Float(1.0) must group together into 1 group"
5595 );
5596 assert_eq!(
5598 rs.row(0)[0],
5599 Some(i(1)),
5600 "key column must display Int(1) (first-seen)"
5601 );
5602 assert_eq!(rs.row(0)[1], Some(i(2)), "unified group must have count=2");
5603 }
5604
5605 #[test]
5606 fn distinct_collapses_duplicate_projected_values() {
5607 let mut fx = Fx::new();
5608 fx.add("N", "a1", vec![("city", s("Austin"))]);
5609 fx.add("N", "a2", vec![("city", s("Austin"))]);
5610 let v = fx.view();
5611 let params = BTreeMap::new();
5612 let rs = run(&v, "MATCH (n:N) RETURN DISTINCT n.city", ¶ms).unwrap();
5613 assert_eq!(rs.len(), 1);
5614 assert_eq!(rs.row(0)[0], Some(s("Austin")));
5615 }
5616
5617 #[test]
5618 fn distinct_int_float_unify() {
5619 let mut fx = Fx::new();
5620 fx.add("N", "a", vec![("score", i(1))]);
5621 fx.add("N", "b", vec![("score", f(1.0))]);
5622 let v = fx.view();
5623 let params = BTreeMap::new();
5624 let rs = run(&v, "MATCH (n:N) RETURN DISTINCT n.score", ¶ms).unwrap();
5625 assert_eq!(
5626 rs.len(),
5627 1,
5628 "Int(1) and Float(1.0) must DISTINCT as one row"
5629 );
5630 assert_eq!(rs.row(0)[0], Some(i(1)), "first-seen Int(1) wins display");
5631 }
5632
5633 #[test]
5634 fn distinct_caps_at_intermediate_row_budget() {
5635 let mut fx = Fx::new();
5636 fx.add("N", "a", vec![("city", s("Austin"))]);
5637 fx.add("N", "b", vec![("city", s("Paris"))]);
5638 let v = fx.view();
5639 let params = BTreeMap::new();
5640 let err = super::with_max_intermediate_rows(1, || {
5641 run(&v, "MATCH (n:N) RETURN DISTINCT n.city", ¶ms)
5642 })
5643 .expect_err("two distinct cities must exceed cap=1");
5644 assert!(
5645 err.contains("1") || err.contains("row"),
5646 "cap error must mention the budget, got: {err}"
5647 );
5648 }
5649
5650 #[test]
5651 fn where_in_list_filters_rows() {
5652 let mut fx = Fx::new();
5653 fx.add("N", "a", vec![("city", s("Austin"))]);
5654 fx.add("N", "p", vec![("city", s("Paris"))]);
5655 fx.add("N", "l", vec![("city", s("London"))]);
5656 let v = fx.view();
5657 let mut params = BTreeMap::new();
5658 params.insert("c".into(), s("Paris"));
5659 let rs = run(
5660 &v,
5661 "MATCH (n:N) WHERE n.city IN ['Austin', $c] RETURN n.city",
5662 ¶ms,
5663 )
5664 .unwrap();
5665 assert_eq!(rs.len(), 2);
5666 }
5667
5668 #[test]
5669 fn pure_int_keys_display_as_int() {
5670 let mut fx = Fx::new();
5675 fx.add("N", "a", vec![("age", i(10))]);
5676 fx.add("N", "b", vec![("age", i(20))]);
5677 fx.add("N", "c", vec![("age", i(10))]);
5678 let v = fx.view();
5679 let params = BTreeMap::new();
5680
5681 let rs = run(
5682 &v,
5683 "MATCH (n:N) RETURN n.age, COUNT(*) AS cnt ORDER BY n.age",
5684 ¶ms,
5685 )
5686 .expect("pure-Int group keys must succeed");
5687 assert_eq!(rs.len(), 2, "must have 2 groups: age=10 and age=20");
5688 assert_eq!(
5690 rs.row(0)[0],
5691 Some(i(10)),
5692 "age=10 key column must display as Int(10)"
5693 );
5694 assert_eq!(rs.row(0)[1], Some(i(2)), "age=10 group has 2 nodes");
5695 assert_eq!(
5696 rs.row(1)[0],
5697 Some(i(20)),
5698 "age=20 key column must display as Int(20)"
5699 );
5700 assert_eq!(rs.row(1)[1], Some(i(1)), "age=20 group has 1 node");
5701 }
5702
5703 #[test]
5708 fn var_expand_group_aggregate_staged_path() {
5709 let mut fx = Fx::new();
5719 let a = fx.add("N", "a", vec![("key", s("a"))]);
5720 let b = fx.add("N", "b", vec![("key", s("b"))]);
5721 let c = fx.add("N", "c", vec![("key", s("c"))]);
5722 fx.edge("T", a, b, vec![]);
5723 fx.edge("T", b, c, vec![]);
5724 let v = fx.view();
5725 let params = BTreeMap::new();
5726
5727 let rs = run(
5728 &v,
5729 "MATCH (x:N)-[*1..2]->(y:N) RETURN y.key, COUNT(*) AS cnt ORDER BY y.key",
5730 ¶ms,
5731 )
5732 .expect("VarExpand + GroupAggregate must succeed");
5733
5734 assert_eq!(rs.len(), 2, "must have 2 destination groups: b and c");
5735 assert_eq!(rs.row(0)[0], Some(s("b")), "first group key must be 'b'");
5736 assert_eq!(rs.row(0)[1], Some(i(1)), "b is reached via 1 path");
5737 assert_eq!(rs.row(1)[0], Some(s("c")), "second group key must be 'c'");
5738 assert_eq!(
5739 rs.row(1)[1],
5740 Some(i(2)),
5741 "c is reached via 2 paths (1-hop and 2-hop)"
5742 );
5743 }
5744
5745 #[test]
5753 fn pull_rows_var_expand_arm_returns_named_err() {
5754 let fx = Fx::new();
5755 let view = fx.view();
5756 let vars = super::VarTable {
5757 names: vec!["a".into(), "b".into()],
5758 };
5759 let project_items: Vec<crate::cypher::ast::RetItem> = vec![];
5760 let empty_params = BTreeMap::new();
5761 let params = super::Params(&empty_params);
5762 let ctx = super::PullCtx {
5763 view: &view,
5764 vars: &vars,
5765 project_items: &project_items,
5766 params: ¶ms,
5767 bound: 100,
5768 };
5769 let ops = vec![PlanOp::VarExpand {
5770 from: "a".into(),
5771 rel_var: None,
5772 etype: None,
5773 dir: crate::cypher::RelDir::Right,
5774 to: "b".into(),
5775 min: 1,
5776 max: 3,
5777 }];
5778 let mut row = vec![None; vars.names.len()];
5779 let mut result = Vec::new();
5780 let err = super::pull_rows(&ctx, &ops, &mut row, &mut result)
5781 .expect_err("VarExpand must Err in pull_rows");
5782 assert!(
5783 err.contains("VarExpand") && err.contains("pull executor"),
5784 "error must name VarExpand and pull executor, got: {err}"
5785 );
5786 }
5787
5788 #[test]
5789 fn pull_rows_shortest_path_arm_returns_named_err() {
5790 let fx = Fx::new();
5791 let view = fx.view();
5792 let vars = super::VarTable {
5793 names: vec!["a".into(), "b".into()],
5794 };
5795 let project_items: Vec<crate::cypher::ast::RetItem> = vec![];
5796 let empty_params = BTreeMap::new();
5797 let params = super::Params(&empty_params);
5798 let ctx = super::PullCtx {
5799 view: &view,
5800 vars: &vars,
5801 project_items: &project_items,
5802 params: ¶ms,
5803 bound: 100,
5804 };
5805 let ops = vec![PlanOp::ShortestPath {
5806 from: "a".into(),
5807 rel_var: None,
5808 etype: None,
5809 dir: crate::cypher::RelDir::Right,
5810 to: "b".into(),
5811 max_hops: 5,
5812 }];
5813 let mut row = vec![None; vars.names.len()];
5814 let mut result = Vec::new();
5815 let err = super::pull_rows(&ctx, &ops, &mut row, &mut result)
5816 .expect_err("ShortestPath must Err in pull_rows");
5817 assert!(
5818 err.contains("ShortestPath") && err.contains("pull executor"),
5819 "error must name ShortestPath and pull executor, got: {err}"
5820 );
5821 }
5822
5823 fn city_graph() -> Fx {
5827 let mut fx = Fx::new();
5828 fx.add(
5829 "Person",
5830 "alice",
5831 vec![("city", s("Boston")), ("age", i(30))],
5832 );
5833 fx.add("Person", "bob", vec![("city", s("Boston")), ("age", i(25))]);
5834 fx.add(
5835 "Person",
5836 "carol",
5837 vec![("city", s("Austin")), ("age", i(35))],
5838 );
5839 fx.add(
5840 "Person",
5841 "dave",
5842 vec![("city", s("Austin")), ("age", i(28))],
5843 );
5844 fx.add("Person", "eve", vec![("city", s("Boston")), ("age", i(22))]);
5845 fx
5846 }
5847
5848 #[test]
5851 fn with_aggregate_having_filters_groups() {
5852 let fx = city_graph();
5853 let view = fx.view();
5854 let params = BTreeMap::new();
5855 let rs = run(
5856 &view,
5857 "MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt WHERE cnt > 2 RETURN city, cnt",
5858 ¶ms,
5859 )
5860 .expect("WITH HAVING query must succeed");
5861 assert_eq!(rs.len(), 1, "only Boston group has cnt > 2");
5862 assert_eq!(rs.get(0, "city"), Some(&s("Boston")));
5863 assert_eq!(rs.get(0, "cnt"), Some(&i(3)));
5864 }
5865
5866 #[test]
5868 fn with_order_limit_then_return() {
5869 let fx = city_graph();
5870 let view = fx.view();
5871 let params = BTreeMap::new();
5872 let rs = run(&view, "MATCH (p:Person) WITH p, p.age AS age ORDER BY age DESC LIMIT 2 RETURN p.city AS city, age", ¶ms)
5874 .expect("WITH ORDER LIMIT must succeed");
5875 assert_eq!(rs.len(), 2, "LIMIT 2");
5876 let ages: Vec<Option<Value>> = (0..rs.len()).map(|i| rs.get(i, "age").cloned()).collect();
5878 assert_eq!(ages[0], Some(i(35)), "oldest person first");
5879 assert_eq!(ages[1], Some(i(30)), "second oldest");
5880 }
5881
5882 #[test]
5884 fn chained_with_stages() {
5885 let fx = city_graph();
5886 let view = fx.view();
5887 let params = BTreeMap::new();
5888 let rs = run(&view,
5890 "MATCH (p:Person) WITH p.city AS city, p.age AS age WITH city, age WHERE age > 25 RETURN city, age",
5891 ¶ms).expect("chained WITH must succeed");
5892 assert_eq!(rs.len(), 3, "3 people with age > 25");
5894 }
5895
5896 #[test]
5898 fn with_then_match_reentry() {
5899 let mut fx = Fx::new();
5900 let alice = fx.add("Person", "alice", vec![]);
5901 let bob = fx.add("Person", "bob", vec![]);
5902 let corp = fx.add("Company", "acme", vec![]);
5903 fx.edge("WORKS_AT", alice, corp, vec![("years", i(5))]);
5904 fx.edge("WORKS_AT", bob, corp, vec![("years", i(3))]);
5905 let view = fx.view();
5906 let params = BTreeMap::new();
5907 let rs = run(&view,
5909 "MATCH (p:Person)-[r:WORKS_AT]->(c:Company) WITH p, c MATCH (c)-[r2:WORKS_AT]-(colleague:Person) RETURN p, colleague",
5910 ¶ms).expect("WITH MATCH re-entry must succeed");
5911 assert!(rs.len() >= 2, "cross join via company: got {}", rs.len());
5913 }
5914
5915 #[test]
5917 fn unwind_literal_list_produces_rows() {
5918 let fx = city_graph();
5919 let view = fx.view();
5920 let params = BTreeMap::new();
5921 let rs = run(
5922 &view,
5923 "MATCH (p:Person) WHERE p.city = 'Boston' UNWIND [1, 2, 3] AS x RETURN p, x",
5924 ¶ms,
5925 )
5926 .expect("UNWIND literal must succeed");
5927 assert_eq!(rs.len(), 9, "UNWIND [1,2,3] × 3 Boston people = 9 rows");
5929 let xs: Vec<Option<Value>> = (0..rs.len())
5931 .map(|row_i| rs.get(row_i, "x").cloned())
5932 .collect();
5933 assert!(xs.contains(&Some(i(1))));
5934 assert!(xs.contains(&Some(i(2))));
5935 assert!(xs.contains(&Some(i(3))));
5936 }
5937
5938 #[test]
5940 fn unwind_list_property() {
5941 let mut fx = Fx::new();
5942 fx.add(
5943 "Tag",
5944 "post1",
5945 vec![("tags", Value::List(vec![s("rust"), s("graph")]))],
5946 );
5947 fx.add("Tag", "post2", vec![("tags", Value::List(vec![s("db")]))]);
5948 let view = fx.view();
5949 let params = BTreeMap::new();
5950 let rs = run(
5951 &view,
5952 "MATCH (p:Tag) UNWIND p.tags AS tag RETURN p, tag",
5953 ¶ms,
5954 )
5955 .expect("UNWIND property must succeed");
5956 assert_eq!(rs.len(), 3, "2+1 tag elements");
5957 }
5958
5959 #[test]
5961 fn unwind_empty_list_yields_zero_rows() {
5962 let fx = city_graph();
5963 let view = fx.view();
5964 let params = BTreeMap::new();
5965 let rs = run(
5966 &view,
5967 "MATCH (p:Person) WHERE p.city = 'Boston' UNWIND [] AS x RETURN x",
5968 ¶ms,
5969 )
5970 .expect("UNWIND [] must succeed with 0 rows");
5971 assert_eq!(rs.len(), 0, "UNWIND [] should produce 0 rows");
5972 }
5973
5974 #[test]
5976 fn unwind_non_list_is_named_error() {
5977 let fx = city_graph();
5978 let view = fx.view();
5979 let params = BTreeMap::new();
5980 let err = run(
5982 &view,
5983 "MATCH (p:Person) WHERE p.city = 'Boston' UNWIND p.city AS x RETURN x",
5984 ¶ms,
5985 )
5986 .expect_err("UNWIND non-list must error");
5987 assert!(
5988 err.contains("UNWIND") && err.contains("list"),
5989 "error must mention UNWIND and list: {err}"
5990 );
5991 }
5992
5993 #[test]
5995 fn unwind_cross_product_trips_budget() {
5996 let mut fx = Fx::new();
5997 for i in 0..10 {
5999 fx.add("N", &format!("n{i}"), vec![]);
6000 }
6001 let view = fx.view();
6002 let params = BTreeMap::new();
6003 let err = super::with_max_intermediate_rows(5, || {
6004 run(
6005 &view,
6006 "MATCH (n:N) UNWIND [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] AS x RETURN n, x",
6007 ¶ms,
6008 )
6009 })
6010 .expect_err("must hit budget");
6011 assert!(
6012 err.contains("intermediate result exceeds"),
6013 "error must mention intermediate result limit: {err}"
6014 );
6015 }
6016
6017 #[test]
6020 fn routing_with_plan_uses_staged_path() {
6021 use crate::cypher::plan::row_bound;
6022 let ops_with = compile("MATCH (a) WITH a RETURN a");
6023 assert_eq!(
6024 row_bound(&ops_with),
6025 None,
6026 "WITH plan must have row_bound=None"
6027 );
6028
6029 let ops_unwind = compile("MATCH (a) UNWIND [1, 2] AS x RETURN a");
6030 assert_eq!(
6031 row_bound(&ops_unwind),
6032 None,
6033 "UNWIND plan must have row_bound=None"
6034 );
6035
6036 let ops_plain = compile("MATCH (a) RETURN a LIMIT 5");
6038 assert!(
6039 row_bound(&ops_plain).is_some(),
6040 "plain LIMIT plan must have a row bound"
6041 );
6042 }
6043
6044 #[test]
6047 fn unwind_null_property_yields_zero_rows_for_that_node() {
6048 let mut fx = Fx::new();
6049 fx.add(
6051 "Post",
6052 "post1",
6053 vec![("tags", Value::List(vec![s("rust"), s("graph")]))],
6054 );
6055 fx.add("Post", "post2", vec![("other", Value::Str("hello".into()))]);
6056 let view = fx.view();
6057 let params = BTreeMap::new();
6058 let rs = run(
6059 &view,
6060 "MATCH (p:Post) UNWIND p.tags AS tag RETURN tag",
6061 ¶ms,
6062 )
6063 .expect("UNWIND null property must succeed (not error)");
6064 assert_eq!(
6066 rs.len(),
6067 2,
6068 "null property must yield 0 rows; list property yields N rows"
6069 );
6070 let tag0 = rs.get(0, "tag");
6071 let tag1 = rs.get(1, "tag");
6072 let tags = [tag0, tag1];
6073 assert!(tags.contains(&Some(&s("rust"))), "rust tag present");
6074 assert!(tags.contains(&Some(&s("graph"))), "graph tag present");
6075 }
6076
6077 #[test]
6079 fn where_before_unwind_filters_nodes() {
6080 let mut fx = Fx::new();
6081 fx.add(
6082 "P",
6083 "a",
6084 vec![
6085 ("group", Value::Str("keep".into())),
6086 ("items", Value::List(vec![Value::Int(1), Value::Int(2)])),
6087 ],
6088 );
6089 fx.add(
6090 "P",
6091 "b",
6092 vec![
6093 ("group", Value::Str("drop".into())),
6094 ("items", Value::List(vec![Value::Int(3), Value::Int(4)])),
6095 ],
6096 );
6097 let view = fx.view();
6098 let params = BTreeMap::new();
6099 let rs = run(
6101 &view,
6102 "MATCH (n:P) WHERE n.group = 'keep' UNWIND n.items AS x RETURN x",
6103 ¶ms,
6104 )
6105 .expect("WHERE before UNWIND must succeed");
6106 assert_eq!(rs.len(), 2, "only matching node expands; 2 items");
6107 }
6108
6109 #[test]
6111 fn where_after_unwind_filters_expanded_rows() {
6112 let mut fx = Fx::new();
6113 fx.add(
6114 "N",
6115 "n1",
6116 vec![(
6117 "vals",
6118 Value::List(vec![Value::Int(1), Value::Int(3), Value::Int(5)]),
6119 )],
6120 );
6121 let view = fx.view();
6122 let params = BTreeMap::new();
6123 let rs = run(
6125 &view,
6126 "MATCH (n:N) UNWIND n.vals AS x WHERE x > 2 RETURN x",
6127 ¶ms,
6128 )
6129 .expect("WHERE after UNWIND must succeed");
6130 assert_eq!(rs.len(), 2, "x=3 and x=5 pass; x=1 filtered out");
6131 let v0 = rs.get(0, "x").cloned();
6132 let v1 = rs.get(1, "x").cloned();
6133 let vals = [v0, v1];
6134 assert!(vals.contains(&Some(Value::Int(3))), "x=3 present");
6135 assert!(vals.contains(&Some(Value::Int(5))), "x=5 present");
6136 }
6137
6138 #[test]
6140 fn aggregate_with_order_by_unknown_alias_is_named_error() {
6141 use crate::cypher::{lex, parser::parse, plan::plan};
6142 let src = "MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt ORDER BY cntt DESC RETURN city, cnt";
6144 let plan_result = plan(&parse(&lex(src).expect("lex")).expect("parse"));
6145 let err = plan_result
6146 .expect_err("typo'd ORDER BY alias in aggregate WITH must be a named plan error");
6147 assert!(
6148 err.contains("cntt") || err.contains("unbound"),
6149 "error must mention the unknown alias: {err}"
6150 );
6151 }
6152
6153 #[test]
6155 fn non_aggregate_with_order_by_unknown_alias_is_named_error() {
6156 use crate::cypher::{lex, parser::parse, plan::plan};
6157 let src = "MATCH (p:Person) WITH p, p.age AS age ORDER BY nope DESC RETURN p.city AS city";
6158 let plan_result = plan(&parse(&lex(src).expect("lex")).expect("parse"));
6159 let err = plan_result
6160 .expect_err("typo'd ORDER BY alias in non-aggregate WITH must be a named plan error");
6161 assert!(
6162 err.contains("nope") || err.contains("unbound"),
6163 "error must mention the unknown alias: {err}"
6164 );
6165 }
6166
6167 #[test]
6171 fn post_unwind_where_references_pre_match_variable() {
6172 let mut fx = Fx::new();
6173 fx.add(
6175 "N",
6176 "n1",
6177 vec![
6178 ("threshold", Value::Int(3)),
6179 (
6180 "xs",
6181 Value::List(vec![
6182 Value::Int(1),
6183 Value::Int(2),
6184 Value::Int(3),
6185 Value::Int(4),
6186 Value::Int(5),
6187 ]),
6188 ),
6189 ],
6190 );
6191 fx.add(
6193 "N",
6194 "n2",
6195 vec![
6196 ("threshold", Value::Int(10)),
6197 ("xs", Value::List(vec![Value::Int(1), Value::Int(2)])),
6198 ],
6199 );
6200 let view = fx.view();
6201 let params = BTreeMap::new();
6202 let rs = run(
6203 &view,
6204 "MATCH (n:N) UNWIND n.xs AS x WHERE x > n.threshold RETURN x",
6205 ¶ms,
6206 )
6207 .expect("post-UNWIND WHERE referencing MATCH variable must succeed");
6208 assert_eq!(rs.len(), 2, "exactly 2 rows: x=4 and x=5 from n1");
6210 let v0 = rs.get(0, "x").cloned();
6211 let v1 = rs.get(1, "x").cloned();
6212 let got = [v0, v1];
6213 assert!(got.contains(&Some(Value::Int(4))), "x=4 present");
6214 assert!(got.contains(&Some(Value::Int(5))), "x=5 present");
6215 }
6216
6217 #[test]
6221 fn with_stage_aggregate_order_by_descending() {
6222 let fx = city_graph();
6225 let view = fx.view();
6226 let params = BTreeMap::new();
6227 let rs = run(
6228 &view,
6229 "MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt ORDER BY cnt DESC RETURN city, cnt",
6230 ¶ms,
6231 )
6232 .expect("aggregate WITH ORDER BY must succeed");
6233 assert_eq!(rs.len(), 2, "two city groups");
6234 assert_eq!(
6236 rs.get(0, "cnt"),
6237 Some(&Value::Int(3)),
6238 "first row must be the group with cnt=3 (Boston)"
6239 );
6240 assert_eq!(
6241 rs.get(1, "cnt"),
6242 Some(&Value::Int(2)),
6243 "second row must be the group with cnt=2 (Austin)"
6244 );
6245 }
6246
6247 #[test]
6251 fn unwind_then_with_composition() {
6252 let mut fx = Fx::new();
6253 fx.add(
6254 "Doc",
6255 "d1",
6256 vec![(
6257 "scores",
6258 Value::List(vec![Value::Int(10), Value::Int(20), Value::Int(30)]),
6259 )],
6260 );
6261 fx.add(
6262 "Doc",
6263 "d2",
6264 vec![("scores", Value::List(vec![Value::Int(5), Value::Int(15)]))],
6265 );
6266 let view = fx.view();
6267 let params = BTreeMap::new();
6268
6269 let rs = run(
6271 &view,
6272 "MATCH (n:Doc) UNWIND n.scores AS x WITH x RETURN x",
6273 ¶ms,
6274 )
6275 .expect("UNWIND then WITH pass-through must succeed");
6276 assert_eq!(rs.len(), 5, "3 + 2 = 5 expanded rows carried through WITH");
6277
6278 let rs2 = run(
6280 &view,
6281 "MATCH (n:Doc) UNWIND n.scores AS x WITH COUNT(*) AS total RETURN total",
6282 ¶ms,
6283 )
6284 .expect("UNWIND then aggregate WITH must succeed");
6285 assert_eq!(rs2.len(), 1, "single aggregate row");
6286 assert_eq!(
6287 rs2.get(0, "total"),
6288 Some(&Value::Int(5)),
6289 "total must be 5 (3+2 expanded rows)"
6290 );
6291 }
6292
6293 #[test]
6299 fn binarith_div_min_over_neg1_does_not_panic() {
6300 let fx = Fx::new();
6301 let view = fx.view();
6302 let vars = VarTable { names: vec![] };
6303 let row: Row = vec![];
6304 let params = BTreeMap::new();
6305
6306 let operand = Operand::BinArith {
6308 op: ArithOp::Div,
6309 left: Box::new(Operand::Lit(Value::Int(i64::MIN))),
6310 right: Box::new(Operand::Lit(Value::Int(-1))),
6311 };
6312
6313 let result = resolve_operand(&view, &vars, &row, &operand, &Params(¶ms));
6314 assert!(
6315 result.is_ok(),
6316 "overflow division must not return Err: {result:?}"
6317 );
6318 assert_eq!(
6319 result.unwrap(),
6320 Some(Value::Int(i64::MAX)),
6321 "i64::MIN / -1 must saturate to i64::MAX, not panic"
6322 );
6323 }
6324
6325 #[test]
6330 fn binarith_missing_param_caught_at_preflight() {
6331 let fx = Fx::new();
6332 let view = fx.view();
6333 let err = run(
6338 &view,
6339 "MATCH (n:X) RETURN abs($missing - 1)",
6340 &BTreeMap::new(),
6341 )
6342 .expect_err("missing param inside BinArith must be caught at preflight");
6343 assert!(
6344 err.contains("missing"),
6345 "error must name the param 'missing', got: {err}"
6346 );
6347 }
6348
6349 #[test]
6353 fn is_null_filters_absent_prop() {
6354 let mut fx = Fx::new();
6355 fx.add("Person", "alice", vec![("age", Value::Int(30))]);
6356 fx.add("Person", "bob", vec![]); let view = fx.view();
6358 let rs = run(
6359 &view,
6360 "MATCH (n:Person) WHERE n.age IS NULL RETURN n",
6361 &BTreeMap::new(),
6362 )
6363 .unwrap();
6364 assert_eq!(rs.len(), 1, "only bob lacks age");
6365 assert_eq!(rs.get(0, "n"), Some(&s("bob")));
6366 }
6367
6368 #[test]
6370 fn is_not_null_filters_present_prop() {
6371 let mut fx = Fx::new();
6372 fx.add("Person", "alice", vec![("age", Value::Int(30))]);
6373 fx.add("Person", "bob", vec![]); let view = fx.view();
6375 let rs = run(
6376 &view,
6377 "MATCH (n:Person) WHERE n.age IS NOT NULL RETURN n",
6378 &BTreeMap::new(),
6379 )
6380 .unwrap();
6381 assert_eq!(rs.len(), 1, "only alice has age");
6382 assert_eq!(rs.get(0, "n"), Some(&s("alice")));
6383 }
6384
6385 #[test]
6388 fn optional_match_is_null_anti_join() {
6389 let mut fx = Fx::new();
6390 let alice = fx.add("Person", "alice", vec![]);
6391 let bob = fx.add("Person", "bob", vec![]);
6392 let carol = fx.add("Person", "carol", vec![]);
6393 fx.edge("KNOWS", alice, carol, vec![]); fx.edge("KNOWS", carol, bob, vec![]); let view = fx.view();
6397 let rs = run(
6398 &view,
6399 "MATCH (a:Person) OPTIONAL MATCH (a)-[:KNOWS]->(b) WITH a, b WHERE b IS NULL RETURN a",
6400 &BTreeMap::new(),
6401 )
6402 .unwrap();
6403 assert_eq!(rs.len(), 1, "only bob has no outgoing KNOWS edge");
6404 assert_eq!(rs.get(0, "a"), Some(&s("bob")));
6405 }
6406
6407 #[test]
6409 fn is_null_combined_with_and_exec() {
6410 let mut fx = Fx::new();
6411 fx.add("N", "a", vec![("x", Value::Int(1))]);
6412 fx.add("N", "b", vec![("x", Value::Int(2)), ("y", Value::Int(9))]);
6413 fx.add("N", "c", vec![]); let view = fx.view();
6415 let rs = run(
6417 &view,
6418 "MATCH (n:N) WHERE n.y IS NULL AND n.x IS NOT NULL RETURN n",
6419 &BTreeMap::new(),
6420 )
6421 .unwrap();
6422 assert_eq!(rs.len(), 1);
6423 assert_eq!(rs.get(0, "n"), Some(&s("a")));
6424 }
6425
6426 #[test]
6430 fn arithmetic_add_in_return() {
6431 let mut fx = Fx::new();
6432 fx.add("Person", "alice", vec![("age", Value::Int(29))]);
6433 let view = fx.view();
6434 let rs = run(
6435 &view,
6436 "MATCH (n:Person) RETURN n.age + 1 AS adjusted",
6437 &BTreeMap::new(),
6438 )
6439 .unwrap();
6440 assert_eq!(rs.len(), 1);
6441 assert_eq!(rs.get(0, "adjusted"), Some(&Value::Int(30)));
6442 }
6443
6444 #[test]
6446 fn arithmetic_precedence_parens_pin() {
6447 let mut fx = Fx::new();
6448 fx.add("N", "x", vec![]);
6449 let view = fx.view();
6450 let rs = run(
6451 &view,
6452 "MATCH (n:N) RETURN (1 + 2) * 3 AS r",
6453 &BTreeMap::new(),
6454 )
6455 .unwrap();
6456 assert_eq!(rs.len(), 1);
6457 assert_eq!(rs.get(0, "r"), Some(&Value::Int(9)));
6458 }
6459
6460 #[test]
6462 fn arithmetic_precedence_mul_over_add_pin() {
6463 let mut fx = Fx::new();
6464 fx.add("N", "x", vec![]);
6465 let view = fx.view();
6466 let rs = run(&view, "MATCH (n:N) RETURN 1 + 2 * 3 AS r", &BTreeMap::new()).unwrap();
6467 assert_eq!(rs.len(), 1);
6468 assert_eq!(rs.get(0, "r"), Some(&Value::Int(7)));
6469 }
6470
6471 #[test]
6473 fn arithmetic_div_by_zero_returns_error() {
6474 let mut fx = Fx::new();
6475 fx.add("N", "x", vec![]);
6476 let view = fx.view();
6477 let err = run(
6478 &view,
6479 "MATCH (n:N) WHERE 1 / 0 > 0 RETURN n",
6480 &BTreeMap::new(),
6481 )
6482 .expect_err("division by zero must error");
6483 assert!(
6484 err.contains("division by zero"),
6485 "error must mention division by zero, got: {err}"
6486 );
6487 }
6488
6489 #[test]
6491 fn arithmetic_null_propagates() {
6492 let mut fx = Fx::new();
6493 fx.add("N", "x", vec![]); let view = fx.view();
6495 let rs = run(
6497 &view,
6498 "MATCH (n:N) WHERE n.val + 1 > 0 RETURN n",
6499 &BTreeMap::new(),
6500 )
6501 .unwrap();
6502 assert_eq!(rs.len(), 0, "null arithmetic must not match");
6503 }
6504
6505 #[test]
6507 fn arithmetic_in_where_comparison() {
6508 let mut fx = Fx::new();
6509 fx.add("Person", "alice", vec![("age", Value::Int(5))]); fx.add("Person", "bob", vec![("age", Value::Int(4))]); let view = fx.view();
6512 let rs = run(
6513 &view,
6514 "MATCH (n:Person) WHERE n.age + 1 > 5 RETURN n",
6515 &BTreeMap::new(),
6516 )
6517 .unwrap();
6518 assert_eq!(rs.len(), 1);
6519 assert_eq!(rs.get(0, "n"), Some(&s("alice")));
6520 }
6521}