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 Ok(Some(Value::Bool(core_storage::fulltext::eval_query_str(
1444 &s, &q,
1445 ))))
1446 }
1447 (Some(Value::List(items)), Some(Value::Str(q))) => Ok(Some(Value::Bool(
1448 core_storage::fulltext::eval_query_str_list(&items, &q),
1449 ))),
1450 _ => Ok(Some(Value::Bool(false))), }
1452 }
1453 _ => Err(format!(
1454 "unknown function `{name}`; supported: {}",
1455 SCALAR_FUNCS.join(", ")
1456 )),
1457 }
1458}
1459
1460fn resolve_operand(
1461 view: &GraphView,
1462 vars: &VarTable,
1463 row: &Row,
1464 operand: &Operand,
1465 params: &Params,
1466) -> Result<Option<Value>, String> {
1467 match operand {
1468 Operand::Lit(v) => Ok(Some(v.clone())),
1469 Operand::Param(name) => match params.0.get(name) {
1470 Some(v) => Ok(Some(v.clone())),
1471 None => Err(format!("missing parameter `{name}`")),
1472 },
1473 Operand::Prop { var, field } => resolve_prop(view, vars, row, var, field),
1474 Operand::Var(name) => {
1475 match vars
1477 .slot(name)
1478 .and_then(|s| row.get(s))
1479 .and_then(|c| c.as_ref())
1480 {
1481 Some(Cell::Scalar(v)) => Ok(Some(v.clone())),
1482 Some(Cell::Node(id)) => match view.ids.key_of(*id) {
1483 Some(key) => Ok(Some(Value::Str(key.to_owned()))),
1484 None => Ok(None),
1485 },
1486 Some(Cell::Path(hops)) => Ok(Some(Value::Int(*hops as i64))),
1487 Some(Cell::Rel(_)) => Err(format!("variable `{name}` is a relationship")),
1488 None => Ok(None),
1489 }
1490 }
1491 Operand::FuncCall { name, args } => eval_func(name, args, view, vars, row, params),
1492 Operand::BinArith { op, left, right } => {
1493 use super::ast::ArithOp;
1494 let lv = resolve_operand(view, vars, row, left, params)?;
1495 let rv = resolve_operand(view, vars, row, right, params)?;
1496 match (lv, rv) {
1497 (None, _) | (_, None) => Ok(None), (Some(Value::Int(a)), Some(Value::Int(b))) => {
1499 let result = match op {
1500 ArithOp::Sub => a.saturating_sub(b),
1501 ArithOp::Mul => a.saturating_mul(b),
1502 ArithOp::Add => a.saturating_add(b),
1503 ArithOp::Div => {
1504 if b == 0 {
1505 return Err("division by zero".into());
1506 }
1507 a.checked_div(b).unwrap_or(i64::MAX)
1508 }
1509 };
1510 Ok(Some(Value::Int(result)))
1511 }
1512 (Some(lv), Some(rv)) => {
1513 let a = match &lv {
1514 Value::Float(f) => *f,
1515 Value::Int(i) => *i as f64,
1516 _ => return Err(format!("arithmetic operand must be numeric, got {lv:?}")),
1517 };
1518 let b = match &rv {
1519 Value::Float(f) => *f,
1520 Value::Int(i) => *i as f64,
1521 _ => return Err(format!("arithmetic operand must be numeric, got {rv:?}")),
1522 };
1523 let result = match op {
1524 ArithOp::Sub => a - b,
1525 ArithOp::Mul => a * b,
1526 ArithOp::Add => a + b,
1527 ArithOp::Div => {
1528 if b == 0.0 {
1529 return Err("division by zero".into());
1530 }
1531 a / b
1532 }
1533 };
1534 Ok(Some(Value::Float(result)))
1535 }
1536 }
1537 }
1538 }
1539}
1540
1541fn resolve_prop(
1542 view: &GraphView,
1543 vars: &VarTable,
1544 row: &Row,
1545 var: &str,
1546 field: &str,
1547) -> Result<Option<Value>, String> {
1548 let slot = vars
1551 .slot(var)
1552 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1553 let cell = match row.get(slot).and_then(|c| c.as_ref()) {
1554 Some(c) => c,
1555 None => return Ok(None),
1557 };
1558 match cell {
1559 Cell::Node(id) => Ok(view.prop(*id, field).map(|vr| vr.into_value())),
1560 Cell::Rel(e) => Ok(view.edge_props.get(e.etype, e.src, e.dst, field)),
1561 Cell::Path(hops) => {
1563 if field == "length" {
1564 Ok(Some(Value::Int(*hops as i64)))
1565 } else {
1566 Ok(None)
1567 }
1568 }
1569 Cell::Scalar(_) => Ok(None),
1571 }
1572}
1573
1574fn node_matches(
1575 view: &GraphView,
1576 vars: &VarTable,
1577 row: &Row,
1578 id: u32,
1579 label: Option<&str>,
1580 props: &[(String, Operand)],
1581 params: &Params,
1582) -> Result<bool, String> {
1583 if let Some(want) = label {
1584 match view.label_of(id) {
1585 Some(got) if got == want => {}
1586 _ => return Ok(false),
1587 }
1588 }
1589 for (field, operand) in props {
1590 let Some(expected) = resolve_operand(view, vars, row, operand, params)? else {
1591 return Ok(false);
1592 };
1593 match view.prop(id, field) {
1594 Some(got) if values_equal(got.as_value(), &expected) => {}
1595 _ => return Ok(false),
1596 }
1597 }
1598 Ok(true)
1599}
1600
1601fn retain_node(
1602 view: &GraphView,
1603 vars: &VarTable,
1604 rows: &[Row],
1605 var: &str,
1606 label: Option<&str>,
1607 props: &[(String, Operand)],
1608 params: &Params,
1609) -> Result<Vec<Row>, String> {
1610 let mut out = Vec::with_capacity(rows.len());
1611 for row in rows {
1612 let id = require_node(row, vars, var)?;
1613 if node_matches(view, vars, row, id, label, props, params)? {
1614 out.push(row.clone());
1615 }
1616 }
1617 Ok(out)
1618}
1619
1620fn map_dir(dir: RelDir) -> Dir {
1621 match dir {
1622 RelDir::Right => Dir::Out,
1623 RelDir::Left => Dir::In,
1624 RelDir::Undirected => Dir::Both,
1625 }
1626}
1627
1628fn neighbor(from: u32, e: &EdgeRef, dir: RelDir) -> u32 {
1629 match dir {
1630 RelDir::Right => e.dst,
1631 RelDir::Left => e.src,
1632 RelDir::Undirected => {
1633 if e.src == from {
1634 e.dst
1635 } else {
1636 e.src
1637 }
1638 }
1639 }
1640}
1641
1642fn row_has_edge(row: &Row, e: &EdgeRef) -> bool {
1643 row.iter()
1644 .any(|c| matches!(c, Some(Cell::Rel(existing)) if existing == e))
1645}
1646
1647fn resolve_etypes(view: &GraphView, etype: Option<&str>) -> Option<Vec<u32>> {
1648 etype.map(|name| view.syms.get(name).into_iter().collect())
1649}
1650
1651fn exec_expand(
1652 view: &GraphView,
1653 vars: &VarTable,
1654 rows: &[Row],
1655 op: &PlanOp,
1656 params: &Params,
1657) -> Result<Vec<Row>, String> {
1658 let PlanOp::Expand {
1659 from,
1660 rel_var,
1661 etype,
1662 dir,
1663 to,
1664 to_label,
1665 to_props,
1666 } = op
1667 else {
1668 return Err("internal: expected Expand".into());
1669 };
1670 let etypes = resolve_etypes(view, etype.as_deref());
1671 let exp_dir = map_dir(*dir);
1672 let to_slot = vars
1673 .slot(to)
1674 .ok_or_else(|| format!("unbound variable `{to}`"))?;
1675 let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
1676 let cap = max_intermediate_rows();
1677 let mut out = Vec::with_capacity(rows.len().saturating_mul(2).min(cap));
1678 for row in rows {
1679 let from_id = require_node(row, vars, from)?;
1680 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
1681 Some(Cell::Node(id)) => Some(*id),
1682 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
1683 return Err(format!("variable `{to}` is not a node"))
1684 }
1685 None => None,
1686 };
1687 for e in expand(view, from_id, etypes.as_deref(), exp_dir) {
1688 if row_has_edge(row, &e) {
1689 continue;
1690 }
1691 let nbr = neighbor(from_id, &e, *dir);
1692 if !view.visible(nbr) {
1693 continue;
1694 }
1695 if let Some(want) = bound_to {
1696 if nbr != want {
1697 continue;
1698 }
1699 }
1700 if !node_matches(view, vars, row, nbr, to_label.as_deref(), to_props, params)? {
1701 continue;
1702 }
1703 if out.len() >= cap {
1704 return Err(row_cap_err(cap));
1705 }
1706 let mut next = row.clone();
1707 if let Some(slot) = rel_slot {
1708 next[slot] = Some(Cell::Rel(e));
1709 }
1710 if bound_to.is_none() {
1711 next[to_slot] = Some(Cell::Node(nbr));
1712 }
1713 out.push(next);
1714 #[cfg(test)]
1715 record_expand_row();
1716 }
1717 }
1718 Ok(out)
1719}
1720
1721#[allow(clippy::too_many_arguments)]
1728fn exec_var_expand(
1729 view: &GraphView,
1730 vars: &VarTable,
1731 rows: &[Row],
1732 from: &str,
1733 rel_var: &Option<String>,
1734 etype: &Option<String>,
1735 dir: RelDir,
1736 to: &str,
1737 min: u8,
1738 max: u8,
1739) -> Result<Vec<Row>, String> {
1740 let etypes = resolve_etypes(view, etype.as_deref());
1741 let exp_dir = map_dir(dir);
1742 let to_slot = vars
1743 .slot(to)
1744 .ok_or_else(|| format!("unbound variable `{to}`"))?;
1745 let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
1746 let cap = max_intermediate_rows();
1747 let mut out: Vec<Row> = Vec::new();
1748
1749 for row in rows {
1750 let from_id = require_node(row, vars, from)?;
1751 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
1752 Some(Cell::Node(id)) => Some(*id),
1753 Some(_) => return Err(format!("variable `{to}` is not a node")),
1754 None => None,
1755 };
1756
1757 struct PathState {
1760 node: u32,
1761 edges: Vec<EdgeRef>,
1762 }
1763
1764 let mut frontier: Vec<PathState> = vec![PathState {
1765 node: from_id,
1766 edges: Vec::new(),
1767 }];
1768
1769 let mut frontier_count: usize = 0;
1774
1775 for depth in 1u8..=max {
1776 let mut next_frontier: Vec<PathState> = Vec::new();
1777 for state in &frontier {
1778 for e in expand(view, state.node, etypes.as_deref(), exp_dir) {
1779 if state.edges.contains(&e) {
1781 continue;
1782 }
1783 let nbr = neighbor(state.node, &e, dir);
1784 if !view.visible(nbr) {
1787 continue;
1788 }
1789
1790 if depth >= min {
1793 let dest_matches = match bound_to {
1794 Some(want) => nbr == want,
1795 None => true,
1796 };
1797 if dest_matches {
1798 if out.len() >= cap {
1799 return Err(row_cap_err(cap));
1800 }
1801 let mut next = row.clone();
1802 if let Some(slot) = rel_slot {
1803 next[slot] = Some(Cell::Path(depth));
1804 }
1805 next[to_slot] = Some(Cell::Node(nbr));
1806 out.push(next);
1807 }
1808 }
1809
1810 if depth < max {
1814 frontier_count += 1;
1815 if frontier_count >= cap {
1816 return Err(row_cap_err(cap));
1817 }
1818 let mut new_edges = state.edges.clone();
1819 new_edges.push(e);
1820 next_frontier.push(PathState {
1821 node: nbr,
1822 edges: new_edges,
1823 });
1824 }
1825 }
1826 }
1827 frontier = next_frontier;
1828 if frontier.is_empty() {
1829 break;
1830 }
1831 }
1832 }
1833 Ok(out)
1834}
1835
1836#[allow(clippy::too_many_arguments)]
1842fn exec_shortest_path(
1843 view: &GraphView,
1844 vars: &VarTable,
1845 rows: &[Row],
1846 from: &str,
1847 rel_var: &Option<String>,
1848 etype: &Option<String>,
1849 dir: RelDir,
1850 to: &str,
1851 max_hops: u8,
1852) -> Result<Vec<Row>, String> {
1853 let etypes = resolve_etypes(view, etype.as_deref());
1854 let exp_dir = map_dir(dir);
1855 let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
1856 let mut out: Vec<Row> = Vec::new();
1857
1858 for row in rows {
1859 let from_id = require_node(row, vars, from)?;
1860 let to_id = require_node(row, vars, to)?;
1861
1862 let mut visited = std::collections::BTreeSet::new();
1864 visited.insert(from_id);
1865 let mut frontier: Vec<u32> = vec![from_id];
1866
1867 'bfs: for depth in 1u8..=max_hops {
1868 let mut next_frontier: Vec<u32> = Vec::new();
1869 for &node in &frontier {
1870 for e in expand(view, node, etypes.as_deref(), exp_dir) {
1871 let nbr = neighbor(node, &e, dir);
1872 if !view.visible(nbr) {
1876 continue;
1877 }
1878 if nbr == to_id {
1879 let mut next = row.clone();
1881 if let Some(slot) = rel_slot {
1882 next[slot] = Some(Cell::Path(depth));
1883 }
1884 out.push(next);
1885 break 'bfs;
1886 }
1887 if !visited.contains(&nbr) {
1888 visited.insert(nbr);
1889 next_frontier.push(nbr);
1890 }
1891 }
1892 }
1893 frontier = next_frontier;
1894 if frontier.is_empty() {
1895 break;
1896 }
1897 }
1898 }
1899 Ok(out)
1900}
1901
1902fn exec_filter(
1903 view: &GraphView,
1904 vars: &VarTable,
1905 rows: &[Row],
1906 expr: &Expr,
1907 params: &Params,
1908) -> Result<Vec<Row>, String> {
1909 let mut out = Vec::with_capacity(rows.len());
1910 for row in rows {
1911 if eval_expr(view, vars, row, expr, params, 0)? {
1912 out.push(row.clone());
1913 }
1914 }
1915 Ok(out)
1916}
1917
1918struct PullCtx<'a> {
1934 view: &'a GraphView<'a>,
1935 vars: &'a VarTable,
1936 project_items: &'a [RetItem],
1937 params: &'a Params<'a>,
1938 bound: usize,
1939}
1940
1941fn execute_pull(
1942 view: &GraphView,
1943 plan: &[PlanOp],
1944 params: &Params,
1945 bound: usize,
1946) -> Result<ResultSet, String> {
1947 let proj_pos = match plan
1948 .iter()
1949 .position(|op| matches!(op, PlanOp::Project { .. }))
1950 {
1951 Some(p) => p,
1952 None => return Ok(ResultSet::new(vec![])),
1953 };
1954 let producers = &plan[..proj_pos];
1955 let project_items = match &plan[proj_pos] {
1956 PlanOp::Project { items } => items,
1957 _ => unreachable!(),
1958 };
1959 let columns: Vec<String> = project_items.iter().map(column_name).collect();
1960 let vars = collect_vars(plan);
1961 let ctx = PullCtx {
1962 view,
1963 vars: &vars,
1964 project_items,
1965 params,
1966 bound,
1967 };
1968 let mut initial_row: Row = vec![None; vars.names.len()];
1969 let mut result_rows: Vec<Vec<Option<Value>>> = Vec::with_capacity(bound);
1970 pull_rows(&ctx, producers, &mut initial_row, &mut result_rows)?;
1971 let skip_n = plan[proj_pos + 1..]
1974 .iter()
1975 .find_map(|op| match op {
1976 PlanOp::Skip(ls) => Some(resolve_ls(ls, params)),
1977 _ => None,
1978 })
1979 .transpose()?
1980 .unwrap_or(0);
1981 let skip_n = usize::try_from(skip_n).unwrap_or(usize::MAX);
1982 let mut rs = ResultSet::new(columns);
1983 for row in result_rows.into_iter().skip(skip_n) {
1984 rs.push_row(row);
1985 }
1986 Ok(rs)
1987}
1988
1989enum AggAcc {
2003 Count(u64),
2004 Sum { val: f64, has_value: bool },
2005 Avg { sum: f64, n: u64 },
2006 Min(Option<Value>),
2007 Max(Option<Value>),
2008}
2009
2010impl AggAcc {
2011 fn new(func: &AggFunc) -> Self {
2012 match func {
2013 AggFunc::Count => AggAcc::Count(0),
2014 AggFunc::Sum => AggAcc::Sum {
2015 val: 0.0,
2016 has_value: false,
2017 },
2018 AggFunc::Avg => AggAcc::Avg { sum: 0.0, n: 0 },
2019 AggFunc::Min => AggAcc::Min(None),
2020 AggFunc::Max => AggAcc::Max(None),
2021 }
2022 }
2023
2024 fn finish(self) -> Option<Value> {
2025 match self {
2026 AggAcc::Count(n) => Some(Value::Int(i64::try_from(n).unwrap_or(i64::MAX))),
2031 AggAcc::Sum { val, has_value } => {
2032 if has_value {
2033 Some(Value::Float(val))
2034 } else {
2035 None
2036 }
2037 }
2038 AggAcc::Avg { sum, n } => {
2039 if n > 0 {
2040 Some(Value::Float(sum / n as f64))
2041 } else {
2042 None
2043 }
2044 }
2045 AggAcc::Min(v) => v,
2046 AggAcc::Max(v) => v,
2047 }
2048 }
2049}
2050
2051fn numeric_val(v: &Value) -> Option<f64> {
2054 match v {
2055 Value::Int(n) => Some(*n as f64),
2056 Value::Float(f) => Some(*f),
2057 _ => None,
2058 }
2059}
2060
2061struct AggStreamCtx<'a> {
2063 view: &'a GraphView<'a>,
2064 vars: &'a VarTable,
2065 params: &'a Params<'a>,
2066 func: &'a AggFunc,
2067 arg: &'a AggArg,
2068}
2069
2070fn execute_aggregate(
2075 view: &GraphView,
2076 plan: &[PlanOp],
2077 params: &Params,
2078) -> Result<ResultSet, String> {
2079 let agg_pos = match plan
2080 .iter()
2081 .position(|op| matches!(op, PlanOp::Aggregate { .. }))
2082 {
2083 Some(p) => p,
2084 None => return Ok(ResultSet::new(vec![])),
2085 };
2086 let producers = &plan[..agg_pos];
2087 let (func, arg, column) = match &plan[agg_pos] {
2088 PlanOp::Aggregate { func, arg, column } => (func, arg, column),
2089 _ => unreachable!(),
2090 };
2091 let vars = collect_vars(plan);
2092 let ctx = AggStreamCtx {
2093 view,
2094 vars: &vars,
2095 params,
2096 func,
2097 arg,
2098 };
2099 let initial_row: Row = vec![None; vars.names.len()];
2100 let mut acc = AggAcc::new(func);
2101 agg_stream(&ctx, producers, &initial_row, &mut acc)?;
2102
2103 let value = acc.finish();
2104 let mut rs = ResultSet::new(vec![column.clone()]);
2105 rs.push_row(vec![value]);
2106 Ok(rs)
2107}
2108
2109fn update_acc(
2115 view: &GraphView,
2116 vars: &VarTable,
2117 row: &Row,
2118 func: &AggFunc,
2119 arg: &AggArg,
2120 acc: &mut AggAcc,
2121) -> Result<(), String> {
2122 match (func, arg) {
2123 (AggFunc::Count, AggArg::Star) => {
2124 if let AggAcc::Count(n) = acc {
2125 *n += 1;
2126 }
2127 }
2128 (AggFunc::Count, AggArg::Var(v)) => {
2129 let slot = vars.slot(v);
2130 let is_bound = slot
2131 .and_then(|s| row.get(s))
2132 .and_then(|c| c.as_ref())
2133 .is_some();
2134 if is_bound {
2135 if let AggAcc::Count(n) = acc {
2136 *n += 1;
2137 }
2138 }
2139 }
2140 (AggFunc::Count, AggArg::Prop { var, field }) => {
2141 let val = resolve_prop(view, vars, row, var, field)?;
2142 if val.is_some() {
2143 if let AggAcc::Count(n) = acc {
2144 *n += 1;
2145 }
2146 }
2147 }
2148 (AggFunc::Sum, AggArg::Prop { var, field }) => {
2149 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2150 if let Some(num) = numeric_val(&v) {
2151 if let AggAcc::Sum { val, has_value } = acc {
2152 *val += num;
2153 *has_value = true;
2154 }
2155 }
2156 }
2157 }
2158 (AggFunc::Avg, AggArg::Prop { var, field }) => {
2159 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2160 if let Some(num) = numeric_val(&v) {
2161 if let AggAcc::Avg { sum, n } = acc {
2162 *sum += num;
2163 *n += 1;
2164 }
2165 }
2166 }
2167 }
2168 (AggFunc::Min, AggArg::Prop { var, field }) => {
2169 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2170 if numeric_val(&v).is_some() {
2171 if let AggAcc::Min(current) = acc {
2172 *current = Some(match current.take() {
2173 None => v,
2174 Some(prev) => {
2175 if cmp_optional(Some(&prev), Some(&v), false)
2176 == std::cmp::Ordering::Greater
2177 {
2178 v
2179 } else {
2180 prev
2181 }
2182 }
2183 });
2184 }
2185 }
2186 }
2187 }
2188 (AggFunc::Max, AggArg::Prop { var, field }) => {
2189 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2190 if numeric_val(&v).is_some() {
2191 if let AggAcc::Max(current) = acc {
2192 *current = Some(match current.take() {
2193 None => v,
2194 Some(prev) => {
2195 if cmp_optional(Some(&prev), Some(&v), true)
2196 == std::cmp::Ordering::Greater
2197 {
2198 v
2199 } else {
2200 prev
2201 }
2202 }
2203 });
2204 }
2205 }
2206 }
2207 }
2208 _ => {}
2211 }
2212 Ok(())
2213}
2214
2215fn agg_stream(
2218 ctx: &AggStreamCtx<'_>,
2219 ops: &[PlanOp],
2220 row: &Row,
2221 acc: &mut AggAcc,
2222) -> Result<(), String> {
2223 let (op, rest) = match ops.split_first() {
2224 Some(pair) => pair,
2225 None => {
2226 update_acc(ctx.view, ctx.vars, row, ctx.func, ctx.arg, acc)?;
2228 return Ok(());
2229 }
2230 };
2231
2232 match op {
2233 PlanOp::ScanLabel { var, label } => {
2234 let ids = scan_ids(ctx.view, label.as_deref());
2235 let slot = ctx
2236 .vars
2237 .slot(var)
2238 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2239 for &id in &ids {
2240 let mut next = row.clone();
2241 next[slot] = Some(Cell::Node(id));
2242 agg_stream(ctx, rest, &next, acc)?;
2243 }
2244 }
2245 PlanOp::ScanKey { var, key, label } => {
2246 let slot = ctx
2247 .vars
2248 .slot(var)
2249 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2250 if let Some(id) =
2251 resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
2252 {
2253 let mut next = row.clone();
2254 next[slot] = Some(Cell::Node(id));
2255 agg_stream(ctx, rest, &next, acc)?;
2256 }
2257 }
2258 PlanOp::Expand {
2259 from,
2260 rel_var,
2261 etype,
2262 dir,
2263 to,
2264 to_label,
2265 to_props,
2266 } => {
2267 let etypes = resolve_etypes(ctx.view, etype.as_deref());
2268 let exp_dir = map_dir(*dir);
2269 let to_slot = ctx
2270 .vars
2271 .slot(to)
2272 .ok_or_else(|| format!("unbound variable `{to}`"))?;
2273 let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
2274 let from_id = require_node(row, ctx.vars, from)?;
2275 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
2276 Some(Cell::Node(id)) => Some(*id),
2277 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
2278 return Err(format!("variable `{to}` is not a node"))
2279 }
2280 None => None,
2281 };
2282 for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
2283 if row_has_edge(row, &e) {
2284 continue;
2285 }
2286 let nbr = neighbor(from_id, &e, *dir);
2287 if !ctx.view.visible(nbr) {
2288 continue;
2289 }
2290 if let Some(want) = bound_to {
2291 if nbr != want {
2292 continue;
2293 }
2294 }
2295 if !node_matches(
2296 ctx.view,
2297 ctx.vars,
2298 row,
2299 nbr,
2300 to_label.as_deref(),
2301 to_props,
2302 ctx.params,
2303 )? {
2304 continue;
2305 }
2306 let mut next = row.clone();
2307 if let Some(slot) = rel_slot {
2308 next[slot] = Some(Cell::Rel(e));
2309 }
2310 if bound_to.is_none() {
2311 next[to_slot] = Some(Cell::Node(nbr));
2312 }
2313 agg_stream(ctx, rest, &next, acc)?;
2314 }
2315 }
2316 PlanOp::Filter { expr } => {
2317 if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
2318 agg_stream(ctx, rest, row, acc)?;
2319 }
2320 }
2321 PlanOp::LookupProps { var, props } => {
2322 let id = require_node(row, ctx.vars, var)?;
2323 if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
2324 agg_stream(ctx, rest, row, acc)?;
2325 }
2326 }
2327 PlanOp::JoinBound { var, label, props } => {
2328 let id = require_node(row, ctx.vars, var)?;
2329 if node_matches(
2330 ctx.view,
2331 ctx.vars,
2332 row,
2333 id,
2334 label.as_deref(),
2335 props,
2336 ctx.params,
2337 )? {
2338 agg_stream(ctx, rest, row, acc)?;
2339 }
2340 }
2341 PlanOp::VarExpand {
2342 from,
2343 rel_var,
2344 etype,
2345 dir,
2346 to,
2347 min,
2348 max,
2349 } => {
2350 let new_rows = exec_var_expand(
2351 ctx.view,
2352 ctx.vars,
2353 std::slice::from_ref(row),
2354 from,
2355 rel_var,
2356 etype,
2357 *dir,
2358 to,
2359 *min,
2360 *max,
2361 )?;
2362 for nr in &new_rows {
2363 agg_stream(ctx, rest, nr, acc)?;
2364 }
2365 }
2366 PlanOp::ShortestPath {
2367 from,
2368 rel_var,
2369 etype,
2370 dir,
2371 to,
2372 max_hops,
2373 } => {
2374 let new_rows = exec_shortest_path(
2375 ctx.view,
2376 ctx.vars,
2377 std::slice::from_ref(row),
2378 from,
2379 rel_var,
2380 etype,
2381 *dir,
2382 to,
2383 *max_hops,
2384 )?;
2385 for nr in &new_rows {
2386 agg_stream(ctx, rest, nr, acc)?;
2387 }
2388 }
2389 PlanOp::Project { .. } => {
2391 return Err(
2392 "agg executor: Project in producer slice — plan is structurally malformed"
2393 .to_string(),
2394 );
2395 }
2396 PlanOp::Distinct => {
2397 return Err(
2398 "agg executor: Distinct in producer slice — plan is structurally malformed"
2399 .to_string(),
2400 );
2401 }
2402 PlanOp::OrderBy { .. } => {
2403 return Err(
2404 "agg executor: OrderBy in producer slice — structurally malformed".to_string(),
2405 );
2406 }
2407 PlanOp::Skip(_) => {
2408 return Err(
2409 "agg executor: Skip in producer slice — structurally malformed".to_string(),
2410 );
2411 }
2412 PlanOp::Limit(_) => {
2413 return Err(
2414 "agg executor: Limit in producer slice — structurally malformed".to_string(),
2415 );
2416 }
2417 PlanOp::Aggregate { .. } => {
2418 return Err(
2419 "agg executor: nested Aggregate in producer slice — structurally malformed"
2420 .to_string(),
2421 );
2422 }
2423 PlanOp::GroupAggregate { .. } => {
2424 return Err(
2425 "agg executor: GroupAggregate in producer slice — structurally malformed"
2426 .to_string(),
2427 );
2428 }
2429 PlanOp::With { .. } => {
2430 return Err(
2431 "agg executor: With in producer slice — structurally malformed".to_string(),
2432 );
2433 }
2434 PlanOp::Unwind { .. } => {
2435 return Err(
2436 "agg executor: Unwind in producer slice — structurally malformed".to_string(),
2437 );
2438 }
2439 PlanOp::LeftOuterApply { .. } => {
2440 return Err(
2441 "agg executor: LeftOuterApply in producer slice — structurally malformed"
2442 .to_string(),
2443 );
2444 }
2445 }
2446 Ok(())
2447}
2448
2449struct GroupStreamCtx<'a> {
2457 view: &'a GraphView<'a>,
2458 vars: &'a VarTable,
2459 params: &'a Params<'a>,
2460 keys: &'a [(String, RetItem)],
2461 aggs: &'a [(AggFunc, AggArg, String)],
2462}
2463
2464fn build_group_projected(
2466 keys: &[(String, RetItem)],
2467 aggs: &[(AggFunc, AggArg, String)],
2468 key_order: Vec<GroupKey>,
2469 groups: &mut HashMap<GroupKey, GroupEntry>,
2470) -> Projected {
2471 let columns: Vec<String> = keys
2472 .iter()
2473 .map(|(col, _)| col.clone())
2474 .chain(aggs.iter().map(|(_, _, col)| col.clone()))
2475 .collect();
2476 let mut rows: Vec<Vec<Option<Value>>> = Vec::with_capacity(key_order.len());
2477 for gk in key_order {
2478 let (display_keys, accs) = groups.remove(&gk).unwrap_or_default();
2479 let mut row: Vec<Option<Value>> = Vec::with_capacity(columns.len());
2480 for display_val in display_keys {
2483 row.push(display_val);
2484 }
2485 for acc in accs {
2486 row.push(acc.finish());
2487 }
2488 rows.push(row);
2489 }
2490 Projected { columns, rows }
2491}
2492
2493fn group_result_to_rows(
2500 keys: &[(String, RetItem)],
2501 aggs: &[(AggFunc, AggArg, String)],
2502 key_order: Vec<GroupKey>,
2503 groups: &mut HashMap<GroupKey, GroupEntry>,
2504 vars: &VarTable,
2505) -> Vec<Row> {
2506 let row_len = vars.names.len();
2507 let mut out: Vec<Row> = Vec::with_capacity(key_order.len());
2508 for gk in key_order {
2509 let (display_keys, accs) = groups.remove(&gk).unwrap_or_default();
2510 let mut row: Row = vec![None; row_len];
2511 for ((col, _), val) in keys.iter().zip(display_keys) {
2512 if let Some(slot) = vars.slot(col) {
2513 row[slot] = val.map(Cell::Scalar);
2514 }
2515 }
2516 for ((_, _, col), acc) in aggs.iter().zip(accs) {
2517 if let Some(slot) = vars.slot(col) {
2518 row[slot] = acc.finish().map(Cell::Scalar);
2519 }
2520 }
2521 out.push(row);
2522 }
2523 out
2524}
2525
2526fn exec_order_by_rows(vars: &VarTable, rows: &mut Vec<Row>, items: &[OrderItem], view: &GraphView) {
2533 let mut key_table: Vec<Vec<Option<Value>>> = Vec::with_capacity(rows.len());
2536 for row in rows.iter() {
2537 let mut row_key: Vec<Option<Value>> = Vec::with_capacity(items.len());
2538 for item in items {
2539 let val = match &item.target {
2540 OrderTarget::Alias(name) | OrderTarget::Var(name) => vars
2541 .slot(name)
2542 .and_then(|s| row.get(s))
2543 .and_then(|c| c.as_ref())
2544 .and_then(|c| match c {
2545 Cell::Scalar(v) => Some(v.clone()),
2546 Cell::Node(id) => view.ids.key_of(*id).map(|k| Value::Str(k.to_owned())),
2547 Cell::Path(hops) => Some(Value::Int(*hops as i64)),
2548 Cell::Rel(_) => None,
2549 }),
2550 OrderTarget::Prop { var, field } => vars
2551 .slot(var)
2552 .and_then(|s| row.get(s))
2553 .and_then(|c| c.as_ref())
2554 .and_then(|c| match c {
2555 Cell::Node(id) => view.prop(*id, field).map(|vr| vr.into_value()),
2556 Cell::Rel(e) => view.edge_props.get(e.etype, e.src, e.dst, field),
2557 _ => None,
2558 }),
2559 };
2560 row_key.push(val);
2561 }
2562 key_table.push(row_key);
2563 }
2564 let mut indices: Vec<usize> = (0..rows.len()).collect();
2566 indices.sort_by(|&a, &b| {
2567 for (ki, item) in items.iter().enumerate() {
2568 let c = cmp_optional(
2569 key_table[a].get(ki).and_then(|x| x.as_ref()),
2570 key_table[b].get(ki).and_then(|x| x.as_ref()),
2571 item.descending,
2572 );
2573 if c != std::cmp::Ordering::Equal {
2574 return c;
2575 }
2576 }
2577 std::cmp::Ordering::Equal
2578 });
2579 let sorted: Vec<Row> = indices.into_iter().map(|i| rows[i].clone()).collect();
2580 *rows = sorted;
2581}
2582
2583fn execute_group_aggregate(
2589 view: &GraphView,
2590 plan: &[PlanOp],
2591 params: &Params,
2592) -> Result<ResultSet, String> {
2593 let gagg_pos = plan
2594 .iter()
2595 .position(|op| matches!(op, PlanOp::GroupAggregate { .. }))
2596 .ok_or_else(|| "internal: GroupAggregate op not found in plan".to_string())?;
2597 let producers = &plan[..gagg_pos];
2598 let (keys, aggs) = match &plan[gagg_pos] {
2599 PlanOp::GroupAggregate { keys, aggs } => (keys, aggs),
2600 _ => unreachable!(),
2601 };
2602 let tail = &plan[gagg_pos + 1..];
2603 let vars = collect_vars(plan);
2604 let initial_row: Row = vec![None; vars.names.len()];
2605 let mut groups: HashMap<GroupKey, GroupEntry> = HashMap::new();
2606 let mut key_order: Vec<GroupKey> = Vec::new();
2607 let ctx = GroupStreamCtx {
2608 view,
2609 vars: &vars,
2610 params,
2611 keys,
2612 aggs,
2613 };
2614 group_stream(&ctx, producers, &initial_row, &mut groups, &mut key_order)?;
2615 if keys.is_empty() && key_order.is_empty() {
2619 let empty_key: GroupKey = vec![];
2620 key_order.push(empty_key.clone());
2621 groups.insert(
2622 empty_key,
2623 (
2624 vec![],
2625 aggs.iter().map(|(f, _, _)| AggAcc::new(f)).collect(),
2626 ),
2627 );
2628 }
2629 let mut projected = build_group_projected(keys, aggs, key_order, &mut groups);
2630 for op in tail {
2632 match op {
2633 PlanOp::OrderBy { items } => exec_order_by(&mut projected, items)?,
2634 PlanOp::Skip(ls) => {
2635 let n = resolve_ls(ls, params)?;
2636 apply_skip(&mut projected.rows, n);
2637 }
2638 PlanOp::Limit(ls) => {
2639 let n = resolve_ls(ls, params)?;
2640 apply_limit(&mut projected.rows, n);
2641 }
2642 _ => {} }
2644 }
2645 Ok(finish(projected))
2646}
2647
2648fn group_stream(
2655 ctx: &GroupStreamCtx<'_>,
2656 ops: &[PlanOp],
2657 row: &Row,
2658 groups: &mut HashMap<GroupKey, GroupEntry>,
2659 key_order: &mut Vec<GroupKey>,
2660) -> Result<(), String> {
2661 let (op, rest) = match ops.split_first() {
2662 Some(pair) => pair,
2663 None => {
2664 let mut gk: GroupKey = Vec::with_capacity(ctx.keys.len());
2667 let mut display_vals: Vec<Option<Value>> = Vec::with_capacity(ctx.keys.len());
2668 for (_, item) in ctx.keys {
2669 let val = project_item(ctx.view, ctx.vars, row, item, ctx.params)?;
2670 gk.push(val.as_ref().and_then(group_key_normalize));
2671 display_vals.push(val);
2672 }
2673 if !groups.contains_key(&gk) {
2674 if groups.len() >= max_groups() {
2675 return Err(group_cap_err());
2676 }
2677 key_order.push(gk.clone());
2678 let init: Vec<AggAcc> = ctx.aggs.iter().map(|(f, _, _)| AggAcc::new(f)).collect();
2679 groups.insert(gk.clone(), (display_vals, init));
2680 }
2681 let (_, accs) = groups.get_mut(&gk).unwrap();
2682 for (acc, (func, arg, _)) in accs.iter_mut().zip(ctx.aggs.iter()) {
2683 update_acc(ctx.view, ctx.vars, row, func, arg, acc)?;
2684 }
2685 return Ok(());
2686 }
2687 };
2688
2689 match op {
2690 PlanOp::ScanLabel { var, label } => {
2691 let ids = scan_ids(ctx.view, label.as_deref());
2692 let slot = ctx
2693 .vars
2694 .slot(var)
2695 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2696 for &id in &ids {
2697 let mut next = row.clone();
2698 next[slot] = Some(Cell::Node(id));
2699 group_stream(ctx, rest, &next, groups, key_order)?;
2700 }
2701 }
2702 PlanOp::ScanKey { var, key, label } => {
2703 let slot = ctx
2704 .vars
2705 .slot(var)
2706 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2707 if let Some(id) =
2708 resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
2709 {
2710 let mut next = row.clone();
2711 next[slot] = Some(Cell::Node(id));
2712 group_stream(ctx, rest, &next, groups, key_order)?;
2713 }
2714 }
2715 PlanOp::Expand {
2716 from,
2717 rel_var,
2718 etype,
2719 dir,
2720 to,
2721 to_label,
2722 to_props,
2723 } => {
2724 let etypes = resolve_etypes(ctx.view, etype.as_deref());
2725 let exp_dir = map_dir(*dir);
2726 let to_slot = ctx
2727 .vars
2728 .slot(to)
2729 .ok_or_else(|| format!("unbound variable `{to}`"))?;
2730 let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
2731 let from_id = require_node(row, ctx.vars, from)?;
2732 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
2733 Some(Cell::Node(id)) => Some(*id),
2734 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
2735 return Err(format!("variable `{to}` is not a node"))
2736 }
2737 None => None,
2738 };
2739 for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
2740 if row_has_edge(row, &e) {
2741 continue;
2742 }
2743 let nbr = neighbor(from_id, &e, *dir);
2744 if !ctx.view.visible(nbr) {
2745 continue;
2746 }
2747 if let Some(want) = bound_to {
2748 if nbr != want {
2749 continue;
2750 }
2751 }
2752 if !node_matches(
2753 ctx.view,
2754 ctx.vars,
2755 row,
2756 nbr,
2757 to_label.as_deref(),
2758 to_props,
2759 ctx.params,
2760 )? {
2761 continue;
2762 }
2763 let mut next = row.clone();
2764 if let Some(slot) = rel_slot {
2765 next[slot] = Some(Cell::Rel(e));
2766 }
2767 if bound_to.is_none() {
2768 next[to_slot] = Some(Cell::Node(nbr));
2769 }
2770 group_stream(ctx, rest, &next, groups, key_order)?;
2771 }
2772 }
2773 PlanOp::Filter { expr } => {
2774 if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
2775 group_stream(ctx, rest, row, groups, key_order)?;
2776 }
2777 }
2778 PlanOp::LookupProps { var, props } => {
2779 let id = require_node(row, ctx.vars, var)?;
2780 if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
2781 group_stream(ctx, rest, row, groups, key_order)?;
2782 }
2783 }
2784 PlanOp::JoinBound { var, label, props } => {
2785 let id = require_node(row, ctx.vars, var)?;
2786 if node_matches(
2787 ctx.view,
2788 ctx.vars,
2789 row,
2790 id,
2791 label.as_deref(),
2792 props,
2793 ctx.params,
2794 )? {
2795 group_stream(ctx, rest, row, groups, key_order)?;
2796 }
2797 }
2798 PlanOp::VarExpand {
2799 from,
2800 rel_var,
2801 etype,
2802 dir,
2803 to,
2804 min,
2805 max,
2806 } => {
2807 let new_rows = exec_var_expand(
2808 ctx.view,
2809 ctx.vars,
2810 std::slice::from_ref(row),
2811 from,
2812 rel_var,
2813 etype,
2814 *dir,
2815 to,
2816 *min,
2817 *max,
2818 )?;
2819 for nr in &new_rows {
2820 group_stream(ctx, rest, nr, groups, key_order)?;
2821 }
2822 }
2823 PlanOp::ShortestPath {
2824 from,
2825 rel_var,
2826 etype,
2827 dir,
2828 to,
2829 max_hops,
2830 } => {
2831 let new_rows = exec_shortest_path(
2832 ctx.view,
2833 ctx.vars,
2834 std::slice::from_ref(row),
2835 from,
2836 rel_var,
2837 etype,
2838 *dir,
2839 to,
2840 *max_hops,
2841 )?;
2842 for nr in &new_rows {
2843 group_stream(ctx, rest, nr, groups, key_order)?;
2844 }
2845 }
2846 PlanOp::Project { .. } => {
2848 return Err(
2849 "group executor: Project in producer slice — structurally malformed".to_string(),
2850 );
2851 }
2852 PlanOp::Distinct => {
2853 return Err(
2854 "group executor: Distinct in producer slice — structurally malformed".to_string(),
2855 );
2856 }
2857 PlanOp::OrderBy { .. } => {
2858 return Err(
2859 "group executor: OrderBy in producer slice — structurally malformed".to_string(),
2860 );
2861 }
2862 PlanOp::Skip(_) => {
2863 return Err(
2864 "group executor: Skip in producer slice — structurally malformed".to_string(),
2865 );
2866 }
2867 PlanOp::Limit(_) => {
2868 return Err(
2869 "group executor: Limit in producer slice — structurally malformed".to_string(),
2870 );
2871 }
2872 PlanOp::Aggregate { .. } => {
2873 return Err(
2874 "group executor: Aggregate in producer slice — structurally malformed".to_string(),
2875 );
2876 }
2877 PlanOp::GroupAggregate { .. } => {
2878 return Err(
2879 "group executor: nested GroupAggregate in producer slice — structurally malformed"
2880 .to_string(),
2881 );
2882 }
2883 PlanOp::With { .. } => {
2884 return Err(
2885 "group executor: With in producer slice — structurally malformed".to_string(),
2886 );
2887 }
2888 PlanOp::Unwind { .. } => {
2889 return Err(
2890 "group executor: Unwind in producer slice — structurally malformed".to_string(),
2891 );
2892 }
2893 PlanOp::LeftOuterApply { .. } => {
2894 return Err(
2895 "group executor: LeftOuterApply in producer slice — structurally malformed"
2896 .to_string(),
2897 );
2898 }
2899 }
2900 Ok(())
2901}
2902
2903fn pull_rows(
2917 ctx: &PullCtx<'_>,
2918 ops: &[PlanOp],
2919 row: &mut Row,
2920 result: &mut Vec<Vec<Option<Value>>>,
2921) -> Result<(), String> {
2922 if result.len() >= ctx.bound {
2923 return Ok(());
2924 }
2925 let (op, rest) = match ops.split_first() {
2926 Some(pair) => pair,
2927 None => {
2928 let mut cells = Vec::with_capacity(ctx.project_items.len());
2930 for item in ctx.project_items {
2931 cells.push(project_item(ctx.view, ctx.vars, row, item, ctx.params)?);
2932 }
2933 result.push(cells);
2934 return Ok(());
2935 }
2936 };
2937 match op {
2938 PlanOp::ScanLabel { var, label } => {
2939 let slot = ctx
2940 .vars
2941 .slot(var)
2942 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2943 let want_sym = label.as_deref().and_then(|l| ctx.view.syms.get(l));
2946 if label.is_some() && want_sym.is_none() {
2947 return Ok(());
2948 }
2949 let prev = row[slot].clone();
2951
2952 let fused_filter = rest.first().and_then(|next_op| {
2958 if let PlanOp::Filter {
2959 expr:
2960 Expr::Cmp {
2961 lhs:
2962 Operand::Prop {
2963 var: ref fv,
2964 field: ref f,
2965 },
2966 op: ref cmp_op_ref,
2967 rhs: Operand::Lit(ref lit),
2968 },
2969 } = *next_op
2970 {
2971 if fv == var {
2972 return Some((f.as_str(), cmp_op_ref, lit));
2973 }
2974 }
2975 None
2976 });
2977
2978 if let Some((field, cmp_op_ref, lit)) = fused_filter {
2979 #[cfg(test)]
2982 FUSED_SCAN_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2983 let col = ctx.view.props.column(field);
2984 let rest_after_filter = &rest[1..];
2985 for (i, &sym) in ctx.view.labels.iter().enumerate() {
2986 if result.len() >= ctx.bound {
2987 break;
2988 }
2989 if sym == u32::MAX {
2990 continue;
2991 }
2992 if let Some(ws) = want_sym {
2993 if sym != ws {
2994 continue;
2995 }
2996 }
2997 let id = i as u32;
2998 if !ctx.view.visible(id) {
2999 continue;
3000 }
3001 if let Some(v) = col.get(id) {
3002 if eval_cmp(cmp_op_ref, v, lit) {
3003 row[slot] = Some(Cell::Node(id));
3004 pull_rows(ctx, rest_after_filter, row, result)?;
3005 }
3006 }
3007 }
3008 } else {
3009 for (i, &sym) in ctx.view.labels.iter().enumerate() {
3012 if result.len() >= ctx.bound {
3013 break;
3014 }
3015 if sym == u32::MAX {
3017 continue;
3018 }
3019 if let Some(ws) = want_sym {
3021 if sym != ws {
3022 continue;
3023 }
3024 }
3025 let id = i as u32;
3026 if !ctx.view.visible(id) {
3027 continue;
3028 }
3029 row[slot] = Some(Cell::Node(id));
3030 pull_rows(ctx, rest, row, result)?;
3031 }
3032 }
3033 row[slot] = prev;
3039 }
3040 PlanOp::ScanKey { var, key, label } => {
3041 let slot = ctx
3042 .vars
3043 .slot(var)
3044 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3045 if let Some(id) =
3046 resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
3047 {
3048 let prev = row[slot].clone();
3049 row[slot] = Some(Cell::Node(id));
3050 pull_rows(ctx, rest, row, result)?;
3051 row[slot] = prev;
3052 }
3053 }
3054 PlanOp::Expand {
3055 from,
3056 rel_var,
3057 etype,
3058 dir,
3059 to,
3060 to_label,
3061 to_props,
3062 } => {
3063 let etypes = resolve_etypes(ctx.view, etype.as_deref());
3064 let exp_dir = map_dir(*dir);
3065 let to_slot = ctx
3066 .vars
3067 .slot(to)
3068 .ok_or_else(|| format!("unbound variable `{to}`"))?;
3069 let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
3070 let from_id = require_node(row, ctx.vars, from)?;
3071 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
3072 Some(Cell::Node(id)) => Some(*id),
3073 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
3074 return Err(format!("variable `{to}` is not a node"))
3075 }
3076 None => None,
3077 };
3078 for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
3079 if result.len() >= ctx.bound {
3080 break;
3081 }
3082 if row_has_edge(row, &e) {
3083 continue;
3084 }
3085 let nbr = neighbor(from_id, &e, *dir);
3086 if !ctx.view.visible(nbr) {
3087 continue;
3088 }
3089 if let Some(want) = bound_to {
3090 if nbr != want {
3091 continue;
3092 }
3093 }
3094 if !node_matches(
3095 ctx.view,
3096 ctx.vars,
3097 row,
3098 nbr,
3099 to_label.as_deref(),
3100 to_props,
3101 ctx.params,
3102 )? {
3103 continue;
3104 }
3105 let mut next = row.clone();
3106 if let Some(slot) = rel_slot {
3107 next[slot] = Some(Cell::Rel(e));
3108 }
3109 if bound_to.is_none() {
3110 next[to_slot] = Some(Cell::Node(nbr));
3111 }
3112 #[cfg(test)]
3113 record_expand_row();
3114 pull_rows(ctx, rest, &mut next, result)?;
3115 }
3116 }
3117 PlanOp::Filter { expr } => {
3118 if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
3119 pull_rows(ctx, rest, row, result)?;
3120 }
3121 }
3122 PlanOp::LookupProps { var, props } => {
3123 let id = require_node(row, ctx.vars, var)?;
3124 if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
3125 pull_rows(ctx, rest, row, result)?;
3126 }
3127 }
3128 PlanOp::JoinBound { var, label, props } => {
3129 let id = require_node(row, ctx.vars, var)?;
3130 if node_matches(
3131 ctx.view,
3132 ctx.vars,
3133 row,
3134 id,
3135 label.as_deref(),
3136 props,
3137 ctx.params,
3138 )? {
3139 pull_rows(ctx, rest, row, result)?;
3140 }
3141 }
3142 PlanOp::Project { .. } => {
3147 return Err(
3148 "pull executor: Project reached pull_rows — plan is structurally malformed"
3149 .to_string(),
3150 );
3151 }
3152 PlanOp::Distinct => {
3153 return Err(
3154 "pull executor: Distinct reached pull_rows — DISTINCT queries must use \
3155 the staged path (row_bound returns None)"
3156 .to_string(),
3157 );
3158 }
3159 PlanOp::OrderBy { .. } => {
3160 return Err(
3161 "pull executor: OrderBy reached pull_rows — queries with ORDER BY \
3162 must use the staged path (row_bound returns None)"
3163 .to_string(),
3164 );
3165 }
3166 PlanOp::Skip(_) => {
3167 return Err(
3168 "pull executor: Skip reached pull_rows — Skip must appear after Project"
3169 .to_string(),
3170 );
3171 }
3172 PlanOp::Limit(_) => {
3173 return Err(
3174 "pull executor: Limit reached pull_rows — Limit must appear after Project"
3175 .to_string(),
3176 );
3177 }
3178 PlanOp::Aggregate { .. } => {
3179 return Err(
3180 "pull executor: Aggregate reached pull_rows — aggregate plans must use \
3181 the execute_aggregate path (routed before pull in execute_inner)"
3182 .to_string(),
3183 );
3184 }
3185 PlanOp::VarExpand { .. } => {
3190 return Err(
3191 "pull executor: VarExpand reached pull_rows — variable-length path \
3192 plans must use the staged path (row_bound returns None)"
3193 .to_string(),
3194 );
3195 }
3196 PlanOp::ShortestPath { .. } => {
3197 return Err(
3198 "pull executor: ShortestPath reached pull_rows — shortestPath plans \
3199 must use the staged path (row_bound returns None)"
3200 .to_string(),
3201 );
3202 }
3203 PlanOp::GroupAggregate { .. } => {
3204 return Err(
3205 "pull executor: GroupAggregate reached pull_rows — grouped aggregate plans \
3206 must use the execute_group_aggregate path (routed before pull in execute_inner)"
3207 .to_string(),
3208 );
3209 }
3210 PlanOp::With { .. } => {
3211 return Err(
3212 "pull executor: With reached pull_rows — pipeline plans must use the staged path \
3213 (row_bound returns None for plans containing With)"
3214 .to_string(),
3215 );
3216 }
3217 PlanOp::Unwind { .. } => {
3218 return Err(
3219 "pull executor: Unwind reached pull_rows — pipeline plans must use the staged path \
3220 (row_bound returns None for plans containing Unwind)"
3221 .to_string(),
3222 );
3223 }
3224 PlanOp::LeftOuterApply { .. } => {
3225 return Err(
3226 "pull executor: LeftOuterApply reached pull_rows — OPTIONAL MATCH plans must use \
3227 the staged path (row_bound returns None for plans containing LeftOuterApply)"
3228 .to_string(),
3229 );
3230 }
3231 }
3232 Ok(())
3233}
3234
3235fn eval_expr(
3236 view: &GraphView,
3237 vars: &VarTable,
3238 row: &Row,
3239 expr: &Expr,
3240 params: &Params,
3241 depth: u32,
3242) -> Result<bool, String> {
3243 if depth > 256 {
3244 return Err("expression nesting too deep".into());
3245 }
3246 match expr {
3247 Expr::And(lhs, rhs) => {
3248 let l = eval_expr(view, vars, row, lhs, params, depth + 1)?;
3249 let r = eval_expr(view, vars, row, rhs, params, depth + 1)?;
3250 Ok(l && r)
3251 }
3252 Expr::Or(lhs, rhs) => {
3253 let l = eval_expr(view, vars, row, lhs, params, depth + 1)?;
3254 let r = eval_expr(view, vars, row, rhs, params, depth + 1)?;
3255 Ok(l || r)
3256 }
3257 Expr::Not(inner) => Ok(!eval_expr(view, vars, row, inner, params, depth + 1)?),
3258 Expr::Cmp { lhs, op, rhs } => {
3259 let l = resolve_operand(view, vars, row, lhs, params)?;
3260 let r = resolve_operand(view, vars, row, rhs, params)?;
3261 match (l, r) {
3262 (Some(a), Some(b)) => Ok(eval_cmp(op, &a, &b)),
3263 _ => Ok(false),
3264 }
3265 }
3266 Expr::Truthy(op) => {
3267 let val = resolve_operand(view, vars, row, op, params)?;
3268 Ok(match val {
3269 None => false,
3270 Some(Value::Bool(b)) => b,
3271 Some(Value::Int(n)) => n != 0,
3272 Some(Value::Float(f)) => f != 0.0,
3273 Some(Value::Str(s)) => !s.is_empty(),
3274 Some(Value::List(v)) => !v.is_empty(),
3275 Some(Value::Map(m)) => !m.is_empty(),
3276 })
3277 }
3278 Expr::IsNull(op) => {
3279 let val = resolve_operand(view, vars, row, op, params)?;
3280 Ok(val.is_none())
3281 }
3282 Expr::IsNotNull(op) => {
3283 let val = resolve_operand(view, vars, row, op, params)?;
3284 Ok(val.is_some())
3285 }
3286 Expr::In { expr, list } => eval_in(view, vars, row, expr, list, params),
3287 }
3288}
3289
3290fn eval_in(
3291 view: &GraphView,
3292 vars: &VarTable,
3293 row: &Row,
3294 expr: &Operand,
3295 list: &[Operand],
3296 params: &Params,
3297) -> Result<bool, String> {
3298 let Some(needle) = resolve_operand(view, vars, row, expr, params)? else {
3299 return Ok(false);
3300 };
3301 for item_op in list {
3302 match resolve_operand(view, vars, row, item_op, params)? {
3303 None => {}
3304 Some(Value::List(items)) => {
3305 for item in items {
3306 if crate::filter::eval_cmp(&crate::filter::CmpOp::Eq, &needle, &item) {
3307 return Ok(true);
3308 }
3309 }
3310 }
3311 Some(item) if crate::filter::eval_cmp(&crate::filter::CmpOp::Eq, &needle, &item) => {
3312 return Ok(true);
3313 }
3314 Some(_) => {}
3315 }
3316 }
3317 Ok(false)
3318}
3319
3320fn exec_distinct(table: &mut Projected) -> Result<(), String> {
3322 let cap = max_intermediate_rows();
3323 let mut seen: BTreeSet<Vec<Option<ValueKey>>> = BTreeSet::new();
3324 let mut out = Vec::with_capacity(table.rows.len().min(cap));
3325 for row in table.rows.drain(..) {
3326 let key: Vec<Option<ValueKey>> = row
3327 .iter()
3328 .map(|cell| cell.as_ref().and_then(group_key_normalize))
3329 .collect();
3330 if seen.insert(key) {
3331 if out.len() >= cap {
3332 return Err(row_cap_err(cap));
3333 }
3334 out.push(row);
3335 }
3336 }
3337 table.rows = out;
3338 Ok(())
3339}
3340
3341fn column_name(item: &RetItem) -> String {
3342 if let Some(alias) = &item.alias {
3343 return alias.clone();
3344 }
3345 match &item.value {
3346 RetVal::Var(v) => v.clone(),
3347 RetVal::Prop { var, field } => format!("{var}.{field}"),
3348 RetVal::Agg { func, arg } => {
3351 use crate::cypher::ast::AggArg;
3352 let f = match func {
3353 AggFunc::Count => "COUNT",
3354 AggFunc::Sum => "SUM",
3355 AggFunc::Avg => "AVG",
3356 AggFunc::Min => "MIN",
3357 AggFunc::Max => "MAX",
3358 };
3359 let a = match arg {
3360 AggArg::Star => "*".to_string(),
3361 AggArg::Var(v) => v.clone(),
3362 AggArg::Prop { var, field } => format!("{var}.{field}"),
3363 };
3364 format!("{f}({a})")
3365 }
3366 RetVal::FuncCall { name, args } => {
3367 let arg_strs: Vec<String> = args
3368 .iter()
3369 .map(|a| match a {
3370 Operand::Var(v) => v.clone(),
3371 Operand::Prop { var, field } => format!("{var}.{field}"),
3372 Operand::Lit(_) => "<lit>".to_string(),
3373 Operand::Param(p) => format!("${p}"),
3374 Operand::FuncCall { name: n, .. } => format!("{n}(...)"),
3375 Operand::BinArith { .. } => "<arith>".to_string(),
3376 })
3377 .collect();
3378 format!("{name}({})", arg_strs.join(", "))
3379 }
3380 RetVal::ScalarExpr(_) => "<expr>".to_string(),
3381 }
3382}
3383
3384fn exec_project(
3385 view: &GraphView,
3386 vars: &VarTable,
3387 rows: &[Row],
3388 items: &[RetItem],
3389 params: &Params,
3390) -> Result<Projected, String> {
3391 let columns: Vec<String> = items.iter().map(column_name).collect();
3392 let mut out_rows = Vec::with_capacity(rows.len());
3393 for row in rows {
3394 let mut cells = Vec::with_capacity(items.len());
3395 for item in items {
3396 cells.push(project_item(view, vars, row, item, params)?);
3397 }
3398 out_rows.push(cells);
3399 }
3400 Ok(Projected {
3401 columns,
3402 rows: out_rows,
3403 })
3404}
3405
3406fn project_item(
3407 view: &GraphView,
3408 vars: &VarTable,
3409 row: &Row,
3410 item: &RetItem,
3411 params: &Params,
3412) -> Result<Option<Value>, String> {
3413 match &item.value {
3414 RetVal::Var(v) => {
3415 let slot = vars.slot(v).ok_or_else(|| format!("unbound variable `{v}`"))?;
3417 match row.get(slot).and_then(|c| c.as_ref()) {
3418 None => Ok(None),
3420 Some(Cell::Node(id)) => match view.ids.key_of(*id) {
3421 Some(key) => Ok(Some(Value::Str(key.to_owned()))),
3422 None => Err(format!("unknown node id {id}")),
3423 },
3424 Some(Cell::Scalar(val)) => Ok(Some(val.clone())),
3426 Some(Cell::Rel(_)) => Err(format!(
3427 "variable `{v}` is a relationship; return its properties ({v}.field) instead"
3428 )),
3429 Some(Cell::Path(hops)) => Ok(Some(Value::Int(*hops as i64))),
3430 }
3431 }
3432 RetVal::Prop { var, field } => resolve_prop(view, vars, row, var, field),
3433 RetVal::Agg { .. } => Err(
3436 "project_item: Agg variant reached exec_project — aggregate plans must not contain Project"
3437 .to_string(),
3438 ),
3439 RetVal::FuncCall { name, args } => eval_func(name, args, view, vars, row, params),
3440 RetVal::ScalarExpr(op) => resolve_operand(view, vars, row, op, params),
3441 }
3442}
3443
3444fn order_column(item: &OrderItem) -> String {
3445 match &item.target {
3446 OrderTarget::Alias(name) | OrderTarget::Var(name) => name.clone(),
3447 OrderTarget::Prop { var, field } => format!("{var}.{field}"),
3448 }
3449}
3450
3451fn exec_order_by(table: &mut Projected, items: &[OrderItem]) -> Result<(), String> {
3452 let mut keys = Vec::with_capacity(items.len());
3453 for item in items {
3454 let name = order_column(item);
3455 let idx = table
3456 .columns
3457 .iter()
3458 .position(|c| c == &name)
3459 .ok_or_else(|| format!("ORDER BY target `{name}` is not a projected column"))?;
3460 keys.push((idx, item.descending));
3461 }
3462 table.rows.sort_by(|a, b| {
3463 for &(idx, desc) in &keys {
3464 let c = cmp_optional(
3465 a.get(idx).and_then(|x| x.as_ref()),
3466 b.get(idx).and_then(|x| x.as_ref()),
3467 desc,
3468 );
3469 if c != std::cmp::Ordering::Equal {
3470 return c;
3471 }
3472 }
3473 std::cmp::Ordering::Equal
3474 });
3475 Ok(())
3476}
3477
3478fn resolve_ls(ls: &LimitSkip, params: &Params) -> Result<u64, String> {
3483 match ls {
3484 LimitSkip::Exact(n) => Ok(*n),
3485 LimitSkip::Param(name) => {
3486 let val = params
3487 .0
3488 .get(name)
3489 .ok_or_else(|| format!("missing parameter `{name}` (used in LIMIT/SKIP)"))?;
3490 match val {
3491 Value::Int(i) if *i >= 0 => Ok(*i as u64),
3492 Value::Int(i) => Err(format!(
3493 "LIMIT/SKIP parameter `{name}` must be a non-negative integer, got {i}"
3494 )),
3495 other => Err(format!(
3496 "LIMIT/SKIP parameter `{name}` must be an integer, got {other:?}"
3497 )),
3498 }
3499 }
3500 }
3501}
3502
3503fn apply_skip<T>(rows: &mut Vec<T>, n: u64) {
3504 let n = usize::try_from(n).unwrap_or(usize::MAX);
3505 if n >= rows.len() {
3506 rows.clear();
3507 } else {
3508 rows.drain(0..n);
3509 }
3510}
3511
3512fn apply_limit<T>(rows: &mut Vec<T>, n: u64) {
3513 let n = usize::try_from(n).unwrap_or(usize::MAX);
3514 rows.truncate(n);
3515}
3516
3517#[cfg(test)]
3518mod tests {
3519 use super::{execute, resolve_operand, Params, Row, VarTable};
3520 use crate::cypher::ast::{
3521 ArithOp, LimitSkip, Operand, OrderItem, OrderTarget, RetItem, RetVal,
3522 };
3523 use crate::cypher::plan::{plan, PlanOp};
3524 use crate::cypher::{lex, parse, RelDir};
3525 use crate::result::ResultSet;
3526 use crate::view::GraphView;
3527 use core_storage::v8::seam::{ColumnsView, EdgePropsView, TopologyView};
3528 use core_storage::{ColumnStore, EdgeProps, IdMap, Interner, Topology, Value};
3529 use proptest::prelude::*;
3530 use std::collections::BTreeMap;
3531
3532 struct Fx {
3533 ids: IdMap,
3534 syms: Interner,
3535 labels: Vec<u32>,
3536 props: ColumnStore,
3537 topo: Topology,
3538 eprops: EdgeProps,
3539 }
3540
3541 impl Fx {
3542 fn new() -> Self {
3543 Fx {
3544 ids: IdMap::new(),
3545 syms: Interner::new(),
3546 labels: vec![],
3547 props: ColumnStore::new(),
3548 topo: Topology::new(),
3549 eprops: EdgeProps::new(),
3550 }
3551 }
3552
3553 fn add(&mut self, label: &str, key: &str, props: Vec<(&str, Value)>) -> u32 {
3554 let id = self.ids.get_or_insert(key);
3555 let sym = self.syms.intern(label);
3556 self.labels.resize(id as usize + 1, u32::MAX);
3557 self.labels[id as usize] = sym;
3558 for (f, v) in props {
3559 self.props.set(id, f, v);
3560 }
3561 id
3562 }
3563
3564 fn edge(&mut self, etype: &str, src: u32, dst: u32, props: Vec<(&str, Value)>) {
3565 let et = self.syms.intern(etype);
3566 self.topo.add_edge(et, src, dst);
3567 for (f, v) in props {
3568 self.eprops.set(et, src, dst, f, v);
3569 }
3570 }
3571
3572 fn view(&self) -> GraphView<'_> {
3573 GraphView {
3574 ids: &self.ids,
3575 syms: &self.syms,
3576 labels: &self.labels,
3577 props: ColumnsView::owned(&self.props),
3578 topo: TopologyView::owned(&self.topo),
3579 edge_props: EdgePropsView::owned(&self.eprops),
3580 mask: None,
3581 }
3582 }
3583 }
3584
3585 fn compile(src: &str) -> Vec<PlanOp> {
3586 plan(&parse(&lex(src).expect("lex")).expect("parse")).expect("plan")
3587 }
3588
3589 fn run(
3590 view: &GraphView,
3591 src: &str,
3592 params: &BTreeMap<String, Value>,
3593 ) -> Result<ResultSet, String> {
3594 execute(view, &compile(src), &Params(params))
3595 }
3596
3597 fn s(v: &str) -> Value {
3598 Value::Str(v.into())
3599 }
3600
3601 fn f(v: f64) -> Value {
3602 Value::Float(v)
3603 }
3604
3605 fn i(v: i64) -> Value {
3606 Value::Int(v)
3607 }
3608
3609 fn rows_of(rs: &ResultSet) -> Vec<Vec<Option<Value>>> {
3610 (0..rs.len()).map(|i| rs.row(i).to_vec()).collect()
3611 }
3612
3613 fn col(rs: &ResultSet, name: &str) -> Vec<Option<Value>> {
3614 (0..rs.len()).map(|i| rs.get(i, name).cloned()).collect()
3615 }
3616
3617 fn hop_graph() -> Fx {
3618 let mut fx = Fx::new();
3619 let ada = fx.add("Person", "ada", vec![]);
3620 let bob = fx.add("Person", "bob", vec![]);
3621 let cam = fx.add("Person", "cam", vec![]);
3622 let acme = fx.add("Company", "acme", vec![]);
3623 fx.edge("KNOWS", ada, bob, vec![]);
3624 fx.edge("KNOWS", ada, cam, vec![]);
3625 fx.edge("KNOWS", bob, cam, vec![]);
3626 fx.edge("LIKES", ada, acme, vec![]);
3627 fx
3628 }
3629
3630 fn undirected_graph() -> Fx {
3631 let mut fx = Fx::new();
3632 let a = fx.add("N", "a", vec![]);
3633 let b = fx.add("N", "b", vec![]);
3634 fx.edge("T", a, b, vec![("w", i(42))]);
3635 fx
3636 }
3637
3638 fn triangle() -> Fx {
3639 let mut fx = Fx::new();
3640 let a = fx.add("N", "a", vec![]);
3641 let b = fx.add("N", "b", vec![]);
3642 let c = fx.add("N", "c", vec![]);
3643 fx.edge("T", a, b, vec![("eid", i(1))]);
3644 fx.edge("T", b, c, vec![("eid", i(2))]);
3645 fx.edge("T", c, a, vec![("eid", i(3))]);
3646 fx
3647 }
3648
3649 fn single_edge() -> (Fx, u32, u32) {
3650 let mut fx = Fx::new();
3651 let a = fx.add("N", "a", vec![]);
3652 let b = fx.add("N", "b", vec![]);
3653 fx.edge("T", a, b, vec![]);
3654 (fx, a, b)
3655 }
3656
3657 fn dogfood_graph() -> Fx {
3659 let mut fx = Fx::new();
3660 let t1 = fx.add("Talent", "t1", vec![("id", s("t1"))]);
3661 let acme = fx.add("Company", "acme", vec![]);
3662 let beta = fx.add("Company", "beta", vec![]);
3663 let gamma = fx.add("Company", "gamma", vec![]);
3664 let delta = fx.add("Company", "delta", vec![]);
3665 let echo = fx.add("Company", "echo", vec![]);
3666 let foxtrot = fx.add("Company", "foxtrot", vec![]);
3667 let zeta = fx.add("Company", "zeta", vec![]);
3668 fx.edge("INDUSTRY_ALIGNMENT", acme, t1, vec![("score", f(0.9))]);
3669 fx.edge("SPECIALTY_MATCH", acme, t1, vec![("score", f(0.8))]);
3670 fx.edge("INDUSTRY_ALIGNMENT", beta, t1, vec![("score", f(0.6))]);
3671 fx.edge("SPECIALTY_MATCH", beta, t1, vec![("score", f(0.7))]);
3672 fx.edge("INDUSTRY_ALIGNMENT", gamma, t1, vec![("score", f(0.4))]);
3673 fx.edge("SPECIALTY_MATCH", gamma, t1, vec![("score", f(0.9))]);
3674 fx.edge("INDUSTRY_ALIGNMENT", delta, t1, vec![("score", f(0.8))]);
3675 fx.edge("SPECIALTY_MATCH", delta, t1, vec![("score", f(0.3))]);
3676 fx.edge("INDUSTRY_ALIGNMENT", echo, t1, vec![("score", f(0.5))]);
3677 fx.edge("SPECIALTY_MATCH", echo, t1, vec![("score", f(0.5))]);
3678 fx.edge("INDUSTRY_ALIGNMENT", foxtrot, t1, vec![("score", f(0.95))]);
3679 fx.edge("INDUSTRY_ALIGNMENT", zeta, t1, vec![("score", f(0.9))]);
3680 fx.edge("SPECIALTY_MATCH", zeta, t1, vec![("score", f(0.6))]);
3681 fx
3682 }
3683
3684 const DOGFOOD: &str = "\
3685MATCH (t:Talent {id: $tid}) \
3686MATCH (c:Company)-[i:INDUSTRY_ALIGNMENT]->(t) \
3687MATCH (c)-[s:SPECIALTY_MATCH]->(t) \
3688WHERE i.score >= 0.5 AND s.score >= 0.5 \
3689RETURN c, i.score AS industry, s.score AS specialty \
3690ORDER BY industry DESC, specialty DESC \
3691LIMIT 10";
3692
3693 fn tid_params() -> BTreeMap<String, Value> {
3694 let mut p = BTreeMap::new();
3695 p.insert("tid".into(), s("t1"));
3696 p
3697 }
3698
3699 #[test]
3700 fn single_hop_match_label_and_etype_filters() {
3701 let fx = hop_graph();
3702 let v = fx.view();
3703 let rs = run(
3704 &v,
3705 "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b",
3706 &BTreeMap::new(),
3707 )
3708 .expect("single-hop");
3709 assert_eq!(rs.columns(), &["a".to_string(), "b".to_string()]);
3710 assert_eq!(
3711 rows_of(&rs),
3712 vec![
3713 vec![Some(s("ada")), Some(s("bob"))],
3714 vec![Some(s("ada")), Some(s("cam"))],
3715 vec![Some(s("bob")), Some(s("cam"))],
3716 ]
3717 );
3718 let likes = run(
3720 &v,
3721 "MATCH (a:Person)-[:LIKES]->(b:Company) RETURN a, b",
3722 &BTreeMap::new(),
3723 )
3724 .unwrap();
3725 assert_eq!(rows_of(&likes), vec![vec![Some(s("ada")), Some(s("acme"))]]);
3726 let no_combo = run(
3727 &v,
3728 "MATCH (a:Person)-[:KNOWS]->(b:Company) RETURN a, b",
3729 &BTreeMap::new(),
3730 )
3731 .unwrap();
3732 assert!(no_combo.is_empty());
3733 }
3734
3735 #[test]
3736 fn undirected_match_finds_both_orientations_and_binds_true_triple() {
3737 let fx = undirected_graph();
3738 let v = fx.view();
3739 let rs =
3742 run(&v, "MATCH (x)-[r:T]-(y) RETURN x, y, r.w", &BTreeMap::new()).expect("undirected");
3743 assert_eq!(
3744 rows_of(&rs),
3745 vec![
3746 vec![Some(s("a")), Some(s("b")), Some(i(42))],
3747 vec![Some(s("b")), Some(s("a")), Some(i(42))],
3748 ]
3749 );
3750
3751 let left = run(
3752 &v,
3753 "MATCH (y)<-[r:T]-(x) RETURN x, y, r.w",
3754 &BTreeMap::new(),
3755 )
3756 .unwrap();
3757 assert_eq!(
3758 rows_of(&left),
3759 vec![vec![Some(s("a")), Some(s("b")), Some(i(42))]]
3760 );
3761 }
3762
3763 #[test]
3764 fn relationship_uniqueness_triangle_and_two_hop_cycle() {
3765 let tri = triangle();
3766 let v = tri.view();
3767 let rs = run(
3768 &v,
3769 "MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z, r1.eid, r2.eid",
3770 &BTreeMap::new(),
3771 )
3772 .expect("triangle 2-hop");
3773 assert_eq!(
3774 rows_of(&rs),
3775 vec![
3776 vec![
3777 Some(s("a")),
3778 Some(s("b")),
3779 Some(s("c")),
3780 Some(i(1)),
3781 Some(i(2))
3782 ],
3783 vec![
3784 Some(s("b")),
3785 Some(s("c")),
3786 Some(s("a")),
3787 Some(i(2)),
3788 Some(i(3))
3789 ],
3790 vec![
3791 Some(s("c")),
3792 Some(s("a")),
3793 Some(s("b")),
3794 Some(i(3)),
3795 Some(i(1))
3796 ],
3797 ]
3798 );
3799 for row in rows_of(&rs) {
3800 assert_ne!(row[3], row[4], "r1 must never bind the same edge as r2");
3801 }
3802
3803 let (mut one, a, b) = single_edge();
3804 let v = one.view();
3805 let cycle = run(
3806 &v,
3807 "MATCH (x)-[:T]->(y)-[:T]->(x) RETURN x",
3808 &BTreeMap::new(),
3809 )
3810 .expect("2-hop cycle");
3811 assert!(
3812 cycle.is_empty(),
3813 "single directed edge cannot close a 2-hop cycle"
3814 );
3815
3816 let undirected_cycle = run(&v, "MATCH (x)-[:T]-(y)-[:T]-(x) RETURN x", &BTreeMap::new())
3818 .expect("undirected uniqueness");
3819 assert!(
3820 undirected_cycle.is_empty(),
3821 "relationship uniqueness must reject walking the same triple back"
3822 );
3823
3824 one.edge("T", b, a, vec![]);
3825 let v = one.view();
3826 let with_recip = run(
3827 &v,
3828 "MATCH (x)-[:T]->(y)-[:T]->(x) RETURN x",
3829 &BTreeMap::new(),
3830 )
3831 .unwrap();
3832 assert_eq!(col(&with_recip, "x"), vec![Some(s("a")), Some(s("b"))]);
3833 }
3834
3835 #[test]
3836 fn multi_match_join_bound_and_bound_destination_expand() {
3837 let fx = dogfood_graph();
3838 let v = fx.view();
3839 let rs = run(
3842 &v,
3843 "MATCH (t:Talent {id: $tid}) \
3844 MATCH (c:Company)-[i:INDUSTRY_ALIGNMENT]->(t) \
3845 MATCH (c)-[s:SPECIALTY_MATCH]->(t) \
3846 RETURN c",
3847 &tid_params(),
3848 )
3849 .expect("join + bound dest");
3850 assert_eq!(
3851 col(&rs, "c"),
3852 vec![
3853 Some(s("acme")),
3854 Some(s("beta")),
3855 Some(s("gamma")),
3856 Some(s("delta")),
3857 Some(s("echo")),
3858 Some(s("zeta")),
3859 ]
3860 );
3861
3862 let keep = run(&v, "MATCH (c:Company) MATCH (c) RETURN c", &BTreeMap::new()).unwrap();
3864 assert_eq!(keep.len(), 7);
3865 let drop = run(
3867 &v,
3868 "MATCH (c:Company) MATCH (c:Talent) RETURN c",
3869 &BTreeMap::new(),
3870 )
3871 .unwrap();
3872 assert!(drop.is_empty());
3873 }
3874
3875 #[test]
3876 fn scan_key_exec_does_not_use_label_scan() {
3877 let mut fx = Fx::new();
3878 fx.add("Person", "p1", vec![]);
3879 fx.add("Person", "p2", vec![]);
3880 fx.add("Person", "p3", vec![]);
3881 fx.add("Company", "c1", vec![]);
3882 let v = fx.view();
3883 let params = BTreeMap::new();
3884
3885 let fires_before = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
3886 let rs = run(&v, "MATCH (n:Person {id: 'p2'}) RETURN n", ¶ms).expect("scan key");
3887 let fires_after = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
3888 assert!(
3889 fires_after > fires_before,
3890 "SCAN_KEY_FIRES must increment; before={fires_before} after={fires_after}"
3891 );
3892 assert_eq!(rows_of(&rs), vec![vec![Some(s("p2"))]]);
3893
3894 let miss = run(&v, "MATCH (n:Person {id: 'nope'}) RETURN n", ¶ms).unwrap();
3895 assert!(
3896 miss.is_empty(),
3897 "missing key must be zero rows, not an error"
3898 );
3899
3900 let wrong = run(&v, "MATCH (n:Person {id: 'c1'}) RETURN n", ¶ms).unwrap();
3901 assert!(
3902 wrong.is_empty(),
3903 "wrong label must be zero rows, not an error"
3904 );
3905 }
3906
3907 #[test]
3908 fn rel_var_edge_prop_filter() {
3909 let mut fx = Fx::new();
3910 let a = fx.add("N", "a", vec![]);
3911 let b = fx.add("N", "b", vec![]);
3912 let c = fx.add("N", "c", vec![]);
3913 fx.edge("T", a, b, vec![("w", f(0.7))]);
3914 fx.edge("T", a, c, vec![("w", f(0.3))]);
3915 let v = fx.view();
3916 let rs = run(
3917 &v,
3918 "MATCH (x)-[r:T]->(y) WHERE r.w >= 0.5 RETURN y, r.w",
3919 &BTreeMap::new(),
3920 )
3921 .expect("edge-prop filter");
3922 assert_eq!(rows_of(&rs), vec![vec![Some(s("b")), Some(f(0.7))]]);
3923 let fail = run(
3924 &v,
3925 "MATCH (x)-[r:T]->(y) WHERE r.w >= 0.8 RETURN y",
3926 &BTreeMap::new(),
3927 )
3928 .unwrap();
3929 assert!(fail.is_empty());
3930 }
3931
3932 #[test]
3933 fn params_present_resolve_missing_is_err_before_rows() {
3934 let fx = dogfood_graph();
3935 let v = fx.view();
3936 let hit =
3937 run(&v, "MATCH (t:Talent {id: $tid}) RETURN t", &tid_params()).expect("present param");
3938 assert_eq!(col(&hit, "t"), vec![Some(s("t1"))]);
3939
3940 let err = run(
3942 &v,
3943 "MATCH (t:NoSuchLabel {id: $tid}) RETURN t",
3944 &BTreeMap::new(),
3945 )
3946 .expect_err("missing param must be Err, not Ok(empty)");
3947 assert!(
3948 err.contains("tid")
3949 && (err.contains("param") || err.contains("Param") || err.contains("missing")),
3950 "missing-param error must name the parameter, got: {err}"
3951 );
3952
3953 let err = run(
3954 &v,
3955 "MATCH (t:Talent) WHERE t.id = $tid RETURN t",
3956 &BTreeMap::new(),
3957 )
3958 .expect_err("missing WHERE param");
3959 assert!(err.contains("tid"), "got: {err}");
3960 }
3961
3962 #[test]
3963 fn order_by_none_last_then_skip_limit() {
3964 let mut fx = Fx::new();
3965 fx.add("Person", "ada", vec![("age", i(30))]);
3966 fx.add("Person", "bob", vec![]); fx.add("Person", "cam", vec![("age", i(10))]);
3968 fx.add("Person", "dan", vec![("age", i(20))]);
3969 let v = fx.view();
3970
3971 let asc = run(
3972 &v,
3973 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age",
3974 &BTreeMap::new(),
3975 )
3976 .unwrap();
3977 assert_eq!(
3978 rows_of(&asc),
3979 vec![
3980 vec![Some(s("cam")), Some(i(10))],
3981 vec![Some(s("dan")), Some(i(20))],
3982 vec![Some(s("ada")), Some(i(30))],
3983 vec![Some(s("bob")), None],
3984 ]
3985 );
3986
3987 let desc = run(
3988 &v,
3989 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age DESC",
3990 &BTreeMap::new(),
3991 )
3992 .unwrap();
3993 assert_eq!(
3994 rows_of(&desc),
3995 vec![
3996 vec![Some(s("ada")), Some(i(30))],
3997 vec![Some(s("dan")), Some(i(20))],
3998 vec![Some(s("cam")), Some(i(10))],
3999 vec![Some(s("bob")), None],
4000 ]
4001 );
4002
4003 let skip_lim = run(
4004 &v,
4005 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age SKIP 1 LIMIT 2",
4006 &BTreeMap::new(),
4007 )
4008 .unwrap();
4009 assert_eq!(
4010 rows_of(&skip_lim),
4011 vec![
4012 vec![Some(s("dan")), Some(i(20))],
4013 vec![Some(s("ada")), Some(i(30))],
4014 ]
4015 );
4016
4017 let desc_sl = run(
4018 &v,
4019 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age DESC SKIP 1 LIMIT 2",
4020 &BTreeMap::new(),
4021 )
4022 .unwrap();
4023 assert_eq!(
4024 rows_of(&desc_sl),
4025 vec![
4026 vec![Some(s("dan")), Some(i(20))],
4027 vec![Some(s("cam")), Some(i(10))],
4028 ]
4029 );
4030 }
4031
4032 #[test]
4033 fn unknown_label_and_etype_are_ok_empty() {
4034 let fx = hop_graph();
4035 let v = fx.view();
4036 let lab = run(&v, "MATCH (x:Nope) RETURN x", &BTreeMap::new()).expect("unknown label");
4037 assert!(lab.is_empty());
4038 let et = run(
4039 &v,
4040 "MATCH (a)-[:NO_SUCH_ETYPE]->(b) RETURN a",
4041 &BTreeMap::new(),
4042 )
4043 .expect("unknown etype");
4044 assert!(et.is_empty());
4045 }
4046
4047 #[test]
4048 fn execute_is_deterministic() {
4049 let fx = dogfood_graph();
4050 let v = fx.view();
4051 let p = tid_params();
4052 let a = run(&v, DOGFOOD, &p).expect("first");
4053 let b = run(&v, DOGFOOD, &p).expect("second");
4054 assert_eq!(a, b);
4055 let hop = hop_graph();
4056 let hv = hop.view();
4057 let q = "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b";
4058 assert_eq!(run(&hv, q, &BTreeMap::new()), run(&hv, q, &BTreeMap::new()));
4059 }
4060
4061 #[test]
4062 fn dogfood_pipeline_exact_rows() {
4063 let fx = dogfood_graph();
4064 let v = fx.view();
4065 let rs = run(&v, DOGFOOD, &tid_params()).expect("dogfood");
4066 assert_eq!(
4067 rs.columns(),
4068 &[
4069 "c".to_string(),
4070 "industry".to_string(),
4071 "specialty".to_string()
4072 ]
4073 );
4074 assert_eq!(
4076 rows_of(&rs),
4077 vec![
4078 vec![Some(s("acme")), Some(f(0.9)), Some(f(0.8))],
4079 vec![Some(s("zeta")), Some(f(0.9)), Some(f(0.6))],
4080 vec![Some(s("beta")), Some(f(0.6)), Some(f(0.7))],
4081 vec![Some(s("echo")), Some(f(0.5)), Some(f(0.5))],
4082 ]
4083 );
4084 }
4085
4086 #[test]
4087 fn unknown_var_in_op_is_err_not_panic() {
4088 let fx = hop_graph();
4089 let v = fx.view();
4090 let plan = vec![
4091 PlanOp::ScanLabel {
4092 var: "a".into(),
4093 label: None,
4094 },
4095 PlanOp::Expand {
4096 from: "zzz".into(),
4097 rel_var: Some("r".into()),
4098 etype: None,
4099 dir: RelDir::Right,
4100 to: "b".into(),
4101 to_label: None,
4102 to_props: vec![],
4103 },
4104 PlanOp::Project {
4105 items: vec![RetItem {
4106 value: RetVal::Var("a".into()),
4107 alias: None,
4108 }],
4109 },
4110 ];
4111 let params = BTreeMap::new();
4112 let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
4113 execute(&v, &plan, &Params(¶ms))
4114 }));
4115 assert!(caught.is_ok(), "execute panicked on unknown var");
4116 let err = caught.unwrap().expect_err("unknown var must be Err");
4117 assert!(
4118 err.contains("zzz") && err.to_ascii_lowercase().contains("unbound"),
4119 "got: {err}"
4120 );
4121
4122 let join = vec![
4123 PlanOp::JoinBound {
4124 var: "ghost".into(),
4125 label: None,
4126 props: vec![],
4127 },
4128 PlanOp::Project {
4129 items: vec![RetItem {
4130 value: RetVal::Var("ghost".into()),
4131 alias: None,
4132 }],
4133 },
4134 ];
4135 let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
4136 execute(&v, &join, &Params(¶ms))
4137 }));
4138 assert!(caught.is_ok(), "execute panicked on JoinBound unknown var");
4139 assert!(caught.unwrap().is_err());
4140 }
4141
4142 #[test]
4143 fn dest_props_on_bound_expand_are_applied() {
4144 let fx = dogfood_graph();
4145 let v = fx.view();
4146 let rs = run(
4147 &v,
4148 "MATCH (t:Talent {id: $tid}) \
4149 MATCH (c:Company)-[r:INDUSTRY_ALIGNMENT]->(t:Talent {id: $tid}) \
4150 RETURN c",
4151 &tid_params(),
4152 )
4153 .unwrap();
4154 assert_eq!(
4156 col(&rs, "c"),
4157 vec![
4158 Some(s("acme")),
4159 Some(s("beta")),
4160 Some(s("gamma")),
4161 Some(s("delta")),
4162 Some(s("echo")),
4163 Some(s("foxtrot")),
4164 Some(s("zeta")),
4165 ]
4166 );
4167 let miss = run(
4168 &v,
4169 "MATCH (t:Talent {id: $tid}) \
4170 MATCH (c:Company)-[r:INDUSTRY_ALIGNMENT]->(t {id: 'nope'}) \
4171 RETURN c",
4172 &tid_params(),
4173 )
4174 .unwrap();
4175 assert!(miss.is_empty());
4176 }
4177
4178 #[test]
4179 fn missing_node_prop_projects_none_and_pattern_misses() {
4180 let mut fx = Fx::new();
4181 fx.add("Person", "ada", vec![("age", i(30))]);
4182 fx.add("Person", "bob", vec![]);
4183 let v = fx.view();
4184 let rs = run(&v, "MATCH (p:Person) RETURN p.age", &BTreeMap::new()).unwrap();
4185 assert_eq!(col(&rs, "p.age"), vec![Some(i(30)), None]);
4186 let pat = run(&v, "MATCH (p:Person {age: 30}) RETURN p", &BTreeMap::new()).unwrap();
4187 assert_eq!(col(&pat, "p"), vec![Some(s("ada"))]);
4188 }
4189
4190 #[test]
4191 fn unlabeled_match_does_not_project_sentinel_ghost_key() {
4192 let mut fx = Fx::new();
4193 fx.add("Person", "ada", vec![]);
4194 fx.ids.get_or_insert("ghost");
4197 fx.labels.resize(fx.ids.len(), u32::MAX);
4198 fx.add("Person", "bob", vec![]);
4199 let v = fx.view();
4200 let rs = run(&v, "MATCH (n) RETURN n", &BTreeMap::new()).expect("unlabeled scan");
4201 assert_eq!(col(&rs, "n"), vec![Some(s("ada")), Some(s("bob"))]);
4202 assert!(
4203 !rows_of(&rs)
4204 .iter()
4205 .any(|row| row.iter().any(|c| *c == Some(s("ghost")))),
4206 "sentinel slot must not project a ghost key"
4207 );
4208 }
4209
4210 #[test]
4211 fn execute_never_panics_on_hostile_plans() {
4212 let fx = hop_graph();
4213 let v = fx.view();
4214 let params = BTreeMap::new();
4215 let hostile = vec![
4216 vec![],
4217 vec![PlanOp::Project { items: vec![] }],
4218 vec![PlanOp::OrderBy {
4219 items: vec![OrderItem {
4220 target: OrderTarget::Alias("nope".into()),
4221 descending: false,
4222 }],
4223 }],
4224 vec![PlanOp::Filter {
4225 expr: crate::cypher::ast::Expr::Cmp {
4226 lhs: Operand::Prop {
4227 var: "missing".into(),
4228 field: "x".into(),
4229 },
4230 op: crate::filter::CmpOp::Eq,
4231 rhs: Operand::Lit(i(1)),
4232 },
4233 }],
4234 vec![
4235 PlanOp::ScanLabel {
4236 var: "a".into(),
4237 label: None,
4238 },
4239 PlanOp::LookupProps {
4240 var: "zzz".into(),
4241 props: vec![("k".into(), Operand::Lit(i(1)))],
4242 },
4243 ],
4244 vec![
4245 PlanOp::Skip(LimitSkip::Exact(99)),
4246 PlanOp::Limit(LimitSkip::Exact(0)),
4247 ],
4248 ];
4249 for plan in hostile {
4250 let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
4251 execute(&v, &plan, &Params(¶ms))
4252 }));
4253 assert!(caught.is_ok(), "execute panicked on hostile plan: {plan:?}");
4254 }
4255 }
4256
4257 #[test]
4258 fn unlabeled_and_labeled_scans_skip_tombstoned_ids() {
4259 let mut fx = Fx::new();
4260 let ada = fx.add("Person", "ada", vec![]);
4261 let bob = fx.add("Person", "bob", vec![]);
4262 fx.edge("KNOWS", ada, bob, vec![]);
4263 fx.ids.delete("ada");
4265 fx.labels[ada as usize] = u32::MAX;
4266 let knows = fx.syms.get("KNOWS").unwrap();
4267 fx.topo.remove_edge(knows, ada, bob);
4268 let v = fx.view();
4269
4270 let labeled = run(&v, "MATCH (p:Person) RETURN p", &BTreeMap::new()).unwrap();
4271 assert_eq!(col(&labeled, "p"), vec![Some(s("bob"))]);
4272 let unlabeled = run(&v, "MATCH (n) RETURN n", &BTreeMap::new()).unwrap();
4273 assert_eq!(col(&unlabeled, "n"), vec![Some(s("bob"))]);
4274 let hop = run(&v, "MATCH (x)-[:KNOWS]->(y) RETURN x, y", &BTreeMap::new()).unwrap();
4275 assert!(
4276 hop.is_empty(),
4277 "expand cannot yield edges to a deleted node once topology is swept"
4278 );
4279 }
4280
4281 proptest! {
4286 #[test]
4299 fn prop_bounded_equals_unbounded_slice(
4300 n_nodes in 2u32..10u32,
4301 edge_pairs in proptest::collection::vec(
4302 (any::<u32>(), any::<u32>()), 0..20usize
4303 ),
4304 recip_pairs in proptest::collection::vec(
4306 (any::<u32>(), any::<u32>()), 0..8usize
4307 ),
4308 n_hops in 1u32..4u32, use_filter in any::<bool>(),
4310 threshold in 0i64..8i64, limit in 1u64..10u64,
4312 skip in 0u64..4u64,
4313 ) {
4314 let mut fx = Fx::new();
4315 let mut node_ids = Vec::new();
4316 for idx in 0..n_nodes {
4317 let id = fx.add("N", &format!("n{idx}"), vec![("v", i(idx as i64 % 8))]);
4319 node_ids.push(id);
4320 }
4321 let n = node_ids.len();
4322
4323 for (si, di) in &edge_pairs {
4325 let si = (*si as usize) % n;
4326 let di = (*di as usize) % n;
4327 if si != di {
4328 fx.edge("T", node_ids[si], node_ids[di], vec![]);
4329 }
4330 }
4331 for (si, di) in &recip_pairs {
4334 let si = (*si as usize) % n;
4335 let di = (*di as usize) % n;
4336 if si != di {
4337 fx.edge("T", node_ids[si], node_ids[di], vec![]);
4338 fx.edge("T", node_ids[di], node_ids[si], vec![]);
4339 }
4340 }
4341
4342 let v = fx.view();
4343 let params = BTreeMap::new();
4344
4345 let var_names = ["a", "b", "c", "d"];
4348 let hop_count = n_hops as usize;
4349 let mut pattern = format!("({}:N)", var_names[0]);
4350 for h in 0..hop_count {
4351 pattern.push_str(&format!("-[:T]->({}", var_names[h + 1]));
4352 if h + 1 == hop_count {
4354 pattern.push_str(":N)");
4355 } else {
4356 pattern.push(')');
4357 }
4358 }
4359 let last_var = var_names[hop_count];
4360 let where_clause = if use_filter {
4361 format!(" WHERE {last_var}.v > {threshold}")
4362 } else {
4363 String::new()
4364 };
4365 let ret_vars: Vec<&str> = var_names[..=hop_count].to_vec();
4366 let ret_clause = ret_vars.join(", ");
4367
4368 let full_q = format!("MATCH {pattern}{where_clause} RETURN {ret_clause}");
4369 let bounded_q = format!(
4370 "MATCH {pattern}{where_clause} RETURN {ret_clause} SKIP {skip} LIMIT {limit}"
4371 );
4372
4373 let full_plan = compile(&full_q);
4374 let unbounded = super::with_max_intermediate_rows(100_000, || {
4377 super::execute_unbounded(&v, &full_plan, &Params(¶ms))
4378 });
4379 let unbounded = match unbounded {
4382 Ok(rs) => rs,
4383 Err(_) => return Ok(()),
4384 };
4385 let total = unbounded.len();
4386 let full_rows = rows_of(&unbounded);
4387
4388 let bounded = super::with_max_intermediate_rows(100_000, || {
4389 run(&v, &bounded_q, ¶ms)
4390 }).expect("bounded must not error");
4391
4392 let s = (skip as usize).min(total);
4393 let e = (skip as usize + limit as usize).min(total);
4394 prop_assert_eq!(
4395 rows_of(&bounded),
4396 full_rows[s..e].to_vec(),
4397 "hops={} filter={} threshold={} SKIP {} LIMIT {}: \
4398 bounded != unbounded[{}..{}]",
4399 hop_count, use_filter, threshold, skip, limit, s, e
4400 );
4401 }
4402 }
4403
4404 proptest! {
4405 #[test]
4423 fn prop_scan_filter_fused_equals_unbounded_slice(
4424 n_nodes in 0u32..20u32,
4425 prop_mask in any::<u32>(),
4427 float_mask in any::<u32>(),
4429 threshold in -1i64..8i64,
4432 op_idx in 0u32..6u32,
4434 skip in 0u64..5u64,
4435 limit in 1u64..8u64,
4436 ) {
4437 let op_str = match op_idx {
4438 0 => "=",
4439 1 => "<>",
4440 2 => "<",
4441 3 => "<=",
4442 4 => ">",
4443 _ => ">=",
4444 };
4445
4446 let mut fx = Fx::new();
4447 for idx in 0..n_nodes {
4448 let has_prop = (prop_mask >> (idx % 32)) & 1 == 1;
4449 let use_float = (float_mask >> (idx % 32)) & 1 == 1;
4450 let props: Vec<(&str, Value)> = if has_prop {
4452 if use_float {
4453 vec![("v", f(idx as f64 % 7.0))]
4454 } else {
4455 vec![("v", i(idx as i64 % 7))]
4456 }
4457 } else {
4458 vec![]
4459 };
4460 fx.add("N", &format!("n{idx}"), props);
4461 }
4462
4463 let v = fx.view();
4464 let params = BTreeMap::new();
4465
4466 let full_q = format!("MATCH (n:N) WHERE n.v {op_str} {threshold} RETURN n");
4468 let bounded_q = format!(
4469 "MATCH (n:N) WHERE n.v {op_str} {threshold} RETURN n SKIP {skip} LIMIT {limit}"
4470 );
4471
4472 let full_plan = compile(&full_q);
4473 let unbounded = super::execute_unbounded(&v, &full_plan, &Params(¶ms));
4474 let unbounded = match unbounded {
4475 Ok(rs) => rs,
4476 Err(_) => return Ok(()),
4477 };
4478 let total = unbounded.len();
4479 let full_rows = rows_of(&unbounded);
4480
4481 let fires_before =
4483 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
4484 let bounded = run(&v, &bounded_q, ¶ms).expect("fused bounded must not error");
4485 let fires_after =
4486 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
4487
4488 if n_nodes > 0 {
4493 prop_assert!(
4494 fires_after > fires_before,
4495 "fused arm did NOT fire for op={} threshold={} n_nodes={}: \
4496 counter before={} after={}",
4497 op_str,
4498 threshold,
4499 n_nodes,
4500 fires_before,
4501 fires_after
4502 );
4503 }
4504
4505 let s = (skip as usize).min(total);
4506 let e = (skip as usize + limit as usize).min(total);
4507 prop_assert_eq!(
4508 rows_of(&bounded),
4509 full_rows[s..e].to_vec(),
4510 "fused path: op={} threshold={} n_nodes={} SKIP {} LIMIT {}: \
4511 bounded != unbounded[{}..{}]",
4512 op_str, threshold, n_nodes, skip, limit, s, e
4513 );
4514
4515 let compound_q = format!(
4519 "MATCH (n:N) WHERE n.v {} {} AND n.v >= -999 RETURN n SKIP {} LIMIT {}",
4520 op_str, threshold, skip, limit
4521 );
4522 let fires_before_compound =
4523 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
4524 let compound_bounded =
4525 run(&v, &compound_q, ¶ms).expect("compound-AND bounded must not error");
4526 let fires_after_compound =
4527 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
4528
4529 prop_assert_eq!(
4531 fires_after_compound,
4532 fires_before_compound,
4533 "fused arm fired for compound-AND shape (should use generic path): \
4534 op={} threshold={} n_nodes={}",
4535 op_str,
4536 threshold,
4537 n_nodes
4538 );
4539 prop_assert_eq!(
4540 rows_of(&compound_bounded),
4541 full_rows[s..e].to_vec(),
4542 "compound-AND fallback: op={} threshold={} n_nodes={} SKIP {} LIMIT {}: \
4543 result differs from fused",
4544 op_str, threshold, n_nodes, skip, limit
4545 );
4546 }
4547 }
4548
4549 #[test]
4560 fn dense_hop1_with_filter_survives_pull() {
4561 const LEAVES: usize = 120; const CAP: usize = 100;
4563
4564 let mut fx = Fx::new();
4565 let src = fx.add("Src", "src", vec![]);
4566 for idx in 0..LEAVES {
4567 let leaf = fx.add("Leaf", &format!("l{idx}"), vec![("v", i(idx as i64))]);
4568 fx.edge("T", src, leaf, vec![]);
4569 }
4570 let v = fx.view();
4571 let params = BTreeMap::new();
4572
4573 let staged_err = super::with_max_intermediate_rows(CAP, || {
4575 super::execute_unbounded(
4576 &v,
4577 &compile("MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v >= 110 RETURN l, l.v"),
4578 &Params(¶ms),
4579 )
4580 });
4581 assert!(
4582 staged_err.is_err(),
4583 "staged path must error on 120 leaves with cap={CAP}"
4584 );
4585 assert!(
4586 staged_err
4587 .unwrap_err()
4588 .contains("intermediate result exceeds"),
4589 "wrong error message"
4590 );
4591
4592 let ok = super::with_max_intermediate_rows(CAP, || {
4596 run(
4597 &v,
4598 "MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v >= 110 RETURN l, l.v LIMIT 5",
4599 ¶ms,
4600 )
4601 });
4602 let rs = ok.expect("pull-based must survive despite dense hop-1 exceeding cap");
4603 assert_eq!(rs.len(), 5, "LIMIT 5 must return exactly 5 rows");
4604
4605 let vs: Vec<i64> = (0..rs.len())
4607 .filter_map(|i| match rs.get(i, "l.v") {
4608 Some(Value::Int(n)) => Some(*n),
4609 _ => None,
4610 })
4611 .collect();
4612 assert_eq!(vs.len(), 5, "all projected rows must have v");
4613 for v_val in &vs {
4614 assert!(*v_val >= 110, "filter must hold: v={v_val} is not >= 110");
4615 }
4616
4617 let (pull_result, pull_produced) = super::with_expand_counter(|| {
4621 super::with_max_intermediate_rows(1_000_000, || {
4622 run(
4623 &v,
4624 "MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v < 10 RETURN l LIMIT 5",
4625 ¶ms,
4626 )
4627 })
4628 });
4629 pull_result.expect("pull must succeed without cap");
4630 assert!(
4632 pull_produced <= 5,
4633 "pull expand count {pull_produced} should be ≤ 5 (stops after 5 passing leaves)"
4634 );
4635
4636 let (staged_result, staged_produced) = super::with_expand_counter(|| {
4637 super::with_max_intermediate_rows(1_000_000, || {
4638 super::execute_unbounded(
4639 &v,
4640 &compile("MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v < 10 RETURN l"),
4641 &Params(¶ms),
4642 )
4643 })
4644 });
4645 staged_result.expect("staged must succeed with 1M cap");
4646 assert_eq!(
4647 staged_produced, LEAVES,
4648 "staged must expand all {LEAVES} leaves"
4649 );
4650
4651 assert!(
4652 staged_produced >= pull_produced * 10,
4653 "staged ({staged_produced}) must be ≥ 10× pull ({pull_produced})"
4654 );
4655 }
4656
4657 #[test]
4675 fn harness_shape_two_hop_dense_survives_pull() {
4676 const N_TALENT: usize = 70;
4677 const N_COMPANY: usize = 20;
4678 const N_INDUSTRY: usize = 3;
4679 const CAP: usize = 100;
4680 const LIMIT: usize = 10;
4681
4682 let mut fx = Fx::new();
4683
4684 let mut talent_ids: Vec<u32> = Vec::new();
4686 let mut talent_industry: Vec<usize> = Vec::new();
4687 for i in 0..N_TALENT {
4688 let ind = i % N_INDUSTRY;
4689 let id = fx.add(
4690 "Talent",
4691 &format!("t{i}"),
4692 vec![("industry", s(&ind.to_string()))],
4693 );
4694 talent_ids.push(id);
4695 talent_industry.push(ind);
4696 }
4697
4698 let mut company_ids: Vec<u32> = Vec::new();
4700 let mut company_industry: Vec<usize> = Vec::new();
4701 for i in 0..N_COMPANY {
4702 let ind = i % N_INDUSTRY;
4703 let id = fx.add(
4704 "Company",
4705 &format!("c{i}"),
4706 vec![("industry", s(&ind.to_string()))],
4707 );
4708 company_ids.push(id);
4709 company_industry.push(ind);
4710 }
4711
4712 for (ti, &tid) in talent_ids.iter().enumerate() {
4714 for (ci, &cid) in company_ids.iter().enumerate() {
4715 if talent_industry[ti] == company_industry[ci] {
4716 fx.edge("INDUSTRY_ALIGNMENT", tid, cid, vec![]);
4717 }
4718 }
4719 }
4720
4721 let v = fx.view();
4722 let params = BTreeMap::new();
4723 let query = format!(
4724 "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company)\
4725 <-[:INDUSTRY_ALIGNMENT]-(t2:Talent) RETURN t, c, t2 LIMIT {LIMIT}"
4726 );
4727
4728 const UNBOUNDED_Q: &str = "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company)\
4730 <-[:INDUSTRY_ALIGNMENT]-(t2:Talent) RETURN t, c, t2";
4731 let staged_err = super::with_max_intermediate_rows(CAP, || {
4732 super::execute_unbounded(&v, &compile(UNBOUNDED_Q), &Params(¶ms))
4733 });
4734 assert!(
4735 staged_err.is_err(),
4736 "staged must error with cap={CAP} on harness-shape graph"
4737 );
4738 assert!(
4739 staged_err
4740 .unwrap_err()
4741 .contains("intermediate result exceeds"),
4742 "wrong error"
4743 );
4744
4745 let ok = super::with_max_intermediate_rows(CAP, || run(&v, &query, ¶ms));
4747 let rs = ok.expect("pull-based must complete on harness-shape with LIMIT 10");
4748 assert_eq!(rs.len(), LIMIT, "must return exactly {LIMIT} rows");
4749
4750 for i in 0..rs.len() {
4755 let row = rs.row(i);
4756 assert_eq!(row.len(), 3, "each row must have 3 columns (t, c, t2)");
4757 assert!(row.iter().all(|c| c.is_some()), "all cells must be Some");
4758 }
4759 }
4760
4761 #[test]
4762 fn intermediate_row_cap_errors_on_scan_and_expand() {
4763 let cap_msg = |n: usize| {
4764 format!(
4765 "intermediate result exceeds {n} rows; add a LIMIT or constrain patterns with shared variables"
4766 )
4767 };
4768
4769 let mut scan_fx = Fx::new();
4770 scan_fx.add("N", "a", vec![]);
4771 scan_fx.add("N", "b", vec![]);
4772 scan_fx.add("N", "c", vec![]);
4773 let sv = scan_fx.view();
4774 let scan_err = super::with_max_intermediate_rows(2, || {
4775 run(&sv, "MATCH (n:N) RETURN n", &BTreeMap::new())
4776 })
4777 .expect_err("3-row scan must exceed cap 2");
4778 assert_eq!(scan_err, cap_msg(2));
4779
4780 let mut exp_fx = Fx::new();
4781 let src = exp_fx.add("Src", "s", vec![]);
4782 let d1 = exp_fx.add("Dst", "d1", vec![]);
4783 let d2 = exp_fx.add("Dst", "d2", vec![]);
4784 exp_fx.edge("T", src, d1, vec![]);
4785 exp_fx.edge("T", src, d2, vec![]);
4786 let ev = exp_fx.view();
4787 let exp_err = super::with_max_intermediate_rows(1, || {
4789 run(&ev, "MATCH (x:Src)-[:T]->(y) RETURN x, y", &BTreeMap::new())
4790 })
4791 .expect_err("2-row expand must exceed cap 1");
4792 assert_eq!(exp_err, cap_msg(1));
4793 }
4794
4795 #[test]
4805 fn bounded_matches_unbounded_slice_various_limits() {
4806 let fx = hop_graph();
4808 let v = fx.view();
4809 let params = BTreeMap::new();
4810
4811 let full_plan = compile("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b");
4813 let full_rs = super::execute_unbounded(&v, &full_plan, &Params(¶ms)).unwrap();
4814 let full_rows = rows_of(&full_rs);
4815 assert_eq!(full_rows.len(), 3, "hop_graph has exactly 3 KNOWS paths");
4816
4817 for limit in [1u64, 2, 3, 10] {
4818 let q = format!("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b LIMIT {limit}");
4819 let rs = run(&v, &q, ¶ms).unwrap();
4820 let expected_len = (limit as usize).min(full_rows.len());
4821 assert_eq!(
4822 rs.len(),
4823 expected_len,
4824 "LIMIT {limit}: expected {expected_len} rows, got {}",
4825 rs.len()
4826 );
4827 assert_eq!(
4828 rows_of(&rs),
4829 full_rows[..expected_len],
4830 "LIMIT {limit}: rows differ from reference slice"
4831 );
4832 }
4833
4834 let skip_rs = run(
4836 &v,
4837 "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b SKIP 1 LIMIT 2",
4838 ¶ms,
4839 )
4840 .unwrap();
4841 assert_eq!(rows_of(&skip_rs), full_rows[1..3]);
4842
4843 let mut fx2 = Fx::new();
4845 fx2.add("N", "a", vec![("v", i(1))]);
4846 fx2.add("N", "b", vec![("v", i(2))]);
4847 fx2.add("N", "c", vec![("v", i(3))]);
4848 let v2 = fx2.view();
4849 let filter_full = super::execute_unbounded(
4851 &v2,
4852 &compile("MATCH (n:N) WHERE n.v > 1 RETURN n"),
4853 &Params(¶ms),
4854 )
4855 .unwrap();
4856 assert_eq!(filter_full.len(), 2);
4857
4858 let filter_lim = run(&v2, "MATCH (n:N) WHERE n.v > 1 RETURN n LIMIT 1", ¶ms).unwrap();
4859 assert_eq!(filter_lim.len(), 1, "LIMIT 1 on filter query");
4860 assert_eq!(rows_of(&filter_lim), rows_of(&filter_full)[..1]);
4861
4862 let tri = triangle();
4864 let tv = tri.view();
4865 let tri_full = super::execute_unbounded(
4867 &tv,
4868 &compile("MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z"),
4869 &Params(¶ms),
4870 )
4871 .unwrap();
4872 assert_eq!(tri_full.len(), 3, "triangle has 3 unique two-hop paths");
4873 for limit in [1u64, 2, 3, 5] {
4874 let q = format!("MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z LIMIT {limit}");
4875 let rs = run(&tv, &q, ¶ms).unwrap();
4876 let expected_len = (limit as usize).min(3);
4877 assert_eq!(
4878 rs.len(),
4879 expected_len,
4880 "triangle LIMIT {limit}: got {} rows",
4881 rs.len()
4882 );
4883 assert_eq!(
4884 rows_of(&rs),
4885 rows_of(&tri_full)[..expected_len],
4886 "triangle LIMIT {limit}: rows differ"
4887 );
4888 }
4889 }
4890
4891 #[test]
4895 fn expand_terminates_early_with_row_bound() {
4896 const LEAVES: usize = 500;
4897 let mut fx = Fx::new();
4898 let hub = fx.add("Hub", "hub", vec![]);
4899 for i in 0..LEAVES {
4900 let leaf = fx.add("Leaf", &format!("leaf-{i}"), vec![]);
4901 fx.edge("T", hub, leaf, vec![]);
4902 }
4903 let v = fx.view();
4904 let params = BTreeMap::new();
4905 let full_plan = compile("MATCH (h:Hub)-[:T]->(x:Leaf) RETURN x");
4906
4907 let (bounded_result, bounded_produced) = super::with_expand_counter(|| {
4909 run(&v, "MATCH (h:Hub)-[:T]->(x:Leaf) RETURN x LIMIT 5", ¶ms)
4910 });
4911 let bounded_rs = bounded_result.unwrap();
4912 assert_eq!(bounded_rs.len(), 5, "LIMIT 5 must return exactly 5 rows");
4913 assert!(
4914 bounded_produced <= 5,
4915 "bounded: exec_expand emitted {bounded_produced} rows, expected ≤ 5"
4916 );
4917
4918 let (unbounded_result, unbounded_produced) = super::with_expand_counter(|| {
4920 super::execute_unbounded(&v, &full_plan, &Params(¶ms))
4921 });
4922 let unbounded_rs = unbounded_result.unwrap();
4923 assert_eq!(
4924 unbounded_rs.len(),
4925 LEAVES,
4926 "unbounded must return all {LEAVES} rows"
4927 );
4928 assert_eq!(
4929 unbounded_produced, LEAVES,
4930 "unbounded: exec_expand must emit all {LEAVES} rows"
4931 );
4932
4933 assert!(
4935 unbounded_produced >= bounded_produced * 100,
4936 "unbounded ({unbounded_produced}) must be ≥ 100× bounded ({bounded_produced})"
4937 );
4938 }
4939
4940 #[test]
4946 fn bounded_query_survives_low_intermediate_row_cap() {
4947 let mut fx = Fx::new();
4949 let src = fx.add("Src", "s", vec![]);
4950 for i in 0..20usize {
4951 let dst = fx.add("Dst", &format!("d{i}"), vec![]);
4952 fx.edge("T", src, dst, vec![]);
4953 }
4954 let v = fx.view();
4955 let params = BTreeMap::new();
4956 let full_plan = compile("MATCH (s:Src)-[:T]->(d:Dst) RETURN d");
4957
4958 let cap_err = super::with_max_intermediate_rows(10, || {
4960 super::execute_unbounded(&v, &full_plan, &Params(¶ms))
4961 });
4962 assert!(
4963 cap_err.is_err(),
4964 "unbounded must hit the intermediate-row cap"
4965 );
4966 assert!(
4967 cap_err.unwrap_err().contains("intermediate result exceeds"),
4968 "cap error message must be the budget message"
4969 );
4970
4971 let ok = super::with_max_intermediate_rows(10, || {
4973 run(&v, "MATCH (s:Src)-[:T]->(d:Dst) RETURN d LIMIT 5", ¶ms)
4974 });
4975 assert_eq!(
4976 ok.unwrap().len(),
4977 5,
4978 "bounded (LIMIT 5) must complete with 5 rows, not a cap error"
4979 );
4980
4981 let at_cap = super::with_max_intermediate_rows(10, || {
4983 run(&v, "MATCH (s:Src)-[:T]->(d:Dst) RETURN d LIMIT 10", ¶ms)
4984 });
4985 assert_eq!(
4986 at_cap.unwrap().len(),
4987 10,
4988 "bounded at LIMIT==cap must complete with 10 rows"
4989 );
4990 }
4991
4992 #[test]
4997 fn count_star_returns_total_node_count() {
4998 let mut fx = Fx::new();
4999 fx.add("Person", "ada", vec![]);
5000 fx.add("Person", "bob", vec![]);
5001 fx.add("Person", "cam", vec![]);
5002 let v = fx.view();
5003 let params = BTreeMap::new();
5004
5005 let rs = run(&v, "MATCH (n:Person) RETURN COUNT(*)", ¶ms).expect("COUNT(*)");
5006 assert_eq!(rs.columns(), &["COUNT(*)".to_string()]);
5007 assert_eq!(rs.len(), 1);
5008 assert_eq!(rs.row(0), &[Some(i(3))]);
5009
5010 let rs_empty = run(&v, "MATCH (n:Ghost) RETURN COUNT(*)", ¶ms).expect("COUNT(*) empty");
5012 assert_eq!(rs_empty.row(0), &[Some(i(0))]);
5013 }
5014
5015 #[test]
5016 fn count_star_alias_sets_column_name() {
5017 let mut fx = Fx::new();
5018 fx.add("N", "a", vec![]);
5019 let v = fx.view();
5020 let params = BTreeMap::new();
5021 let rs = run(&v, "MATCH (n:N) RETURN COUNT(*) AS total", ¶ms).expect("COUNT AS");
5022 assert_eq!(rs.columns(), &["total".to_string()]);
5023 assert_eq!(rs.row(0), &[Some(i(1))]);
5024 }
5025
5026 #[test]
5027 fn count_var_skips_null_node_bindings() {
5028 let mut fx = Fx::new();
5031 fx.add("N", "a", vec![]);
5032 fx.add("N", "b", vec![]);
5033 let v = fx.view();
5034 let params = BTreeMap::new();
5035 let rs = run(&v, "MATCH (n:N) RETURN COUNT(n)", ¶ms).expect("COUNT(n)");
5036 assert_eq!(rs.columns(), &["COUNT(n)".to_string()]);
5037 assert_eq!(rs.row(0), &[Some(i(2))]);
5038 }
5039
5040 #[test]
5041 fn sum_numeric_prop_ignores_null_and_non_numeric() {
5042 let mut fx = Fx::new();
5043 fx.add("N", "a", vec![("v", i(10))]);
5044 fx.add("N", "b", vec![("v", i(20))]);
5045 fx.add("N", "c", vec![]); let v = fx.view();
5047 let params = BTreeMap::new();
5048 let rs = run(&v, "MATCH (n:N) RETURN SUM(n.v)", ¶ms).expect("SUM");
5049 assert_eq!(rs.columns(), &["SUM(n.v)".to_string()]);
5050 assert_eq!(rs.row(0), &[Some(f(30.0))]);
5052
5053 let rs_null = run(&v, "MATCH (n:N) RETURN SUM(n.missing)", ¶ms).expect("SUM null");
5055 assert_eq!(rs_null.row(0), &[None]);
5056 }
5057
5058 #[test]
5059 fn avg_numeric_prop() {
5060 let mut fx = Fx::new();
5061 fx.add("N", "a", vec![("v", i(10))]);
5062 fx.add("N", "b", vec![("v", i(30))]);
5063 let v = fx.view();
5064 let params = BTreeMap::new();
5065 let rs = run(&v, "MATCH (n:N) RETURN AVG(n.v) AS avg_v", ¶ms).expect("AVG");
5066 assert_eq!(rs.columns(), &["avg_v".to_string()]);
5067 assert_eq!(rs.row(0), &[Some(f(20.0))]);
5069
5070 let rs_empty = run(&v, "MATCH (n:Ghost) RETURN AVG(n.v)", ¶ms).expect("AVG empty");
5072 assert_eq!(rs_empty.row(0), &[None]);
5073 }
5074
5075 #[test]
5076 fn min_max_numeric_prop() {
5077 let mut fx = Fx::new();
5078 fx.add("N", "a", vec![("v", i(5))]);
5079 fx.add("N", "b", vec![("v", i(1))]);
5080 fx.add("N", "c", vec![("v", i(9))]);
5081 fx.add("N", "d", vec![]); let v = fx.view();
5083 let params = BTreeMap::new();
5084
5085 let min_rs = run(&v, "MATCH (n:N) RETURN MIN(n.v)", ¶ms).expect("MIN");
5086 assert_eq!(min_rs.row(0), &[Some(i(1))]);
5087
5088 let max_rs = run(&v, "MATCH (n:N) RETURN MAX(n.v)", ¶ms).expect("MAX");
5089 assert_eq!(max_rs.row(0), &[Some(i(9))]);
5090 }
5091
5092 #[test]
5095 fn min_max_mixed_int_float_props() {
5096 let mut fx = Fx::new();
5097 fx.add("N", "a", vec![("v", i(3))]);
5099 fx.add("N", "b", vec![("v", f(1.5))]);
5100 fx.add("N", "c", vec![("v", i(7))]);
5101 fx.add("N", "d", vec![("v", f(2.0))]);
5102 let v = fx.view();
5103 let params = BTreeMap::new();
5104
5105 let min_rs = run(&v, "MATCH (n:N) RETURN MIN(n.v)", ¶ms).expect("MIN mixed");
5106 assert_eq!(min_rs.row(0), &[Some(f(1.5))]);
5108
5109 let max_rs = run(&v, "MATCH (n:N) RETURN MAX(n.v)", ¶ms).expect("MAX mixed");
5110 assert_eq!(max_rs.row(0), &[Some(i(7))]);
5112 }
5113
5114 #[test]
5117 fn aggregate_limit_skip_order_by_are_no_ops() {
5118 let mut fx = Fx::new();
5119 fx.add("N", "a", vec![]);
5120 fx.add("N", "b", vec![]);
5121 fx.add("N", "c", vec![]);
5122 let v = fx.view();
5123 let params = BTreeMap::new();
5124
5125 let rs_lim5 =
5127 run(&v, "MATCH (n:N) RETURN COUNT(*) LIMIT 5", ¶ms).expect("COUNT(*) LIMIT 5");
5128 assert_eq!(
5129 rs_lim5.len(),
5130 1,
5131 "aggregate with LIMIT 5 must still return 1 row"
5132 );
5133 assert_eq!(rs_lim5.row(0), &[Some(i(3))]);
5134
5135 let rs_lim0 =
5137 run(&v, "MATCH (n:N) RETURN COUNT(*) LIMIT 0", ¶ms).expect("COUNT(*) LIMIT 0");
5138 assert_eq!(
5139 rs_lim0.len(),
5140 1,
5141 "aggregate with LIMIT 0 must still return 1 row"
5142 );
5143 assert_eq!(rs_lim0.row(0), &[Some(i(3))]);
5144
5145 let rs_skip =
5147 run(&v, "MATCH (n:N) RETURN COUNT(*) SKIP 100", ¶ms).expect("COUNT(*) SKIP 100");
5148 assert_eq!(
5149 rs_skip.len(),
5150 1,
5151 "aggregate with large SKIP must still return 1 row"
5152 );
5153
5154 let rs_ord = plan_src("MATCH (n:N) RETURN COUNT(*) ORDER BY n");
5158 assert!(
5161 rs_ord.is_ok(),
5162 "COUNT(*) ORDER BY should plan without error (ORDER BY dropped)"
5163 );
5164 let plan_ops = rs_ord.unwrap();
5165 assert!(
5167 !plan_ops
5168 .iter()
5169 .any(|op| matches!(op, crate::cypher::plan::PlanOp::OrderBy { .. })),
5170 "aggregate plan must not contain OrderBy"
5171 );
5172 }
5173
5174 #[test]
5175 fn count_star_no_budget_cap_applies() {
5176 let mut fx = Fx::new();
5179 let src = fx.add("Src", "s", vec![]);
5180 for i in 0..30usize {
5181 let dst = fx.add("Dst", &format!("d{i}"), vec![]);
5182 fx.edge("T", src, dst, vec![]);
5183 }
5184 let v = fx.view();
5185 let params = BTreeMap::new();
5186
5187 let cap_err =
5189 super::with_max_intermediate_rows(10, || run(&v, "MATCH (n:Dst) RETURN n", ¶ms));
5190 assert!(
5191 cap_err.is_err(),
5192 "staged path must error on 30 nodes with cap=10"
5193 );
5194
5195 let agg_ok = super::with_max_intermediate_rows(10, || {
5197 run(&v, "MATCH (n:Dst) RETURN COUNT(*)", ¶ms)
5198 })
5199 .expect("aggregate must not hit the intermediate-row cap");
5200 assert_eq!(
5201 agg_ok.row(0),
5202 &[Some(i(30))],
5203 "COUNT(*) must count all 30 nodes regardless of cap"
5204 );
5205 }
5206
5207 fn plan_src(src: &str) -> Result<Vec<crate::cypher::plan::PlanOp>, String> {
5208 use crate::cypher::{lex, parse, plan};
5209 let toks = lex(src).map_err(|e| format!("lex: {e}"))?;
5210 let ast = parse(&toks).map_err(|e| format!("parse: {e}"))?;
5211 plan(&ast).map_err(|e| format!("plan: {e}"))
5212 }
5213
5214 #[test]
5217 fn grouped_aggregation_plan_routing() {
5218 use crate::cypher::plan::PlanOp;
5219
5220 let ops = plan_src("MATCH (a:N) RETURN a, COUNT(*)")
5222 .expect("grouped aggregation must now succeed");
5223 assert!(
5224 ops.iter()
5225 .any(|op| matches!(op, PlanOp::GroupAggregate { .. })),
5226 "grouped aggregation plan must contain GroupAggregate op, got: {ops:?}"
5227 );
5228
5229 let ops2 = plan_src("MATCH (a:N) RETURN COUNT(*), COUNT(a)")
5231 .expect("multi-aggregate must now succeed");
5232 assert!(
5233 ops2.iter()
5234 .any(|op| matches!(op, PlanOp::GroupAggregate { .. })),
5235 "multi-aggregate plan must contain GroupAggregate op, got: {ops2:?}"
5236 );
5237
5238 let err3 = plan_src("MATCH (a:N) RETURN SUM(*)").expect_err("SUM(*) must be plan error");
5240 assert!(
5241 err3.to_ascii_lowercase().contains("sum") || err3.to_ascii_lowercase().contains("*"),
5242 "error must mention SUM or *, got: {err3}"
5243 );
5244 }
5245
5246 #[test]
5249 fn grouped_single_key_count() {
5250 let mut fx = Fx::new();
5252 fx.add("N", "a", vec![("t", s("X"))]);
5253 fx.add("N", "b", vec![("t", s("X"))]);
5254 fx.add("N", "c", vec![("t", s("Y"))]);
5255 let v = fx.view();
5256 let params = BTreeMap::new();
5257
5258 let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
5259 .expect("single-key grouped COUNT must succeed");
5260 assert_eq!(
5261 rs.columns(),
5262 &["n.t".to_string(), "cnt".to_string()],
5263 "columns must match RETURN clause"
5264 );
5265 assert_eq!(rs.len(), 2, "must produce exactly 2 groups (X and Y)");
5266
5267 let find = |label: &Value| (0..rs.len()).find(|&i| rs.row(i)[0].as_ref() == Some(label));
5269 let xi = find(&s("X")).expect("group X must exist");
5270 let yi = find(&s("Y")).expect("group Y must exist");
5271 assert_eq!(rs.row(xi)[1], Some(i(2)), "X group count must be 2");
5272 assert_eq!(rs.row(yi)[1], Some(i(1)), "Y group count must be 1");
5273 }
5274
5275 #[test]
5276 fn grouped_two_keys_sum_avg() {
5277 let mut fx = Fx::new();
5279 fx.add(
5280 "N",
5281 "a",
5282 vec![("cat", s("A")), ("sub", s("1")), ("v", i(10))],
5283 );
5284 fx.add(
5285 "N",
5286 "b",
5287 vec![("cat", s("A")), ("sub", s("1")), ("v", i(20))],
5288 );
5289 fx.add(
5290 "N",
5291 "c",
5292 vec![("cat", s("A")), ("sub", s("2")), ("v", i(5))],
5293 );
5294 fx.add(
5295 "N",
5296 "d",
5297 vec![("cat", s("B")), ("sub", s("1")), ("v", i(100))],
5298 );
5299 let v = fx.view();
5300 let params = BTreeMap::new();
5301
5302 let rs = run(
5303 &v,
5304 "MATCH (n:N) RETURN n.cat, n.sub, SUM(n.v) AS total, AVG(n.v) AS avg_v",
5305 ¶ms,
5306 )
5307 .expect("two-key SUM + AVG must succeed");
5308 assert_eq!(
5309 rs.columns(),
5310 &[
5311 "n.cat".to_string(),
5312 "n.sub".to_string(),
5313 "total".to_string(),
5314 "avg_v".to_string()
5315 ]
5316 );
5317 assert_eq!(rs.len(), 3, "must produce 3 groups: (A,1), (A,2), (B,1)");
5318
5319 let find = |cat: &Value, sub: &Value| {
5320 (0..rs.len())
5321 .find(|&i| rs.row(i)[0].as_ref() == Some(cat) && rs.row(i)[1].as_ref() == Some(sub))
5322 };
5323 let a1 = find(&s("A"), &s("1")).expect("group (A,1) must exist");
5324 assert_eq!(rs.row(a1)[2], Some(f(30.0)), "(A,1) SUM must be 30.0");
5325 assert_eq!(rs.row(a1)[3], Some(f(15.0)), "(A,1) AVG must be 15.0");
5326
5327 let a2 = find(&s("A"), &s("2")).expect("group (A,2) must exist");
5328 assert_eq!(rs.row(a2)[2], Some(f(5.0)), "(A,2) SUM must be 5.0");
5329
5330 let b1 = find(&s("B"), &s("1")).expect("group (B,1) must exist");
5331 assert_eq!(rs.row(b1)[2], Some(f(100.0)), "(B,1) SUM must be 100.0");
5332 assert_eq!(rs.row(b1)[3], Some(f(100.0)), "(B,1) AVG must be 100.0");
5333 }
5334
5335 #[test]
5336 fn grouped_order_by_count_desc_limit() {
5337 let mut fx = Fx::new();
5339 fx.add("N", "a1", vec![("cat", s("A"))]);
5340 fx.add("N", "a2", vec![("cat", s("A"))]);
5341 fx.add("N", "a3", vec![("cat", s("A"))]);
5342 fx.add("N", "b1", vec![("cat", s("B"))]);
5343 fx.add("N", "b2", vec![("cat", s("B"))]);
5344 fx.add("N", "c1", vec![("cat", s("C"))]);
5345 fx.add("N", "d1", vec![("cat", s("D"))]);
5346 fx.add("N", "d2", vec![("cat", s("D"))]);
5347 fx.add("N", "d3", vec![("cat", s("D"))]);
5348 fx.add("N", "d4", vec![("cat", s("D"))]);
5349 fx.add("N", "e1", vec![("cat", s("E"))]);
5350 let v = fx.view();
5351 let params = BTreeMap::new();
5352
5353 let rs = run(
5354 &v,
5355 "MATCH (n:N) RETURN n.cat, COUNT(*) AS cnt ORDER BY cnt DESC LIMIT 3",
5356 ¶ms,
5357 )
5358 .expect("ORDER BY count DESC LIMIT 3 must succeed");
5359 assert_eq!(rs.len(), 3, "LIMIT 3 must return exactly 3 groups");
5360 assert_eq!(rs.row(0)[1], Some(i(4)), "row 0 must be count 4");
5362 assert_eq!(rs.row(0)[0], Some(s("D")), "row 0 must be category D");
5363 assert_eq!(rs.row(1)[1], Some(i(3)), "row 1 must be count 3");
5364 assert_eq!(rs.row(1)[0], Some(s("A")), "row 1 must be category A");
5365 assert_eq!(rs.row(2)[1], Some(i(2)), "row 2 must be count 2");
5366 assert_eq!(rs.row(2)[0], Some(s("B")), "row 2 must be category B");
5367
5368 let plan_ops =
5370 plan_src("MATCH (n:N) RETURN n.cat, COUNT(*) AS cnt ORDER BY cnt DESC LIMIT 3")
5371 .expect("plan must succeed");
5372 assert_eq!(
5373 crate::cypher::plan::row_bound(&plan_ops),
5374 None,
5375 "GroupAggregate plan with LIMIT must have row_bound = None"
5376 );
5377 }
5378
5379 #[test]
5380 fn grouped_empty_input_yields_zero_groups() {
5381 let fx = Fx::new(); let v = fx.view();
5383 let params = BTreeMap::new();
5384
5385 let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
5386 .expect("grouped aggregate on empty graph must succeed");
5387 assert_eq!(rs.len(), 0, "empty input must yield zero groups");
5388 assert_eq!(
5389 rs.columns(),
5390 &["n.t".to_string(), "cnt".to_string()],
5391 "columns must still be present even with zero rows"
5392 );
5393 }
5394
5395 #[test]
5396 fn grouped_null_key_groups_together() {
5397 let mut fx = Fx::new();
5399 fx.add("N", "a", vec![("t", s("X"))]);
5400 fx.add("N", "b", vec![]); fx.add("N", "c", vec![]); fx.add("N", "d", vec![("t", s("Y"))]);
5403 let v = fx.view();
5404 let params = BTreeMap::new();
5405
5406 let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
5407 .expect("null-key grouped aggregate must succeed");
5408 assert_eq!(rs.len(), 3, "must produce 3 groups: X, null, Y");
5409
5410 let null_row = (0..rs.len())
5412 .find(|&i| rs.row(i)[0].is_none())
5413 .expect("null group must be present");
5414 assert_eq!(
5415 rs.row(null_row)[1],
5416 Some(i(2)),
5417 "null group must count 2 rows (b and c)"
5418 );
5419
5420 let x_row = (0..rs.len())
5421 .find(|&i| rs.row(i)[0] == Some(s("X")))
5422 .expect("X group must exist");
5423 assert_eq!(rs.row(x_row)[1], Some(i(1)));
5424
5425 let y_row = (0..rs.len())
5426 .find(|&i| rs.row(i)[0] == Some(s("Y")))
5427 .expect("Y group must exist");
5428 assert_eq!(rs.row(y_row)[1], Some(i(1)));
5429 }
5430
5431 #[test]
5432 fn grouped_cap_error_on_high_cardinality() {
5433 let mut fx = Fx::new();
5436 fx.add("N", "a", vec![("t", s("A"))]);
5437 fx.add("N", "b", vec![("t", s("B"))]);
5438 fx.add("N", "c", vec![("t", s("C"))]);
5439 let v = fx.view();
5440 let params = BTreeMap::new();
5441
5442 let err = super::with_max_groups(2, || {
5443 run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
5444 })
5445 .expect_err("must error when group count exceeds cap");
5446 assert!(
5447 err.to_ascii_lowercase().contains("group count"),
5448 "error must mention group count, got: {err}"
5449 );
5450 }
5451
5452 #[test]
5453 fn multi_aggregate_no_keys() {
5454 let mut fx = Fx::new();
5457 fx.add("N", "a", vec![]);
5458 fx.add("N", "b", vec![]);
5459 let v = fx.view();
5460 let params = BTreeMap::new();
5461
5462 let rs = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
5463 .expect("multi-aggregate no keys must succeed");
5464 assert_eq!(rs.len(), 1, "must produce exactly one result row");
5465 assert_eq!(rs.row(0)[0], Some(i(2)), "COUNT(*) must be 2");
5466 assert_eq!(rs.row(0)[1], Some(i(2)), "COUNT(n) must be 2");
5467 }
5468
5469 #[test]
5470 fn aggregate_over_hop_counts_edges() {
5471 let fx = hop_graph();
5472 let v = fx.view();
5473 let params = BTreeMap::new();
5474 let rs = run(
5476 &v,
5477 "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN COUNT(*)",
5478 ¶ms,
5479 )
5480 .expect("COUNT(*) on hop graph");
5481 assert_eq!(rs.row(0), &[Some(i(3))]);
5482 }
5483
5484 #[test]
5489 fn multi_aggregate_no_keys_empty_graph() {
5490 let fx = Fx::new(); let v = fx.view();
5496 let params = BTreeMap::new();
5497
5498 let rs = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
5499 .expect("empty-graph multi-agg must succeed");
5500 assert_eq!(rs.len(), 1, "must produce exactly 1 row on empty input");
5501 assert_eq!(
5502 rs.row(0)[0],
5503 Some(i(0)),
5504 "COUNT(*) on empty graph must be 0"
5505 );
5506 assert_eq!(
5507 rs.row(0)[1],
5508 Some(i(0)),
5509 "COUNT(n) on empty graph must be 0"
5510 );
5511 }
5512
5513 #[test]
5514 fn fast_path_and_group_path_agree_on_empty_input() {
5515 let fx = Fx::new();
5521 let v = fx.view();
5522 let params = BTreeMap::new();
5523
5524 let fast = run(&v, "MATCH (n:N) RETURN COUNT(*)", ¶ms)
5525 .expect("fast-path COUNT on empty graph");
5526 assert_eq!(fast.len(), 1, "fast path: 1 row on empty input");
5527 let fast_count = fast.row(0)[0].clone();
5528
5529 let grouped = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
5530 .expect("group-path COUNT on empty graph");
5531 assert_eq!(grouped.len(), 1, "group path: 1 row on empty input");
5532 let group_count = grouped.row(0)[0].clone();
5533
5534 assert_eq!(
5535 fast_count, group_count,
5536 "fast path and group path must agree on COUNT(*) for empty input"
5537 );
5538
5539 let mut fx2 = Fx::new();
5541 fx2.add("N", "x", vec![]);
5542 fx2.add("N", "y", vec![]);
5543 let v2 = fx2.view();
5544
5545 let fast2 = run(&v2, "MATCH (n:N) RETURN COUNT(*)", ¶ms)
5546 .expect("fast-path COUNT on 2-node graph");
5547 assert_eq!(fast2.row(0)[0], Some(i(2)), "fast path: COUNT(*) = 2");
5548
5549 let grouped2 = run(&v2, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
5550 .expect("group-path COUNT on 2-node graph");
5551 assert_eq!(grouped2.row(0)[0], Some(i(2)), "group path: COUNT(*) = 2");
5552 assert_eq!(grouped2.row(0)[1], Some(i(2)), "group path: COUNT(n) = 2");
5553 }
5554
5555 #[test]
5560 fn grouped_int_float_key_unification() {
5561 let mut fx = Fx::new();
5568 fx.add("N", "a", vec![("score", i(1))]); fx.add("N", "b", vec![("score", f(1.0))]);
5570 let v = fx.view();
5571 let params = BTreeMap::new();
5572
5573 let rs = run(&v, "MATCH (n:N) RETURN n.score, COUNT(*) AS cnt", ¶ms)
5574 .expect("int/float unification must succeed");
5575 assert_eq!(
5576 rs.len(),
5577 1,
5578 "Int(1) and Float(1.0) must group together into 1 group"
5579 );
5580 assert_eq!(
5582 rs.row(0)[0],
5583 Some(i(1)),
5584 "key column must display Int(1) (first-seen)"
5585 );
5586 assert_eq!(rs.row(0)[1], Some(i(2)), "unified group must have count=2");
5587 }
5588
5589 #[test]
5590 fn distinct_collapses_duplicate_projected_values() {
5591 let mut fx = Fx::new();
5592 fx.add("N", "a1", vec![("city", s("Austin"))]);
5593 fx.add("N", "a2", vec![("city", s("Austin"))]);
5594 let v = fx.view();
5595 let params = BTreeMap::new();
5596 let rs = run(&v, "MATCH (n:N) RETURN DISTINCT n.city", ¶ms).unwrap();
5597 assert_eq!(rs.len(), 1);
5598 assert_eq!(rs.row(0)[0], Some(s("Austin")));
5599 }
5600
5601 #[test]
5602 fn distinct_int_float_unify() {
5603 let mut fx = Fx::new();
5604 fx.add("N", "a", vec![("score", i(1))]);
5605 fx.add("N", "b", vec![("score", f(1.0))]);
5606 let v = fx.view();
5607 let params = BTreeMap::new();
5608 let rs = run(&v, "MATCH (n:N) RETURN DISTINCT n.score", ¶ms).unwrap();
5609 assert_eq!(
5610 rs.len(),
5611 1,
5612 "Int(1) and Float(1.0) must DISTINCT as one row"
5613 );
5614 assert_eq!(rs.row(0)[0], Some(i(1)), "first-seen Int(1) wins display");
5615 }
5616
5617 #[test]
5618 fn distinct_caps_at_intermediate_row_budget() {
5619 let mut fx = Fx::new();
5620 fx.add("N", "a", vec![("city", s("Austin"))]);
5621 fx.add("N", "b", vec![("city", s("Paris"))]);
5622 let v = fx.view();
5623 let params = BTreeMap::new();
5624 let err = super::with_max_intermediate_rows(1, || {
5625 run(&v, "MATCH (n:N) RETURN DISTINCT n.city", ¶ms)
5626 })
5627 .expect_err("two distinct cities must exceed cap=1");
5628 assert!(
5629 err.contains("1") || err.contains("row"),
5630 "cap error must mention the budget, got: {err}"
5631 );
5632 }
5633
5634 #[test]
5635 fn where_in_list_filters_rows() {
5636 let mut fx = Fx::new();
5637 fx.add("N", "a", vec![("city", s("Austin"))]);
5638 fx.add("N", "p", vec![("city", s("Paris"))]);
5639 fx.add("N", "l", vec![("city", s("London"))]);
5640 let v = fx.view();
5641 let mut params = BTreeMap::new();
5642 params.insert("c".into(), s("Paris"));
5643 let rs = run(
5644 &v,
5645 "MATCH (n:N) WHERE n.city IN ['Austin', $c] RETURN n.city",
5646 ¶ms,
5647 )
5648 .unwrap();
5649 assert_eq!(rs.len(), 2);
5650 }
5651
5652 #[test]
5653 fn pure_int_keys_display_as_int() {
5654 let mut fx = Fx::new();
5659 fx.add("N", "a", vec![("age", i(10))]);
5660 fx.add("N", "b", vec![("age", i(20))]);
5661 fx.add("N", "c", vec![("age", i(10))]);
5662 let v = fx.view();
5663 let params = BTreeMap::new();
5664
5665 let rs = run(
5666 &v,
5667 "MATCH (n:N) RETURN n.age, COUNT(*) AS cnt ORDER BY n.age",
5668 ¶ms,
5669 )
5670 .expect("pure-Int group keys must succeed");
5671 assert_eq!(rs.len(), 2, "must have 2 groups: age=10 and age=20");
5672 assert_eq!(
5674 rs.row(0)[0],
5675 Some(i(10)),
5676 "age=10 key column must display as Int(10)"
5677 );
5678 assert_eq!(rs.row(0)[1], Some(i(2)), "age=10 group has 2 nodes");
5679 assert_eq!(
5680 rs.row(1)[0],
5681 Some(i(20)),
5682 "age=20 key column must display as Int(20)"
5683 );
5684 assert_eq!(rs.row(1)[1], Some(i(1)), "age=20 group has 1 node");
5685 }
5686
5687 #[test]
5692 fn var_expand_group_aggregate_staged_path() {
5693 let mut fx = Fx::new();
5703 let a = fx.add("N", "a", vec![("key", s("a"))]);
5704 let b = fx.add("N", "b", vec![("key", s("b"))]);
5705 let c = fx.add("N", "c", vec![("key", s("c"))]);
5706 fx.edge("T", a, b, vec![]);
5707 fx.edge("T", b, c, vec![]);
5708 let v = fx.view();
5709 let params = BTreeMap::new();
5710
5711 let rs = run(
5712 &v,
5713 "MATCH (x:N)-[*1..2]->(y:N) RETURN y.key, COUNT(*) AS cnt ORDER BY y.key",
5714 ¶ms,
5715 )
5716 .expect("VarExpand + GroupAggregate must succeed");
5717
5718 assert_eq!(rs.len(), 2, "must have 2 destination groups: b and c");
5719 assert_eq!(rs.row(0)[0], Some(s("b")), "first group key must be 'b'");
5720 assert_eq!(rs.row(0)[1], Some(i(1)), "b is reached via 1 path");
5721 assert_eq!(rs.row(1)[0], Some(s("c")), "second group key must be 'c'");
5722 assert_eq!(
5723 rs.row(1)[1],
5724 Some(i(2)),
5725 "c is reached via 2 paths (1-hop and 2-hop)"
5726 );
5727 }
5728
5729 #[test]
5737 fn pull_rows_var_expand_arm_returns_named_err() {
5738 let fx = Fx::new();
5739 let view = fx.view();
5740 let vars = super::VarTable {
5741 names: vec!["a".into(), "b".into()],
5742 };
5743 let project_items: Vec<crate::cypher::ast::RetItem> = vec![];
5744 let empty_params = BTreeMap::new();
5745 let params = super::Params(&empty_params);
5746 let ctx = super::PullCtx {
5747 view: &view,
5748 vars: &vars,
5749 project_items: &project_items,
5750 params: ¶ms,
5751 bound: 100,
5752 };
5753 let ops = vec![PlanOp::VarExpand {
5754 from: "a".into(),
5755 rel_var: None,
5756 etype: None,
5757 dir: crate::cypher::RelDir::Right,
5758 to: "b".into(),
5759 min: 1,
5760 max: 3,
5761 }];
5762 let mut row = vec![None; vars.names.len()];
5763 let mut result = Vec::new();
5764 let err = super::pull_rows(&ctx, &ops, &mut row, &mut result)
5765 .expect_err("VarExpand must Err in pull_rows");
5766 assert!(
5767 err.contains("VarExpand") && err.contains("pull executor"),
5768 "error must name VarExpand and pull executor, got: {err}"
5769 );
5770 }
5771
5772 #[test]
5773 fn pull_rows_shortest_path_arm_returns_named_err() {
5774 let fx = Fx::new();
5775 let view = fx.view();
5776 let vars = super::VarTable {
5777 names: vec!["a".into(), "b".into()],
5778 };
5779 let project_items: Vec<crate::cypher::ast::RetItem> = vec![];
5780 let empty_params = BTreeMap::new();
5781 let params = super::Params(&empty_params);
5782 let ctx = super::PullCtx {
5783 view: &view,
5784 vars: &vars,
5785 project_items: &project_items,
5786 params: ¶ms,
5787 bound: 100,
5788 };
5789 let ops = vec![PlanOp::ShortestPath {
5790 from: "a".into(),
5791 rel_var: None,
5792 etype: None,
5793 dir: crate::cypher::RelDir::Right,
5794 to: "b".into(),
5795 max_hops: 5,
5796 }];
5797 let mut row = vec![None; vars.names.len()];
5798 let mut result = Vec::new();
5799 let err = super::pull_rows(&ctx, &ops, &mut row, &mut result)
5800 .expect_err("ShortestPath must Err in pull_rows");
5801 assert!(
5802 err.contains("ShortestPath") && err.contains("pull executor"),
5803 "error must name ShortestPath and pull executor, got: {err}"
5804 );
5805 }
5806
5807 fn city_graph() -> Fx {
5811 let mut fx = Fx::new();
5812 fx.add(
5813 "Person",
5814 "alice",
5815 vec![("city", s("Boston")), ("age", i(30))],
5816 );
5817 fx.add("Person", "bob", vec![("city", s("Boston")), ("age", i(25))]);
5818 fx.add(
5819 "Person",
5820 "carol",
5821 vec![("city", s("Austin")), ("age", i(35))],
5822 );
5823 fx.add(
5824 "Person",
5825 "dave",
5826 vec![("city", s("Austin")), ("age", i(28))],
5827 );
5828 fx.add("Person", "eve", vec![("city", s("Boston")), ("age", i(22))]);
5829 fx
5830 }
5831
5832 #[test]
5835 fn with_aggregate_having_filters_groups() {
5836 let fx = city_graph();
5837 let view = fx.view();
5838 let params = BTreeMap::new();
5839 let rs = run(
5840 &view,
5841 "MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt WHERE cnt > 2 RETURN city, cnt",
5842 ¶ms,
5843 )
5844 .expect("WITH HAVING query must succeed");
5845 assert_eq!(rs.len(), 1, "only Boston group has cnt > 2");
5846 assert_eq!(rs.get(0, "city"), Some(&s("Boston")));
5847 assert_eq!(rs.get(0, "cnt"), Some(&i(3)));
5848 }
5849
5850 #[test]
5852 fn with_order_limit_then_return() {
5853 let fx = city_graph();
5854 let view = fx.view();
5855 let params = BTreeMap::new();
5856 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)
5858 .expect("WITH ORDER LIMIT must succeed");
5859 assert_eq!(rs.len(), 2, "LIMIT 2");
5860 let ages: Vec<Option<Value>> = (0..rs.len()).map(|i| rs.get(i, "age").cloned()).collect();
5862 assert_eq!(ages[0], Some(i(35)), "oldest person first");
5863 assert_eq!(ages[1], Some(i(30)), "second oldest");
5864 }
5865
5866 #[test]
5868 fn chained_with_stages() {
5869 let fx = city_graph();
5870 let view = fx.view();
5871 let params = BTreeMap::new();
5872 let rs = run(&view,
5874 "MATCH (p:Person) WITH p.city AS city, p.age AS age WITH city, age WHERE age > 25 RETURN city, age",
5875 ¶ms).expect("chained WITH must succeed");
5876 assert_eq!(rs.len(), 3, "3 people with age > 25");
5878 }
5879
5880 #[test]
5882 fn with_then_match_reentry() {
5883 let mut fx = Fx::new();
5884 let alice = fx.add("Person", "alice", vec![]);
5885 let bob = fx.add("Person", "bob", vec![]);
5886 let corp = fx.add("Company", "acme", vec![]);
5887 fx.edge("WORKS_AT", alice, corp, vec![("years", i(5))]);
5888 fx.edge("WORKS_AT", bob, corp, vec![("years", i(3))]);
5889 let view = fx.view();
5890 let params = BTreeMap::new();
5891 let rs = run(&view,
5893 "MATCH (p:Person)-[r:WORKS_AT]->(c:Company) WITH p, c MATCH (c)-[r2:WORKS_AT]-(colleague:Person) RETURN p, colleague",
5894 ¶ms).expect("WITH MATCH re-entry must succeed");
5895 assert!(rs.len() >= 2, "cross join via company: got {}", rs.len());
5897 }
5898
5899 #[test]
5901 fn unwind_literal_list_produces_rows() {
5902 let fx = city_graph();
5903 let view = fx.view();
5904 let params = BTreeMap::new();
5905 let rs = run(
5906 &view,
5907 "MATCH (p:Person) WHERE p.city = 'Boston' UNWIND [1, 2, 3] AS x RETURN p, x",
5908 ¶ms,
5909 )
5910 .expect("UNWIND literal must succeed");
5911 assert_eq!(rs.len(), 9, "UNWIND [1,2,3] × 3 Boston people = 9 rows");
5913 let xs: Vec<Option<Value>> = (0..rs.len())
5915 .map(|row_i| rs.get(row_i, "x").cloned())
5916 .collect();
5917 assert!(xs.contains(&Some(i(1))));
5918 assert!(xs.contains(&Some(i(2))));
5919 assert!(xs.contains(&Some(i(3))));
5920 }
5921
5922 #[test]
5924 fn unwind_list_property() {
5925 let mut fx = Fx::new();
5926 fx.add(
5927 "Tag",
5928 "post1",
5929 vec![("tags", Value::List(vec![s("rust"), s("graph")]))],
5930 );
5931 fx.add("Tag", "post2", vec![("tags", Value::List(vec![s("db")]))]);
5932 let view = fx.view();
5933 let params = BTreeMap::new();
5934 let rs = run(
5935 &view,
5936 "MATCH (p:Tag) UNWIND p.tags AS tag RETURN p, tag",
5937 ¶ms,
5938 )
5939 .expect("UNWIND property must succeed");
5940 assert_eq!(rs.len(), 3, "2+1 tag elements");
5941 }
5942
5943 #[test]
5945 fn unwind_empty_list_yields_zero_rows() {
5946 let fx = city_graph();
5947 let view = fx.view();
5948 let params = BTreeMap::new();
5949 let rs = run(
5950 &view,
5951 "MATCH (p:Person) WHERE p.city = 'Boston' UNWIND [] AS x RETURN x",
5952 ¶ms,
5953 )
5954 .expect("UNWIND [] must succeed with 0 rows");
5955 assert_eq!(rs.len(), 0, "UNWIND [] should produce 0 rows");
5956 }
5957
5958 #[test]
5960 fn unwind_non_list_is_named_error() {
5961 let fx = city_graph();
5962 let view = fx.view();
5963 let params = BTreeMap::new();
5964 let err = run(
5966 &view,
5967 "MATCH (p:Person) WHERE p.city = 'Boston' UNWIND p.city AS x RETURN x",
5968 ¶ms,
5969 )
5970 .expect_err("UNWIND non-list must error");
5971 assert!(
5972 err.contains("UNWIND") && err.contains("list"),
5973 "error must mention UNWIND and list: {err}"
5974 );
5975 }
5976
5977 #[test]
5979 fn unwind_cross_product_trips_budget() {
5980 let mut fx = Fx::new();
5981 for i in 0..10 {
5983 fx.add("N", &format!("n{i}"), vec![]);
5984 }
5985 let view = fx.view();
5986 let params = BTreeMap::new();
5987 let err = super::with_max_intermediate_rows(5, || {
5988 run(
5989 &view,
5990 "MATCH (n:N) UNWIND [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] AS x RETURN n, x",
5991 ¶ms,
5992 )
5993 })
5994 .expect_err("must hit budget");
5995 assert!(
5996 err.contains("intermediate result exceeds"),
5997 "error must mention intermediate result limit: {err}"
5998 );
5999 }
6000
6001 #[test]
6004 fn routing_with_plan_uses_staged_path() {
6005 use crate::cypher::plan::row_bound;
6006 let ops_with = compile("MATCH (a) WITH a RETURN a");
6007 assert_eq!(
6008 row_bound(&ops_with),
6009 None,
6010 "WITH plan must have row_bound=None"
6011 );
6012
6013 let ops_unwind = compile("MATCH (a) UNWIND [1, 2] AS x RETURN a");
6014 assert_eq!(
6015 row_bound(&ops_unwind),
6016 None,
6017 "UNWIND plan must have row_bound=None"
6018 );
6019
6020 let ops_plain = compile("MATCH (a) RETURN a LIMIT 5");
6022 assert!(
6023 row_bound(&ops_plain).is_some(),
6024 "plain LIMIT plan must have a row bound"
6025 );
6026 }
6027
6028 #[test]
6031 fn unwind_null_property_yields_zero_rows_for_that_node() {
6032 let mut fx = Fx::new();
6033 fx.add(
6035 "Post",
6036 "post1",
6037 vec![("tags", Value::List(vec![s("rust"), s("graph")]))],
6038 );
6039 fx.add("Post", "post2", vec![("other", Value::Str("hello".into()))]);
6040 let view = fx.view();
6041 let params = BTreeMap::new();
6042 let rs = run(
6043 &view,
6044 "MATCH (p:Post) UNWIND p.tags AS tag RETURN tag",
6045 ¶ms,
6046 )
6047 .expect("UNWIND null property must succeed (not error)");
6048 assert_eq!(
6050 rs.len(),
6051 2,
6052 "null property must yield 0 rows; list property yields N rows"
6053 );
6054 let tag0 = rs.get(0, "tag");
6055 let tag1 = rs.get(1, "tag");
6056 let tags = [tag0, tag1];
6057 assert!(tags.contains(&Some(&s("rust"))), "rust tag present");
6058 assert!(tags.contains(&Some(&s("graph"))), "graph tag present");
6059 }
6060
6061 #[test]
6063 fn where_before_unwind_filters_nodes() {
6064 let mut fx = Fx::new();
6065 fx.add(
6066 "P",
6067 "a",
6068 vec![
6069 ("group", Value::Str("keep".into())),
6070 ("items", Value::List(vec![Value::Int(1), Value::Int(2)])),
6071 ],
6072 );
6073 fx.add(
6074 "P",
6075 "b",
6076 vec![
6077 ("group", Value::Str("drop".into())),
6078 ("items", Value::List(vec![Value::Int(3), Value::Int(4)])),
6079 ],
6080 );
6081 let view = fx.view();
6082 let params = BTreeMap::new();
6083 let rs = run(
6085 &view,
6086 "MATCH (n:P) WHERE n.group = 'keep' UNWIND n.items AS x RETURN x",
6087 ¶ms,
6088 )
6089 .expect("WHERE before UNWIND must succeed");
6090 assert_eq!(rs.len(), 2, "only matching node expands; 2 items");
6091 }
6092
6093 #[test]
6095 fn where_after_unwind_filters_expanded_rows() {
6096 let mut fx = Fx::new();
6097 fx.add(
6098 "N",
6099 "n1",
6100 vec![(
6101 "vals",
6102 Value::List(vec![Value::Int(1), Value::Int(3), Value::Int(5)]),
6103 )],
6104 );
6105 let view = fx.view();
6106 let params = BTreeMap::new();
6107 let rs = run(
6109 &view,
6110 "MATCH (n:N) UNWIND n.vals AS x WHERE x > 2 RETURN x",
6111 ¶ms,
6112 )
6113 .expect("WHERE after UNWIND must succeed");
6114 assert_eq!(rs.len(), 2, "x=3 and x=5 pass; x=1 filtered out");
6115 let v0 = rs.get(0, "x").cloned();
6116 let v1 = rs.get(1, "x").cloned();
6117 let vals = [v0, v1];
6118 assert!(vals.contains(&Some(Value::Int(3))), "x=3 present");
6119 assert!(vals.contains(&Some(Value::Int(5))), "x=5 present");
6120 }
6121
6122 #[test]
6124 fn aggregate_with_order_by_unknown_alias_is_named_error() {
6125 use crate::cypher::{lex, parser::parse, plan::plan};
6126 let src = "MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt ORDER BY cntt DESC RETURN city, cnt";
6128 let plan_result = plan(&parse(&lex(src).expect("lex")).expect("parse"));
6129 let err = plan_result
6130 .expect_err("typo'd ORDER BY alias in aggregate WITH must be a named plan error");
6131 assert!(
6132 err.contains("cntt") || err.contains("unbound"),
6133 "error must mention the unknown alias: {err}"
6134 );
6135 }
6136
6137 #[test]
6139 fn non_aggregate_with_order_by_unknown_alias_is_named_error() {
6140 use crate::cypher::{lex, parser::parse, plan::plan};
6141 let src = "MATCH (p:Person) WITH p, p.age AS age ORDER BY nope DESC RETURN p.city AS city";
6142 let plan_result = plan(&parse(&lex(src).expect("lex")).expect("parse"));
6143 let err = plan_result
6144 .expect_err("typo'd ORDER BY alias in non-aggregate WITH must be a named plan error");
6145 assert!(
6146 err.contains("nope") || err.contains("unbound"),
6147 "error must mention the unknown alias: {err}"
6148 );
6149 }
6150
6151 #[test]
6155 fn post_unwind_where_references_pre_match_variable() {
6156 let mut fx = Fx::new();
6157 fx.add(
6159 "N",
6160 "n1",
6161 vec![
6162 ("threshold", Value::Int(3)),
6163 (
6164 "xs",
6165 Value::List(vec![
6166 Value::Int(1),
6167 Value::Int(2),
6168 Value::Int(3),
6169 Value::Int(4),
6170 Value::Int(5),
6171 ]),
6172 ),
6173 ],
6174 );
6175 fx.add(
6177 "N",
6178 "n2",
6179 vec![
6180 ("threshold", Value::Int(10)),
6181 ("xs", Value::List(vec![Value::Int(1), Value::Int(2)])),
6182 ],
6183 );
6184 let view = fx.view();
6185 let params = BTreeMap::new();
6186 let rs = run(
6187 &view,
6188 "MATCH (n:N) UNWIND n.xs AS x WHERE x > n.threshold RETURN x",
6189 ¶ms,
6190 )
6191 .expect("post-UNWIND WHERE referencing MATCH variable must succeed");
6192 assert_eq!(rs.len(), 2, "exactly 2 rows: x=4 and x=5 from n1");
6194 let v0 = rs.get(0, "x").cloned();
6195 let v1 = rs.get(1, "x").cloned();
6196 let got = [v0, v1];
6197 assert!(got.contains(&Some(Value::Int(4))), "x=4 present");
6198 assert!(got.contains(&Some(Value::Int(5))), "x=5 present");
6199 }
6200
6201 #[test]
6205 fn with_stage_aggregate_order_by_descending() {
6206 let fx = city_graph();
6209 let view = fx.view();
6210 let params = BTreeMap::new();
6211 let rs = run(
6212 &view,
6213 "MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt ORDER BY cnt DESC RETURN city, cnt",
6214 ¶ms,
6215 )
6216 .expect("aggregate WITH ORDER BY must succeed");
6217 assert_eq!(rs.len(), 2, "two city groups");
6218 assert_eq!(
6220 rs.get(0, "cnt"),
6221 Some(&Value::Int(3)),
6222 "first row must be the group with cnt=3 (Boston)"
6223 );
6224 assert_eq!(
6225 rs.get(1, "cnt"),
6226 Some(&Value::Int(2)),
6227 "second row must be the group with cnt=2 (Austin)"
6228 );
6229 }
6230
6231 #[test]
6235 fn unwind_then_with_composition() {
6236 let mut fx = Fx::new();
6237 fx.add(
6238 "Doc",
6239 "d1",
6240 vec![(
6241 "scores",
6242 Value::List(vec![Value::Int(10), Value::Int(20), Value::Int(30)]),
6243 )],
6244 );
6245 fx.add(
6246 "Doc",
6247 "d2",
6248 vec![("scores", Value::List(vec![Value::Int(5), Value::Int(15)]))],
6249 );
6250 let view = fx.view();
6251 let params = BTreeMap::new();
6252
6253 let rs = run(
6255 &view,
6256 "MATCH (n:Doc) UNWIND n.scores AS x WITH x RETURN x",
6257 ¶ms,
6258 )
6259 .expect("UNWIND then WITH pass-through must succeed");
6260 assert_eq!(rs.len(), 5, "3 + 2 = 5 expanded rows carried through WITH");
6261
6262 let rs2 = run(
6264 &view,
6265 "MATCH (n:Doc) UNWIND n.scores AS x WITH COUNT(*) AS total RETURN total",
6266 ¶ms,
6267 )
6268 .expect("UNWIND then aggregate WITH must succeed");
6269 assert_eq!(rs2.len(), 1, "single aggregate row");
6270 assert_eq!(
6271 rs2.get(0, "total"),
6272 Some(&Value::Int(5)),
6273 "total must be 5 (3+2 expanded rows)"
6274 );
6275 }
6276
6277 #[test]
6283 fn binarith_div_min_over_neg1_does_not_panic() {
6284 let fx = Fx::new();
6285 let view = fx.view();
6286 let vars = VarTable { names: vec![] };
6287 let row: Row = vec![];
6288 let params = BTreeMap::new();
6289
6290 let operand = Operand::BinArith {
6292 op: ArithOp::Div,
6293 left: Box::new(Operand::Lit(Value::Int(i64::MIN))),
6294 right: Box::new(Operand::Lit(Value::Int(-1))),
6295 };
6296
6297 let result = resolve_operand(&view, &vars, &row, &operand, &Params(¶ms));
6298 assert!(
6299 result.is_ok(),
6300 "overflow division must not return Err: {result:?}"
6301 );
6302 assert_eq!(
6303 result.unwrap(),
6304 Some(Value::Int(i64::MAX)),
6305 "i64::MIN / -1 must saturate to i64::MAX, not panic"
6306 );
6307 }
6308
6309 #[test]
6314 fn binarith_missing_param_caught_at_preflight() {
6315 let fx = Fx::new();
6316 let view = fx.view();
6317 let err = run(
6322 &view,
6323 "MATCH (n:X) RETURN abs($missing - 1)",
6324 &BTreeMap::new(),
6325 )
6326 .expect_err("missing param inside BinArith must be caught at preflight");
6327 assert!(
6328 err.contains("missing"),
6329 "error must name the param 'missing', got: {err}"
6330 );
6331 }
6332
6333 #[test]
6337 fn is_null_filters_absent_prop() {
6338 let mut fx = Fx::new();
6339 fx.add("Person", "alice", vec![("age", Value::Int(30))]);
6340 fx.add("Person", "bob", vec![]); let view = fx.view();
6342 let rs = run(
6343 &view,
6344 "MATCH (n:Person) WHERE n.age IS NULL RETURN n",
6345 &BTreeMap::new(),
6346 )
6347 .unwrap();
6348 assert_eq!(rs.len(), 1, "only bob lacks age");
6349 assert_eq!(rs.get(0, "n"), Some(&s("bob")));
6350 }
6351
6352 #[test]
6354 fn is_not_null_filters_present_prop() {
6355 let mut fx = Fx::new();
6356 fx.add("Person", "alice", vec![("age", Value::Int(30))]);
6357 fx.add("Person", "bob", vec![]); let view = fx.view();
6359 let rs = run(
6360 &view,
6361 "MATCH (n:Person) WHERE n.age IS NOT NULL RETURN n",
6362 &BTreeMap::new(),
6363 )
6364 .unwrap();
6365 assert_eq!(rs.len(), 1, "only alice has age");
6366 assert_eq!(rs.get(0, "n"), Some(&s("alice")));
6367 }
6368
6369 #[test]
6372 fn optional_match_is_null_anti_join() {
6373 let mut fx = Fx::new();
6374 let alice = fx.add("Person", "alice", vec![]);
6375 let bob = fx.add("Person", "bob", vec![]);
6376 let carol = fx.add("Person", "carol", vec![]);
6377 fx.edge("KNOWS", alice, carol, vec![]); fx.edge("KNOWS", carol, bob, vec![]); let view = fx.view();
6381 let rs = run(
6382 &view,
6383 "MATCH (a:Person) OPTIONAL MATCH (a)-[:KNOWS]->(b) WITH a, b WHERE b IS NULL RETURN a",
6384 &BTreeMap::new(),
6385 )
6386 .unwrap();
6387 assert_eq!(rs.len(), 1, "only bob has no outgoing KNOWS edge");
6388 assert_eq!(rs.get(0, "a"), Some(&s("bob")));
6389 }
6390
6391 #[test]
6393 fn is_null_combined_with_and_exec() {
6394 let mut fx = Fx::new();
6395 fx.add("N", "a", vec![("x", Value::Int(1))]);
6396 fx.add("N", "b", vec![("x", Value::Int(2)), ("y", Value::Int(9))]);
6397 fx.add("N", "c", vec![]); let view = fx.view();
6399 let rs = run(
6401 &view,
6402 "MATCH (n:N) WHERE n.y IS NULL AND n.x IS NOT NULL RETURN n",
6403 &BTreeMap::new(),
6404 )
6405 .unwrap();
6406 assert_eq!(rs.len(), 1);
6407 assert_eq!(rs.get(0, "n"), Some(&s("a")));
6408 }
6409
6410 #[test]
6414 fn arithmetic_add_in_return() {
6415 let mut fx = Fx::new();
6416 fx.add("Person", "alice", vec![("age", Value::Int(29))]);
6417 let view = fx.view();
6418 let rs = run(
6419 &view,
6420 "MATCH (n:Person) RETURN n.age + 1 AS adjusted",
6421 &BTreeMap::new(),
6422 )
6423 .unwrap();
6424 assert_eq!(rs.len(), 1);
6425 assert_eq!(rs.get(0, "adjusted"), Some(&Value::Int(30)));
6426 }
6427
6428 #[test]
6430 fn arithmetic_precedence_parens_pin() {
6431 let mut fx = Fx::new();
6432 fx.add("N", "x", vec![]);
6433 let view = fx.view();
6434 let rs = run(
6435 &view,
6436 "MATCH (n:N) RETURN (1 + 2) * 3 AS r",
6437 &BTreeMap::new(),
6438 )
6439 .unwrap();
6440 assert_eq!(rs.len(), 1);
6441 assert_eq!(rs.get(0, "r"), Some(&Value::Int(9)));
6442 }
6443
6444 #[test]
6446 fn arithmetic_precedence_mul_over_add_pin() {
6447 let mut fx = Fx::new();
6448 fx.add("N", "x", vec![]);
6449 let view = fx.view();
6450 let rs = run(&view, "MATCH (n:N) RETURN 1 + 2 * 3 AS r", &BTreeMap::new()).unwrap();
6451 assert_eq!(rs.len(), 1);
6452 assert_eq!(rs.get(0, "r"), Some(&Value::Int(7)));
6453 }
6454
6455 #[test]
6457 fn arithmetic_div_by_zero_returns_error() {
6458 let mut fx = Fx::new();
6459 fx.add("N", "x", vec![]);
6460 let view = fx.view();
6461 let err = run(
6462 &view,
6463 "MATCH (n:N) WHERE 1 / 0 > 0 RETURN n",
6464 &BTreeMap::new(),
6465 )
6466 .expect_err("division by zero must error");
6467 assert!(
6468 err.contains("division by zero"),
6469 "error must mention division by zero, got: {err}"
6470 );
6471 }
6472
6473 #[test]
6475 fn arithmetic_null_propagates() {
6476 let mut fx = Fx::new();
6477 fx.add("N", "x", vec![]); let view = fx.view();
6479 let rs = run(
6481 &view,
6482 "MATCH (n:N) WHERE n.val + 1 > 0 RETURN n",
6483 &BTreeMap::new(),
6484 )
6485 .unwrap();
6486 assert_eq!(rs.len(), 0, "null arithmetic must not match");
6487 }
6488
6489 #[test]
6491 fn arithmetic_in_where_comparison() {
6492 let mut fx = Fx::new();
6493 fx.add("Person", "alice", vec![("age", Value::Int(5))]); fx.add("Person", "bob", vec![("age", Value::Int(4))]); let view = fx.view();
6496 let rs = run(
6497 &view,
6498 "MATCH (n:Person) WHERE n.age + 1 > 5 RETURN n",
6499 &BTreeMap::new(),
6500 )
6501 .unwrap();
6502 assert_eq!(rs.len(), 1);
6503 assert_eq!(rs.get(0, "n"), Some(&s("alice")));
6504 }
6505}