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
28#[cfg(test)]
31pub static INDEX_SCAN_FIRES: std::sync::atomic::AtomicUsize =
32 std::sync::atomic::AtomicUsize::new(0);
33
34#[cfg(test)]
38pub static INDEX_INTERSECT_FIRES: std::sync::atomic::AtomicUsize =
39 std::sync::atomic::AtomicUsize::new(0);
40
41pub struct Params<'a>(pub &'a BTreeMap<String, Value>);
44
45#[derive(Clone, Debug)]
46enum Cell {
47 Node(u32),
48 Rel(EdgeRef),
49 Path(u8),
52 Scalar(Value),
54}
55
56type Row = Vec<Option<Cell>>;
59
60struct VarTable {
61 names: Vec<String>,
62}
63
64impl VarTable {
65 fn intern(&mut self, name: &str) -> usize {
66 if let Some(i) = self.names.iter().position(|n| n == name) {
67 return i;
68 }
69 self.names.push(name.to_string());
70 self.names.len() - 1
71 }
72
73 fn slot(&self, name: &str) -> Option<usize> {
74 self.names.iter().position(|n| n == name)
75 }
76}
77
78struct Projected {
79 columns: Vec<String>,
80 rows: Vec<Vec<Option<Value>>>,
81}
82
83const MAX_INTERMEDIATE_ROWS: usize = 1_000_000;
86
87const MAX_GROUPS: usize = 1_000_000;
89
90type GroupKey = Vec<Option<ValueKey>>;
93type GroupEntry = (Vec<Option<Value>>, Vec<AggAcc>);
98
99#[cfg(test)]
100thread_local! {
101 static TEST_MAX_INTERMEDIATE_ROWS: std::cell::Cell<Option<usize>> =
102 const { std::cell::Cell::new(None) };
103 static TEST_EXPAND_PRODUCED: std::cell::Cell<Option<usize>> =
106 const { std::cell::Cell::new(None) };
107 static TEST_MAX_GROUPS: std::cell::Cell<Option<usize>> =
109 const { std::cell::Cell::new(None) };
110}
111
112fn max_intermediate_rows() -> usize {
113 #[cfg(test)]
114 {
115 TEST_MAX_INTERMEDIATE_ROWS
116 .with(|c| c.get())
117 .unwrap_or(MAX_INTERMEDIATE_ROWS)
118 }
119 #[cfg(not(test))]
120 {
121 MAX_INTERMEDIATE_ROWS
122 }
123}
124
125fn max_groups() -> usize {
126 #[cfg(test)]
127 {
128 TEST_MAX_GROUPS.with(|c| c.get()).unwrap_or(MAX_GROUPS)
129 }
130 #[cfg(not(test))]
131 {
132 MAX_GROUPS
133 }
134}
135
136#[cfg(test)]
139pub(crate) fn with_max_groups<R>(cap: usize, f: impl FnOnce() -> R) -> R {
140 TEST_MAX_GROUPS.with(|c| {
141 let prev = c.replace(Some(cap));
142 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
143 c.set(prev);
144 match result {
145 Ok(v) => v,
146 Err(p) => std::panic::resume_unwind(p),
147 }
148 })
149}
150
151#[cfg(test)]
155pub(crate) fn with_max_intermediate_rows<R>(cap: usize, f: impl FnOnce() -> R) -> R {
156 TEST_MAX_INTERMEDIATE_ROWS.with(|c| {
157 let prev = c.replace(Some(cap));
158 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
159 c.set(prev);
160 match result {
161 Ok(v) => v,
162 Err(p) => std::panic::resume_unwind(p),
163 }
164 })
165}
166
167#[cfg(test)]
169fn record_expand_row() {
170 TEST_EXPAND_PRODUCED.with(|c| {
171 if let Some(prev) = c.get() {
172 c.set(Some(prev + 1));
173 }
174 });
175}
176
177#[cfg(test)]
183pub(crate) fn with_expand_counter<R>(f: impl FnOnce() -> R) -> (R, usize) {
184 TEST_EXPAND_PRODUCED.with(|c| c.set(Some(0)));
185 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
186 let count = TEST_EXPAND_PRODUCED.with(|c| c.get().unwrap_or(0));
187 TEST_EXPAND_PRODUCED.with(|c| c.set(None));
188 match result {
189 Ok(v) => (v, count),
190 Err(p) => std::panic::resume_unwind(p),
191 }
192}
193
194fn row_cap_err(cap: usize) -> String {
195 format!(
196 "intermediate result exceeds {cap} rows; add a LIMIT or constrain patterns with shared variables"
197 )
198}
199
200fn group_cap_err() -> String {
201 format!(
202 "group count exceeds {} distinct keys; add a WHERE clause or constrain the grouping key",
203 max_groups()
204 )
205}
206
207fn group_key_normalize(v: &Value) -> Option<ValueKey> {
218 match v {
219 Value::Int(n) => Some(ValueKey::FloatBits((*n as f64).to_bits())),
220 Value::Float(f) => Some(ValueKey::FloatBits(f.to_bits())),
221 _ => ValueKey::from_value(v),
222 }
223}
224
225pub fn execute(view: &GraphView, plan: &[PlanOp], params: &Params) -> Result<ResultSet, String> {
237 use crate::cypher::plan::row_bound;
238 execute_inner(view, plan, params, row_bound(plan))
239}
240
241pub fn execute_union(
246 view: &GraphView,
247 union: &crate::cypher::parser::UnionQuery,
248 params: &Params,
249) -> Result<ResultSet, String> {
250 let mut acc: Option<ResultSet> = None;
251 for (i, part) in union.parts.iter().enumerate() {
252 let ops = crate::cypher::plan::plan(part)?;
253 let rs = execute(view, &ops, params)?;
254 acc = Some(match acc {
255 None => rs,
256 Some(prev) => {
257 if prev.columns() != rs.columns() {
258 return Err(format!(
259 "UNION requires matching column names across all parts; \
260 got {:?} then {:?}",
261 prev.columns(),
262 rs.columns()
263 ));
264 }
265 let all = union.all_flags.get(i - 1).copied().unwrap_or(false);
267 combine_result_sets(prev, rs, all)
268 }
269 });
270 }
271 Ok(acc.expect("UNION query has at least one part"))
273}
274
275fn combine_result_sets(mut acc: ResultSet, other: ResultSet, all: bool) -> ResultSet {
276 for i in 0..other.len() {
277 acc.push_row(other.row(i).to_vec());
278 }
279 if !all {
280 let mut seen: Vec<Vec<Option<Value>>> = Vec::new();
282 for i in 0..acc.len() {
283 let r = acc.row(i).to_vec();
284 if !seen.contains(&r) {
285 seen.push(r);
286 }
287 }
288 let mut deduped = ResultSet::new(acc.columns().to_vec());
289 for r in seen {
290 deduped.push_row(r);
291 }
292 return deduped;
293 }
294 acc
295}
296
297#[cfg(test)]
302pub(crate) fn execute_unbounded(
303 view: &GraphView,
304 plan: &[PlanOp],
305 params: &Params,
306) -> Result<ResultSet, String> {
307 execute_inner(view, plan, params, None)
308}
309
310fn execute_inner(
311 view: &GraphView,
312 plan: &[PlanOp],
313 params: &Params,
314 row_bound: Option<usize>,
315) -> Result<ResultSet, String> {
316 check_params(plan, params)?;
317
318 let has_var_expand = plan
324 .iter()
325 .any(|op| matches!(op, PlanOp::VarExpand { .. } | PlanOp::ShortestPath { .. }));
326
327 let is_pipeline = plan.iter().any(|op| {
333 matches!(
334 op,
335 PlanOp::With { .. } | PlanOp::Unwind { .. } | PlanOp::LeftOuterApply { .. }
336 )
337 }) || {
338 let mut saw_gagg = false;
339 plan.iter().any(|op| {
340 if matches!(op, PlanOp::GroupAggregate { .. }) {
341 saw_gagg = true;
342 false
343 } else {
344 saw_gagg && matches!(op, PlanOp::Filter { .. })
345 }
346 })
347 };
348
349 if !has_var_expand
353 && !is_pipeline
354 && plan
355 .iter()
356 .any(|op| matches!(op, PlanOp::GroupAggregate { .. }))
357 {
358 return execute_group_aggregate(view, plan, params);
359 }
360
361 if !has_var_expand
365 && !is_pipeline
366 && plan.iter().any(|op| matches!(op, PlanOp::Aggregate { .. }))
367 {
368 return execute_aggregate(view, plan, params);
369 }
370
371 if let Some(bound) = row_bound {
375 return execute_pull(view, plan, params, bound);
376 }
377
378 let vars = collect_vars(plan);
380 let mut rows: Vec<Row> = vec![vec![None; vars.names.len()]];
381 let mut projected: Option<Projected> = None;
382 let mut pipeline_group_columns: Option<Vec<String>> = None;
385
386 for op in plan {
387 match op {
388 PlanOp::ScanLabel { var, label } => {
389 rows = scan_label(view, &vars, &rows, var, label.as_deref())?;
390 }
391 PlanOp::ScanKey { var, key, label } => {
392 rows = scan_key(view, &vars, &rows, var, key, label.as_deref(), params)?;
393 }
394 PlanOp::IndexScan {
395 var,
396 label,
397 field,
398 value,
399 } => {
400 rows = scan_index(
401 view,
402 &vars,
403 &rows,
404 var,
405 label.as_deref(),
406 field,
407 value,
408 params,
409 )?;
410 }
411 PlanOp::IndexIntersect {
412 var,
413 label,
414 equalities,
415 } => {
416 rows = scan_intersect(
417 view,
418 &vars,
419 &rows,
420 var,
421 label.as_deref(),
422 equalities,
423 params,
424 )?;
425 }
426 PlanOp::LookupProps { var, props } => {
427 rows = retain_node(view, &vars, &rows, var, None, props, params)?;
428 }
429 PlanOp::JoinBound { var, label, props } => {
430 rows = retain_node(view, &vars, &rows, var, label.as_deref(), props, params)?;
431 }
432 PlanOp::Expand { .. } => {
433 rows = exec_expand(view, &vars, &rows, op, params)?;
434 }
435 PlanOp::VarExpand {
436 from,
437 rel_var,
438 etypes,
439 dir,
440 to,
441 min,
442 max,
443 } => {
444 rows = exec_var_expand(
445 view, &vars, &rows, from, rel_var, etypes, *dir, to, *min, *max,
446 )?;
447 }
448 PlanOp::ShortestPath {
449 from,
450 rel_var,
451 etypes,
452 dir,
453 to,
454 max_hops,
455 } => {
456 rows = exec_shortest_path(
457 view, &vars, &rows, from, rel_var, etypes, *dir, to, *max_hops,
458 )?;
459 }
460 PlanOp::Filter { expr } => {
461 rows = exec_filter(view, &vars, &rows, expr, params)?;
462 }
463 PlanOp::Project { items } => {
464 projected = Some(exec_project(view, &vars, &rows, items, params)?);
465 }
466 PlanOp::Distinct => {
467 if let Some(table) = projected.as_mut() {
468 exec_distinct(table)?;
469 } else {
470 return Err("DISTINCT requires a Project".into());
471 }
472 }
473 PlanOp::OrderBy { items } => {
474 if let Some(table) = projected.as_mut() {
475 exec_order_by(table, items)?;
476 } else {
477 exec_order_by_rows(&vars, &mut rows, items, view);
479 }
480 }
481 PlanOp::Skip(ls) => {
482 let n = resolve_ls(ls, params)?;
483 if let Some(table) = projected.as_mut() {
484 apply_skip(&mut table.rows, n);
485 } else {
486 apply_skip(&mut rows, n);
487 }
488 }
489 PlanOp::Limit(ls) => {
490 let n = resolve_ls(ls, params)?;
491 if let Some(table) = projected.as_mut() {
492 apply_limit(&mut table.rows, n);
493 } else {
494 apply_limit(&mut rows, n);
495 }
496 }
497 PlanOp::GroupAggregate { keys, aggs } => {
502 let mut grp_groups: HashMap<GroupKey, GroupEntry> = HashMap::new();
503 let mut grp_key_order: Vec<GroupKey> = Vec::new();
504 for row in &rows {
505 let mut gk: GroupKey = Vec::with_capacity(keys.len());
506 let mut display_vals: Vec<Option<Value>> = Vec::with_capacity(keys.len());
507 for (_, item) in keys {
508 let val = project_item(view, &vars, row, item, params)?;
509 gk.push(val.as_ref().and_then(group_key_normalize));
510 display_vals.push(val);
511 }
512 if !grp_groups.contains_key(&gk) {
513 if grp_groups.len() >= max_groups() {
514 return Err(group_cap_err());
515 }
516 grp_key_order.push(gk.clone());
517 grp_groups.insert(
518 gk.clone(),
519 (
520 display_vals,
521 aggs.iter().map(|(f, _, _)| AggAcc::new(f)).collect(),
522 ),
523 );
524 }
525 let (_, accs) = grp_groups.get_mut(&gk).unwrap();
526 for (acc, (func, arg, _)) in accs.iter_mut().zip(aggs.iter()) {
527 update_acc(view, &vars, row, func, arg, acc)?;
528 }
529 }
530 if keys.is_empty() && grp_key_order.is_empty() {
533 let empty_key: GroupKey = vec![];
534 grp_key_order.push(empty_key.clone());
535 grp_groups.insert(
536 empty_key,
537 (
538 vec![],
539 aggs.iter().map(|(f, _, _)| AggAcc::new(f)).collect(),
540 ),
541 );
542 }
543 if is_pipeline {
544 rows = group_result_to_rows(keys, aggs, grp_key_order, &mut grp_groups, &vars);
547 projected = None;
548 let mut cols: Vec<String> = keys.iter().map(|(c, _)| c.clone()).collect();
551 cols.extend(aggs.iter().map(|(_, _, c)| c.clone()));
552 pipeline_group_columns = Some(cols);
553 } else {
554 projected = Some(build_group_projected(
555 keys,
556 aggs,
557 grp_key_order,
558 &mut grp_groups,
559 ));
560 pipeline_group_columns = None;
561 }
562 }
563 PlanOp::With {
566 items,
567 where_expr,
568 order_by,
569 skip,
570 limit,
571 } => {
572 let row_len = vars.names.len();
576 let mut new_rows: Vec<Row> = Vec::with_capacity(rows.len());
577 for row in &rows {
578 let mut new_row: Row = vec![None; row_len];
579 for item in items {
580 let col = column_name(item);
581 let Some(dst_slot) = vars.slot(&col) else {
582 continue;
583 };
584 match &item.value {
585 RetVal::Var(v) => {
586 if let Some(src_slot) = vars.slot(v) {
588 new_row[dst_slot] = row.get(src_slot).cloned().flatten();
589 }
590 }
591 RetVal::Prop { var, field } => {
592 let val = resolve_prop(view, &vars, row, var, field)?;
593 new_row[dst_slot] = val.map(Cell::Scalar);
594 }
595 RetVal::Agg { .. } => {} RetVal::FuncCall { name, args } => {
597 let val = eval_func(name, args, view, &vars, row, params)?;
598 new_row[dst_slot] = val.map(Cell::Scalar);
599 }
600 RetVal::ScalarExpr(op) => {
601 let val = resolve_operand(view, &vars, row, op, params)?;
602 new_row[dst_slot] = val.map(Cell::Scalar);
603 }
604 }
605 }
606 new_rows.push(new_row);
607 }
608 rows = new_rows;
609 if let Some(expr) = where_expr {
611 rows = exec_filter(view, &vars, &rows, expr, params)?;
612 }
613 if !order_by.is_empty() {
615 exec_order_by_rows(&vars, &mut rows, order_by, view);
616 }
617 if let Some(ls) = skip {
618 let n = resolve_ls(ls, params)?;
619 apply_skip(&mut rows, n);
620 }
621 if let Some(ls) = limit {
622 let n = resolve_ls(ls, params)?;
623 apply_limit(&mut rows, n);
624 }
625 }
626 PlanOp::Unwind { expr, alias } => {
628 let alias_slot = vars
629 .slot(alias)
630 .ok_or_else(|| format!("UNWIND alias `{alias}` not in VarTable"))?;
631 let cap = max_intermediate_rows();
632 let mut new_rows: Vec<Row> = Vec::new();
633 for row in &rows {
634 let list_val: Option<Value> = match expr {
635 UnwindExpr::Lit(vals) => Some(Value::List(vals.clone())),
636 UnwindExpr::Prop { var, field } => {
637 resolve_prop(view, &vars, row, var, field)?
638 }
639 UnwindExpr::Var(name) => {
640 let slot = vars
641 .slot(name)
642 .ok_or_else(|| format!("UNWIND variable `{name}` is not bound"))?;
643 match row.get(slot).and_then(|c| c.as_ref()) {
644 Some(Cell::Scalar(v)) => Some(v.clone()),
645 Some(Cell::Node(_) | Cell::Rel(_) | Cell::Path(_)) => {
646 return Err(format!(
647 "UNWIND requires a list; `{name}` is bound to a graph element"
648 ));
649 }
650 None => None,
651 }
652 }
653 };
654 match list_val {
655 None => {} Some(Value::List(items_list)) => {
657 for item_val in items_list {
659 if new_rows.len() >= cap {
660 return Err(row_cap_err(cap));
661 }
662 let mut new_row = row.clone();
663 new_row[alias_slot] = Some(Cell::Scalar(item_val));
664 new_rows.push(new_row);
665 }
666 }
667 Some(other) => {
668 let type_name = match &other {
669 Value::Int(_) => "Int",
670 Value::Float(_) => "Float",
671 Value::Str(_) => "Str",
672 Value::Bool(_) => "Bool",
673 Value::List(_) => unreachable!(),
674 Value::Map(_) => "Map",
675 };
676 return Err(format!(
677 "UNWIND requires a list; got {type_name} value for `{alias}`"
678 ));
679 }
680 }
681 }
682 rows = new_rows;
683 }
684 PlanOp::Aggregate { func, arg, column } => {
689 let ctx = AggStreamCtx {
690 view,
691 vars: &vars,
692 params,
693 func,
694 arg,
695 };
696 let mut acc = AggAcc::new(func);
697 for row in &rows {
698 agg_stream(&ctx, &[], row, &mut acc)?;
700 }
701 let value = acc.finish();
702 let mut rs = ResultSet::new(vec![column.clone()]);
703 rs.push_row(vec![value]);
704 return Ok(rs);
705 }
706 PlanOp::LeftOuterApply {
712 inner,
713 optional_vars,
714 } => {
715 let cap = max_intermediate_rows();
716 let mut new_rows: Vec<Row> = Vec::new();
717 for outer_row in &rows {
718 let inner_seed: Vec<Row> = vec![outer_row.clone()];
720 let inner_result =
723 exec_left_outer_inner(view, &vars, inner_seed, inner, params)?;
724 if inner_result.is_empty() {
725 let mut null_row = outer_row.clone();
727 for v in optional_vars {
728 if let Some(slot) = vars.slot(v) {
729 null_row[slot] = None;
730 }
731 }
732 if new_rows.len() >= cap {
733 return Err(row_cap_err(cap));
734 }
735 new_rows.push(null_row);
736 } else {
737 for r in inner_result {
738 if new_rows.len() >= cap {
739 return Err(row_cap_err(cap));
740 }
741 new_rows.push(r);
742 }
743 }
744 }
745 rows = new_rows;
746 }
747 }
748 }
749
750 Ok(match projected {
751 Some(table) => finish(table),
752 None => {
753 if let Some(cols) = pipeline_group_columns {
756 let mut rs = ResultSet::new(cols.clone());
757 for row in rows {
758 let vals: Vec<Option<Value>> = cols
759 .iter()
760 .map(|col| {
761 vars.slot(col)
762 .and_then(|s| row.get(s))
763 .and_then(|c| c.as_ref())
764 .and_then(|c| match c {
765 Cell::Scalar(v) => Some(v.clone()),
766 Cell::Node(id) => {
767 view.ids.key_of(*id).map(|k| Value::Str(k.to_owned()))
768 }
769 Cell::Path(h) => Some(Value::Int(*h as i64)),
770 Cell::Rel(_) => None,
771 })
772 })
773 .collect();
774 rs.push_row(vals);
775 }
776 rs
777 } else {
778 ResultSet::new(vec![])
779 }
780 }
781 })
782}
783
784fn exec_left_outer_inner(
793 view: &GraphView,
794 vars: &VarTable,
795 mut rows: Vec<Row>,
796 inner: &[PlanOp],
797 params: &Params,
798) -> Result<Vec<Row>, String> {
799 for op in inner {
800 match op {
801 PlanOp::ScanLabel { var, label } => {
802 rows = scan_label(view, vars, &rows, var, label.as_deref())?;
803 }
804 PlanOp::ScanKey { var, key, label } => {
805 rows = scan_key(view, vars, &rows, var, key, label.as_deref(), params)?;
806 }
807 PlanOp::LookupProps { var, props } => {
808 rows = retain_node(view, vars, &rows, var, None, props, params)?;
809 }
810 PlanOp::JoinBound { var, label, props } => {
811 rows = retain_node(view, vars, &rows, var, label.as_deref(), props, params)?;
812 }
813 PlanOp::Expand { .. } => {
814 rows = exec_expand(view, vars, &rows, op, params)?;
815 }
816 PlanOp::Filter { expr } => {
817 rows = exec_filter(view, vars, &rows, expr, params)?;
818 }
819 other => {
820 return Err(format!(
821 "unsupported op inside OPTIONAL MATCH inner plan: {other:?}"
822 ));
823 }
824 }
825 }
826 Ok(rows)
827}
828
829fn finish(table: Projected) -> ResultSet {
830 let mut rs = ResultSet::new(table.columns);
831 for row in table.rows {
832 rs.push_row(row);
833 }
834 rs
835}
836
837fn check_params(plan: &[PlanOp], params: &Params) -> Result<(), String> {
838 let mut names = Vec::new();
839 let mut seen = BTreeSet::new();
840 collect_params_from_ops(plan, &mut names, &mut seen)?;
841 for name in names {
842 if !params.0.contains_key(&name) {
843 return Err(format!("missing parameter `{name}`"));
844 }
845 }
846 Ok(())
847}
848
849fn collect_params_from_ops(
850 plan: &[PlanOp],
851 names: &mut Vec<String>,
852 seen: &mut BTreeSet<String>,
853) -> Result<(), String> {
854 for op in plan {
855 match op {
856 PlanOp::ScanKey { key, .. } => collect_operand(key, names, seen),
857 PlanOp::LookupProps { props, .. }
858 | PlanOp::JoinBound { props, .. }
859 | PlanOp::Expand {
860 to_props: props, ..
861 } => {
862 for (_, operand) in props {
863 collect_operand(operand, names, seen);
864 }
865 }
866 PlanOp::Filter { expr } => collect_expr(expr, names, seen, 0)?,
867 PlanOp::LeftOuterApply { inner, .. } => {
868 collect_params_from_ops(inner, names, seen)?;
869 }
870 PlanOp::Project { items } => {
873 for item in items {
874 collect_ret_item_params(item, names, seen);
875 }
876 }
877 PlanOp::With {
878 items, where_expr, ..
879 } => {
880 for item in items {
881 collect_ret_item_params(item, names, seen);
882 }
883 if let Some(expr) = where_expr {
884 let _ = collect_expr(expr, names, seen, 0);
885 }
886 }
887 PlanOp::GroupAggregate { keys, aggs } => {
888 for (_, item) in keys {
889 collect_ret_item_params(item, names, seen);
890 }
891 for (_, arg, _) in aggs {
892 match arg {
893 AggArg::Star => {}
894 AggArg::Var(_) => {}
895 AggArg::Prop { .. } => {}
896 }
897 }
898 }
899 PlanOp::Skip(LimitSkip::Param(n)) | PlanOp::Limit(LimitSkip::Param(n))
902 if seen.insert(n.clone()) =>
903 {
904 names.push(n.clone());
905 }
906 _ => {}
907 }
908 }
909 Ok(())
910}
911
912fn collect_ret_item_params(item: &RetItem, names: &mut Vec<String>, seen: &mut BTreeSet<String>) {
914 match &item.value {
915 RetVal::FuncCall { args, .. } => {
916 for arg in args {
917 collect_operand(arg, names, seen);
918 }
919 }
920 RetVal::ScalarExpr(op) => {
921 collect_operand(op, names, seen);
922 }
923 RetVal::Prop { .. } | RetVal::Var(_) | RetVal::Agg { .. } => {}
924 }
925}
926
927fn collect_operand(op: &Operand, names: &mut Vec<String>, seen: &mut BTreeSet<String>) {
928 match op {
929 Operand::Param(n) => {
930 if seen.insert(n.clone()) {
931 names.push(n.clone());
932 }
933 }
934 Operand::FuncCall { args, .. } => {
935 for arg in args {
936 collect_operand(arg, names, seen);
937 }
938 }
939 Operand::BinArith { left, right, .. } => {
940 collect_operand(left, names, seen);
941 collect_operand(right, names, seen);
942 }
943 _ => {}
944 }
945}
946
947fn collect_expr(
948 expr: &Expr,
949 names: &mut Vec<String>,
950 seen: &mut BTreeSet<String>,
951 depth: u32,
952) -> Result<(), String> {
953 if depth > 256 {
954 return Err("expression nesting too deep".into());
955 }
956 match expr {
957 Expr::And(lhs, rhs) | Expr::Or(lhs, rhs) => {
958 collect_expr(lhs, names, seen, depth + 1)?;
959 collect_expr(rhs, names, seen, depth + 1)
960 }
961 Expr::Not(inner) => collect_expr(inner, names, seen, depth + 1),
962 Expr::Cmp { lhs, rhs, .. } => {
963 collect_operand(lhs, names, seen);
964 collect_operand(rhs, names, seen);
965 Ok(())
966 }
967 Expr::Truthy(op) => {
968 collect_operand(op, names, seen);
969 Ok(())
970 }
971 Expr::IsNull(op) | Expr::IsNotNull(op) => {
972 collect_operand(op, names, seen);
973 Ok(())
974 }
975 Expr::In { expr, list } => {
976 collect_operand(expr, names, seen);
977 for item in list {
978 collect_operand(item, names, seen);
979 }
980 Ok(())
981 }
982 }
983}
984
985fn collect_vars(plan: &[PlanOp]) -> VarTable {
986 let mut vars = VarTable { names: Vec::new() };
987 for op in plan {
988 match op {
989 PlanOp::ScanLabel { var, .. } => {
990 vars.intern(var);
991 }
992 PlanOp::ScanKey { var, key, .. } => {
993 vars.intern(var);
994 intern_operand(&mut vars, key);
995 }
996 PlanOp::IndexScan { var, value, .. } => {
997 vars.intern(var);
998 intern_operand(&mut vars, value);
999 }
1000 PlanOp::IndexIntersect {
1001 var, equalities, ..
1002 } => {
1003 vars.intern(var);
1004 for (_, operand) in equalities {
1005 intern_operand(&mut vars, operand);
1006 }
1007 }
1008 PlanOp::LookupProps { var, props } | PlanOp::JoinBound { var, props, .. } => {
1009 vars.intern(var);
1010 for (_, operand) in props {
1011 intern_operand(&mut vars, operand);
1012 }
1013 }
1014 PlanOp::Expand {
1015 from,
1016 rel_var,
1017 to,
1018 to_props,
1019 ..
1020 } => {
1021 vars.intern(from);
1022 vars.intern(to);
1023 if let Some(r) = rel_var {
1024 vars.intern(r);
1025 }
1026 for (_, operand) in to_props {
1027 intern_operand(&mut vars, operand);
1028 }
1029 }
1030 PlanOp::VarExpand {
1031 from, rel_var, to, ..
1032 } => {
1033 vars.intern(from);
1034 vars.intern(to);
1035 if let Some(r) = rel_var {
1036 vars.intern(r);
1037 }
1038 }
1039 PlanOp::ShortestPath {
1040 from, rel_var, to, ..
1041 } => {
1042 vars.intern(from);
1043 vars.intern(to);
1044 if let Some(r) = rel_var {
1045 vars.intern(r);
1046 }
1047 }
1048 PlanOp::Filter { expr } => intern_expr(&mut vars, expr),
1049 PlanOp::Project { items } => {
1050 for item in items {
1051 match &item.value {
1052 RetVal::Var(name) | RetVal::Prop { var: name, .. } => {
1053 vars.intern(name);
1054 }
1055 RetVal::Agg { .. } => {} RetVal::FuncCall { args, .. } => {
1057 for arg in args {
1058 intern_operand(&mut vars, arg);
1059 }
1060 }
1061 RetVal::ScalarExpr(op) => {
1062 intern_operand(&mut vars, op);
1063 }
1064 }
1065 }
1066 }
1067 PlanOp::Aggregate { arg, .. } => match arg {
1068 AggArg::Star => {}
1069 AggArg::Var(v) => {
1070 vars.intern(v);
1071 }
1072 AggArg::Prop { var, .. } => {
1073 vars.intern(var);
1074 }
1075 },
1076 PlanOp::GroupAggregate { keys, aggs } => {
1077 for (col, item) in keys {
1079 match &item.value {
1080 RetVal::Var(name) | RetVal::Prop { var: name, .. } => {
1081 vars.intern(name);
1082 }
1083 RetVal::Agg { .. } => {}
1084 RetVal::FuncCall { args, .. } => {
1085 for arg in args {
1086 intern_operand(&mut vars, arg);
1087 }
1088 }
1089 RetVal::ScalarExpr(op) => {
1090 intern_operand(&mut vars, op);
1091 }
1092 }
1093 vars.intern(col);
1095 }
1096 for (_, arg, col) in aggs {
1097 match arg {
1098 AggArg::Star => {}
1099 AggArg::Var(v) => {
1100 vars.intern(v);
1101 }
1102 AggArg::Prop { var, .. } => {
1103 vars.intern(var);
1104 }
1105 }
1106 vars.intern(col);
1108 }
1109 }
1110 PlanOp::With {
1111 items,
1112 where_expr,
1113 order_by,
1114 ..
1115 } => {
1116 for item in items {
1117 match &item.value {
1118 RetVal::Var(name) | RetVal::Prop { var: name, .. } => {
1119 vars.intern(name);
1120 }
1121 RetVal::Agg { .. } => {}
1122 RetVal::FuncCall { args, .. } => {
1123 for arg in args {
1124 intern_operand(&mut vars, arg);
1125 }
1126 }
1127 RetVal::ScalarExpr(op) => {
1128 intern_operand(&mut vars, op);
1129 }
1130 }
1131 if let Some(alias) = &item.alias {
1133 vars.intern(alias);
1134 } else {
1135 match &item.value {
1136 RetVal::Var(name) => {
1137 vars.intern(name);
1138 }
1139 RetVal::Prop { var, field } => {
1140 let col = format!("{var}.{field}");
1141 vars.intern(&col);
1142 }
1143 RetVal::Agg { .. } => {}
1144 RetVal::ScalarExpr(_) => {
1145 vars.intern("<expr>");
1146 }
1147 RetVal::FuncCall { name, args } => {
1148 let arg_strs: Vec<String> = args
1150 .iter()
1151 .map(|a| match a {
1152 Operand::Var(v) => v.clone(),
1153 Operand::Prop { var, field } => format!("{var}.{field}"),
1154 Operand::Lit(_) => "<lit>".to_string(),
1155 Operand::Param(p) => format!("${p}"),
1156 Operand::FuncCall { name: n, .. } => format!("{n}(...)"),
1157 Operand::BinArith { .. } => "<arith>".to_string(),
1158 Operand::Case { .. } => "<case>".to_string(),
1159 })
1160 .collect();
1161 let col = format!("{name}({})", arg_strs.join(", "));
1162 vars.intern(&col);
1163 }
1164 }
1165 }
1166 }
1167 if let Some(expr) = where_expr {
1168 intern_expr(&mut vars, expr);
1169 }
1170 for oi in order_by {
1171 match &oi.target {
1172 OrderTarget::Alias(name) | OrderTarget::Var(name) => {
1173 vars.intern(name);
1174 }
1175 OrderTarget::Prop { var, .. } => {
1176 vars.intern(var);
1177 }
1178 }
1179 }
1180 }
1181 PlanOp::Unwind { expr, alias } => {
1182 vars.intern(alias);
1183 match expr {
1184 UnwindExpr::Prop { var, .. } => {
1185 vars.intern(var);
1186 }
1187 UnwindExpr::Var(name) => {
1188 vars.intern(name);
1189 }
1190 UnwindExpr::Lit(_) => {}
1191 }
1192 }
1193 PlanOp::LeftOuterApply {
1194 inner,
1195 optional_vars,
1196 } => {
1197 for op in inner {
1199 match op {
1203 PlanOp::ScanLabel { var, .. } => {
1204 vars.intern(var);
1205 }
1206 PlanOp::ScanKey { var, key, .. } => {
1207 vars.intern(var);
1208 intern_operand(&mut vars, key);
1209 }
1210 PlanOp::IndexScan { var, value, .. } => {
1211 vars.intern(var);
1212 intern_operand(&mut vars, value);
1213 }
1214 PlanOp::IndexIntersect {
1215 var, equalities, ..
1216 } => {
1217 vars.intern(var);
1218 for (_, operand) in equalities {
1219 intern_operand(&mut vars, operand);
1220 }
1221 }
1222 PlanOp::Expand {
1223 from, rel_var, to, ..
1224 } => {
1225 vars.intern(from);
1226 vars.intern(to);
1227 if let Some(r) = rel_var {
1228 vars.intern(r);
1229 }
1230 }
1231 PlanOp::JoinBound { var, .. } | PlanOp::LookupProps { var, .. } => {
1232 vars.intern(var);
1233 }
1234 PlanOp::Filter { expr } => intern_expr(&mut vars, expr),
1235 _ => {}
1236 }
1237 }
1238 for v in optional_vars {
1239 vars.intern(v);
1240 }
1241 }
1242 _ => {}
1243 }
1244 }
1245 vars
1246}
1247
1248fn intern_operand(vars: &mut VarTable, operand: &Operand) {
1249 match operand {
1250 Operand::Prop { var, .. } | Operand::Var(var) => {
1251 vars.intern(var);
1252 }
1253 Operand::Lit(_) | Operand::Param(_) => {}
1254 Operand::BinArith { left, right, .. } => {
1255 intern_operand(vars, left);
1256 intern_operand(vars, right);
1257 }
1258 Operand::FuncCall { args, .. } => {
1259 for arg in args {
1260 intern_operand(vars, arg);
1261 }
1262 }
1263 Operand::Case { branches, default } => {
1264 for (cond, value) in branches {
1265 intern_expr(vars, cond);
1266 intern_operand(vars, value);
1267 }
1268 if let Some(d) = default {
1269 intern_operand(vars, d);
1270 }
1271 }
1272 }
1273}
1274
1275fn intern_expr(vars: &mut VarTable, expr: &Expr) {
1276 match expr {
1277 Expr::And(lhs, rhs) | Expr::Or(lhs, rhs) => {
1278 intern_expr(vars, lhs);
1279 intern_expr(vars, rhs);
1280 }
1281 Expr::Not(inner) => intern_expr(vars, inner),
1282 Expr::Cmp { lhs, rhs, .. } => {
1283 intern_operand(vars, lhs);
1284 intern_operand(vars, rhs);
1285 }
1286 Expr::Truthy(op) => intern_operand(vars, op),
1287 Expr::IsNull(op) | Expr::IsNotNull(op) => intern_operand(vars, op),
1288 Expr::In { expr, list } => {
1289 intern_operand(vars, expr);
1290 for item in list {
1291 intern_operand(vars, item);
1292 }
1293 }
1294 }
1295}
1296
1297fn scan_ids(view: &GraphView, label: Option<&str>) -> Vec<u32> {
1298 let ids = match label {
1299 Some(label) => view.nodes_with_label(label),
1300 None => (0..view.ids.len() as u32)
1302 .filter(|&id| view.label_of(id).is_some())
1303 .collect(),
1304 };
1305 if view.mask.is_some() {
1306 ids.into_iter().filter(|&id| view.visible(id)).collect()
1307 } else {
1308 ids
1309 }
1310}
1311
1312fn resolve_scan_key_id(
1315 view: &GraphView,
1316 vars: &VarTable,
1317 row: &Row,
1318 key: &Operand,
1319 label: Option<&str>,
1320 params: &Params,
1321) -> Result<Option<u32>, String> {
1322 #[cfg(test)]
1323 SCAN_KEY_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1324
1325 let Some(val) = resolve_operand(view, vars, row, key, params)? else {
1326 return Ok(None);
1327 };
1328 let Value::Str(s) = val else {
1329 return Ok(None);
1330 };
1331 let Some(id) = view.node_id(&s) else {
1332 return Ok(None);
1333 };
1334 if !view.visible(id) {
1335 return Ok(None);
1336 }
1337 if let Some(want) = label {
1338 match view.label_of(id) {
1339 Some(got) if got == want => {}
1340 _ => return Ok(None),
1341 }
1342 }
1343 Ok(Some(id))
1344}
1345
1346fn scan_key(
1347 view: &GraphView,
1348 vars: &VarTable,
1349 rows: &[Row],
1350 var: &str,
1351 key: &Operand,
1352 label: Option<&str>,
1353 params: &Params,
1354) -> Result<Vec<Row>, String> {
1355 let slot = vars
1356 .slot(var)
1357 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1358 let cap = max_intermediate_rows();
1359 let mut out = Vec::new();
1360 for row in rows {
1361 let Some(id) = resolve_scan_key_id(view, vars, row, key, label, params)? else {
1362 continue;
1363 };
1364 if out.len() >= cap {
1365 return Err(row_cap_err(cap));
1366 }
1367 let mut next = row.clone();
1368 next[slot] = Some(Cell::Node(id));
1369 out.push(next);
1370 }
1371 Ok(out)
1372}
1373
1374fn scan_label(
1375 view: &GraphView,
1376 vars: &VarTable,
1377 rows: &[Row],
1378 var: &str,
1379 label: Option<&str>,
1380) -> Result<Vec<Row>, String> {
1381 let ids = scan_ids(view, label);
1382 let slot = vars
1383 .slot(var)
1384 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1385 let cap = max_intermediate_rows();
1386 let mut out = Vec::with_capacity(rows.len().saturating_mul(ids.len()).min(cap));
1387 for row in rows {
1388 for &id in &ids {
1389 if out.len() >= cap {
1390 return Err(row_cap_err(cap));
1391 }
1392 let mut next = row.clone();
1393 next[slot] = Some(Cell::Node(id));
1394 out.push(next);
1395 }
1396 }
1397 Ok(out)
1398}
1399
1400#[allow(clippy::too_many_arguments)]
1408fn index_scan_ids(
1409 view: &GraphView,
1410 vars: &VarTable,
1411 row: &Row,
1412 label: Option<&str>,
1413 field: &str,
1414 value: &Operand,
1415 params: &Params,
1416) -> Result<Vec<u32>, String> {
1417 let resolved = resolve_operand(view, vars, row, value, params)?;
1418 if let (Some(label_str), Some(val)) = (label, resolved.as_ref()) {
1419 if let Some(ids) = view.nodes_with_prop(label_str, field, val) {
1420 #[cfg(test)]
1421 INDEX_SCAN_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1422 return Ok(ids);
1423 }
1424 }
1425 let props = [(field.to_string(), value.clone())];
1427 let mut out = Vec::new();
1428 for id in scan_ids(view, label) {
1429 if node_matches(view, vars, row, id, None, &props, params)? {
1430 out.push(id);
1431 }
1432 }
1433 Ok(out)
1434}
1435
1436#[allow(clippy::too_many_arguments)]
1439fn scan_index(
1440 view: &GraphView,
1441 vars: &VarTable,
1442 rows: &[Row],
1443 var: &str,
1444 label: Option<&str>,
1445 field: &str,
1446 value: &Operand,
1447 params: &Params,
1448) -> Result<Vec<Row>, String> {
1449 let Some(first) = rows.first() else {
1450 return Ok(Vec::new());
1451 };
1452 let ids = index_scan_ids(view, vars, first, label, field, value, params)?;
1453 let slot = vars
1454 .slot(var)
1455 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1456 let cap = max_intermediate_rows();
1457 let mut out = Vec::new();
1458 for row in rows {
1459 for &id in &ids {
1460 if out.len() >= cap {
1461 return Err(row_cap_err(cap));
1462 }
1463 let mut next = row.clone();
1464 next[slot] = Some(Cell::Node(id));
1465 out.push(next);
1466 }
1467 }
1468 Ok(out)
1469}
1470
1471#[allow(clippy::too_many_arguments)]
1480fn index_intersect_ids(
1481 view: &GraphView,
1482 vars: &VarTable,
1483 row: &Row,
1484 label: Option<&str>,
1485 equalities: &[(String, Operand)],
1486 params: &Params,
1487) -> Result<Vec<u32>, String> {
1488 let mut resolved: Vec<(String, Option<Value>)> = Vec::with_capacity(equalities.len());
1490 for (field, operand) in equalities {
1491 let val = resolve_operand(view, vars, row, operand, params)?;
1492 resolved.push((field.clone(), val));
1493 }
1494
1495 let mut indexed_lists: Vec<Vec<u32>> = Vec::new();
1497 let mut unindexed_props: Vec<(String, Operand)> = Vec::new();
1498
1499 for ((field, val_opt), (_, operand)) in resolved.iter().zip(equalities.iter()) {
1500 if let (Some(label_str), Some(val)) = (label, val_opt.as_ref()) {
1501 if let Some(ids) = view.nodes_with_prop(label_str, field, val) {
1502 indexed_lists.push(ids);
1503 continue;
1504 }
1505 }
1506 unindexed_props.push((field.clone(), operand.clone()));
1507 }
1508
1509 if indexed_lists.is_empty() {
1510 let mut out = Vec::new();
1512 for id in scan_ids(view, label) {
1513 if node_matches(view, vars, row, id, None, equalities, params)? {
1514 out.push(id);
1515 }
1516 }
1517 return Ok(out);
1518 }
1519
1520 #[cfg(test)]
1521 INDEX_INTERSECT_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1522
1523 indexed_lists.sort_unstable_by_key(|v| v.len());
1525 let mut result = indexed_lists.remove(0);
1526 for other in indexed_lists {
1527 let mut merged = Vec::new();
1528 let (mut i, mut j) = (0, 0);
1529 while i < result.len() && j < other.len() {
1530 match result[i].cmp(&other[j]) {
1531 std::cmp::Ordering::Equal => {
1532 merged.push(result[i]);
1533 i += 1;
1534 j += 1;
1535 }
1536 std::cmp::Ordering::Less => i += 1,
1537 std::cmp::Ordering::Greater => j += 1,
1538 }
1539 }
1540 result = merged;
1541 }
1542
1543 if !unindexed_props.is_empty() {
1547 let mut filtered = Vec::with_capacity(result.len());
1548 for id in result {
1549 if node_matches(view, vars, row, id, None, &unindexed_props, params)? {
1550 filtered.push(id);
1551 }
1552 }
1553 result = filtered;
1554 }
1555
1556 Ok(result)
1557}
1558
1559#[allow(clippy::too_many_arguments)]
1562fn scan_intersect(
1563 view: &GraphView,
1564 vars: &VarTable,
1565 rows: &[Row],
1566 var: &str,
1567 label: Option<&str>,
1568 equalities: &[(String, Operand)],
1569 params: &Params,
1570) -> Result<Vec<Row>, String> {
1571 let Some(first) = rows.first() else {
1572 return Ok(Vec::new());
1573 };
1574 let ids = index_intersect_ids(view, vars, first, label, equalities, params)?;
1575 let slot = vars
1576 .slot(var)
1577 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1578 let cap = max_intermediate_rows();
1579 let mut out = Vec::new();
1580 for row in rows {
1581 for &id in &ids {
1582 if out.len() >= cap {
1583 return Err(row_cap_err(cap));
1584 }
1585 let mut next = row.clone();
1586 next[slot] = Some(Cell::Node(id));
1587 out.push(next);
1588 }
1589 }
1590 Ok(out)
1591}
1592
1593fn require_cell<'a>(row: &'a Row, vars: &VarTable, var: &str) -> Result<&'a Cell, String> {
1594 let slot = vars
1595 .slot(var)
1596 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1597 row.get(slot)
1598 .and_then(|c| c.as_ref())
1599 .ok_or_else(|| format!("unbound variable `{var}`"))
1600}
1601
1602fn require_node(row: &Row, vars: &VarTable, var: &str) -> Result<u32, String> {
1603 match require_cell(row, vars, var)? {
1604 Cell::Node(id) => Ok(*id),
1605 Cell::Rel(_) => Err(format!("variable `{var}` is not a node")),
1606 Cell::Path(_) => Err(format!("variable `{var}` is a path, not a node")),
1607 Cell::Scalar(_) => Err(format!("variable `{var}` is a scalar value, not a node")),
1608 }
1609}
1610
1611const SCALAR_FUNCS: &[&str] = &[
1613 "toLower",
1614 "toUpper",
1615 "size",
1616 "coalesce",
1617 "type",
1618 "abs",
1619 "round",
1620 "textMatches",
1621 "contains",
1622 "startsWith",
1623 "endsWith",
1624 "toInteger",
1625 "toFloat",
1626 "toString",
1627 "decay",
1628];
1629
1630fn eval_string_predicate(
1633 name: &str,
1634 args: &[Operand],
1635 view: &GraphView,
1636 vars: &VarTable,
1637 row: &Row,
1638 params: &Params,
1639 f: impl Fn(&str, &str) -> bool,
1640) -> Result<Option<Value>, String> {
1641 if args.len() != 2 {
1642 return Err(format!(
1643 "{name}() requires exactly 2 arguments, got {}",
1644 args.len()
1645 ));
1646 }
1647 let a = resolve_operand(view, vars, row, &args[0], params)?;
1648 let b = resolve_operand(view, vars, row, &args[1], params)?;
1649 match (a, b) {
1650 (Some(Value::Str(s)), Some(Value::Str(sub))) => Ok(Some(Value::Bool(f(&s, &sub)))),
1651 (None, _) | (_, None) => Ok(None),
1652 _ => Ok(None),
1653 }
1654}
1655
1656fn eval_func(
1660 name: &str,
1661 args: &[Operand],
1662 view: &GraphView,
1663 vars: &VarTable,
1664 row: &Row,
1665 params: &Params,
1666) -> Result<Option<Value>, String> {
1667 let norm = name.to_ascii_lowercase();
1668 match norm.as_str() {
1669 "tolower" => {
1670 if args.len() != 1 {
1671 return Err(format!(
1672 "toLower() requires exactly 1 argument, got {}",
1673 args.len()
1674 ));
1675 }
1676 let v = resolve_operand(view, vars, row, &args[0], params)?;
1677 Ok(v.map(|val| match val {
1678 Value::Str(s) => Value::Str(s.to_ascii_lowercase()),
1679 other => other, }))
1681 }
1682 "toupper" => {
1683 if args.len() != 1 {
1684 return Err(format!(
1685 "toUpper() requires exactly 1 argument, got {}",
1686 args.len()
1687 ));
1688 }
1689 let v = resolve_operand(view, vars, row, &args[0], params)?;
1690 Ok(v.map(|val| match val {
1691 Value::Str(s) => Value::Str(s.to_ascii_uppercase()),
1692 other => other,
1693 }))
1694 }
1695 "size" => {
1696 if args.len() != 1 {
1697 return Err(format!(
1698 "size() requires exactly 1 argument, got {}",
1699 args.len()
1700 ));
1701 }
1702 let v = resolve_operand(view, vars, row, &args[0], params)?;
1703 match v {
1704 None => Ok(None), Some(Value::Str(s)) => Ok(Some(Value::Int(s.len() as i64))),
1706 Some(Value::List(items)) => Ok(Some(Value::Int(items.len() as i64))),
1707 Some(_) => Ok(None), }
1709 }
1710 "coalesce" => {
1711 for arg in args {
1713 if let Some(v) = resolve_operand(view, vars, row, arg, params)? {
1714 return Ok(Some(v));
1715 }
1716 }
1717 Ok(None)
1718 }
1719 "type" => {
1720 if args.len() != 1 {
1721 return Err(format!(
1722 "type() requires exactly 1 argument, got {}",
1723 args.len()
1724 ));
1725 }
1726 let Operand::Var(var_name) = &args[0] else {
1728 return Err(
1729 "type() argument must be a relationship variable (e.g. type(r))".to_string(),
1730 );
1731 };
1732 let slot = vars
1733 .slot(var_name)
1734 .ok_or_else(|| format!("unbound variable `{var_name}` in type()"))?;
1735 match row.get(slot).and_then(|c| c.as_ref()) {
1736 Some(Cell::Rel(e)) => {
1737 let etype = view.syms.resolve(e.etype).unwrap_or("").to_owned();
1739 Ok(Some(Value::Str(etype)))
1740 }
1741 Some(Cell::Node(_)) => Err(format!(
1742 "type() argument `{var_name}` is a node, not a relationship"
1743 )),
1744 Some(Cell::Scalar(_) | Cell::Path(_)) => Err(format!(
1745 "type() argument `{var_name}` is not a relationship"
1746 )),
1747 None => Ok(None), }
1749 }
1750 "abs" => {
1751 if args.len() != 1 {
1752 return Err(format!(
1753 "abs() requires exactly 1 argument, got {}",
1754 args.len()
1755 ));
1756 }
1757 let v = resolve_operand(view, vars, row, &args[0], params)?;
1758 match v {
1759 None => Ok(None),
1760 Some(Value::Int(n)) => Ok(Some(Value::Int(n.abs()))),
1761 Some(Value::Float(f)) => Ok(Some(Value::Float(f.abs()))),
1762 Some(_) => Ok(None), }
1764 }
1765 "round" => {
1766 if args.len() != 1 {
1767 return Err(format!(
1768 "round() requires exactly 1 argument, got {}",
1769 args.len()
1770 ));
1771 }
1772 let v = resolve_operand(view, vars, row, &args[0], params)?;
1773 match v {
1774 None => Ok(None),
1775 Some(Value::Int(n)) => Ok(Some(Value::Int(n))), Some(Value::Float(f)) => Ok(Some(Value::Float(f.round()))),
1777 Some(_) => Ok(None), }
1779 }
1780 "textmatches" => {
1781 if args.len() != 2 {
1793 return Err(format!(
1794 "textMatches() requires exactly 2 arguments (field_value, query), got {}",
1795 args.len()
1796 ));
1797 }
1798 let field_val = resolve_operand(view, vars, row, &args[0], params)?;
1799 let query_val = resolve_operand(view, vars, row, &args[1], params)?;
1800 match (field_val, query_val) {
1801 (None, _) | (_, None) => Ok(None),
1802 (Some(Value::Str(s)), Some(Value::Str(q))) => {
1803 Ok(Some(Value::Bool(core_storage::fulltext::eval_query_str(
1805 &s, &q,
1806 ))))
1807 }
1808 (Some(Value::List(items)), Some(Value::Str(q))) => Ok(Some(Value::Bool(
1809 core_storage::fulltext::eval_query_str_list(&items, &q),
1810 ))),
1811 _ => Ok(Some(Value::Bool(false))), }
1813 }
1814 "contains" => eval_string_predicate("contains", args, view, vars, row, params, |s, sub| {
1815 s.contains(sub)
1816 }),
1817 "startswith" => {
1818 eval_string_predicate("startsWith", args, view, vars, row, params, |s, p| {
1819 s.starts_with(p)
1820 })
1821 }
1822 "endswith" => eval_string_predicate("endsWith", args, view, vars, row, params, |s, p| {
1823 s.ends_with(p)
1824 }),
1825 "tointeger" => {
1826 if args.len() != 1 {
1827 return Err(format!(
1828 "toInteger() requires exactly 1 argument, got {}",
1829 args.len()
1830 ));
1831 }
1832 let v = resolve_operand(view, vars, row, &args[0], params)?;
1833 Ok(match v {
1834 None => None,
1835 Some(Value::Int(n)) => Some(Value::Int(n)),
1836 Some(Value::Float(f)) => Some(Value::Int(f.trunc() as i64)),
1837 Some(Value::Str(s)) => s
1840 .trim()
1841 .parse::<i64>()
1842 .ok()
1843 .or_else(|| s.trim().parse::<f64>().ok().map(|f| f.trunc() as i64))
1844 .map(Value::Int),
1845 Some(_) => None,
1846 })
1847 }
1848 "tofloat" => {
1849 if args.len() != 1 {
1850 return Err(format!(
1851 "toFloat() requires exactly 1 argument, got {}",
1852 args.len()
1853 ));
1854 }
1855 let v = resolve_operand(view, vars, row, &args[0], params)?;
1856 Ok(match v {
1857 None => None,
1858 Some(Value::Float(f)) => Some(Value::Float(f)),
1859 Some(Value::Int(n)) => Some(Value::Float(n as f64)),
1860 Some(Value::Str(s)) => s.trim().parse::<f64>().ok().map(Value::Float),
1861 Some(_) => None,
1862 })
1863 }
1864 "tostring" => {
1865 if args.len() != 1 {
1866 return Err(format!(
1867 "toString() requires exactly 1 argument, got {}",
1868 args.len()
1869 ));
1870 }
1871 let v = resolve_operand(view, vars, row, &args[0], params)?;
1872 Ok(match v {
1873 None => None,
1874 Some(Value::Str(s)) => Some(Value::Str(s)),
1875 Some(Value::Int(n)) => Some(Value::Str(n.to_string())),
1876 Some(Value::Float(f)) => Some(Value::Str(f.to_string())),
1877 Some(Value::Bool(b)) => Some(Value::Str(b.to_string())),
1878 Some(_) => None, })
1880 }
1881 "decay" => {
1882 if args.len() != 3 {
1887 return Err(format!(
1888 "decay() requires exactly 3 arguments, got {}",
1889 args.len()
1890 ));
1891 }
1892 let base = resolve_operand(view, vars, row, &args[0], params)?;
1893 let age = resolve_operand(view, vars, row, &args[1], params)?;
1894 let halflife = resolve_operand(view, vars, row, &args[2], params)?;
1895 match (base, age, halflife) {
1896 (None, _, _) | (_, None, _) | (_, _, None) => Ok(None),
1897 (Some(b), Some(a), Some(h)) => {
1898 let b = numeric_val(&b)
1899 .ok_or_else(|| "decay() requires numeric arguments".to_string())?;
1900 let a = numeric_val(&a)
1901 .ok_or_else(|| "decay() requires numeric arguments".to_string())?;
1902 let h = numeric_val(&h)
1903 .ok_or_else(|| "decay() requires numeric arguments".to_string())?;
1904 if h <= 0.0 {
1905 return Err("decay() requires halflife > 0".to_string());
1906 }
1907 Ok(Some(Value::Float(b * 0.5f64.powf(a / h))))
1908 }
1909 }
1910 }
1911 _ => Err(format!(
1912 "unknown function `{name}`; supported: {}",
1913 SCALAR_FUNCS.join(", ")
1914 )),
1915 }
1916}
1917
1918fn resolve_operand(
1919 view: &GraphView,
1920 vars: &VarTable,
1921 row: &Row,
1922 operand: &Operand,
1923 params: &Params,
1924) -> Result<Option<Value>, String> {
1925 match operand {
1926 Operand::Lit(v) => Ok(Some(v.clone())),
1927 Operand::Param(name) => match params.0.get(name) {
1928 Some(v) => Ok(Some(v.clone())),
1929 None => Err(format!("missing parameter `{name}`")),
1930 },
1931 Operand::Prop { var, field } => resolve_prop(view, vars, row, var, field),
1932 Operand::Var(name) => {
1933 match vars
1935 .slot(name)
1936 .and_then(|s| row.get(s))
1937 .and_then(|c| c.as_ref())
1938 {
1939 Some(Cell::Scalar(v)) => Ok(Some(v.clone())),
1940 Some(Cell::Node(id)) => match view.ids.key_of(*id) {
1941 Some(key) => Ok(Some(Value::Str(key.to_owned()))),
1942 None => Ok(None),
1943 },
1944 Some(Cell::Path(hops)) => Ok(Some(Value::Int(*hops as i64))),
1945 Some(Cell::Rel(_)) => Err(format!("variable `{name}` is a relationship")),
1946 None => Ok(None),
1947 }
1948 }
1949 Operand::FuncCall { name, args } => eval_func(name, args, view, vars, row, params),
1950 Operand::Case { branches, default } => {
1951 for (cond, value) in branches {
1952 if eval_expr(view, vars, row, cond, params, 0)? {
1953 return resolve_operand(view, vars, row, value, params);
1954 }
1955 }
1956 match default {
1957 Some(d) => resolve_operand(view, vars, row, d, params),
1958 None => Ok(None),
1959 }
1960 }
1961 Operand::BinArith { op, left, right } => {
1962 use super::ast::ArithOp;
1963 let lv = resolve_operand(view, vars, row, left, params)?;
1964 let rv = resolve_operand(view, vars, row, right, params)?;
1965 match (lv, rv) {
1966 (None, _) | (_, None) => Ok(None), (Some(Value::Int(a)), Some(Value::Int(b))) => {
1968 let result = match op {
1969 ArithOp::Sub => a.saturating_sub(b),
1970 ArithOp::Mul => a.saturating_mul(b),
1971 ArithOp::Add => a.saturating_add(b),
1972 ArithOp::Div => {
1973 if b == 0 {
1974 return Err("division by zero".into());
1975 }
1976 a.checked_div(b).unwrap_or(i64::MAX)
1977 }
1978 };
1979 Ok(Some(Value::Int(result)))
1980 }
1981 (Some(lv), Some(rv)) => {
1982 let a = match &lv {
1983 Value::Float(f) => *f,
1984 Value::Int(i) => *i as f64,
1985 _ => return Err(format!("arithmetic operand must be numeric, got {lv:?}")),
1986 };
1987 let b = match &rv {
1988 Value::Float(f) => *f,
1989 Value::Int(i) => *i as f64,
1990 _ => return Err(format!("arithmetic operand must be numeric, got {rv:?}")),
1991 };
1992 let result = match op {
1993 ArithOp::Sub => a - b,
1994 ArithOp::Mul => a * b,
1995 ArithOp::Add => a + b,
1996 ArithOp::Div => {
1997 if b == 0.0 {
1998 return Err("division by zero".into());
1999 }
2000 a / b
2001 }
2002 };
2003 Ok(Some(Value::Float(result)))
2004 }
2005 }
2006 }
2007 }
2008}
2009
2010fn resolve_prop(
2011 view: &GraphView,
2012 vars: &VarTable,
2013 row: &Row,
2014 var: &str,
2015 field: &str,
2016) -> Result<Option<Value>, String> {
2017 let slot = vars
2020 .slot(var)
2021 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2022 let cell = match row.get(slot).and_then(|c| c.as_ref()) {
2023 Some(c) => c,
2024 None => return Ok(None),
2026 };
2027 match cell {
2028 Cell::Node(id) => Ok(view.prop(*id, field).map(|vr| vr.into_value())),
2029 Cell::Rel(e) => Ok(view.edge_props.get(e.etype, e.src, e.dst, field)),
2030 Cell::Path(hops) => {
2032 if field == "length" {
2033 Ok(Some(Value::Int(*hops as i64)))
2034 } else {
2035 Ok(None)
2036 }
2037 }
2038 Cell::Scalar(_) => Ok(None),
2040 }
2041}
2042
2043fn node_matches(
2044 view: &GraphView,
2045 vars: &VarTable,
2046 row: &Row,
2047 id: u32,
2048 label: Option<&str>,
2049 props: &[(String, Operand)],
2050 params: &Params,
2051) -> Result<bool, String> {
2052 if let Some(want) = label {
2053 match view.label_of(id) {
2054 Some(got) if got == want => {}
2055 _ => return Ok(false),
2056 }
2057 }
2058 for (field, operand) in props {
2059 let Some(expected) = resolve_operand(view, vars, row, operand, params)? else {
2060 return Ok(false);
2061 };
2062 match view.prop(id, field) {
2063 Some(got) if values_equal(got.as_value(), &expected) => {}
2064 _ => return Ok(false),
2065 }
2066 }
2067 Ok(true)
2068}
2069
2070fn retain_node(
2071 view: &GraphView,
2072 vars: &VarTable,
2073 rows: &[Row],
2074 var: &str,
2075 label: Option<&str>,
2076 props: &[(String, Operand)],
2077 params: &Params,
2078) -> Result<Vec<Row>, String> {
2079 let mut out = Vec::with_capacity(rows.len());
2080 for row in rows {
2081 let id = require_node(row, vars, var)?;
2082 if node_matches(view, vars, row, id, label, props, params)? {
2083 out.push(row.clone());
2084 }
2085 }
2086 Ok(out)
2087}
2088
2089fn map_dir(dir: RelDir) -> Dir {
2090 match dir {
2091 RelDir::Right => Dir::Out,
2092 RelDir::Left => Dir::In,
2093 RelDir::Undirected => Dir::Both,
2094 }
2095}
2096
2097fn neighbor(from: u32, e: &EdgeRef, dir: RelDir) -> u32 {
2098 match dir {
2099 RelDir::Right => e.dst,
2100 RelDir::Left => e.src,
2101 RelDir::Undirected => {
2102 if e.src == from {
2103 e.dst
2104 } else {
2105 e.src
2106 }
2107 }
2108 }
2109}
2110
2111fn row_has_edge(row: &Row, e: &EdgeRef) -> bool {
2112 row.iter()
2113 .any(|c| matches!(c, Some(Cell::Rel(existing)) if existing == e))
2114}
2115
2116fn resolve_etypes(view: &GraphView, etypes: &[String]) -> Option<Vec<u32>> {
2117 if etypes.is_empty() {
2118 None } else {
2120 Some(etypes.iter().filter_map(|n| view.syms.get(n)).collect())
2123 }
2124}
2125
2126fn exec_expand(
2127 view: &GraphView,
2128 vars: &VarTable,
2129 rows: &[Row],
2130 op: &PlanOp,
2131 params: &Params,
2132) -> Result<Vec<Row>, String> {
2133 let PlanOp::Expand {
2134 from,
2135 rel_var,
2136 etypes,
2137 dir,
2138 to,
2139 to_label,
2140 to_props,
2141 } = op
2142 else {
2143 return Err("internal: expected Expand".into());
2144 };
2145 let etypes = resolve_etypes(view, etypes);
2146 let exp_dir = map_dir(*dir);
2147 let to_slot = vars
2148 .slot(to)
2149 .ok_or_else(|| format!("unbound variable `{to}`"))?;
2150 let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
2151 let cap = max_intermediate_rows();
2152 let mut out = Vec::with_capacity(rows.len().saturating_mul(2).min(cap));
2153 for row in rows {
2154 let from_id = require_node(row, vars, from)?;
2155 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
2156 Some(Cell::Node(id)) => Some(*id),
2157 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
2158 return Err(format!("variable `{to}` is not a node"))
2159 }
2160 None => None,
2161 };
2162 for e in expand(view, from_id, etypes.as_deref(), exp_dir) {
2163 if row_has_edge(row, &e) {
2164 continue;
2165 }
2166 let nbr = neighbor(from_id, &e, *dir);
2167 if !view.visible(nbr) {
2168 continue;
2169 }
2170 if let Some(want) = bound_to {
2171 if nbr != want {
2172 continue;
2173 }
2174 }
2175 if !node_matches(view, vars, row, nbr, to_label.as_deref(), to_props, params)? {
2176 continue;
2177 }
2178 if out.len() >= cap {
2179 return Err(row_cap_err(cap));
2180 }
2181 let mut next = row.clone();
2182 if let Some(slot) = rel_slot {
2183 next[slot] = Some(Cell::Rel(e));
2184 }
2185 if bound_to.is_none() {
2186 next[to_slot] = Some(Cell::Node(nbr));
2187 }
2188 out.push(next);
2189 #[cfg(test)]
2190 record_expand_row();
2191 }
2192 }
2193 Ok(out)
2194}
2195
2196#[allow(clippy::too_many_arguments)]
2203fn exec_var_expand(
2204 view: &GraphView,
2205 vars: &VarTable,
2206 rows: &[Row],
2207 from: &str,
2208 rel_var: &Option<String>,
2209 etypes: &[String],
2210 dir: RelDir,
2211 to: &str,
2212 min: u8,
2213 max: u8,
2214) -> Result<Vec<Row>, String> {
2215 let etypes = resolve_etypes(view, etypes);
2216 let exp_dir = map_dir(dir);
2217 let to_slot = vars
2218 .slot(to)
2219 .ok_or_else(|| format!("unbound variable `{to}`"))?;
2220 let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
2221 let cap = max_intermediate_rows();
2222 let mut out: Vec<Row> = Vec::new();
2223
2224 for row in rows {
2225 let from_id = require_node(row, vars, from)?;
2226 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
2227 Some(Cell::Node(id)) => Some(*id),
2228 Some(_) => return Err(format!("variable `{to}` is not a node")),
2229 None => None,
2230 };
2231
2232 struct PathState {
2235 node: u32,
2236 edges: Vec<EdgeRef>,
2237 }
2238
2239 let mut frontier: Vec<PathState> = vec![PathState {
2240 node: from_id,
2241 edges: Vec::new(),
2242 }];
2243
2244 let mut frontier_count: usize = 0;
2249
2250 for depth in 1u8..=max {
2251 let mut next_frontier: Vec<PathState> = Vec::new();
2252 for state in &frontier {
2253 for e in expand(view, state.node, etypes.as_deref(), exp_dir) {
2254 if state.edges.contains(&e) {
2256 continue;
2257 }
2258 let nbr = neighbor(state.node, &e, dir);
2259 if !view.visible(nbr) {
2262 continue;
2263 }
2264
2265 if depth >= min {
2268 let dest_matches = match bound_to {
2269 Some(want) => nbr == want,
2270 None => true,
2271 };
2272 if dest_matches {
2273 if out.len() >= cap {
2274 return Err(row_cap_err(cap));
2275 }
2276 let mut next = row.clone();
2277 if let Some(slot) = rel_slot {
2278 next[slot] = Some(Cell::Path(depth));
2279 }
2280 next[to_slot] = Some(Cell::Node(nbr));
2281 out.push(next);
2282 }
2283 }
2284
2285 if depth < max {
2289 frontier_count += 1;
2290 if frontier_count >= cap {
2291 return Err(row_cap_err(cap));
2292 }
2293 let mut new_edges = state.edges.clone();
2294 new_edges.push(e);
2295 next_frontier.push(PathState {
2296 node: nbr,
2297 edges: new_edges,
2298 });
2299 }
2300 }
2301 }
2302 frontier = next_frontier;
2303 if frontier.is_empty() {
2304 break;
2305 }
2306 }
2307 }
2308 Ok(out)
2309}
2310
2311#[allow(clippy::too_many_arguments)]
2317fn exec_shortest_path(
2318 view: &GraphView,
2319 vars: &VarTable,
2320 rows: &[Row],
2321 from: &str,
2322 rel_var: &Option<String>,
2323 etypes: &[String],
2324 dir: RelDir,
2325 to: &str,
2326 max_hops: u8,
2327) -> Result<Vec<Row>, String> {
2328 let etypes = resolve_etypes(view, etypes);
2329 let exp_dir = map_dir(dir);
2330 let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
2331 let mut out: Vec<Row> = Vec::new();
2332
2333 for row in rows {
2334 let from_id = require_node(row, vars, from)?;
2335 let to_id = require_node(row, vars, to)?;
2336
2337 let mut visited = std::collections::BTreeSet::new();
2339 visited.insert(from_id);
2340 let mut frontier: Vec<u32> = vec![from_id];
2341
2342 'bfs: for depth in 1u8..=max_hops {
2343 let mut next_frontier: Vec<u32> = Vec::new();
2344 for &node in &frontier {
2345 for e in expand(view, node, etypes.as_deref(), exp_dir) {
2346 let nbr = neighbor(node, &e, dir);
2347 if !view.visible(nbr) {
2351 continue;
2352 }
2353 if nbr == to_id {
2354 let mut next = row.clone();
2356 if let Some(slot) = rel_slot {
2357 next[slot] = Some(Cell::Path(depth));
2358 }
2359 out.push(next);
2360 break 'bfs;
2361 }
2362 if !visited.contains(&nbr) {
2363 visited.insert(nbr);
2364 next_frontier.push(nbr);
2365 }
2366 }
2367 }
2368 frontier = next_frontier;
2369 if frontier.is_empty() {
2370 break;
2371 }
2372 }
2373 }
2374 Ok(out)
2375}
2376
2377fn exec_filter(
2378 view: &GraphView,
2379 vars: &VarTable,
2380 rows: &[Row],
2381 expr: &Expr,
2382 params: &Params,
2383) -> Result<Vec<Row>, String> {
2384 let mut out = Vec::with_capacity(rows.len());
2385 for row in rows {
2386 if eval_expr(view, vars, row, expr, params, 0)? {
2387 out.push(row.clone());
2388 }
2389 }
2390 Ok(out)
2391}
2392
2393struct PullCtx<'a> {
2409 view: &'a GraphView<'a>,
2410 vars: &'a VarTable,
2411 project_items: &'a [RetItem],
2412 params: &'a Params<'a>,
2413 bound: usize,
2414}
2415
2416fn execute_pull(
2417 view: &GraphView,
2418 plan: &[PlanOp],
2419 params: &Params,
2420 bound: usize,
2421) -> Result<ResultSet, String> {
2422 let proj_pos = match plan
2423 .iter()
2424 .position(|op| matches!(op, PlanOp::Project { .. }))
2425 {
2426 Some(p) => p,
2427 None => return Ok(ResultSet::new(vec![])),
2428 };
2429 let producers = &plan[..proj_pos];
2430 let project_items = match &plan[proj_pos] {
2431 PlanOp::Project { items } => items,
2432 _ => unreachable!(),
2433 };
2434 let columns: Vec<String> = project_items.iter().map(column_name).collect();
2435 let vars = collect_vars(plan);
2436 let ctx = PullCtx {
2437 view,
2438 vars: &vars,
2439 project_items,
2440 params,
2441 bound,
2442 };
2443 let mut initial_row: Row = vec![None; vars.names.len()];
2444 let mut result_rows: Vec<Vec<Option<Value>>> = Vec::with_capacity(bound);
2445 pull_rows(&ctx, producers, &mut initial_row, &mut result_rows)?;
2446 let skip_n = plan[proj_pos + 1..]
2449 .iter()
2450 .find_map(|op| match op {
2451 PlanOp::Skip(ls) => Some(resolve_ls(ls, params)),
2452 _ => None,
2453 })
2454 .transpose()?
2455 .unwrap_or(0);
2456 let skip_n = usize::try_from(skip_n).unwrap_or(usize::MAX);
2457 let mut rs = ResultSet::new(columns);
2458 for row in result_rows.into_iter().skip(skip_n) {
2459 rs.push_row(row);
2460 }
2461 Ok(rs)
2462}
2463
2464enum AggAcc {
2478 Count(u64),
2479 Sum { val: f64, has_value: bool },
2480 Avg { sum: f64, n: u64 },
2481 Min(Option<Value>),
2482 Max(Option<Value>),
2483 Collect(Vec<Value>),
2484}
2485
2486impl AggAcc {
2487 fn new(func: &AggFunc) -> Self {
2488 match func {
2489 AggFunc::Count => AggAcc::Count(0),
2490 AggFunc::Sum => AggAcc::Sum {
2491 val: 0.0,
2492 has_value: false,
2493 },
2494 AggFunc::Avg => AggAcc::Avg { sum: 0.0, n: 0 },
2495 AggFunc::Min => AggAcc::Min(None),
2496 AggFunc::Max => AggAcc::Max(None),
2497 AggFunc::Collect => AggAcc::Collect(Vec::new()),
2498 }
2499 }
2500
2501 fn finish(self) -> Option<Value> {
2502 match self {
2503 AggAcc::Count(n) => Some(Value::Int(i64::try_from(n).unwrap_or(i64::MAX))),
2508 AggAcc::Sum { val, has_value } => {
2509 if has_value {
2510 Some(Value::Float(val))
2511 } else {
2512 None
2513 }
2514 }
2515 AggAcc::Avg { sum, n } => {
2516 if n > 0 {
2517 Some(Value::Float(sum / n as f64))
2518 } else {
2519 None
2520 }
2521 }
2522 AggAcc::Min(v) => v,
2523 AggAcc::Max(v) => v,
2524 AggAcc::Collect(items) => Some(Value::List(items)),
2526 }
2527 }
2528}
2529
2530fn numeric_val(v: &Value) -> Option<f64> {
2533 match v {
2534 Value::Int(n) => Some(*n as f64),
2535 Value::Float(f) => Some(*f),
2536 _ => None,
2537 }
2538}
2539
2540struct AggStreamCtx<'a> {
2542 view: &'a GraphView<'a>,
2543 vars: &'a VarTable,
2544 params: &'a Params<'a>,
2545 func: &'a AggFunc,
2546 arg: &'a AggArg,
2547}
2548
2549fn execute_aggregate(
2554 view: &GraphView,
2555 plan: &[PlanOp],
2556 params: &Params,
2557) -> Result<ResultSet, String> {
2558 let agg_pos = match plan
2559 .iter()
2560 .position(|op| matches!(op, PlanOp::Aggregate { .. }))
2561 {
2562 Some(p) => p,
2563 None => return Ok(ResultSet::new(vec![])),
2564 };
2565 let producers = &plan[..agg_pos];
2566 let (func, arg, column) = match &plan[agg_pos] {
2567 PlanOp::Aggregate { func, arg, column } => (func, arg, column),
2568 _ => unreachable!(),
2569 };
2570 let vars = collect_vars(plan);
2571 let ctx = AggStreamCtx {
2572 view,
2573 vars: &vars,
2574 params,
2575 func,
2576 arg,
2577 };
2578 let initial_row: Row = vec![None; vars.names.len()];
2579 let mut acc = AggAcc::new(func);
2580 agg_stream(&ctx, producers, &initial_row, &mut acc)?;
2581
2582 let value = acc.finish();
2583 let mut rs = ResultSet::new(vec![column.clone()]);
2584 rs.push_row(vec![value]);
2585 Ok(rs)
2586}
2587
2588fn update_acc(
2594 view: &GraphView,
2595 vars: &VarTable,
2596 row: &Row,
2597 func: &AggFunc,
2598 arg: &AggArg,
2599 acc: &mut AggAcc,
2600) -> Result<(), String> {
2601 match (func, arg) {
2602 (AggFunc::Count, AggArg::Star) => {
2603 if let AggAcc::Count(n) = acc {
2604 *n += 1;
2605 }
2606 }
2607 (AggFunc::Count, AggArg::Var(v)) => {
2608 let slot = vars.slot(v);
2609 let is_bound = slot
2610 .and_then(|s| row.get(s))
2611 .and_then(|c| c.as_ref())
2612 .is_some();
2613 if is_bound {
2614 if let AggAcc::Count(n) = acc {
2615 *n += 1;
2616 }
2617 }
2618 }
2619 (AggFunc::Count, AggArg::Prop { var, field }) => {
2620 let val = resolve_prop(view, vars, row, var, field)?;
2621 if val.is_some() {
2622 if let AggAcc::Count(n) = acc {
2623 *n += 1;
2624 }
2625 }
2626 }
2627 (AggFunc::Sum, AggArg::Prop { var, field }) => {
2628 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2629 if let Some(num) = numeric_val(&v) {
2630 if let AggAcc::Sum { val, has_value } = acc {
2631 *val += num;
2632 *has_value = true;
2633 }
2634 }
2635 }
2636 }
2637 (AggFunc::Avg, AggArg::Prop { var, field }) => {
2638 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2639 if let Some(num) = numeric_val(&v) {
2640 if let AggAcc::Avg { sum, n } = acc {
2641 *sum += num;
2642 *n += 1;
2643 }
2644 }
2645 }
2646 }
2647 (AggFunc::Min, AggArg::Prop { var, field }) => {
2648 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2649 if numeric_val(&v).is_some() {
2650 if let AggAcc::Min(current) = acc {
2651 *current = Some(match current.take() {
2652 None => v,
2653 Some(prev) => {
2654 if cmp_optional(Some(&prev), Some(&v), false)
2655 == std::cmp::Ordering::Greater
2656 {
2657 v
2658 } else {
2659 prev
2660 }
2661 }
2662 });
2663 }
2664 }
2665 }
2666 }
2667 (AggFunc::Max, AggArg::Prop { var, field }) => {
2668 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2669 if numeric_val(&v).is_some() {
2670 if let AggAcc::Max(current) = acc {
2671 *current = Some(match current.take() {
2672 None => v,
2673 Some(prev) => {
2674 if cmp_optional(Some(&prev), Some(&v), true)
2675 == std::cmp::Ordering::Greater
2676 {
2677 v
2678 } else {
2679 prev
2680 }
2681 }
2682 });
2683 }
2684 }
2685 }
2686 }
2687 (AggFunc::Collect, AggArg::Var(v)) => {
2688 let val = vars
2691 .slot(v)
2692 .and_then(|s| row.get(s))
2693 .and_then(|c| c.as_ref())
2694 .and_then(|cell| match cell {
2695 Cell::Scalar(x) => Some(x.clone()),
2696 Cell::Node(id) => view.ids.key_of(*id).map(|k| Value::Str(k.to_owned())),
2697 Cell::Path(h) => Some(Value::Int(*h as i64)),
2698 Cell::Rel(_) => None,
2699 });
2700 if let Some(val) = val {
2701 if let AggAcc::Collect(items) = acc {
2702 items.push(val);
2703 }
2704 }
2705 }
2706 (AggFunc::Collect, AggArg::Prop { var, field }) => {
2707 if let Some(val) = resolve_prop(view, vars, row, var, field)? {
2708 if let AggAcc::Collect(items) = acc {
2709 items.push(val);
2710 }
2711 }
2712 }
2713 _ => {}
2716 }
2717 Ok(())
2718}
2719
2720fn agg_stream(
2723 ctx: &AggStreamCtx<'_>,
2724 ops: &[PlanOp],
2725 row: &Row,
2726 acc: &mut AggAcc,
2727) -> Result<(), String> {
2728 let (op, rest) = match ops.split_first() {
2729 Some(pair) => pair,
2730 None => {
2731 update_acc(ctx.view, ctx.vars, row, ctx.func, ctx.arg, acc)?;
2733 return Ok(());
2734 }
2735 };
2736
2737 match op {
2738 PlanOp::ScanLabel { var, label } => {
2739 let ids = scan_ids(ctx.view, label.as_deref());
2740 let slot = ctx
2741 .vars
2742 .slot(var)
2743 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2744 for &id in &ids {
2745 let mut next = row.clone();
2746 next[slot] = Some(Cell::Node(id));
2747 agg_stream(ctx, rest, &next, acc)?;
2748 }
2749 }
2750 PlanOp::ScanKey { var, key, label } => {
2751 let slot = ctx
2752 .vars
2753 .slot(var)
2754 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2755 if let Some(id) =
2756 resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
2757 {
2758 let mut next = row.clone();
2759 next[slot] = Some(Cell::Node(id));
2760 agg_stream(ctx, rest, &next, acc)?;
2761 }
2762 }
2763 PlanOp::IndexScan {
2764 var,
2765 label,
2766 field,
2767 value,
2768 } => {
2769 let ids = index_scan_ids(
2770 ctx.view,
2771 ctx.vars,
2772 row,
2773 label.as_deref(),
2774 field,
2775 value,
2776 ctx.params,
2777 )?;
2778 let slot = ctx
2779 .vars
2780 .slot(var)
2781 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2782 for &id in &ids {
2783 let mut next = row.clone();
2784 next[slot] = Some(Cell::Node(id));
2785 agg_stream(ctx, rest, &next, acc)?;
2786 }
2787 }
2788 PlanOp::IndexIntersect {
2789 var,
2790 label,
2791 equalities,
2792 } => {
2793 let ids = index_intersect_ids(
2794 ctx.view,
2795 ctx.vars,
2796 row,
2797 label.as_deref(),
2798 equalities,
2799 ctx.params,
2800 )?;
2801 let slot = ctx
2802 .vars
2803 .slot(var)
2804 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2805 for id in ids {
2806 let mut next = row.clone();
2807 next[slot] = Some(Cell::Node(id));
2808 agg_stream(ctx, rest, &next, acc)?;
2809 }
2810 }
2811 PlanOp::Expand {
2812 from,
2813 rel_var,
2814 etypes,
2815 dir,
2816 to,
2817 to_label,
2818 to_props,
2819 } => {
2820 let etypes = resolve_etypes(ctx.view, etypes);
2821 let exp_dir = map_dir(*dir);
2822 let to_slot = ctx
2823 .vars
2824 .slot(to)
2825 .ok_or_else(|| format!("unbound variable `{to}`"))?;
2826 let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
2827 let from_id = require_node(row, ctx.vars, from)?;
2828 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
2829 Some(Cell::Node(id)) => Some(*id),
2830 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
2831 return Err(format!("variable `{to}` is not a node"))
2832 }
2833 None => None,
2834 };
2835 for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
2836 if row_has_edge(row, &e) {
2837 continue;
2838 }
2839 let nbr = neighbor(from_id, &e, *dir);
2840 if !ctx.view.visible(nbr) {
2841 continue;
2842 }
2843 if let Some(want) = bound_to {
2844 if nbr != want {
2845 continue;
2846 }
2847 }
2848 if !node_matches(
2849 ctx.view,
2850 ctx.vars,
2851 row,
2852 nbr,
2853 to_label.as_deref(),
2854 to_props,
2855 ctx.params,
2856 )? {
2857 continue;
2858 }
2859 let mut next = row.clone();
2860 if let Some(slot) = rel_slot {
2861 next[slot] = Some(Cell::Rel(e));
2862 }
2863 if bound_to.is_none() {
2864 next[to_slot] = Some(Cell::Node(nbr));
2865 }
2866 agg_stream(ctx, rest, &next, acc)?;
2867 }
2868 }
2869 PlanOp::Filter { expr } => {
2870 if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
2871 agg_stream(ctx, rest, row, acc)?;
2872 }
2873 }
2874 PlanOp::LookupProps { var, props } => {
2875 let id = require_node(row, ctx.vars, var)?;
2876 if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
2877 agg_stream(ctx, rest, row, acc)?;
2878 }
2879 }
2880 PlanOp::JoinBound { var, label, props } => {
2881 let id = require_node(row, ctx.vars, var)?;
2882 if node_matches(
2883 ctx.view,
2884 ctx.vars,
2885 row,
2886 id,
2887 label.as_deref(),
2888 props,
2889 ctx.params,
2890 )? {
2891 agg_stream(ctx, rest, row, acc)?;
2892 }
2893 }
2894 PlanOp::VarExpand {
2895 from,
2896 rel_var,
2897 etypes,
2898 dir,
2899 to,
2900 min,
2901 max,
2902 } => {
2903 let new_rows = exec_var_expand(
2904 ctx.view,
2905 ctx.vars,
2906 std::slice::from_ref(row),
2907 from,
2908 rel_var,
2909 etypes,
2910 *dir,
2911 to,
2912 *min,
2913 *max,
2914 )?;
2915 for nr in &new_rows {
2916 agg_stream(ctx, rest, nr, acc)?;
2917 }
2918 }
2919 PlanOp::ShortestPath {
2920 from,
2921 rel_var,
2922 etypes,
2923 dir,
2924 to,
2925 max_hops,
2926 } => {
2927 let new_rows = exec_shortest_path(
2928 ctx.view,
2929 ctx.vars,
2930 std::slice::from_ref(row),
2931 from,
2932 rel_var,
2933 etypes,
2934 *dir,
2935 to,
2936 *max_hops,
2937 )?;
2938 for nr in &new_rows {
2939 agg_stream(ctx, rest, nr, acc)?;
2940 }
2941 }
2942 PlanOp::Project { .. } => {
2944 return Err(
2945 "agg executor: Project in producer slice — plan is structurally malformed"
2946 .to_string(),
2947 );
2948 }
2949 PlanOp::Distinct => {
2950 return Err(
2951 "agg executor: Distinct in producer slice — plan is structurally malformed"
2952 .to_string(),
2953 );
2954 }
2955 PlanOp::OrderBy { .. } => {
2956 return Err(
2957 "agg executor: OrderBy in producer slice — structurally malformed".to_string(),
2958 );
2959 }
2960 PlanOp::Skip(_) => {
2961 return Err(
2962 "agg executor: Skip in producer slice — structurally malformed".to_string(),
2963 );
2964 }
2965 PlanOp::Limit(_) => {
2966 return Err(
2967 "agg executor: Limit in producer slice — structurally malformed".to_string(),
2968 );
2969 }
2970 PlanOp::Aggregate { .. } => {
2971 return Err(
2972 "agg executor: nested Aggregate in producer slice — structurally malformed"
2973 .to_string(),
2974 );
2975 }
2976 PlanOp::GroupAggregate { .. } => {
2977 return Err(
2978 "agg executor: GroupAggregate in producer slice — structurally malformed"
2979 .to_string(),
2980 );
2981 }
2982 PlanOp::With { .. } => {
2983 return Err(
2984 "agg executor: With in producer slice — structurally malformed".to_string(),
2985 );
2986 }
2987 PlanOp::Unwind { .. } => {
2988 return Err(
2989 "agg executor: Unwind in producer slice — structurally malformed".to_string(),
2990 );
2991 }
2992 PlanOp::LeftOuterApply { .. } => {
2993 return Err(
2994 "agg executor: LeftOuterApply in producer slice — structurally malformed"
2995 .to_string(),
2996 );
2997 }
2998 }
2999 Ok(())
3000}
3001
3002struct GroupStreamCtx<'a> {
3010 view: &'a GraphView<'a>,
3011 vars: &'a VarTable,
3012 params: &'a Params<'a>,
3013 keys: &'a [(String, RetItem)],
3014 aggs: &'a [(AggFunc, AggArg, String)],
3015}
3016
3017fn build_group_projected(
3019 keys: &[(String, RetItem)],
3020 aggs: &[(AggFunc, AggArg, String)],
3021 key_order: Vec<GroupKey>,
3022 groups: &mut HashMap<GroupKey, GroupEntry>,
3023) -> Projected {
3024 let columns: Vec<String> = keys
3025 .iter()
3026 .map(|(col, _)| col.clone())
3027 .chain(aggs.iter().map(|(_, _, col)| col.clone()))
3028 .collect();
3029 let mut rows: Vec<Vec<Option<Value>>> = Vec::with_capacity(key_order.len());
3030 for gk in key_order {
3031 let (display_keys, accs) = groups.remove(&gk).unwrap_or_default();
3032 let mut row: Vec<Option<Value>> = Vec::with_capacity(columns.len());
3033 for display_val in display_keys {
3036 row.push(display_val);
3037 }
3038 for acc in accs {
3039 row.push(acc.finish());
3040 }
3041 rows.push(row);
3042 }
3043 Projected { columns, rows }
3044}
3045
3046fn group_result_to_rows(
3053 keys: &[(String, RetItem)],
3054 aggs: &[(AggFunc, AggArg, String)],
3055 key_order: Vec<GroupKey>,
3056 groups: &mut HashMap<GroupKey, GroupEntry>,
3057 vars: &VarTable,
3058) -> Vec<Row> {
3059 let row_len = vars.names.len();
3060 let mut out: Vec<Row> = Vec::with_capacity(key_order.len());
3061 for gk in key_order {
3062 let (display_keys, accs) = groups.remove(&gk).unwrap_or_default();
3063 let mut row: Row = vec![None; row_len];
3064 for ((col, _), val) in keys.iter().zip(display_keys) {
3065 if let Some(slot) = vars.slot(col) {
3066 row[slot] = val.map(Cell::Scalar);
3067 }
3068 }
3069 for ((_, _, col), acc) in aggs.iter().zip(accs) {
3070 if let Some(slot) = vars.slot(col) {
3071 row[slot] = acc.finish().map(Cell::Scalar);
3072 }
3073 }
3074 out.push(row);
3075 }
3076 out
3077}
3078
3079fn exec_order_by_rows(vars: &VarTable, rows: &mut Vec<Row>, items: &[OrderItem], view: &GraphView) {
3086 let mut key_table: Vec<Vec<Option<Value>>> = Vec::with_capacity(rows.len());
3089 for row in rows.iter() {
3090 let mut row_key: Vec<Option<Value>> = Vec::with_capacity(items.len());
3091 for item in items {
3092 let val = match &item.target {
3093 OrderTarget::Alias(name) | OrderTarget::Var(name) => vars
3094 .slot(name)
3095 .and_then(|s| row.get(s))
3096 .and_then(|c| c.as_ref())
3097 .and_then(|c| match c {
3098 Cell::Scalar(v) => Some(v.clone()),
3099 Cell::Node(id) => view.ids.key_of(*id).map(|k| Value::Str(k.to_owned())),
3100 Cell::Path(hops) => Some(Value::Int(*hops as i64)),
3101 Cell::Rel(_) => None,
3102 }),
3103 OrderTarget::Prop { var, field } => vars
3104 .slot(var)
3105 .and_then(|s| row.get(s))
3106 .and_then(|c| c.as_ref())
3107 .and_then(|c| match c {
3108 Cell::Node(id) => view.prop(*id, field).map(|vr| vr.into_value()),
3109 Cell::Rel(e) => view.edge_props.get(e.etype, e.src, e.dst, field),
3110 _ => None,
3111 }),
3112 };
3113 row_key.push(val);
3114 }
3115 key_table.push(row_key);
3116 }
3117 let mut indices: Vec<usize> = (0..rows.len()).collect();
3119 indices.sort_by(|&a, &b| {
3120 for (ki, item) in items.iter().enumerate() {
3121 let c = cmp_optional(
3122 key_table[a].get(ki).and_then(|x| x.as_ref()),
3123 key_table[b].get(ki).and_then(|x| x.as_ref()),
3124 item.descending,
3125 );
3126 if c != std::cmp::Ordering::Equal {
3127 return c;
3128 }
3129 }
3130 std::cmp::Ordering::Equal
3131 });
3132 let sorted: Vec<Row> = indices.into_iter().map(|i| rows[i].clone()).collect();
3133 *rows = sorted;
3134}
3135
3136fn execute_group_aggregate(
3142 view: &GraphView,
3143 plan: &[PlanOp],
3144 params: &Params,
3145) -> Result<ResultSet, String> {
3146 let gagg_pos = plan
3147 .iter()
3148 .position(|op| matches!(op, PlanOp::GroupAggregate { .. }))
3149 .ok_or_else(|| "internal: GroupAggregate op not found in plan".to_string())?;
3150 let producers = &plan[..gagg_pos];
3151 let (keys, aggs) = match &plan[gagg_pos] {
3152 PlanOp::GroupAggregate { keys, aggs } => (keys, aggs),
3153 _ => unreachable!(),
3154 };
3155 let tail = &plan[gagg_pos + 1..];
3156 let vars = collect_vars(plan);
3157 let initial_row: Row = vec![None; vars.names.len()];
3158 let mut groups: HashMap<GroupKey, GroupEntry> = HashMap::new();
3159 let mut key_order: Vec<GroupKey> = Vec::new();
3160 let ctx = GroupStreamCtx {
3161 view,
3162 vars: &vars,
3163 params,
3164 keys,
3165 aggs,
3166 };
3167 group_stream(&ctx, producers, &initial_row, &mut groups, &mut key_order)?;
3168 if keys.is_empty() && key_order.is_empty() {
3172 let empty_key: GroupKey = vec![];
3173 key_order.push(empty_key.clone());
3174 groups.insert(
3175 empty_key,
3176 (
3177 vec![],
3178 aggs.iter().map(|(f, _, _)| AggAcc::new(f)).collect(),
3179 ),
3180 );
3181 }
3182 let mut projected = build_group_projected(keys, aggs, key_order, &mut groups);
3183 for op in tail {
3185 match op {
3186 PlanOp::OrderBy { items } => exec_order_by(&mut projected, items)?,
3187 PlanOp::Skip(ls) => {
3188 let n = resolve_ls(ls, params)?;
3189 apply_skip(&mut projected.rows, n);
3190 }
3191 PlanOp::Limit(ls) => {
3192 let n = resolve_ls(ls, params)?;
3193 apply_limit(&mut projected.rows, n);
3194 }
3195 _ => {} }
3197 }
3198 Ok(finish(projected))
3199}
3200
3201fn group_stream(
3208 ctx: &GroupStreamCtx<'_>,
3209 ops: &[PlanOp],
3210 row: &Row,
3211 groups: &mut HashMap<GroupKey, GroupEntry>,
3212 key_order: &mut Vec<GroupKey>,
3213) -> Result<(), String> {
3214 let (op, rest) = match ops.split_first() {
3215 Some(pair) => pair,
3216 None => {
3217 let mut gk: GroupKey = Vec::with_capacity(ctx.keys.len());
3220 let mut display_vals: Vec<Option<Value>> = Vec::with_capacity(ctx.keys.len());
3221 for (_, item) in ctx.keys {
3222 let val = project_item(ctx.view, ctx.vars, row, item, ctx.params)?;
3223 gk.push(val.as_ref().and_then(group_key_normalize));
3224 display_vals.push(val);
3225 }
3226 if !groups.contains_key(&gk) {
3227 if groups.len() >= max_groups() {
3228 return Err(group_cap_err());
3229 }
3230 key_order.push(gk.clone());
3231 let init: Vec<AggAcc> = ctx.aggs.iter().map(|(f, _, _)| AggAcc::new(f)).collect();
3232 groups.insert(gk.clone(), (display_vals, init));
3233 }
3234 let (_, accs) = groups.get_mut(&gk).unwrap();
3235 for (acc, (func, arg, _)) in accs.iter_mut().zip(ctx.aggs.iter()) {
3236 update_acc(ctx.view, ctx.vars, row, func, arg, acc)?;
3237 }
3238 return Ok(());
3239 }
3240 };
3241
3242 match op {
3243 PlanOp::ScanLabel { var, label } => {
3244 let ids = scan_ids(ctx.view, label.as_deref());
3245 let slot = ctx
3246 .vars
3247 .slot(var)
3248 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3249 for &id in &ids {
3250 let mut next = row.clone();
3251 next[slot] = Some(Cell::Node(id));
3252 group_stream(ctx, rest, &next, groups, key_order)?;
3253 }
3254 }
3255 PlanOp::ScanKey { var, key, label } => {
3256 let slot = ctx
3257 .vars
3258 .slot(var)
3259 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3260 if let Some(id) =
3261 resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
3262 {
3263 let mut next = row.clone();
3264 next[slot] = Some(Cell::Node(id));
3265 group_stream(ctx, rest, &next, groups, key_order)?;
3266 }
3267 }
3268 PlanOp::IndexScan {
3269 var,
3270 label,
3271 field,
3272 value,
3273 } => {
3274 let ids = index_scan_ids(
3275 ctx.view,
3276 ctx.vars,
3277 row,
3278 label.as_deref(),
3279 field,
3280 value,
3281 ctx.params,
3282 )?;
3283 let slot = ctx
3284 .vars
3285 .slot(var)
3286 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3287 for &id in &ids {
3288 let mut next = row.clone();
3289 next[slot] = Some(Cell::Node(id));
3290 group_stream(ctx, rest, &next, groups, key_order)?;
3291 }
3292 }
3293 PlanOp::IndexIntersect {
3294 var,
3295 label,
3296 equalities,
3297 } => {
3298 let ids = index_intersect_ids(
3299 ctx.view,
3300 ctx.vars,
3301 row,
3302 label.as_deref(),
3303 equalities,
3304 ctx.params,
3305 )?;
3306 let slot = ctx
3307 .vars
3308 .slot(var)
3309 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3310 for id in ids {
3311 let mut next = row.clone();
3312 next[slot] = Some(Cell::Node(id));
3313 group_stream(ctx, rest, &next, groups, key_order)?;
3314 }
3315 }
3316 PlanOp::Expand {
3317 from,
3318 rel_var,
3319 etypes,
3320 dir,
3321 to,
3322 to_label,
3323 to_props,
3324 } => {
3325 let etypes = resolve_etypes(ctx.view, etypes);
3326 let exp_dir = map_dir(*dir);
3327 let to_slot = ctx
3328 .vars
3329 .slot(to)
3330 .ok_or_else(|| format!("unbound variable `{to}`"))?;
3331 let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
3332 let from_id = require_node(row, ctx.vars, from)?;
3333 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
3334 Some(Cell::Node(id)) => Some(*id),
3335 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
3336 return Err(format!("variable `{to}` is not a node"))
3337 }
3338 None => None,
3339 };
3340 for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
3341 if row_has_edge(row, &e) {
3342 continue;
3343 }
3344 let nbr = neighbor(from_id, &e, *dir);
3345 if !ctx.view.visible(nbr) {
3346 continue;
3347 }
3348 if let Some(want) = bound_to {
3349 if nbr != want {
3350 continue;
3351 }
3352 }
3353 if !node_matches(
3354 ctx.view,
3355 ctx.vars,
3356 row,
3357 nbr,
3358 to_label.as_deref(),
3359 to_props,
3360 ctx.params,
3361 )? {
3362 continue;
3363 }
3364 let mut next = row.clone();
3365 if let Some(slot) = rel_slot {
3366 next[slot] = Some(Cell::Rel(e));
3367 }
3368 if bound_to.is_none() {
3369 next[to_slot] = Some(Cell::Node(nbr));
3370 }
3371 group_stream(ctx, rest, &next, groups, key_order)?;
3372 }
3373 }
3374 PlanOp::Filter { expr } => {
3375 if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
3376 group_stream(ctx, rest, row, groups, key_order)?;
3377 }
3378 }
3379 PlanOp::LookupProps { var, props } => {
3380 let id = require_node(row, ctx.vars, var)?;
3381 if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
3382 group_stream(ctx, rest, row, groups, key_order)?;
3383 }
3384 }
3385 PlanOp::JoinBound { var, label, props } => {
3386 let id = require_node(row, ctx.vars, var)?;
3387 if node_matches(
3388 ctx.view,
3389 ctx.vars,
3390 row,
3391 id,
3392 label.as_deref(),
3393 props,
3394 ctx.params,
3395 )? {
3396 group_stream(ctx, rest, row, groups, key_order)?;
3397 }
3398 }
3399 PlanOp::VarExpand {
3400 from,
3401 rel_var,
3402 etypes,
3403 dir,
3404 to,
3405 min,
3406 max,
3407 } => {
3408 let new_rows = exec_var_expand(
3409 ctx.view,
3410 ctx.vars,
3411 std::slice::from_ref(row),
3412 from,
3413 rel_var,
3414 etypes,
3415 *dir,
3416 to,
3417 *min,
3418 *max,
3419 )?;
3420 for nr in &new_rows {
3421 group_stream(ctx, rest, nr, groups, key_order)?;
3422 }
3423 }
3424 PlanOp::ShortestPath {
3425 from,
3426 rel_var,
3427 etypes,
3428 dir,
3429 to,
3430 max_hops,
3431 } => {
3432 let new_rows = exec_shortest_path(
3433 ctx.view,
3434 ctx.vars,
3435 std::slice::from_ref(row),
3436 from,
3437 rel_var,
3438 etypes,
3439 *dir,
3440 to,
3441 *max_hops,
3442 )?;
3443 for nr in &new_rows {
3444 group_stream(ctx, rest, nr, groups, key_order)?;
3445 }
3446 }
3447 PlanOp::Project { .. } => {
3449 return Err(
3450 "group executor: Project in producer slice — structurally malformed".to_string(),
3451 );
3452 }
3453 PlanOp::Distinct => {
3454 return Err(
3455 "group executor: Distinct in producer slice — structurally malformed".to_string(),
3456 );
3457 }
3458 PlanOp::OrderBy { .. } => {
3459 return Err(
3460 "group executor: OrderBy in producer slice — structurally malformed".to_string(),
3461 );
3462 }
3463 PlanOp::Skip(_) => {
3464 return Err(
3465 "group executor: Skip in producer slice — structurally malformed".to_string(),
3466 );
3467 }
3468 PlanOp::Limit(_) => {
3469 return Err(
3470 "group executor: Limit in producer slice — structurally malformed".to_string(),
3471 );
3472 }
3473 PlanOp::Aggregate { .. } => {
3474 return Err(
3475 "group executor: Aggregate in producer slice — structurally malformed".to_string(),
3476 );
3477 }
3478 PlanOp::GroupAggregate { .. } => {
3479 return Err(
3480 "group executor: nested GroupAggregate in producer slice — structurally malformed"
3481 .to_string(),
3482 );
3483 }
3484 PlanOp::With { .. } => {
3485 return Err(
3486 "group executor: With in producer slice — structurally malformed".to_string(),
3487 );
3488 }
3489 PlanOp::Unwind { .. } => {
3490 return Err(
3491 "group executor: Unwind in producer slice — structurally malformed".to_string(),
3492 );
3493 }
3494 PlanOp::LeftOuterApply { .. } => {
3495 return Err(
3496 "group executor: LeftOuterApply in producer slice — structurally malformed"
3497 .to_string(),
3498 );
3499 }
3500 }
3501 Ok(())
3502}
3503
3504fn pull_rows(
3518 ctx: &PullCtx<'_>,
3519 ops: &[PlanOp],
3520 row: &mut Row,
3521 result: &mut Vec<Vec<Option<Value>>>,
3522) -> Result<(), String> {
3523 if result.len() >= ctx.bound {
3524 return Ok(());
3525 }
3526 let (op, rest) = match ops.split_first() {
3527 Some(pair) => pair,
3528 None => {
3529 let mut cells = Vec::with_capacity(ctx.project_items.len());
3531 for item in ctx.project_items {
3532 cells.push(project_item(ctx.view, ctx.vars, row, item, ctx.params)?);
3533 }
3534 result.push(cells);
3535 return Ok(());
3536 }
3537 };
3538 match op {
3539 PlanOp::ScanLabel { var, label } => {
3540 let slot = ctx
3541 .vars
3542 .slot(var)
3543 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3544 let want_sym = label.as_deref().and_then(|l| ctx.view.syms.get(l));
3547 if label.is_some() && want_sym.is_none() {
3548 return Ok(());
3549 }
3550 let prev = row[slot].clone();
3552
3553 let fused_filter = rest.first().and_then(|next_op| {
3559 if let PlanOp::Filter {
3560 expr:
3561 Expr::Cmp {
3562 lhs:
3563 Operand::Prop {
3564 var: ref fv,
3565 field: ref f,
3566 },
3567 op: ref cmp_op_ref,
3568 rhs: Operand::Lit(ref lit),
3569 },
3570 } = *next_op
3571 {
3572 if fv == var {
3573 return Some((f.as_str(), cmp_op_ref, lit));
3574 }
3575 }
3576 None
3577 });
3578
3579 if let Some((field, cmp_op_ref, lit)) = fused_filter {
3580 #[cfg(test)]
3583 FUSED_SCAN_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
3584 let col = ctx.view.props.column(field);
3585 let rest_after_filter = &rest[1..];
3586 for (i, &sym) in ctx.view.labels.iter().enumerate() {
3587 if result.len() >= ctx.bound {
3588 break;
3589 }
3590 if sym == u32::MAX {
3591 continue;
3592 }
3593 if let Some(ws) = want_sym {
3594 if sym != ws {
3595 continue;
3596 }
3597 }
3598 let id = i as u32;
3599 if !ctx.view.visible(id) {
3600 continue;
3601 }
3602 if let Some(v) = col.get(id) {
3603 if eval_cmp(cmp_op_ref, v, lit) {
3604 row[slot] = Some(Cell::Node(id));
3605 pull_rows(ctx, rest_after_filter, row, result)?;
3606 }
3607 }
3608 }
3609 } else {
3610 for (i, &sym) in ctx.view.labels.iter().enumerate() {
3613 if result.len() >= ctx.bound {
3614 break;
3615 }
3616 if sym == u32::MAX {
3618 continue;
3619 }
3620 if let Some(ws) = want_sym {
3622 if sym != ws {
3623 continue;
3624 }
3625 }
3626 let id = i as u32;
3627 if !ctx.view.visible(id) {
3628 continue;
3629 }
3630 row[slot] = Some(Cell::Node(id));
3631 pull_rows(ctx, rest, row, result)?;
3632 }
3633 }
3634 row[slot] = prev;
3640 }
3641 PlanOp::ScanKey { var, key, label } => {
3642 let slot = ctx
3643 .vars
3644 .slot(var)
3645 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3646 if let Some(id) =
3647 resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
3648 {
3649 let prev = row[slot].clone();
3650 row[slot] = Some(Cell::Node(id));
3651 pull_rows(ctx, rest, row, result)?;
3652 row[slot] = prev;
3653 }
3654 }
3655 PlanOp::IndexScan {
3656 var,
3657 label,
3658 field,
3659 value,
3660 } => {
3661 let slot = ctx
3662 .vars
3663 .slot(var)
3664 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3665 let ids = index_scan_ids(
3666 ctx.view,
3667 ctx.vars,
3668 row,
3669 label.as_deref(),
3670 field,
3671 value,
3672 ctx.params,
3673 )?;
3674 let prev = row[slot].clone();
3675 for id in ids {
3676 if result.len() >= ctx.bound {
3677 break;
3678 }
3679 row[slot] = Some(Cell::Node(id));
3680 pull_rows(ctx, rest, row, result)?;
3681 }
3682 row[slot] = prev;
3683 }
3684 PlanOp::IndexIntersect {
3685 var,
3686 label,
3687 equalities,
3688 } => {
3689 let slot = ctx
3690 .vars
3691 .slot(var)
3692 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3693 let ids = index_intersect_ids(
3694 ctx.view,
3695 ctx.vars,
3696 row,
3697 label.as_deref(),
3698 equalities,
3699 ctx.params,
3700 )?;
3701 let prev = row[slot].clone();
3702 for id in ids {
3703 if result.len() >= ctx.bound {
3704 break;
3705 }
3706 row[slot] = Some(Cell::Node(id));
3707 pull_rows(ctx, rest, row, result)?;
3708 }
3709 row[slot] = prev;
3710 }
3711 PlanOp::Expand {
3712 from,
3713 rel_var,
3714 etypes,
3715 dir,
3716 to,
3717 to_label,
3718 to_props,
3719 } => {
3720 let etypes = resolve_etypes(ctx.view, etypes);
3721 let exp_dir = map_dir(*dir);
3722 let to_slot = ctx
3723 .vars
3724 .slot(to)
3725 .ok_or_else(|| format!("unbound variable `{to}`"))?;
3726 let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
3727 let from_id = require_node(row, ctx.vars, from)?;
3728 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
3729 Some(Cell::Node(id)) => Some(*id),
3730 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
3731 return Err(format!("variable `{to}` is not a node"))
3732 }
3733 None => None,
3734 };
3735 for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
3736 if result.len() >= ctx.bound {
3737 break;
3738 }
3739 if row_has_edge(row, &e) {
3740 continue;
3741 }
3742 let nbr = neighbor(from_id, &e, *dir);
3743 if !ctx.view.visible(nbr) {
3744 continue;
3745 }
3746 if let Some(want) = bound_to {
3747 if nbr != want {
3748 continue;
3749 }
3750 }
3751 if !node_matches(
3752 ctx.view,
3753 ctx.vars,
3754 row,
3755 nbr,
3756 to_label.as_deref(),
3757 to_props,
3758 ctx.params,
3759 )? {
3760 continue;
3761 }
3762 let mut next = row.clone();
3763 if let Some(slot) = rel_slot {
3764 next[slot] = Some(Cell::Rel(e));
3765 }
3766 if bound_to.is_none() {
3767 next[to_slot] = Some(Cell::Node(nbr));
3768 }
3769 #[cfg(test)]
3770 record_expand_row();
3771 pull_rows(ctx, rest, &mut next, result)?;
3772 }
3773 }
3774 PlanOp::Filter { expr } => {
3775 if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
3776 pull_rows(ctx, rest, row, result)?;
3777 }
3778 }
3779 PlanOp::LookupProps { var, props } => {
3780 let id = require_node(row, ctx.vars, var)?;
3781 if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
3782 pull_rows(ctx, rest, row, result)?;
3783 }
3784 }
3785 PlanOp::JoinBound { var, label, props } => {
3786 let id = require_node(row, ctx.vars, var)?;
3787 if node_matches(
3788 ctx.view,
3789 ctx.vars,
3790 row,
3791 id,
3792 label.as_deref(),
3793 props,
3794 ctx.params,
3795 )? {
3796 pull_rows(ctx, rest, row, result)?;
3797 }
3798 }
3799 PlanOp::Project { .. } => {
3804 return Err(
3805 "pull executor: Project reached pull_rows — plan is structurally malformed"
3806 .to_string(),
3807 );
3808 }
3809 PlanOp::Distinct => {
3810 return Err(
3811 "pull executor: Distinct reached pull_rows — DISTINCT queries must use \
3812 the staged path (row_bound returns None)"
3813 .to_string(),
3814 );
3815 }
3816 PlanOp::OrderBy { .. } => {
3817 return Err(
3818 "pull executor: OrderBy reached pull_rows — queries with ORDER BY \
3819 must use the staged path (row_bound returns None)"
3820 .to_string(),
3821 );
3822 }
3823 PlanOp::Skip(_) => {
3824 return Err(
3825 "pull executor: Skip reached pull_rows — Skip must appear after Project"
3826 .to_string(),
3827 );
3828 }
3829 PlanOp::Limit(_) => {
3830 return Err(
3831 "pull executor: Limit reached pull_rows — Limit must appear after Project"
3832 .to_string(),
3833 );
3834 }
3835 PlanOp::Aggregate { .. } => {
3836 return Err(
3837 "pull executor: Aggregate reached pull_rows — aggregate plans must use \
3838 the execute_aggregate path (routed before pull in execute_inner)"
3839 .to_string(),
3840 );
3841 }
3842 PlanOp::VarExpand { .. } => {
3847 return Err(
3848 "pull executor: VarExpand reached pull_rows — variable-length path \
3849 plans must use the staged path (row_bound returns None)"
3850 .to_string(),
3851 );
3852 }
3853 PlanOp::ShortestPath { .. } => {
3854 return Err(
3855 "pull executor: ShortestPath reached pull_rows — shortestPath plans \
3856 must use the staged path (row_bound returns None)"
3857 .to_string(),
3858 );
3859 }
3860 PlanOp::GroupAggregate { .. } => {
3861 return Err(
3862 "pull executor: GroupAggregate reached pull_rows — grouped aggregate plans \
3863 must use the execute_group_aggregate path (routed before pull in execute_inner)"
3864 .to_string(),
3865 );
3866 }
3867 PlanOp::With { .. } => {
3868 return Err(
3869 "pull executor: With reached pull_rows — pipeline plans must use the staged path \
3870 (row_bound returns None for plans containing With)"
3871 .to_string(),
3872 );
3873 }
3874 PlanOp::Unwind { .. } => {
3875 return Err(
3876 "pull executor: Unwind reached pull_rows — pipeline plans must use the staged path \
3877 (row_bound returns None for plans containing Unwind)"
3878 .to_string(),
3879 );
3880 }
3881 PlanOp::LeftOuterApply { .. } => {
3882 return Err(
3883 "pull executor: LeftOuterApply reached pull_rows — OPTIONAL MATCH plans must use \
3884 the staged path (row_bound returns None for plans containing LeftOuterApply)"
3885 .to_string(),
3886 );
3887 }
3888 }
3889 Ok(())
3890}
3891
3892fn eval_expr(
3893 view: &GraphView,
3894 vars: &VarTable,
3895 row: &Row,
3896 expr: &Expr,
3897 params: &Params,
3898 depth: u32,
3899) -> Result<bool, String> {
3900 if depth > 256 {
3901 return Err("expression nesting too deep".into());
3902 }
3903 match expr {
3904 Expr::And(lhs, rhs) => {
3905 let l = eval_expr(view, vars, row, lhs, params, depth + 1)?;
3906 let r = eval_expr(view, vars, row, rhs, params, depth + 1)?;
3907 Ok(l && r)
3908 }
3909 Expr::Or(lhs, rhs) => {
3910 let l = eval_expr(view, vars, row, lhs, params, depth + 1)?;
3911 let r = eval_expr(view, vars, row, rhs, params, depth + 1)?;
3912 Ok(l || r)
3913 }
3914 Expr::Not(inner) => Ok(!eval_expr(view, vars, row, inner, params, depth + 1)?),
3915 Expr::Cmp { lhs, op, rhs } => {
3916 let l = resolve_operand(view, vars, row, lhs, params)?;
3917 let r = resolve_operand(view, vars, row, rhs, params)?;
3918 match (l, r) {
3919 (Some(a), Some(b)) => Ok(eval_cmp(op, &a, &b)),
3920 _ => Ok(false),
3921 }
3922 }
3923 Expr::Truthy(op) => {
3924 let val = resolve_operand(view, vars, row, op, params)?;
3925 Ok(match val {
3926 None => false,
3927 Some(Value::Bool(b)) => b,
3928 Some(Value::Int(n)) => n != 0,
3929 Some(Value::Float(f)) => f != 0.0,
3930 Some(Value::Str(s)) => !s.is_empty(),
3931 Some(Value::List(v)) => !v.is_empty(),
3932 Some(Value::Map(m)) => !m.is_empty(),
3933 })
3934 }
3935 Expr::IsNull(op) => {
3936 let val = resolve_operand(view, vars, row, op, params)?;
3937 Ok(val.is_none())
3938 }
3939 Expr::IsNotNull(op) => {
3940 let val = resolve_operand(view, vars, row, op, params)?;
3941 Ok(val.is_some())
3942 }
3943 Expr::In { expr, list } => eval_in(view, vars, row, expr, list, params),
3944 }
3945}
3946
3947fn eval_in(
3948 view: &GraphView,
3949 vars: &VarTable,
3950 row: &Row,
3951 expr: &Operand,
3952 list: &[Operand],
3953 params: &Params,
3954) -> Result<bool, String> {
3955 let Some(needle) = resolve_operand(view, vars, row, expr, params)? else {
3956 return Ok(false);
3957 };
3958 for item_op in list {
3959 match resolve_operand(view, vars, row, item_op, params)? {
3960 None => {}
3961 Some(Value::List(items)) => {
3962 for item in items {
3963 if crate::filter::eval_cmp(&crate::filter::CmpOp::Eq, &needle, &item) {
3964 return Ok(true);
3965 }
3966 }
3967 }
3968 Some(item) if crate::filter::eval_cmp(&crate::filter::CmpOp::Eq, &needle, &item) => {
3969 return Ok(true);
3970 }
3971 Some(_) => {}
3972 }
3973 }
3974 Ok(false)
3975}
3976
3977fn exec_distinct(table: &mut Projected) -> Result<(), String> {
3979 let cap = max_intermediate_rows();
3980 let mut seen: BTreeSet<Vec<Option<ValueKey>>> = BTreeSet::new();
3981 let mut out = Vec::with_capacity(table.rows.len().min(cap));
3982 for row in table.rows.drain(..) {
3983 let key: Vec<Option<ValueKey>> = row
3984 .iter()
3985 .map(|cell| cell.as_ref().and_then(group_key_normalize))
3986 .collect();
3987 if seen.insert(key) {
3988 if out.len() >= cap {
3989 return Err(row_cap_err(cap));
3990 }
3991 out.push(row);
3992 }
3993 }
3994 table.rows = out;
3995 Ok(())
3996}
3997
3998fn column_name(item: &RetItem) -> String {
3999 if let Some(alias) = &item.alias {
4000 return alias.clone();
4001 }
4002 match &item.value {
4003 RetVal::Var(v) => v.clone(),
4004 RetVal::Prop { var, field } => format!("{var}.{field}"),
4005 RetVal::Agg { func, arg } => {
4008 use crate::cypher::ast::AggArg;
4009 let f = match func {
4010 AggFunc::Count => "COUNT",
4011 AggFunc::Sum => "SUM",
4012 AggFunc::Avg => "AVG",
4013 AggFunc::Min => "MIN",
4014 AggFunc::Max => "MAX",
4015 AggFunc::Collect => "COLLECT",
4016 };
4017 let a = match arg {
4018 AggArg::Star => "*".to_string(),
4019 AggArg::Var(v) => v.clone(),
4020 AggArg::Prop { var, field } => format!("{var}.{field}"),
4021 };
4022 format!("{f}({a})")
4023 }
4024 RetVal::FuncCall { name, args } => {
4025 let arg_strs: Vec<String> = args
4026 .iter()
4027 .map(|a| match a {
4028 Operand::Var(v) => v.clone(),
4029 Operand::Prop { var, field } => format!("{var}.{field}"),
4030 Operand::Lit(_) => "<lit>".to_string(),
4031 Operand::Param(p) => format!("${p}"),
4032 Operand::FuncCall { name: n, .. } => format!("{n}(...)"),
4033 Operand::BinArith { .. } => "<arith>".to_string(),
4034 Operand::Case { .. } => "<case>".to_string(),
4035 })
4036 .collect();
4037 format!("{name}({})", arg_strs.join(", "))
4038 }
4039 RetVal::ScalarExpr(_) => "<expr>".to_string(),
4040 }
4041}
4042
4043fn exec_project(
4044 view: &GraphView,
4045 vars: &VarTable,
4046 rows: &[Row],
4047 items: &[RetItem],
4048 params: &Params,
4049) -> Result<Projected, String> {
4050 let columns: Vec<String> = items.iter().map(column_name).collect();
4051 let mut out_rows = Vec::with_capacity(rows.len());
4052 for row in rows {
4053 let mut cells = Vec::with_capacity(items.len());
4054 for item in items {
4055 cells.push(project_item(view, vars, row, item, params)?);
4056 }
4057 out_rows.push(cells);
4058 }
4059 Ok(Projected {
4060 columns,
4061 rows: out_rows,
4062 })
4063}
4064
4065fn project_item(
4066 view: &GraphView,
4067 vars: &VarTable,
4068 row: &Row,
4069 item: &RetItem,
4070 params: &Params,
4071) -> Result<Option<Value>, String> {
4072 match &item.value {
4073 RetVal::Var(v) => {
4074 let slot = vars.slot(v).ok_or_else(|| format!("unbound variable `{v}`"))?;
4076 match row.get(slot).and_then(|c| c.as_ref()) {
4077 None => Ok(None),
4079 Some(Cell::Node(id)) => match view.ids.key_of(*id) {
4080 Some(key) => Ok(Some(Value::Str(key.to_owned()))),
4081 None => Err(format!("unknown node id {id}")),
4082 },
4083 Some(Cell::Scalar(val)) => Ok(Some(val.clone())),
4085 Some(Cell::Rel(_)) => Err(format!(
4086 "variable `{v}` is a relationship; return its properties ({v}.field) instead"
4087 )),
4088 Some(Cell::Path(hops)) => Ok(Some(Value::Int(*hops as i64))),
4089 }
4090 }
4091 RetVal::Prop { var, field } => resolve_prop(view, vars, row, var, field),
4092 RetVal::Agg { .. } => Err(
4095 "project_item: Agg variant reached exec_project — aggregate plans must not contain Project"
4096 .to_string(),
4097 ),
4098 RetVal::FuncCall { name, args } => eval_func(name, args, view, vars, row, params),
4099 RetVal::ScalarExpr(op) => resolve_operand(view, vars, row, op, params),
4100 }
4101}
4102
4103fn order_column(item: &OrderItem) -> String {
4104 match &item.target {
4105 OrderTarget::Alias(name) | OrderTarget::Var(name) => name.clone(),
4106 OrderTarget::Prop { var, field } => format!("{var}.{field}"),
4107 }
4108}
4109
4110fn exec_order_by(table: &mut Projected, items: &[OrderItem]) -> Result<(), String> {
4111 let mut keys = Vec::with_capacity(items.len());
4112 for item in items {
4113 let name = order_column(item);
4114 let idx = table
4115 .columns
4116 .iter()
4117 .position(|c| c == &name)
4118 .ok_or_else(|| format!("ORDER BY target `{name}` is not a projected column"))?;
4119 keys.push((idx, item.descending));
4120 }
4121 table.rows.sort_by(|a, b| {
4122 for &(idx, desc) in &keys {
4123 let c = cmp_optional(
4124 a.get(idx).and_then(|x| x.as_ref()),
4125 b.get(idx).and_then(|x| x.as_ref()),
4126 desc,
4127 );
4128 if c != std::cmp::Ordering::Equal {
4129 return c;
4130 }
4131 }
4132 std::cmp::Ordering::Equal
4133 });
4134 Ok(())
4135}
4136
4137fn resolve_ls(ls: &LimitSkip, params: &Params) -> Result<u64, String> {
4142 match ls {
4143 LimitSkip::Exact(n) => Ok(*n),
4144 LimitSkip::Param(name) => {
4145 let val = params
4146 .0
4147 .get(name)
4148 .ok_or_else(|| format!("missing parameter `{name}` (used in LIMIT/SKIP)"))?;
4149 match val {
4150 Value::Int(i) if *i >= 0 => Ok(*i as u64),
4151 Value::Int(i) => Err(format!(
4152 "LIMIT/SKIP parameter `{name}` must be a non-negative integer, got {i}"
4153 )),
4154 other => Err(format!(
4155 "LIMIT/SKIP parameter `{name}` must be an integer, got {other:?}"
4156 )),
4157 }
4158 }
4159 }
4160}
4161
4162fn apply_skip<T>(rows: &mut Vec<T>, n: u64) {
4163 let n = usize::try_from(n).unwrap_or(usize::MAX);
4164 if n >= rows.len() {
4165 rows.clear();
4166 } else {
4167 rows.drain(0..n);
4168 }
4169}
4170
4171fn apply_limit<T>(rows: &mut Vec<T>, n: u64) {
4172 let n = usize::try_from(n).unwrap_or(usize::MAX);
4173 rows.truncate(n);
4174}
4175
4176#[cfg(test)]
4177mod tests {
4178 use super::{execute, resolve_operand, Params, Row, VarTable};
4179 use crate::cypher::ast::{
4180 ArithOp, LimitSkip, Operand, OrderItem, OrderTarget, RetItem, RetVal,
4181 };
4182 use crate::cypher::plan::{plan, PlanOp};
4183 use crate::cypher::{lex, parse, RelDir};
4184 use crate::result::ResultSet;
4185 use crate::view::GraphView;
4186 use core_storage::v8::seam::{ColumnsView, EdgePropsView, TopologyView};
4187 use core_storage::{ColumnStore, EdgeProps, IdMap, Interner, Topology, Value};
4188 use proptest::prelude::*;
4189 use std::collections::BTreeMap;
4190
4191 struct Fx {
4192 ids: IdMap,
4193 syms: Interner,
4194 labels: Vec<u32>,
4195 props: ColumnStore,
4196 topo: Topology,
4197 eprops: EdgeProps,
4198 }
4199
4200 impl Fx {
4201 fn new() -> Self {
4202 Fx {
4203 ids: IdMap::new(),
4204 syms: Interner::new(),
4205 labels: vec![],
4206 props: ColumnStore::new(),
4207 topo: Topology::new(),
4208 eprops: EdgeProps::new(),
4209 }
4210 }
4211
4212 fn add(&mut self, label: &str, key: &str, props: Vec<(&str, Value)>) -> u32 {
4213 let id = self.ids.get_or_insert(key);
4214 let sym = self.syms.intern(label);
4215 self.labels.resize(id as usize + 1, u32::MAX);
4216 self.labels[id as usize] = sym;
4217 for (f, v) in props {
4218 self.props.set(id, f, v);
4219 }
4220 id
4221 }
4222
4223 fn edge(&mut self, etype: &str, src: u32, dst: u32, props: Vec<(&str, Value)>) {
4224 let et = self.syms.intern(etype);
4225 self.topo.add_edge(et, src, dst);
4226 for (f, v) in props {
4227 self.eprops.set(et, src, dst, f, v);
4228 }
4229 }
4230
4231 fn view(&self) -> GraphView<'_> {
4232 GraphView {
4233 ids: &self.ids,
4234 syms: &self.syms,
4235 labels: &self.labels,
4236 props: ColumnsView::owned(&self.props),
4237 topo: TopologyView::owned(&self.topo),
4238 edge_props: EdgePropsView::owned(&self.eprops),
4239 mask: None,
4240 prop_index: None,
4241 }
4242 }
4243
4244 fn view_indexed<'a>(
4245 &'a self,
4246 index: &'a core_storage::property_index::PropertyIndex,
4247 ) -> GraphView<'a> {
4248 GraphView {
4249 prop_index: Some(index),
4250 ..self.view()
4251 }
4252 }
4253 }
4254
4255 fn compile(src: &str) -> Vec<PlanOp> {
4256 plan(&parse(&lex(src).expect("lex")).expect("parse")).expect("plan")
4257 }
4258
4259 fn run(
4260 view: &GraphView,
4261 src: &str,
4262 params: &BTreeMap<String, Value>,
4263 ) -> Result<ResultSet, String> {
4264 execute(view, &compile(src), &Params(params))
4265 }
4266
4267 fn s(v: &str) -> Value {
4268 Value::Str(v.into())
4269 }
4270
4271 fn f(v: f64) -> Value {
4272 Value::Float(v)
4273 }
4274
4275 fn i(v: i64) -> Value {
4276 Value::Int(v)
4277 }
4278
4279 fn rows_of(rs: &ResultSet) -> Vec<Vec<Option<Value>>> {
4280 (0..rs.len()).map(|i| rs.row(i).to_vec()).collect()
4281 }
4282
4283 fn col(rs: &ResultSet, name: &str) -> Vec<Option<Value>> {
4284 (0..rs.len()).map(|i| rs.get(i, name).cloned()).collect()
4285 }
4286
4287 fn hop_graph() -> Fx {
4288 let mut fx = Fx::new();
4289 let ada = fx.add("Person", "ada", vec![]);
4290 let bob = fx.add("Person", "bob", vec![]);
4291 let cam = fx.add("Person", "cam", vec![]);
4292 let acme = fx.add("Company", "acme", vec![]);
4293 fx.edge("KNOWS", ada, bob, vec![]);
4294 fx.edge("KNOWS", ada, cam, vec![]);
4295 fx.edge("KNOWS", bob, cam, vec![]);
4296 fx.edge("LIKES", ada, acme, vec![]);
4297 fx
4298 }
4299
4300 fn undirected_graph() -> Fx {
4301 let mut fx = Fx::new();
4302 let a = fx.add("N", "a", vec![]);
4303 let b = fx.add("N", "b", vec![]);
4304 fx.edge("T", a, b, vec![("w", i(42))]);
4305 fx
4306 }
4307
4308 fn triangle() -> Fx {
4309 let mut fx = Fx::new();
4310 let a = fx.add("N", "a", vec![]);
4311 let b = fx.add("N", "b", vec![]);
4312 let c = fx.add("N", "c", vec![]);
4313 fx.edge("T", a, b, vec![("eid", i(1))]);
4314 fx.edge("T", b, c, vec![("eid", i(2))]);
4315 fx.edge("T", c, a, vec![("eid", i(3))]);
4316 fx
4317 }
4318
4319 fn single_edge() -> (Fx, u32, u32) {
4320 let mut fx = Fx::new();
4321 let a = fx.add("N", "a", vec![]);
4322 let b = fx.add("N", "b", vec![]);
4323 fx.edge("T", a, b, vec![]);
4324 (fx, a, b)
4325 }
4326
4327 fn dogfood_graph() -> Fx {
4329 let mut fx = Fx::new();
4330 let t1 = fx.add("Talent", "t1", vec![("id", s("t1"))]);
4331 let acme = fx.add("Company", "acme", vec![]);
4332 let beta = fx.add("Company", "beta", vec![]);
4333 let gamma = fx.add("Company", "gamma", vec![]);
4334 let delta = fx.add("Company", "delta", vec![]);
4335 let echo = fx.add("Company", "echo", vec![]);
4336 let foxtrot = fx.add("Company", "foxtrot", vec![]);
4337 let zeta = fx.add("Company", "zeta", vec![]);
4338 fx.edge("INDUSTRY_ALIGNMENT", acme, t1, vec![("score", f(0.9))]);
4339 fx.edge("SPECIALTY_MATCH", acme, t1, vec![("score", f(0.8))]);
4340 fx.edge("INDUSTRY_ALIGNMENT", beta, t1, vec![("score", f(0.6))]);
4341 fx.edge("SPECIALTY_MATCH", beta, t1, vec![("score", f(0.7))]);
4342 fx.edge("INDUSTRY_ALIGNMENT", gamma, t1, vec![("score", f(0.4))]);
4343 fx.edge("SPECIALTY_MATCH", gamma, t1, vec![("score", f(0.9))]);
4344 fx.edge("INDUSTRY_ALIGNMENT", delta, t1, vec![("score", f(0.8))]);
4345 fx.edge("SPECIALTY_MATCH", delta, t1, vec![("score", f(0.3))]);
4346 fx.edge("INDUSTRY_ALIGNMENT", echo, t1, vec![("score", f(0.5))]);
4347 fx.edge("SPECIALTY_MATCH", echo, t1, vec![("score", f(0.5))]);
4348 fx.edge("INDUSTRY_ALIGNMENT", foxtrot, t1, vec![("score", f(0.95))]);
4349 fx.edge("INDUSTRY_ALIGNMENT", zeta, t1, vec![("score", f(0.9))]);
4350 fx.edge("SPECIALTY_MATCH", zeta, t1, vec![("score", f(0.6))]);
4351 fx
4352 }
4353
4354 const DOGFOOD: &str = "\
4355MATCH (t:Talent {id: $tid}) \
4356MATCH (c:Company)-[i:INDUSTRY_ALIGNMENT]->(t) \
4357MATCH (c)-[s:SPECIALTY_MATCH]->(t) \
4358WHERE i.score >= 0.5 AND s.score >= 0.5 \
4359RETURN c, i.score AS industry, s.score AS specialty \
4360ORDER BY industry DESC, specialty DESC \
4361LIMIT 10";
4362
4363 fn tid_params() -> BTreeMap<String, Value> {
4364 let mut p = BTreeMap::new();
4365 p.insert("tid".into(), s("t1"));
4366 p
4367 }
4368
4369 #[test]
4370 fn single_hop_match_label_and_etype_filters() {
4371 let fx = hop_graph();
4372 let v = fx.view();
4373 let rs = run(
4374 &v,
4375 "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b",
4376 &BTreeMap::new(),
4377 )
4378 .expect("single-hop");
4379 assert_eq!(rs.columns(), &["a".to_string(), "b".to_string()]);
4380 assert_eq!(
4381 rows_of(&rs),
4382 vec![
4383 vec![Some(s("ada")), Some(s("bob"))],
4384 vec![Some(s("ada")), Some(s("cam"))],
4385 vec![Some(s("bob")), Some(s("cam"))],
4386 ]
4387 );
4388 let likes = run(
4390 &v,
4391 "MATCH (a:Person)-[:LIKES]->(b:Company) RETURN a, b",
4392 &BTreeMap::new(),
4393 )
4394 .unwrap();
4395 assert_eq!(rows_of(&likes), vec![vec![Some(s("ada")), Some(s("acme"))]]);
4396 let no_combo = run(
4397 &v,
4398 "MATCH (a:Person)-[:KNOWS]->(b:Company) RETURN a, b",
4399 &BTreeMap::new(),
4400 )
4401 .unwrap();
4402 assert!(no_combo.is_empty());
4403 }
4404
4405 #[test]
4406 fn undirected_match_finds_both_orientations_and_binds_true_triple() {
4407 let fx = undirected_graph();
4408 let v = fx.view();
4409 let rs =
4412 run(&v, "MATCH (x)-[r:T]-(y) RETURN x, y, r.w", &BTreeMap::new()).expect("undirected");
4413 assert_eq!(
4414 rows_of(&rs),
4415 vec![
4416 vec![Some(s("a")), Some(s("b")), Some(i(42))],
4417 vec![Some(s("b")), Some(s("a")), Some(i(42))],
4418 ]
4419 );
4420
4421 let left = run(
4422 &v,
4423 "MATCH (y)<-[r:T]-(x) RETURN x, y, r.w",
4424 &BTreeMap::new(),
4425 )
4426 .unwrap();
4427 assert_eq!(
4428 rows_of(&left),
4429 vec![vec![Some(s("a")), Some(s("b")), Some(i(42))]]
4430 );
4431 }
4432
4433 #[test]
4434 fn relationship_uniqueness_triangle_and_two_hop_cycle() {
4435 let tri = triangle();
4436 let v = tri.view();
4437 let rs = run(
4438 &v,
4439 "MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z, r1.eid, r2.eid",
4440 &BTreeMap::new(),
4441 )
4442 .expect("triangle 2-hop");
4443 assert_eq!(
4444 rows_of(&rs),
4445 vec![
4446 vec![
4447 Some(s("a")),
4448 Some(s("b")),
4449 Some(s("c")),
4450 Some(i(1)),
4451 Some(i(2))
4452 ],
4453 vec![
4454 Some(s("b")),
4455 Some(s("c")),
4456 Some(s("a")),
4457 Some(i(2)),
4458 Some(i(3))
4459 ],
4460 vec![
4461 Some(s("c")),
4462 Some(s("a")),
4463 Some(s("b")),
4464 Some(i(3)),
4465 Some(i(1))
4466 ],
4467 ]
4468 );
4469 for row in rows_of(&rs) {
4470 assert_ne!(row[3], row[4], "r1 must never bind the same edge as r2");
4471 }
4472
4473 let (mut one, a, b) = single_edge();
4474 let v = one.view();
4475 let cycle = run(
4476 &v,
4477 "MATCH (x)-[:T]->(y)-[:T]->(x) RETURN x",
4478 &BTreeMap::new(),
4479 )
4480 .expect("2-hop cycle");
4481 assert!(
4482 cycle.is_empty(),
4483 "single directed edge cannot close a 2-hop cycle"
4484 );
4485
4486 let undirected_cycle = run(&v, "MATCH (x)-[:T]-(y)-[:T]-(x) RETURN x", &BTreeMap::new())
4488 .expect("undirected uniqueness");
4489 assert!(
4490 undirected_cycle.is_empty(),
4491 "relationship uniqueness must reject walking the same triple back"
4492 );
4493
4494 one.edge("T", b, a, vec![]);
4495 let v = one.view();
4496 let with_recip = run(
4497 &v,
4498 "MATCH (x)-[:T]->(y)-[:T]->(x) RETURN x",
4499 &BTreeMap::new(),
4500 )
4501 .unwrap();
4502 assert_eq!(col(&with_recip, "x"), vec![Some(s("a")), Some(s("b"))]);
4503 }
4504
4505 #[test]
4506 fn multi_match_join_bound_and_bound_destination_expand() {
4507 let fx = dogfood_graph();
4508 let v = fx.view();
4509 let rs = run(
4512 &v,
4513 "MATCH (t:Talent {id: $tid}) \
4514 MATCH (c:Company)-[i:INDUSTRY_ALIGNMENT]->(t) \
4515 MATCH (c)-[s:SPECIALTY_MATCH]->(t) \
4516 RETURN c",
4517 &tid_params(),
4518 )
4519 .expect("join + bound dest");
4520 assert_eq!(
4521 col(&rs, "c"),
4522 vec![
4523 Some(s("acme")),
4524 Some(s("beta")),
4525 Some(s("gamma")),
4526 Some(s("delta")),
4527 Some(s("echo")),
4528 Some(s("zeta")),
4529 ]
4530 );
4531
4532 let keep = run(&v, "MATCH (c:Company) MATCH (c) RETURN c", &BTreeMap::new()).unwrap();
4534 assert_eq!(keep.len(), 7);
4535 let drop = run(
4537 &v,
4538 "MATCH (c:Company) MATCH (c:Talent) RETURN c",
4539 &BTreeMap::new(),
4540 )
4541 .unwrap();
4542 assert!(drop.is_empty());
4543 }
4544
4545 #[test]
4546 fn scan_key_exec_does_not_use_label_scan() {
4547 let mut fx = Fx::new();
4548 fx.add("Person", "p1", vec![]);
4549 fx.add("Person", "p2", vec![]);
4550 fx.add("Person", "p3", vec![]);
4551 fx.add("Company", "c1", vec![]);
4552 let v = fx.view();
4553 let params = BTreeMap::new();
4554
4555 let fires_before = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
4556 let rs = run(&v, "MATCH (n:Person {id: 'p2'}) RETURN n", ¶ms).expect("scan key");
4557 let fires_after = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
4558 assert!(
4559 fires_after > fires_before,
4560 "SCAN_KEY_FIRES must increment; before={fires_before} after={fires_after}"
4561 );
4562 assert_eq!(rows_of(&rs), vec![vec![Some(s("p2"))]]);
4563
4564 let miss = run(&v, "MATCH (n:Person {id: 'nope'}) RETURN n", ¶ms).unwrap();
4565 assert!(
4566 miss.is_empty(),
4567 "missing key must be zero rows, not an error"
4568 );
4569
4570 let wrong = run(&v, "MATCH (n:Person {id: 'c1'}) RETURN n", ¶ms).unwrap();
4571 assert!(
4572 wrong.is_empty(),
4573 "wrong label must be zero rows, not an error"
4574 );
4575 }
4576
4577 #[test]
4578 fn rel_var_edge_prop_filter() {
4579 let mut fx = Fx::new();
4580 let a = fx.add("N", "a", vec![]);
4581 let b = fx.add("N", "b", vec![]);
4582 let c = fx.add("N", "c", vec![]);
4583 fx.edge("T", a, b, vec![("w", f(0.7))]);
4584 fx.edge("T", a, c, vec![("w", f(0.3))]);
4585 let v = fx.view();
4586 let rs = run(
4587 &v,
4588 "MATCH (x)-[r:T]->(y) WHERE r.w >= 0.5 RETURN y, r.w",
4589 &BTreeMap::new(),
4590 )
4591 .expect("edge-prop filter");
4592 assert_eq!(rows_of(&rs), vec![vec![Some(s("b")), Some(f(0.7))]]);
4593 let fail = run(
4594 &v,
4595 "MATCH (x)-[r:T]->(y) WHERE r.w >= 0.8 RETURN y",
4596 &BTreeMap::new(),
4597 )
4598 .unwrap();
4599 assert!(fail.is_empty());
4600 }
4601
4602 #[test]
4603 fn params_present_resolve_missing_is_err_before_rows() {
4604 let fx = dogfood_graph();
4605 let v = fx.view();
4606 let hit =
4607 run(&v, "MATCH (t:Talent {id: $tid}) RETURN t", &tid_params()).expect("present param");
4608 assert_eq!(col(&hit, "t"), vec![Some(s("t1"))]);
4609
4610 let err = run(
4612 &v,
4613 "MATCH (t:NoSuchLabel {id: $tid}) RETURN t",
4614 &BTreeMap::new(),
4615 )
4616 .expect_err("missing param must be Err, not Ok(empty)");
4617 assert!(
4618 err.contains("tid")
4619 && (err.contains("param") || err.contains("Param") || err.contains("missing")),
4620 "missing-param error must name the parameter, got: {err}"
4621 );
4622
4623 let err = run(
4624 &v,
4625 "MATCH (t:Talent) WHERE t.id = $tid RETURN t",
4626 &BTreeMap::new(),
4627 )
4628 .expect_err("missing WHERE param");
4629 assert!(err.contains("tid"), "got: {err}");
4630 }
4631
4632 #[test]
4633 fn order_by_none_last_then_skip_limit() {
4634 let mut fx = Fx::new();
4635 fx.add("Person", "ada", vec![("age", i(30))]);
4636 fx.add("Person", "bob", vec![]); fx.add("Person", "cam", vec![("age", i(10))]);
4638 fx.add("Person", "dan", vec![("age", i(20))]);
4639 let v = fx.view();
4640
4641 let asc = run(
4642 &v,
4643 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age",
4644 &BTreeMap::new(),
4645 )
4646 .unwrap();
4647 assert_eq!(
4648 rows_of(&asc),
4649 vec![
4650 vec![Some(s("cam")), Some(i(10))],
4651 vec![Some(s("dan")), Some(i(20))],
4652 vec![Some(s("ada")), Some(i(30))],
4653 vec![Some(s("bob")), None],
4654 ]
4655 );
4656
4657 let desc = run(
4658 &v,
4659 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age DESC",
4660 &BTreeMap::new(),
4661 )
4662 .unwrap();
4663 assert_eq!(
4664 rows_of(&desc),
4665 vec![
4666 vec![Some(s("ada")), Some(i(30))],
4667 vec![Some(s("dan")), Some(i(20))],
4668 vec![Some(s("cam")), Some(i(10))],
4669 vec![Some(s("bob")), None],
4670 ]
4671 );
4672
4673 let skip_lim = run(
4674 &v,
4675 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age SKIP 1 LIMIT 2",
4676 &BTreeMap::new(),
4677 )
4678 .unwrap();
4679 assert_eq!(
4680 rows_of(&skip_lim),
4681 vec![
4682 vec![Some(s("dan")), Some(i(20))],
4683 vec![Some(s("ada")), Some(i(30))],
4684 ]
4685 );
4686
4687 let desc_sl = run(
4688 &v,
4689 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age DESC SKIP 1 LIMIT 2",
4690 &BTreeMap::new(),
4691 )
4692 .unwrap();
4693 assert_eq!(
4694 rows_of(&desc_sl),
4695 vec![
4696 vec![Some(s("dan")), Some(i(20))],
4697 vec![Some(s("cam")), Some(i(10))],
4698 ]
4699 );
4700 }
4701
4702 #[test]
4703 fn unknown_label_and_etype_are_ok_empty() {
4704 let fx = hop_graph();
4705 let v = fx.view();
4706 let lab = run(&v, "MATCH (x:Nope) RETURN x", &BTreeMap::new()).expect("unknown label");
4707 assert!(lab.is_empty());
4708 let et = run(
4709 &v,
4710 "MATCH (a)-[:NO_SUCH_ETYPE]->(b) RETURN a",
4711 &BTreeMap::new(),
4712 )
4713 .expect("unknown etype");
4714 assert!(et.is_empty());
4715 }
4716
4717 #[test]
4718 fn execute_is_deterministic() {
4719 let fx = dogfood_graph();
4720 let v = fx.view();
4721 let p = tid_params();
4722 let a = run(&v, DOGFOOD, &p).expect("first");
4723 let b = run(&v, DOGFOOD, &p).expect("second");
4724 assert_eq!(a, b);
4725 let hop = hop_graph();
4726 let hv = hop.view();
4727 let q = "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b";
4728 assert_eq!(run(&hv, q, &BTreeMap::new()), run(&hv, q, &BTreeMap::new()));
4729 }
4730
4731 #[test]
4732 fn dogfood_pipeline_exact_rows() {
4733 let fx = dogfood_graph();
4734 let v = fx.view();
4735 let rs = run(&v, DOGFOOD, &tid_params()).expect("dogfood");
4736 assert_eq!(
4737 rs.columns(),
4738 &[
4739 "c".to_string(),
4740 "industry".to_string(),
4741 "specialty".to_string()
4742 ]
4743 );
4744 assert_eq!(
4746 rows_of(&rs),
4747 vec![
4748 vec![Some(s("acme")), Some(f(0.9)), Some(f(0.8))],
4749 vec![Some(s("zeta")), Some(f(0.9)), Some(f(0.6))],
4750 vec![Some(s("beta")), Some(f(0.6)), Some(f(0.7))],
4751 vec![Some(s("echo")), Some(f(0.5)), Some(f(0.5))],
4752 ]
4753 );
4754 }
4755
4756 #[test]
4757 fn unknown_var_in_op_is_err_not_panic() {
4758 let fx = hop_graph();
4759 let v = fx.view();
4760 let plan = vec![
4761 PlanOp::ScanLabel {
4762 var: "a".into(),
4763 label: None,
4764 },
4765 PlanOp::Expand {
4766 from: "zzz".into(),
4767 rel_var: Some("r".into()),
4768 etypes: vec![],
4769 dir: RelDir::Right,
4770 to: "b".into(),
4771 to_label: None,
4772 to_props: vec![],
4773 },
4774 PlanOp::Project {
4775 items: vec![RetItem {
4776 value: RetVal::Var("a".into()),
4777 alias: None,
4778 }],
4779 },
4780 ];
4781 let params = BTreeMap::new();
4782 let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
4783 execute(&v, &plan, &Params(¶ms))
4784 }));
4785 assert!(caught.is_ok(), "execute panicked on unknown var");
4786 let err = caught.unwrap().expect_err("unknown var must be Err");
4787 assert!(
4788 err.contains("zzz") && err.to_ascii_lowercase().contains("unbound"),
4789 "got: {err}"
4790 );
4791
4792 let join = vec![
4793 PlanOp::JoinBound {
4794 var: "ghost".into(),
4795 label: None,
4796 props: vec![],
4797 },
4798 PlanOp::Project {
4799 items: vec![RetItem {
4800 value: RetVal::Var("ghost".into()),
4801 alias: None,
4802 }],
4803 },
4804 ];
4805 let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
4806 execute(&v, &join, &Params(¶ms))
4807 }));
4808 assert!(caught.is_ok(), "execute panicked on JoinBound unknown var");
4809 assert!(caught.unwrap().is_err());
4810 }
4811
4812 #[test]
4813 fn dest_props_on_bound_expand_are_applied() {
4814 let fx = dogfood_graph();
4815 let v = fx.view();
4816 let rs = run(
4817 &v,
4818 "MATCH (t:Talent {id: $tid}) \
4819 MATCH (c:Company)-[r:INDUSTRY_ALIGNMENT]->(t:Talent {id: $tid}) \
4820 RETURN c",
4821 &tid_params(),
4822 )
4823 .unwrap();
4824 assert_eq!(
4826 col(&rs, "c"),
4827 vec![
4828 Some(s("acme")),
4829 Some(s("beta")),
4830 Some(s("gamma")),
4831 Some(s("delta")),
4832 Some(s("echo")),
4833 Some(s("foxtrot")),
4834 Some(s("zeta")),
4835 ]
4836 );
4837 let miss = run(
4838 &v,
4839 "MATCH (t:Talent {id: $tid}) \
4840 MATCH (c:Company)-[r:INDUSTRY_ALIGNMENT]->(t {id: 'nope'}) \
4841 RETURN c",
4842 &tid_params(),
4843 )
4844 .unwrap();
4845 assert!(miss.is_empty());
4846 }
4847
4848 #[test]
4849 fn missing_node_prop_projects_none_and_pattern_misses() {
4850 let mut fx = Fx::new();
4851 fx.add("Person", "ada", vec![("age", i(30))]);
4852 fx.add("Person", "bob", vec![]);
4853 let v = fx.view();
4854 let rs = run(&v, "MATCH (p:Person) RETURN p.age", &BTreeMap::new()).unwrap();
4855 assert_eq!(col(&rs, "p.age"), vec![Some(i(30)), None]);
4856 let pat = run(&v, "MATCH (p:Person {age: 30}) RETURN p", &BTreeMap::new()).unwrap();
4857 assert_eq!(col(&pat, "p"), vec![Some(s("ada"))]);
4858 }
4859
4860 #[test]
4861 fn unlabeled_match_does_not_project_sentinel_ghost_key() {
4862 let mut fx = Fx::new();
4863 fx.add("Person", "ada", vec![]);
4864 fx.ids.get_or_insert("ghost");
4867 fx.labels.resize(fx.ids.len(), u32::MAX);
4868 fx.add("Person", "bob", vec![]);
4869 let v = fx.view();
4870 let rs = run(&v, "MATCH (n) RETURN n", &BTreeMap::new()).expect("unlabeled scan");
4871 assert_eq!(col(&rs, "n"), vec![Some(s("ada")), Some(s("bob"))]);
4872 assert!(
4873 !rows_of(&rs)
4874 .iter()
4875 .any(|row| row.iter().any(|c| *c == Some(s("ghost")))),
4876 "sentinel slot must not project a ghost key"
4877 );
4878 }
4879
4880 #[test]
4881 fn execute_never_panics_on_hostile_plans() {
4882 let fx = hop_graph();
4883 let v = fx.view();
4884 let params = BTreeMap::new();
4885 let hostile = vec![
4886 vec![],
4887 vec![PlanOp::Project { items: vec![] }],
4888 vec![PlanOp::OrderBy {
4889 items: vec![OrderItem {
4890 target: OrderTarget::Alias("nope".into()),
4891 descending: false,
4892 }],
4893 }],
4894 vec![PlanOp::Filter {
4895 expr: crate::cypher::ast::Expr::Cmp {
4896 lhs: Operand::Prop {
4897 var: "missing".into(),
4898 field: "x".into(),
4899 },
4900 op: crate::filter::CmpOp::Eq,
4901 rhs: Operand::Lit(i(1)),
4902 },
4903 }],
4904 vec![
4905 PlanOp::ScanLabel {
4906 var: "a".into(),
4907 label: None,
4908 },
4909 PlanOp::LookupProps {
4910 var: "zzz".into(),
4911 props: vec![("k".into(), Operand::Lit(i(1)))],
4912 },
4913 ],
4914 vec![
4915 PlanOp::Skip(LimitSkip::Exact(99)),
4916 PlanOp::Limit(LimitSkip::Exact(0)),
4917 ],
4918 ];
4919 for plan in hostile {
4920 let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
4921 execute(&v, &plan, &Params(¶ms))
4922 }));
4923 assert!(caught.is_ok(), "execute panicked on hostile plan: {plan:?}");
4924 }
4925 }
4926
4927 #[test]
4928 fn unlabeled_and_labeled_scans_skip_tombstoned_ids() {
4929 let mut fx = Fx::new();
4930 let ada = fx.add("Person", "ada", vec![]);
4931 let bob = fx.add("Person", "bob", vec![]);
4932 fx.edge("KNOWS", ada, bob, vec![]);
4933 fx.ids.delete("ada");
4935 fx.labels[ada as usize] = u32::MAX;
4936 let knows = fx.syms.get("KNOWS").unwrap();
4937 fx.topo.remove_edge(knows, ada, bob);
4938 let v = fx.view();
4939
4940 let labeled = run(&v, "MATCH (p:Person) RETURN p", &BTreeMap::new()).unwrap();
4941 assert_eq!(col(&labeled, "p"), vec![Some(s("bob"))]);
4942 let unlabeled = run(&v, "MATCH (n) RETURN n", &BTreeMap::new()).unwrap();
4943 assert_eq!(col(&unlabeled, "n"), vec![Some(s("bob"))]);
4944 let hop = run(&v, "MATCH (x)-[:KNOWS]->(y) RETURN x, y", &BTreeMap::new()).unwrap();
4945 assert!(
4946 hop.is_empty(),
4947 "expand cannot yield edges to a deleted node once topology is swept"
4948 );
4949 }
4950
4951 proptest! {
4956 #[test]
4969 fn prop_bounded_equals_unbounded_slice(
4970 n_nodes in 2u32..10u32,
4971 edge_pairs in proptest::collection::vec(
4972 (any::<u32>(), any::<u32>()), 0..20usize
4973 ),
4974 recip_pairs in proptest::collection::vec(
4976 (any::<u32>(), any::<u32>()), 0..8usize
4977 ),
4978 n_hops in 1u32..4u32, use_filter in any::<bool>(),
4980 threshold in 0i64..8i64, limit in 1u64..10u64,
4982 skip in 0u64..4u64,
4983 ) {
4984 let mut fx = Fx::new();
4985 let mut node_ids = Vec::new();
4986 for idx in 0..n_nodes {
4987 let id = fx.add("N", &format!("n{idx}"), vec![("v", i(idx as i64 % 8))]);
4989 node_ids.push(id);
4990 }
4991 let n = node_ids.len();
4992
4993 for (si, di) in &edge_pairs {
4995 let si = (*si as usize) % n;
4996 let di = (*di as usize) % n;
4997 if si != di {
4998 fx.edge("T", node_ids[si], node_ids[di], vec![]);
4999 }
5000 }
5001 for (si, di) in &recip_pairs {
5004 let si = (*si as usize) % n;
5005 let di = (*di as usize) % n;
5006 if si != di {
5007 fx.edge("T", node_ids[si], node_ids[di], vec![]);
5008 fx.edge("T", node_ids[di], node_ids[si], vec![]);
5009 }
5010 }
5011
5012 let v = fx.view();
5013 let params = BTreeMap::new();
5014
5015 let var_names = ["a", "b", "c", "d"];
5018 let hop_count = n_hops as usize;
5019 let mut pattern = format!("({}:N)", var_names[0]);
5020 for h in 0..hop_count {
5021 pattern.push_str(&format!("-[:T]->({}", var_names[h + 1]));
5022 if h + 1 == hop_count {
5024 pattern.push_str(":N)");
5025 } else {
5026 pattern.push(')');
5027 }
5028 }
5029 let last_var = var_names[hop_count];
5030 let where_clause = if use_filter {
5031 format!(" WHERE {last_var}.v > {threshold}")
5032 } else {
5033 String::new()
5034 };
5035 let ret_vars: Vec<&str> = var_names[..=hop_count].to_vec();
5036 let ret_clause = ret_vars.join(", ");
5037
5038 let full_q = format!("MATCH {pattern}{where_clause} RETURN {ret_clause}");
5039 let bounded_q = format!(
5040 "MATCH {pattern}{where_clause} RETURN {ret_clause} SKIP {skip} LIMIT {limit}"
5041 );
5042
5043 let full_plan = compile(&full_q);
5044 let unbounded = super::with_max_intermediate_rows(100_000, || {
5047 super::execute_unbounded(&v, &full_plan, &Params(¶ms))
5048 });
5049 let unbounded = match unbounded {
5052 Ok(rs) => rs,
5053 Err(_) => return Ok(()),
5054 };
5055 let total = unbounded.len();
5056 let full_rows = rows_of(&unbounded);
5057
5058 let bounded = super::with_max_intermediate_rows(100_000, || {
5059 run(&v, &bounded_q, ¶ms)
5060 }).expect("bounded must not error");
5061
5062 let s = (skip as usize).min(total);
5063 let e = (skip as usize + limit as usize).min(total);
5064 prop_assert_eq!(
5065 rows_of(&bounded),
5066 full_rows[s..e].to_vec(),
5067 "hops={} filter={} threshold={} SKIP {} LIMIT {}: \
5068 bounded != unbounded[{}..{}]",
5069 hop_count, use_filter, threshold, skip, limit, s, e
5070 );
5071 }
5072 }
5073
5074 proptest! {
5075 #[test]
5093 fn prop_scan_filter_fused_equals_unbounded_slice(
5094 n_nodes in 0u32..20u32,
5095 prop_mask in any::<u32>(),
5097 float_mask in any::<u32>(),
5099 threshold in -1i64..8i64,
5102 op_idx in 0u32..6u32,
5104 skip in 0u64..5u64,
5105 limit in 1u64..8u64,
5106 ) {
5107 let op_str = match op_idx {
5108 0 => "=",
5109 1 => "<>",
5110 2 => "<",
5111 3 => "<=",
5112 4 => ">",
5113 _ => ">=",
5114 };
5115
5116 let mut fx = Fx::new();
5117 for idx in 0..n_nodes {
5118 let has_prop = (prop_mask >> (idx % 32)) & 1 == 1;
5119 let use_float = (float_mask >> (idx % 32)) & 1 == 1;
5120 let props: Vec<(&str, Value)> = if has_prop {
5122 if use_float {
5123 vec![("v", f(idx as f64 % 7.0))]
5124 } else {
5125 vec![("v", i(idx as i64 % 7))]
5126 }
5127 } else {
5128 vec![]
5129 };
5130 fx.add("N", &format!("n{idx}"), props);
5131 }
5132
5133 let v = fx.view();
5134 let params = BTreeMap::new();
5135
5136 let full_q = format!("MATCH (n:N) WHERE n.v {op_str} {threshold} RETURN n");
5138 let bounded_q = format!(
5139 "MATCH (n:N) WHERE n.v {op_str} {threshold} RETURN n SKIP {skip} LIMIT {limit}"
5140 );
5141
5142 let full_plan = compile(&full_q);
5143 let unbounded = super::execute_unbounded(&v, &full_plan, &Params(¶ms));
5144 let unbounded = match unbounded {
5145 Ok(rs) => rs,
5146 Err(_) => return Ok(()),
5147 };
5148 let total = unbounded.len();
5149 let full_rows = rows_of(&unbounded);
5150
5151 let fires_before =
5153 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
5154 let bounded = run(&v, &bounded_q, ¶ms).expect("fused bounded must not error");
5155 let fires_after =
5156 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
5157
5158 if n_nodes > 0 && op_idx != 0 {
5166 prop_assert!(
5167 fires_after > fires_before,
5168 "fused arm did NOT fire for op={} threshold={} n_nodes={}: \
5169 counter before={} after={}",
5170 op_str,
5171 threshold,
5172 n_nodes,
5173 fires_before,
5174 fires_after
5175 );
5176 }
5177
5178 let s = (skip as usize).min(total);
5179 let e = (skip as usize + limit as usize).min(total);
5180 prop_assert_eq!(
5181 rows_of(&bounded),
5182 full_rows[s..e].to_vec(),
5183 "fused path: op={} threshold={} n_nodes={} SKIP {} LIMIT {}: \
5184 bounded != unbounded[{}..{}]",
5185 op_str, threshold, n_nodes, skip, limit, s, e
5186 );
5187
5188 let compound_q = format!(
5192 "MATCH (n:N) WHERE n.v {} {} AND n.v >= -999 RETURN n SKIP {} LIMIT {}",
5193 op_str, threshold, skip, limit
5194 );
5195 let fires_before_compound =
5196 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
5197 let compound_bounded =
5198 run(&v, &compound_q, ¶ms).expect("compound-AND bounded must not error");
5199 let fires_after_compound =
5200 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
5201
5202 prop_assert_eq!(
5204 fires_after_compound,
5205 fires_before_compound,
5206 "fused arm fired for compound-AND shape (should use generic path): \
5207 op={} threshold={} n_nodes={}",
5208 op_str,
5209 threshold,
5210 n_nodes
5211 );
5212 prop_assert_eq!(
5213 rows_of(&compound_bounded),
5214 full_rows[s..e].to_vec(),
5215 "compound-AND fallback: op={} threshold={} n_nodes={} SKIP {} LIMIT {}: \
5216 result differs from fused",
5217 op_str, threshold, n_nodes, skip, limit
5218 );
5219 }
5220 }
5221
5222 #[test]
5233 fn dense_hop1_with_filter_survives_pull() {
5234 const LEAVES: usize = 120; const CAP: usize = 100;
5236
5237 let mut fx = Fx::new();
5238 let src = fx.add("Src", "src", vec![]);
5239 for idx in 0..LEAVES {
5240 let leaf = fx.add("Leaf", &format!("l{idx}"), vec![("v", i(idx as i64))]);
5241 fx.edge("T", src, leaf, vec![]);
5242 }
5243 let v = fx.view();
5244 let params = BTreeMap::new();
5245
5246 let staged_err = super::with_max_intermediate_rows(CAP, || {
5248 super::execute_unbounded(
5249 &v,
5250 &compile("MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v >= 110 RETURN l, l.v"),
5251 &Params(¶ms),
5252 )
5253 });
5254 assert!(
5255 staged_err.is_err(),
5256 "staged path must error on 120 leaves with cap={CAP}"
5257 );
5258 assert!(
5259 staged_err
5260 .unwrap_err()
5261 .contains("intermediate result exceeds"),
5262 "wrong error message"
5263 );
5264
5265 let ok = super::with_max_intermediate_rows(CAP, || {
5269 run(
5270 &v,
5271 "MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v >= 110 RETURN l, l.v LIMIT 5",
5272 ¶ms,
5273 )
5274 });
5275 let rs = ok.expect("pull-based must survive despite dense hop-1 exceeding cap");
5276 assert_eq!(rs.len(), 5, "LIMIT 5 must return exactly 5 rows");
5277
5278 let vs: Vec<i64> = (0..rs.len())
5280 .filter_map(|i| match rs.get(i, "l.v") {
5281 Some(Value::Int(n)) => Some(*n),
5282 _ => None,
5283 })
5284 .collect();
5285 assert_eq!(vs.len(), 5, "all projected rows must have v");
5286 for v_val in &vs {
5287 assert!(*v_val >= 110, "filter must hold: v={v_val} is not >= 110");
5288 }
5289
5290 let (pull_result, pull_produced) = super::with_expand_counter(|| {
5294 super::with_max_intermediate_rows(1_000_000, || {
5295 run(
5296 &v,
5297 "MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v < 10 RETURN l LIMIT 5",
5298 ¶ms,
5299 )
5300 })
5301 });
5302 pull_result.expect("pull must succeed without cap");
5303 assert!(
5305 pull_produced <= 5,
5306 "pull expand count {pull_produced} should be ≤ 5 (stops after 5 passing leaves)"
5307 );
5308
5309 let (staged_result, staged_produced) = super::with_expand_counter(|| {
5310 super::with_max_intermediate_rows(1_000_000, || {
5311 super::execute_unbounded(
5312 &v,
5313 &compile("MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v < 10 RETURN l"),
5314 &Params(¶ms),
5315 )
5316 })
5317 });
5318 staged_result.expect("staged must succeed with 1M cap");
5319 assert_eq!(
5320 staged_produced, LEAVES,
5321 "staged must expand all {LEAVES} leaves"
5322 );
5323
5324 assert!(
5325 staged_produced >= pull_produced * 10,
5326 "staged ({staged_produced}) must be ≥ 10× pull ({pull_produced})"
5327 );
5328 }
5329
5330 #[test]
5348 fn harness_shape_two_hop_dense_survives_pull() {
5349 const N_TALENT: usize = 70;
5350 const N_COMPANY: usize = 20;
5351 const N_INDUSTRY: usize = 3;
5352 const CAP: usize = 100;
5353 const LIMIT: usize = 10;
5354
5355 let mut fx = Fx::new();
5356
5357 let mut talent_ids: Vec<u32> = Vec::new();
5359 let mut talent_industry: Vec<usize> = Vec::new();
5360 for i in 0..N_TALENT {
5361 let ind = i % N_INDUSTRY;
5362 let id = fx.add(
5363 "Talent",
5364 &format!("t{i}"),
5365 vec![("industry", s(&ind.to_string()))],
5366 );
5367 talent_ids.push(id);
5368 talent_industry.push(ind);
5369 }
5370
5371 let mut company_ids: Vec<u32> = Vec::new();
5373 let mut company_industry: Vec<usize> = Vec::new();
5374 for i in 0..N_COMPANY {
5375 let ind = i % N_INDUSTRY;
5376 let id = fx.add(
5377 "Company",
5378 &format!("c{i}"),
5379 vec![("industry", s(&ind.to_string()))],
5380 );
5381 company_ids.push(id);
5382 company_industry.push(ind);
5383 }
5384
5385 for (ti, &tid) in talent_ids.iter().enumerate() {
5387 for (ci, &cid) in company_ids.iter().enumerate() {
5388 if talent_industry[ti] == company_industry[ci] {
5389 fx.edge("INDUSTRY_ALIGNMENT", tid, cid, vec![]);
5390 }
5391 }
5392 }
5393
5394 let v = fx.view();
5395 let params = BTreeMap::new();
5396 let query = format!(
5397 "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company)\
5398 <-[:INDUSTRY_ALIGNMENT]-(t2:Talent) RETURN t, c, t2 LIMIT {LIMIT}"
5399 );
5400
5401 const UNBOUNDED_Q: &str = "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company)\
5403 <-[:INDUSTRY_ALIGNMENT]-(t2:Talent) RETURN t, c, t2";
5404 let staged_err = super::with_max_intermediate_rows(CAP, || {
5405 super::execute_unbounded(&v, &compile(UNBOUNDED_Q), &Params(¶ms))
5406 });
5407 assert!(
5408 staged_err.is_err(),
5409 "staged must error with cap={CAP} on harness-shape graph"
5410 );
5411 assert!(
5412 staged_err
5413 .unwrap_err()
5414 .contains("intermediate result exceeds"),
5415 "wrong error"
5416 );
5417
5418 let ok = super::with_max_intermediate_rows(CAP, || run(&v, &query, ¶ms));
5420 let rs = ok.expect("pull-based must complete on harness-shape with LIMIT 10");
5421 assert_eq!(rs.len(), LIMIT, "must return exactly {LIMIT} rows");
5422
5423 for i in 0..rs.len() {
5428 let row = rs.row(i);
5429 assert_eq!(row.len(), 3, "each row must have 3 columns (t, c, t2)");
5430 assert!(row.iter().all(|c| c.is_some()), "all cells must be Some");
5431 }
5432 }
5433
5434 #[test]
5435 fn intermediate_row_cap_errors_on_scan_and_expand() {
5436 let cap_msg = |n: usize| {
5437 format!(
5438 "intermediate result exceeds {n} rows; add a LIMIT or constrain patterns with shared variables"
5439 )
5440 };
5441
5442 let mut scan_fx = Fx::new();
5443 scan_fx.add("N", "a", vec![]);
5444 scan_fx.add("N", "b", vec![]);
5445 scan_fx.add("N", "c", vec![]);
5446 let sv = scan_fx.view();
5447 let scan_err = super::with_max_intermediate_rows(2, || {
5448 run(&sv, "MATCH (n:N) RETURN n", &BTreeMap::new())
5449 })
5450 .expect_err("3-row scan must exceed cap 2");
5451 assert_eq!(scan_err, cap_msg(2));
5452
5453 let mut exp_fx = Fx::new();
5454 let src = exp_fx.add("Src", "s", vec![]);
5455 let d1 = exp_fx.add("Dst", "d1", vec![]);
5456 let d2 = exp_fx.add("Dst", "d2", vec![]);
5457 exp_fx.edge("T", src, d1, vec![]);
5458 exp_fx.edge("T", src, d2, vec![]);
5459 let ev = exp_fx.view();
5460 let exp_err = super::with_max_intermediate_rows(1, || {
5462 run(&ev, "MATCH (x:Src)-[:T]->(y) RETURN x, y", &BTreeMap::new())
5463 })
5464 .expect_err("2-row expand must exceed cap 1");
5465 assert_eq!(exp_err, cap_msg(1));
5466 }
5467
5468 #[test]
5478 fn bounded_matches_unbounded_slice_various_limits() {
5479 let fx = hop_graph();
5481 let v = fx.view();
5482 let params = BTreeMap::new();
5483
5484 let full_plan = compile("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b");
5486 let full_rs = super::execute_unbounded(&v, &full_plan, &Params(¶ms)).unwrap();
5487 let full_rows = rows_of(&full_rs);
5488 assert_eq!(full_rows.len(), 3, "hop_graph has exactly 3 KNOWS paths");
5489
5490 for limit in [1u64, 2, 3, 10] {
5491 let q = format!("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b LIMIT {limit}");
5492 let rs = run(&v, &q, ¶ms).unwrap();
5493 let expected_len = (limit as usize).min(full_rows.len());
5494 assert_eq!(
5495 rs.len(),
5496 expected_len,
5497 "LIMIT {limit}: expected {expected_len} rows, got {}",
5498 rs.len()
5499 );
5500 assert_eq!(
5501 rows_of(&rs),
5502 full_rows[..expected_len],
5503 "LIMIT {limit}: rows differ from reference slice"
5504 );
5505 }
5506
5507 let skip_rs = run(
5509 &v,
5510 "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b SKIP 1 LIMIT 2",
5511 ¶ms,
5512 )
5513 .unwrap();
5514 assert_eq!(rows_of(&skip_rs), full_rows[1..3]);
5515
5516 let mut fx2 = Fx::new();
5518 fx2.add("N", "a", vec![("v", i(1))]);
5519 fx2.add("N", "b", vec![("v", i(2))]);
5520 fx2.add("N", "c", vec![("v", i(3))]);
5521 let v2 = fx2.view();
5522 let filter_full = super::execute_unbounded(
5524 &v2,
5525 &compile("MATCH (n:N) WHERE n.v > 1 RETURN n"),
5526 &Params(¶ms),
5527 )
5528 .unwrap();
5529 assert_eq!(filter_full.len(), 2);
5530
5531 let filter_lim = run(&v2, "MATCH (n:N) WHERE n.v > 1 RETURN n LIMIT 1", ¶ms).unwrap();
5532 assert_eq!(filter_lim.len(), 1, "LIMIT 1 on filter query");
5533 assert_eq!(rows_of(&filter_lim), rows_of(&filter_full)[..1]);
5534
5535 let tri = triangle();
5537 let tv = tri.view();
5538 let tri_full = super::execute_unbounded(
5540 &tv,
5541 &compile("MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z"),
5542 &Params(¶ms),
5543 )
5544 .unwrap();
5545 assert_eq!(tri_full.len(), 3, "triangle has 3 unique two-hop paths");
5546 for limit in [1u64, 2, 3, 5] {
5547 let q = format!("MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z LIMIT {limit}");
5548 let rs = run(&tv, &q, ¶ms).unwrap();
5549 let expected_len = (limit as usize).min(3);
5550 assert_eq!(
5551 rs.len(),
5552 expected_len,
5553 "triangle LIMIT {limit}: got {} rows",
5554 rs.len()
5555 );
5556 assert_eq!(
5557 rows_of(&rs),
5558 rows_of(&tri_full)[..expected_len],
5559 "triangle LIMIT {limit}: rows differ"
5560 );
5561 }
5562 }
5563
5564 #[test]
5568 fn expand_terminates_early_with_row_bound() {
5569 const LEAVES: usize = 500;
5570 let mut fx = Fx::new();
5571 let hub = fx.add("Hub", "hub", vec![]);
5572 for i in 0..LEAVES {
5573 let leaf = fx.add("Leaf", &format!("leaf-{i}"), vec![]);
5574 fx.edge("T", hub, leaf, vec![]);
5575 }
5576 let v = fx.view();
5577 let params = BTreeMap::new();
5578 let full_plan = compile("MATCH (h:Hub)-[:T]->(x:Leaf) RETURN x");
5579
5580 let (bounded_result, bounded_produced) = super::with_expand_counter(|| {
5582 run(&v, "MATCH (h:Hub)-[:T]->(x:Leaf) RETURN x LIMIT 5", ¶ms)
5583 });
5584 let bounded_rs = bounded_result.unwrap();
5585 assert_eq!(bounded_rs.len(), 5, "LIMIT 5 must return exactly 5 rows");
5586 assert!(
5587 bounded_produced <= 5,
5588 "bounded: exec_expand emitted {bounded_produced} rows, expected ≤ 5"
5589 );
5590
5591 let (unbounded_result, unbounded_produced) = super::with_expand_counter(|| {
5593 super::execute_unbounded(&v, &full_plan, &Params(¶ms))
5594 });
5595 let unbounded_rs = unbounded_result.unwrap();
5596 assert_eq!(
5597 unbounded_rs.len(),
5598 LEAVES,
5599 "unbounded must return all {LEAVES} rows"
5600 );
5601 assert_eq!(
5602 unbounded_produced, LEAVES,
5603 "unbounded: exec_expand must emit all {LEAVES} rows"
5604 );
5605
5606 assert!(
5608 unbounded_produced >= bounded_produced * 100,
5609 "unbounded ({unbounded_produced}) must be ≥ 100× bounded ({bounded_produced})"
5610 );
5611 }
5612
5613 #[test]
5619 fn bounded_query_survives_low_intermediate_row_cap() {
5620 let mut fx = Fx::new();
5622 let src = fx.add("Src", "s", vec![]);
5623 for i in 0..20usize {
5624 let dst = fx.add("Dst", &format!("d{i}"), vec![]);
5625 fx.edge("T", src, dst, vec![]);
5626 }
5627 let v = fx.view();
5628 let params = BTreeMap::new();
5629 let full_plan = compile("MATCH (s:Src)-[:T]->(d:Dst) RETURN d");
5630
5631 let cap_err = super::with_max_intermediate_rows(10, || {
5633 super::execute_unbounded(&v, &full_plan, &Params(¶ms))
5634 });
5635 assert!(
5636 cap_err.is_err(),
5637 "unbounded must hit the intermediate-row cap"
5638 );
5639 assert!(
5640 cap_err.unwrap_err().contains("intermediate result exceeds"),
5641 "cap error message must be the budget message"
5642 );
5643
5644 let ok = super::with_max_intermediate_rows(10, || {
5646 run(&v, "MATCH (s:Src)-[:T]->(d:Dst) RETURN d LIMIT 5", ¶ms)
5647 });
5648 assert_eq!(
5649 ok.unwrap().len(),
5650 5,
5651 "bounded (LIMIT 5) must complete with 5 rows, not a cap error"
5652 );
5653
5654 let at_cap = super::with_max_intermediate_rows(10, || {
5656 run(&v, "MATCH (s:Src)-[:T]->(d:Dst) RETURN d LIMIT 10", ¶ms)
5657 });
5658 assert_eq!(
5659 at_cap.unwrap().len(),
5660 10,
5661 "bounded at LIMIT==cap must complete with 10 rows"
5662 );
5663 }
5664
5665 #[test]
5670 fn count_star_returns_total_node_count() {
5671 let mut fx = Fx::new();
5672 fx.add("Person", "ada", vec![]);
5673 fx.add("Person", "bob", vec![]);
5674 fx.add("Person", "cam", vec![]);
5675 let v = fx.view();
5676 let params = BTreeMap::new();
5677
5678 let rs = run(&v, "MATCH (n:Person) RETURN COUNT(*)", ¶ms).expect("COUNT(*)");
5679 assert_eq!(rs.columns(), &["COUNT(*)".to_string()]);
5680 assert_eq!(rs.len(), 1);
5681 assert_eq!(rs.row(0), &[Some(i(3))]);
5682
5683 let rs_empty = run(&v, "MATCH (n:Ghost) RETURN COUNT(*)", ¶ms).expect("COUNT(*) empty");
5685 assert_eq!(rs_empty.row(0), &[Some(i(0))]);
5686 }
5687
5688 #[test]
5689 fn count_star_alias_sets_column_name() {
5690 let mut fx = Fx::new();
5691 fx.add("N", "a", vec![]);
5692 let v = fx.view();
5693 let params = BTreeMap::new();
5694 let rs = run(&v, "MATCH (n:N) RETURN COUNT(*) AS total", ¶ms).expect("COUNT AS");
5695 assert_eq!(rs.columns(), &["total".to_string()]);
5696 assert_eq!(rs.row(0), &[Some(i(1))]);
5697 }
5698
5699 #[test]
5700 fn count_var_skips_null_node_bindings() {
5701 let mut fx = Fx::new();
5704 fx.add("N", "a", vec![]);
5705 fx.add("N", "b", vec![]);
5706 let v = fx.view();
5707 let params = BTreeMap::new();
5708 let rs = run(&v, "MATCH (n:N) RETURN COUNT(n)", ¶ms).expect("COUNT(n)");
5709 assert_eq!(rs.columns(), &["COUNT(n)".to_string()]);
5710 assert_eq!(rs.row(0), &[Some(i(2))]);
5711 }
5712
5713 #[test]
5714 fn sum_numeric_prop_ignores_null_and_non_numeric() {
5715 let mut fx = Fx::new();
5716 fx.add("N", "a", vec![("v", i(10))]);
5717 fx.add("N", "b", vec![("v", i(20))]);
5718 fx.add("N", "c", vec![]); let v = fx.view();
5720 let params = BTreeMap::new();
5721 let rs = run(&v, "MATCH (n:N) RETURN SUM(n.v)", ¶ms).expect("SUM");
5722 assert_eq!(rs.columns(), &["SUM(n.v)".to_string()]);
5723 assert_eq!(rs.row(0), &[Some(f(30.0))]);
5725
5726 let rs_null = run(&v, "MATCH (n:N) RETURN SUM(n.missing)", ¶ms).expect("SUM null");
5728 assert_eq!(rs_null.row(0), &[None]);
5729 }
5730
5731 #[test]
5732 fn avg_numeric_prop() {
5733 let mut fx = Fx::new();
5734 fx.add("N", "a", vec![("v", i(10))]);
5735 fx.add("N", "b", vec![("v", i(30))]);
5736 let v = fx.view();
5737 let params = BTreeMap::new();
5738 let rs = run(&v, "MATCH (n:N) RETURN AVG(n.v) AS avg_v", ¶ms).expect("AVG");
5739 assert_eq!(rs.columns(), &["avg_v".to_string()]);
5740 assert_eq!(rs.row(0), &[Some(f(20.0))]);
5742
5743 let rs_empty = run(&v, "MATCH (n:Ghost) RETURN AVG(n.v)", ¶ms).expect("AVG empty");
5745 assert_eq!(rs_empty.row(0), &[None]);
5746 }
5747
5748 #[test]
5749 fn min_max_numeric_prop() {
5750 let mut fx = Fx::new();
5751 fx.add("N", "a", vec![("v", i(5))]);
5752 fx.add("N", "b", vec![("v", i(1))]);
5753 fx.add("N", "c", vec![("v", i(9))]);
5754 fx.add("N", "d", vec![]); let v = fx.view();
5756 let params = BTreeMap::new();
5757
5758 let min_rs = run(&v, "MATCH (n:N) RETURN MIN(n.v)", ¶ms).expect("MIN");
5759 assert_eq!(min_rs.row(0), &[Some(i(1))]);
5760
5761 let max_rs = run(&v, "MATCH (n:N) RETURN MAX(n.v)", ¶ms).expect("MAX");
5762 assert_eq!(max_rs.row(0), &[Some(i(9))]);
5763 }
5764
5765 #[test]
5768 fn min_max_mixed_int_float_props() {
5769 let mut fx = Fx::new();
5770 fx.add("N", "a", vec![("v", i(3))]);
5772 fx.add("N", "b", vec![("v", f(1.5))]);
5773 fx.add("N", "c", vec![("v", i(7))]);
5774 fx.add("N", "d", vec![("v", f(2.0))]);
5775 let v = fx.view();
5776 let params = BTreeMap::new();
5777
5778 let min_rs = run(&v, "MATCH (n:N) RETURN MIN(n.v)", ¶ms).expect("MIN mixed");
5779 assert_eq!(min_rs.row(0), &[Some(f(1.5))]);
5781
5782 let max_rs = run(&v, "MATCH (n:N) RETURN MAX(n.v)", ¶ms).expect("MAX mixed");
5783 assert_eq!(max_rs.row(0), &[Some(i(7))]);
5785 }
5786
5787 #[test]
5790 fn aggregate_limit_skip_order_by_are_no_ops() {
5791 let mut fx = Fx::new();
5792 fx.add("N", "a", vec![]);
5793 fx.add("N", "b", vec![]);
5794 fx.add("N", "c", vec![]);
5795 let v = fx.view();
5796 let params = BTreeMap::new();
5797
5798 let rs_lim5 =
5800 run(&v, "MATCH (n:N) RETURN COUNT(*) LIMIT 5", ¶ms).expect("COUNT(*) LIMIT 5");
5801 assert_eq!(
5802 rs_lim5.len(),
5803 1,
5804 "aggregate with LIMIT 5 must still return 1 row"
5805 );
5806 assert_eq!(rs_lim5.row(0), &[Some(i(3))]);
5807
5808 let rs_lim0 =
5810 run(&v, "MATCH (n:N) RETURN COUNT(*) LIMIT 0", ¶ms).expect("COUNT(*) LIMIT 0");
5811 assert_eq!(
5812 rs_lim0.len(),
5813 1,
5814 "aggregate with LIMIT 0 must still return 1 row"
5815 );
5816 assert_eq!(rs_lim0.row(0), &[Some(i(3))]);
5817
5818 let rs_skip =
5820 run(&v, "MATCH (n:N) RETURN COUNT(*) SKIP 100", ¶ms).expect("COUNT(*) SKIP 100");
5821 assert_eq!(
5822 rs_skip.len(),
5823 1,
5824 "aggregate with large SKIP must still return 1 row"
5825 );
5826
5827 let rs_ord = plan_src("MATCH (n:N) RETURN COUNT(*) ORDER BY n");
5831 assert!(
5834 rs_ord.is_ok(),
5835 "COUNT(*) ORDER BY should plan without error (ORDER BY dropped)"
5836 );
5837 let plan_ops = rs_ord.unwrap();
5838 assert!(
5840 !plan_ops
5841 .iter()
5842 .any(|op| matches!(op, crate::cypher::plan::PlanOp::OrderBy { .. })),
5843 "aggregate plan must not contain OrderBy"
5844 );
5845 }
5846
5847 #[test]
5848 fn count_star_no_budget_cap_applies() {
5849 let mut fx = Fx::new();
5852 let src = fx.add("Src", "s", vec![]);
5853 for i in 0..30usize {
5854 let dst = fx.add("Dst", &format!("d{i}"), vec![]);
5855 fx.edge("T", src, dst, vec![]);
5856 }
5857 let v = fx.view();
5858 let params = BTreeMap::new();
5859
5860 let cap_err =
5862 super::with_max_intermediate_rows(10, || run(&v, "MATCH (n:Dst) RETURN n", ¶ms));
5863 assert!(
5864 cap_err.is_err(),
5865 "staged path must error on 30 nodes with cap=10"
5866 );
5867
5868 let agg_ok = super::with_max_intermediate_rows(10, || {
5870 run(&v, "MATCH (n:Dst) RETURN COUNT(*)", ¶ms)
5871 })
5872 .expect("aggregate must not hit the intermediate-row cap");
5873 assert_eq!(
5874 agg_ok.row(0),
5875 &[Some(i(30))],
5876 "COUNT(*) must count all 30 nodes regardless of cap"
5877 );
5878 }
5879
5880 fn plan_src(src: &str) -> Result<Vec<crate::cypher::plan::PlanOp>, String> {
5881 use crate::cypher::{lex, parse, plan};
5882 let toks = lex(src).map_err(|e| format!("lex: {e}"))?;
5883 let ast = parse(&toks).map_err(|e| format!("parse: {e}"))?;
5884 plan(&ast).map_err(|e| format!("plan: {e}"))
5885 }
5886
5887 #[test]
5890 fn grouped_aggregation_plan_routing() {
5891 use crate::cypher::plan::PlanOp;
5892
5893 let ops = plan_src("MATCH (a:N) RETURN a, COUNT(*)")
5895 .expect("grouped aggregation must now succeed");
5896 assert!(
5897 ops.iter()
5898 .any(|op| matches!(op, PlanOp::GroupAggregate { .. })),
5899 "grouped aggregation plan must contain GroupAggregate op, got: {ops:?}"
5900 );
5901
5902 let ops2 = plan_src("MATCH (a:N) RETURN COUNT(*), COUNT(a)")
5904 .expect("multi-aggregate must now succeed");
5905 assert!(
5906 ops2.iter()
5907 .any(|op| matches!(op, PlanOp::GroupAggregate { .. })),
5908 "multi-aggregate plan must contain GroupAggregate op, got: {ops2:?}"
5909 );
5910
5911 let err3 = plan_src("MATCH (a:N) RETURN SUM(*)").expect_err("SUM(*) must be plan error");
5913 assert!(
5914 err3.to_ascii_lowercase().contains("sum") || err3.to_ascii_lowercase().contains("*"),
5915 "error must mention SUM or *, got: {err3}"
5916 );
5917 }
5918
5919 #[test]
5922 fn grouped_single_key_count() {
5923 let mut fx = Fx::new();
5925 fx.add("N", "a", vec![("t", s("X"))]);
5926 fx.add("N", "b", vec![("t", s("X"))]);
5927 fx.add("N", "c", vec![("t", s("Y"))]);
5928 let v = fx.view();
5929 let params = BTreeMap::new();
5930
5931 let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
5932 .expect("single-key grouped COUNT must succeed");
5933 assert_eq!(
5934 rs.columns(),
5935 &["n.t".to_string(), "cnt".to_string()],
5936 "columns must match RETURN clause"
5937 );
5938 assert_eq!(rs.len(), 2, "must produce exactly 2 groups (X and Y)");
5939
5940 let find = |label: &Value| (0..rs.len()).find(|&i| rs.row(i)[0].as_ref() == Some(label));
5942 let xi = find(&s("X")).expect("group X must exist");
5943 let yi = find(&s("Y")).expect("group Y must exist");
5944 assert_eq!(rs.row(xi)[1], Some(i(2)), "X group count must be 2");
5945 assert_eq!(rs.row(yi)[1], Some(i(1)), "Y group count must be 1");
5946 }
5947
5948 #[test]
5949 fn grouped_two_keys_sum_avg() {
5950 let mut fx = Fx::new();
5952 fx.add(
5953 "N",
5954 "a",
5955 vec![("cat", s("A")), ("sub", s("1")), ("v", i(10))],
5956 );
5957 fx.add(
5958 "N",
5959 "b",
5960 vec![("cat", s("A")), ("sub", s("1")), ("v", i(20))],
5961 );
5962 fx.add(
5963 "N",
5964 "c",
5965 vec![("cat", s("A")), ("sub", s("2")), ("v", i(5))],
5966 );
5967 fx.add(
5968 "N",
5969 "d",
5970 vec![("cat", s("B")), ("sub", s("1")), ("v", i(100))],
5971 );
5972 let v = fx.view();
5973 let params = BTreeMap::new();
5974
5975 let rs = run(
5976 &v,
5977 "MATCH (n:N) RETURN n.cat, n.sub, SUM(n.v) AS total, AVG(n.v) AS avg_v",
5978 ¶ms,
5979 )
5980 .expect("two-key SUM + AVG must succeed");
5981 assert_eq!(
5982 rs.columns(),
5983 &[
5984 "n.cat".to_string(),
5985 "n.sub".to_string(),
5986 "total".to_string(),
5987 "avg_v".to_string()
5988 ]
5989 );
5990 assert_eq!(rs.len(), 3, "must produce 3 groups: (A,1), (A,2), (B,1)");
5991
5992 let find = |cat: &Value, sub: &Value| {
5993 (0..rs.len())
5994 .find(|&i| rs.row(i)[0].as_ref() == Some(cat) && rs.row(i)[1].as_ref() == Some(sub))
5995 };
5996 let a1 = find(&s("A"), &s("1")).expect("group (A,1) must exist");
5997 assert_eq!(rs.row(a1)[2], Some(f(30.0)), "(A,1) SUM must be 30.0");
5998 assert_eq!(rs.row(a1)[3], Some(f(15.0)), "(A,1) AVG must be 15.0");
5999
6000 let a2 = find(&s("A"), &s("2")).expect("group (A,2) must exist");
6001 assert_eq!(rs.row(a2)[2], Some(f(5.0)), "(A,2) SUM must be 5.0");
6002
6003 let b1 = find(&s("B"), &s("1")).expect("group (B,1) must exist");
6004 assert_eq!(rs.row(b1)[2], Some(f(100.0)), "(B,1) SUM must be 100.0");
6005 assert_eq!(rs.row(b1)[3], Some(f(100.0)), "(B,1) AVG must be 100.0");
6006 }
6007
6008 #[test]
6009 fn grouped_order_by_count_desc_limit() {
6010 let mut fx = Fx::new();
6012 fx.add("N", "a1", vec![("cat", s("A"))]);
6013 fx.add("N", "a2", vec![("cat", s("A"))]);
6014 fx.add("N", "a3", vec![("cat", s("A"))]);
6015 fx.add("N", "b1", vec![("cat", s("B"))]);
6016 fx.add("N", "b2", vec![("cat", s("B"))]);
6017 fx.add("N", "c1", vec![("cat", s("C"))]);
6018 fx.add("N", "d1", vec![("cat", s("D"))]);
6019 fx.add("N", "d2", vec![("cat", s("D"))]);
6020 fx.add("N", "d3", vec![("cat", s("D"))]);
6021 fx.add("N", "d4", vec![("cat", s("D"))]);
6022 fx.add("N", "e1", vec![("cat", s("E"))]);
6023 let v = fx.view();
6024 let params = BTreeMap::new();
6025
6026 let rs = run(
6027 &v,
6028 "MATCH (n:N) RETURN n.cat, COUNT(*) AS cnt ORDER BY cnt DESC LIMIT 3",
6029 ¶ms,
6030 )
6031 .expect("ORDER BY count DESC LIMIT 3 must succeed");
6032 assert_eq!(rs.len(), 3, "LIMIT 3 must return exactly 3 groups");
6033 assert_eq!(rs.row(0)[1], Some(i(4)), "row 0 must be count 4");
6035 assert_eq!(rs.row(0)[0], Some(s("D")), "row 0 must be category D");
6036 assert_eq!(rs.row(1)[1], Some(i(3)), "row 1 must be count 3");
6037 assert_eq!(rs.row(1)[0], Some(s("A")), "row 1 must be category A");
6038 assert_eq!(rs.row(2)[1], Some(i(2)), "row 2 must be count 2");
6039 assert_eq!(rs.row(2)[0], Some(s("B")), "row 2 must be category B");
6040
6041 let plan_ops =
6043 plan_src("MATCH (n:N) RETURN n.cat, COUNT(*) AS cnt ORDER BY cnt DESC LIMIT 3")
6044 .expect("plan must succeed");
6045 assert_eq!(
6046 crate::cypher::plan::row_bound(&plan_ops),
6047 None,
6048 "GroupAggregate plan with LIMIT must have row_bound = None"
6049 );
6050 }
6051
6052 #[test]
6053 fn grouped_empty_input_yields_zero_groups() {
6054 let fx = Fx::new(); let v = fx.view();
6056 let params = BTreeMap::new();
6057
6058 let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
6059 .expect("grouped aggregate on empty graph must succeed");
6060 assert_eq!(rs.len(), 0, "empty input must yield zero groups");
6061 assert_eq!(
6062 rs.columns(),
6063 &["n.t".to_string(), "cnt".to_string()],
6064 "columns must still be present even with zero rows"
6065 );
6066 }
6067
6068 #[test]
6069 fn grouped_null_key_groups_together() {
6070 let mut fx = Fx::new();
6072 fx.add("N", "a", vec![("t", s("X"))]);
6073 fx.add("N", "b", vec![]); fx.add("N", "c", vec![]); fx.add("N", "d", vec![("t", s("Y"))]);
6076 let v = fx.view();
6077 let params = BTreeMap::new();
6078
6079 let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
6080 .expect("null-key grouped aggregate must succeed");
6081 assert_eq!(rs.len(), 3, "must produce 3 groups: X, null, Y");
6082
6083 let null_row = (0..rs.len())
6085 .find(|&i| rs.row(i)[0].is_none())
6086 .expect("null group must be present");
6087 assert_eq!(
6088 rs.row(null_row)[1],
6089 Some(i(2)),
6090 "null group must count 2 rows (b and c)"
6091 );
6092
6093 let x_row = (0..rs.len())
6094 .find(|&i| rs.row(i)[0] == Some(s("X")))
6095 .expect("X group must exist");
6096 assert_eq!(rs.row(x_row)[1], Some(i(1)));
6097
6098 let y_row = (0..rs.len())
6099 .find(|&i| rs.row(i)[0] == Some(s("Y")))
6100 .expect("Y group must exist");
6101 assert_eq!(rs.row(y_row)[1], Some(i(1)));
6102 }
6103
6104 #[test]
6105 fn grouped_cap_error_on_high_cardinality() {
6106 let mut fx = Fx::new();
6109 fx.add("N", "a", vec![("t", s("A"))]);
6110 fx.add("N", "b", vec![("t", s("B"))]);
6111 fx.add("N", "c", vec![("t", s("C"))]);
6112 let v = fx.view();
6113 let params = BTreeMap::new();
6114
6115 let err = super::with_max_groups(2, || {
6116 run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
6117 })
6118 .expect_err("must error when group count exceeds cap");
6119 assert!(
6120 err.to_ascii_lowercase().contains("group count"),
6121 "error must mention group count, got: {err}"
6122 );
6123 }
6124
6125 #[test]
6126 fn multi_aggregate_no_keys() {
6127 let mut fx = Fx::new();
6130 fx.add("N", "a", vec![]);
6131 fx.add("N", "b", vec![]);
6132 let v = fx.view();
6133 let params = BTreeMap::new();
6134
6135 let rs = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
6136 .expect("multi-aggregate no keys must succeed");
6137 assert_eq!(rs.len(), 1, "must produce exactly one result row");
6138 assert_eq!(rs.row(0)[0], Some(i(2)), "COUNT(*) must be 2");
6139 assert_eq!(rs.row(0)[1], Some(i(2)), "COUNT(n) must be 2");
6140 }
6141
6142 #[test]
6143 fn aggregate_over_hop_counts_edges() {
6144 let fx = hop_graph();
6145 let v = fx.view();
6146 let params = BTreeMap::new();
6147 let rs = run(
6149 &v,
6150 "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN COUNT(*)",
6151 ¶ms,
6152 )
6153 .expect("COUNT(*) on hop graph");
6154 assert_eq!(rs.row(0), &[Some(i(3))]);
6155 }
6156
6157 #[test]
6162 fn multi_aggregate_no_keys_empty_graph() {
6163 let fx = Fx::new(); let v = fx.view();
6169 let params = BTreeMap::new();
6170
6171 let rs = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
6172 .expect("empty-graph multi-agg must succeed");
6173 assert_eq!(rs.len(), 1, "must produce exactly 1 row on empty input");
6174 assert_eq!(
6175 rs.row(0)[0],
6176 Some(i(0)),
6177 "COUNT(*) on empty graph must be 0"
6178 );
6179 assert_eq!(
6180 rs.row(0)[1],
6181 Some(i(0)),
6182 "COUNT(n) on empty graph must be 0"
6183 );
6184 }
6185
6186 #[test]
6187 fn fast_path_and_group_path_agree_on_empty_input() {
6188 let fx = Fx::new();
6194 let v = fx.view();
6195 let params = BTreeMap::new();
6196
6197 let fast = run(&v, "MATCH (n:N) RETURN COUNT(*)", ¶ms)
6198 .expect("fast-path COUNT on empty graph");
6199 assert_eq!(fast.len(), 1, "fast path: 1 row on empty input");
6200 let fast_count = fast.row(0)[0].clone();
6201
6202 let grouped = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
6203 .expect("group-path COUNT on empty graph");
6204 assert_eq!(grouped.len(), 1, "group path: 1 row on empty input");
6205 let group_count = grouped.row(0)[0].clone();
6206
6207 assert_eq!(
6208 fast_count, group_count,
6209 "fast path and group path must agree on COUNT(*) for empty input"
6210 );
6211
6212 let mut fx2 = Fx::new();
6214 fx2.add("N", "x", vec![]);
6215 fx2.add("N", "y", vec![]);
6216 let v2 = fx2.view();
6217
6218 let fast2 = run(&v2, "MATCH (n:N) RETURN COUNT(*)", ¶ms)
6219 .expect("fast-path COUNT on 2-node graph");
6220 assert_eq!(fast2.row(0)[0], Some(i(2)), "fast path: COUNT(*) = 2");
6221
6222 let grouped2 = run(&v2, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
6223 .expect("group-path COUNT on 2-node graph");
6224 assert_eq!(grouped2.row(0)[0], Some(i(2)), "group path: COUNT(*) = 2");
6225 assert_eq!(grouped2.row(0)[1], Some(i(2)), "group path: COUNT(n) = 2");
6226 }
6227
6228 #[test]
6233 fn grouped_int_float_key_unification() {
6234 let mut fx = Fx::new();
6241 fx.add("N", "a", vec![("score", i(1))]); fx.add("N", "b", vec![("score", f(1.0))]);
6243 let v = fx.view();
6244 let params = BTreeMap::new();
6245
6246 let rs = run(&v, "MATCH (n:N) RETURN n.score, COUNT(*) AS cnt", ¶ms)
6247 .expect("int/float unification must succeed");
6248 assert_eq!(
6249 rs.len(),
6250 1,
6251 "Int(1) and Float(1.0) must group together into 1 group"
6252 );
6253 assert_eq!(
6255 rs.row(0)[0],
6256 Some(i(1)),
6257 "key column must display Int(1) (first-seen)"
6258 );
6259 assert_eq!(rs.row(0)[1], Some(i(2)), "unified group must have count=2");
6260 }
6261
6262 #[test]
6263 fn distinct_collapses_duplicate_projected_values() {
6264 let mut fx = Fx::new();
6265 fx.add("N", "a1", vec![("city", s("Austin"))]);
6266 fx.add("N", "a2", vec![("city", s("Austin"))]);
6267 let v = fx.view();
6268 let params = BTreeMap::new();
6269 let rs = run(&v, "MATCH (n:N) RETURN DISTINCT n.city", ¶ms).unwrap();
6270 assert_eq!(rs.len(), 1);
6271 assert_eq!(rs.row(0)[0], Some(s("Austin")));
6272 }
6273
6274 #[test]
6275 fn distinct_int_float_unify() {
6276 let mut fx = Fx::new();
6277 fx.add("N", "a", vec![("score", i(1))]);
6278 fx.add("N", "b", vec![("score", f(1.0))]);
6279 let v = fx.view();
6280 let params = BTreeMap::new();
6281 let rs = run(&v, "MATCH (n:N) RETURN DISTINCT n.score", ¶ms).unwrap();
6282 assert_eq!(
6283 rs.len(),
6284 1,
6285 "Int(1) and Float(1.0) must DISTINCT as one row"
6286 );
6287 assert_eq!(rs.row(0)[0], Some(i(1)), "first-seen Int(1) wins display");
6288 }
6289
6290 #[test]
6291 fn distinct_caps_at_intermediate_row_budget() {
6292 let mut fx = Fx::new();
6293 fx.add("N", "a", vec![("city", s("Austin"))]);
6294 fx.add("N", "b", vec![("city", s("Paris"))]);
6295 let v = fx.view();
6296 let params = BTreeMap::new();
6297 let err = super::with_max_intermediate_rows(1, || {
6298 run(&v, "MATCH (n:N) RETURN DISTINCT n.city", ¶ms)
6299 })
6300 .expect_err("two distinct cities must exceed cap=1");
6301 assert!(
6302 err.contains("1") || err.contains("row"),
6303 "cap error must mention the budget, got: {err}"
6304 );
6305 }
6306
6307 #[test]
6308 fn where_in_list_filters_rows() {
6309 let mut fx = Fx::new();
6310 fx.add("N", "a", vec![("city", s("Austin"))]);
6311 fx.add("N", "p", vec![("city", s("Paris"))]);
6312 fx.add("N", "l", vec![("city", s("London"))]);
6313 let v = fx.view();
6314 let mut params = BTreeMap::new();
6315 params.insert("c".into(), s("Paris"));
6316 let rs = run(
6317 &v,
6318 "MATCH (n:N) WHERE n.city IN ['Austin', $c] RETURN n.city",
6319 ¶ms,
6320 )
6321 .unwrap();
6322 assert_eq!(rs.len(), 2);
6323 }
6324
6325 #[test]
6326 fn pure_int_keys_display_as_int() {
6327 let mut fx = Fx::new();
6332 fx.add("N", "a", vec![("age", i(10))]);
6333 fx.add("N", "b", vec![("age", i(20))]);
6334 fx.add("N", "c", vec![("age", i(10))]);
6335 let v = fx.view();
6336 let params = BTreeMap::new();
6337
6338 let rs = run(
6339 &v,
6340 "MATCH (n:N) RETURN n.age, COUNT(*) AS cnt ORDER BY n.age",
6341 ¶ms,
6342 )
6343 .expect("pure-Int group keys must succeed");
6344 assert_eq!(rs.len(), 2, "must have 2 groups: age=10 and age=20");
6345 assert_eq!(
6347 rs.row(0)[0],
6348 Some(i(10)),
6349 "age=10 key column must display as Int(10)"
6350 );
6351 assert_eq!(rs.row(0)[1], Some(i(2)), "age=10 group has 2 nodes");
6352 assert_eq!(
6353 rs.row(1)[0],
6354 Some(i(20)),
6355 "age=20 key column must display as Int(20)"
6356 );
6357 assert_eq!(rs.row(1)[1], Some(i(1)), "age=20 group has 1 node");
6358 }
6359
6360 #[test]
6365 fn var_expand_group_aggregate_staged_path() {
6366 let mut fx = Fx::new();
6376 let a = fx.add("N", "a", vec![("key", s("a"))]);
6377 let b = fx.add("N", "b", vec![("key", s("b"))]);
6378 let c = fx.add("N", "c", vec![("key", s("c"))]);
6379 fx.edge("T", a, b, vec![]);
6380 fx.edge("T", b, c, vec![]);
6381 let v = fx.view();
6382 let params = BTreeMap::new();
6383
6384 let rs = run(
6385 &v,
6386 "MATCH (x:N)-[*1..2]->(y:N) RETURN y.key, COUNT(*) AS cnt ORDER BY y.key",
6387 ¶ms,
6388 )
6389 .expect("VarExpand + GroupAggregate must succeed");
6390
6391 assert_eq!(rs.len(), 2, "must have 2 destination groups: b and c");
6392 assert_eq!(rs.row(0)[0], Some(s("b")), "first group key must be 'b'");
6393 assert_eq!(rs.row(0)[1], Some(i(1)), "b is reached via 1 path");
6394 assert_eq!(rs.row(1)[0], Some(s("c")), "second group key must be 'c'");
6395 assert_eq!(
6396 rs.row(1)[1],
6397 Some(i(2)),
6398 "c is reached via 2 paths (1-hop and 2-hop)"
6399 );
6400 }
6401
6402 #[test]
6410 fn pull_rows_var_expand_arm_returns_named_err() {
6411 let fx = Fx::new();
6412 let view = fx.view();
6413 let vars = super::VarTable {
6414 names: vec!["a".into(), "b".into()],
6415 };
6416 let project_items: Vec<crate::cypher::ast::RetItem> = vec![];
6417 let empty_params = BTreeMap::new();
6418 let params = super::Params(&empty_params);
6419 let ctx = super::PullCtx {
6420 view: &view,
6421 vars: &vars,
6422 project_items: &project_items,
6423 params: ¶ms,
6424 bound: 100,
6425 };
6426 let ops = vec![PlanOp::VarExpand {
6427 from: "a".into(),
6428 rel_var: None,
6429 etypes: vec![],
6430 dir: crate::cypher::RelDir::Right,
6431 to: "b".into(),
6432 min: 1,
6433 max: 3,
6434 }];
6435 let mut row = vec![None; vars.names.len()];
6436 let mut result = Vec::new();
6437 let err = super::pull_rows(&ctx, &ops, &mut row, &mut result)
6438 .expect_err("VarExpand must Err in pull_rows");
6439 assert!(
6440 err.contains("VarExpand") && err.contains("pull executor"),
6441 "error must name VarExpand and pull executor, got: {err}"
6442 );
6443 }
6444
6445 #[test]
6446 fn pull_rows_shortest_path_arm_returns_named_err() {
6447 let fx = Fx::new();
6448 let view = fx.view();
6449 let vars = super::VarTable {
6450 names: vec!["a".into(), "b".into()],
6451 };
6452 let project_items: Vec<crate::cypher::ast::RetItem> = vec![];
6453 let empty_params = BTreeMap::new();
6454 let params = super::Params(&empty_params);
6455 let ctx = super::PullCtx {
6456 view: &view,
6457 vars: &vars,
6458 project_items: &project_items,
6459 params: ¶ms,
6460 bound: 100,
6461 };
6462 let ops = vec![PlanOp::ShortestPath {
6463 from: "a".into(),
6464 rel_var: None,
6465 etypes: vec![],
6466 dir: crate::cypher::RelDir::Right,
6467 to: "b".into(),
6468 max_hops: 5,
6469 }];
6470 let mut row = vec![None; vars.names.len()];
6471 let mut result = Vec::new();
6472 let err = super::pull_rows(&ctx, &ops, &mut row, &mut result)
6473 .expect_err("ShortestPath must Err in pull_rows");
6474 assert!(
6475 err.contains("ShortestPath") && err.contains("pull executor"),
6476 "error must name ShortestPath and pull executor, got: {err}"
6477 );
6478 }
6479
6480 fn city_graph() -> Fx {
6484 let mut fx = Fx::new();
6485 fx.add(
6486 "Person",
6487 "alice",
6488 vec![("city", s("Boston")), ("age", i(30))],
6489 );
6490 fx.add("Person", "bob", vec![("city", s("Boston")), ("age", i(25))]);
6491 fx.add(
6492 "Person",
6493 "carol",
6494 vec![("city", s("Austin")), ("age", i(35))],
6495 );
6496 fx.add(
6497 "Person",
6498 "dave",
6499 vec![("city", s("Austin")), ("age", i(28))],
6500 );
6501 fx.add("Person", "eve", vec![("city", s("Boston")), ("age", i(22))]);
6502 fx
6503 }
6504
6505 #[test]
6508 fn with_aggregate_having_filters_groups() {
6509 let fx = city_graph();
6510 let view = fx.view();
6511 let params = BTreeMap::new();
6512 let rs = run(
6513 &view,
6514 "MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt WHERE cnt > 2 RETURN city, cnt",
6515 ¶ms,
6516 )
6517 .expect("WITH HAVING query must succeed");
6518 assert_eq!(rs.len(), 1, "only Boston group has cnt > 2");
6519 assert_eq!(rs.get(0, "city"), Some(&s("Boston")));
6520 assert_eq!(rs.get(0, "cnt"), Some(&i(3)));
6521 }
6522
6523 #[test]
6525 fn with_order_limit_then_return() {
6526 let fx = city_graph();
6527 let view = fx.view();
6528 let params = BTreeMap::new();
6529 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)
6531 .expect("WITH ORDER LIMIT must succeed");
6532 assert_eq!(rs.len(), 2, "LIMIT 2");
6533 let ages: Vec<Option<Value>> = (0..rs.len()).map(|i| rs.get(i, "age").cloned()).collect();
6535 assert_eq!(ages[0], Some(i(35)), "oldest person first");
6536 assert_eq!(ages[1], Some(i(30)), "second oldest");
6537 }
6538
6539 #[test]
6541 fn chained_with_stages() {
6542 let fx = city_graph();
6543 let view = fx.view();
6544 let params = BTreeMap::new();
6545 let rs = run(&view,
6547 "MATCH (p:Person) WITH p.city AS city, p.age AS age WITH city, age WHERE age > 25 RETURN city, age",
6548 ¶ms).expect("chained WITH must succeed");
6549 assert_eq!(rs.len(), 3, "3 people with age > 25");
6551 }
6552
6553 #[test]
6555 fn with_then_match_reentry() {
6556 let mut fx = Fx::new();
6557 let alice = fx.add("Person", "alice", vec![]);
6558 let bob = fx.add("Person", "bob", vec![]);
6559 let corp = fx.add("Company", "acme", vec![]);
6560 fx.edge("WORKS_AT", alice, corp, vec![("years", i(5))]);
6561 fx.edge("WORKS_AT", bob, corp, vec![("years", i(3))]);
6562 let view = fx.view();
6563 let params = BTreeMap::new();
6564 let rs = run(&view,
6566 "MATCH (p:Person)-[r:WORKS_AT]->(c:Company) WITH p, c MATCH (c)-[r2:WORKS_AT]-(colleague:Person) RETURN p, colleague",
6567 ¶ms).expect("WITH MATCH re-entry must succeed");
6568 assert!(rs.len() >= 2, "cross join via company: got {}", rs.len());
6570 }
6571
6572 #[test]
6574 fn unwind_literal_list_produces_rows() {
6575 let fx = city_graph();
6576 let view = fx.view();
6577 let params = BTreeMap::new();
6578 let rs = run(
6579 &view,
6580 "MATCH (p:Person) WHERE p.city = 'Boston' UNWIND [1, 2, 3] AS x RETURN p, x",
6581 ¶ms,
6582 )
6583 .expect("UNWIND literal must succeed");
6584 assert_eq!(rs.len(), 9, "UNWIND [1,2,3] × 3 Boston people = 9 rows");
6586 let xs: Vec<Option<Value>> = (0..rs.len())
6588 .map(|row_i| rs.get(row_i, "x").cloned())
6589 .collect();
6590 assert!(xs.contains(&Some(i(1))));
6591 assert!(xs.contains(&Some(i(2))));
6592 assert!(xs.contains(&Some(i(3))));
6593 }
6594
6595 #[test]
6597 fn unwind_list_property() {
6598 let mut fx = Fx::new();
6599 fx.add(
6600 "Tag",
6601 "post1",
6602 vec![("tags", Value::List(vec![s("rust"), s("graph")]))],
6603 );
6604 fx.add("Tag", "post2", vec![("tags", Value::List(vec![s("db")]))]);
6605 let view = fx.view();
6606 let params = BTreeMap::new();
6607 let rs = run(
6608 &view,
6609 "MATCH (p:Tag) UNWIND p.tags AS tag RETURN p, tag",
6610 ¶ms,
6611 )
6612 .expect("UNWIND property must succeed");
6613 assert_eq!(rs.len(), 3, "2+1 tag elements");
6614 }
6615
6616 #[test]
6618 fn unwind_empty_list_yields_zero_rows() {
6619 let fx = city_graph();
6620 let view = fx.view();
6621 let params = BTreeMap::new();
6622 let rs = run(
6623 &view,
6624 "MATCH (p:Person) WHERE p.city = 'Boston' UNWIND [] AS x RETURN x",
6625 ¶ms,
6626 )
6627 .expect("UNWIND [] must succeed with 0 rows");
6628 assert_eq!(rs.len(), 0, "UNWIND [] should produce 0 rows");
6629 }
6630
6631 #[test]
6633 fn unwind_non_list_is_named_error() {
6634 let fx = city_graph();
6635 let view = fx.view();
6636 let params = BTreeMap::new();
6637 let err = run(
6639 &view,
6640 "MATCH (p:Person) WHERE p.city = 'Boston' UNWIND p.city AS x RETURN x",
6641 ¶ms,
6642 )
6643 .expect_err("UNWIND non-list must error");
6644 assert!(
6645 err.contains("UNWIND") && err.contains("list"),
6646 "error must mention UNWIND and list: {err}"
6647 );
6648 }
6649
6650 #[test]
6652 fn unwind_cross_product_trips_budget() {
6653 let mut fx = Fx::new();
6654 for i in 0..10 {
6656 fx.add("N", &format!("n{i}"), vec![]);
6657 }
6658 let view = fx.view();
6659 let params = BTreeMap::new();
6660 let err = super::with_max_intermediate_rows(5, || {
6661 run(
6662 &view,
6663 "MATCH (n:N) UNWIND [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] AS x RETURN n, x",
6664 ¶ms,
6665 )
6666 })
6667 .expect_err("must hit budget");
6668 assert!(
6669 err.contains("intermediate result exceeds"),
6670 "error must mention intermediate result limit: {err}"
6671 );
6672 }
6673
6674 #[test]
6677 fn routing_with_plan_uses_staged_path() {
6678 use crate::cypher::plan::row_bound;
6679 let ops_with = compile("MATCH (a) WITH a RETURN a");
6680 assert_eq!(
6681 row_bound(&ops_with),
6682 None,
6683 "WITH plan must have row_bound=None"
6684 );
6685
6686 let ops_unwind = compile("MATCH (a) UNWIND [1, 2] AS x RETURN a");
6687 assert_eq!(
6688 row_bound(&ops_unwind),
6689 None,
6690 "UNWIND plan must have row_bound=None"
6691 );
6692
6693 let ops_plain = compile("MATCH (a) RETURN a LIMIT 5");
6695 assert!(
6696 row_bound(&ops_plain).is_some(),
6697 "plain LIMIT plan must have a row bound"
6698 );
6699 }
6700
6701 #[test]
6704 fn unwind_null_property_yields_zero_rows_for_that_node() {
6705 let mut fx = Fx::new();
6706 fx.add(
6708 "Post",
6709 "post1",
6710 vec![("tags", Value::List(vec![s("rust"), s("graph")]))],
6711 );
6712 fx.add("Post", "post2", vec![("other", Value::Str("hello".into()))]);
6713 let view = fx.view();
6714 let params = BTreeMap::new();
6715 let rs = run(
6716 &view,
6717 "MATCH (p:Post) UNWIND p.tags AS tag RETURN tag",
6718 ¶ms,
6719 )
6720 .expect("UNWIND null property must succeed (not error)");
6721 assert_eq!(
6723 rs.len(),
6724 2,
6725 "null property must yield 0 rows; list property yields N rows"
6726 );
6727 let tag0 = rs.get(0, "tag");
6728 let tag1 = rs.get(1, "tag");
6729 let tags = [tag0, tag1];
6730 assert!(tags.contains(&Some(&s("rust"))), "rust tag present");
6731 assert!(tags.contains(&Some(&s("graph"))), "graph tag present");
6732 }
6733
6734 #[test]
6736 fn where_before_unwind_filters_nodes() {
6737 let mut fx = Fx::new();
6738 fx.add(
6739 "P",
6740 "a",
6741 vec![
6742 ("group", Value::Str("keep".into())),
6743 ("items", Value::List(vec![Value::Int(1), Value::Int(2)])),
6744 ],
6745 );
6746 fx.add(
6747 "P",
6748 "b",
6749 vec![
6750 ("group", Value::Str("drop".into())),
6751 ("items", Value::List(vec![Value::Int(3), Value::Int(4)])),
6752 ],
6753 );
6754 let view = fx.view();
6755 let params = BTreeMap::new();
6756 let rs = run(
6758 &view,
6759 "MATCH (n:P) WHERE n.group = 'keep' UNWIND n.items AS x RETURN x",
6760 ¶ms,
6761 )
6762 .expect("WHERE before UNWIND must succeed");
6763 assert_eq!(rs.len(), 2, "only matching node expands; 2 items");
6764 }
6765
6766 #[test]
6768 fn where_after_unwind_filters_expanded_rows() {
6769 let mut fx = Fx::new();
6770 fx.add(
6771 "N",
6772 "n1",
6773 vec![(
6774 "vals",
6775 Value::List(vec![Value::Int(1), Value::Int(3), Value::Int(5)]),
6776 )],
6777 );
6778 let view = fx.view();
6779 let params = BTreeMap::new();
6780 let rs = run(
6782 &view,
6783 "MATCH (n:N) UNWIND n.vals AS x WHERE x > 2 RETURN x",
6784 ¶ms,
6785 )
6786 .expect("WHERE after UNWIND must succeed");
6787 assert_eq!(rs.len(), 2, "x=3 and x=5 pass; x=1 filtered out");
6788 let v0 = rs.get(0, "x").cloned();
6789 let v1 = rs.get(1, "x").cloned();
6790 let vals = [v0, v1];
6791 assert!(vals.contains(&Some(Value::Int(3))), "x=3 present");
6792 assert!(vals.contains(&Some(Value::Int(5))), "x=5 present");
6793 }
6794
6795 #[test]
6797 fn aggregate_with_order_by_unknown_alias_is_named_error() {
6798 use crate::cypher::{lex, parser::parse, plan::plan};
6799 let src = "MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt ORDER BY cntt DESC RETURN city, cnt";
6801 let plan_result = plan(&parse(&lex(src).expect("lex")).expect("parse"));
6802 let err = plan_result
6803 .expect_err("typo'd ORDER BY alias in aggregate WITH must be a named plan error");
6804 assert!(
6805 err.contains("cntt") || err.contains("unbound"),
6806 "error must mention the unknown alias: {err}"
6807 );
6808 }
6809
6810 #[test]
6812 fn non_aggregate_with_order_by_unknown_alias_is_named_error() {
6813 use crate::cypher::{lex, parser::parse, plan::plan};
6814 let src = "MATCH (p:Person) WITH p, p.age AS age ORDER BY nope DESC RETURN p.city AS city";
6815 let plan_result = plan(&parse(&lex(src).expect("lex")).expect("parse"));
6816 let err = plan_result
6817 .expect_err("typo'd ORDER BY alias in non-aggregate WITH must be a named plan error");
6818 assert!(
6819 err.contains("nope") || err.contains("unbound"),
6820 "error must mention the unknown alias: {err}"
6821 );
6822 }
6823
6824 #[test]
6828 fn post_unwind_where_references_pre_match_variable() {
6829 let mut fx = Fx::new();
6830 fx.add(
6832 "N",
6833 "n1",
6834 vec![
6835 ("threshold", Value::Int(3)),
6836 (
6837 "xs",
6838 Value::List(vec![
6839 Value::Int(1),
6840 Value::Int(2),
6841 Value::Int(3),
6842 Value::Int(4),
6843 Value::Int(5),
6844 ]),
6845 ),
6846 ],
6847 );
6848 fx.add(
6850 "N",
6851 "n2",
6852 vec![
6853 ("threshold", Value::Int(10)),
6854 ("xs", Value::List(vec![Value::Int(1), Value::Int(2)])),
6855 ],
6856 );
6857 let view = fx.view();
6858 let params = BTreeMap::new();
6859 let rs = run(
6860 &view,
6861 "MATCH (n:N) UNWIND n.xs AS x WHERE x > n.threshold RETURN x",
6862 ¶ms,
6863 )
6864 .expect("post-UNWIND WHERE referencing MATCH variable must succeed");
6865 assert_eq!(rs.len(), 2, "exactly 2 rows: x=4 and x=5 from n1");
6867 let v0 = rs.get(0, "x").cloned();
6868 let v1 = rs.get(1, "x").cloned();
6869 let got = [v0, v1];
6870 assert!(got.contains(&Some(Value::Int(4))), "x=4 present");
6871 assert!(got.contains(&Some(Value::Int(5))), "x=5 present");
6872 }
6873
6874 #[test]
6878 fn with_stage_aggregate_order_by_descending() {
6879 let fx = city_graph();
6882 let view = fx.view();
6883 let params = BTreeMap::new();
6884 let rs = run(
6885 &view,
6886 "MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt ORDER BY cnt DESC RETURN city, cnt",
6887 ¶ms,
6888 )
6889 .expect("aggregate WITH ORDER BY must succeed");
6890 assert_eq!(rs.len(), 2, "two city groups");
6891 assert_eq!(
6893 rs.get(0, "cnt"),
6894 Some(&Value::Int(3)),
6895 "first row must be the group with cnt=3 (Boston)"
6896 );
6897 assert_eq!(
6898 rs.get(1, "cnt"),
6899 Some(&Value::Int(2)),
6900 "second row must be the group with cnt=2 (Austin)"
6901 );
6902 }
6903
6904 #[test]
6908 fn unwind_then_with_composition() {
6909 let mut fx = Fx::new();
6910 fx.add(
6911 "Doc",
6912 "d1",
6913 vec![(
6914 "scores",
6915 Value::List(vec![Value::Int(10), Value::Int(20), Value::Int(30)]),
6916 )],
6917 );
6918 fx.add(
6919 "Doc",
6920 "d2",
6921 vec![("scores", Value::List(vec![Value::Int(5), Value::Int(15)]))],
6922 );
6923 let view = fx.view();
6924 let params = BTreeMap::new();
6925
6926 let rs = run(
6928 &view,
6929 "MATCH (n:Doc) UNWIND n.scores AS x WITH x RETURN x",
6930 ¶ms,
6931 )
6932 .expect("UNWIND then WITH pass-through must succeed");
6933 assert_eq!(rs.len(), 5, "3 + 2 = 5 expanded rows carried through WITH");
6934
6935 let rs2 = run(
6937 &view,
6938 "MATCH (n:Doc) UNWIND n.scores AS x WITH COUNT(*) AS total RETURN total",
6939 ¶ms,
6940 )
6941 .expect("UNWIND then aggregate WITH must succeed");
6942 assert_eq!(rs2.len(), 1, "single aggregate row");
6943 assert_eq!(
6944 rs2.get(0, "total"),
6945 Some(&Value::Int(5)),
6946 "total must be 5 (3+2 expanded rows)"
6947 );
6948 }
6949
6950 #[test]
6956 fn binarith_div_min_over_neg1_does_not_panic() {
6957 let fx = Fx::new();
6958 let view = fx.view();
6959 let vars = VarTable { names: vec![] };
6960 let row: Row = vec![];
6961 let params = BTreeMap::new();
6962
6963 let operand = Operand::BinArith {
6965 op: ArithOp::Div,
6966 left: Box::new(Operand::Lit(Value::Int(i64::MIN))),
6967 right: Box::new(Operand::Lit(Value::Int(-1))),
6968 };
6969
6970 let result = resolve_operand(&view, &vars, &row, &operand, &Params(¶ms));
6971 assert!(
6972 result.is_ok(),
6973 "overflow division must not return Err: {result:?}"
6974 );
6975 assert_eq!(
6976 result.unwrap(),
6977 Some(Value::Int(i64::MAX)),
6978 "i64::MIN / -1 must saturate to i64::MAX, not panic"
6979 );
6980 }
6981
6982 #[test]
6987 fn binarith_missing_param_caught_at_preflight() {
6988 let fx = Fx::new();
6989 let view = fx.view();
6990 let err = run(
6995 &view,
6996 "MATCH (n:X) RETURN abs($missing - 1)",
6997 &BTreeMap::new(),
6998 )
6999 .expect_err("missing param inside BinArith must be caught at preflight");
7000 assert!(
7001 err.contains("missing"),
7002 "error must name the param 'missing', got: {err}"
7003 );
7004 }
7005
7006 #[test]
7010 fn is_null_filters_absent_prop() {
7011 let mut fx = Fx::new();
7012 fx.add("Person", "alice", vec![("age", Value::Int(30))]);
7013 fx.add("Person", "bob", vec![]); let view = fx.view();
7015 let rs = run(
7016 &view,
7017 "MATCH (n:Person) WHERE n.age IS NULL RETURN n",
7018 &BTreeMap::new(),
7019 )
7020 .unwrap();
7021 assert_eq!(rs.len(), 1, "only bob lacks age");
7022 assert_eq!(rs.get(0, "n"), Some(&s("bob")));
7023 }
7024
7025 #[test]
7027 fn is_not_null_filters_present_prop() {
7028 let mut fx = Fx::new();
7029 fx.add("Person", "alice", vec![("age", Value::Int(30))]);
7030 fx.add("Person", "bob", vec![]); let view = fx.view();
7032 let rs = run(
7033 &view,
7034 "MATCH (n:Person) WHERE n.age IS NOT NULL RETURN n",
7035 &BTreeMap::new(),
7036 )
7037 .unwrap();
7038 assert_eq!(rs.len(), 1, "only alice has age");
7039 assert_eq!(rs.get(0, "n"), Some(&s("alice")));
7040 }
7041
7042 #[test]
7045 fn optional_match_is_null_anti_join() {
7046 let mut fx = Fx::new();
7047 let alice = fx.add("Person", "alice", vec![]);
7048 let bob = fx.add("Person", "bob", vec![]);
7049 let carol = fx.add("Person", "carol", vec![]);
7050 fx.edge("KNOWS", alice, carol, vec![]); fx.edge("KNOWS", carol, bob, vec![]); let view = fx.view();
7054 let rs = run(
7055 &view,
7056 "MATCH (a:Person) OPTIONAL MATCH (a)-[:KNOWS]->(b) WITH a, b WHERE b IS NULL RETURN a",
7057 &BTreeMap::new(),
7058 )
7059 .unwrap();
7060 assert_eq!(rs.len(), 1, "only bob has no outgoing KNOWS edge");
7061 assert_eq!(rs.get(0, "a"), Some(&s("bob")));
7062 }
7063
7064 #[test]
7066 fn is_null_combined_with_and_exec() {
7067 let mut fx = Fx::new();
7068 fx.add("N", "a", vec![("x", Value::Int(1))]);
7069 fx.add("N", "b", vec![("x", Value::Int(2)), ("y", Value::Int(9))]);
7070 fx.add("N", "c", vec![]); let view = fx.view();
7072 let rs = run(
7074 &view,
7075 "MATCH (n:N) WHERE n.y IS NULL AND n.x IS NOT NULL RETURN n",
7076 &BTreeMap::new(),
7077 )
7078 .unwrap();
7079 assert_eq!(rs.len(), 1);
7080 assert_eq!(rs.get(0, "n"), Some(&s("a")));
7081 }
7082
7083 #[test]
7087 fn arithmetic_add_in_return() {
7088 let mut fx = Fx::new();
7089 fx.add("Person", "alice", vec![("age", Value::Int(29))]);
7090 let view = fx.view();
7091 let rs = run(
7092 &view,
7093 "MATCH (n:Person) RETURN n.age + 1 AS adjusted",
7094 &BTreeMap::new(),
7095 )
7096 .unwrap();
7097 assert_eq!(rs.len(), 1);
7098 assert_eq!(rs.get(0, "adjusted"), Some(&Value::Int(30)));
7099 }
7100
7101 #[test]
7103 fn arithmetic_precedence_parens_pin() {
7104 let mut fx = Fx::new();
7105 fx.add("N", "x", vec![]);
7106 let view = fx.view();
7107 let rs = run(
7108 &view,
7109 "MATCH (n:N) RETURN (1 + 2) * 3 AS r",
7110 &BTreeMap::new(),
7111 )
7112 .unwrap();
7113 assert_eq!(rs.len(), 1);
7114 assert_eq!(rs.get(0, "r"), Some(&Value::Int(9)));
7115 }
7116
7117 #[test]
7119 fn arithmetic_precedence_mul_over_add_pin() {
7120 let mut fx = Fx::new();
7121 fx.add("N", "x", vec![]);
7122 let view = fx.view();
7123 let rs = run(&view, "MATCH (n:N) RETURN 1 + 2 * 3 AS r", &BTreeMap::new()).unwrap();
7124 assert_eq!(rs.len(), 1);
7125 assert_eq!(rs.get(0, "r"), Some(&Value::Int(7)));
7126 }
7127
7128 #[test]
7130 fn arithmetic_div_by_zero_returns_error() {
7131 let mut fx = Fx::new();
7132 fx.add("N", "x", vec![]);
7133 let view = fx.view();
7134 let err = run(
7135 &view,
7136 "MATCH (n:N) WHERE 1 / 0 > 0 RETURN n",
7137 &BTreeMap::new(),
7138 )
7139 .expect_err("division by zero must error");
7140 assert!(
7141 err.contains("division by zero"),
7142 "error must mention division by zero, got: {err}"
7143 );
7144 }
7145
7146 #[test]
7148 fn arithmetic_null_propagates() {
7149 let mut fx = Fx::new();
7150 fx.add("N", "x", vec![]); let view = fx.view();
7152 let rs = run(
7154 &view,
7155 "MATCH (n:N) WHERE n.val + 1 > 0 RETURN n",
7156 &BTreeMap::new(),
7157 )
7158 .unwrap();
7159 assert_eq!(rs.len(), 0, "null arithmetic must not match");
7160 }
7161
7162 #[test]
7164 fn case_when_expression_in_return() {
7165 let mut fx = Fx::new();
7166 fx.add("N", "a", vec![("id", s("a")), ("age", i(20))]);
7167 fx.add("N", "b", vec![("id", s("b")), ("age", i(65))]);
7168 let v = fx.view();
7169 let rs = run(
7170 &v,
7171 "MATCH (n:N) RETURN n.id AS id, \
7172 CASE WHEN n.age >= 65 THEN 'senior' ELSE 'other' END AS band",
7173 &BTreeMap::new(),
7174 )
7175 .unwrap();
7176 let band_of = |who: &str| {
7177 (0..rs.len())
7178 .find(|&i| rs.get(i, "id") == Some(&s(who)))
7179 .and_then(|i| rs.get(i, "band").cloned())
7180 };
7181 assert_eq!(band_of("a"), Some(s("other")));
7182 assert_eq!(band_of("b"), Some(s("senior")));
7183 }
7184
7185 #[test]
7186 fn case_when_no_else_yields_null() {
7187 let mut fx = Fx::new();
7188 fx.add("N", "a", vec![("age", i(20))]);
7189 let v = fx.view();
7190 let rs = run(
7191 &v,
7192 "MATCH (n:N) RETURN CASE WHEN n.age >= 65 THEN 'senior' END AS band",
7193 &BTreeMap::new(),
7194 )
7195 .unwrap();
7196 assert_eq!(rs.get(0, "band"), None);
7197 }
7198
7199 #[test]
7200 fn multi_relationship_type_pattern_matches_either() {
7201 let mut fx = Fx::new();
7202 let a = fx.add("N", "a", vec![("id", s("a"))]);
7203 let b = fx.add("N", "b", vec![("id", s("b"))]);
7204 let c = fx.add("N", "c", vec![("id", s("c"))]);
7205 let d = fx.add("N", "d", vec![("id", s("d"))]);
7206 fx.edge("KNOWS", a, b, vec![]);
7207 fx.edge("LIKES", a, c, vec![]);
7208 fx.edge("HATES", a, d, vec![]);
7209 let v = fx.view();
7210 let rs = run(
7212 &v,
7213 "MATCH (a:N {id: 'a'})-[r:KNOWS|:LIKES]->(x) RETURN x",
7214 &BTreeMap::new(),
7215 )
7216 .unwrap();
7217 assert_eq!(rs.len(), 2, "KNOWS|LIKES reaches exactly b and c");
7218 }
7219
7220 #[test]
7221 fn collect_grouped_gathers_values_per_group() {
7222 let mut fx = Fx::new();
7223 fx.add("P", "a", vec![("city", s("austin")), ("name", s("Ann"))]);
7224 fx.add("P", "b", vec![("city", s("austin")), ("name", s("Bob"))]);
7225 fx.add("P", "c", vec![("city", s("boston")), ("name", s("Cy"))]);
7226 let v = fx.view();
7227 let rs = run(
7228 &v,
7229 "MATCH (n:P) RETURN n.city AS city, collect(n.name) AS names",
7230 &BTreeMap::new(),
7231 )
7232 .unwrap();
7233 assert_eq!(rs.len(), 2, "two city groups");
7234 let austin = (0..rs.len())
7236 .find(|&i| rs.get(i, "city") == Some(&s("austin")))
7237 .expect("austin group present");
7238 assert_eq!(
7239 rs.get(austin, "names"),
7240 Some(&Value::List(vec![s("Ann"), s("Bob")]))
7241 );
7242 }
7243
7244 #[test]
7245 fn collect_ungrouped_gathers_all_into_one_list() {
7246 let mut fx = Fx::new();
7247 fx.add("P", "a", vec![("name", s("Ann"))]);
7248 fx.add("P", "b", vec![("name", s("Bob"))]);
7249 let v = fx.view();
7250 let rs = run(
7251 &v,
7252 "MATCH (n:P) RETURN collect(n.name) AS names",
7253 &BTreeMap::new(),
7254 )
7255 .unwrap();
7256 assert_eq!(rs.len(), 1);
7257 assert_eq!(
7258 rs.get(0, "names"),
7259 Some(&Value::List(vec![s("Ann"), s("Bob")]))
7260 );
7261 }
7262
7263 #[test]
7264 fn string_predicate_functions_in_where() {
7265 let mut fx = Fx::new();
7266 fx.add("N", "a", vec![("email", s("alice@acme.com"))]);
7267 fx.add("N", "b", vec![("email", s("bob@other.org"))]);
7268 let v = fx.view();
7269 let p = BTreeMap::new();
7270 assert_eq!(
7271 run(
7272 &v,
7273 "MATCH (n:N) WHERE endsWith(n.email, '.com') RETURN n",
7274 &p
7275 )
7276 .unwrap()
7277 .len(),
7278 1
7279 );
7280 assert_eq!(
7281 run(
7282 &v,
7283 "MATCH (n:N) WHERE startsWith(n.email, 'bob') RETURN n",
7284 &p
7285 )
7286 .unwrap()
7287 .len(),
7288 1
7289 );
7290 assert_eq!(
7291 run(
7292 &v,
7293 "MATCH (n:N) WHERE contains(n.email, 'acme') RETURN n",
7294 &p
7295 )
7296 .unwrap()
7297 .len(),
7298 1
7299 );
7300 }
7301
7302 #[test]
7303 fn coercion_functions_in_return() {
7304 let mut fx = Fx::new();
7305 fx.add("N", "a", vec![("s", s("42")), ("n", i(7)), ("g", f(3.9))]);
7306 let v = fx.view();
7307 let rs = run(
7308 &v,
7309 "MATCH (n:N) RETURN toInteger(n.s) AS ti, toFloat(n.n) AS tf, toString(n.g) AS ts",
7310 &BTreeMap::new(),
7311 )
7312 .unwrap();
7313 assert_eq!(rs.get(0, "ti"), Some(&Value::Int(42)));
7314 assert_eq!(rs.get(0, "tf"), Some(&Value::Float(7.0)));
7315 assert_eq!(rs.get(0, "ts"), Some(&Value::Str("3.9".into())));
7316 }
7317
7318 #[test]
7319 fn to_integer_unparseable_string_is_null() {
7320 let mut fx = Fx::new();
7321 fx.add("N", "a", vec![("s", s("not-a-number"))]);
7322 let v = fx.view();
7323 let rs = run(
7324 &v,
7325 "MATCH (n:N) RETURN toInteger(n.s) AS ti",
7326 &BTreeMap::new(),
7327 )
7328 .unwrap();
7329 assert_eq!(rs.get(0, "ti"), None);
7330 }
7331
7332 #[test]
7333 fn index_scan_uses_index_and_matches_fallback() {
7334 use core_storage::property_index::PropertyIndex;
7335 use std::sync::atomic::Ordering;
7336
7337 let mut fx = Fx::new();
7338 let a = fx.add("Person", "a", vec![("city", s("austin"))]);
7339 let _b = fx.add("Person", "b", vec![("city", s("boston"))]);
7340 let c = fx.add("Person", "c", vec![("city", s("austin"))]);
7341
7342 let mut pi = PropertyIndex::new();
7343 pi.enable("Person", "city");
7344 pi.set("Person", "city", a, &s("austin"));
7345 pi.set("Person", "city", 1, &s("boston"));
7346 pi.set("Person", "city", c, &s("austin"));
7347
7348 let q = "MATCH (n:Person {city: 'austin'}) RETURN n";
7349
7350 let before = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7352 let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
7353 let after = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7354 assert!(after > before, "IndexScan must take the indexed path");
7355 assert_eq!(indexed.len(), 2);
7356
7357 let before2 = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7360 let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7361 let after2 = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7362 assert_eq!(after2, before2, "fallback must not touch the index counter");
7363 assert_eq!(
7364 fallback.len(),
7365 indexed.len(),
7366 "fallback matches indexed result"
7367 );
7368 }
7369
7370 #[test]
7371 fn arithmetic_in_where_comparison() {
7372 let mut fx = Fx::new();
7373 fx.add("Person", "alice", vec![("age", Value::Int(5))]); fx.add("Person", "bob", vec![("age", Value::Int(4))]); let view = fx.view();
7376 let rs = run(
7377 &view,
7378 "MATCH (n:Person) WHERE n.age + 1 > 5 RETURN n",
7379 &BTreeMap::new(),
7380 )
7381 .unwrap();
7382 assert_eq!(rs.len(), 1);
7383 assert_eq!(rs.get(0, "n"), Some(&s("alice")));
7384 }
7385
7386 #[test]
7390 fn where_fold_indexed_matches_fallback() {
7391 use core_storage::property_index::PropertyIndex;
7392 use std::sync::atomic::Ordering;
7393
7394 let mut fx = Fx::new();
7395 let a = fx.add("Person", "alice", vec![("city", s("austin"))]);
7396 let b = fx.add("Person", "bob", vec![("city", s("boston"))]);
7397 let c = fx.add("Person", "carol", vec![("city", s("austin"))]);
7398
7399 let mut pi = PropertyIndex::new();
7400 pi.enable("Person", "city");
7401 pi.set("Person", "city", a, &s("austin"));
7402 pi.set("Person", "city", b, &s("boston"));
7403 pi.set("Person", "city", c, &s("austin"));
7404
7405 let q = "MATCH (n:Person) WHERE n.city = 'austin' RETURN n";
7406
7407 let before = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7409 let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
7410 let after = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7411 assert!(
7412 after > before,
7413 "WHERE equality fold must take the indexed path"
7414 );
7415 assert_eq!(indexed.len(), 2);
7416
7417 let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7419 assert_eq!(
7420 rows_of(&fallback),
7421 rows_of(&indexed),
7422 "fallback must return identical rows, not just the same count"
7423 );
7424 }
7425
7426 #[test]
7428 fn where_fold_miss_returns_empty() {
7429 let mut fx = Fx::new();
7430 fx.add("Person", "alice", vec![("city", s("austin"))]);
7431 let q = "MATCH (n:Person) WHERE n.city = 'berlin' RETURN n";
7432 let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7433 assert_eq!(rs.len(), 0);
7434 }
7435
7436 #[test]
7438 fn where_fold_param_uses_index() {
7439 use core_storage::property_index::PropertyIndex;
7440 use std::sync::atomic::Ordering;
7441
7442 let mut fx = Fx::new();
7443 let a = fx.add("Person", "alice", vec![("city", s("austin"))]);
7444 let b = fx.add("Person", "bob", vec![("city", s("boston"))]);
7445
7446 let mut pi = PropertyIndex::new();
7447 pi.enable("Person", "city");
7448 pi.set("Person", "city", a, &s("austin"));
7449 pi.set("Person", "city", b, &s("boston"));
7450
7451 let q = "MATCH (n:Person) WHERE n.city = $c RETURN n";
7452 let mut params = BTreeMap::new();
7453 params.insert("c".to_string(), s("austin"));
7454
7455 let before = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7456 let rs = run(&fx.view_indexed(&pi), q, ¶ms).unwrap();
7457 let after = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7458 assert!(after > before, "$param WHERE equality must use index");
7459 assert_eq!(rs.len(), 1);
7460 }
7461
7462 #[test]
7464 fn where_fold_residual_filter_applied() {
7465 let mut fx = Fx::new();
7466 fx.add(
7467 "Person",
7468 "young-austin",
7469 vec![("city", s("austin")), ("age", Value::Int(20))],
7470 );
7471 fx.add(
7472 "Person",
7473 "old-austin",
7474 vec![("city", s("austin")), ("age", Value::Int(40))],
7475 );
7476 fx.add(
7477 "Person",
7478 "boston",
7479 vec![("city", s("boston")), ("age", Value::Int(20))],
7480 );
7481
7482 let q = "MATCH (n:Person) WHERE n.city = 'austin' AND n.age > 30 RETURN n";
7483 let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7484 assert_eq!(rs.len(), 1, "only old-austin should match city+age filter");
7485 assert_eq!(rs.get(0, "n"), Some(&s("old-austin")));
7486 }
7487
7488 #[test]
7494 fn where_fold_unindexed_field_fallback() {
7495 let mut fx = Fx::new();
7496 fx.add("Person", "a", vec![("notindexed", s("x"))]);
7497 fx.add("Person", "b", vec![("notindexed", s("y"))]);
7498 let rs = run(
7499 &fx.view(),
7500 "MATCH (n:Person) WHERE n.notindexed = 'x' RETURN n",
7501 &BTreeMap::new(),
7502 )
7503 .unwrap();
7504 assert_eq!(rs.len(), 1);
7505 assert_eq!(rs.get(0, "n"), Some(&s("a")));
7506 }
7507
7508 #[test]
7512 fn index_intersect_both_indexed_fires() {
7513 use core_storage::property_index::PropertyIndex;
7514 use std::sync::atomic::Ordering;
7515
7516 let mut fx = Fx::new();
7517 let a = fx.add(
7519 "Person",
7520 "alice",
7521 vec![("city", s("austin")), ("age", Value::Int(30))],
7522 );
7523 let b = fx.add(
7525 "Person",
7526 "bob",
7527 vec![("city", s("austin")), ("age", Value::Int(25))],
7528 );
7529 let c = fx.add(
7531 "Person",
7532 "carol",
7533 vec![("city", s("boston")), ("age", Value::Int(30))],
7534 );
7535
7536 let mut pi = PropertyIndex::new();
7537 pi.enable("Person", "city");
7538 pi.enable("Person", "age");
7539 pi.set("Person", "city", a, &s("austin"));
7540 pi.set("Person", "city", b, &s("austin"));
7541 pi.set("Person", "city", c, &s("boston"));
7542 pi.set("Person", "age", a, &Value::Int(30));
7543 pi.set("Person", "age", b, &Value::Int(25));
7544 pi.set("Person", "age", c, &Value::Int(30));
7545
7546 let q = "MATCH (n:Person) WHERE n.city = 'austin' AND n.age = 30 RETURN n";
7547
7548 let before = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
7549 let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
7550 let after = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
7551 assert!(
7552 after > before,
7553 "IndexIntersect must advance counter on indexed path"
7554 );
7555 assert_eq!(indexed.len(), 1);
7556 assert_eq!(indexed.get(0, "n"), Some(&s("alice")));
7557
7558 let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7560 assert_eq!(
7561 rows_of(&fallback),
7562 rows_of(&indexed),
7563 "fallback must return identical rows"
7564 );
7565 }
7566
7567 #[test]
7569 fn index_intersect_one_indexed_one_not() {
7570 use core_storage::property_index::PropertyIndex;
7571 use std::sync::atomic::Ordering;
7572
7573 let mut fx = Fx::new();
7574 let a = fx.add(
7575 "Person",
7576 "alice",
7577 vec![("city", s("austin")), ("role", s("eng"))],
7578 );
7579 let b = fx.add(
7580 "Person",
7581 "bob",
7582 vec![("city", s("austin")), ("role", s("mgr"))],
7583 );
7584 let _c = fx.add(
7585 "Person",
7586 "carol",
7587 vec![("city", s("boston")), ("role", s("eng"))],
7588 );
7589
7590 let mut pi = PropertyIndex::new();
7592 pi.enable("Person", "city");
7593 pi.set("Person", "city", a, &s("austin"));
7594 pi.set("Person", "city", b, &s("austin"));
7595
7596 let q = "MATCH (n:Person) WHERE n.city = 'austin' AND n.role = 'eng' RETURN n";
7597
7598 let before = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
7599 let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
7600 let after = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
7601 assert!(
7602 after > before,
7603 "IndexIntersect must fire when at least one field is indexed"
7604 );
7605 assert_eq!(indexed.len(), 1);
7606 assert_eq!(indexed.get(0, "n"), Some(&s("alice")));
7607
7608 let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7609 assert_eq!(
7610 rows_of(&fallback),
7611 rows_of(&indexed),
7612 "fallback must return identical rows"
7613 );
7614 }
7615
7616 #[test]
7621 fn index_intersect_no_indexed_fallback() {
7622 let mut fx = Fx::new();
7623 fx.add("Person", "alice", vec![("x", s("1")), ("y", s("a"))]);
7624 fx.add("Person", "bob", vec![("x", s("1")), ("y", s("b"))]);
7625 fx.add("Person", "carol", vec![("x", s("2")), ("y", s("a"))]);
7626
7627 let q = "MATCH (n:Person) WHERE n.x = '1' AND n.y = 'a' RETURN n";
7628 let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7629 assert_eq!(rs.len(), 1);
7630 assert_eq!(rs.get(0, "n"), Some(&s("alice")));
7631 }
7632
7633 #[test]
7635 fn index_intersect_empty_intersection() {
7636 use core_storage::property_index::PropertyIndex;
7637
7638 let mut fx = Fx::new();
7639 let a = fx.add(
7640 "Person",
7641 "alice",
7642 vec![("city", s("austin")), ("age", Value::Int(30))],
7643 );
7644 let b = fx.add(
7645 "Person",
7646 "bob",
7647 vec![("city", s("boston")), ("age", Value::Int(25))],
7648 );
7649
7650 let mut pi = PropertyIndex::new();
7651 pi.enable("Person", "city");
7652 pi.enable("Person", "age");
7653 pi.set("Person", "city", a, &s("austin"));
7654 pi.set("Person", "city", b, &s("boston"));
7655 pi.set("Person", "age", a, &Value::Int(30));
7656 pi.set("Person", "age", b, &Value::Int(25));
7657
7658 let q = "MATCH (n:Person) WHERE n.city = 'boston' AND n.age = 30 RETURN n";
7660 let rs = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
7661 assert_eq!(rs.len(), 0);
7662 }
7663
7664 #[test]
7666 fn index_intersect_with_params() {
7667 use core_storage::property_index::PropertyIndex;
7668 use std::sync::atomic::Ordering;
7669
7670 let mut fx = Fx::new();
7671 let a = fx.add(
7672 "Person",
7673 "alice",
7674 vec![("city", s("austin")), ("age", Value::Int(30))],
7675 );
7676 let b = fx.add(
7677 "Person",
7678 "bob",
7679 vec![("city", s("boston")), ("age", Value::Int(30))],
7680 );
7681
7682 let mut pi = PropertyIndex::new();
7683 pi.enable("Person", "city");
7684 pi.enable("Person", "age");
7685 pi.set("Person", "city", a, &s("austin"));
7686 pi.set("Person", "city", b, &s("boston"));
7687 pi.set("Person", "age", a, &Value::Int(30));
7688 pi.set("Person", "age", b, &Value::Int(30));
7689
7690 let q = "MATCH (n:Person) WHERE n.city = $c AND n.age = $a RETURN n";
7691 let mut params = BTreeMap::new();
7692 params.insert("c".to_string(), s("austin"));
7693 params.insert("a".to_string(), Value::Int(30));
7694
7695 let before = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
7696 let indexed = run(&fx.view_indexed(&pi), q, ¶ms).unwrap();
7697 let after = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
7698 assert!(after > before, "$param intersect must fire indexed path");
7699 assert_eq!(indexed.len(), 1);
7700 assert_eq!(indexed.get(0, "n"), Some(&s("alice")));
7701
7702 let fallback = run(&fx.view(), q, ¶ms).unwrap();
7703 assert_eq!(
7704 rows_of(&fallback),
7705 rows_of(&indexed),
7706 "fallback must return identical rows"
7707 );
7708 }
7709
7710 #[test]
7712 fn index_intersect_three_fields() {
7713 use core_storage::property_index::PropertyIndex;
7714
7715 let mut fx = Fx::new();
7716 let a = fx.add(
7718 "Person",
7719 "alice",
7720 vec![
7721 ("city", s("austin")),
7722 ("age", Value::Int(30)),
7723 ("role", s("eng")),
7724 ],
7725 );
7726 let b = fx.add(
7728 "Person",
7729 "bob",
7730 vec![
7731 ("city", s("austin")),
7732 ("age", Value::Int(30)),
7733 ("role", s("mgr")),
7734 ],
7735 );
7736
7737 let mut pi = PropertyIndex::new();
7738 pi.enable("Person", "city");
7739 pi.enable("Person", "age");
7740 pi.set("Person", "city", a, &s("austin"));
7742 pi.set("Person", "city", b, &s("austin"));
7743 pi.set("Person", "age", a, &Value::Int(30));
7744 pi.set("Person", "age", b, &Value::Int(30));
7745
7746 let q =
7747 "MATCH (n:Person) WHERE n.city = 'austin' AND n.age = 30 AND n.role = 'eng' RETURN n";
7748 let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
7749 assert_eq!(indexed.len(), 1);
7750 assert_eq!(indexed.get(0, "n"), Some(&s("alice")));
7751
7752 let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7753 assert_eq!(
7754 rows_of(&fallback),
7755 rows_of(&indexed),
7756 "fallback must return identical rows"
7757 );
7758 }
7759}