1use crate::cypher::ast::{
4 ret_val_label, AggArg, AggFunc, Expr, LimitSkip, Operand, OrderItem, OrderTarget, RetItem,
5 RetVal, UnwindExpr,
6};
7use crate::cypher::plan::PlanOp;
8use crate::cypher::RelDir;
9use crate::filter::eval_cmp;
10use crate::result::ResultSet;
11use crate::traverse::{expand, Dir, EdgeRef};
12use crate::value_ops::{cmp_optional, values_equal};
13use crate::view::GraphView;
14use core_storage::{Value, ValueKey};
15use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
16
17#[cfg(test)]
21static FUSED_SCAN_FIRES: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
22
23#[cfg(test)]
27static SCAN_KEY_FIRES: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
28
29#[cfg(test)]
32pub static INDEX_SCAN_FIRES: std::sync::atomic::AtomicUsize =
33 std::sync::atomic::AtomicUsize::new(0);
34
35#[cfg(test)]
39pub static INDEX_INTERSECT_FIRES: std::sync::atomic::AtomicUsize =
40 std::sync::atomic::AtomicUsize::new(0);
41
42pub struct Params<'a>(pub &'a BTreeMap<String, Value>);
45
46#[derive(Clone, Debug)]
47enum Cell {
48 Node(u32),
49 Rel(EdgeRef),
50 Path(u8),
53 Scalar(Value),
55}
56
57type Row = Vec<Option<Cell>>;
60
61struct VarTable {
62 names: Vec<String>,
63}
64
65impl VarTable {
66 fn intern(&mut self, name: &str) -> usize {
67 if let Some(i) = self.names.iter().position(|n| n == name) {
68 return i;
69 }
70 self.names.push(name.to_string());
71 self.names.len() - 1
72 }
73
74 fn slot(&self, name: &str) -> Option<usize> {
75 self.names.iter().position(|n| n == name)
76 }
77}
78
79struct Projected {
80 columns: Vec<String>,
81 rows: Vec<Vec<Option<Value>>>,
82}
83
84const MAX_INTERMEDIATE_ROWS: usize = 1_000_000;
87
88const MAX_GROUPS: usize = 1_000_000;
90
91type GroupKey = Vec<Option<ValueKey>>;
94type GroupEntry = (Vec<Option<Value>>, Vec<AggAcc>);
99
100#[cfg(test)]
101thread_local! {
102 static TEST_MAX_INTERMEDIATE_ROWS: std::cell::Cell<Option<usize>> =
103 const { std::cell::Cell::new(None) };
104 static TEST_EXPAND_PRODUCED: std::cell::Cell<Option<usize>> =
107 const { std::cell::Cell::new(None) };
108 static TEST_MAX_GROUPS: std::cell::Cell<Option<usize>> =
110 const { std::cell::Cell::new(None) };
111}
112
113fn max_intermediate_rows() -> usize {
114 #[cfg(test)]
115 {
116 TEST_MAX_INTERMEDIATE_ROWS
117 .with(|c| c.get())
118 .unwrap_or(MAX_INTERMEDIATE_ROWS)
119 }
120 #[cfg(not(test))]
121 {
122 MAX_INTERMEDIATE_ROWS
123 }
124}
125
126fn max_groups() -> usize {
127 #[cfg(test)]
128 {
129 TEST_MAX_GROUPS.with(|c| c.get()).unwrap_or(MAX_GROUPS)
130 }
131 #[cfg(not(test))]
132 {
133 MAX_GROUPS
134 }
135}
136
137#[cfg(test)]
140pub(crate) fn with_max_groups<R>(cap: usize, f: impl FnOnce() -> R) -> R {
141 TEST_MAX_GROUPS.with(|c| {
142 let prev = c.replace(Some(cap));
143 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
144 c.set(prev);
145 match result {
146 Ok(v) => v,
147 Err(p) => std::panic::resume_unwind(p),
148 }
149 })
150}
151
152#[cfg(test)]
156pub(crate) fn with_max_intermediate_rows<R>(cap: usize, f: impl FnOnce() -> R) -> R {
157 TEST_MAX_INTERMEDIATE_ROWS.with(|c| {
158 let prev = c.replace(Some(cap));
159 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
160 c.set(prev);
161 match result {
162 Ok(v) => v,
163 Err(p) => std::panic::resume_unwind(p),
164 }
165 })
166}
167
168#[cfg(test)]
170fn record_expand_row() {
171 TEST_EXPAND_PRODUCED.with(|c| {
172 if let Some(prev) = c.get() {
173 c.set(Some(prev + 1));
174 }
175 });
176}
177
178#[cfg(test)]
184pub(crate) fn with_expand_counter<R>(f: impl FnOnce() -> R) -> (R, usize) {
185 TEST_EXPAND_PRODUCED.with(|c| c.set(Some(0)));
186 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
187 let count = TEST_EXPAND_PRODUCED.with(|c| c.get().unwrap_or(0));
188 TEST_EXPAND_PRODUCED.with(|c| c.set(None));
189 match result {
190 Ok(v) => (v, count),
191 Err(p) => std::panic::resume_unwind(p),
192 }
193}
194
195fn row_cap_err(cap: usize) -> String {
196 format!(
197 "intermediate result exceeds {cap} rows; add a LIMIT or constrain patterns with shared variables"
198 )
199}
200
201fn group_cap_err() -> String {
202 format!(
203 "group count exceeds {} distinct keys; add a WHERE clause or constrain the grouping key",
204 max_groups()
205 )
206}
207
208fn group_key_normalize(v: &Value) -> Option<ValueKey> {
219 match v {
220 Value::Int(n) => Some(ValueKey::FloatBits((*n as f64).to_bits())),
221 Value::Float(f) => Some(ValueKey::FloatBits(f.to_bits())),
222 _ => ValueKey::from_value(v),
223 }
224}
225
226pub fn execute(view: &GraphView, plan: &[PlanOp], params: &Params) -> Result<ResultSet, String> {
238 use crate::cypher::plan::row_bound;
239 execute_inner(view, plan, params, row_bound(plan))
240}
241
242pub fn execute_union(
247 view: &GraphView,
248 union: &crate::cypher::parser::UnionQuery,
249 params: &Params,
250) -> Result<ResultSet, String> {
251 let mut acc: Option<ResultSet> = None;
252 for (i, part) in union.parts.iter().enumerate() {
253 let ops = crate::cypher::plan::plan(part)?;
254 let rs = execute(view, &ops, params)?;
255 acc = Some(match acc {
256 None => rs,
257 Some(prev) => {
258 if prev.columns() != rs.columns() {
259 return Err(format!(
260 "UNION requires matching column names across all parts; \
261 got {:?} then {:?}",
262 prev.columns(),
263 rs.columns()
264 ));
265 }
266 let all = union.all_flags.get(i - 1).copied().unwrap_or(false);
268 combine_result_sets(prev, rs, all)
269 }
270 });
271 }
272 Ok(acc.expect("UNION query has at least one part"))
274}
275
276fn combine_result_sets(mut acc: ResultSet, other: ResultSet, all: bool) -> ResultSet {
277 for i in 0..other.len() {
278 acc.push_row(other.row(i).to_vec());
279 }
280 if !all {
281 let mut seen: Vec<Vec<Option<Value>>> = Vec::new();
283 for i in 0..acc.len() {
284 let r = acc.row(i).to_vec();
285 if !seen.contains(&r) {
286 seen.push(r);
287 }
288 }
289 let mut deduped = ResultSet::new(acc.columns().to_vec());
290 for r in seen {
291 deduped.push_row(r);
292 }
293 return deduped;
294 }
295 acc
296}
297
298#[cfg(test)]
303pub(crate) fn execute_unbounded(
304 view: &GraphView,
305 plan: &[PlanOp],
306 params: &Params,
307) -> Result<ResultSet, String> {
308 execute_inner(view, plan, params, None)
309}
310
311fn execute_inner(
312 view: &GraphView,
313 plan: &[PlanOp],
314 params: &Params,
315 row_bound: Option<usize>,
316) -> Result<ResultSet, String> {
317 check_params(plan, params)?;
318
319 let has_var_expand = plan
325 .iter()
326 .any(|op| matches!(op, PlanOp::VarExpand { .. } | PlanOp::ShortestPath { .. }));
327
328 let is_pipeline = plan.iter().any(|op| {
345 matches!(
346 op,
347 PlanOp::With { .. } | PlanOp::Unwind { .. } | PlanOp::LeftOuterApply { .. }
348 )
349 }) || {
350 let mut saw_gagg = false;
351 plan.iter().any(|op| {
352 if matches!(op, PlanOp::GroupAggregate { .. }) {
353 saw_gagg = true;
354 false
355 } else {
356 saw_gagg && matches!(op, PlanOp::Filter { .. } | PlanOp::Project { .. })
357 }
358 })
359 };
360
361 if !has_var_expand
365 && !is_pipeline
366 && plan
367 .iter()
368 .any(|op| matches!(op, PlanOp::GroupAggregate { .. }))
369 {
370 return execute_group_aggregate(view, plan, params);
371 }
372
373 if !has_var_expand
377 && !is_pipeline
378 && plan.iter().any(|op| matches!(op, PlanOp::Aggregate { .. }))
379 {
380 return execute_aggregate(view, plan, params);
381 }
382
383 if let Some(bound) = row_bound {
387 return execute_pull(view, plan, params, bound);
388 }
389
390 let vars = collect_vars(plan);
392 let mut rows: Vec<Row> = vec![vec![None; vars.names.len()]];
393 let mut projected: Option<Projected> = None;
394 let mut pipeline_group_columns: Option<Vec<String>> = None;
397
398 for op in plan {
399 match op {
400 PlanOp::ScanLabel { var, label } => {
401 rows = scan_label(view, &vars, &rows, var, label.as_deref())?;
402 }
403 PlanOp::ScanKey { var, key, label } => {
404 rows = scan_key(view, &vars, &rows, var, key, label.as_deref(), params)?;
405 }
406 PlanOp::IndexScan {
407 var,
408 label,
409 field,
410 value,
411 } => {
412 rows = scan_index(
413 view,
414 &vars,
415 &rows,
416 var,
417 label.as_deref(),
418 field,
419 value,
420 params,
421 )?;
422 }
423 PlanOp::IndexIntersect {
424 var,
425 label,
426 equalities,
427 } => {
428 rows = scan_intersect(
429 view,
430 &vars,
431 &rows,
432 var,
433 label.as_deref(),
434 equalities,
435 params,
436 )?;
437 }
438 PlanOp::LookupProps { var, props } => {
439 rows = retain_node(view, &vars, &rows, var, None, props, params)?;
440 }
441 PlanOp::JoinBound { var, label, props } => {
442 rows = retain_node(view, &vars, &rows, var, label.as_deref(), props, params)?;
443 }
444 PlanOp::Expand { .. } => {
445 rows = exec_expand(view, &vars, &rows, op, params)?;
446 }
447 PlanOp::VarExpand {
448 from,
449 rel_var,
450 etypes,
451 dir,
452 to,
453 min,
454 max,
455 } => {
456 rows = exec_var_expand(
457 view, &vars, &rows, from, rel_var, etypes, *dir, to, *min, *max,
458 )?;
459 }
460 PlanOp::ShortestPath {
461 from,
462 rel_var,
463 etypes,
464 dir,
465 to,
466 max_hops,
467 } => {
468 rows = exec_shortest_path(
469 view, &vars, &rows, from, rel_var, etypes, *dir, to, *max_hops,
470 )?;
471 }
472 PlanOp::Filter { expr } => {
473 rows = exec_filter(view, &vars, &rows, expr, params)?;
474 }
475 PlanOp::Project { items } => {
476 projected = Some(exec_project(view, &vars, &rows, items, params)?);
477 }
478 PlanOp::Distinct => {
479 if let Some(table) = projected.as_mut() {
480 exec_distinct(table)?;
481 } else {
482 return Err("DISTINCT requires a Project".into());
483 }
484 }
485 PlanOp::OrderBy { items } => {
486 if let Some(table) = projected.as_mut() {
487 exec_order_by(table, items)?;
488 } else {
489 exec_order_by_rows(&vars, &mut rows, items, view);
491 }
492 }
493 PlanOp::Skip(ls) => {
494 let n = resolve_ls(ls, params)?;
495 if let Some(table) = projected.as_mut() {
496 apply_skip(&mut table.rows, n);
497 } else {
498 apply_skip(&mut rows, n);
499 }
500 }
501 PlanOp::Limit(ls) => {
502 let n = resolve_ls(ls, params)?;
503 if let Some(table) = projected.as_mut() {
504 apply_limit(&mut table.rows, n);
505 } else {
506 apply_limit(&mut rows, n);
507 }
508 }
509 PlanOp::GroupAggregate { keys, aggs } => {
514 let mut grp_groups: HashMap<GroupKey, GroupEntry> = HashMap::new();
515 let mut grp_key_order: Vec<GroupKey> = Vec::new();
516 let mut grp_cells: HashMap<GroupKey, Vec<Option<Cell>>> = HashMap::new();
523 for row in &rows {
524 let mut gk: GroupKey = Vec::with_capacity(keys.len());
525 let mut display_vals: Vec<Option<Value>> = Vec::with_capacity(keys.len());
526 for (_, item) in keys {
527 let val = project_item(view, &vars, row, item, params)?;
528 gk.push(val.as_ref().and_then(group_key_normalize));
529 display_vals.push(val);
530 }
531 if !grp_groups.contains_key(&gk) {
532 if grp_groups.len() >= max_groups() {
533 return Err(group_cap_err());
534 }
535 grp_key_order.push(gk.clone());
536 grp_cells.insert(gk.clone(), key_source_cells(&vars, row, keys));
537 grp_groups.insert(
538 gk.clone(),
539 (
540 display_vals,
541 aggs.iter().map(|(f, a, _)| AggAcc::for_arg(f, a)).collect(),
542 ),
543 );
544 }
545 let (_, accs) = grp_groups.get_mut(&gk).unwrap();
546 for (acc, (func, arg, _)) in accs.iter_mut().zip(aggs.iter()) {
547 update_acc(view, &vars, row, func, arg, acc)?;
548 }
549 }
550 if keys.is_empty() && grp_key_order.is_empty() {
553 let empty_key: GroupKey = vec![];
554 grp_key_order.push(empty_key.clone());
555 grp_groups.insert(
556 empty_key,
557 (
558 vec![],
559 aggs.iter().map(|(f, a, _)| AggAcc::for_arg(f, a)).collect(),
560 ),
561 );
562 }
563 if is_pipeline {
564 rows = group_result_to_rows(
567 keys,
568 aggs,
569 grp_key_order,
570 &mut grp_groups,
571 &mut grp_cells,
572 &vars,
573 );
574 projected = None;
575 let mut cols: Vec<String> = keys.iter().map(|(c, _)| c.clone()).collect();
578 cols.extend(aggs.iter().map(|(_, _, c)| c.clone()));
579 pipeline_group_columns = Some(cols);
580 } else {
581 projected = Some(build_group_projected(
582 keys,
583 aggs,
584 grp_key_order,
585 &mut grp_groups,
586 ));
587 pipeline_group_columns = None;
588 }
589 }
590 PlanOp::With {
593 items,
594 where_expr,
595 order_by,
596 skip,
597 limit,
598 } => {
599 let row_len = vars.names.len();
603 let mut new_rows: Vec<Row> = Vec::with_capacity(rows.len());
604 for row in &rows {
605 let mut new_row: Row = vec![None; row_len];
606 for item in items {
607 let col = column_name(item);
608 let Some(dst_slot) = vars.slot(&col) else {
609 continue;
610 };
611 match &item.value {
612 RetVal::Var(v) => {
613 if let Some(src_slot) = vars.slot(v) {
615 new_row[dst_slot] = row.get(src_slot).cloned().flatten();
616 }
617 }
618 RetVal::Prop { var, field } => {
619 let val = resolve_prop(view, &vars, row, var, field)?;
620 new_row[dst_slot] = val.map(Cell::Scalar);
621 }
622 RetVal::Agg { .. } => {} RetVal::FuncCall { name, args } => {
624 let val = eval_func(name, args, view, &vars, row, params)?;
625 new_row[dst_slot] = val.map(Cell::Scalar);
626 }
627 RetVal::ScalarExpr(op) => {
628 let val = resolve_operand(view, &vars, row, op, params)?;
629 new_row[dst_slot] = val.map(Cell::Scalar);
630 }
631 }
632 }
633 new_rows.push(new_row);
634 }
635 rows = new_rows;
636 if let Some(expr) = where_expr {
638 rows = exec_filter(view, &vars, &rows, expr, params)?;
639 }
640 if !order_by.is_empty() {
642 exec_order_by_rows(&vars, &mut rows, order_by, view);
643 }
644 if let Some(ls) = skip {
645 let n = resolve_ls(ls, params)?;
646 apply_skip(&mut rows, n);
647 }
648 if let Some(ls) = limit {
649 let n = resolve_ls(ls, params)?;
650 apply_limit(&mut rows, n);
651 }
652 }
653 PlanOp::Unwind { expr, alias } => {
655 let alias_slot = vars
656 .slot(alias)
657 .ok_or_else(|| format!("UNWIND alias `{alias}` not in VarTable"))?;
658 let cap = max_intermediate_rows();
659 let mut new_rows: Vec<Row> = Vec::new();
660 for row in &rows {
661 let list_val: Option<Value> = match expr {
662 UnwindExpr::Lit(vals) => Some(Value::List(vals.clone())),
663 UnwindExpr::Prop { var, field } => {
664 resolve_prop(view, &vars, row, var, field)?
665 }
666 UnwindExpr::Var(name) => {
667 let slot = vars
668 .slot(name)
669 .ok_or_else(|| format!("UNWIND variable `{name}` is not bound"))?;
670 match row.get(slot).and_then(|c| c.as_ref()) {
671 Some(Cell::Scalar(v)) => Some(v.clone()),
672 Some(Cell::Node(_) | Cell::Rel(_) | Cell::Path(_)) => {
673 return Err(format!(
674 "UNWIND requires a list; `{name}` is bound to a graph element"
675 ));
676 }
677 None => None,
678 }
679 }
680 };
681 match list_val {
682 None => {} Some(Value::List(items_list)) => {
684 for item_val in items_list {
686 if new_rows.len() >= cap {
687 return Err(row_cap_err(cap));
688 }
689 let mut new_row = row.clone();
690 new_row[alias_slot] = Some(Cell::Scalar(item_val));
691 new_rows.push(new_row);
692 }
693 }
694 Some(other) => {
695 let type_name = match &other {
696 Value::Int(_) => "Int",
697 Value::Float(_) => "Float",
698 Value::Str(_) => "Str",
699 Value::Bool(_) => "Bool",
700 Value::List(_) => unreachable!(),
701 Value::Map(_) => "Map",
702 };
703 return Err(format!(
704 "UNWIND requires a list; got {type_name} value for `{alias}`"
705 ));
706 }
707 }
708 }
709 rows = new_rows;
710 }
711 PlanOp::Aggregate { func, arg, column } => {
716 let ctx = AggStreamCtx {
717 view,
718 vars: &vars,
719 params,
720 func,
721 arg,
722 };
723 let mut acc = AggAcc::for_arg(func, arg);
724 for row in &rows {
725 agg_stream(&ctx, &[], row, &mut acc)?;
727 }
728 let value = acc.finish();
729 let mut rs = ResultSet::new(vec![column.clone()]);
730 rs.push_row(vec![value]);
731 return Ok(rs);
732 }
733 PlanOp::LeftOuterApply {
739 inner,
740 optional_vars,
741 } => {
742 let cap = max_intermediate_rows();
743 let mut new_rows: Vec<Row> = Vec::new();
744 for outer_row in &rows {
745 let inner_seed: Vec<Row> = vec![outer_row.clone()];
747 let inner_result =
750 exec_left_outer_inner(view, &vars, inner_seed, inner, params)?;
751 if inner_result.is_empty() {
752 let mut null_row = outer_row.clone();
754 for v in optional_vars {
755 if let Some(slot) = vars.slot(v) {
756 null_row[slot] = None;
757 }
758 }
759 if new_rows.len() >= cap {
760 return Err(row_cap_err(cap));
761 }
762 new_rows.push(null_row);
763 } else {
764 for r in inner_result {
765 if new_rows.len() >= cap {
766 return Err(row_cap_err(cap));
767 }
768 new_rows.push(r);
769 }
770 }
771 }
772 rows = new_rows;
773 }
774 }
775 }
776
777 Ok(match projected {
778 Some(table) => finish(table),
779 None => {
780 if let Some(cols) = pipeline_group_columns {
783 let mut rs = ResultSet::new(cols.clone());
784 for row in rows {
785 let vals: Vec<Option<Value>> = cols
786 .iter()
787 .map(|col| {
788 vars.slot(col)
789 .and_then(|s| row.get(s))
790 .and_then(|c| c.as_ref())
791 .and_then(|c| match c {
792 Cell::Scalar(v) => Some(v.clone()),
793 Cell::Node(id) => {
794 view.ids.key_of(*id).map(|k| Value::Str(k.to_owned()))
795 }
796 Cell::Path(h) => Some(Value::Int(*h as i64)),
797 Cell::Rel(_) => None,
798 })
799 })
800 .collect();
801 rs.push_row(vals);
802 }
803 rs
804 } else {
805 ResultSet::new(vec![])
806 }
807 }
808 })
809}
810
811fn exec_left_outer_inner(
820 view: &GraphView,
821 vars: &VarTable,
822 mut rows: Vec<Row>,
823 inner: &[PlanOp],
824 params: &Params,
825) -> Result<Vec<Row>, String> {
826 for op in inner {
827 match op {
828 PlanOp::ScanLabel { var, label } => {
829 rows = scan_label(view, vars, &rows, var, label.as_deref())?;
830 }
831 PlanOp::ScanKey { var, key, label } => {
832 rows = scan_key(view, vars, &rows, var, key, label.as_deref(), params)?;
833 }
834 PlanOp::LookupProps { var, props } => {
835 rows = retain_node(view, vars, &rows, var, None, props, params)?;
836 }
837 PlanOp::JoinBound { var, label, props } => {
838 rows = retain_node(view, vars, &rows, var, label.as_deref(), props, params)?;
839 }
840 PlanOp::Expand { .. } => {
841 rows = exec_expand(view, vars, &rows, op, params)?;
842 }
843 PlanOp::Filter { expr } => {
844 rows = exec_filter(view, vars, &rows, expr, params)?;
845 }
846 other => {
847 return Err(format!(
848 "unsupported op inside OPTIONAL MATCH inner plan: {other:?}"
849 ));
850 }
851 }
852 }
853 Ok(rows)
854}
855
856fn finish(table: Projected) -> ResultSet {
857 let mut rs = ResultSet::new(table.columns);
858 for row in table.rows {
859 rs.push_row(row);
860 }
861 rs
862}
863
864fn check_params(plan: &[PlanOp], params: &Params) -> Result<(), String> {
865 let mut names = Vec::new();
866 let mut seen = BTreeSet::new();
867 collect_params_from_ops(plan, &mut names, &mut seen)?;
868 for name in names {
869 if !params.0.contains_key(&name) {
870 return Err(format!("missing parameter `{name}`"));
871 }
872 }
873 Ok(())
874}
875
876fn collect_params_from_ops(
877 plan: &[PlanOp],
878 names: &mut Vec<String>,
879 seen: &mut BTreeSet<String>,
880) -> Result<(), String> {
881 for op in plan {
882 match op {
883 PlanOp::ScanKey { key, .. } => collect_operand(key, names, seen),
884 PlanOp::LookupProps { props, .. }
885 | PlanOp::JoinBound { props, .. }
886 | PlanOp::Expand {
887 to_props: props, ..
888 } => {
889 for (_, operand) in props {
890 collect_operand(operand, names, seen);
891 }
892 }
893 PlanOp::Filter { expr } => collect_expr(expr, names, seen, 0)?,
894 PlanOp::LeftOuterApply { inner, .. } => {
895 collect_params_from_ops(inner, names, seen)?;
896 }
897 PlanOp::Project { items } => {
900 for item in items {
901 collect_ret_item_params(item, names, seen);
902 }
903 }
904 PlanOp::With {
905 items, where_expr, ..
906 } => {
907 for item in items {
908 collect_ret_item_params(item, names, seen);
909 }
910 if let Some(expr) = where_expr {
911 let _ = collect_expr(expr, names, seen, 0);
912 }
913 }
914 PlanOp::GroupAggregate { keys, aggs } => {
915 for (_, item) in keys {
916 collect_ret_item_params(item, names, seen);
917 }
918 for (_, arg, _) in aggs {
919 match arg {
920 AggArg::Star => {}
921 AggArg::Var(_) => {}
922 AggArg::Prop { .. } => {}
923 AggArg::Distinct(_) => {}
924 }
925 }
926 }
927 PlanOp::Skip(LimitSkip::Param(n)) | PlanOp::Limit(LimitSkip::Param(n))
930 if seen.insert(n.clone()) =>
931 {
932 names.push(n.clone());
933 }
934 _ => {}
935 }
936 }
937 Ok(())
938}
939
940fn collect_ret_item_params(item: &RetItem, names: &mut Vec<String>, seen: &mut BTreeSet<String>) {
942 match &item.value {
943 RetVal::FuncCall { args, .. } => {
944 for arg in args {
945 collect_operand(arg, names, seen);
946 }
947 }
948 RetVal::ScalarExpr(op) => {
949 collect_operand(op, names, seen);
950 }
951 RetVal::Prop { .. } | RetVal::Var(_) | RetVal::Agg { .. } => {}
952 }
953}
954
955fn collect_operand(op: &Operand, names: &mut Vec<String>, seen: &mut BTreeSet<String>) {
956 match op {
957 Operand::Param(n) => {
958 if seen.insert(n.clone()) {
959 names.push(n.clone());
960 }
961 }
962 Operand::FuncCall { args, .. } => {
963 for arg in args {
964 collect_operand(arg, names, seen);
965 }
966 }
967 Operand::BinArith { left, right, .. } => {
968 collect_operand(left, names, seen);
969 collect_operand(right, names, seen);
970 }
971 _ => {}
972 }
973}
974
975fn collect_expr(
976 expr: &Expr,
977 names: &mut Vec<String>,
978 seen: &mut BTreeSet<String>,
979 depth: u32,
980) -> Result<(), String> {
981 if depth > 256 {
982 return Err("expression nesting too deep".into());
983 }
984 match expr {
985 Expr::And(lhs, rhs) | Expr::Or(lhs, rhs) => {
986 collect_expr(lhs, names, seen, depth + 1)?;
987 collect_expr(rhs, names, seen, depth + 1)
988 }
989 Expr::Not(inner) => collect_expr(inner, names, seen, depth + 1),
990 Expr::Cmp { lhs, rhs, .. } => {
991 collect_operand(lhs, names, seen);
992 collect_operand(rhs, names, seen);
993 Ok(())
994 }
995 Expr::Truthy(op) => {
996 collect_operand(op, names, seen);
997 Ok(())
998 }
999 Expr::IsNull(op) | Expr::IsNotNull(op) => {
1000 collect_operand(op, names, seen);
1001 Ok(())
1002 }
1003 Expr::In { expr, list } => {
1004 collect_operand(expr, names, seen);
1005 for item in list {
1006 collect_operand(item, names, seen);
1007 }
1008 Ok(())
1009 }
1010 }
1011}
1012
1013fn collect_vars(plan: &[PlanOp]) -> VarTable {
1014 let mut vars = VarTable { names: Vec::new() };
1015 for op in plan {
1016 match op {
1017 PlanOp::ScanLabel { var, .. } => {
1018 vars.intern(var);
1019 }
1020 PlanOp::ScanKey { var, key, .. } => {
1021 vars.intern(var);
1022 intern_operand(&mut vars, key);
1023 }
1024 PlanOp::IndexScan { var, value, .. } => {
1025 vars.intern(var);
1026 intern_operand(&mut vars, value);
1027 }
1028 PlanOp::IndexIntersect {
1029 var, equalities, ..
1030 } => {
1031 vars.intern(var);
1032 for (_, operand) in equalities {
1033 intern_operand(&mut vars, operand);
1034 }
1035 }
1036 PlanOp::LookupProps { var, props } | PlanOp::JoinBound { var, props, .. } => {
1037 vars.intern(var);
1038 for (_, operand) in props {
1039 intern_operand(&mut vars, operand);
1040 }
1041 }
1042 PlanOp::Expand {
1043 from,
1044 rel_var,
1045 to,
1046 to_props,
1047 ..
1048 } => {
1049 vars.intern(from);
1050 vars.intern(to);
1051 if let Some(r) = rel_var {
1052 vars.intern(r);
1053 }
1054 for (_, operand) in to_props {
1055 intern_operand(&mut vars, operand);
1056 }
1057 }
1058 PlanOp::VarExpand {
1059 from, rel_var, to, ..
1060 } => {
1061 vars.intern(from);
1062 vars.intern(to);
1063 if let Some(r) = rel_var {
1064 vars.intern(r);
1065 }
1066 }
1067 PlanOp::ShortestPath {
1068 from, rel_var, to, ..
1069 } => {
1070 vars.intern(from);
1071 vars.intern(to);
1072 if let Some(r) = rel_var {
1073 vars.intern(r);
1074 }
1075 }
1076 PlanOp::Filter { expr } => intern_expr(&mut vars, expr),
1077 PlanOp::Project { items } => {
1078 for item in items {
1079 match &item.value {
1080 RetVal::Var(name) | RetVal::Prop { var: name, .. } => {
1081 vars.intern(name);
1082 }
1083 RetVal::Agg { .. } => {} 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 }
1094 }
1095 PlanOp::Aggregate { arg, .. } => intern_agg_arg(&mut vars, arg),
1096 PlanOp::GroupAggregate { keys, aggs } => {
1097 for (col, item) in keys {
1099 match &item.value {
1100 RetVal::Var(name) | RetVal::Prop { var: name, .. } => {
1101 vars.intern(name);
1102 }
1103 RetVal::Agg { .. } => {}
1104 RetVal::FuncCall { args, .. } => {
1105 for arg in args {
1106 intern_operand(&mut vars, arg);
1107 }
1108 }
1109 RetVal::ScalarExpr(op) => {
1110 intern_operand(&mut vars, op);
1111 }
1112 }
1113 vars.intern(col);
1115 }
1116 for (_, arg, col) in aggs {
1117 intern_agg_arg(&mut vars, arg);
1118 vars.intern(col);
1120 }
1121 }
1122 PlanOp::With {
1123 items,
1124 where_expr,
1125 order_by,
1126 ..
1127 } => {
1128 for item in items {
1129 match &item.value {
1130 RetVal::Var(name) | RetVal::Prop { var: name, .. } => {
1131 vars.intern(name);
1132 }
1133 RetVal::Agg { .. } => {}
1134 RetVal::FuncCall { args, .. } => {
1135 for arg in args {
1136 intern_operand(&mut vars, arg);
1137 }
1138 }
1139 RetVal::ScalarExpr(op) => {
1140 intern_operand(&mut vars, op);
1141 }
1142 }
1143 if let Some(alias) = &item.alias {
1148 vars.intern(alias);
1149 } else if let Some(col) = ret_val_label(&item.value) {
1150 vars.intern(&col);
1151 }
1152 }
1153 if let Some(expr) = where_expr {
1154 intern_expr(&mut vars, expr);
1155 }
1156 for oi in order_by {
1157 match &oi.target {
1158 OrderTarget::Alias(name) | OrderTarget::Var(name) => {
1159 vars.intern(name);
1160 }
1161 OrderTarget::Prop { var, .. } => {
1162 vars.intern(var);
1163 }
1164 }
1165 }
1166 }
1167 PlanOp::Unwind { expr, alias } => {
1168 vars.intern(alias);
1169 match expr {
1170 UnwindExpr::Prop { var, .. } => {
1171 vars.intern(var);
1172 }
1173 UnwindExpr::Var(name) => {
1174 vars.intern(name);
1175 }
1176 UnwindExpr::Lit(_) => {}
1177 }
1178 }
1179 PlanOp::LeftOuterApply {
1180 inner,
1181 optional_vars,
1182 } => {
1183 for op in inner {
1185 match op {
1189 PlanOp::ScanLabel { var, .. } => {
1190 vars.intern(var);
1191 }
1192 PlanOp::ScanKey { var, key, .. } => {
1193 vars.intern(var);
1194 intern_operand(&mut vars, key);
1195 }
1196 PlanOp::IndexScan { var, value, .. } => {
1197 vars.intern(var);
1198 intern_operand(&mut vars, value);
1199 }
1200 PlanOp::IndexIntersect {
1201 var, equalities, ..
1202 } => {
1203 vars.intern(var);
1204 for (_, operand) in equalities {
1205 intern_operand(&mut vars, operand);
1206 }
1207 }
1208 PlanOp::Expand {
1209 from, rel_var, to, ..
1210 } => {
1211 vars.intern(from);
1212 vars.intern(to);
1213 if let Some(r) = rel_var {
1214 vars.intern(r);
1215 }
1216 }
1217 PlanOp::JoinBound { var, .. } | PlanOp::LookupProps { var, .. } => {
1218 vars.intern(var);
1219 }
1220 PlanOp::Filter { expr } => intern_expr(&mut vars, expr),
1221 _ => {}
1222 }
1223 }
1224 for v in optional_vars {
1225 vars.intern(v);
1226 }
1227 }
1228 _ => {}
1229 }
1230 }
1231 vars
1232}
1233
1234fn intern_operand(vars: &mut VarTable, operand: &Operand) {
1235 match operand {
1236 Operand::Prop { var, .. } | Operand::Var(var) => {
1237 vars.intern(var);
1238 }
1239 Operand::Lit(_) | Operand::Param(_) => {}
1240 Operand::BinArith { left, right, .. } => {
1241 intern_operand(vars, left);
1242 intern_operand(vars, right);
1243 }
1244 Operand::FuncCall { args, .. } => {
1245 for arg in args {
1246 intern_operand(vars, arg);
1247 }
1248 }
1249 Operand::Case { branches, default } => {
1250 for (cond, value) in branches {
1251 intern_expr(vars, cond);
1252 intern_operand(vars, value);
1253 }
1254 if let Some(d) = default {
1255 intern_operand(vars, d);
1256 }
1257 }
1258 Operand::Index { base, index } => {
1259 intern_operand(vars, base);
1260 intern_operand(vars, index);
1261 }
1262 }
1263}
1264
1265fn intern_agg_arg(vars: &mut VarTable, arg: &AggArg) {
1267 match arg {
1268 AggArg::Star => {}
1269 AggArg::Var(v) => {
1270 vars.intern(v);
1271 }
1272 AggArg::Prop { var, .. } => {
1273 vars.intern(var);
1274 }
1275 AggArg::Distinct(inner) => intern_agg_arg(vars, inner),
1276 }
1277}
1278
1279fn agg_arg_label(arg: &AggArg) -> String {
1281 match arg {
1282 AggArg::Star => "*".to_string(),
1283 AggArg::Var(v) => v.clone(),
1284 AggArg::Prop { var, field } => format!("{var}.{field}"),
1285 AggArg::Distinct(inner) => format!("DISTINCT {}", agg_arg_label(inner)),
1286 }
1287}
1288
1289fn intern_expr(vars: &mut VarTable, expr: &Expr) {
1290 match expr {
1291 Expr::And(lhs, rhs) | Expr::Or(lhs, rhs) => {
1292 intern_expr(vars, lhs);
1293 intern_expr(vars, rhs);
1294 }
1295 Expr::Not(inner) => intern_expr(vars, inner),
1296 Expr::Cmp { lhs, rhs, .. } => {
1297 intern_operand(vars, lhs);
1298 intern_operand(vars, rhs);
1299 }
1300 Expr::Truthy(op) => intern_operand(vars, op),
1301 Expr::IsNull(op) | Expr::IsNotNull(op) => intern_operand(vars, op),
1302 Expr::In { expr, list } => {
1303 intern_operand(vars, expr);
1304 for item in list {
1305 intern_operand(vars, item);
1306 }
1307 }
1308 }
1309}
1310
1311fn scan_ids(view: &GraphView, label: Option<&str>) -> Vec<u32> {
1312 let ids = match label {
1313 Some(label) => view.nodes_with_label(label),
1314 None => (0..view.ids.len() as u32)
1316 .filter(|&id| view.label_of(id).is_some())
1317 .collect(),
1318 };
1319 if view.mask.is_some() {
1320 ids.into_iter().filter(|&id| view.visible(id)).collect()
1321 } else {
1322 ids
1323 }
1324}
1325
1326fn resolve_scan_key_id(
1329 view: &GraphView,
1330 vars: &VarTable,
1331 row: &Row,
1332 key: &Operand,
1333 label: Option<&str>,
1334 params: &Params,
1335) -> Result<Option<u32>, String> {
1336 #[cfg(test)]
1337 SCAN_KEY_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1338
1339 let Some(val) = resolve_operand(view, vars, row, key, params)? else {
1340 return Ok(None);
1341 };
1342 let Value::Str(s) = val else {
1343 return Ok(None);
1344 };
1345 let Some(id) = view.node_id(&s) else {
1346 return Ok(None);
1347 };
1348 if !view.visible(id) {
1349 return Ok(None);
1350 }
1351 if let Some(want) = label {
1352 match view.label_of(id) {
1353 Some(got) if got == want => {}
1354 _ => return Ok(None),
1355 }
1356 }
1357 Ok(Some(id))
1358}
1359
1360fn scan_key(
1361 view: &GraphView,
1362 vars: &VarTable,
1363 rows: &[Row],
1364 var: &str,
1365 key: &Operand,
1366 label: Option<&str>,
1367 params: &Params,
1368) -> Result<Vec<Row>, String> {
1369 let slot = vars
1370 .slot(var)
1371 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1372 let cap = max_intermediate_rows();
1373 let mut out = Vec::new();
1374 for row in rows {
1375 let Some(id) = resolve_scan_key_id(view, vars, row, key, label, params)? else {
1376 continue;
1377 };
1378 if out.len() >= cap {
1379 return Err(row_cap_err(cap));
1380 }
1381 let mut next = row.clone();
1382 next[slot] = Some(Cell::Node(id));
1383 out.push(next);
1384 }
1385 Ok(out)
1386}
1387
1388fn scan_label(
1389 view: &GraphView,
1390 vars: &VarTable,
1391 rows: &[Row],
1392 var: &str,
1393 label: Option<&str>,
1394) -> Result<Vec<Row>, String> {
1395 let ids = scan_ids(view, label);
1396 let slot = vars
1397 .slot(var)
1398 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1399 let cap = max_intermediate_rows();
1400 let mut out = Vec::with_capacity(rows.len().saturating_mul(ids.len()).min(cap));
1401 for row in rows {
1402 for &id in &ids {
1403 if out.len() >= cap {
1404 return Err(row_cap_err(cap));
1405 }
1406 let mut next = row.clone();
1407 next[slot] = Some(Cell::Node(id));
1408 out.push(next);
1409 }
1410 }
1411 Ok(out)
1412}
1413
1414#[allow(clippy::too_many_arguments)]
1422fn index_scan_ids(
1423 view: &GraphView,
1424 vars: &VarTable,
1425 row: &Row,
1426 label: Option<&str>,
1427 field: &str,
1428 value: &Operand,
1429 params: &Params,
1430) -> Result<Vec<u32>, String> {
1431 let resolved = resolve_operand(view, vars, row, value, params)?;
1432 if let (Some(label_str), Some(val)) = (label, resolved.as_ref()) {
1433 if let Some(ids) = view.nodes_with_prop(label_str, field, val) {
1434 #[cfg(test)]
1435 INDEX_SCAN_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1436 return Ok(ids);
1437 }
1438 }
1439 let props = [(field.to_string(), value.clone())];
1441 let mut out = Vec::new();
1442 for id in scan_ids(view, label) {
1443 if node_matches(view, vars, row, id, None, &props, params)? {
1444 out.push(id);
1445 }
1446 }
1447 Ok(out)
1448}
1449
1450#[allow(clippy::too_many_arguments)]
1453fn scan_index(
1454 view: &GraphView,
1455 vars: &VarTable,
1456 rows: &[Row],
1457 var: &str,
1458 label: Option<&str>,
1459 field: &str,
1460 value: &Operand,
1461 params: &Params,
1462) -> Result<Vec<Row>, String> {
1463 let Some(first) = rows.first() else {
1464 return Ok(Vec::new());
1465 };
1466 let ids = index_scan_ids(view, vars, first, label, field, value, params)?;
1467 let slot = vars
1468 .slot(var)
1469 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1470 let cap = max_intermediate_rows();
1471 let mut out = Vec::new();
1472 for row in rows {
1473 for &id in &ids {
1474 if out.len() >= cap {
1475 return Err(row_cap_err(cap));
1476 }
1477 let mut next = row.clone();
1478 next[slot] = Some(Cell::Node(id));
1479 out.push(next);
1480 }
1481 }
1482 Ok(out)
1483}
1484
1485#[allow(clippy::too_many_arguments)]
1494fn index_intersect_ids(
1495 view: &GraphView,
1496 vars: &VarTable,
1497 row: &Row,
1498 label: Option<&str>,
1499 equalities: &[(String, Operand)],
1500 params: &Params,
1501) -> Result<Vec<u32>, String> {
1502 let mut resolved: Vec<(String, Option<Value>)> = Vec::with_capacity(equalities.len());
1504 for (field, operand) in equalities {
1505 let val = resolve_operand(view, vars, row, operand, params)?;
1506 resolved.push((field.clone(), val));
1507 }
1508
1509 let mut indexed_lists: Vec<Vec<u32>> = Vec::new();
1511 let mut unindexed_props: Vec<(String, Operand)> = Vec::new();
1512
1513 for ((field, val_opt), (_, operand)) in resolved.iter().zip(equalities.iter()) {
1514 if let (Some(label_str), Some(val)) = (label, val_opt.as_ref()) {
1515 if let Some(ids) = view.nodes_with_prop(label_str, field, val) {
1516 indexed_lists.push(ids);
1517 continue;
1518 }
1519 }
1520 unindexed_props.push((field.clone(), operand.clone()));
1521 }
1522
1523 if indexed_lists.is_empty() {
1524 let mut out = Vec::new();
1526 for id in scan_ids(view, label) {
1527 if node_matches(view, vars, row, id, None, equalities, params)? {
1528 out.push(id);
1529 }
1530 }
1531 return Ok(out);
1532 }
1533
1534 #[cfg(test)]
1535 INDEX_INTERSECT_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1536
1537 indexed_lists.sort_unstable_by_key(|v| v.len());
1539 let mut result = indexed_lists.remove(0);
1540 for other in indexed_lists {
1541 let mut merged = Vec::new();
1542 let (mut i, mut j) = (0, 0);
1543 while i < result.len() && j < other.len() {
1544 match result[i].cmp(&other[j]) {
1545 std::cmp::Ordering::Equal => {
1546 merged.push(result[i]);
1547 i += 1;
1548 j += 1;
1549 }
1550 std::cmp::Ordering::Less => i += 1,
1551 std::cmp::Ordering::Greater => j += 1,
1552 }
1553 }
1554 result = merged;
1555 }
1556
1557 if !unindexed_props.is_empty() {
1561 let mut filtered = Vec::with_capacity(result.len());
1562 for id in result {
1563 if node_matches(view, vars, row, id, None, &unindexed_props, params)? {
1564 filtered.push(id);
1565 }
1566 }
1567 result = filtered;
1568 }
1569
1570 Ok(result)
1571}
1572
1573#[allow(clippy::too_many_arguments)]
1576fn scan_intersect(
1577 view: &GraphView,
1578 vars: &VarTable,
1579 rows: &[Row],
1580 var: &str,
1581 label: Option<&str>,
1582 equalities: &[(String, Operand)],
1583 params: &Params,
1584) -> Result<Vec<Row>, String> {
1585 let Some(first) = rows.first() else {
1586 return Ok(Vec::new());
1587 };
1588 let ids = index_intersect_ids(view, vars, first, label, equalities, params)?;
1589 let slot = vars
1590 .slot(var)
1591 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1592 let cap = max_intermediate_rows();
1593 let mut out = Vec::new();
1594 for row in rows {
1595 for &id in &ids {
1596 if out.len() >= cap {
1597 return Err(row_cap_err(cap));
1598 }
1599 let mut next = row.clone();
1600 next[slot] = Some(Cell::Node(id));
1601 out.push(next);
1602 }
1603 }
1604 Ok(out)
1605}
1606
1607fn require_cell<'a>(row: &'a Row, vars: &VarTable, var: &str) -> Result<&'a Cell, String> {
1608 let slot = vars
1609 .slot(var)
1610 .ok_or_else(|| format!("unbound variable `{var}`"))?;
1611 row.get(slot)
1612 .and_then(|c| c.as_ref())
1613 .ok_or_else(|| format!("unbound variable `{var}`"))
1614}
1615
1616fn require_node(row: &Row, vars: &VarTable, var: &str) -> Result<u32, String> {
1617 match require_cell(row, vars, var)? {
1618 Cell::Node(id) => Ok(*id),
1619 Cell::Rel(_) => Err(format!("variable `{var}` is not a node")),
1620 Cell::Path(_) => Err(format!("variable `{var}` is a path, not a node")),
1621 Cell::Scalar(_) => Err(format!("variable `{var}` is a scalar value, not a node")),
1622 }
1623}
1624
1625const SCALAR_FUNCS: &[&str] = &[
1627 "toLower",
1628 "toUpper",
1629 "size",
1630 "coalesce",
1631 "type",
1632 "abs",
1633 "round",
1634 "textMatches",
1635 "contains",
1636 "startsWith",
1637 "endsWith",
1638 "toInteger",
1639 "toFloat",
1640 "toString",
1641 "decay",
1642 "key",
1643 "labels",
1644];
1645
1646fn eval_string_predicate(
1649 name: &str,
1650 args: &[Operand],
1651 view: &GraphView,
1652 vars: &VarTable,
1653 row: &Row,
1654 params: &Params,
1655 f: impl Fn(&str, &str) -> bool,
1656) -> Result<Option<Value>, String> {
1657 if args.len() != 2 {
1658 return Err(format!(
1659 "{name}() requires exactly 2 arguments, got {}",
1660 args.len()
1661 ));
1662 }
1663 let a = resolve_operand(view, vars, row, &args[0], params)?;
1664 let b = resolve_operand(view, vars, row, &args[1], params)?;
1665 match (a, b) {
1666 (Some(Value::Str(s)), Some(Value::Str(sub))) => Ok(Some(Value::Bool(f(&s, &sub)))),
1667 (None, _) | (_, None) => Ok(None),
1668 _ => Ok(None),
1669 }
1670}
1671
1672fn eval_func(
1676 name: &str,
1677 args: &[Operand],
1678 view: &GraphView,
1679 vars: &VarTable,
1680 row: &Row,
1681 params: &Params,
1682) -> Result<Option<Value>, String> {
1683 let norm = name.to_ascii_lowercase();
1684 match norm.as_str() {
1685 "tolower" => {
1686 if args.len() != 1 {
1687 return Err(format!(
1688 "toLower() requires exactly 1 argument, got {}",
1689 args.len()
1690 ));
1691 }
1692 let v = resolve_operand(view, vars, row, &args[0], params)?;
1693 Ok(v.map(|val| match val {
1694 Value::Str(s) => Value::Str(s.to_ascii_lowercase()),
1695 other => other, }))
1697 }
1698 "toupper" => {
1699 if args.len() != 1 {
1700 return Err(format!(
1701 "toUpper() requires exactly 1 argument, got {}",
1702 args.len()
1703 ));
1704 }
1705 let v = resolve_operand(view, vars, row, &args[0], params)?;
1706 Ok(v.map(|val| match val {
1707 Value::Str(s) => Value::Str(s.to_ascii_uppercase()),
1708 other => other,
1709 }))
1710 }
1711 "size" => {
1712 if args.len() != 1 {
1713 return Err(format!(
1714 "size() requires exactly 1 argument, got {}",
1715 args.len()
1716 ));
1717 }
1718 let v = resolve_operand(view, vars, row, &args[0], params)?;
1719 match v {
1720 None => Ok(None), Some(Value::Str(s)) => Ok(Some(Value::Int(s.len() as i64))),
1722 Some(Value::List(items)) => Ok(Some(Value::Int(items.len() as i64))),
1723 Some(_) => Ok(None), }
1725 }
1726 "coalesce" => {
1727 for arg in args {
1729 if let Some(v) = resolve_operand(view, vars, row, arg, params)? {
1730 return Ok(Some(v));
1731 }
1732 }
1733 Ok(None)
1734 }
1735 "type" => {
1736 if args.len() != 1 {
1737 return Err(format!(
1738 "type() requires exactly 1 argument, got {}",
1739 args.len()
1740 ));
1741 }
1742 let Operand::Var(var_name) = &args[0] else {
1744 return Err(
1745 "type() argument must be a relationship variable (e.g. type(r))".to_string(),
1746 );
1747 };
1748 let slot = vars
1749 .slot(var_name)
1750 .ok_or_else(|| format!("unbound variable `{var_name}` in type()"))?;
1751 match row.get(slot).and_then(|c| c.as_ref()) {
1752 Some(Cell::Rel(e)) => {
1753 let etype = view.syms.resolve(e.etype).unwrap_or("").to_owned();
1755 Ok(Some(Value::Str(etype)))
1756 }
1757 Some(Cell::Node(_)) => Err(format!(
1758 "type() argument `{var_name}` is a node, not a relationship"
1759 )),
1760 Some(Cell::Scalar(_) | Cell::Path(_)) => Err(format!(
1761 "type() argument `{var_name}` is not a relationship"
1762 )),
1763 None => Ok(None), }
1765 }
1766 "key" => {
1767 if args.len() != 1 {
1768 return Err(format!(
1769 "key() requires exactly 1 argument, got {}",
1770 args.len()
1771 ));
1772 }
1773 let Operand::Var(var_name) = &args[0] else {
1776 return Err("key() argument must be a node variable (e.g. key(n))".to_string());
1777 };
1778 let slot = vars
1779 .slot(var_name)
1780 .ok_or_else(|| format!("unbound variable `{var_name}` in key()"))?;
1781 match row.get(slot).and_then(|c| c.as_ref()) {
1782 Some(Cell::Node(id)) => Ok(Some(Value::Str(view.key_of(*id).to_owned()))),
1783 Some(Cell::Rel(_)) => Err(format!(
1784 "key() argument `{var_name}` is a relationship, not a node"
1785 )),
1786 Some(Cell::Scalar(_) | Cell::Path(_)) => {
1787 Err(format!("key() argument `{var_name}` is not a node"))
1788 }
1789 None => Ok(None), }
1791 }
1792 "labels" => {
1793 if args.len() != 1 {
1794 return Err(format!(
1795 "labels() requires exactly 1 argument, got {}",
1796 args.len()
1797 ));
1798 }
1799 let Operand::Var(var_name) = &args[0] else {
1803 return Err(
1804 "labels() argument must be a node variable (e.g. labels(n))".to_string()
1805 );
1806 };
1807 let slot = vars
1808 .slot(var_name)
1809 .ok_or_else(|| format!("unbound variable `{var_name}` in labels()"))?;
1810 match row.get(slot).and_then(|c| c.as_ref()) {
1811 Some(Cell::Node(id)) => Ok(Some(Value::List(
1812 view.label_of(*id)
1813 .map(|l| vec![Value::Str(l.to_owned())])
1814 .unwrap_or_default(),
1815 ))),
1816 Some(Cell::Rel(_)) => Err(format!(
1817 "labels() argument `{var_name}` is a relationship, not a node"
1818 )),
1819 Some(Cell::Scalar(_) | Cell::Path(_)) => {
1820 Err(format!("labels() argument `{var_name}` is not a node"))
1821 }
1822 None => Ok(None), }
1824 }
1825 "abs" => {
1826 if args.len() != 1 {
1827 return Err(format!(
1828 "abs() requires exactly 1 argument, got {}",
1829 args.len()
1830 ));
1831 }
1832 let v = resolve_operand(view, vars, row, &args[0], params)?;
1833 match v {
1834 None => Ok(None),
1835 Some(Value::Int(n)) => Ok(Some(Value::Int(n.abs()))),
1836 Some(Value::Float(f)) => Ok(Some(Value::Float(f.abs()))),
1837 Some(_) => Ok(None), }
1839 }
1840 "round" => {
1841 if args.len() != 1 {
1842 return Err(format!(
1843 "round() requires exactly 1 argument, got {}",
1844 args.len()
1845 ));
1846 }
1847 let v = resolve_operand(view, vars, row, &args[0], params)?;
1848 match v {
1849 None => Ok(None),
1850 Some(Value::Int(n)) => Ok(Some(Value::Int(n))), Some(Value::Float(f)) => Ok(Some(Value::Float(f.round()))),
1852 Some(_) => Ok(None), }
1854 }
1855 "textmatches" => {
1856 if args.len() != 2 {
1868 return Err(format!(
1869 "textMatches() requires exactly 2 arguments (field_value, query), got {}",
1870 args.len()
1871 ));
1872 }
1873 let field_val = resolve_operand(view, vars, row, &args[0], params)?;
1874 let query_val = resolve_operand(view, vars, row, &args[1], params)?;
1875 match (field_val, query_val) {
1876 (None, _) | (_, None) => Ok(None),
1877 (Some(Value::Str(s)), Some(Value::Str(q))) => {
1878 Ok(Some(Value::Bool(core_storage::fulltext::eval_query_str(
1880 &s, &q,
1881 ))))
1882 }
1883 (Some(Value::List(items)), Some(Value::Str(q))) => Ok(Some(Value::Bool(
1884 core_storage::fulltext::eval_query_str_list(&items, &q),
1885 ))),
1886 _ => Ok(Some(Value::Bool(false))), }
1888 }
1889 "contains" => eval_string_predicate("contains", args, view, vars, row, params, |s, sub| {
1890 s.contains(sub)
1891 }),
1892 "startswith" => {
1893 eval_string_predicate("startsWith", args, view, vars, row, params, |s, p| {
1894 s.starts_with(p)
1895 })
1896 }
1897 "endswith" => eval_string_predicate("endsWith", args, view, vars, row, params, |s, p| {
1898 s.ends_with(p)
1899 }),
1900 "tointeger" => {
1901 if args.len() != 1 {
1902 return Err(format!(
1903 "toInteger() requires exactly 1 argument, got {}",
1904 args.len()
1905 ));
1906 }
1907 let v = resolve_operand(view, vars, row, &args[0], params)?;
1908 Ok(match v {
1909 None => None,
1910 Some(Value::Int(n)) => Some(Value::Int(n)),
1911 Some(Value::Float(f)) => Some(Value::Int(f.trunc() as i64)),
1912 Some(Value::Str(s)) => s
1915 .trim()
1916 .parse::<i64>()
1917 .ok()
1918 .or_else(|| s.trim().parse::<f64>().ok().map(|f| f.trunc() as i64))
1919 .map(Value::Int),
1920 Some(_) => None,
1921 })
1922 }
1923 "tofloat" => {
1924 if args.len() != 1 {
1925 return Err(format!(
1926 "toFloat() requires exactly 1 argument, got {}",
1927 args.len()
1928 ));
1929 }
1930 let v = resolve_operand(view, vars, row, &args[0], params)?;
1931 Ok(match v {
1932 None => None,
1933 Some(Value::Float(f)) => Some(Value::Float(f)),
1934 Some(Value::Int(n)) => Some(Value::Float(n as f64)),
1935 Some(Value::Str(s)) => s.trim().parse::<f64>().ok().map(Value::Float),
1936 Some(_) => None,
1937 })
1938 }
1939 "tostring" => {
1940 if args.len() != 1 {
1941 return Err(format!(
1942 "toString() requires exactly 1 argument, got {}",
1943 args.len()
1944 ));
1945 }
1946 let v = resolve_operand(view, vars, row, &args[0], params)?;
1947 Ok(match v {
1948 None => None,
1949 Some(Value::Str(s)) => Some(Value::Str(s)),
1950 Some(Value::Int(n)) => Some(Value::Str(n.to_string())),
1951 Some(Value::Float(f)) => Some(Value::Str(f.to_string())),
1952 Some(Value::Bool(b)) => Some(Value::Str(b.to_string())),
1953 Some(_) => None, })
1955 }
1956 "decay" => {
1957 if args.len() != 3 {
1962 return Err(format!(
1963 "decay() requires exactly 3 arguments, got {}",
1964 args.len()
1965 ));
1966 }
1967 let base = resolve_operand(view, vars, row, &args[0], params)?;
1968 let age = resolve_operand(view, vars, row, &args[1], params)?;
1969 let halflife = resolve_operand(view, vars, row, &args[2], params)?;
1970 match (base, age, halflife) {
1971 (None, _, _) | (_, None, _) | (_, _, None) => Ok(None),
1972 (Some(b), Some(a), Some(h)) => {
1973 let b = numeric_val(&b)
1974 .ok_or_else(|| "decay() requires numeric arguments".to_string())?;
1975 let a = numeric_val(&a)
1976 .ok_or_else(|| "decay() requires numeric arguments".to_string())?;
1977 let h = numeric_val(&h)
1978 .ok_or_else(|| "decay() requires numeric arguments".to_string())?;
1979 if h <= 0.0 {
1980 return Err("decay() requires halflife > 0".to_string());
1981 }
1982 Ok(Some(Value::Float(b * 0.5f64.powf(a / h))))
1983 }
1984 }
1985 }
1986 _ => Err(format!(
1987 "unknown function `{name}`; supported: {}",
1988 SCALAR_FUNCS.join(", ")
1989 )),
1990 }
1991}
1992
1993fn resolve_operand(
1994 view: &GraphView,
1995 vars: &VarTable,
1996 row: &Row,
1997 operand: &Operand,
1998 params: &Params,
1999) -> Result<Option<Value>, String> {
2000 match operand {
2001 Operand::Lit(v) => Ok(Some(v.clone())),
2002 Operand::Param(name) => match params.0.get(name) {
2003 Some(v) => Ok(Some(v.clone())),
2004 None => Err(format!("missing parameter `{name}`")),
2005 },
2006 Operand::Prop { var, field } => resolve_prop(view, vars, row, var, field),
2007 Operand::Var(name) => {
2008 match vars
2010 .slot(name)
2011 .and_then(|s| row.get(s))
2012 .and_then(|c| c.as_ref())
2013 {
2014 Some(Cell::Scalar(v)) => Ok(Some(v.clone())),
2015 Some(Cell::Node(id)) => match view.ids.key_of(*id) {
2016 Some(key) => Ok(Some(Value::Str(key.to_owned()))),
2017 None => Ok(None),
2018 },
2019 Some(Cell::Path(hops)) => Ok(Some(Value::Int(*hops as i64))),
2020 Some(Cell::Rel(_)) => Err(format!("variable `{name}` is a relationship")),
2021 None => Ok(None),
2022 }
2023 }
2024 Operand::FuncCall { name, args } => eval_func(name, args, view, vars, row, params),
2025 Operand::Index { base, index } => {
2026 let base_val = resolve_operand(view, vars, row, base, params)?;
2027 let idx_val = resolve_operand(view, vars, row, index, params)?;
2028 Ok(index_list(base_val, idx_val))
2029 }
2030 Operand::Case { branches, default } => {
2031 for (cond, value) in branches {
2032 if eval_expr(view, vars, row, cond, params, 0)? {
2033 return resolve_operand(view, vars, row, value, params);
2034 }
2035 }
2036 match default {
2037 Some(d) => resolve_operand(view, vars, row, d, params),
2038 None => Ok(None),
2039 }
2040 }
2041 Operand::BinArith { op, left, right } => {
2042 use super::ast::ArithOp;
2043 let lv = resolve_operand(view, vars, row, left, params)?;
2044 let rv = resolve_operand(view, vars, row, right, params)?;
2045 match (lv, rv) {
2046 (None, _) | (_, None) => Ok(None), (Some(Value::Int(a)), Some(Value::Int(b))) => {
2048 let result = match op {
2049 ArithOp::Sub => a.saturating_sub(b),
2050 ArithOp::Mul => a.saturating_mul(b),
2051 ArithOp::Add => a.saturating_add(b),
2052 ArithOp::Div => {
2053 if b == 0 {
2054 return Err("division by zero".into());
2055 }
2056 a.checked_div(b).unwrap_or(i64::MAX)
2057 }
2058 };
2059 Ok(Some(Value::Int(result)))
2060 }
2061 (Some(lv), Some(rv)) => {
2062 let a = match &lv {
2063 Value::Float(f) => *f,
2064 Value::Int(i) => *i as f64,
2065 _ => return Err(format!("arithmetic operand must be numeric, got {lv:?}")),
2066 };
2067 let b = match &rv {
2068 Value::Float(f) => *f,
2069 Value::Int(i) => *i as f64,
2070 _ => return Err(format!("arithmetic operand must be numeric, got {rv:?}")),
2071 };
2072 let result = match op {
2073 ArithOp::Sub => a - b,
2074 ArithOp::Mul => a * b,
2075 ArithOp::Add => a + b,
2076 ArithOp::Div => {
2077 if b == 0.0 {
2078 return Err("division by zero".into());
2079 }
2080 a / b
2081 }
2082 };
2083 Ok(Some(Value::Float(result)))
2084 }
2085 }
2086 }
2087 }
2088}
2089
2090fn index_list(base: Option<Value>, index: Option<Value>) -> Option<Value> {
2096 let (Some(Value::List(items)), Some(idx)) = (base, index) else {
2097 return None;
2098 };
2099 let i = match idx {
2100 Value::Int(n) => n,
2101 Value::Float(f) if f.fract() == 0.0 && f.is_finite() => f as i64,
2102 _ => return None,
2103 };
2104 let len = i64::try_from(items.len()).ok()?;
2105 let pos = if i < 0 { len.checked_add(i)? } else { i };
2106 if pos < 0 || pos >= len {
2107 return None;
2108 }
2109 items.into_iter().nth(pos as usize)
2110}
2111
2112fn is_identity_field(field: &str) -> bool {
2115 field == "key" || field == "label"
2116}
2117
2118fn node_identity_prop(view: &GraphView, id: u32, field: &str) -> Option<Value> {
2125 match field {
2126 "key" => view.ids.key_of(id).map(|k| Value::Str(k.to_owned())),
2127 "label" => view.label_of(id).map(|l| Value::Str(l.to_owned())),
2128 _ => None,
2129 }
2130}
2131
2132fn resolve_prop(
2133 view: &GraphView,
2134 vars: &VarTable,
2135 row: &Row,
2136 var: &str,
2137 field: &str,
2138) -> Result<Option<Value>, String> {
2139 let slot = vars
2142 .slot(var)
2143 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2144 let cell = match row.get(slot).and_then(|c| c.as_ref()) {
2145 Some(c) => c,
2146 None => return Ok(None),
2148 };
2149 match cell {
2150 Cell::Node(id) => Ok(view
2151 .prop(*id, field)
2152 .map(|vr| vr.into_value())
2153 .or_else(|| node_identity_prop(view, *id, field))),
2154 Cell::Rel(e) => Ok(view.edge_props.get(e.etype, e.src, e.dst, field)),
2155 Cell::Path(hops) => {
2157 if field == "length" {
2158 Ok(Some(Value::Int(*hops as i64)))
2159 } else {
2160 Ok(None)
2161 }
2162 }
2163 Cell::Scalar(_) => Ok(None),
2165 }
2166}
2167
2168fn node_matches(
2169 view: &GraphView,
2170 vars: &VarTable,
2171 row: &Row,
2172 id: u32,
2173 label: Option<&str>,
2174 props: &[(String, Operand)],
2175 params: &Params,
2176) -> Result<bool, String> {
2177 if let Some(want) = label {
2178 match view.label_of(id) {
2179 Some(got) if got == want => {}
2180 _ => return Ok(false),
2181 }
2182 }
2183 for (field, operand) in props {
2184 let Some(expected) = resolve_operand(view, vars, row, operand, params)? else {
2185 return Ok(false);
2186 };
2187 let got = view
2188 .prop(id, field)
2189 .map(|vr| vr.into_value())
2190 .or_else(|| node_identity_prop(view, id, field));
2191 match got {
2192 Some(got) if values_equal(&got, &expected) => {}
2193 _ => return Ok(false),
2194 }
2195 }
2196 Ok(true)
2197}
2198
2199fn retain_node(
2200 view: &GraphView,
2201 vars: &VarTable,
2202 rows: &[Row],
2203 var: &str,
2204 label: Option<&str>,
2205 props: &[(String, Operand)],
2206 params: &Params,
2207) -> Result<Vec<Row>, String> {
2208 let mut out = Vec::with_capacity(rows.len());
2209 for row in rows {
2210 let id = require_node(row, vars, var)?;
2211 if node_matches(view, vars, row, id, label, props, params)? {
2212 out.push(row.clone());
2213 }
2214 }
2215 Ok(out)
2216}
2217
2218fn map_dir(dir: RelDir) -> Dir {
2219 match dir {
2220 RelDir::Right => Dir::Out,
2221 RelDir::Left => Dir::In,
2222 RelDir::Undirected => Dir::Both,
2223 }
2224}
2225
2226fn neighbor(from: u32, e: &EdgeRef, dir: RelDir) -> u32 {
2227 match dir {
2228 RelDir::Right => e.dst,
2229 RelDir::Left => e.src,
2230 RelDir::Undirected => {
2231 if e.src == from {
2232 e.dst
2233 } else {
2234 e.src
2235 }
2236 }
2237 }
2238}
2239
2240fn row_has_edge(row: &Row, e: &EdgeRef) -> bool {
2241 row.iter()
2242 .any(|c| matches!(c, Some(Cell::Rel(existing)) if existing == e))
2243}
2244
2245fn resolve_etypes(view: &GraphView, etypes: &[String]) -> Option<Vec<u32>> {
2246 if etypes.is_empty() {
2247 None } else {
2249 Some(etypes.iter().filter_map(|n| view.syms.get(n)).collect())
2252 }
2253}
2254
2255fn exec_expand(
2256 view: &GraphView,
2257 vars: &VarTable,
2258 rows: &[Row],
2259 op: &PlanOp,
2260 params: &Params,
2261) -> Result<Vec<Row>, String> {
2262 let PlanOp::Expand {
2263 from,
2264 rel_var,
2265 etypes,
2266 dir,
2267 to,
2268 to_label,
2269 to_props,
2270 } = op
2271 else {
2272 return Err("internal: expected Expand".into());
2273 };
2274 let etypes = resolve_etypes(view, etypes);
2275 let exp_dir = map_dir(*dir);
2276 let to_slot = vars
2277 .slot(to)
2278 .ok_or_else(|| format!("unbound variable `{to}`"))?;
2279 let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
2280 let cap = max_intermediate_rows();
2281 let mut out = Vec::with_capacity(rows.len().saturating_mul(2).min(cap));
2282 for row in rows {
2283 let from_id = require_node(row, vars, from)?;
2284 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
2285 Some(Cell::Node(id)) => Some(*id),
2286 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
2287 return Err(format!("variable `{to}` is not a node"))
2288 }
2289 None => None,
2290 };
2291 for e in expand(view, from_id, etypes.as_deref(), exp_dir) {
2292 if row_has_edge(row, &e) {
2293 continue;
2294 }
2295 let nbr = neighbor(from_id, &e, *dir);
2296 if !view.visible(nbr) {
2297 continue;
2298 }
2299 if let Some(want) = bound_to {
2300 if nbr != want {
2301 continue;
2302 }
2303 }
2304 if !node_matches(view, vars, row, nbr, to_label.as_deref(), to_props, params)? {
2305 continue;
2306 }
2307 if out.len() >= cap {
2308 return Err(row_cap_err(cap));
2309 }
2310 let mut next = row.clone();
2311 if let Some(slot) = rel_slot {
2312 next[slot] = Some(Cell::Rel(e));
2313 }
2314 if bound_to.is_none() {
2315 next[to_slot] = Some(Cell::Node(nbr));
2316 }
2317 out.push(next);
2318 #[cfg(test)]
2319 record_expand_row();
2320 }
2321 }
2322 Ok(out)
2323}
2324
2325#[allow(clippy::too_many_arguments)]
2332fn exec_var_expand(
2333 view: &GraphView,
2334 vars: &VarTable,
2335 rows: &[Row],
2336 from: &str,
2337 rel_var: &Option<String>,
2338 etypes: &[String],
2339 dir: RelDir,
2340 to: &str,
2341 min: u8,
2342 max: u8,
2343) -> Result<Vec<Row>, String> {
2344 let etypes = resolve_etypes(view, etypes);
2345 let exp_dir = map_dir(dir);
2346 let to_slot = vars
2347 .slot(to)
2348 .ok_or_else(|| format!("unbound variable `{to}`"))?;
2349 let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
2350 let cap = max_intermediate_rows();
2351 let mut out: Vec<Row> = Vec::new();
2352
2353 for row in rows {
2354 let from_id = require_node(row, vars, from)?;
2355 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
2356 Some(Cell::Node(id)) => Some(*id),
2357 Some(_) => return Err(format!("variable `{to}` is not a node")),
2358 None => None,
2359 };
2360
2361 struct PathState {
2364 node: u32,
2365 edges: Vec<EdgeRef>,
2366 }
2367
2368 let mut frontier: Vec<PathState> = vec![PathState {
2369 node: from_id,
2370 edges: Vec::new(),
2371 }];
2372
2373 let mut frontier_count: usize = 0;
2378
2379 for depth in 1u8..=max {
2380 let mut next_frontier: Vec<PathState> = Vec::new();
2381 for state in &frontier {
2382 for e in expand(view, state.node, etypes.as_deref(), exp_dir) {
2383 if state.edges.contains(&e) {
2385 continue;
2386 }
2387 let nbr = neighbor(state.node, &e, dir);
2388 if !view.visible(nbr) {
2391 continue;
2392 }
2393
2394 if depth >= min {
2397 let dest_matches = match bound_to {
2398 Some(want) => nbr == want,
2399 None => true,
2400 };
2401 if dest_matches {
2402 if out.len() >= cap {
2403 return Err(row_cap_err(cap));
2404 }
2405 let mut next = row.clone();
2406 if let Some(slot) = rel_slot {
2407 next[slot] = Some(Cell::Path(depth));
2408 }
2409 next[to_slot] = Some(Cell::Node(nbr));
2410 out.push(next);
2411 }
2412 }
2413
2414 if depth < max {
2418 frontier_count += 1;
2419 if frontier_count >= cap {
2420 return Err(row_cap_err(cap));
2421 }
2422 let mut new_edges = state.edges.clone();
2423 new_edges.push(e);
2424 next_frontier.push(PathState {
2425 node: nbr,
2426 edges: new_edges,
2427 });
2428 }
2429 }
2430 }
2431 frontier = next_frontier;
2432 if frontier.is_empty() {
2433 break;
2434 }
2435 }
2436 }
2437 Ok(out)
2438}
2439
2440#[allow(clippy::too_many_arguments)]
2446fn exec_shortest_path(
2447 view: &GraphView,
2448 vars: &VarTable,
2449 rows: &[Row],
2450 from: &str,
2451 rel_var: &Option<String>,
2452 etypes: &[String],
2453 dir: RelDir,
2454 to: &str,
2455 max_hops: u8,
2456) -> Result<Vec<Row>, String> {
2457 let etypes = resolve_etypes(view, etypes);
2458 let exp_dir = map_dir(dir);
2459 let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
2460 let mut out: Vec<Row> = Vec::new();
2461
2462 for row in rows {
2463 let from_id = require_node(row, vars, from)?;
2464 let to_id = require_node(row, vars, to)?;
2465
2466 let mut visited = std::collections::BTreeSet::new();
2468 visited.insert(from_id);
2469 let mut frontier: Vec<u32> = vec![from_id];
2470
2471 'bfs: for depth in 1u8..=max_hops {
2472 let mut next_frontier: Vec<u32> = Vec::new();
2473 for &node in &frontier {
2474 for e in expand(view, node, etypes.as_deref(), exp_dir) {
2475 let nbr = neighbor(node, &e, dir);
2476 if !view.visible(nbr) {
2480 continue;
2481 }
2482 if nbr == to_id {
2483 let mut next = row.clone();
2485 if let Some(slot) = rel_slot {
2486 next[slot] = Some(Cell::Path(depth));
2487 }
2488 out.push(next);
2489 break 'bfs;
2490 }
2491 if !visited.contains(&nbr) {
2492 visited.insert(nbr);
2493 next_frontier.push(nbr);
2494 }
2495 }
2496 }
2497 frontier = next_frontier;
2498 if frontier.is_empty() {
2499 break;
2500 }
2501 }
2502 }
2503 Ok(out)
2504}
2505
2506fn exec_filter(
2507 view: &GraphView,
2508 vars: &VarTable,
2509 rows: &[Row],
2510 expr: &Expr,
2511 params: &Params,
2512) -> Result<Vec<Row>, String> {
2513 let mut out = Vec::with_capacity(rows.len());
2514 for row in rows {
2515 if eval_expr(view, vars, row, expr, params, 0)? {
2516 out.push(row.clone());
2517 }
2518 }
2519 Ok(out)
2520}
2521
2522struct PullCtx<'a> {
2538 view: &'a GraphView<'a>,
2539 vars: &'a VarTable,
2540 project_items: &'a [RetItem],
2541 params: &'a Params<'a>,
2542 bound: usize,
2543}
2544
2545fn execute_pull(
2546 view: &GraphView,
2547 plan: &[PlanOp],
2548 params: &Params,
2549 bound: usize,
2550) -> Result<ResultSet, String> {
2551 let proj_pos = match plan
2552 .iter()
2553 .position(|op| matches!(op, PlanOp::Project { .. }))
2554 {
2555 Some(p) => p,
2556 None => return Ok(ResultSet::new(vec![])),
2557 };
2558 let producers = &plan[..proj_pos];
2559 let project_items = match &plan[proj_pos] {
2560 PlanOp::Project { items } => items,
2561 _ => unreachable!(),
2562 };
2563 let columns: Vec<String> = project_items.iter().map(column_name).collect();
2564 let vars = collect_vars(plan);
2565 let ctx = PullCtx {
2566 view,
2567 vars: &vars,
2568 project_items,
2569 params,
2570 bound,
2571 };
2572 let mut initial_row: Row = vec![None; vars.names.len()];
2573 let mut result_rows: Vec<Vec<Option<Value>>> = Vec::with_capacity(bound);
2574 pull_rows(&ctx, producers, &mut initial_row, &mut result_rows)?;
2575 let skip_n = plan[proj_pos + 1..]
2578 .iter()
2579 .find_map(|op| match op {
2580 PlanOp::Skip(ls) => Some(resolve_ls(ls, params)),
2581 _ => None,
2582 })
2583 .transpose()?
2584 .unwrap_or(0);
2585 let skip_n = usize::try_from(skip_n).unwrap_or(usize::MAX);
2586 let mut rs = ResultSet::new(columns);
2587 for row in result_rows.into_iter().skip(skip_n) {
2588 rs.push_row(row);
2589 }
2590 Ok(rs)
2591}
2592
2593enum AggAcc {
2607 Count(u64),
2608 Sum {
2609 val: f64,
2610 has_value: bool,
2611 },
2612 Avg {
2613 sum: f64,
2614 n: u64,
2615 },
2616 Min(Option<Value>),
2617 Max(Option<Value>),
2618 Collect(Vec<Value>),
2619 Distinct {
2622 seen: HashSet<Option<ValueKey>>,
2623 inner: Box<AggAcc>,
2624 },
2625}
2626
2627impl AggAcc {
2628 fn for_arg(func: &AggFunc, arg: &AggArg) -> Self {
2631 match arg {
2632 AggArg::Distinct(_) => AggAcc::Distinct {
2633 seen: HashSet::new(),
2634 inner: Box::new(AggAcc::new(func)),
2635 },
2636 _ => AggAcc::new(func),
2637 }
2638 }
2639
2640 fn new(func: &AggFunc) -> Self {
2641 match func {
2642 AggFunc::Count => AggAcc::Count(0),
2643 AggFunc::Sum => AggAcc::Sum {
2644 val: 0.0,
2645 has_value: false,
2646 },
2647 AggFunc::Avg => AggAcc::Avg { sum: 0.0, n: 0 },
2648 AggFunc::Min => AggAcc::Min(None),
2649 AggFunc::Max => AggAcc::Max(None),
2650 AggFunc::Collect => AggAcc::Collect(Vec::new()),
2651 }
2652 }
2653
2654 fn finish(self) -> Option<Value> {
2655 match self {
2656 AggAcc::Count(n) => Some(Value::Int(i64::try_from(n).unwrap_or(i64::MAX))),
2661 AggAcc::Sum { val, has_value } => {
2662 if has_value {
2663 Some(Value::Float(val))
2664 } else {
2665 None
2666 }
2667 }
2668 AggAcc::Avg { sum, n } => {
2669 if n > 0 {
2670 Some(Value::Float(sum / n as f64))
2671 } else {
2672 None
2673 }
2674 }
2675 AggAcc::Min(v) => v,
2676 AggAcc::Max(v) => v,
2677 AggAcc::Collect(items) => Some(Value::List(items)),
2679 AggAcc::Distinct { inner, .. } => inner.finish(),
2680 }
2681 }
2682}
2683
2684fn numeric_val(v: &Value) -> Option<f64> {
2687 match v {
2688 Value::Int(n) => Some(*n as f64),
2689 Value::Float(f) => Some(*f),
2690 _ => None,
2691 }
2692}
2693
2694struct AggStreamCtx<'a> {
2696 view: &'a GraphView<'a>,
2697 vars: &'a VarTable,
2698 params: &'a Params<'a>,
2699 func: &'a AggFunc,
2700 arg: &'a AggArg,
2701}
2702
2703fn execute_aggregate(
2708 view: &GraphView,
2709 plan: &[PlanOp],
2710 params: &Params,
2711) -> Result<ResultSet, String> {
2712 let agg_pos = match plan
2713 .iter()
2714 .position(|op| matches!(op, PlanOp::Aggregate { .. }))
2715 {
2716 Some(p) => p,
2717 None => return Ok(ResultSet::new(vec![])),
2718 };
2719 let producers = &plan[..agg_pos];
2720 let (func, arg, column) = match &plan[agg_pos] {
2721 PlanOp::Aggregate { func, arg, column } => (func, arg, column),
2722 _ => unreachable!(),
2723 };
2724 let vars = collect_vars(plan);
2725 let ctx = AggStreamCtx {
2726 view,
2727 vars: &vars,
2728 params,
2729 func,
2730 arg,
2731 };
2732 let initial_row: Row = vec![None; vars.names.len()];
2733 let mut acc = AggAcc::for_arg(func, arg);
2734 agg_stream(&ctx, producers, &initial_row, &mut acc)?;
2735
2736 let value = acc.finish();
2737 let mut rs = ResultSet::new(vec![column.clone()]);
2738 rs.push_row(vec![value]);
2739 Ok(rs)
2740}
2741
2742fn distinct_value(
2766 view: &GraphView,
2767 vars: &VarTable,
2768 row: &Row,
2769 arg: &AggArg,
2770) -> Result<Option<Value>, String> {
2771 match arg {
2772 AggArg::Star => Ok(None),
2773 AggArg::Var(v) => {
2774 let Some(slot) = vars.slot(v) else {
2775 return Ok(None);
2776 };
2777 Ok(match row.get(slot).and_then(|c| c.as_ref()) {
2778 Some(Cell::Node(id)) => view.ids.key_of(*id).map(|k| Value::Str(k.to_owned())),
2779 Some(Cell::Scalar(val)) => Some(val.clone()),
2780 Some(Cell::Path(hops)) => Some(Value::Int(*hops as i64)),
2781 Some(Cell::Rel(e)) => Some(Value::Str(format!(
2782 "{}\u{1}{}\u{1}{}",
2783 e.etype, e.src, e.dst
2784 ))),
2785 None => None,
2786 })
2787 }
2788 AggArg::Prop { var, field } => resolve_prop(view, vars, row, var, field),
2789 AggArg::Distinct(inner) => distinct_value(view, vars, row, inner),
2791 }
2792}
2793
2794fn update_acc(
2800 view: &GraphView,
2801 vars: &VarTable,
2802 row: &Row,
2803 func: &AggFunc,
2804 arg: &AggArg,
2805 acc: &mut AggAcc,
2806) -> Result<(), String> {
2807 if let AggArg::Distinct(inner_arg) = arg {
2811 let AggAcc::Distinct { seen, inner } = acc else {
2812 return Ok(());
2813 };
2814 let val = distinct_value(view, vars, row, inner_arg)?;
2815 let Some(val) = val else {
2818 return Ok(());
2819 };
2820 if !seen.insert(group_key_normalize(&val)) {
2821 return Ok(());
2822 }
2823 return update_acc(view, vars, row, func, inner_arg, inner);
2824 }
2825 match (func, arg) {
2826 (AggFunc::Count, AggArg::Star) => {
2827 if let AggAcc::Count(n) = acc {
2828 *n += 1;
2829 }
2830 }
2831 (AggFunc::Count, AggArg::Var(v)) => {
2832 let slot = vars.slot(v);
2833 let is_bound = slot
2834 .and_then(|s| row.get(s))
2835 .and_then(|c| c.as_ref())
2836 .is_some();
2837 if is_bound {
2838 if let AggAcc::Count(n) = acc {
2839 *n += 1;
2840 }
2841 }
2842 }
2843 (AggFunc::Count, AggArg::Prop { var, field }) => {
2844 let val = resolve_prop(view, vars, row, var, field)?;
2845 if val.is_some() {
2846 if let AggAcc::Count(n) = acc {
2847 *n += 1;
2848 }
2849 }
2850 }
2851 (AggFunc::Sum, AggArg::Prop { var, field }) => {
2852 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2853 if let Some(num) = numeric_val(&v) {
2854 if let AggAcc::Sum { val, has_value } = acc {
2855 *val += num;
2856 *has_value = true;
2857 }
2858 }
2859 }
2860 }
2861 (AggFunc::Avg, AggArg::Prop { var, field }) => {
2862 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2863 if let Some(num) = numeric_val(&v) {
2864 if let AggAcc::Avg { sum, n } = acc {
2865 *sum += num;
2866 *n += 1;
2867 }
2868 }
2869 }
2870 }
2871 (AggFunc::Min, AggArg::Prop { var, field }) => {
2872 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2873 if numeric_val(&v).is_some() {
2874 if let AggAcc::Min(current) = acc {
2875 *current = Some(match current.take() {
2876 None => v,
2877 Some(prev) => {
2878 if cmp_optional(Some(&prev), Some(&v), false)
2879 == std::cmp::Ordering::Greater
2880 {
2881 v
2882 } else {
2883 prev
2884 }
2885 }
2886 });
2887 }
2888 }
2889 }
2890 }
2891 (AggFunc::Max, AggArg::Prop { var, field }) => {
2892 if let Some(v) = resolve_prop(view, vars, row, var, field)? {
2893 if numeric_val(&v).is_some() {
2894 if let AggAcc::Max(current) = acc {
2895 *current = Some(match current.take() {
2896 None => v,
2897 Some(prev) => {
2898 if cmp_optional(Some(&prev), Some(&v), true)
2899 == std::cmp::Ordering::Greater
2900 {
2901 v
2902 } else {
2903 prev
2904 }
2905 }
2906 });
2907 }
2908 }
2909 }
2910 }
2911 (AggFunc::Collect, AggArg::Var(v)) => {
2912 let val = vars
2915 .slot(v)
2916 .and_then(|s| row.get(s))
2917 .and_then(|c| c.as_ref())
2918 .and_then(|cell| match cell {
2919 Cell::Scalar(x) => Some(x.clone()),
2920 Cell::Node(id) => view.ids.key_of(*id).map(|k| Value::Str(k.to_owned())),
2921 Cell::Path(h) => Some(Value::Int(*h as i64)),
2922 Cell::Rel(_) => None,
2923 });
2924 if let Some(val) = val {
2925 if let AggAcc::Collect(items) = acc {
2926 items.push(val);
2927 }
2928 }
2929 }
2930 (AggFunc::Collect, AggArg::Prop { var, field }) => {
2931 if let Some(val) = resolve_prop(view, vars, row, var, field)? {
2932 if let AggAcc::Collect(items) = acc {
2933 items.push(val);
2934 }
2935 }
2936 }
2937 _ => {}
2940 }
2941 Ok(())
2942}
2943
2944fn agg_stream(
2947 ctx: &AggStreamCtx<'_>,
2948 ops: &[PlanOp],
2949 row: &Row,
2950 acc: &mut AggAcc,
2951) -> Result<(), String> {
2952 let (op, rest) = match ops.split_first() {
2953 Some(pair) => pair,
2954 None => {
2955 update_acc(ctx.view, ctx.vars, row, ctx.func, ctx.arg, acc)?;
2957 return Ok(());
2958 }
2959 };
2960
2961 match op {
2962 PlanOp::ScanLabel { var, label } => {
2963 let ids = scan_ids(ctx.view, label.as_deref());
2964 let slot = ctx
2965 .vars
2966 .slot(var)
2967 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2968 for &id in &ids {
2969 let mut next = row.clone();
2970 next[slot] = Some(Cell::Node(id));
2971 agg_stream(ctx, rest, &next, acc)?;
2972 }
2973 }
2974 PlanOp::ScanKey { var, key, label } => {
2975 let slot = ctx
2976 .vars
2977 .slot(var)
2978 .ok_or_else(|| format!("unbound variable `{var}`"))?;
2979 if let Some(id) =
2980 resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
2981 {
2982 let mut next = row.clone();
2983 next[slot] = Some(Cell::Node(id));
2984 agg_stream(ctx, rest, &next, acc)?;
2985 }
2986 }
2987 PlanOp::IndexScan {
2988 var,
2989 label,
2990 field,
2991 value,
2992 } => {
2993 let ids = index_scan_ids(
2994 ctx.view,
2995 ctx.vars,
2996 row,
2997 label.as_deref(),
2998 field,
2999 value,
3000 ctx.params,
3001 )?;
3002 let slot = ctx
3003 .vars
3004 .slot(var)
3005 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3006 for &id in &ids {
3007 let mut next = row.clone();
3008 next[slot] = Some(Cell::Node(id));
3009 agg_stream(ctx, rest, &next, acc)?;
3010 }
3011 }
3012 PlanOp::IndexIntersect {
3013 var,
3014 label,
3015 equalities,
3016 } => {
3017 let ids = index_intersect_ids(
3018 ctx.view,
3019 ctx.vars,
3020 row,
3021 label.as_deref(),
3022 equalities,
3023 ctx.params,
3024 )?;
3025 let slot = ctx
3026 .vars
3027 .slot(var)
3028 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3029 for id in ids {
3030 let mut next = row.clone();
3031 next[slot] = Some(Cell::Node(id));
3032 agg_stream(ctx, rest, &next, acc)?;
3033 }
3034 }
3035 PlanOp::Expand {
3036 from,
3037 rel_var,
3038 etypes,
3039 dir,
3040 to,
3041 to_label,
3042 to_props,
3043 } => {
3044 let etypes = resolve_etypes(ctx.view, etypes);
3045 let exp_dir = map_dir(*dir);
3046 let to_slot = ctx
3047 .vars
3048 .slot(to)
3049 .ok_or_else(|| format!("unbound variable `{to}`"))?;
3050 let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
3051 let from_id = require_node(row, ctx.vars, from)?;
3052 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
3053 Some(Cell::Node(id)) => Some(*id),
3054 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
3055 return Err(format!("variable `{to}` is not a node"))
3056 }
3057 None => None,
3058 };
3059 for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
3060 if row_has_edge(row, &e) {
3061 continue;
3062 }
3063 let nbr = neighbor(from_id, &e, *dir);
3064 if !ctx.view.visible(nbr) {
3065 continue;
3066 }
3067 if let Some(want) = bound_to {
3068 if nbr != want {
3069 continue;
3070 }
3071 }
3072 if !node_matches(
3073 ctx.view,
3074 ctx.vars,
3075 row,
3076 nbr,
3077 to_label.as_deref(),
3078 to_props,
3079 ctx.params,
3080 )? {
3081 continue;
3082 }
3083 let mut next = row.clone();
3084 if let Some(slot) = rel_slot {
3085 next[slot] = Some(Cell::Rel(e));
3086 }
3087 if bound_to.is_none() {
3088 next[to_slot] = Some(Cell::Node(nbr));
3089 }
3090 agg_stream(ctx, rest, &next, acc)?;
3091 }
3092 }
3093 PlanOp::Filter { expr } => {
3094 if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
3095 agg_stream(ctx, rest, row, acc)?;
3096 }
3097 }
3098 PlanOp::LookupProps { var, props } => {
3099 let id = require_node(row, ctx.vars, var)?;
3100 if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
3101 agg_stream(ctx, rest, row, acc)?;
3102 }
3103 }
3104 PlanOp::JoinBound { var, label, props } => {
3105 let id = require_node(row, ctx.vars, var)?;
3106 if node_matches(
3107 ctx.view,
3108 ctx.vars,
3109 row,
3110 id,
3111 label.as_deref(),
3112 props,
3113 ctx.params,
3114 )? {
3115 agg_stream(ctx, rest, row, acc)?;
3116 }
3117 }
3118 PlanOp::VarExpand {
3119 from,
3120 rel_var,
3121 etypes,
3122 dir,
3123 to,
3124 min,
3125 max,
3126 } => {
3127 let new_rows = exec_var_expand(
3128 ctx.view,
3129 ctx.vars,
3130 std::slice::from_ref(row),
3131 from,
3132 rel_var,
3133 etypes,
3134 *dir,
3135 to,
3136 *min,
3137 *max,
3138 )?;
3139 for nr in &new_rows {
3140 agg_stream(ctx, rest, nr, acc)?;
3141 }
3142 }
3143 PlanOp::ShortestPath {
3144 from,
3145 rel_var,
3146 etypes,
3147 dir,
3148 to,
3149 max_hops,
3150 } => {
3151 let new_rows = exec_shortest_path(
3152 ctx.view,
3153 ctx.vars,
3154 std::slice::from_ref(row),
3155 from,
3156 rel_var,
3157 etypes,
3158 *dir,
3159 to,
3160 *max_hops,
3161 )?;
3162 for nr in &new_rows {
3163 agg_stream(ctx, rest, nr, acc)?;
3164 }
3165 }
3166 PlanOp::Project { .. } => {
3168 return Err(
3169 "agg executor: Project in producer slice — plan is structurally malformed"
3170 .to_string(),
3171 );
3172 }
3173 PlanOp::Distinct => {
3174 return Err(
3175 "agg executor: Distinct in producer slice — plan is structurally malformed"
3176 .to_string(),
3177 );
3178 }
3179 PlanOp::OrderBy { .. } => {
3180 return Err(
3181 "agg executor: OrderBy in producer slice — structurally malformed".to_string(),
3182 );
3183 }
3184 PlanOp::Skip(_) => {
3185 return Err(
3186 "agg executor: Skip in producer slice — structurally malformed".to_string(),
3187 );
3188 }
3189 PlanOp::Limit(_) => {
3190 return Err(
3191 "agg executor: Limit in producer slice — structurally malformed".to_string(),
3192 );
3193 }
3194 PlanOp::Aggregate { .. } => {
3195 return Err(
3196 "agg executor: nested Aggregate in producer slice — structurally malformed"
3197 .to_string(),
3198 );
3199 }
3200 PlanOp::GroupAggregate { .. } => {
3201 return Err(
3202 "agg executor: GroupAggregate in producer slice — structurally malformed"
3203 .to_string(),
3204 );
3205 }
3206 PlanOp::With { .. } => {
3207 return Err(
3208 "agg executor: With in producer slice — structurally malformed".to_string(),
3209 );
3210 }
3211 PlanOp::Unwind { .. } => {
3212 return Err(
3213 "agg executor: Unwind in producer slice — structurally malformed".to_string(),
3214 );
3215 }
3216 PlanOp::LeftOuterApply { .. } => {
3217 return Err(
3218 "agg executor: LeftOuterApply in producer slice — structurally malformed"
3219 .to_string(),
3220 );
3221 }
3222 }
3223 Ok(())
3224}
3225
3226struct GroupStreamCtx<'a> {
3234 view: &'a GraphView<'a>,
3235 vars: &'a VarTable,
3236 params: &'a Params<'a>,
3237 keys: &'a [(String, RetItem)],
3238 aggs: &'a [(AggFunc, AggArg, String)],
3239}
3240
3241fn build_group_projected(
3243 keys: &[(String, RetItem)],
3244 aggs: &[(AggFunc, AggArg, String)],
3245 key_order: Vec<GroupKey>,
3246 groups: &mut HashMap<GroupKey, GroupEntry>,
3247) -> Projected {
3248 let columns: Vec<String> = keys
3249 .iter()
3250 .map(|(col, _)| col.clone())
3251 .chain(aggs.iter().map(|(_, _, col)| col.clone()))
3252 .collect();
3253 let mut rows: Vec<Vec<Option<Value>>> = Vec::with_capacity(key_order.len());
3254 for gk in key_order {
3255 let (display_keys, accs) = groups.remove(&gk).unwrap_or_default();
3256 let mut row: Vec<Option<Value>> = Vec::with_capacity(columns.len());
3257 for display_val in display_keys {
3260 row.push(display_val);
3261 }
3262 for acc in accs {
3263 row.push(acc.finish());
3264 }
3265 rows.push(row);
3266 }
3267 Projected { columns, rows }
3268}
3269
3270fn key_source_cells(vars: &VarTable, row: &Row, keys: &[(String, RetItem)]) -> Vec<Option<Cell>> {
3281 keys.iter()
3282 .map(|(_, item)| match &item.value {
3283 RetVal::Var(v) => vars
3284 .slot(v)
3285 .and_then(|s| row.get(s))
3286 .and_then(|c| c.clone()),
3287 _ => None,
3288 })
3289 .collect()
3290}
3291
3292fn group_result_to_rows(
3293 keys: &[(String, RetItem)],
3294 aggs: &[(AggFunc, AggArg, String)],
3295 key_order: Vec<GroupKey>,
3296 groups: &mut HashMap<GroupKey, GroupEntry>,
3297 key_cells: &mut HashMap<GroupKey, Vec<Option<Cell>>>,
3298 vars: &VarTable,
3299) -> Vec<Row> {
3300 let row_len = vars.names.len();
3301 let mut out: Vec<Row> = Vec::with_capacity(key_order.len());
3302 for gk in key_order {
3303 let (display_keys, accs) = groups.remove(&gk).unwrap_or_default();
3304 let mut cells = key_cells.remove(&gk).unwrap_or_default();
3305 cells.resize(keys.len(), None);
3306 let mut row: Row = vec![None; row_len];
3307 for (((col, _), val), cell) in keys.iter().zip(display_keys).zip(cells) {
3308 if let Some(slot) = vars.slot(col) {
3309 row[slot] = cell.or_else(|| val.map(Cell::Scalar));
3310 }
3311 }
3312 for ((_, _, col), acc) in aggs.iter().zip(accs) {
3313 if let Some(slot) = vars.slot(col) {
3314 row[slot] = acc.finish().map(Cell::Scalar);
3315 }
3316 }
3317 out.push(row);
3318 }
3319 out
3320}
3321
3322fn exec_order_by_rows(vars: &VarTable, rows: &mut Vec<Row>, items: &[OrderItem], view: &GraphView) {
3329 let mut key_table: Vec<Vec<Option<Value>>> = Vec::with_capacity(rows.len());
3332 for row in rows.iter() {
3333 let mut row_key: Vec<Option<Value>> = Vec::with_capacity(items.len());
3334 for item in items {
3335 let val = match &item.target {
3336 OrderTarget::Alias(name) | OrderTarget::Var(name) => vars
3337 .slot(name)
3338 .and_then(|s| row.get(s))
3339 .and_then(|c| c.as_ref())
3340 .and_then(|c| match c {
3341 Cell::Scalar(v) => Some(v.clone()),
3342 Cell::Node(id) => view.ids.key_of(*id).map(|k| Value::Str(k.to_owned())),
3343 Cell::Path(hops) => Some(Value::Int(*hops as i64)),
3344 Cell::Rel(_) => None,
3345 }),
3346 OrderTarget::Prop { var, field } => vars
3347 .slot(var)
3348 .and_then(|s| row.get(s))
3349 .and_then(|c| c.as_ref())
3350 .and_then(|c| match c {
3351 Cell::Node(id) => view
3352 .prop(*id, field)
3353 .map(|vr| vr.into_value())
3354 .or_else(|| node_identity_prop(view, *id, field)),
3355 Cell::Rel(e) => view.edge_props.get(e.etype, e.src, e.dst, field),
3356 _ => None,
3357 }),
3358 };
3359 row_key.push(val);
3360 }
3361 key_table.push(row_key);
3362 }
3363 let mut indices: Vec<usize> = (0..rows.len()).collect();
3365 indices.sort_by(|&a, &b| {
3366 for (ki, item) in items.iter().enumerate() {
3367 let c = cmp_optional(
3368 key_table[a].get(ki).and_then(|x| x.as_ref()),
3369 key_table[b].get(ki).and_then(|x| x.as_ref()),
3370 item.descending,
3371 );
3372 if c != std::cmp::Ordering::Equal {
3373 return c;
3374 }
3375 }
3376 std::cmp::Ordering::Equal
3377 });
3378 let sorted: Vec<Row> = indices.into_iter().map(|i| rows[i].clone()).collect();
3379 *rows = sorted;
3380}
3381
3382fn execute_group_aggregate(
3388 view: &GraphView,
3389 plan: &[PlanOp],
3390 params: &Params,
3391) -> Result<ResultSet, String> {
3392 let gagg_pos = plan
3393 .iter()
3394 .position(|op| matches!(op, PlanOp::GroupAggregate { .. }))
3395 .ok_or_else(|| "internal: GroupAggregate op not found in plan".to_string())?;
3396 let producers = &plan[..gagg_pos];
3397 let (keys, aggs) = match &plan[gagg_pos] {
3398 PlanOp::GroupAggregate { keys, aggs } => (keys, aggs),
3399 _ => unreachable!(),
3400 };
3401 let tail = &plan[gagg_pos + 1..];
3402 let vars = collect_vars(plan);
3403 let initial_row: Row = vec![None; vars.names.len()];
3404 let mut groups: HashMap<GroupKey, GroupEntry> = HashMap::new();
3405 let mut key_order: Vec<GroupKey> = Vec::new();
3406 let ctx = GroupStreamCtx {
3407 view,
3408 vars: &vars,
3409 params,
3410 keys,
3411 aggs,
3412 };
3413 group_stream(&ctx, producers, &initial_row, &mut groups, &mut key_order)?;
3414 if keys.is_empty() && key_order.is_empty() {
3418 let empty_key: GroupKey = vec![];
3419 key_order.push(empty_key.clone());
3420 groups.insert(
3421 empty_key,
3422 (
3423 vec![],
3424 aggs.iter().map(|(f, a, _)| AggAcc::for_arg(f, a)).collect(),
3425 ),
3426 );
3427 }
3428 let mut projected = build_group_projected(keys, aggs, key_order, &mut groups);
3429 for op in tail {
3431 match op {
3432 PlanOp::OrderBy { items } => exec_order_by(&mut projected, items)?,
3433 PlanOp::Skip(ls) => {
3434 let n = resolve_ls(ls, params)?;
3435 apply_skip(&mut projected.rows, n);
3436 }
3437 PlanOp::Limit(ls) => {
3438 let n = resolve_ls(ls, params)?;
3439 apply_limit(&mut projected.rows, n);
3440 }
3441 _ => {} }
3443 }
3444 Ok(finish(projected))
3445}
3446
3447fn group_stream(
3454 ctx: &GroupStreamCtx<'_>,
3455 ops: &[PlanOp],
3456 row: &Row,
3457 groups: &mut HashMap<GroupKey, GroupEntry>,
3458 key_order: &mut Vec<GroupKey>,
3459) -> Result<(), String> {
3460 let (op, rest) = match ops.split_first() {
3461 Some(pair) => pair,
3462 None => {
3463 let mut gk: GroupKey = Vec::with_capacity(ctx.keys.len());
3466 let mut display_vals: Vec<Option<Value>> = Vec::with_capacity(ctx.keys.len());
3467 for (_, item) in ctx.keys {
3468 let val = project_item(ctx.view, ctx.vars, row, item, ctx.params)?;
3469 gk.push(val.as_ref().and_then(group_key_normalize));
3470 display_vals.push(val);
3471 }
3472 if !groups.contains_key(&gk) {
3473 if groups.len() >= max_groups() {
3474 return Err(group_cap_err());
3475 }
3476 key_order.push(gk.clone());
3477 let init: Vec<AggAcc> = ctx
3478 .aggs
3479 .iter()
3480 .map(|(f, a, _)| AggAcc::for_arg(f, a))
3481 .collect();
3482 groups.insert(gk.clone(), (display_vals, init));
3483 }
3484 let (_, accs) = groups.get_mut(&gk).unwrap();
3485 for (acc, (func, arg, _)) in accs.iter_mut().zip(ctx.aggs.iter()) {
3486 update_acc(ctx.view, ctx.vars, row, func, arg, acc)?;
3487 }
3488 return Ok(());
3489 }
3490 };
3491
3492 match op {
3493 PlanOp::ScanLabel { var, label } => {
3494 let ids = scan_ids(ctx.view, label.as_deref());
3495 let slot = ctx
3496 .vars
3497 .slot(var)
3498 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3499 for &id in &ids {
3500 let mut next = row.clone();
3501 next[slot] = Some(Cell::Node(id));
3502 group_stream(ctx, rest, &next, groups, key_order)?;
3503 }
3504 }
3505 PlanOp::ScanKey { var, key, label } => {
3506 let slot = ctx
3507 .vars
3508 .slot(var)
3509 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3510 if let Some(id) =
3511 resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
3512 {
3513 let mut next = row.clone();
3514 next[slot] = Some(Cell::Node(id));
3515 group_stream(ctx, rest, &next, groups, key_order)?;
3516 }
3517 }
3518 PlanOp::IndexScan {
3519 var,
3520 label,
3521 field,
3522 value,
3523 } => {
3524 let ids = index_scan_ids(
3525 ctx.view,
3526 ctx.vars,
3527 row,
3528 label.as_deref(),
3529 field,
3530 value,
3531 ctx.params,
3532 )?;
3533 let slot = ctx
3534 .vars
3535 .slot(var)
3536 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3537 for &id in &ids {
3538 let mut next = row.clone();
3539 next[slot] = Some(Cell::Node(id));
3540 group_stream(ctx, rest, &next, groups, key_order)?;
3541 }
3542 }
3543 PlanOp::IndexIntersect {
3544 var,
3545 label,
3546 equalities,
3547 } => {
3548 let ids = index_intersect_ids(
3549 ctx.view,
3550 ctx.vars,
3551 row,
3552 label.as_deref(),
3553 equalities,
3554 ctx.params,
3555 )?;
3556 let slot = ctx
3557 .vars
3558 .slot(var)
3559 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3560 for id in ids {
3561 let mut next = row.clone();
3562 next[slot] = Some(Cell::Node(id));
3563 group_stream(ctx, rest, &next, groups, key_order)?;
3564 }
3565 }
3566 PlanOp::Expand {
3567 from,
3568 rel_var,
3569 etypes,
3570 dir,
3571 to,
3572 to_label,
3573 to_props,
3574 } => {
3575 let etypes = resolve_etypes(ctx.view, etypes);
3576 let exp_dir = map_dir(*dir);
3577 let to_slot = ctx
3578 .vars
3579 .slot(to)
3580 .ok_or_else(|| format!("unbound variable `{to}`"))?;
3581 let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
3582 let from_id = require_node(row, ctx.vars, from)?;
3583 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
3584 Some(Cell::Node(id)) => Some(*id),
3585 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
3586 return Err(format!("variable `{to}` is not a node"))
3587 }
3588 None => None,
3589 };
3590 for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
3591 if row_has_edge(row, &e) {
3592 continue;
3593 }
3594 let nbr = neighbor(from_id, &e, *dir);
3595 if !ctx.view.visible(nbr) {
3596 continue;
3597 }
3598 if let Some(want) = bound_to {
3599 if nbr != want {
3600 continue;
3601 }
3602 }
3603 if !node_matches(
3604 ctx.view,
3605 ctx.vars,
3606 row,
3607 nbr,
3608 to_label.as_deref(),
3609 to_props,
3610 ctx.params,
3611 )? {
3612 continue;
3613 }
3614 let mut next = row.clone();
3615 if let Some(slot) = rel_slot {
3616 next[slot] = Some(Cell::Rel(e));
3617 }
3618 if bound_to.is_none() {
3619 next[to_slot] = Some(Cell::Node(nbr));
3620 }
3621 group_stream(ctx, rest, &next, groups, key_order)?;
3622 }
3623 }
3624 PlanOp::Filter { expr } => {
3625 if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
3626 group_stream(ctx, rest, row, groups, key_order)?;
3627 }
3628 }
3629 PlanOp::LookupProps { var, props } => {
3630 let id = require_node(row, ctx.vars, var)?;
3631 if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
3632 group_stream(ctx, rest, row, groups, key_order)?;
3633 }
3634 }
3635 PlanOp::JoinBound { var, label, props } => {
3636 let id = require_node(row, ctx.vars, var)?;
3637 if node_matches(
3638 ctx.view,
3639 ctx.vars,
3640 row,
3641 id,
3642 label.as_deref(),
3643 props,
3644 ctx.params,
3645 )? {
3646 group_stream(ctx, rest, row, groups, key_order)?;
3647 }
3648 }
3649 PlanOp::VarExpand {
3650 from,
3651 rel_var,
3652 etypes,
3653 dir,
3654 to,
3655 min,
3656 max,
3657 } => {
3658 let new_rows = exec_var_expand(
3659 ctx.view,
3660 ctx.vars,
3661 std::slice::from_ref(row),
3662 from,
3663 rel_var,
3664 etypes,
3665 *dir,
3666 to,
3667 *min,
3668 *max,
3669 )?;
3670 for nr in &new_rows {
3671 group_stream(ctx, rest, nr, groups, key_order)?;
3672 }
3673 }
3674 PlanOp::ShortestPath {
3675 from,
3676 rel_var,
3677 etypes,
3678 dir,
3679 to,
3680 max_hops,
3681 } => {
3682 let new_rows = exec_shortest_path(
3683 ctx.view,
3684 ctx.vars,
3685 std::slice::from_ref(row),
3686 from,
3687 rel_var,
3688 etypes,
3689 *dir,
3690 to,
3691 *max_hops,
3692 )?;
3693 for nr in &new_rows {
3694 group_stream(ctx, rest, nr, groups, key_order)?;
3695 }
3696 }
3697 PlanOp::Project { .. } => {
3699 return Err(
3700 "group executor: Project in producer slice — structurally malformed".to_string(),
3701 );
3702 }
3703 PlanOp::Distinct => {
3704 return Err(
3705 "group executor: Distinct in producer slice — structurally malformed".to_string(),
3706 );
3707 }
3708 PlanOp::OrderBy { .. } => {
3709 return Err(
3710 "group executor: OrderBy in producer slice — structurally malformed".to_string(),
3711 );
3712 }
3713 PlanOp::Skip(_) => {
3714 return Err(
3715 "group executor: Skip in producer slice — structurally malformed".to_string(),
3716 );
3717 }
3718 PlanOp::Limit(_) => {
3719 return Err(
3720 "group executor: Limit in producer slice — structurally malformed".to_string(),
3721 );
3722 }
3723 PlanOp::Aggregate { .. } => {
3724 return Err(
3725 "group executor: Aggregate in producer slice — structurally malformed".to_string(),
3726 );
3727 }
3728 PlanOp::GroupAggregate { .. } => {
3729 return Err(
3730 "group executor: nested GroupAggregate in producer slice — structurally malformed"
3731 .to_string(),
3732 );
3733 }
3734 PlanOp::With { .. } => {
3735 return Err(
3736 "group executor: With in producer slice — structurally malformed".to_string(),
3737 );
3738 }
3739 PlanOp::Unwind { .. } => {
3740 return Err(
3741 "group executor: Unwind in producer slice — structurally malformed".to_string(),
3742 );
3743 }
3744 PlanOp::LeftOuterApply { .. } => {
3745 return Err(
3746 "group executor: LeftOuterApply in producer slice — structurally malformed"
3747 .to_string(),
3748 );
3749 }
3750 }
3751 Ok(())
3752}
3753
3754fn pull_rows(
3768 ctx: &PullCtx<'_>,
3769 ops: &[PlanOp],
3770 row: &mut Row,
3771 result: &mut Vec<Vec<Option<Value>>>,
3772) -> Result<(), String> {
3773 if result.len() >= ctx.bound {
3774 return Ok(());
3775 }
3776 let (op, rest) = match ops.split_first() {
3777 Some(pair) => pair,
3778 None => {
3779 let mut cells = Vec::with_capacity(ctx.project_items.len());
3781 for item in ctx.project_items {
3782 cells.push(project_item(ctx.view, ctx.vars, row, item, ctx.params)?);
3783 }
3784 result.push(cells);
3785 return Ok(());
3786 }
3787 };
3788 match op {
3789 PlanOp::ScanLabel { var, label } => {
3790 let slot = ctx
3791 .vars
3792 .slot(var)
3793 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3794 let want_sym = label.as_deref().and_then(|l| ctx.view.syms.get(l));
3797 if label.is_some() && want_sym.is_none() {
3798 return Ok(());
3799 }
3800 let prev = row[slot].clone();
3802
3803 let fused_filter = rest.first().and_then(|next_op| {
3809 if let PlanOp::Filter {
3810 expr:
3811 Expr::Cmp {
3812 lhs:
3813 Operand::Prop {
3814 var: ref fv,
3815 field: ref f,
3816 },
3817 op: ref cmp_op_ref,
3818 rhs: Operand::Lit(ref lit),
3819 },
3820 } = *next_op
3821 {
3822 if fv == var {
3823 return Some((f.as_str(), cmp_op_ref, lit));
3824 }
3825 }
3826 None
3827 });
3828
3829 let fused_filter = fused_filter.filter(|(f, _, _)| !is_identity_field(f));
3833 if let Some((field, cmp_op_ref, lit)) = fused_filter {
3834 #[cfg(test)]
3837 FUSED_SCAN_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
3838 let col = ctx.view.props.column(field);
3839 let rest_after_filter = &rest[1..];
3840 for (i, &sym) in ctx.view.labels.iter().enumerate() {
3841 if result.len() >= ctx.bound {
3842 break;
3843 }
3844 if sym == u32::MAX {
3845 continue;
3846 }
3847 if let Some(ws) = want_sym {
3848 if sym != ws {
3849 continue;
3850 }
3851 }
3852 let id = i as u32;
3853 if !ctx.view.visible(id) {
3854 continue;
3855 }
3856 if let Some(v) = col.get(id) {
3857 if eval_cmp(cmp_op_ref, v, lit) {
3858 row[slot] = Some(Cell::Node(id));
3859 pull_rows(ctx, rest_after_filter, row, result)?;
3860 }
3861 }
3862 }
3863 } else {
3864 for (i, &sym) in ctx.view.labels.iter().enumerate() {
3867 if result.len() >= ctx.bound {
3868 break;
3869 }
3870 if sym == u32::MAX {
3872 continue;
3873 }
3874 if let Some(ws) = want_sym {
3876 if sym != ws {
3877 continue;
3878 }
3879 }
3880 let id = i as u32;
3881 if !ctx.view.visible(id) {
3882 continue;
3883 }
3884 row[slot] = Some(Cell::Node(id));
3885 pull_rows(ctx, rest, row, result)?;
3886 }
3887 }
3888 row[slot] = prev;
3894 }
3895 PlanOp::ScanKey { var, key, label } => {
3896 let slot = ctx
3897 .vars
3898 .slot(var)
3899 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3900 if let Some(id) =
3901 resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
3902 {
3903 let prev = row[slot].clone();
3904 row[slot] = Some(Cell::Node(id));
3905 pull_rows(ctx, rest, row, result)?;
3906 row[slot] = prev;
3907 }
3908 }
3909 PlanOp::IndexScan {
3910 var,
3911 label,
3912 field,
3913 value,
3914 } => {
3915 let slot = ctx
3916 .vars
3917 .slot(var)
3918 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3919 let ids = index_scan_ids(
3920 ctx.view,
3921 ctx.vars,
3922 row,
3923 label.as_deref(),
3924 field,
3925 value,
3926 ctx.params,
3927 )?;
3928 let prev = row[slot].clone();
3929 for id in ids {
3930 if result.len() >= ctx.bound {
3931 break;
3932 }
3933 row[slot] = Some(Cell::Node(id));
3934 pull_rows(ctx, rest, row, result)?;
3935 }
3936 row[slot] = prev;
3937 }
3938 PlanOp::IndexIntersect {
3939 var,
3940 label,
3941 equalities,
3942 } => {
3943 let slot = ctx
3944 .vars
3945 .slot(var)
3946 .ok_or_else(|| format!("unbound variable `{var}`"))?;
3947 let ids = index_intersect_ids(
3948 ctx.view,
3949 ctx.vars,
3950 row,
3951 label.as_deref(),
3952 equalities,
3953 ctx.params,
3954 )?;
3955 let prev = row[slot].clone();
3956 for id in ids {
3957 if result.len() >= ctx.bound {
3958 break;
3959 }
3960 row[slot] = Some(Cell::Node(id));
3961 pull_rows(ctx, rest, row, result)?;
3962 }
3963 row[slot] = prev;
3964 }
3965 PlanOp::Expand {
3966 from,
3967 rel_var,
3968 etypes,
3969 dir,
3970 to,
3971 to_label,
3972 to_props,
3973 } => {
3974 let etypes = resolve_etypes(ctx.view, etypes);
3975 let exp_dir = map_dir(*dir);
3976 let to_slot = ctx
3977 .vars
3978 .slot(to)
3979 .ok_or_else(|| format!("unbound variable `{to}`"))?;
3980 let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
3981 let from_id = require_node(row, ctx.vars, from)?;
3982 let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
3983 Some(Cell::Node(id)) => Some(*id),
3984 Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
3985 return Err(format!("variable `{to}` is not a node"))
3986 }
3987 None => None,
3988 };
3989 for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
3990 if result.len() >= ctx.bound {
3991 break;
3992 }
3993 if row_has_edge(row, &e) {
3994 continue;
3995 }
3996 let nbr = neighbor(from_id, &e, *dir);
3997 if !ctx.view.visible(nbr) {
3998 continue;
3999 }
4000 if let Some(want) = bound_to {
4001 if nbr != want {
4002 continue;
4003 }
4004 }
4005 if !node_matches(
4006 ctx.view,
4007 ctx.vars,
4008 row,
4009 nbr,
4010 to_label.as_deref(),
4011 to_props,
4012 ctx.params,
4013 )? {
4014 continue;
4015 }
4016 let mut next = row.clone();
4017 if let Some(slot) = rel_slot {
4018 next[slot] = Some(Cell::Rel(e));
4019 }
4020 if bound_to.is_none() {
4021 next[to_slot] = Some(Cell::Node(nbr));
4022 }
4023 #[cfg(test)]
4024 record_expand_row();
4025 pull_rows(ctx, rest, &mut next, result)?;
4026 }
4027 }
4028 PlanOp::Filter { expr } => {
4029 if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
4030 pull_rows(ctx, rest, row, result)?;
4031 }
4032 }
4033 PlanOp::LookupProps { var, props } => {
4034 let id = require_node(row, ctx.vars, var)?;
4035 if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
4036 pull_rows(ctx, rest, row, result)?;
4037 }
4038 }
4039 PlanOp::JoinBound { var, label, props } => {
4040 let id = require_node(row, ctx.vars, var)?;
4041 if node_matches(
4042 ctx.view,
4043 ctx.vars,
4044 row,
4045 id,
4046 label.as_deref(),
4047 props,
4048 ctx.params,
4049 )? {
4050 pull_rows(ctx, rest, row, result)?;
4051 }
4052 }
4053 PlanOp::Project { .. } => {
4058 return Err(
4059 "pull executor: Project reached pull_rows — plan is structurally malformed"
4060 .to_string(),
4061 );
4062 }
4063 PlanOp::Distinct => {
4064 return Err(
4065 "pull executor: Distinct reached pull_rows — DISTINCT queries must use \
4066 the staged path (row_bound returns None)"
4067 .to_string(),
4068 );
4069 }
4070 PlanOp::OrderBy { .. } => {
4071 return Err(
4072 "pull executor: OrderBy reached pull_rows — queries with ORDER BY \
4073 must use the staged path (row_bound returns None)"
4074 .to_string(),
4075 );
4076 }
4077 PlanOp::Skip(_) => {
4078 return Err(
4079 "pull executor: Skip reached pull_rows — Skip must appear after Project"
4080 .to_string(),
4081 );
4082 }
4083 PlanOp::Limit(_) => {
4084 return Err(
4085 "pull executor: Limit reached pull_rows — Limit must appear after Project"
4086 .to_string(),
4087 );
4088 }
4089 PlanOp::Aggregate { .. } => {
4090 return Err(
4091 "pull executor: Aggregate reached pull_rows — aggregate plans must use \
4092 the execute_aggregate path (routed before pull in execute_inner)"
4093 .to_string(),
4094 );
4095 }
4096 PlanOp::VarExpand { .. } => {
4101 return Err(
4102 "pull executor: VarExpand reached pull_rows — variable-length path \
4103 plans must use the staged path (row_bound returns None)"
4104 .to_string(),
4105 );
4106 }
4107 PlanOp::ShortestPath { .. } => {
4108 return Err(
4109 "pull executor: ShortestPath reached pull_rows — shortestPath plans \
4110 must use the staged path (row_bound returns None)"
4111 .to_string(),
4112 );
4113 }
4114 PlanOp::GroupAggregate { .. } => {
4115 return Err(
4116 "pull executor: GroupAggregate reached pull_rows — grouped aggregate plans \
4117 must use the execute_group_aggregate path (routed before pull in execute_inner)"
4118 .to_string(),
4119 );
4120 }
4121 PlanOp::With { .. } => {
4122 return Err(
4123 "pull executor: With reached pull_rows — pipeline plans must use the staged path \
4124 (row_bound returns None for plans containing With)"
4125 .to_string(),
4126 );
4127 }
4128 PlanOp::Unwind { .. } => {
4129 return Err(
4130 "pull executor: Unwind reached pull_rows — pipeline plans must use the staged path \
4131 (row_bound returns None for plans containing Unwind)"
4132 .to_string(),
4133 );
4134 }
4135 PlanOp::LeftOuterApply { .. } => {
4136 return Err(
4137 "pull executor: LeftOuterApply reached pull_rows — OPTIONAL MATCH plans must use \
4138 the staged path (row_bound returns None for plans containing LeftOuterApply)"
4139 .to_string(),
4140 );
4141 }
4142 }
4143 Ok(())
4144}
4145
4146fn eval_expr(
4147 view: &GraphView,
4148 vars: &VarTable,
4149 row: &Row,
4150 expr: &Expr,
4151 params: &Params,
4152 depth: u32,
4153) -> Result<bool, String> {
4154 if depth > 256 {
4155 return Err("expression nesting too deep".into());
4156 }
4157 match expr {
4158 Expr::And(lhs, rhs) => {
4159 let l = eval_expr(view, vars, row, lhs, params, depth + 1)?;
4160 let r = eval_expr(view, vars, row, rhs, params, depth + 1)?;
4161 Ok(l && r)
4162 }
4163 Expr::Or(lhs, rhs) => {
4164 let l = eval_expr(view, vars, row, lhs, params, depth + 1)?;
4165 let r = eval_expr(view, vars, row, rhs, params, depth + 1)?;
4166 Ok(l || r)
4167 }
4168 Expr::Not(inner) => Ok(!eval_expr(view, vars, row, inner, params, depth + 1)?),
4169 Expr::Cmp { lhs, op, rhs } => {
4170 let l = resolve_operand(view, vars, row, lhs, params)?;
4171 let r = resolve_operand(view, vars, row, rhs, params)?;
4172 match (l, r) {
4173 (Some(a), Some(b)) => Ok(eval_cmp(op, &a, &b)),
4174 _ => Ok(false),
4175 }
4176 }
4177 Expr::Truthy(op) => {
4178 let val = resolve_operand(view, vars, row, op, params)?;
4179 Ok(match val {
4180 None => false,
4181 Some(Value::Bool(b)) => b,
4182 Some(Value::Int(n)) => n != 0,
4183 Some(Value::Float(f)) => f != 0.0,
4184 Some(Value::Str(s)) => !s.is_empty(),
4185 Some(Value::List(v)) => !v.is_empty(),
4186 Some(Value::Map(m)) => !m.is_empty(),
4187 })
4188 }
4189 Expr::IsNull(op) => {
4190 let val = resolve_operand(view, vars, row, op, params)?;
4191 Ok(val.is_none())
4192 }
4193 Expr::IsNotNull(op) => {
4194 let val = resolve_operand(view, vars, row, op, params)?;
4195 Ok(val.is_some())
4196 }
4197 Expr::In { expr, list } => eval_in(view, vars, row, expr, list, params),
4198 }
4199}
4200
4201fn eval_in(
4202 view: &GraphView,
4203 vars: &VarTable,
4204 row: &Row,
4205 expr: &Operand,
4206 list: &[Operand],
4207 params: &Params,
4208) -> Result<bool, String> {
4209 let Some(needle) = resolve_operand(view, vars, row, expr, params)? else {
4210 return Ok(false);
4211 };
4212 for item_op in list {
4213 match resolve_operand(view, vars, row, item_op, params)? {
4214 None => {}
4215 Some(Value::List(items)) => {
4216 for item in items {
4217 if crate::filter::eval_cmp(&crate::filter::CmpOp::Eq, &needle, &item) {
4218 return Ok(true);
4219 }
4220 }
4221 }
4222 Some(item) if crate::filter::eval_cmp(&crate::filter::CmpOp::Eq, &needle, &item) => {
4223 return Ok(true);
4224 }
4225 Some(_) => {}
4226 }
4227 }
4228 Ok(false)
4229}
4230
4231fn exec_distinct(table: &mut Projected) -> Result<(), String> {
4233 let cap = max_intermediate_rows();
4234 let mut seen: BTreeSet<Vec<Option<ValueKey>>> = BTreeSet::new();
4235 let mut out = Vec::with_capacity(table.rows.len().min(cap));
4236 for row in table.rows.drain(..) {
4237 let key: Vec<Option<ValueKey>> = row
4238 .iter()
4239 .map(|cell| cell.as_ref().and_then(group_key_normalize))
4240 .collect();
4241 if seen.insert(key) {
4242 if out.len() >= cap {
4243 return Err(row_cap_err(cap));
4244 }
4245 out.push(row);
4246 }
4247 }
4248 table.rows = out;
4249 Ok(())
4250}
4251
4252fn column_name(item: &RetItem) -> String {
4253 if let Some(alias) = &item.alias {
4254 return alias.clone();
4255 }
4256 ret_val_label(&item.value).unwrap_or_else(|| match &item.value {
4260 RetVal::Agg { func, arg } => {
4261 let f = match func {
4262 AggFunc::Count => "COUNT",
4263 AggFunc::Sum => "SUM",
4264 AggFunc::Avg => "AVG",
4265 AggFunc::Min => "MIN",
4266 AggFunc::Max => "MAX",
4267 AggFunc::Collect => "COLLECT",
4268 };
4269 format!("{f}({})", agg_arg_label(arg))
4270 }
4271 _ => unreachable!("ret_val_label names every non-aggregate item"),
4272 })
4273}
4274
4275fn exec_project(
4276 view: &GraphView,
4277 vars: &VarTable,
4278 rows: &[Row],
4279 items: &[RetItem],
4280 params: &Params,
4281) -> Result<Projected, String> {
4282 let columns: Vec<String> = items.iter().map(column_name).collect();
4283 let mut out_rows = Vec::with_capacity(rows.len());
4284 for row in rows {
4285 let mut cells = Vec::with_capacity(items.len());
4286 for item in items {
4287 cells.push(project_item(view, vars, row, item, params)?);
4288 }
4289 out_rows.push(cells);
4290 }
4291 Ok(Projected {
4292 columns,
4293 rows: out_rows,
4294 })
4295}
4296
4297fn project_item(
4298 view: &GraphView,
4299 vars: &VarTable,
4300 row: &Row,
4301 item: &RetItem,
4302 params: &Params,
4303) -> Result<Option<Value>, String> {
4304 match &item.value {
4305 RetVal::Var(v) => {
4306 let slot = vars.slot(v).ok_or_else(|| format!("unbound variable `{v}`"))?;
4308 match row.get(slot).and_then(|c| c.as_ref()) {
4309 None => Ok(None),
4311 Some(Cell::Node(id)) => match view.ids.key_of(*id) {
4312 Some(key) => Ok(Some(Value::Str(key.to_owned()))),
4313 None => Err(format!("unknown node id {id}")),
4314 },
4315 Some(Cell::Scalar(val)) => Ok(Some(val.clone())),
4317 Some(Cell::Rel(_)) => Err(format!(
4318 "variable `{v}` is a relationship; return its properties ({v}.field) instead"
4319 )),
4320 Some(Cell::Path(hops)) => Ok(Some(Value::Int(*hops as i64))),
4321 }
4322 }
4323 RetVal::Prop { var, field } => resolve_prop(view, vars, row, var, field),
4324 RetVal::Agg { .. } => Err(
4327 "project_item: Agg variant reached exec_project — aggregate plans must not contain Project"
4328 .to_string(),
4329 ),
4330 RetVal::FuncCall { name, args } => eval_func(name, args, view, vars, row, params),
4331 RetVal::ScalarExpr(op) => resolve_operand(view, vars, row, op, params),
4332 }
4333}
4334
4335fn order_column(item: &OrderItem) -> String {
4336 match &item.target {
4337 OrderTarget::Alias(name) | OrderTarget::Var(name) => name.clone(),
4338 OrderTarget::Prop { var, field } => format!("{var}.{field}"),
4339 }
4340}
4341
4342fn exec_order_by(table: &mut Projected, items: &[OrderItem]) -> Result<(), String> {
4343 let mut keys = Vec::with_capacity(items.len());
4344 for item in items {
4345 let name = order_column(item);
4346 let idx = table
4347 .columns
4348 .iter()
4349 .position(|c| c == &name)
4350 .ok_or_else(|| format!("ORDER BY target `{name}` is not a projected column"))?;
4351 keys.push((idx, item.descending));
4352 }
4353 table.rows.sort_by(|a, b| {
4354 for &(idx, desc) in &keys {
4355 let c = cmp_optional(
4356 a.get(idx).and_then(|x| x.as_ref()),
4357 b.get(idx).and_then(|x| x.as_ref()),
4358 desc,
4359 );
4360 if c != std::cmp::Ordering::Equal {
4361 return c;
4362 }
4363 }
4364 std::cmp::Ordering::Equal
4365 });
4366 Ok(())
4367}
4368
4369fn resolve_ls(ls: &LimitSkip, params: &Params) -> Result<u64, String> {
4374 match ls {
4375 LimitSkip::Exact(n) => Ok(*n),
4376 LimitSkip::Param(name) => {
4377 let val = params
4378 .0
4379 .get(name)
4380 .ok_or_else(|| format!("missing parameter `{name}` (used in LIMIT/SKIP)"))?;
4381 match val {
4382 Value::Int(i) if *i >= 0 => Ok(*i as u64),
4383 Value::Int(i) => Err(format!(
4384 "LIMIT/SKIP parameter `{name}` must be a non-negative integer, got {i}"
4385 )),
4386 other => Err(format!(
4387 "LIMIT/SKIP parameter `{name}` must be an integer, got {other:?}"
4388 )),
4389 }
4390 }
4391 }
4392}
4393
4394fn apply_skip<T>(rows: &mut Vec<T>, n: u64) {
4395 let n = usize::try_from(n).unwrap_or(usize::MAX);
4396 if n >= rows.len() {
4397 rows.clear();
4398 } else {
4399 rows.drain(0..n);
4400 }
4401}
4402
4403fn apply_limit<T>(rows: &mut Vec<T>, n: u64) {
4404 let n = usize::try_from(n).unwrap_or(usize::MAX);
4405 rows.truncate(n);
4406}
4407
4408#[cfg(test)]
4409mod tests {
4410 use super::{execute, resolve_operand, Params, Row, VarTable};
4411 use crate::cypher::ast::{
4412 ArithOp, LimitSkip, Operand, OrderItem, OrderTarget, RetItem, RetVal,
4413 };
4414 use crate::cypher::plan::{plan, PlanOp};
4415 use crate::cypher::{lex, parse, RelDir};
4416 use crate::result::ResultSet;
4417 use crate::view::GraphView;
4418 use core_storage::v8::seam::{ColumnsView, EdgePropsView, TopologyView};
4419 use core_storage::{ColumnStore, EdgeProps, IdMap, Interner, Topology, Value};
4420 use proptest::prelude::*;
4421 use std::collections::BTreeMap;
4422
4423 struct Fx {
4424 ids: IdMap,
4425 syms: Interner,
4426 labels: Vec<u32>,
4427 props: ColumnStore,
4428 topo: Topology,
4429 eprops: EdgeProps,
4430 }
4431
4432 impl Fx {
4433 fn new() -> Self {
4434 Fx {
4435 ids: IdMap::new(),
4436 syms: Interner::new(),
4437 labels: vec![],
4438 props: ColumnStore::new(),
4439 topo: Topology::new(),
4440 eprops: EdgeProps::new(),
4441 }
4442 }
4443
4444 fn add(&mut self, label: &str, key: &str, props: Vec<(&str, Value)>) -> u32 {
4445 let id = self.ids.get_or_insert(key);
4446 let sym = self.syms.intern(label);
4447 self.labels.resize(id as usize + 1, u32::MAX);
4448 self.labels[id as usize] = sym;
4449 for (f, v) in props {
4450 self.props.set(id, f, v);
4451 }
4452 id
4453 }
4454
4455 fn edge(&mut self, etype: &str, src: u32, dst: u32, props: Vec<(&str, Value)>) {
4456 let et = self.syms.intern(etype);
4457 self.topo.add_edge(et, src, dst);
4458 for (f, v) in props {
4459 self.eprops.set(et, src, dst, f, v);
4460 }
4461 }
4462
4463 fn view(&self) -> GraphView<'_> {
4464 GraphView {
4465 ids: &self.ids,
4466 syms: &self.syms,
4467 labels: &self.labels,
4468 props: ColumnsView::owned(&self.props),
4469 topo: TopologyView::owned(&self.topo),
4470 edge_props: EdgePropsView::owned(&self.eprops),
4471 mask: None,
4472 prop_index: None,
4473 }
4474 }
4475
4476 fn view_indexed<'a>(
4477 &'a self,
4478 index: &'a core_storage::property_index::PropertyIndex,
4479 ) -> GraphView<'a> {
4480 GraphView {
4481 prop_index: Some(index),
4482 ..self.view()
4483 }
4484 }
4485 }
4486
4487 fn compile(src: &str) -> Vec<PlanOp> {
4488 plan(&parse(&lex(src).expect("lex")).expect("parse")).expect("plan")
4489 }
4490
4491 fn run(
4492 view: &GraphView,
4493 src: &str,
4494 params: &BTreeMap<String, Value>,
4495 ) -> Result<ResultSet, String> {
4496 execute(view, &compile(src), &Params(params))
4497 }
4498
4499 fn s(v: &str) -> Value {
4500 Value::Str(v.into())
4501 }
4502
4503 fn f(v: f64) -> Value {
4504 Value::Float(v)
4505 }
4506
4507 fn i(v: i64) -> Value {
4508 Value::Int(v)
4509 }
4510
4511 fn rows_of(rs: &ResultSet) -> Vec<Vec<Option<Value>>> {
4512 (0..rs.len()).map(|i| rs.row(i).to_vec()).collect()
4513 }
4514
4515 fn col(rs: &ResultSet, name: &str) -> Vec<Option<Value>> {
4516 (0..rs.len()).map(|i| rs.get(i, name).cloned()).collect()
4517 }
4518
4519 fn hop_graph() -> Fx {
4520 let mut fx = Fx::new();
4521 let ada = fx.add("Person", "ada", vec![]);
4522 let bob = fx.add("Person", "bob", vec![]);
4523 let cam = fx.add("Person", "cam", vec![]);
4524 let acme = fx.add("Company", "acme", vec![]);
4525 fx.edge("KNOWS", ada, bob, vec![]);
4526 fx.edge("KNOWS", ada, cam, vec![]);
4527 fx.edge("KNOWS", bob, cam, vec![]);
4528 fx.edge("LIKES", ada, acme, vec![]);
4529 fx
4530 }
4531
4532 fn undirected_graph() -> Fx {
4533 let mut fx = Fx::new();
4534 let a = fx.add("N", "a", vec![]);
4535 let b = fx.add("N", "b", vec![]);
4536 fx.edge("T", a, b, vec![("w", i(42))]);
4537 fx
4538 }
4539
4540 fn triangle() -> Fx {
4541 let mut fx = Fx::new();
4542 let a = fx.add("N", "a", vec![]);
4543 let b = fx.add("N", "b", vec![]);
4544 let c = fx.add("N", "c", vec![]);
4545 fx.edge("T", a, b, vec![("eid", i(1))]);
4546 fx.edge("T", b, c, vec![("eid", i(2))]);
4547 fx.edge("T", c, a, vec![("eid", i(3))]);
4548 fx
4549 }
4550
4551 fn single_edge() -> (Fx, u32, u32) {
4552 let mut fx = Fx::new();
4553 let a = fx.add("N", "a", vec![]);
4554 let b = fx.add("N", "b", vec![]);
4555 fx.edge("T", a, b, vec![]);
4556 (fx, a, b)
4557 }
4558
4559 fn dogfood_graph() -> Fx {
4561 let mut fx = Fx::new();
4562 let t1 = fx.add("Talent", "t1", vec![("id", s("t1"))]);
4563 let acme = fx.add("Company", "acme", vec![]);
4564 let beta = fx.add("Company", "beta", vec![]);
4565 let gamma = fx.add("Company", "gamma", vec![]);
4566 let delta = fx.add("Company", "delta", vec![]);
4567 let echo = fx.add("Company", "echo", vec![]);
4568 let foxtrot = fx.add("Company", "foxtrot", vec![]);
4569 let zeta = fx.add("Company", "zeta", vec![]);
4570 fx.edge("INDUSTRY_ALIGNMENT", acme, t1, vec![("score", f(0.9))]);
4571 fx.edge("SPECIALTY_MATCH", acme, t1, vec![("score", f(0.8))]);
4572 fx.edge("INDUSTRY_ALIGNMENT", beta, t1, vec![("score", f(0.6))]);
4573 fx.edge("SPECIALTY_MATCH", beta, t1, vec![("score", f(0.7))]);
4574 fx.edge("INDUSTRY_ALIGNMENT", gamma, t1, vec![("score", f(0.4))]);
4575 fx.edge("SPECIALTY_MATCH", gamma, t1, vec![("score", f(0.9))]);
4576 fx.edge("INDUSTRY_ALIGNMENT", delta, t1, vec![("score", f(0.8))]);
4577 fx.edge("SPECIALTY_MATCH", delta, t1, vec![("score", f(0.3))]);
4578 fx.edge("INDUSTRY_ALIGNMENT", echo, t1, vec![("score", f(0.5))]);
4579 fx.edge("SPECIALTY_MATCH", echo, t1, vec![("score", f(0.5))]);
4580 fx.edge("INDUSTRY_ALIGNMENT", foxtrot, t1, vec![("score", f(0.95))]);
4581 fx.edge("INDUSTRY_ALIGNMENT", zeta, t1, vec![("score", f(0.9))]);
4582 fx.edge("SPECIALTY_MATCH", zeta, t1, vec![("score", f(0.6))]);
4583 fx
4584 }
4585
4586 const DOGFOOD: &str = "\
4587MATCH (t:Talent {id: $tid}) \
4588MATCH (c:Company)-[i:INDUSTRY_ALIGNMENT]->(t) \
4589MATCH (c)-[s:SPECIALTY_MATCH]->(t) \
4590WHERE i.score >= 0.5 AND s.score >= 0.5 \
4591RETURN c, i.score AS industry, s.score AS specialty \
4592ORDER BY industry DESC, specialty DESC \
4593LIMIT 10";
4594
4595 fn tid_params() -> BTreeMap<String, Value> {
4596 let mut p = BTreeMap::new();
4597 p.insert("tid".into(), s("t1"));
4598 p
4599 }
4600
4601 #[test]
4602 fn single_hop_match_label_and_etype_filters() {
4603 let fx = hop_graph();
4604 let v = fx.view();
4605 let rs = run(
4606 &v,
4607 "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b",
4608 &BTreeMap::new(),
4609 )
4610 .expect("single-hop");
4611 assert_eq!(rs.columns(), &["a".to_string(), "b".to_string()]);
4612 assert_eq!(
4613 rows_of(&rs),
4614 vec![
4615 vec![Some(s("ada")), Some(s("bob"))],
4616 vec![Some(s("ada")), Some(s("cam"))],
4617 vec![Some(s("bob")), Some(s("cam"))],
4618 ]
4619 );
4620 let likes = run(
4622 &v,
4623 "MATCH (a:Person)-[:LIKES]->(b:Company) RETURN a, b",
4624 &BTreeMap::new(),
4625 )
4626 .unwrap();
4627 assert_eq!(rows_of(&likes), vec![vec![Some(s("ada")), Some(s("acme"))]]);
4628 let no_combo = run(
4629 &v,
4630 "MATCH (a:Person)-[:KNOWS]->(b:Company) RETURN a, b",
4631 &BTreeMap::new(),
4632 )
4633 .unwrap();
4634 assert!(no_combo.is_empty());
4635 }
4636
4637 #[test]
4638 fn undirected_match_finds_both_orientations_and_binds_true_triple() {
4639 let fx = undirected_graph();
4640 let v = fx.view();
4641 let rs =
4644 run(&v, "MATCH (x)-[r:T]-(y) RETURN x, y, r.w", &BTreeMap::new()).expect("undirected");
4645 assert_eq!(
4646 rows_of(&rs),
4647 vec![
4648 vec![Some(s("a")), Some(s("b")), Some(i(42))],
4649 vec![Some(s("b")), Some(s("a")), Some(i(42))],
4650 ]
4651 );
4652
4653 let left = run(
4654 &v,
4655 "MATCH (y)<-[r:T]-(x) RETURN x, y, r.w",
4656 &BTreeMap::new(),
4657 )
4658 .unwrap();
4659 assert_eq!(
4660 rows_of(&left),
4661 vec![vec![Some(s("a")), Some(s("b")), Some(i(42))]]
4662 );
4663 }
4664
4665 #[test]
4666 fn relationship_uniqueness_triangle_and_two_hop_cycle() {
4667 let tri = triangle();
4668 let v = tri.view();
4669 let rs = run(
4670 &v,
4671 "MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z, r1.eid, r2.eid",
4672 &BTreeMap::new(),
4673 )
4674 .expect("triangle 2-hop");
4675 assert_eq!(
4676 rows_of(&rs),
4677 vec![
4678 vec![
4679 Some(s("a")),
4680 Some(s("b")),
4681 Some(s("c")),
4682 Some(i(1)),
4683 Some(i(2))
4684 ],
4685 vec![
4686 Some(s("b")),
4687 Some(s("c")),
4688 Some(s("a")),
4689 Some(i(2)),
4690 Some(i(3))
4691 ],
4692 vec![
4693 Some(s("c")),
4694 Some(s("a")),
4695 Some(s("b")),
4696 Some(i(3)),
4697 Some(i(1))
4698 ],
4699 ]
4700 );
4701 for row in rows_of(&rs) {
4702 assert_ne!(row[3], row[4], "r1 must never bind the same edge as r2");
4703 }
4704
4705 let (mut one, a, b) = single_edge();
4706 let v = one.view();
4707 let cycle = run(
4708 &v,
4709 "MATCH (x)-[:T]->(y)-[:T]->(x) RETURN x",
4710 &BTreeMap::new(),
4711 )
4712 .expect("2-hop cycle");
4713 assert!(
4714 cycle.is_empty(),
4715 "single directed edge cannot close a 2-hop cycle"
4716 );
4717
4718 let undirected_cycle = run(&v, "MATCH (x)-[:T]-(y)-[:T]-(x) RETURN x", &BTreeMap::new())
4720 .expect("undirected uniqueness");
4721 assert!(
4722 undirected_cycle.is_empty(),
4723 "relationship uniqueness must reject walking the same triple back"
4724 );
4725
4726 one.edge("T", b, a, vec![]);
4727 let v = one.view();
4728 let with_recip = run(
4729 &v,
4730 "MATCH (x)-[:T]->(y)-[:T]->(x) RETURN x",
4731 &BTreeMap::new(),
4732 )
4733 .unwrap();
4734 assert_eq!(col(&with_recip, "x"), vec![Some(s("a")), Some(s("b"))]);
4735 }
4736
4737 #[test]
4738 fn multi_match_join_bound_and_bound_destination_expand() {
4739 let fx = dogfood_graph();
4740 let v = fx.view();
4741 let rs = run(
4744 &v,
4745 "MATCH (t:Talent {id: $tid}) \
4746 MATCH (c:Company)-[i:INDUSTRY_ALIGNMENT]->(t) \
4747 MATCH (c)-[s:SPECIALTY_MATCH]->(t) \
4748 RETURN c",
4749 &tid_params(),
4750 )
4751 .expect("join + bound dest");
4752 assert_eq!(
4753 col(&rs, "c"),
4754 vec![
4755 Some(s("acme")),
4756 Some(s("beta")),
4757 Some(s("gamma")),
4758 Some(s("delta")),
4759 Some(s("echo")),
4760 Some(s("zeta")),
4761 ]
4762 );
4763
4764 let keep = run(&v, "MATCH (c:Company) MATCH (c) RETURN c", &BTreeMap::new()).unwrap();
4766 assert_eq!(keep.len(), 7);
4767 let drop = run(
4769 &v,
4770 "MATCH (c:Company) MATCH (c:Talent) RETURN c",
4771 &BTreeMap::new(),
4772 )
4773 .unwrap();
4774 assert!(drop.is_empty());
4775 }
4776
4777 #[test]
4778 fn scan_key_exec_does_not_use_label_scan() {
4779 let mut fx = Fx::new();
4780 fx.add("Person", "p1", vec![]);
4781 fx.add("Person", "p2", vec![]);
4782 fx.add("Person", "p3", vec![]);
4783 fx.add("Company", "c1", vec![]);
4784 let v = fx.view();
4785 let params = BTreeMap::new();
4786
4787 let fires_before = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
4788 let rs = run(&v, "MATCH (n:Person {id: 'p2'}) RETURN n", ¶ms).expect("scan key");
4789 let fires_after = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
4790 assert!(
4791 fires_after > fires_before,
4792 "SCAN_KEY_FIRES must increment; before={fires_before} after={fires_after}"
4793 );
4794 assert_eq!(rows_of(&rs), vec![vec![Some(s("p2"))]]);
4795
4796 let miss = run(&v, "MATCH (n:Person {id: 'nope'}) RETURN n", ¶ms).unwrap();
4797 assert!(
4798 miss.is_empty(),
4799 "missing key must be zero rows, not an error"
4800 );
4801
4802 let wrong = run(&v, "MATCH (n:Person {id: 'c1'}) RETURN n", ¶ms).unwrap();
4803 assert!(
4804 wrong.is_empty(),
4805 "wrong label must be zero rows, not an error"
4806 );
4807 }
4808
4809 #[test]
4810 fn rel_var_edge_prop_filter() {
4811 let mut fx = Fx::new();
4812 let a = fx.add("N", "a", vec![]);
4813 let b = fx.add("N", "b", vec![]);
4814 let c = fx.add("N", "c", vec![]);
4815 fx.edge("T", a, b, vec![("w", f(0.7))]);
4816 fx.edge("T", a, c, vec![("w", f(0.3))]);
4817 let v = fx.view();
4818 let rs = run(
4819 &v,
4820 "MATCH (x)-[r:T]->(y) WHERE r.w >= 0.5 RETURN y, r.w",
4821 &BTreeMap::new(),
4822 )
4823 .expect("edge-prop filter");
4824 assert_eq!(rows_of(&rs), vec![vec![Some(s("b")), Some(f(0.7))]]);
4825 let fail = run(
4826 &v,
4827 "MATCH (x)-[r:T]->(y) WHERE r.w >= 0.8 RETURN y",
4828 &BTreeMap::new(),
4829 )
4830 .unwrap();
4831 assert!(fail.is_empty());
4832 }
4833
4834 #[test]
4835 fn params_present_resolve_missing_is_err_before_rows() {
4836 let fx = dogfood_graph();
4837 let v = fx.view();
4838 let hit =
4839 run(&v, "MATCH (t:Talent {id: $tid}) RETURN t", &tid_params()).expect("present param");
4840 assert_eq!(col(&hit, "t"), vec![Some(s("t1"))]);
4841
4842 let err = run(
4844 &v,
4845 "MATCH (t:NoSuchLabel {id: $tid}) RETURN t",
4846 &BTreeMap::new(),
4847 )
4848 .expect_err("missing param must be Err, not Ok(empty)");
4849 assert!(
4850 err.contains("tid")
4851 && (err.contains("param") || err.contains("Param") || err.contains("missing")),
4852 "missing-param error must name the parameter, got: {err}"
4853 );
4854
4855 let err = run(
4856 &v,
4857 "MATCH (t:Talent) WHERE t.id = $tid RETURN t",
4858 &BTreeMap::new(),
4859 )
4860 .expect_err("missing WHERE param");
4861 assert!(err.contains("tid"), "got: {err}");
4862 }
4863
4864 #[test]
4865 fn order_by_none_last_then_skip_limit() {
4866 let mut fx = Fx::new();
4867 fx.add("Person", "ada", vec![("age", i(30))]);
4868 fx.add("Person", "bob", vec![]); fx.add("Person", "cam", vec![("age", i(10))]);
4870 fx.add("Person", "dan", vec![("age", i(20))]);
4871 let v = fx.view();
4872
4873 let asc = run(
4874 &v,
4875 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age",
4876 &BTreeMap::new(),
4877 )
4878 .unwrap();
4879 assert_eq!(
4880 rows_of(&asc),
4881 vec![
4882 vec![Some(s("cam")), Some(i(10))],
4883 vec![Some(s("dan")), Some(i(20))],
4884 vec![Some(s("ada")), Some(i(30))],
4885 vec![Some(s("bob")), None],
4886 ]
4887 );
4888
4889 let desc = run(
4890 &v,
4891 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age DESC",
4892 &BTreeMap::new(),
4893 )
4894 .unwrap();
4895 assert_eq!(
4896 rows_of(&desc),
4897 vec![
4898 vec![Some(s("ada")), Some(i(30))],
4899 vec![Some(s("dan")), Some(i(20))],
4900 vec![Some(s("cam")), Some(i(10))],
4901 vec![Some(s("bob")), None],
4902 ]
4903 );
4904
4905 let skip_lim = run(
4906 &v,
4907 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age SKIP 1 LIMIT 2",
4908 &BTreeMap::new(),
4909 )
4910 .unwrap();
4911 assert_eq!(
4912 rows_of(&skip_lim),
4913 vec![
4914 vec![Some(s("dan")), Some(i(20))],
4915 vec![Some(s("ada")), Some(i(30))],
4916 ]
4917 );
4918
4919 let desc_sl = run(
4920 &v,
4921 "MATCH (p:Person) RETURN p, p.age AS age ORDER BY age DESC SKIP 1 LIMIT 2",
4922 &BTreeMap::new(),
4923 )
4924 .unwrap();
4925 assert_eq!(
4926 rows_of(&desc_sl),
4927 vec![
4928 vec![Some(s("dan")), Some(i(20))],
4929 vec![Some(s("cam")), Some(i(10))],
4930 ]
4931 );
4932 }
4933
4934 #[test]
4935 fn unknown_label_and_etype_are_ok_empty() {
4936 let fx = hop_graph();
4937 let v = fx.view();
4938 let lab = run(&v, "MATCH (x:Nope) RETURN x", &BTreeMap::new()).expect("unknown label");
4939 assert!(lab.is_empty());
4940 let et = run(
4941 &v,
4942 "MATCH (a)-[:NO_SUCH_ETYPE]->(b) RETURN a",
4943 &BTreeMap::new(),
4944 )
4945 .expect("unknown etype");
4946 assert!(et.is_empty());
4947 }
4948
4949 #[test]
4950 fn execute_is_deterministic() {
4951 let fx = dogfood_graph();
4952 let v = fx.view();
4953 let p = tid_params();
4954 let a = run(&v, DOGFOOD, &p).expect("first");
4955 let b = run(&v, DOGFOOD, &p).expect("second");
4956 assert_eq!(a, b);
4957 let hop = hop_graph();
4958 let hv = hop.view();
4959 let q = "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b";
4960 assert_eq!(run(&hv, q, &BTreeMap::new()), run(&hv, q, &BTreeMap::new()));
4961 }
4962
4963 #[test]
4964 fn dogfood_pipeline_exact_rows() {
4965 let fx = dogfood_graph();
4966 let v = fx.view();
4967 let rs = run(&v, DOGFOOD, &tid_params()).expect("dogfood");
4968 assert_eq!(
4969 rs.columns(),
4970 &[
4971 "c".to_string(),
4972 "industry".to_string(),
4973 "specialty".to_string()
4974 ]
4975 );
4976 assert_eq!(
4978 rows_of(&rs),
4979 vec![
4980 vec![Some(s("acme")), Some(f(0.9)), Some(f(0.8))],
4981 vec![Some(s("zeta")), Some(f(0.9)), Some(f(0.6))],
4982 vec![Some(s("beta")), Some(f(0.6)), Some(f(0.7))],
4983 vec![Some(s("echo")), Some(f(0.5)), Some(f(0.5))],
4984 ]
4985 );
4986 }
4987
4988 #[test]
4989 fn unknown_var_in_op_is_err_not_panic() {
4990 let fx = hop_graph();
4991 let v = fx.view();
4992 let plan = vec![
4993 PlanOp::ScanLabel {
4994 var: "a".into(),
4995 label: None,
4996 },
4997 PlanOp::Expand {
4998 from: "zzz".into(),
4999 rel_var: Some("r".into()),
5000 etypes: vec![],
5001 dir: RelDir::Right,
5002 to: "b".into(),
5003 to_label: None,
5004 to_props: vec![],
5005 },
5006 PlanOp::Project {
5007 items: vec![RetItem {
5008 value: RetVal::Var("a".into()),
5009 alias: None,
5010 }],
5011 },
5012 ];
5013 let params = BTreeMap::new();
5014 let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
5015 execute(&v, &plan, &Params(¶ms))
5016 }));
5017 assert!(caught.is_ok(), "execute panicked on unknown var");
5018 let err = caught.unwrap().expect_err("unknown var must be Err");
5019 assert!(
5020 err.contains("zzz") && err.to_ascii_lowercase().contains("unbound"),
5021 "got: {err}"
5022 );
5023
5024 let join = vec![
5025 PlanOp::JoinBound {
5026 var: "ghost".into(),
5027 label: None,
5028 props: vec![],
5029 },
5030 PlanOp::Project {
5031 items: vec![RetItem {
5032 value: RetVal::Var("ghost".into()),
5033 alias: None,
5034 }],
5035 },
5036 ];
5037 let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
5038 execute(&v, &join, &Params(¶ms))
5039 }));
5040 assert!(caught.is_ok(), "execute panicked on JoinBound unknown var");
5041 assert!(caught.unwrap().is_err());
5042 }
5043
5044 #[test]
5045 fn dest_props_on_bound_expand_are_applied() {
5046 let fx = dogfood_graph();
5047 let v = fx.view();
5048 let rs = run(
5049 &v,
5050 "MATCH (t:Talent {id: $tid}) \
5051 MATCH (c:Company)-[r:INDUSTRY_ALIGNMENT]->(t:Talent {id: $tid}) \
5052 RETURN c",
5053 &tid_params(),
5054 )
5055 .unwrap();
5056 assert_eq!(
5058 col(&rs, "c"),
5059 vec![
5060 Some(s("acme")),
5061 Some(s("beta")),
5062 Some(s("gamma")),
5063 Some(s("delta")),
5064 Some(s("echo")),
5065 Some(s("foxtrot")),
5066 Some(s("zeta")),
5067 ]
5068 );
5069 let miss = run(
5070 &v,
5071 "MATCH (t:Talent {id: $tid}) \
5072 MATCH (c:Company)-[r:INDUSTRY_ALIGNMENT]->(t {id: 'nope'}) \
5073 RETURN c",
5074 &tid_params(),
5075 )
5076 .unwrap();
5077 assert!(miss.is_empty());
5078 }
5079
5080 #[test]
5081 fn missing_node_prop_projects_none_and_pattern_misses() {
5082 let mut fx = Fx::new();
5083 fx.add("Person", "ada", vec![("age", i(30))]);
5084 fx.add("Person", "bob", vec![]);
5085 let v = fx.view();
5086 let rs = run(&v, "MATCH (p:Person) RETURN p.age", &BTreeMap::new()).unwrap();
5087 assert_eq!(col(&rs, "p.age"), vec![Some(i(30)), None]);
5088 let pat = run(&v, "MATCH (p:Person {age: 30}) RETURN p", &BTreeMap::new()).unwrap();
5089 assert_eq!(col(&pat, "p"), vec![Some(s("ada"))]);
5090 }
5091
5092 #[test]
5093 fn unlabeled_match_does_not_project_sentinel_ghost_key() {
5094 let mut fx = Fx::new();
5095 fx.add("Person", "ada", vec![]);
5096 fx.ids.get_or_insert("ghost");
5099 fx.labels.resize(fx.ids.len(), u32::MAX);
5100 fx.add("Person", "bob", vec![]);
5101 let v = fx.view();
5102 let rs = run(&v, "MATCH (n) RETURN n", &BTreeMap::new()).expect("unlabeled scan");
5103 assert_eq!(col(&rs, "n"), vec![Some(s("ada")), Some(s("bob"))]);
5104 assert!(
5105 !rows_of(&rs)
5106 .iter()
5107 .any(|row| row.iter().any(|c| *c == Some(s("ghost")))),
5108 "sentinel slot must not project a ghost key"
5109 );
5110 }
5111
5112 #[test]
5113 fn execute_never_panics_on_hostile_plans() {
5114 let fx = hop_graph();
5115 let v = fx.view();
5116 let params = BTreeMap::new();
5117 let hostile = vec![
5118 vec![],
5119 vec![PlanOp::Project { items: vec![] }],
5120 vec![PlanOp::OrderBy {
5121 items: vec![OrderItem {
5122 target: OrderTarget::Alias("nope".into()),
5123 descending: false,
5124 }],
5125 }],
5126 vec![PlanOp::Filter {
5127 expr: crate::cypher::ast::Expr::Cmp {
5128 lhs: Operand::Prop {
5129 var: "missing".into(),
5130 field: "x".into(),
5131 },
5132 op: crate::filter::CmpOp::Eq,
5133 rhs: Operand::Lit(i(1)),
5134 },
5135 }],
5136 vec![
5137 PlanOp::ScanLabel {
5138 var: "a".into(),
5139 label: None,
5140 },
5141 PlanOp::LookupProps {
5142 var: "zzz".into(),
5143 props: vec![("k".into(), Operand::Lit(i(1)))],
5144 },
5145 ],
5146 vec![
5147 PlanOp::Skip(LimitSkip::Exact(99)),
5148 PlanOp::Limit(LimitSkip::Exact(0)),
5149 ],
5150 ];
5151 for plan in hostile {
5152 let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
5153 execute(&v, &plan, &Params(¶ms))
5154 }));
5155 assert!(caught.is_ok(), "execute panicked on hostile plan: {plan:?}");
5156 }
5157 }
5158
5159 #[test]
5160 fn unlabeled_and_labeled_scans_skip_tombstoned_ids() {
5161 let mut fx = Fx::new();
5162 let ada = fx.add("Person", "ada", vec![]);
5163 let bob = fx.add("Person", "bob", vec![]);
5164 fx.edge("KNOWS", ada, bob, vec![]);
5165 fx.ids.delete("ada");
5167 fx.labels[ada as usize] = u32::MAX;
5168 let knows = fx.syms.get("KNOWS").unwrap();
5169 fx.topo.remove_edge(knows, ada, bob);
5170 let v = fx.view();
5171
5172 let labeled = run(&v, "MATCH (p:Person) RETURN p", &BTreeMap::new()).unwrap();
5173 assert_eq!(col(&labeled, "p"), vec![Some(s("bob"))]);
5174 let unlabeled = run(&v, "MATCH (n) RETURN n", &BTreeMap::new()).unwrap();
5175 assert_eq!(col(&unlabeled, "n"), vec![Some(s("bob"))]);
5176 let hop = run(&v, "MATCH (x)-[:KNOWS]->(y) RETURN x, y", &BTreeMap::new()).unwrap();
5177 assert!(
5178 hop.is_empty(),
5179 "expand cannot yield edges to a deleted node once topology is swept"
5180 );
5181 }
5182
5183 proptest! {
5188 #[test]
5201 fn prop_bounded_equals_unbounded_slice(
5202 n_nodes in 2u32..10u32,
5203 edge_pairs in proptest::collection::vec(
5204 (any::<u32>(), any::<u32>()), 0..20usize
5205 ),
5206 recip_pairs in proptest::collection::vec(
5208 (any::<u32>(), any::<u32>()), 0..8usize
5209 ),
5210 n_hops in 1u32..4u32, use_filter in any::<bool>(),
5212 threshold in 0i64..8i64, limit in 1u64..10u64,
5214 skip in 0u64..4u64,
5215 ) {
5216 let mut fx = Fx::new();
5217 let mut node_ids = Vec::new();
5218 for idx in 0..n_nodes {
5219 let id = fx.add("N", &format!("n{idx}"), vec![("v", i(idx as i64 % 8))]);
5221 node_ids.push(id);
5222 }
5223 let n = node_ids.len();
5224
5225 for (si, di) in &edge_pairs {
5227 let si = (*si as usize) % n;
5228 let di = (*di as usize) % n;
5229 if si != di {
5230 fx.edge("T", node_ids[si], node_ids[di], vec![]);
5231 }
5232 }
5233 for (si, di) in &recip_pairs {
5236 let si = (*si as usize) % n;
5237 let di = (*di as usize) % n;
5238 if si != di {
5239 fx.edge("T", node_ids[si], node_ids[di], vec![]);
5240 fx.edge("T", node_ids[di], node_ids[si], vec![]);
5241 }
5242 }
5243
5244 let v = fx.view();
5245 let params = BTreeMap::new();
5246
5247 let var_names = ["a", "b", "c", "d"];
5250 let hop_count = n_hops as usize;
5251 let mut pattern = format!("({}:N)", var_names[0]);
5252 for h in 0..hop_count {
5253 pattern.push_str(&format!("-[:T]->({}", var_names[h + 1]));
5254 if h + 1 == hop_count {
5256 pattern.push_str(":N)");
5257 } else {
5258 pattern.push(')');
5259 }
5260 }
5261 let last_var = var_names[hop_count];
5262 let where_clause = if use_filter {
5263 format!(" WHERE {last_var}.v > {threshold}")
5264 } else {
5265 String::new()
5266 };
5267 let ret_vars: Vec<&str> = var_names[..=hop_count].to_vec();
5268 let ret_clause = ret_vars.join(", ");
5269
5270 let full_q = format!("MATCH {pattern}{where_clause} RETURN {ret_clause}");
5271 let bounded_q = format!(
5272 "MATCH {pattern}{where_clause} RETURN {ret_clause} SKIP {skip} LIMIT {limit}"
5273 );
5274
5275 let full_plan = compile(&full_q);
5276 let unbounded = super::with_max_intermediate_rows(100_000, || {
5279 super::execute_unbounded(&v, &full_plan, &Params(¶ms))
5280 });
5281 let unbounded = match unbounded {
5284 Ok(rs) => rs,
5285 Err(_) => return Ok(()),
5286 };
5287 let total = unbounded.len();
5288 let full_rows = rows_of(&unbounded);
5289
5290 let bounded = super::with_max_intermediate_rows(100_000, || {
5291 run(&v, &bounded_q, ¶ms)
5292 }).expect("bounded must not error");
5293
5294 let s = (skip as usize).min(total);
5295 let e = (skip as usize + limit as usize).min(total);
5296 prop_assert_eq!(
5297 rows_of(&bounded),
5298 full_rows[s..e].to_vec(),
5299 "hops={} filter={} threshold={} SKIP {} LIMIT {}: \
5300 bounded != unbounded[{}..{}]",
5301 hop_count, use_filter, threshold, skip, limit, s, e
5302 );
5303 }
5304 }
5305
5306 proptest! {
5307 #[test]
5325 fn prop_scan_filter_fused_equals_unbounded_slice(
5326 n_nodes in 0u32..20u32,
5327 prop_mask in any::<u32>(),
5329 float_mask in any::<u32>(),
5331 threshold in -1i64..8i64,
5334 op_idx in 0u32..6u32,
5336 skip in 0u64..5u64,
5337 limit in 1u64..8u64,
5338 ) {
5339 let op_str = match op_idx {
5340 0 => "=",
5341 1 => "<>",
5342 2 => "<",
5343 3 => "<=",
5344 4 => ">",
5345 _ => ">=",
5346 };
5347
5348 let mut fx = Fx::new();
5349 for idx in 0..n_nodes {
5350 let has_prop = (prop_mask >> (idx % 32)) & 1 == 1;
5351 let use_float = (float_mask >> (idx % 32)) & 1 == 1;
5352 let props: Vec<(&str, Value)> = if has_prop {
5354 if use_float {
5355 vec![("v", f(idx as f64 % 7.0))]
5356 } else {
5357 vec![("v", i(idx as i64 % 7))]
5358 }
5359 } else {
5360 vec![]
5361 };
5362 fx.add("N", &format!("n{idx}"), props);
5363 }
5364
5365 let v = fx.view();
5366 let params = BTreeMap::new();
5367
5368 let full_q = format!("MATCH (n:N) WHERE n.v {op_str} {threshold} RETURN n");
5370 let bounded_q = format!(
5371 "MATCH (n:N) WHERE n.v {op_str} {threshold} RETURN n SKIP {skip} LIMIT {limit}"
5372 );
5373
5374 let full_plan = compile(&full_q);
5375 let unbounded = super::execute_unbounded(&v, &full_plan, &Params(¶ms));
5376 let unbounded = match unbounded {
5377 Ok(rs) => rs,
5378 Err(_) => return Ok(()),
5379 };
5380 let total = unbounded.len();
5381 let full_rows = rows_of(&unbounded);
5382
5383 let fires_before =
5385 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
5386 let bounded = run(&v, &bounded_q, ¶ms).expect("fused bounded must not error");
5387 let fires_after =
5388 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
5389
5390 if n_nodes > 0 && op_idx != 0 {
5398 prop_assert!(
5399 fires_after > fires_before,
5400 "fused arm did NOT fire for op={} threshold={} n_nodes={}: \
5401 counter before={} after={}",
5402 op_str,
5403 threshold,
5404 n_nodes,
5405 fires_before,
5406 fires_after
5407 );
5408 }
5409
5410 let s = (skip as usize).min(total);
5411 let e = (skip as usize + limit as usize).min(total);
5412 prop_assert_eq!(
5413 rows_of(&bounded),
5414 full_rows[s..e].to_vec(),
5415 "fused path: op={} threshold={} n_nodes={} SKIP {} LIMIT {}: \
5416 bounded != unbounded[{}..{}]",
5417 op_str, threshold, n_nodes, skip, limit, s, e
5418 );
5419
5420 let compound_q = format!(
5424 "MATCH (n:N) WHERE n.v {} {} AND n.v >= -999 RETURN n SKIP {} LIMIT {}",
5425 op_str, threshold, skip, limit
5426 );
5427 let fires_before_compound =
5428 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
5429 let compound_bounded =
5430 run(&v, &compound_q, ¶ms).expect("compound-AND bounded must not error");
5431 let fires_after_compound =
5432 super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
5433
5434 prop_assert_eq!(
5436 fires_after_compound,
5437 fires_before_compound,
5438 "fused arm fired for compound-AND shape (should use generic path): \
5439 op={} threshold={} n_nodes={}",
5440 op_str,
5441 threshold,
5442 n_nodes
5443 );
5444 prop_assert_eq!(
5445 rows_of(&compound_bounded),
5446 full_rows[s..e].to_vec(),
5447 "compound-AND fallback: op={} threshold={} n_nodes={} SKIP {} LIMIT {}: \
5448 result differs from fused",
5449 op_str, threshold, n_nodes, skip, limit
5450 );
5451 }
5452 }
5453
5454 #[test]
5465 fn dense_hop1_with_filter_survives_pull() {
5466 const LEAVES: usize = 120; const CAP: usize = 100;
5468
5469 let mut fx = Fx::new();
5470 let src = fx.add("Src", "src", vec![]);
5471 for idx in 0..LEAVES {
5472 let leaf = fx.add("Leaf", &format!("l{idx}"), vec![("v", i(idx as i64))]);
5473 fx.edge("T", src, leaf, vec![]);
5474 }
5475 let v = fx.view();
5476 let params = BTreeMap::new();
5477
5478 let staged_err = super::with_max_intermediate_rows(CAP, || {
5480 super::execute_unbounded(
5481 &v,
5482 &compile("MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v >= 110 RETURN l, l.v"),
5483 &Params(¶ms),
5484 )
5485 });
5486 assert!(
5487 staged_err.is_err(),
5488 "staged path must error on 120 leaves with cap={CAP}"
5489 );
5490 assert!(
5491 staged_err
5492 .unwrap_err()
5493 .contains("intermediate result exceeds"),
5494 "wrong error message"
5495 );
5496
5497 let ok = super::with_max_intermediate_rows(CAP, || {
5501 run(
5502 &v,
5503 "MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v >= 110 RETURN l, l.v LIMIT 5",
5504 ¶ms,
5505 )
5506 });
5507 let rs = ok.expect("pull-based must survive despite dense hop-1 exceeding cap");
5508 assert_eq!(rs.len(), 5, "LIMIT 5 must return exactly 5 rows");
5509
5510 let vs: Vec<i64> = (0..rs.len())
5512 .filter_map(|i| match rs.get(i, "l.v") {
5513 Some(Value::Int(n)) => Some(*n),
5514 _ => None,
5515 })
5516 .collect();
5517 assert_eq!(vs.len(), 5, "all projected rows must have v");
5518 for v_val in &vs {
5519 assert!(*v_val >= 110, "filter must hold: v={v_val} is not >= 110");
5520 }
5521
5522 let (pull_result, pull_produced) = super::with_expand_counter(|| {
5526 super::with_max_intermediate_rows(1_000_000, || {
5527 run(
5528 &v,
5529 "MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v < 10 RETURN l LIMIT 5",
5530 ¶ms,
5531 )
5532 })
5533 });
5534 pull_result.expect("pull must succeed without cap");
5535 assert!(
5537 pull_produced <= 5,
5538 "pull expand count {pull_produced} should be ≤ 5 (stops after 5 passing leaves)"
5539 );
5540
5541 let (staged_result, staged_produced) = super::with_expand_counter(|| {
5542 super::with_max_intermediate_rows(1_000_000, || {
5543 super::execute_unbounded(
5544 &v,
5545 &compile("MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v < 10 RETURN l"),
5546 &Params(¶ms),
5547 )
5548 })
5549 });
5550 staged_result.expect("staged must succeed with 1M cap");
5551 assert_eq!(
5552 staged_produced, LEAVES,
5553 "staged must expand all {LEAVES} leaves"
5554 );
5555
5556 assert!(
5557 staged_produced >= pull_produced * 10,
5558 "staged ({staged_produced}) must be ≥ 10× pull ({pull_produced})"
5559 );
5560 }
5561
5562 #[test]
5580 fn harness_shape_two_hop_dense_survives_pull() {
5581 const N_TALENT: usize = 70;
5582 const N_COMPANY: usize = 20;
5583 const N_INDUSTRY: usize = 3;
5584 const CAP: usize = 100;
5585 const LIMIT: usize = 10;
5586
5587 let mut fx = Fx::new();
5588
5589 let mut talent_ids: Vec<u32> = Vec::new();
5591 let mut talent_industry: Vec<usize> = Vec::new();
5592 for i in 0..N_TALENT {
5593 let ind = i % N_INDUSTRY;
5594 let id = fx.add(
5595 "Talent",
5596 &format!("t{i}"),
5597 vec![("industry", s(&ind.to_string()))],
5598 );
5599 talent_ids.push(id);
5600 talent_industry.push(ind);
5601 }
5602
5603 let mut company_ids: Vec<u32> = Vec::new();
5605 let mut company_industry: Vec<usize> = Vec::new();
5606 for i in 0..N_COMPANY {
5607 let ind = i % N_INDUSTRY;
5608 let id = fx.add(
5609 "Company",
5610 &format!("c{i}"),
5611 vec![("industry", s(&ind.to_string()))],
5612 );
5613 company_ids.push(id);
5614 company_industry.push(ind);
5615 }
5616
5617 for (ti, &tid) in talent_ids.iter().enumerate() {
5619 for (ci, &cid) in company_ids.iter().enumerate() {
5620 if talent_industry[ti] == company_industry[ci] {
5621 fx.edge("INDUSTRY_ALIGNMENT", tid, cid, vec![]);
5622 }
5623 }
5624 }
5625
5626 let v = fx.view();
5627 let params = BTreeMap::new();
5628 let query = format!(
5629 "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company)\
5630 <-[:INDUSTRY_ALIGNMENT]-(t2:Talent) RETURN t, c, t2 LIMIT {LIMIT}"
5631 );
5632
5633 const UNBOUNDED_Q: &str = "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company)\
5635 <-[:INDUSTRY_ALIGNMENT]-(t2:Talent) RETURN t, c, t2";
5636 let staged_err = super::with_max_intermediate_rows(CAP, || {
5637 super::execute_unbounded(&v, &compile(UNBOUNDED_Q), &Params(¶ms))
5638 });
5639 assert!(
5640 staged_err.is_err(),
5641 "staged must error with cap={CAP} on harness-shape graph"
5642 );
5643 assert!(
5644 staged_err
5645 .unwrap_err()
5646 .contains("intermediate result exceeds"),
5647 "wrong error"
5648 );
5649
5650 let ok = super::with_max_intermediate_rows(CAP, || run(&v, &query, ¶ms));
5652 let rs = ok.expect("pull-based must complete on harness-shape with LIMIT 10");
5653 assert_eq!(rs.len(), LIMIT, "must return exactly {LIMIT} rows");
5654
5655 for i in 0..rs.len() {
5660 let row = rs.row(i);
5661 assert_eq!(row.len(), 3, "each row must have 3 columns (t, c, t2)");
5662 assert!(row.iter().all(|c| c.is_some()), "all cells must be Some");
5663 }
5664 }
5665
5666 #[test]
5667 fn intermediate_row_cap_errors_on_scan_and_expand() {
5668 let cap_msg = |n: usize| {
5669 format!(
5670 "intermediate result exceeds {n} rows; add a LIMIT or constrain patterns with shared variables"
5671 )
5672 };
5673
5674 let mut scan_fx = Fx::new();
5675 scan_fx.add("N", "a", vec![]);
5676 scan_fx.add("N", "b", vec![]);
5677 scan_fx.add("N", "c", vec![]);
5678 let sv = scan_fx.view();
5679 let scan_err = super::with_max_intermediate_rows(2, || {
5680 run(&sv, "MATCH (n:N) RETURN n", &BTreeMap::new())
5681 })
5682 .expect_err("3-row scan must exceed cap 2");
5683 assert_eq!(scan_err, cap_msg(2));
5684
5685 let mut exp_fx = Fx::new();
5686 let src = exp_fx.add("Src", "s", vec![]);
5687 let d1 = exp_fx.add("Dst", "d1", vec![]);
5688 let d2 = exp_fx.add("Dst", "d2", vec![]);
5689 exp_fx.edge("T", src, d1, vec![]);
5690 exp_fx.edge("T", src, d2, vec![]);
5691 let ev = exp_fx.view();
5692 let exp_err = super::with_max_intermediate_rows(1, || {
5694 run(&ev, "MATCH (x:Src)-[:T]->(y) RETURN x, y", &BTreeMap::new())
5695 })
5696 .expect_err("2-row expand must exceed cap 1");
5697 assert_eq!(exp_err, cap_msg(1));
5698 }
5699
5700 #[test]
5710 fn bounded_matches_unbounded_slice_various_limits() {
5711 let fx = hop_graph();
5713 let v = fx.view();
5714 let params = BTreeMap::new();
5715
5716 let full_plan = compile("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b");
5718 let full_rs = super::execute_unbounded(&v, &full_plan, &Params(¶ms)).unwrap();
5719 let full_rows = rows_of(&full_rs);
5720 assert_eq!(full_rows.len(), 3, "hop_graph has exactly 3 KNOWS paths");
5721
5722 for limit in [1u64, 2, 3, 10] {
5723 let q = format!("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b LIMIT {limit}");
5724 let rs = run(&v, &q, ¶ms).unwrap();
5725 let expected_len = (limit as usize).min(full_rows.len());
5726 assert_eq!(
5727 rs.len(),
5728 expected_len,
5729 "LIMIT {limit}: expected {expected_len} rows, got {}",
5730 rs.len()
5731 );
5732 assert_eq!(
5733 rows_of(&rs),
5734 full_rows[..expected_len],
5735 "LIMIT {limit}: rows differ from reference slice"
5736 );
5737 }
5738
5739 let skip_rs = run(
5741 &v,
5742 "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b SKIP 1 LIMIT 2",
5743 ¶ms,
5744 )
5745 .unwrap();
5746 assert_eq!(rows_of(&skip_rs), full_rows[1..3]);
5747
5748 let mut fx2 = Fx::new();
5750 fx2.add("N", "a", vec![("v", i(1))]);
5751 fx2.add("N", "b", vec![("v", i(2))]);
5752 fx2.add("N", "c", vec![("v", i(3))]);
5753 let v2 = fx2.view();
5754 let filter_full = super::execute_unbounded(
5756 &v2,
5757 &compile("MATCH (n:N) WHERE n.v > 1 RETURN n"),
5758 &Params(¶ms),
5759 )
5760 .unwrap();
5761 assert_eq!(filter_full.len(), 2);
5762
5763 let filter_lim = run(&v2, "MATCH (n:N) WHERE n.v > 1 RETURN n LIMIT 1", ¶ms).unwrap();
5764 assert_eq!(filter_lim.len(), 1, "LIMIT 1 on filter query");
5765 assert_eq!(rows_of(&filter_lim), rows_of(&filter_full)[..1]);
5766
5767 let tri = triangle();
5769 let tv = tri.view();
5770 let tri_full = super::execute_unbounded(
5772 &tv,
5773 &compile("MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z"),
5774 &Params(¶ms),
5775 )
5776 .unwrap();
5777 assert_eq!(tri_full.len(), 3, "triangle has 3 unique two-hop paths");
5778 for limit in [1u64, 2, 3, 5] {
5779 let q = format!("MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z LIMIT {limit}");
5780 let rs = run(&tv, &q, ¶ms).unwrap();
5781 let expected_len = (limit as usize).min(3);
5782 assert_eq!(
5783 rs.len(),
5784 expected_len,
5785 "triangle LIMIT {limit}: got {} rows",
5786 rs.len()
5787 );
5788 assert_eq!(
5789 rows_of(&rs),
5790 rows_of(&tri_full)[..expected_len],
5791 "triangle LIMIT {limit}: rows differ"
5792 );
5793 }
5794 }
5795
5796 #[test]
5800 fn expand_terminates_early_with_row_bound() {
5801 const LEAVES: usize = 500;
5802 let mut fx = Fx::new();
5803 let hub = fx.add("Hub", "hub", vec![]);
5804 for i in 0..LEAVES {
5805 let leaf = fx.add("Leaf", &format!("leaf-{i}"), vec![]);
5806 fx.edge("T", hub, leaf, vec![]);
5807 }
5808 let v = fx.view();
5809 let params = BTreeMap::new();
5810 let full_plan = compile("MATCH (h:Hub)-[:T]->(x:Leaf) RETURN x");
5811
5812 let (bounded_result, bounded_produced) = super::with_expand_counter(|| {
5814 run(&v, "MATCH (h:Hub)-[:T]->(x:Leaf) RETURN x LIMIT 5", ¶ms)
5815 });
5816 let bounded_rs = bounded_result.unwrap();
5817 assert_eq!(bounded_rs.len(), 5, "LIMIT 5 must return exactly 5 rows");
5818 assert!(
5819 bounded_produced <= 5,
5820 "bounded: exec_expand emitted {bounded_produced} rows, expected ≤ 5"
5821 );
5822
5823 let (unbounded_result, unbounded_produced) = super::with_expand_counter(|| {
5825 super::execute_unbounded(&v, &full_plan, &Params(¶ms))
5826 });
5827 let unbounded_rs = unbounded_result.unwrap();
5828 assert_eq!(
5829 unbounded_rs.len(),
5830 LEAVES,
5831 "unbounded must return all {LEAVES} rows"
5832 );
5833 assert_eq!(
5834 unbounded_produced, LEAVES,
5835 "unbounded: exec_expand must emit all {LEAVES} rows"
5836 );
5837
5838 assert!(
5840 unbounded_produced >= bounded_produced * 100,
5841 "unbounded ({unbounded_produced}) must be ≥ 100× bounded ({bounded_produced})"
5842 );
5843 }
5844
5845 #[test]
5851 fn bounded_query_survives_low_intermediate_row_cap() {
5852 let mut fx = Fx::new();
5854 let src = fx.add("Src", "s", vec![]);
5855 for i in 0..20usize {
5856 let dst = fx.add("Dst", &format!("d{i}"), vec![]);
5857 fx.edge("T", src, dst, vec![]);
5858 }
5859 let v = fx.view();
5860 let params = BTreeMap::new();
5861 let full_plan = compile("MATCH (s:Src)-[:T]->(d:Dst) RETURN d");
5862
5863 let cap_err = super::with_max_intermediate_rows(10, || {
5865 super::execute_unbounded(&v, &full_plan, &Params(¶ms))
5866 });
5867 assert!(
5868 cap_err.is_err(),
5869 "unbounded must hit the intermediate-row cap"
5870 );
5871 assert!(
5872 cap_err.unwrap_err().contains("intermediate result exceeds"),
5873 "cap error message must be the budget message"
5874 );
5875
5876 let ok = super::with_max_intermediate_rows(10, || {
5878 run(&v, "MATCH (s:Src)-[:T]->(d:Dst) RETURN d LIMIT 5", ¶ms)
5879 });
5880 assert_eq!(
5881 ok.unwrap().len(),
5882 5,
5883 "bounded (LIMIT 5) must complete with 5 rows, not a cap error"
5884 );
5885
5886 let at_cap = super::with_max_intermediate_rows(10, || {
5888 run(&v, "MATCH (s:Src)-[:T]->(d:Dst) RETURN d LIMIT 10", ¶ms)
5889 });
5890 assert_eq!(
5891 at_cap.unwrap().len(),
5892 10,
5893 "bounded at LIMIT==cap must complete with 10 rows"
5894 );
5895 }
5896
5897 #[test]
5902 fn count_star_returns_total_node_count() {
5903 let mut fx = Fx::new();
5904 fx.add("Person", "ada", vec![]);
5905 fx.add("Person", "bob", vec![]);
5906 fx.add("Person", "cam", vec![]);
5907 let v = fx.view();
5908 let params = BTreeMap::new();
5909
5910 let rs = run(&v, "MATCH (n:Person) RETURN COUNT(*)", ¶ms).expect("COUNT(*)");
5911 assert_eq!(rs.columns(), &["COUNT(*)".to_string()]);
5912 assert_eq!(rs.len(), 1);
5913 assert_eq!(rs.row(0), &[Some(i(3))]);
5914
5915 let rs_empty = run(&v, "MATCH (n:Ghost) RETURN COUNT(*)", ¶ms).expect("COUNT(*) empty");
5917 assert_eq!(rs_empty.row(0), &[Some(i(0))]);
5918 }
5919
5920 #[test]
5921 fn count_star_alias_sets_column_name() {
5922 let mut fx = Fx::new();
5923 fx.add("N", "a", vec![]);
5924 let v = fx.view();
5925 let params = BTreeMap::new();
5926 let rs = run(&v, "MATCH (n:N) RETURN COUNT(*) AS total", ¶ms).expect("COUNT AS");
5927 assert_eq!(rs.columns(), &["total".to_string()]);
5928 assert_eq!(rs.row(0), &[Some(i(1))]);
5929 }
5930
5931 #[test]
5932 fn count_var_skips_null_node_bindings() {
5933 let mut fx = Fx::new();
5936 fx.add("N", "a", vec![]);
5937 fx.add("N", "b", vec![]);
5938 let v = fx.view();
5939 let params = BTreeMap::new();
5940 let rs = run(&v, "MATCH (n:N) RETURN COUNT(n)", ¶ms).expect("COUNT(n)");
5941 assert_eq!(rs.columns(), &["COUNT(n)".to_string()]);
5942 assert_eq!(rs.row(0), &[Some(i(2))]);
5943 }
5944
5945 #[test]
5946 fn sum_numeric_prop_ignores_null_and_non_numeric() {
5947 let mut fx = Fx::new();
5948 fx.add("N", "a", vec![("v", i(10))]);
5949 fx.add("N", "b", vec![("v", i(20))]);
5950 fx.add("N", "c", vec![]); let v = fx.view();
5952 let params = BTreeMap::new();
5953 let rs = run(&v, "MATCH (n:N) RETURN SUM(n.v)", ¶ms).expect("SUM");
5954 assert_eq!(rs.columns(), &["SUM(n.v)".to_string()]);
5955 assert_eq!(rs.row(0), &[Some(f(30.0))]);
5957
5958 let rs_null = run(&v, "MATCH (n:N) RETURN SUM(n.missing)", ¶ms).expect("SUM null");
5960 assert_eq!(rs_null.row(0), &[None]);
5961 }
5962
5963 #[test]
5964 fn avg_numeric_prop() {
5965 let mut fx = Fx::new();
5966 fx.add("N", "a", vec![("v", i(10))]);
5967 fx.add("N", "b", vec![("v", i(30))]);
5968 let v = fx.view();
5969 let params = BTreeMap::new();
5970 let rs = run(&v, "MATCH (n:N) RETURN AVG(n.v) AS avg_v", ¶ms).expect("AVG");
5971 assert_eq!(rs.columns(), &["avg_v".to_string()]);
5972 assert_eq!(rs.row(0), &[Some(f(20.0))]);
5974
5975 let rs_empty = run(&v, "MATCH (n:Ghost) RETURN AVG(n.v)", ¶ms).expect("AVG empty");
5977 assert_eq!(rs_empty.row(0), &[None]);
5978 }
5979
5980 #[test]
5981 fn min_max_numeric_prop() {
5982 let mut fx = Fx::new();
5983 fx.add("N", "a", vec![("v", i(5))]);
5984 fx.add("N", "b", vec![("v", i(1))]);
5985 fx.add("N", "c", vec![("v", i(9))]);
5986 fx.add("N", "d", vec![]); let v = fx.view();
5988 let params = BTreeMap::new();
5989
5990 let min_rs = run(&v, "MATCH (n:N) RETURN MIN(n.v)", ¶ms).expect("MIN");
5991 assert_eq!(min_rs.row(0), &[Some(i(1))]);
5992
5993 let max_rs = run(&v, "MATCH (n:N) RETURN MAX(n.v)", ¶ms).expect("MAX");
5994 assert_eq!(max_rs.row(0), &[Some(i(9))]);
5995 }
5996
5997 #[test]
6000 fn min_max_mixed_int_float_props() {
6001 let mut fx = Fx::new();
6002 fx.add("N", "a", vec![("v", i(3))]);
6004 fx.add("N", "b", vec![("v", f(1.5))]);
6005 fx.add("N", "c", vec![("v", i(7))]);
6006 fx.add("N", "d", vec![("v", f(2.0))]);
6007 let v = fx.view();
6008 let params = BTreeMap::new();
6009
6010 let min_rs = run(&v, "MATCH (n:N) RETURN MIN(n.v)", ¶ms).expect("MIN mixed");
6011 assert_eq!(min_rs.row(0), &[Some(f(1.5))]);
6013
6014 let max_rs = run(&v, "MATCH (n:N) RETURN MAX(n.v)", ¶ms).expect("MAX mixed");
6015 assert_eq!(max_rs.row(0), &[Some(i(7))]);
6017 }
6018
6019 #[test]
6022 fn aggregate_limit_skip_order_by_are_no_ops() {
6023 let mut fx = Fx::new();
6024 fx.add("N", "a", vec![]);
6025 fx.add("N", "b", vec![]);
6026 fx.add("N", "c", vec![]);
6027 let v = fx.view();
6028 let params = BTreeMap::new();
6029
6030 let rs_lim5 =
6032 run(&v, "MATCH (n:N) RETURN COUNT(*) LIMIT 5", ¶ms).expect("COUNT(*) LIMIT 5");
6033 assert_eq!(
6034 rs_lim5.len(),
6035 1,
6036 "aggregate with LIMIT 5 must still return 1 row"
6037 );
6038 assert_eq!(rs_lim5.row(0), &[Some(i(3))]);
6039
6040 let rs_lim0 =
6042 run(&v, "MATCH (n:N) RETURN COUNT(*) LIMIT 0", ¶ms).expect("COUNT(*) LIMIT 0");
6043 assert_eq!(
6044 rs_lim0.len(),
6045 1,
6046 "aggregate with LIMIT 0 must still return 1 row"
6047 );
6048 assert_eq!(rs_lim0.row(0), &[Some(i(3))]);
6049
6050 let rs_skip =
6052 run(&v, "MATCH (n:N) RETURN COUNT(*) SKIP 100", ¶ms).expect("COUNT(*) SKIP 100");
6053 assert_eq!(
6054 rs_skip.len(),
6055 1,
6056 "aggregate with large SKIP must still return 1 row"
6057 );
6058
6059 let rs_ord = plan_src("MATCH (n:N) RETURN COUNT(*) ORDER BY n");
6063 assert!(
6066 rs_ord.is_ok(),
6067 "COUNT(*) ORDER BY should plan without error (ORDER BY dropped)"
6068 );
6069 let plan_ops = rs_ord.unwrap();
6070 assert!(
6072 !plan_ops
6073 .iter()
6074 .any(|op| matches!(op, crate::cypher::plan::PlanOp::OrderBy { .. })),
6075 "aggregate plan must not contain OrderBy"
6076 );
6077 }
6078
6079 #[test]
6080 fn count_star_no_budget_cap_applies() {
6081 let mut fx = Fx::new();
6084 let src = fx.add("Src", "s", vec![]);
6085 for i in 0..30usize {
6086 let dst = fx.add("Dst", &format!("d{i}"), vec![]);
6087 fx.edge("T", src, dst, vec![]);
6088 }
6089 let v = fx.view();
6090 let params = BTreeMap::new();
6091
6092 let cap_err =
6094 super::with_max_intermediate_rows(10, || run(&v, "MATCH (n:Dst) RETURN n", ¶ms));
6095 assert!(
6096 cap_err.is_err(),
6097 "staged path must error on 30 nodes with cap=10"
6098 );
6099
6100 let agg_ok = super::with_max_intermediate_rows(10, || {
6102 run(&v, "MATCH (n:Dst) RETURN COUNT(*)", ¶ms)
6103 })
6104 .expect("aggregate must not hit the intermediate-row cap");
6105 assert_eq!(
6106 agg_ok.row(0),
6107 &[Some(i(30))],
6108 "COUNT(*) must count all 30 nodes regardless of cap"
6109 );
6110 }
6111
6112 fn plan_src(src: &str) -> Result<Vec<crate::cypher::plan::PlanOp>, String> {
6113 use crate::cypher::{lex, parse, plan};
6114 let toks = lex(src).map_err(|e| format!("lex: {e}"))?;
6115 let ast = parse(&toks).map_err(|e| format!("parse: {e}"))?;
6116 plan(&ast).map_err(|e| format!("plan: {e}"))
6117 }
6118
6119 #[test]
6122 fn grouped_aggregation_plan_routing() {
6123 use crate::cypher::plan::PlanOp;
6124
6125 let ops = plan_src("MATCH (a:N) RETURN a, COUNT(*)")
6127 .expect("grouped aggregation must now succeed");
6128 assert!(
6129 ops.iter()
6130 .any(|op| matches!(op, PlanOp::GroupAggregate { .. })),
6131 "grouped aggregation plan must contain GroupAggregate op, got: {ops:?}"
6132 );
6133
6134 let ops2 = plan_src("MATCH (a:N) RETURN COUNT(*), COUNT(a)")
6136 .expect("multi-aggregate must now succeed");
6137 assert!(
6138 ops2.iter()
6139 .any(|op| matches!(op, PlanOp::GroupAggregate { .. })),
6140 "multi-aggregate plan must contain GroupAggregate op, got: {ops2:?}"
6141 );
6142
6143 let err3 = plan_src("MATCH (a:N) RETURN SUM(*)").expect_err("SUM(*) must be plan error");
6145 assert!(
6146 err3.to_ascii_lowercase().contains("sum") || err3.to_ascii_lowercase().contains("*"),
6147 "error must mention SUM or *, got: {err3}"
6148 );
6149 }
6150
6151 #[test]
6154 fn grouped_single_key_count() {
6155 let mut fx = Fx::new();
6157 fx.add("N", "a", vec![("t", s("X"))]);
6158 fx.add("N", "b", vec![("t", s("X"))]);
6159 fx.add("N", "c", vec![("t", s("Y"))]);
6160 let v = fx.view();
6161 let params = BTreeMap::new();
6162
6163 let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
6164 .expect("single-key grouped COUNT must succeed");
6165 assert_eq!(
6166 rs.columns(),
6167 &["n.t".to_string(), "cnt".to_string()],
6168 "columns must match RETURN clause"
6169 );
6170 assert_eq!(rs.len(), 2, "must produce exactly 2 groups (X and Y)");
6171
6172 let find = |label: &Value| (0..rs.len()).find(|&i| rs.row(i)[0].as_ref() == Some(label));
6174 let xi = find(&s("X")).expect("group X must exist");
6175 let yi = find(&s("Y")).expect("group Y must exist");
6176 assert_eq!(rs.row(xi)[1], Some(i(2)), "X group count must be 2");
6177 assert_eq!(rs.row(yi)[1], Some(i(1)), "Y group count must be 1");
6178 }
6179
6180 #[test]
6181 fn grouped_two_keys_sum_avg() {
6182 let mut fx = Fx::new();
6184 fx.add(
6185 "N",
6186 "a",
6187 vec![("cat", s("A")), ("sub", s("1")), ("v", i(10))],
6188 );
6189 fx.add(
6190 "N",
6191 "b",
6192 vec![("cat", s("A")), ("sub", s("1")), ("v", i(20))],
6193 );
6194 fx.add(
6195 "N",
6196 "c",
6197 vec![("cat", s("A")), ("sub", s("2")), ("v", i(5))],
6198 );
6199 fx.add(
6200 "N",
6201 "d",
6202 vec![("cat", s("B")), ("sub", s("1")), ("v", i(100))],
6203 );
6204 let v = fx.view();
6205 let params = BTreeMap::new();
6206
6207 let rs = run(
6208 &v,
6209 "MATCH (n:N) RETURN n.cat, n.sub, SUM(n.v) AS total, AVG(n.v) AS avg_v",
6210 ¶ms,
6211 )
6212 .expect("two-key SUM + AVG must succeed");
6213 assert_eq!(
6214 rs.columns(),
6215 &[
6216 "n.cat".to_string(),
6217 "n.sub".to_string(),
6218 "total".to_string(),
6219 "avg_v".to_string()
6220 ]
6221 );
6222 assert_eq!(rs.len(), 3, "must produce 3 groups: (A,1), (A,2), (B,1)");
6223
6224 let find = |cat: &Value, sub: &Value| {
6225 (0..rs.len())
6226 .find(|&i| rs.row(i)[0].as_ref() == Some(cat) && rs.row(i)[1].as_ref() == Some(sub))
6227 };
6228 let a1 = find(&s("A"), &s("1")).expect("group (A,1) must exist");
6229 assert_eq!(rs.row(a1)[2], Some(f(30.0)), "(A,1) SUM must be 30.0");
6230 assert_eq!(rs.row(a1)[3], Some(f(15.0)), "(A,1) AVG must be 15.0");
6231
6232 let a2 = find(&s("A"), &s("2")).expect("group (A,2) must exist");
6233 assert_eq!(rs.row(a2)[2], Some(f(5.0)), "(A,2) SUM must be 5.0");
6234
6235 let b1 = find(&s("B"), &s("1")).expect("group (B,1) must exist");
6236 assert_eq!(rs.row(b1)[2], Some(f(100.0)), "(B,1) SUM must be 100.0");
6237 assert_eq!(rs.row(b1)[3], Some(f(100.0)), "(B,1) AVG must be 100.0");
6238 }
6239
6240 #[test]
6241 fn grouped_order_by_count_desc_limit() {
6242 let mut fx = Fx::new();
6244 fx.add("N", "a1", vec![("cat", s("A"))]);
6245 fx.add("N", "a2", vec![("cat", s("A"))]);
6246 fx.add("N", "a3", vec![("cat", s("A"))]);
6247 fx.add("N", "b1", vec![("cat", s("B"))]);
6248 fx.add("N", "b2", vec![("cat", s("B"))]);
6249 fx.add("N", "c1", vec![("cat", s("C"))]);
6250 fx.add("N", "d1", vec![("cat", s("D"))]);
6251 fx.add("N", "d2", vec![("cat", s("D"))]);
6252 fx.add("N", "d3", vec![("cat", s("D"))]);
6253 fx.add("N", "d4", vec![("cat", s("D"))]);
6254 fx.add("N", "e1", vec![("cat", s("E"))]);
6255 let v = fx.view();
6256 let params = BTreeMap::new();
6257
6258 let rs = run(
6259 &v,
6260 "MATCH (n:N) RETURN n.cat, COUNT(*) AS cnt ORDER BY cnt DESC LIMIT 3",
6261 ¶ms,
6262 )
6263 .expect("ORDER BY count DESC LIMIT 3 must succeed");
6264 assert_eq!(rs.len(), 3, "LIMIT 3 must return exactly 3 groups");
6265 assert_eq!(rs.row(0)[1], Some(i(4)), "row 0 must be count 4");
6267 assert_eq!(rs.row(0)[0], Some(s("D")), "row 0 must be category D");
6268 assert_eq!(rs.row(1)[1], Some(i(3)), "row 1 must be count 3");
6269 assert_eq!(rs.row(1)[0], Some(s("A")), "row 1 must be category A");
6270 assert_eq!(rs.row(2)[1], Some(i(2)), "row 2 must be count 2");
6271 assert_eq!(rs.row(2)[0], Some(s("B")), "row 2 must be category B");
6272
6273 let plan_ops =
6275 plan_src("MATCH (n:N) RETURN n.cat, COUNT(*) AS cnt ORDER BY cnt DESC LIMIT 3")
6276 .expect("plan must succeed");
6277 assert_eq!(
6278 crate::cypher::plan::row_bound(&plan_ops),
6279 None,
6280 "GroupAggregate plan with LIMIT must have row_bound = None"
6281 );
6282 }
6283
6284 #[test]
6285 fn grouped_empty_input_yields_zero_groups() {
6286 let fx = Fx::new(); let v = fx.view();
6288 let params = BTreeMap::new();
6289
6290 let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
6291 .expect("grouped aggregate on empty graph must succeed");
6292 assert_eq!(rs.len(), 0, "empty input must yield zero groups");
6293 assert_eq!(
6294 rs.columns(),
6295 &["n.t".to_string(), "cnt".to_string()],
6296 "columns must still be present even with zero rows"
6297 );
6298 }
6299
6300 #[test]
6301 fn grouped_null_key_groups_together() {
6302 let mut fx = Fx::new();
6304 fx.add("N", "a", vec![("t", s("X"))]);
6305 fx.add("N", "b", vec![]); fx.add("N", "c", vec![]); fx.add("N", "d", vec![("t", s("Y"))]);
6308 let v = fx.view();
6309 let params = BTreeMap::new();
6310
6311 let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
6312 .expect("null-key grouped aggregate must succeed");
6313 assert_eq!(rs.len(), 3, "must produce 3 groups: X, null, Y");
6314
6315 let null_row = (0..rs.len())
6317 .find(|&i| rs.row(i)[0].is_none())
6318 .expect("null group must be present");
6319 assert_eq!(
6320 rs.row(null_row)[1],
6321 Some(i(2)),
6322 "null group must count 2 rows (b and c)"
6323 );
6324
6325 let x_row = (0..rs.len())
6326 .find(|&i| rs.row(i)[0] == Some(s("X")))
6327 .expect("X group must exist");
6328 assert_eq!(rs.row(x_row)[1], Some(i(1)));
6329
6330 let y_row = (0..rs.len())
6331 .find(|&i| rs.row(i)[0] == Some(s("Y")))
6332 .expect("Y group must exist");
6333 assert_eq!(rs.row(y_row)[1], Some(i(1)));
6334 }
6335
6336 #[test]
6337 fn grouped_cap_error_on_high_cardinality() {
6338 let mut fx = Fx::new();
6341 fx.add("N", "a", vec![("t", s("A"))]);
6342 fx.add("N", "b", vec![("t", s("B"))]);
6343 fx.add("N", "c", vec![("t", s("C"))]);
6344 let v = fx.view();
6345 let params = BTreeMap::new();
6346
6347 let err = super::with_max_groups(2, || {
6348 run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
6349 })
6350 .expect_err("must error when group count exceeds cap");
6351 assert!(
6352 err.to_ascii_lowercase().contains("group count"),
6353 "error must mention group count, got: {err}"
6354 );
6355 }
6356
6357 #[test]
6358 fn multi_aggregate_no_keys() {
6359 let mut fx = Fx::new();
6362 fx.add("N", "a", vec![]);
6363 fx.add("N", "b", vec![]);
6364 let v = fx.view();
6365 let params = BTreeMap::new();
6366
6367 let rs = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
6368 .expect("multi-aggregate no keys must succeed");
6369 assert_eq!(rs.len(), 1, "must produce exactly one result row");
6370 assert_eq!(rs.row(0)[0], Some(i(2)), "COUNT(*) must be 2");
6371 assert_eq!(rs.row(0)[1], Some(i(2)), "COUNT(n) must be 2");
6372 }
6373
6374 #[test]
6375 fn aggregate_over_hop_counts_edges() {
6376 let fx = hop_graph();
6377 let v = fx.view();
6378 let params = BTreeMap::new();
6379 let rs = run(
6381 &v,
6382 "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN COUNT(*)",
6383 ¶ms,
6384 )
6385 .expect("COUNT(*) on hop graph");
6386 assert_eq!(rs.row(0), &[Some(i(3))]);
6387 }
6388
6389 #[test]
6394 fn multi_aggregate_no_keys_empty_graph() {
6395 let fx = Fx::new(); let v = fx.view();
6401 let params = BTreeMap::new();
6402
6403 let rs = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
6404 .expect("empty-graph multi-agg must succeed");
6405 assert_eq!(rs.len(), 1, "must produce exactly 1 row on empty input");
6406 assert_eq!(
6407 rs.row(0)[0],
6408 Some(i(0)),
6409 "COUNT(*) on empty graph must be 0"
6410 );
6411 assert_eq!(
6412 rs.row(0)[1],
6413 Some(i(0)),
6414 "COUNT(n) on empty graph must be 0"
6415 );
6416 }
6417
6418 #[test]
6419 fn fast_path_and_group_path_agree_on_empty_input() {
6420 let fx = Fx::new();
6426 let v = fx.view();
6427 let params = BTreeMap::new();
6428
6429 let fast = run(&v, "MATCH (n:N) RETURN COUNT(*)", ¶ms)
6430 .expect("fast-path COUNT on empty graph");
6431 assert_eq!(fast.len(), 1, "fast path: 1 row on empty input");
6432 let fast_count = fast.row(0)[0].clone();
6433
6434 let grouped = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
6435 .expect("group-path COUNT on empty graph");
6436 assert_eq!(grouped.len(), 1, "group path: 1 row on empty input");
6437 let group_count = grouped.row(0)[0].clone();
6438
6439 assert_eq!(
6440 fast_count, group_count,
6441 "fast path and group path must agree on COUNT(*) for empty input"
6442 );
6443
6444 let mut fx2 = Fx::new();
6446 fx2.add("N", "x", vec![]);
6447 fx2.add("N", "y", vec![]);
6448 let v2 = fx2.view();
6449
6450 let fast2 = run(&v2, "MATCH (n:N) RETURN COUNT(*)", ¶ms)
6451 .expect("fast-path COUNT on 2-node graph");
6452 assert_eq!(fast2.row(0)[0], Some(i(2)), "fast path: COUNT(*) = 2");
6453
6454 let grouped2 = run(&v2, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
6455 .expect("group-path COUNT on 2-node graph");
6456 assert_eq!(grouped2.row(0)[0], Some(i(2)), "group path: COUNT(*) = 2");
6457 assert_eq!(grouped2.row(0)[1], Some(i(2)), "group path: COUNT(n) = 2");
6458 }
6459
6460 #[test]
6465 fn grouped_int_float_key_unification() {
6466 let mut fx = Fx::new();
6473 fx.add("N", "a", vec![("score", i(1))]); fx.add("N", "b", vec![("score", f(1.0))]);
6475 let v = fx.view();
6476 let params = BTreeMap::new();
6477
6478 let rs = run(&v, "MATCH (n:N) RETURN n.score, COUNT(*) AS cnt", ¶ms)
6479 .expect("int/float unification must succeed");
6480 assert_eq!(
6481 rs.len(),
6482 1,
6483 "Int(1) and Float(1.0) must group together into 1 group"
6484 );
6485 assert_eq!(
6487 rs.row(0)[0],
6488 Some(i(1)),
6489 "key column must display Int(1) (first-seen)"
6490 );
6491 assert_eq!(rs.row(0)[1], Some(i(2)), "unified group must have count=2");
6492 }
6493
6494 #[test]
6495 fn distinct_collapses_duplicate_projected_values() {
6496 let mut fx = Fx::new();
6497 fx.add("N", "a1", vec![("city", s("Austin"))]);
6498 fx.add("N", "a2", vec![("city", s("Austin"))]);
6499 let v = fx.view();
6500 let params = BTreeMap::new();
6501 let rs = run(&v, "MATCH (n:N) RETURN DISTINCT n.city", ¶ms).unwrap();
6502 assert_eq!(rs.len(), 1);
6503 assert_eq!(rs.row(0)[0], Some(s("Austin")));
6504 }
6505
6506 #[test]
6507 fn distinct_int_float_unify() {
6508 let mut fx = Fx::new();
6509 fx.add("N", "a", vec![("score", i(1))]);
6510 fx.add("N", "b", vec![("score", f(1.0))]);
6511 let v = fx.view();
6512 let params = BTreeMap::new();
6513 let rs = run(&v, "MATCH (n:N) RETURN DISTINCT n.score", ¶ms).unwrap();
6514 assert_eq!(
6515 rs.len(),
6516 1,
6517 "Int(1) and Float(1.0) must DISTINCT as one row"
6518 );
6519 assert_eq!(rs.row(0)[0], Some(i(1)), "first-seen Int(1) wins display");
6520 }
6521
6522 #[test]
6523 fn distinct_caps_at_intermediate_row_budget() {
6524 let mut fx = Fx::new();
6525 fx.add("N", "a", vec![("city", s("Austin"))]);
6526 fx.add("N", "b", vec![("city", s("Paris"))]);
6527 let v = fx.view();
6528 let params = BTreeMap::new();
6529 let err = super::with_max_intermediate_rows(1, || {
6530 run(&v, "MATCH (n:N) RETURN DISTINCT n.city", ¶ms)
6531 })
6532 .expect_err("two distinct cities must exceed cap=1");
6533 assert!(
6534 err.contains("1") || err.contains("row"),
6535 "cap error must mention the budget, got: {err}"
6536 );
6537 }
6538
6539 #[test]
6540 fn where_in_list_filters_rows() {
6541 let mut fx = Fx::new();
6542 fx.add("N", "a", vec![("city", s("Austin"))]);
6543 fx.add("N", "p", vec![("city", s("Paris"))]);
6544 fx.add("N", "l", vec![("city", s("London"))]);
6545 let v = fx.view();
6546 let mut params = BTreeMap::new();
6547 params.insert("c".into(), s("Paris"));
6548 let rs = run(
6549 &v,
6550 "MATCH (n:N) WHERE n.city IN ['Austin', $c] RETURN n.city",
6551 ¶ms,
6552 )
6553 .unwrap();
6554 assert_eq!(rs.len(), 2);
6555 }
6556
6557 #[test]
6558 fn pure_int_keys_display_as_int() {
6559 let mut fx = Fx::new();
6564 fx.add("N", "a", vec![("age", i(10))]);
6565 fx.add("N", "b", vec![("age", i(20))]);
6566 fx.add("N", "c", vec![("age", i(10))]);
6567 let v = fx.view();
6568 let params = BTreeMap::new();
6569
6570 let rs = run(
6571 &v,
6572 "MATCH (n:N) RETURN n.age, COUNT(*) AS cnt ORDER BY n.age",
6573 ¶ms,
6574 )
6575 .expect("pure-Int group keys must succeed");
6576 assert_eq!(rs.len(), 2, "must have 2 groups: age=10 and age=20");
6577 assert_eq!(
6579 rs.row(0)[0],
6580 Some(i(10)),
6581 "age=10 key column must display as Int(10)"
6582 );
6583 assert_eq!(rs.row(0)[1], Some(i(2)), "age=10 group has 2 nodes");
6584 assert_eq!(
6585 rs.row(1)[0],
6586 Some(i(20)),
6587 "age=20 key column must display as Int(20)"
6588 );
6589 assert_eq!(rs.row(1)[1], Some(i(1)), "age=20 group has 1 node");
6590 }
6591
6592 #[test]
6597 fn var_expand_group_aggregate_staged_path() {
6598 let mut fx = Fx::new();
6608 let a = fx.add("N", "a", vec![("key", s("a"))]);
6609 let b = fx.add("N", "b", vec![("key", s("b"))]);
6610 let c = fx.add("N", "c", vec![("key", s("c"))]);
6611 fx.edge("T", a, b, vec![]);
6612 fx.edge("T", b, c, vec![]);
6613 let v = fx.view();
6614 let params = BTreeMap::new();
6615
6616 let rs = run(
6617 &v,
6618 "MATCH (x:N)-[*1..2]->(y:N) RETURN y.key, COUNT(*) AS cnt ORDER BY y.key",
6619 ¶ms,
6620 )
6621 .expect("VarExpand + GroupAggregate must succeed");
6622
6623 assert_eq!(rs.len(), 2, "must have 2 destination groups: b and c");
6624 assert_eq!(rs.row(0)[0], Some(s("b")), "first group key must be 'b'");
6625 assert_eq!(rs.row(0)[1], Some(i(1)), "b is reached via 1 path");
6626 assert_eq!(rs.row(1)[0], Some(s("c")), "second group key must be 'c'");
6627 assert_eq!(
6628 rs.row(1)[1],
6629 Some(i(2)),
6630 "c is reached via 2 paths (1-hop and 2-hop)"
6631 );
6632 }
6633
6634 #[test]
6642 fn pull_rows_var_expand_arm_returns_named_err() {
6643 let fx = Fx::new();
6644 let view = fx.view();
6645 let vars = super::VarTable {
6646 names: vec!["a".into(), "b".into()],
6647 };
6648 let project_items: Vec<crate::cypher::ast::RetItem> = vec![];
6649 let empty_params = BTreeMap::new();
6650 let params = super::Params(&empty_params);
6651 let ctx = super::PullCtx {
6652 view: &view,
6653 vars: &vars,
6654 project_items: &project_items,
6655 params: ¶ms,
6656 bound: 100,
6657 };
6658 let ops = vec![PlanOp::VarExpand {
6659 from: "a".into(),
6660 rel_var: None,
6661 etypes: vec![],
6662 dir: crate::cypher::RelDir::Right,
6663 to: "b".into(),
6664 min: 1,
6665 max: 3,
6666 }];
6667 let mut row = vec![None; vars.names.len()];
6668 let mut result = Vec::new();
6669 let err = super::pull_rows(&ctx, &ops, &mut row, &mut result)
6670 .expect_err("VarExpand must Err in pull_rows");
6671 assert!(
6672 err.contains("VarExpand") && err.contains("pull executor"),
6673 "error must name VarExpand and pull executor, got: {err}"
6674 );
6675 }
6676
6677 #[test]
6678 fn pull_rows_shortest_path_arm_returns_named_err() {
6679 let fx = Fx::new();
6680 let view = fx.view();
6681 let vars = super::VarTable {
6682 names: vec!["a".into(), "b".into()],
6683 };
6684 let project_items: Vec<crate::cypher::ast::RetItem> = vec![];
6685 let empty_params = BTreeMap::new();
6686 let params = super::Params(&empty_params);
6687 let ctx = super::PullCtx {
6688 view: &view,
6689 vars: &vars,
6690 project_items: &project_items,
6691 params: ¶ms,
6692 bound: 100,
6693 };
6694 let ops = vec![PlanOp::ShortestPath {
6695 from: "a".into(),
6696 rel_var: None,
6697 etypes: vec![],
6698 dir: crate::cypher::RelDir::Right,
6699 to: "b".into(),
6700 max_hops: 5,
6701 }];
6702 let mut row = vec![None; vars.names.len()];
6703 let mut result = Vec::new();
6704 let err = super::pull_rows(&ctx, &ops, &mut row, &mut result)
6705 .expect_err("ShortestPath must Err in pull_rows");
6706 assert!(
6707 err.contains("ShortestPath") && err.contains("pull executor"),
6708 "error must name ShortestPath and pull executor, got: {err}"
6709 );
6710 }
6711
6712 fn city_graph() -> Fx {
6716 let mut fx = Fx::new();
6717 fx.add(
6718 "Person",
6719 "alice",
6720 vec![("city", s("Boston")), ("age", i(30))],
6721 );
6722 fx.add("Person", "bob", vec![("city", s("Boston")), ("age", i(25))]);
6723 fx.add(
6724 "Person",
6725 "carol",
6726 vec![("city", s("Austin")), ("age", i(35))],
6727 );
6728 fx.add(
6729 "Person",
6730 "dave",
6731 vec![("city", s("Austin")), ("age", i(28))],
6732 );
6733 fx.add("Person", "eve", vec![("city", s("Boston")), ("age", i(22))]);
6734 fx
6735 }
6736
6737 #[test]
6740 fn with_aggregate_having_filters_groups() {
6741 let fx = city_graph();
6742 let view = fx.view();
6743 let params = BTreeMap::new();
6744 let rs = run(
6745 &view,
6746 "MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt WHERE cnt > 2 RETURN city, cnt",
6747 ¶ms,
6748 )
6749 .expect("WITH HAVING query must succeed");
6750 assert_eq!(rs.len(), 1, "only Boston group has cnt > 2");
6751 assert_eq!(rs.get(0, "city"), Some(&s("Boston")));
6752 assert_eq!(rs.get(0, "cnt"), Some(&i(3)));
6753 }
6754
6755 #[test]
6757 fn with_order_limit_then_return() {
6758 let fx = city_graph();
6759 let view = fx.view();
6760 let params = BTreeMap::new();
6761 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)
6763 .expect("WITH ORDER LIMIT must succeed");
6764 assert_eq!(rs.len(), 2, "LIMIT 2");
6765 let ages: Vec<Option<Value>> = (0..rs.len()).map(|i| rs.get(i, "age").cloned()).collect();
6767 assert_eq!(ages[0], Some(i(35)), "oldest person first");
6768 assert_eq!(ages[1], Some(i(30)), "second oldest");
6769 }
6770
6771 #[test]
6773 fn chained_with_stages() {
6774 let fx = city_graph();
6775 let view = fx.view();
6776 let params = BTreeMap::new();
6777 let rs = run(&view,
6779 "MATCH (p:Person) WITH p.city AS city, p.age AS age WITH city, age WHERE age > 25 RETURN city, age",
6780 ¶ms).expect("chained WITH must succeed");
6781 assert_eq!(rs.len(), 3, "3 people with age > 25");
6783 }
6784
6785 #[test]
6787 fn with_then_match_reentry() {
6788 let mut fx = Fx::new();
6789 let alice = fx.add("Person", "alice", vec![]);
6790 let bob = fx.add("Person", "bob", vec![]);
6791 let corp = fx.add("Company", "acme", vec![]);
6792 fx.edge("WORKS_AT", alice, corp, vec![("years", i(5))]);
6793 fx.edge("WORKS_AT", bob, corp, vec![("years", i(3))]);
6794 let view = fx.view();
6795 let params = BTreeMap::new();
6796 let rs = run(&view,
6798 "MATCH (p:Person)-[r:WORKS_AT]->(c:Company) WITH p, c MATCH (c)-[r2:WORKS_AT]-(colleague:Person) RETURN p, colleague",
6799 ¶ms).expect("WITH MATCH re-entry must succeed");
6800 assert!(rs.len() >= 2, "cross join via company: got {}", rs.len());
6802 }
6803
6804 #[test]
6806 fn unwind_literal_list_produces_rows() {
6807 let fx = city_graph();
6808 let view = fx.view();
6809 let params = BTreeMap::new();
6810 let rs = run(
6811 &view,
6812 "MATCH (p:Person) WHERE p.city = 'Boston' UNWIND [1, 2, 3] AS x RETURN p, x",
6813 ¶ms,
6814 )
6815 .expect("UNWIND literal must succeed");
6816 assert_eq!(rs.len(), 9, "UNWIND [1,2,3] × 3 Boston people = 9 rows");
6818 let xs: Vec<Option<Value>> = (0..rs.len())
6820 .map(|row_i| rs.get(row_i, "x").cloned())
6821 .collect();
6822 assert!(xs.contains(&Some(i(1))));
6823 assert!(xs.contains(&Some(i(2))));
6824 assert!(xs.contains(&Some(i(3))));
6825 }
6826
6827 #[test]
6829 fn unwind_list_property() {
6830 let mut fx = Fx::new();
6831 fx.add(
6832 "Tag",
6833 "post1",
6834 vec![("tags", Value::List(vec![s("rust"), s("graph")]))],
6835 );
6836 fx.add("Tag", "post2", vec![("tags", Value::List(vec![s("db")]))]);
6837 let view = fx.view();
6838 let params = BTreeMap::new();
6839 let rs = run(
6840 &view,
6841 "MATCH (p:Tag) UNWIND p.tags AS tag RETURN p, tag",
6842 ¶ms,
6843 )
6844 .expect("UNWIND property must succeed");
6845 assert_eq!(rs.len(), 3, "2+1 tag elements");
6846 }
6847
6848 #[test]
6850 fn unwind_empty_list_yields_zero_rows() {
6851 let fx = city_graph();
6852 let view = fx.view();
6853 let params = BTreeMap::new();
6854 let rs = run(
6855 &view,
6856 "MATCH (p:Person) WHERE p.city = 'Boston' UNWIND [] AS x RETURN x",
6857 ¶ms,
6858 )
6859 .expect("UNWIND [] must succeed with 0 rows");
6860 assert_eq!(rs.len(), 0, "UNWIND [] should produce 0 rows");
6861 }
6862
6863 #[test]
6865 fn unwind_non_list_is_named_error() {
6866 let fx = city_graph();
6867 let view = fx.view();
6868 let params = BTreeMap::new();
6869 let err = run(
6871 &view,
6872 "MATCH (p:Person) WHERE p.city = 'Boston' UNWIND p.city AS x RETURN x",
6873 ¶ms,
6874 )
6875 .expect_err("UNWIND non-list must error");
6876 assert!(
6877 err.contains("UNWIND") && err.contains("list"),
6878 "error must mention UNWIND and list: {err}"
6879 );
6880 }
6881
6882 #[test]
6884 fn unwind_cross_product_trips_budget() {
6885 let mut fx = Fx::new();
6886 for i in 0..10 {
6888 fx.add("N", &format!("n{i}"), vec![]);
6889 }
6890 let view = fx.view();
6891 let params = BTreeMap::new();
6892 let err = super::with_max_intermediate_rows(5, || {
6893 run(
6894 &view,
6895 "MATCH (n:N) UNWIND [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] AS x RETURN n, x",
6896 ¶ms,
6897 )
6898 })
6899 .expect_err("must hit budget");
6900 assert!(
6901 err.contains("intermediate result exceeds"),
6902 "error must mention intermediate result limit: {err}"
6903 );
6904 }
6905
6906 #[test]
6909 fn routing_with_plan_uses_staged_path() {
6910 use crate::cypher::plan::row_bound;
6911 let ops_with = compile("MATCH (a) WITH a RETURN a");
6912 assert_eq!(
6913 row_bound(&ops_with),
6914 None,
6915 "WITH plan must have row_bound=None"
6916 );
6917
6918 let ops_unwind = compile("MATCH (a) UNWIND [1, 2] AS x RETURN a");
6919 assert_eq!(
6920 row_bound(&ops_unwind),
6921 None,
6922 "UNWIND plan must have row_bound=None"
6923 );
6924
6925 let ops_plain = compile("MATCH (a) RETURN a LIMIT 5");
6927 assert!(
6928 row_bound(&ops_plain).is_some(),
6929 "plain LIMIT plan must have a row bound"
6930 );
6931 }
6932
6933 #[test]
6936 fn unwind_null_property_yields_zero_rows_for_that_node() {
6937 let mut fx = Fx::new();
6938 fx.add(
6940 "Post",
6941 "post1",
6942 vec![("tags", Value::List(vec![s("rust"), s("graph")]))],
6943 );
6944 fx.add("Post", "post2", vec![("other", Value::Str("hello".into()))]);
6945 let view = fx.view();
6946 let params = BTreeMap::new();
6947 let rs = run(
6948 &view,
6949 "MATCH (p:Post) UNWIND p.tags AS tag RETURN tag",
6950 ¶ms,
6951 )
6952 .expect("UNWIND null property must succeed (not error)");
6953 assert_eq!(
6955 rs.len(),
6956 2,
6957 "null property must yield 0 rows; list property yields N rows"
6958 );
6959 let tag0 = rs.get(0, "tag");
6960 let tag1 = rs.get(1, "tag");
6961 let tags = [tag0, tag1];
6962 assert!(tags.contains(&Some(&s("rust"))), "rust tag present");
6963 assert!(tags.contains(&Some(&s("graph"))), "graph tag present");
6964 }
6965
6966 #[test]
6968 fn where_before_unwind_filters_nodes() {
6969 let mut fx = Fx::new();
6970 fx.add(
6971 "P",
6972 "a",
6973 vec![
6974 ("group", Value::Str("keep".into())),
6975 ("items", Value::List(vec![Value::Int(1), Value::Int(2)])),
6976 ],
6977 );
6978 fx.add(
6979 "P",
6980 "b",
6981 vec![
6982 ("group", Value::Str("drop".into())),
6983 ("items", Value::List(vec![Value::Int(3), Value::Int(4)])),
6984 ],
6985 );
6986 let view = fx.view();
6987 let params = BTreeMap::new();
6988 let rs = run(
6990 &view,
6991 "MATCH (n:P) WHERE n.group = 'keep' UNWIND n.items AS x RETURN x",
6992 ¶ms,
6993 )
6994 .expect("WHERE before UNWIND must succeed");
6995 assert_eq!(rs.len(), 2, "only matching node expands; 2 items");
6996 }
6997
6998 #[test]
7000 fn where_after_unwind_filters_expanded_rows() {
7001 let mut fx = Fx::new();
7002 fx.add(
7003 "N",
7004 "n1",
7005 vec![(
7006 "vals",
7007 Value::List(vec![Value::Int(1), Value::Int(3), Value::Int(5)]),
7008 )],
7009 );
7010 let view = fx.view();
7011 let params = BTreeMap::new();
7012 let rs = run(
7014 &view,
7015 "MATCH (n:N) UNWIND n.vals AS x WHERE x > 2 RETURN x",
7016 ¶ms,
7017 )
7018 .expect("WHERE after UNWIND must succeed");
7019 assert_eq!(rs.len(), 2, "x=3 and x=5 pass; x=1 filtered out");
7020 let v0 = rs.get(0, "x").cloned();
7021 let v1 = rs.get(1, "x").cloned();
7022 let vals = [v0, v1];
7023 assert!(vals.contains(&Some(Value::Int(3))), "x=3 present");
7024 assert!(vals.contains(&Some(Value::Int(5))), "x=5 present");
7025 }
7026
7027 #[test]
7029 fn aggregate_with_order_by_unknown_alias_is_named_error() {
7030 use crate::cypher::{lex, parser::parse, plan::plan};
7031 let src = "MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt ORDER BY cntt DESC RETURN city, cnt";
7033 let plan_result = plan(&parse(&lex(src).expect("lex")).expect("parse"));
7034 let err = plan_result
7035 .expect_err("typo'd ORDER BY alias in aggregate WITH must be a named plan error");
7036 assert!(
7037 err.contains("cntt") || err.contains("unbound"),
7038 "error must mention the unknown alias: {err}"
7039 );
7040 }
7041
7042 #[test]
7044 fn non_aggregate_with_order_by_unknown_alias_is_named_error() {
7045 use crate::cypher::{lex, parser::parse, plan::plan};
7046 let src = "MATCH (p:Person) WITH p, p.age AS age ORDER BY nope DESC RETURN p.city AS city";
7047 let plan_result = plan(&parse(&lex(src).expect("lex")).expect("parse"));
7048 let err = plan_result
7049 .expect_err("typo'd ORDER BY alias in non-aggregate WITH must be a named plan error");
7050 assert!(
7051 err.contains("nope") || err.contains("unbound"),
7052 "error must mention the unknown alias: {err}"
7053 );
7054 }
7055
7056 #[test]
7060 fn post_unwind_where_references_pre_match_variable() {
7061 let mut fx = Fx::new();
7062 fx.add(
7064 "N",
7065 "n1",
7066 vec![
7067 ("threshold", Value::Int(3)),
7068 (
7069 "xs",
7070 Value::List(vec![
7071 Value::Int(1),
7072 Value::Int(2),
7073 Value::Int(3),
7074 Value::Int(4),
7075 Value::Int(5),
7076 ]),
7077 ),
7078 ],
7079 );
7080 fx.add(
7082 "N",
7083 "n2",
7084 vec![
7085 ("threshold", Value::Int(10)),
7086 ("xs", Value::List(vec![Value::Int(1), Value::Int(2)])),
7087 ],
7088 );
7089 let view = fx.view();
7090 let params = BTreeMap::new();
7091 let rs = run(
7092 &view,
7093 "MATCH (n:N) UNWIND n.xs AS x WHERE x > n.threshold RETURN x",
7094 ¶ms,
7095 )
7096 .expect("post-UNWIND WHERE referencing MATCH variable must succeed");
7097 assert_eq!(rs.len(), 2, "exactly 2 rows: x=4 and x=5 from n1");
7099 let v0 = rs.get(0, "x").cloned();
7100 let v1 = rs.get(1, "x").cloned();
7101 let got = [v0, v1];
7102 assert!(got.contains(&Some(Value::Int(4))), "x=4 present");
7103 assert!(got.contains(&Some(Value::Int(5))), "x=5 present");
7104 }
7105
7106 #[test]
7110 fn with_stage_aggregate_order_by_descending() {
7111 let fx = city_graph();
7114 let view = fx.view();
7115 let params = BTreeMap::new();
7116 let rs = run(
7117 &view,
7118 "MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt ORDER BY cnt DESC RETURN city, cnt",
7119 ¶ms,
7120 )
7121 .expect("aggregate WITH ORDER BY must succeed");
7122 assert_eq!(rs.len(), 2, "two city groups");
7123 assert_eq!(
7125 rs.get(0, "cnt"),
7126 Some(&Value::Int(3)),
7127 "first row must be the group with cnt=3 (Boston)"
7128 );
7129 assert_eq!(
7130 rs.get(1, "cnt"),
7131 Some(&Value::Int(2)),
7132 "second row must be the group with cnt=2 (Austin)"
7133 );
7134 }
7135
7136 #[test]
7140 fn unwind_then_with_composition() {
7141 let mut fx = Fx::new();
7142 fx.add(
7143 "Doc",
7144 "d1",
7145 vec![(
7146 "scores",
7147 Value::List(vec![Value::Int(10), Value::Int(20), Value::Int(30)]),
7148 )],
7149 );
7150 fx.add(
7151 "Doc",
7152 "d2",
7153 vec![("scores", Value::List(vec![Value::Int(5), Value::Int(15)]))],
7154 );
7155 let view = fx.view();
7156 let params = BTreeMap::new();
7157
7158 let rs = run(
7160 &view,
7161 "MATCH (n:Doc) UNWIND n.scores AS x WITH x RETURN x",
7162 ¶ms,
7163 )
7164 .expect("UNWIND then WITH pass-through must succeed");
7165 assert_eq!(rs.len(), 5, "3 + 2 = 5 expanded rows carried through WITH");
7166
7167 let rs2 = run(
7169 &view,
7170 "MATCH (n:Doc) UNWIND n.scores AS x WITH COUNT(*) AS total RETURN total",
7171 ¶ms,
7172 )
7173 .expect("UNWIND then aggregate WITH must succeed");
7174 assert_eq!(rs2.len(), 1, "single aggregate row");
7175 assert_eq!(
7176 rs2.get(0, "total"),
7177 Some(&Value::Int(5)),
7178 "total must be 5 (3+2 expanded rows)"
7179 );
7180 }
7181
7182 #[test]
7188 fn binarith_div_min_over_neg1_does_not_panic() {
7189 let fx = Fx::new();
7190 let view = fx.view();
7191 let vars = VarTable { names: vec![] };
7192 let row: Row = vec![];
7193 let params = BTreeMap::new();
7194
7195 let operand = Operand::BinArith {
7197 op: ArithOp::Div,
7198 left: Box::new(Operand::Lit(Value::Int(i64::MIN))),
7199 right: Box::new(Operand::Lit(Value::Int(-1))),
7200 };
7201
7202 let result = resolve_operand(&view, &vars, &row, &operand, &Params(¶ms));
7203 assert!(
7204 result.is_ok(),
7205 "overflow division must not return Err: {result:?}"
7206 );
7207 assert_eq!(
7208 result.unwrap(),
7209 Some(Value::Int(i64::MAX)),
7210 "i64::MIN / -1 must saturate to i64::MAX, not panic"
7211 );
7212 }
7213
7214 #[test]
7219 fn binarith_missing_param_caught_at_preflight() {
7220 let fx = Fx::new();
7221 let view = fx.view();
7222 let err = run(
7227 &view,
7228 "MATCH (n:X) RETURN abs($missing - 1)",
7229 &BTreeMap::new(),
7230 )
7231 .expect_err("missing param inside BinArith must be caught at preflight");
7232 assert!(
7233 err.contains("missing"),
7234 "error must name the param 'missing', got: {err}"
7235 );
7236 }
7237
7238 #[test]
7242 fn is_null_filters_absent_prop() {
7243 let mut fx = Fx::new();
7244 fx.add("Person", "alice", vec![("age", Value::Int(30))]);
7245 fx.add("Person", "bob", vec![]); let view = fx.view();
7247 let rs = run(
7248 &view,
7249 "MATCH (n:Person) WHERE n.age IS NULL RETURN n",
7250 &BTreeMap::new(),
7251 )
7252 .unwrap();
7253 assert_eq!(rs.len(), 1, "only bob lacks age");
7254 assert_eq!(rs.get(0, "n"), Some(&s("bob")));
7255 }
7256
7257 #[test]
7259 fn is_not_null_filters_present_prop() {
7260 let mut fx = Fx::new();
7261 fx.add("Person", "alice", vec![("age", Value::Int(30))]);
7262 fx.add("Person", "bob", vec![]); let view = fx.view();
7264 let rs = run(
7265 &view,
7266 "MATCH (n:Person) WHERE n.age IS NOT NULL RETURN n",
7267 &BTreeMap::new(),
7268 )
7269 .unwrap();
7270 assert_eq!(rs.len(), 1, "only alice has age");
7271 assert_eq!(rs.get(0, "n"), Some(&s("alice")));
7272 }
7273
7274 #[test]
7277 fn optional_match_is_null_anti_join() {
7278 let mut fx = Fx::new();
7279 let alice = fx.add("Person", "alice", vec![]);
7280 let bob = fx.add("Person", "bob", vec![]);
7281 let carol = fx.add("Person", "carol", vec![]);
7282 fx.edge("KNOWS", alice, carol, vec![]); fx.edge("KNOWS", carol, bob, vec![]); let view = fx.view();
7286 let rs = run(
7287 &view,
7288 "MATCH (a:Person) OPTIONAL MATCH (a)-[:KNOWS]->(b) WITH a, b WHERE b IS NULL RETURN a",
7289 &BTreeMap::new(),
7290 )
7291 .unwrap();
7292 assert_eq!(rs.len(), 1, "only bob has no outgoing KNOWS edge");
7293 assert_eq!(rs.get(0, "a"), Some(&s("bob")));
7294 }
7295
7296 #[test]
7298 fn is_null_combined_with_and_exec() {
7299 let mut fx = Fx::new();
7300 fx.add("N", "a", vec![("x", Value::Int(1))]);
7301 fx.add("N", "b", vec![("x", Value::Int(2)), ("y", Value::Int(9))]);
7302 fx.add("N", "c", vec![]); let view = fx.view();
7304 let rs = run(
7306 &view,
7307 "MATCH (n:N) WHERE n.y IS NULL AND n.x IS NOT NULL RETURN n",
7308 &BTreeMap::new(),
7309 )
7310 .unwrap();
7311 assert_eq!(rs.len(), 1);
7312 assert_eq!(rs.get(0, "n"), Some(&s("a")));
7313 }
7314
7315 #[test]
7319 fn arithmetic_add_in_return() {
7320 let mut fx = Fx::new();
7321 fx.add("Person", "alice", vec![("age", Value::Int(29))]);
7322 let view = fx.view();
7323 let rs = run(
7324 &view,
7325 "MATCH (n:Person) RETURN n.age + 1 AS adjusted",
7326 &BTreeMap::new(),
7327 )
7328 .unwrap();
7329 assert_eq!(rs.len(), 1);
7330 assert_eq!(rs.get(0, "adjusted"), Some(&Value::Int(30)));
7331 }
7332
7333 #[test]
7335 fn arithmetic_precedence_parens_pin() {
7336 let mut fx = Fx::new();
7337 fx.add("N", "x", vec![]);
7338 let view = fx.view();
7339 let rs = run(
7340 &view,
7341 "MATCH (n:N) RETURN (1 + 2) * 3 AS r",
7342 &BTreeMap::new(),
7343 )
7344 .unwrap();
7345 assert_eq!(rs.len(), 1);
7346 assert_eq!(rs.get(0, "r"), Some(&Value::Int(9)));
7347 }
7348
7349 #[test]
7351 fn arithmetic_precedence_mul_over_add_pin() {
7352 let mut fx = Fx::new();
7353 fx.add("N", "x", vec![]);
7354 let view = fx.view();
7355 let rs = run(&view, "MATCH (n:N) RETURN 1 + 2 * 3 AS r", &BTreeMap::new()).unwrap();
7356 assert_eq!(rs.len(), 1);
7357 assert_eq!(rs.get(0, "r"), Some(&Value::Int(7)));
7358 }
7359
7360 #[test]
7362 fn arithmetic_div_by_zero_returns_error() {
7363 let mut fx = Fx::new();
7364 fx.add("N", "x", vec![]);
7365 let view = fx.view();
7366 let err = run(
7367 &view,
7368 "MATCH (n:N) WHERE 1 / 0 > 0 RETURN n",
7369 &BTreeMap::new(),
7370 )
7371 .expect_err("division by zero must error");
7372 assert!(
7373 err.contains("division by zero"),
7374 "error must mention division by zero, got: {err}"
7375 );
7376 }
7377
7378 #[test]
7380 fn arithmetic_null_propagates() {
7381 let mut fx = Fx::new();
7382 fx.add("N", "x", vec![]); let view = fx.view();
7384 let rs = run(
7386 &view,
7387 "MATCH (n:N) WHERE n.val + 1 > 0 RETURN n",
7388 &BTreeMap::new(),
7389 )
7390 .unwrap();
7391 assert_eq!(rs.len(), 0, "null arithmetic must not match");
7392 }
7393
7394 #[test]
7396 fn case_when_expression_in_return() {
7397 let mut fx = Fx::new();
7398 fx.add("N", "a", vec![("id", s("a")), ("age", i(20))]);
7399 fx.add("N", "b", vec![("id", s("b")), ("age", i(65))]);
7400 let v = fx.view();
7401 let rs = run(
7402 &v,
7403 "MATCH (n:N) RETURN n.id AS id, \
7404 CASE WHEN n.age >= 65 THEN 'senior' ELSE 'other' END AS band",
7405 &BTreeMap::new(),
7406 )
7407 .unwrap();
7408 let band_of = |who: &str| {
7409 (0..rs.len())
7410 .find(|&i| rs.get(i, "id") == Some(&s(who)))
7411 .and_then(|i| rs.get(i, "band").cloned())
7412 };
7413 assert_eq!(band_of("a"), Some(s("other")));
7414 assert_eq!(band_of("b"), Some(s("senior")));
7415 }
7416
7417 #[test]
7418 fn case_when_no_else_yields_null() {
7419 let mut fx = Fx::new();
7420 fx.add("N", "a", vec![("age", i(20))]);
7421 let v = fx.view();
7422 let rs = run(
7423 &v,
7424 "MATCH (n:N) RETURN CASE WHEN n.age >= 65 THEN 'senior' END AS band",
7425 &BTreeMap::new(),
7426 )
7427 .unwrap();
7428 assert_eq!(rs.get(0, "band"), None);
7429 }
7430
7431 #[test]
7432 fn multi_relationship_type_pattern_matches_either() {
7433 let mut fx = Fx::new();
7434 let a = fx.add("N", "a", vec![("id", s("a"))]);
7435 let b = fx.add("N", "b", vec![("id", s("b"))]);
7436 let c = fx.add("N", "c", vec![("id", s("c"))]);
7437 let d = fx.add("N", "d", vec![("id", s("d"))]);
7438 fx.edge("KNOWS", a, b, vec![]);
7439 fx.edge("LIKES", a, c, vec![]);
7440 fx.edge("HATES", a, d, vec![]);
7441 let v = fx.view();
7442 let rs = run(
7444 &v,
7445 "MATCH (a:N {id: 'a'})-[r:KNOWS|:LIKES]->(x) RETURN x",
7446 &BTreeMap::new(),
7447 )
7448 .unwrap();
7449 assert_eq!(rs.len(), 2, "KNOWS|LIKES reaches exactly b and c");
7450 }
7451
7452 #[test]
7453 fn collect_grouped_gathers_values_per_group() {
7454 let mut fx = Fx::new();
7455 fx.add("P", "a", vec![("city", s("austin")), ("name", s("Ann"))]);
7456 fx.add("P", "b", vec![("city", s("austin")), ("name", s("Bob"))]);
7457 fx.add("P", "c", vec![("city", s("boston")), ("name", s("Cy"))]);
7458 let v = fx.view();
7459 let rs = run(
7460 &v,
7461 "MATCH (n:P) RETURN n.city AS city, collect(n.name) AS names",
7462 &BTreeMap::new(),
7463 )
7464 .unwrap();
7465 assert_eq!(rs.len(), 2, "two city groups");
7466 let austin = (0..rs.len())
7468 .find(|&i| rs.get(i, "city") == Some(&s("austin")))
7469 .expect("austin group present");
7470 assert_eq!(
7471 rs.get(austin, "names"),
7472 Some(&Value::List(vec![s("Ann"), s("Bob")]))
7473 );
7474 }
7475
7476 #[test]
7477 fn collect_ungrouped_gathers_all_into_one_list() {
7478 let mut fx = Fx::new();
7479 fx.add("P", "a", vec![("name", s("Ann"))]);
7480 fx.add("P", "b", vec![("name", s("Bob"))]);
7481 let v = fx.view();
7482 let rs = run(
7483 &v,
7484 "MATCH (n:P) RETURN collect(n.name) AS names",
7485 &BTreeMap::new(),
7486 )
7487 .unwrap();
7488 assert_eq!(rs.len(), 1);
7489 assert_eq!(
7490 rs.get(0, "names"),
7491 Some(&Value::List(vec![s("Ann"), s("Bob")]))
7492 );
7493 }
7494
7495 #[test]
7496 fn string_predicate_functions_in_where() {
7497 let mut fx = Fx::new();
7498 fx.add("N", "a", vec![("email", s("alice@acme.com"))]);
7499 fx.add("N", "b", vec![("email", s("bob@other.org"))]);
7500 let v = fx.view();
7501 let p = BTreeMap::new();
7502 assert_eq!(
7503 run(
7504 &v,
7505 "MATCH (n:N) WHERE endsWith(n.email, '.com') RETURN n",
7506 &p
7507 )
7508 .unwrap()
7509 .len(),
7510 1
7511 );
7512 assert_eq!(
7513 run(
7514 &v,
7515 "MATCH (n:N) WHERE startsWith(n.email, 'bob') RETURN n",
7516 &p
7517 )
7518 .unwrap()
7519 .len(),
7520 1
7521 );
7522 assert_eq!(
7523 run(
7524 &v,
7525 "MATCH (n:N) WHERE contains(n.email, 'acme') RETURN n",
7526 &p
7527 )
7528 .unwrap()
7529 .len(),
7530 1
7531 );
7532 }
7533
7534 #[test]
7535 fn coercion_functions_in_return() {
7536 let mut fx = Fx::new();
7537 fx.add("N", "a", vec![("s", s("42")), ("n", i(7)), ("g", f(3.9))]);
7538 let v = fx.view();
7539 let rs = run(
7540 &v,
7541 "MATCH (n:N) RETURN toInteger(n.s) AS ti, toFloat(n.n) AS tf, toString(n.g) AS ts",
7542 &BTreeMap::new(),
7543 )
7544 .unwrap();
7545 assert_eq!(rs.get(0, "ti"), Some(&Value::Int(42)));
7546 assert_eq!(rs.get(0, "tf"), Some(&Value::Float(7.0)));
7547 assert_eq!(rs.get(0, "ts"), Some(&Value::Str("3.9".into())));
7548 }
7549
7550 #[test]
7551 fn to_integer_unparseable_string_is_null() {
7552 let mut fx = Fx::new();
7553 fx.add("N", "a", vec![("s", s("not-a-number"))]);
7554 let v = fx.view();
7555 let rs = run(
7556 &v,
7557 "MATCH (n:N) RETURN toInteger(n.s) AS ti",
7558 &BTreeMap::new(),
7559 )
7560 .unwrap();
7561 assert_eq!(rs.get(0, "ti"), None);
7562 }
7563
7564 #[test]
7565 fn index_scan_uses_index_and_matches_fallback() {
7566 use core_storage::property_index::PropertyIndex;
7567 use std::sync::atomic::Ordering;
7568
7569 let mut fx = Fx::new();
7570 let a = fx.add("Person", "a", vec![("city", s("austin"))]);
7571 let _b = fx.add("Person", "b", vec![("city", s("boston"))]);
7572 let c = fx.add("Person", "c", vec![("city", s("austin"))]);
7573
7574 let mut pi = PropertyIndex::new();
7575 pi.enable("Person", "city");
7576 pi.set("Person", "city", a, &s("austin"));
7577 pi.set("Person", "city", 1, &s("boston"));
7578 pi.set("Person", "city", c, &s("austin"));
7579
7580 let q = "MATCH (n:Person {city: 'austin'}) RETURN n";
7581
7582 let before = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7584 let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
7585 let after = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7586 assert!(after > before, "IndexScan must take the indexed path");
7587 assert_eq!(indexed.len(), 2);
7588
7589 let before2 = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7592 let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7593 let after2 = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7594 assert_eq!(after2, before2, "fallback must not touch the index counter");
7595 assert_eq!(
7596 fallback.len(),
7597 indexed.len(),
7598 "fallback matches indexed result"
7599 );
7600 }
7601
7602 #[test]
7603 fn arithmetic_in_where_comparison() {
7604 let mut fx = Fx::new();
7605 fx.add("Person", "alice", vec![("age", Value::Int(5))]); fx.add("Person", "bob", vec![("age", Value::Int(4))]); let view = fx.view();
7608 let rs = run(
7609 &view,
7610 "MATCH (n:Person) WHERE n.age + 1 > 5 RETURN n",
7611 &BTreeMap::new(),
7612 )
7613 .unwrap();
7614 assert_eq!(rs.len(), 1);
7615 assert_eq!(rs.get(0, "n"), Some(&s("alice")));
7616 }
7617
7618 #[test]
7622 fn where_fold_indexed_matches_fallback() {
7623 use core_storage::property_index::PropertyIndex;
7624 use std::sync::atomic::Ordering;
7625
7626 let mut fx = Fx::new();
7627 let a = fx.add("Person", "alice", vec![("city", s("austin"))]);
7628 let b = fx.add("Person", "bob", vec![("city", s("boston"))]);
7629 let c = fx.add("Person", "carol", vec![("city", s("austin"))]);
7630
7631 let mut pi = PropertyIndex::new();
7632 pi.enable("Person", "city");
7633 pi.set("Person", "city", a, &s("austin"));
7634 pi.set("Person", "city", b, &s("boston"));
7635 pi.set("Person", "city", c, &s("austin"));
7636
7637 let q = "MATCH (n:Person) WHERE n.city = 'austin' RETURN n";
7638
7639 let before = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7641 let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
7642 let after = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7643 assert!(
7644 after > before,
7645 "WHERE equality fold must take the indexed path"
7646 );
7647 assert_eq!(indexed.len(), 2);
7648
7649 let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7651 assert_eq!(
7652 rows_of(&fallback),
7653 rows_of(&indexed),
7654 "fallback must return identical rows, not just the same count"
7655 );
7656 }
7657
7658 #[test]
7660 fn where_fold_miss_returns_empty() {
7661 let mut fx = Fx::new();
7662 fx.add("Person", "alice", vec![("city", s("austin"))]);
7663 let q = "MATCH (n:Person) WHERE n.city = 'berlin' RETURN n";
7664 let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7665 assert_eq!(rs.len(), 0);
7666 }
7667
7668 #[test]
7670 fn where_fold_param_uses_index() {
7671 use core_storage::property_index::PropertyIndex;
7672 use std::sync::atomic::Ordering;
7673
7674 let mut fx = Fx::new();
7675 let a = fx.add("Person", "alice", vec![("city", s("austin"))]);
7676 let b = fx.add("Person", "bob", vec![("city", s("boston"))]);
7677
7678 let mut pi = PropertyIndex::new();
7679 pi.enable("Person", "city");
7680 pi.set("Person", "city", a, &s("austin"));
7681 pi.set("Person", "city", b, &s("boston"));
7682
7683 let q = "MATCH (n:Person) WHERE n.city = $c RETURN n";
7684 let mut params = BTreeMap::new();
7685 params.insert("c".to_string(), s("austin"));
7686
7687 let before = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7688 let rs = run(&fx.view_indexed(&pi), q, ¶ms).unwrap();
7689 let after = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
7690 assert!(after > before, "$param WHERE equality must use index");
7691 assert_eq!(rs.len(), 1);
7692 }
7693
7694 #[test]
7696 fn where_fold_residual_filter_applied() {
7697 let mut fx = Fx::new();
7698 fx.add(
7699 "Person",
7700 "young-austin",
7701 vec![("city", s("austin")), ("age", Value::Int(20))],
7702 );
7703 fx.add(
7704 "Person",
7705 "old-austin",
7706 vec![("city", s("austin")), ("age", Value::Int(40))],
7707 );
7708 fx.add(
7709 "Person",
7710 "boston",
7711 vec![("city", s("boston")), ("age", Value::Int(20))],
7712 );
7713
7714 let q = "MATCH (n:Person) WHERE n.city = 'austin' AND n.age > 30 RETURN n";
7715 let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7716 assert_eq!(rs.len(), 1, "only old-austin should match city+age filter");
7717 assert_eq!(rs.get(0, "n"), Some(&s("old-austin")));
7718 }
7719
7720 #[test]
7726 fn where_fold_unindexed_field_fallback() {
7727 let mut fx = Fx::new();
7728 fx.add("Person", "a", vec![("notindexed", s("x"))]);
7729 fx.add("Person", "b", vec![("notindexed", s("y"))]);
7730 let rs = run(
7731 &fx.view(),
7732 "MATCH (n:Person) WHERE n.notindexed = 'x' RETURN n",
7733 &BTreeMap::new(),
7734 )
7735 .unwrap();
7736 assert_eq!(rs.len(), 1);
7737 assert_eq!(rs.get(0, "n"), Some(&s("a")));
7738 }
7739
7740 #[test]
7744 fn index_intersect_both_indexed_fires() {
7745 use core_storage::property_index::PropertyIndex;
7746 use std::sync::atomic::Ordering;
7747
7748 let mut fx = Fx::new();
7749 let a = fx.add(
7751 "Person",
7752 "alice",
7753 vec![("city", s("austin")), ("age", Value::Int(30))],
7754 );
7755 let b = fx.add(
7757 "Person",
7758 "bob",
7759 vec![("city", s("austin")), ("age", Value::Int(25))],
7760 );
7761 let c = fx.add(
7763 "Person",
7764 "carol",
7765 vec![("city", s("boston")), ("age", Value::Int(30))],
7766 );
7767
7768 let mut pi = PropertyIndex::new();
7769 pi.enable("Person", "city");
7770 pi.enable("Person", "age");
7771 pi.set("Person", "city", a, &s("austin"));
7772 pi.set("Person", "city", b, &s("austin"));
7773 pi.set("Person", "city", c, &s("boston"));
7774 pi.set("Person", "age", a, &Value::Int(30));
7775 pi.set("Person", "age", b, &Value::Int(25));
7776 pi.set("Person", "age", c, &Value::Int(30));
7777
7778 let q = "MATCH (n:Person) WHERE n.city = 'austin' AND n.age = 30 RETURN n";
7779
7780 let before = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
7781 let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
7782 let after = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
7783 assert!(
7784 after > before,
7785 "IndexIntersect must advance counter on indexed path"
7786 );
7787 assert_eq!(indexed.len(), 1);
7788 assert_eq!(indexed.get(0, "n"), Some(&s("alice")));
7789
7790 let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7792 assert_eq!(
7793 rows_of(&fallback),
7794 rows_of(&indexed),
7795 "fallback must return identical rows"
7796 );
7797 }
7798
7799 #[test]
7801 fn index_intersect_one_indexed_one_not() {
7802 use core_storage::property_index::PropertyIndex;
7803 use std::sync::atomic::Ordering;
7804
7805 let mut fx = Fx::new();
7806 let a = fx.add(
7807 "Person",
7808 "alice",
7809 vec![("city", s("austin")), ("role", s("eng"))],
7810 );
7811 let b = fx.add(
7812 "Person",
7813 "bob",
7814 vec![("city", s("austin")), ("role", s("mgr"))],
7815 );
7816 let _c = fx.add(
7817 "Person",
7818 "carol",
7819 vec![("city", s("boston")), ("role", s("eng"))],
7820 );
7821
7822 let mut pi = PropertyIndex::new();
7824 pi.enable("Person", "city");
7825 pi.set("Person", "city", a, &s("austin"));
7826 pi.set("Person", "city", b, &s("austin"));
7827
7828 let q = "MATCH (n:Person) WHERE n.city = 'austin' AND n.role = 'eng' RETURN n";
7829
7830 let before = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
7831 let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
7832 let after = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
7833 assert!(
7834 after > before,
7835 "IndexIntersect must fire when at least one field is indexed"
7836 );
7837 assert_eq!(indexed.len(), 1);
7838 assert_eq!(indexed.get(0, "n"), Some(&s("alice")));
7839
7840 let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7841 assert_eq!(
7842 rows_of(&fallback),
7843 rows_of(&indexed),
7844 "fallback must return identical rows"
7845 );
7846 }
7847
7848 #[test]
7853 fn index_intersect_no_indexed_fallback() {
7854 let mut fx = Fx::new();
7855 fx.add("Person", "alice", vec![("x", s("1")), ("y", s("a"))]);
7856 fx.add("Person", "bob", vec![("x", s("1")), ("y", s("b"))]);
7857 fx.add("Person", "carol", vec![("x", s("2")), ("y", s("a"))]);
7858
7859 let q = "MATCH (n:Person) WHERE n.x = '1' AND n.y = 'a' RETURN n";
7860 let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7861 assert_eq!(rs.len(), 1);
7862 assert_eq!(rs.get(0, "n"), Some(&s("alice")));
7863 }
7864
7865 #[test]
7867 fn index_intersect_empty_intersection() {
7868 use core_storage::property_index::PropertyIndex;
7869
7870 let mut fx = Fx::new();
7871 let a = fx.add(
7872 "Person",
7873 "alice",
7874 vec![("city", s("austin")), ("age", Value::Int(30))],
7875 );
7876 let b = fx.add(
7877 "Person",
7878 "bob",
7879 vec![("city", s("boston")), ("age", Value::Int(25))],
7880 );
7881
7882 let mut pi = PropertyIndex::new();
7883 pi.enable("Person", "city");
7884 pi.enable("Person", "age");
7885 pi.set("Person", "city", a, &s("austin"));
7886 pi.set("Person", "city", b, &s("boston"));
7887 pi.set("Person", "age", a, &Value::Int(30));
7888 pi.set("Person", "age", b, &Value::Int(25));
7889
7890 let q = "MATCH (n:Person) WHERE n.city = 'boston' AND n.age = 30 RETURN n";
7892 let rs = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
7893 assert_eq!(rs.len(), 0);
7894 }
7895
7896 #[test]
7898 fn index_intersect_with_params() {
7899 use core_storage::property_index::PropertyIndex;
7900 use std::sync::atomic::Ordering;
7901
7902 let mut fx = Fx::new();
7903 let a = fx.add(
7904 "Person",
7905 "alice",
7906 vec![("city", s("austin")), ("age", Value::Int(30))],
7907 );
7908 let b = fx.add(
7909 "Person",
7910 "bob",
7911 vec![("city", s("boston")), ("age", Value::Int(30))],
7912 );
7913
7914 let mut pi = PropertyIndex::new();
7915 pi.enable("Person", "city");
7916 pi.enable("Person", "age");
7917 pi.set("Person", "city", a, &s("austin"));
7918 pi.set("Person", "city", b, &s("boston"));
7919 pi.set("Person", "age", a, &Value::Int(30));
7920 pi.set("Person", "age", b, &Value::Int(30));
7921
7922 let q = "MATCH (n:Person) WHERE n.city = $c AND n.age = $a RETURN n";
7923 let mut params = BTreeMap::new();
7924 params.insert("c".to_string(), s("austin"));
7925 params.insert("a".to_string(), Value::Int(30));
7926
7927 let before = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
7928 let indexed = run(&fx.view_indexed(&pi), q, ¶ms).unwrap();
7929 let after = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
7930 assert!(after > before, "$param intersect must fire indexed path");
7931 assert_eq!(indexed.len(), 1);
7932 assert_eq!(indexed.get(0, "n"), Some(&s("alice")));
7933
7934 let fallback = run(&fx.view(), q, ¶ms).unwrap();
7935 assert_eq!(
7936 rows_of(&fallback),
7937 rows_of(&indexed),
7938 "fallback must return identical rows"
7939 );
7940 }
7941
7942 #[test]
7944 fn index_intersect_three_fields() {
7945 use core_storage::property_index::PropertyIndex;
7946
7947 let mut fx = Fx::new();
7948 let a = fx.add(
7950 "Person",
7951 "alice",
7952 vec![
7953 ("city", s("austin")),
7954 ("age", Value::Int(30)),
7955 ("role", s("eng")),
7956 ],
7957 );
7958 let b = fx.add(
7960 "Person",
7961 "bob",
7962 vec![
7963 ("city", s("austin")),
7964 ("age", Value::Int(30)),
7965 ("role", s("mgr")),
7966 ],
7967 );
7968
7969 let mut pi = PropertyIndex::new();
7970 pi.enable("Person", "city");
7971 pi.enable("Person", "age");
7972 pi.set("Person", "city", a, &s("austin"));
7974 pi.set("Person", "city", b, &s("austin"));
7975 pi.set("Person", "age", a, &Value::Int(30));
7976 pi.set("Person", "age", b, &Value::Int(30));
7977
7978 let q =
7979 "MATCH (n:Person) WHERE n.city = 'austin' AND n.age = 30 AND n.role = 'eng' RETURN n";
7980 let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
7981 assert_eq!(indexed.len(), 1);
7982 assert_eq!(indexed.get(0, "n"), Some(&s("alice")));
7983
7984 let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
7985 assert_eq!(
7986 rows_of(&fallback),
7987 rows_of(&indexed),
7988 "fallback must return identical rows"
7989 );
7990 }
7991 fn assoc_graph() -> Fx {
8004 let mut fx = Fx::new();
8005 let t1 = fx.add(
8006 "Talent",
8007 "t1",
8008 vec![
8009 ("status", s("published")),
8010 ("years_of_experience", i(12)),
8011 (
8012 "specialties",
8013 Value::List(vec![s("hospitality"), s("retail")]),
8014 ),
8015 ("location", Value::List(vec![f(40.71), f(-74.01)])),
8016 ],
8017 );
8018 let t2 = fx.add(
8019 "Talent",
8020 "t2",
8021 vec![
8022 ("status", s("published")),
8023 ("years_of_experience", i(11)),
8024 ("location", Value::List(vec![f(41.88), f(-87.63)])),
8025 ],
8026 );
8027 let t3 = fx.add(
8028 "Talent",
8029 "t3",
8030 vec![("status", s("published")), ("years_of_experience", i(3))],
8031 );
8032 let c1 = fx.add("Company", "c1", vec![("name", s("Acme Design Works"))]);
8033 let c2 = fx.add("Company", "c2", vec![("name", s("Beta Studio"))]);
8034 let c3 = fx.add("Company", "c3", vec![("name", s("Gamma Works"))]);
8035 for (t, c) in [(t1, c1), (t2, c1), (t3, c3)] {
8036 fx.edge("INDUSTRY_ALIGNMENT", t, c, vec![]);
8037 fx.edge("SPECIALTY_MATCH", t, c, vec![]);
8038 fx.edge("LOCATION_FIT", t, c, vec![]);
8039 }
8040 fx.edge("INDUSTRY_ALIGNMENT", t1, c2, vec![]);
8042 fx.edge("SPECIALTY_MATCH", t1, c2, vec![]);
8043 fx
8044 }
8045
8046 #[test]
8049 fn node_key_reads_as_a_property() {
8050 let fx = assoc_graph();
8051 let rs = run(
8052 &fx.view(),
8053 "MATCH (n:Company) RETURN n.key",
8054 &BTreeMap::new(),
8055 )
8056 .unwrap();
8057 assert_eq!(
8058 rows_of(&rs),
8059 vec![
8060 vec![Some(s("c1"))],
8061 vec![Some(s("c2"))],
8062 vec![Some(s("c3"))]
8063 ]
8064 );
8065 }
8066
8067 #[test]
8071 fn node_key_survives_a_with_aggregation() {
8072 let fx = assoc_graph();
8073 let rs = run(
8074 &fx.view(),
8075 "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company) \
8076 WITH c, count(*) AS n WHERE n >= 1 RETURN c.key, key(c), n",
8077 &BTreeMap::new(),
8078 )
8079 .unwrap();
8080 assert_eq!(
8081 rows_of(&rs),
8082 vec![
8083 vec![Some(s("c1")), Some(s("c1")), Some(i(2))],
8084 vec![Some(s("c2")), Some(s("c2")), Some(i(1))],
8085 vec![Some(s("c3")), Some(s("c3")), Some(i(1))],
8086 ]
8087 );
8088 }
8089
8090 #[test]
8092 fn stored_key_property_wins_over_node_key() {
8093 let mut fx = Fx::new();
8094 fx.add("N", "a", vec![("key", s("stored"))]);
8095 let rs = run(&fx.view(), "MATCH (n:N) RETURN n.key", &BTreeMap::new()).unwrap();
8096 assert_eq!(rows_of(&rs), vec![vec![Some(s("stored"))]]);
8097 }
8098
8099 #[test]
8102 fn labels_and_label_read_the_node_label() {
8103 let fx = assoc_graph();
8104 let rs = run(
8105 &fx.view(),
8106 "MATCH (n:Company) RETURN labels(n), n.label LIMIT 1",
8107 &BTreeMap::new(),
8108 )
8109 .unwrap();
8110 assert_eq!(
8111 rows_of(&rs),
8112 vec![vec![
8113 Some(Value::List(vec![s("Company")])),
8114 Some(s("Company"))
8115 ]]
8116 );
8117 }
8118
8119 #[test]
8121 fn unknown_function_is_a_named_error() {
8122 let fx = assoc_graph();
8123 let err = run(
8124 &fx.view(),
8125 "MATCH (n:Company) RETURN nodes(n)",
8126 &BTreeMap::new(),
8127 )
8128 .expect_err("unknown function must error");
8129 assert!(err.contains("unknown function `nodes`"), "{err}");
8130 assert!(
8131 err.contains("labels"),
8132 "error must list what is supported: {err}"
8133 );
8134 }
8135
8136 #[test]
8139 fn infix_string_predicates_filter() {
8140 let fx = assoc_graph();
8141 let p = BTreeMap::new();
8142 for (q, want) in [
8143 (
8144 "MATCH (c:Company) WHERE c.name STARTS WITH 'Acme' RETURN c.key",
8145 vec!["c1"],
8146 ),
8147 (
8148 "MATCH (c:Company) WHERE c.name ENDS WITH 'Works' RETURN c.key",
8149 vec!["c1", "c3"],
8150 ),
8151 (
8152 "MATCH (c:Company) WHERE c.name CONTAINS 'Studio' RETURN c.key",
8153 vec!["c2"],
8154 ),
8155 (
8156 "MATCH (c:Company) WHERE NOT c.name CONTAINS 'Works' RETURN c.key",
8157 vec!["c2"],
8158 ),
8159 ] {
8160 let rs = run(&fx.view(), q, &p).unwrap_or_else(|e| panic!("{q}: {e}"));
8161 let got: Vec<String> = (0..rs.len())
8162 .map(|r| match rs.row(r)[0].clone() {
8163 Some(Value::Str(k)) => k,
8164 other => panic!("{q}: {other:?}"),
8165 })
8166 .collect();
8167 assert_eq!(got, want, "{q}");
8168 }
8169 }
8170
8171 #[test]
8174 fn infix_string_predicate_on_missing_property_is_false() {
8175 let fx = assoc_graph();
8176 let rs = run(
8177 &fx.view(),
8178 "MATCH (t:Talent) WHERE t.name STARTS WITH 'x' RETURN t.key",
8179 &BTreeMap::new(),
8180 )
8181 .unwrap();
8182 assert_eq!(rows_of(&rs), Vec::<Vec<Option<Value>>>::new());
8183 }
8184
8185 #[test]
8188 fn starts_without_with_is_a_named_error() {
8189 let err = parse(&lex("MATCH (c:Company) WHERE c.name STARTS 'Acme' RETURN c").unwrap())
8190 .expect_err("must not parse");
8191 assert!(err.contains("expected WITH after STARTS"), "{err}");
8192 }
8193
8194 #[test]
8197 fn list_subscript_reads_one_element() {
8198 let fx = assoc_graph();
8199 let rs = run(
8200 &fx.view(),
8201 "MATCH (t:Talent) WHERE t.key = 't1' \
8202 RETURN t.location[0] AS lat, t.location[1] AS lon, \
8203 t.location[-1] AS last, t.location[7] AS oob, t.status[0] AS notalist",
8204 &BTreeMap::new(),
8205 )
8206 .unwrap();
8207 assert_eq!(
8208 rows_of(&rs),
8209 vec![vec![
8210 Some(f(40.71)),
8211 Some(f(-74.01)),
8212 Some(f(-74.01)),
8213 None,
8214 None
8215 ]]
8216 );
8217 }
8218
8219 #[test]
8223 fn node_key_works_in_every_position() {
8224 let fx = assoc_graph();
8225 let p = BTreeMap::new();
8226 for (q, want) in [
8227 (
8228 "MATCH (c:Company) WHERE c.key = 'c2' RETURN c.key",
8229 vec!["c2"],
8230 ),
8231 (
8232 "MATCH (c:Company) WHERE c.key <> 'c1' RETURN c.key LIMIT 1",
8233 vec!["c2"],
8234 ),
8235 (
8236 "MATCH (c:Company) WITH c ORDER BY c.key DESC RETURN c.key LIMIT 1",
8237 vec!["c3"],
8238 ),
8239 ("MATCH (c:Company {key: 'c3'}) RETURN c.key", vec!["c3"]),
8240 (
8241 "MATCH (c:Company) WHERE c.label = 'Company' RETURN c.key LIMIT 1",
8242 vec!["c1"],
8243 ),
8244 ] {
8245 let rs = run(&fx.view(), q, &p).unwrap_or_else(|e| panic!("{q}: {e}"));
8246 let got: Vec<String> = (0..rs.len())
8247 .map(|r| match rs.row(r)[0].clone() {
8248 Some(Value::Str(k)) => k,
8249 other => panic!("{q}: {other:?}"),
8250 })
8251 .collect();
8252 assert_eq!(got, want, "{q}");
8253 }
8254 }
8255
8256 #[test]
8258 fn list_subscript_filters() {
8259 let fx = assoc_graph();
8260 let rs = run(
8261 &fx.view(),
8262 "MATCH (t:Talent) WHERE t.location[0] > 41.0 RETURN t.key",
8263 &BTreeMap::new(),
8264 )
8265 .unwrap();
8266 assert_eq!(rows_of(&rs), vec![vec![Some(s("t2"))]]);
8267 }
8268
8269 #[test]
8273 fn count_distinct_counts_each_binding_once() {
8274 let fx = assoc_graph();
8275 let q = "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT|:SPECIALTY_MATCH]->(c:Company) \
8276 WITH c, count(t) AS raw, count(DISTINCT t) AS uniq WHERE raw >= 1 \
8277 RETURN c.key, raw, uniq";
8278 let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
8279 assert_eq!(
8280 rows_of(&rs),
8281 vec![
8282 vec![Some(s("c1")), Some(i(4)), Some(i(2))],
8283 vec![Some(s("c2")), Some(i(2)), Some(i(1))],
8284 vec![Some(s("c3")), Some(i(2)), Some(i(1))],
8285 ]
8286 );
8287 }
8288
8289 #[test]
8299 fn count_distinct_on_a_relationship_counts_edges() {
8300 let fx = assoc_graph();
8301 let rs = run(
8302 &fx.view(),
8303 "MATCH (a)-[r]->(b) RETURN count(r) AS raw, count(DISTINCT r) AS uniq",
8304 &BTreeMap::new(),
8305 )
8306 .unwrap();
8307 let row = &rows_of(&rs)[0];
8308 assert_eq!(
8309 row[0], row[1],
8310 "no pair is joined twice by one type: {row:?}"
8311 );
8312 assert_ne!(row[1], Some(i(0)), "a graph full of edges counts them");
8313 }
8314
8315 #[test]
8319 fn count_distinct_on_a_relationship_survives_an_alternation() {
8320 let fx = assoc_graph();
8321 let rs = run(
8322 &fx.view(),
8323 "MATCH (t:Talent)-[r:INDUSTRY_ALIGNMENT|:SPECIALTY_MATCH]->(c:Company) \
8324 RETURN count(r) AS raw, count(DISTINCT r) AS uniq",
8325 &BTreeMap::new(),
8326 )
8327 .unwrap();
8328 let row = &rows_of(&rs)[0];
8329 assert_eq!(row[0], row[1], "every row bound a different edge: {row:?}");
8330 }
8331
8332 #[test]
8334 fn collect_distinct_dedupes() {
8335 let fx = assoc_graph();
8336 let rs = run(
8337 &fx.view(),
8338 "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT|:SPECIALTY_MATCH]->(c:Company) \
8339 WHERE c.key = 'c1' WITH collect(DISTINCT t.status) AS st RETURN st",
8340 &BTreeMap::new(),
8341 )
8342 .unwrap();
8343 assert_eq!(
8344 rows_of(&rs),
8345 vec![vec![Some(Value::List(vec![s("published")]))]]
8346 );
8347 }
8348
8349 #[test]
8351 fn count_distinct_star_is_rejected() {
8352 let err =
8353 parse(&lex("MATCH (n) RETURN count(DISTINCT *)").unwrap()).expect_err("must not parse");
8354 assert!(
8355 err.contains("DISTINCT * is not a valid aggregate argument"),
8356 "{err}"
8357 );
8358 }
8359
8360 #[test]
8362 fn distinct_is_still_usable_as_a_variable_name() {
8363 let q = parse(&lex("MATCH (distinct) RETURN count(distinct)").unwrap()).unwrap();
8364 assert_eq!(
8365 q.returns[0].value,
8366 RetVal::Agg {
8367 func: crate::cypher::ast::AggFunc::Count,
8368 arg: crate::cypher::ast::AggArg::Var("distinct".into()),
8369 }
8370 );
8371 }
8372
8373 #[test]
8377 fn comma_patterns_intersect_on_shared_variables() {
8378 let fx = assoc_graph();
8379 let q = "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company), \
8380 (t)-[:SPECIALTY_MATCH]->(c), \
8381 (t)-[:LOCATION_FIT]->(c) \
8382 WHERE t.status = 'published' AND t.years_of_experience >= 10 \
8383 WITH c, count(DISTINCT t) AS n WHERE n >= 2 \
8384 RETURN c.key, n ORDER BY n DESC";
8385 let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
8386 assert_eq!(rows_of(&rs), vec![vec![Some(s("c1")), Some(i(2))]]);
8387 }
8388
8389 #[test]
8393 fn comma_patterns_exclude_partial_matches() {
8394 let fx = assoc_graph();
8395 let q = "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company), \
8396 (t)-[:SPECIALTY_MATCH]->(c), \
8397 (t)-[:LOCATION_FIT]->(c) \
8398 WHERE t.years_of_experience >= 10 \
8399 WITH c, count(DISTINCT t) AS n RETURN c.key, n";
8400 let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
8401 assert_eq!(rows_of(&rs), vec![vec![Some(s("c1")), Some(i(2))]]);
8402 }
8403
8404 #[test]
8407 fn comma_patterns_equal_separate_match_clauses() {
8408 let commas =
8409 parse(&lex("MATCH (a:A)-[:X]->(b:B), (a)-[:Y]->(b) RETURN a.key").unwrap()).unwrap();
8410 let clauses =
8411 parse(&lex("MATCH (a:A)-[:X]->(b:B) MATCH (a)-[:Y]->(b) RETURN a.key").unwrap())
8412 .unwrap();
8413 assert_eq!(commas.matches, clauses.matches);
8414 }
8415
8416 #[test]
8428 fn aggregate_with_projects_without_a_having_clause() {
8429 let fx = assoc_graph();
8430 let rs = run(
8431 &fx.view(),
8432 "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company) \
8433 WITH c, count(t) AS n RETURN c.name, n * 2 AS dbl",
8434 &BTreeMap::new(),
8435 )
8436 .unwrap();
8437 assert_eq!(rs.columns(), ["c.name", "dbl"]);
8438 assert_eq!(
8439 rows_of(&rs),
8440 vec![
8441 vec![Some(s("Acme Design Works")), Some(i(4))],
8442 vec![Some(s("Beta Studio")), Some(i(2))],
8443 vec![Some(s("Gamma Works")), Some(i(2))],
8444 ]
8445 );
8446 }
8447
8448 #[test]
8451 fn aggregate_with_names_its_projected_columns() {
8452 let fx = assoc_graph();
8453 let rs = run(
8454 &fx.view(),
8455 "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company) \
8456 WITH c, count(t) AS n RETURN key(c), n",
8457 &BTreeMap::new(),
8458 )
8459 .unwrap();
8460 assert_eq!(rs.columns(), ["key(c)", "n"]);
8461 assert_eq!(
8462 rows_of(&rs),
8463 vec![
8464 vec![Some(s("c1")), Some(i(2))],
8465 vec![Some(s("c2")), Some(i(1))],
8466 vec![Some(s("c3")), Some(i(1))],
8467 ]
8468 );
8469 }
8470
8471 #[test]
8474 fn aggregate_with_returning_the_bare_node_keeps_the_key() {
8475 let fx = assoc_graph();
8476 let rs = run(
8477 &fx.view(),
8478 "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company) \
8479 WITH c, count(t) AS n RETURN c",
8480 &BTreeMap::new(),
8481 )
8482 .unwrap();
8483 assert_eq!(rs.columns(), ["c"]);
8484 assert_eq!(
8485 rows_of(&rs),
8486 vec![
8487 vec![Some(s("c1"))],
8488 vec![Some(s("c2"))],
8489 vec![Some(s("c3"))]
8490 ]
8491 );
8492 }
8493
8494 #[test]
8497 fn a_plain_aggregate_is_unchanged() {
8498 let fx = assoc_graph();
8499 let rs = run(
8500 &fx.view(),
8501 "MATCH (c:Company) RETURN c.name, count(*)",
8502 &BTreeMap::new(),
8503 )
8504 .unwrap();
8505 assert_eq!(rs.columns(), ["c.name", "COUNT(*)"]);
8506 assert_eq!(rs.len(), 3);
8507 }
8508
8509 #[test]
8513 fn two_subscripts_of_one_list_are_two_named_columns() {
8514 let fx = assoc_graph();
8515 let rs = run(
8516 &fx.view(),
8517 "MATCH (t:Talent) RETURN t.location[0], t.location[1]",
8518 &BTreeMap::new(),
8519 )
8520 .unwrap();
8521 assert_eq!(rs.columns(), ["t.location[0]", "t.location[1]"]);
8522 assert_eq!(
8523 rows_of(&rs),
8524 vec![
8525 vec![Some(f(40.71)), Some(f(-74.01))],
8526 vec![Some(f(41.88)), Some(f(-87.63))],
8527 vec![None, None],
8529 ]
8530 );
8531 }
8532
8533 #[test]
8537 fn a_subscript_carries_through_a_with_stage() {
8538 let fx = assoc_graph();
8539 let rs = run(
8540 &fx.view(),
8541 "MATCH (t:Talent) WHERE t.location[0] > 41.0 \
8542 WITH t, t.location[0] AS lat, t.location[1] \
8543 RETURN key(t), lat, t.location[1]",
8544 &BTreeMap::new(),
8545 )
8546 .unwrap();
8547 assert_eq!(rs.columns(), ["key(t)", "lat", "t.location[1]"]);
8548 assert_eq!(
8549 rows_of(&rs),
8550 vec![vec![Some(s("t2")), Some(f(41.88)), Some(f(-87.63))]]
8551 );
8552 }
8553
8554 #[test]
8557 fn subscript_column_names_cover_nested_and_computed_forms() {
8558 use crate::cypher::ast::operand_label;
8559 let prop = || Operand::Prop {
8560 var: "t".into(),
8561 field: "location".into(),
8562 };
8563 let at = |idx: Operand| Operand::Index {
8564 base: Box::new(prop()),
8565 index: Box::new(idx),
8566 };
8567 assert_eq!(operand_label(&at(Operand::Lit(i(0)))), "t.location[0]");
8568 assert_eq!(operand_label(&at(Operand::Lit(i(-1)))), "t.location[-1]");
8569 assert_eq!(
8570 operand_label(&at(Operand::Param("k".into()))),
8571 "t.location[$k]"
8572 );
8573 assert_eq!(
8574 operand_label(&at(Operand::Var("j".into()))),
8575 "t.location[j]"
8576 );
8577 assert_eq!(
8578 operand_label(&Operand::Index {
8579 base: Box::new(at(Operand::Lit(i(0)))),
8580 index: Box::new(Operand::Lit(i(1))),
8581 }),
8582 "t.location[0][1]"
8583 );
8584 }
8585
8586 #[test]
8595 fn a_with_alias_is_visible_to_its_where() {
8596 let fx = assoc_graph();
8597 let rs = run(
8598 &fx.view(),
8599 "MATCH (t:Talent) WITH t, t.years_of_experience AS x WHERE x > 11 \
8600 RETURN key(t), x",
8601 &BTreeMap::new(),
8602 )
8603 .unwrap();
8604 assert_eq!(rs.columns(), ["key(t)", "x"]);
8605 assert_eq!(rows_of(&rs), vec![vec![Some(s("t1")), Some(i(12))]]);
8606 }
8607
8608 #[test]
8610 fn a_with_alias_is_visible_to_where_and_order_by_together() {
8611 let fx = assoc_graph();
8612 let rs = run(
8613 &fx.view(),
8614 "MATCH (t:Talent) WITH t, t.years_of_experience AS x WHERE x > 3 \
8615 ORDER BY x DESC RETURN key(t), x",
8616 &BTreeMap::new(),
8617 )
8618 .unwrap();
8619 assert_eq!(rs.columns(), ["key(t)", "x"]);
8620 assert_eq!(
8621 rows_of(&rs),
8622 vec![
8623 vec![Some(s("t1")), Some(i(12))],
8624 vec![Some(s("t2")), Some(i(11))],
8625 ]
8626 );
8627 }
8628
8629 #[test]
8633 fn a_with_carrying_only_a_variable_still_filters_on_it() {
8634 let fx = assoc_graph();
8635 let rs = run(
8636 &fx.view(),
8637 "MATCH (t:Talent) WHERE t.status = 'published' \
8638 WITH t WHERE t.years_of_experience > 11 RETURN key(t)",
8639 &BTreeMap::new(),
8640 )
8641 .unwrap();
8642 assert_eq!(rs.columns(), ["key(t)"]);
8643 assert_eq!(rows_of(&rs), vec![vec![Some(s("t1"))]]);
8644 }
8645
8646 #[test]
8649 fn a_with_alias_projects_under_its_new_name() {
8650 let fx = assoc_graph();
8651 let rs = run(
8652 &fx.view(),
8653 "MATCH (c:Company) WITH c, c.name AS nm RETURN nm",
8654 &BTreeMap::new(),
8655 )
8656 .unwrap();
8657 assert_eq!(rs.columns(), ["nm"]);
8658 assert_eq!(
8659 rows_of(&rs),
8660 vec![
8661 vec![Some(s("Acme Design Works"))],
8662 vec![Some(s("Beta Studio"))],
8663 vec![Some(s("Gamma Works"))],
8664 ]
8665 );
8666 }
8667
8668 #[test]
8672 fn every_with_alias_shape_is_in_scope_for_the_where() {
8673 let fx = assoc_graph();
8674 let p = BTreeMap::new();
8675 for (q, want) in [
8676 (
8677 "MATCH (t:Talent) WITH t, t.years_of_experience + 1 AS x WHERE x > 12 \
8678 RETURN key(t)",
8679 vec!["t1"],
8680 ),
8681 (
8682 "MATCH (t:Talent) WITH t, t.location[0] AS lat WHERE lat > 41.0 RETURN key(t)",
8683 vec!["t2"],
8684 ),
8685 (
8686 "MATCH (t:Talent) WITH t AS u WHERE u.years_of_experience > 11 RETURN key(u)",
8687 vec!["t1"],
8688 ),
8689 (
8690 "MATCH (t:Talent) WITH t.key AS k WHERE k = 't2' RETURN k",
8691 vec!["t2"],
8692 ),
8693 ] {
8694 let rs = run(&fx.view(), q, &p).unwrap_or_else(|e| panic!("{q}: {e}"));
8695 let got: Vec<String> = (0..rs.len())
8696 .map(|r| match rs.row(r)[0].clone() {
8697 Some(Value::Str(k)) => k,
8698 other => panic!("{q}: {other:?}"),
8699 })
8700 .collect();
8701 assert_eq!(got, want, "{q}");
8702 }
8703 }
8704
8705 #[test]
8708 fn an_unknown_name_in_a_with_where_is_still_an_error() {
8709 let err = plan(
8710 &parse(
8711 &lex(
8712 "MATCH (t:Talent) WITH t, t.years_of_experience AS x WHERE nope > 1 \
8713 RETURN key(t)",
8714 )
8715 .unwrap(),
8716 )
8717 .unwrap(),
8718 )
8719 .expect_err("must not plan");
8720 assert_eq!(err, "unbound variable `nope` in WHERE");
8721 }
8722
8723 #[test]
8726 fn comma_patterns_without_shared_vars_are_a_product() {
8727 let mut fx = Fx::new();
8728 fx.add("A", "a1", vec![]);
8729 fx.add("A", "a2", vec![]);
8730 fx.add("B", "b1", vec![]);
8731 let rs = run(
8732 &fx.view(),
8733 "MATCH (a:A), (b:B) RETURN a.key, b.key",
8734 &BTreeMap::new(),
8735 )
8736 .unwrap();
8737 assert_eq!(
8738 rows_of(&rs),
8739 vec![
8740 vec![Some(s("a1")), Some(s("b1"))],
8741 vec![Some(s("a2")), Some(s("b1"))],
8742 ]
8743 );
8744 }
8745}