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