use crate::cypher::ast::{
ret_val_label, AggArg, AggFunc, Expr, LimitSkip, Operand, OrderItem, OrderTarget, RetItem,
RetVal, UnwindExpr,
};
use crate::cypher::plan::PlanOp;
use crate::cypher::RelDir;
use crate::filter::eval_cmp;
use crate::result::ResultSet;
use crate::traverse::{expand, Dir, EdgeRef};
use crate::value_ops::{cmp_optional, values_equal};
use crate::view::GraphView;
use core_storage::{Value, ValueKey};
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
#[cfg(test)]
static FUSED_SCAN_FIRES: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
#[cfg(test)]
static SCAN_KEY_FIRES: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
#[cfg(test)]
pub static INDEX_SCAN_FIRES: std::sync::atomic::AtomicUsize =
std::sync::atomic::AtomicUsize::new(0);
#[cfg(test)]
pub static INDEX_INTERSECT_FIRES: std::sync::atomic::AtomicUsize =
std::sync::atomic::AtomicUsize::new(0);
pub struct Params<'a>(pub &'a BTreeMap<String, Value>);
#[derive(Clone, Debug)]
enum Cell {
Node(u32),
Rel(EdgeRef),
Path(u8),
Scalar(Value),
}
type Row = Vec<Option<Cell>>;
struct VarTable {
names: Vec<String>,
}
impl VarTable {
fn intern(&mut self, name: &str) -> usize {
if let Some(i) = self.names.iter().position(|n| n == name) {
return i;
}
self.names.push(name.to_string());
self.names.len() - 1
}
fn slot(&self, name: &str) -> Option<usize> {
self.names.iter().position(|n| n == name)
}
}
struct Projected {
columns: Vec<String>,
rows: Vec<Vec<Option<Value>>>,
}
const MAX_INTERMEDIATE_ROWS: usize = 1_000_000;
const MAX_GROUPS: usize = 1_000_000;
type GroupKey = Vec<Option<ValueKey>>;
type GroupEntry = (Vec<Option<Value>>, Vec<AggAcc>);
#[cfg(test)]
thread_local! {
static TEST_MAX_INTERMEDIATE_ROWS: std::cell::Cell<Option<usize>> =
const { std::cell::Cell::new(None) };
static TEST_EXPAND_PRODUCED: std::cell::Cell<Option<usize>> =
const { std::cell::Cell::new(None) };
static TEST_MAX_GROUPS: std::cell::Cell<Option<usize>> =
const { std::cell::Cell::new(None) };
}
fn max_intermediate_rows() -> usize {
#[cfg(test)]
{
TEST_MAX_INTERMEDIATE_ROWS
.with(|c| c.get())
.unwrap_or(MAX_INTERMEDIATE_ROWS)
}
#[cfg(not(test))]
{
MAX_INTERMEDIATE_ROWS
}
}
fn max_groups() -> usize {
#[cfg(test)]
{
TEST_MAX_GROUPS.with(|c| c.get()).unwrap_or(MAX_GROUPS)
}
#[cfg(not(test))]
{
MAX_GROUPS
}
}
#[cfg(test)]
pub(crate) fn with_max_groups<R>(cap: usize, f: impl FnOnce() -> R) -> R {
TEST_MAX_GROUPS.with(|c| {
let prev = c.replace(Some(cap));
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
c.set(prev);
match result {
Ok(v) => v,
Err(p) => std::panic::resume_unwind(p),
}
})
}
#[cfg(test)]
pub(crate) fn with_max_intermediate_rows<R>(cap: usize, f: impl FnOnce() -> R) -> R {
TEST_MAX_INTERMEDIATE_ROWS.with(|c| {
let prev = c.replace(Some(cap));
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
c.set(prev);
match result {
Ok(v) => v,
Err(p) => std::panic::resume_unwind(p),
}
})
}
#[cfg(test)]
fn record_expand_row() {
TEST_EXPAND_PRODUCED.with(|c| {
if let Some(prev) = c.get() {
c.set(Some(prev + 1));
}
});
}
#[cfg(test)]
pub(crate) fn with_expand_counter<R>(f: impl FnOnce() -> R) -> (R, usize) {
TEST_EXPAND_PRODUCED.with(|c| c.set(Some(0)));
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(f));
let count = TEST_EXPAND_PRODUCED.with(|c| c.get().unwrap_or(0));
TEST_EXPAND_PRODUCED.with(|c| c.set(None));
match result {
Ok(v) => (v, count),
Err(p) => std::panic::resume_unwind(p),
}
}
fn row_cap_err(cap: usize) -> String {
format!(
"intermediate result exceeds {cap} rows; add a LIMIT or constrain patterns with shared variables"
)
}
fn group_cap_err() -> String {
format!(
"group count exceeds {} distinct keys; add a WHERE clause or constrain the grouping key",
max_groups()
)
}
fn group_key_normalize(v: &Value) -> Option<ValueKey> {
match v {
Value::Int(n) => Some(ValueKey::FloatBits((*n as f64).to_bits())),
Value::Float(f) => Some(ValueKey::FloatBits(f.to_bits())),
_ => ValueKey::from_value(v),
}
}
pub fn execute(view: &GraphView, plan: &[PlanOp], params: &Params) -> Result<ResultSet, String> {
use crate::cypher::plan::row_bound;
execute_inner(view, plan, params, row_bound(plan))
}
pub fn execute_union(
view: &GraphView,
union: &crate::cypher::parser::UnionQuery,
params: &Params,
) -> Result<ResultSet, String> {
let mut acc: Option<ResultSet> = None;
for (i, part) in union.parts.iter().enumerate() {
let ops = crate::cypher::plan::plan(part)?;
let rs = execute(view, &ops, params)?;
acc = Some(match acc {
None => rs,
Some(prev) => {
if prev.columns() != rs.columns() {
return Err(format!(
"UNION requires matching column names across all parts; \
got {:?} then {:?}",
prev.columns(),
rs.columns()
));
}
let all = union.all_flags.get(i - 1).copied().unwrap_or(false);
combine_result_sets(prev, rs, all)
}
});
}
Ok(acc.expect("UNION query has at least one part"))
}
fn combine_result_sets(mut acc: ResultSet, other: ResultSet, all: bool) -> ResultSet {
for i in 0..other.len() {
acc.push_row(other.row(i).to_vec());
}
if !all {
let mut seen: Vec<Vec<Option<Value>>> = Vec::new();
for i in 0..acc.len() {
let r = acc.row(i).to_vec();
if !seen.contains(&r) {
seen.push(r);
}
}
let mut deduped = ResultSet::new(acc.columns().to_vec());
for r in seen {
deduped.push_row(r);
}
return deduped;
}
acc
}
#[cfg(test)]
pub(crate) fn execute_unbounded(
view: &GraphView,
plan: &[PlanOp],
params: &Params,
) -> Result<ResultSet, String> {
execute_inner(view, plan, params, None)
}
fn execute_inner(
view: &GraphView,
plan: &[PlanOp],
params: &Params,
row_bound: Option<usize>,
) -> Result<ResultSet, String> {
check_params(plan, params)?;
let has_var_expand = plan
.iter()
.any(|op| matches!(op, PlanOp::VarExpand { .. } | PlanOp::ShortestPath { .. }));
let is_pipeline = plan.iter().any(|op| {
matches!(
op,
PlanOp::With { .. } | PlanOp::Unwind { .. } | PlanOp::LeftOuterApply { .. }
)
}) || {
let mut saw_gagg = false;
plan.iter().any(|op| {
if matches!(op, PlanOp::GroupAggregate { .. }) {
saw_gagg = true;
false
} else {
saw_gagg && matches!(op, PlanOp::Filter { .. } | PlanOp::Project { .. })
}
})
};
if !has_var_expand
&& !is_pipeline
&& plan
.iter()
.any(|op| matches!(op, PlanOp::GroupAggregate { .. }))
{
return execute_group_aggregate(view, plan, params);
}
if !has_var_expand
&& !is_pipeline
&& plan.iter().any(|op| matches!(op, PlanOp::Aggregate { .. }))
{
return execute_aggregate(view, plan, params);
}
if let Some(bound) = row_bound {
return execute_pull(view, plan, params, bound);
}
let vars = collect_vars(plan);
let mut rows: Vec<Row> = vec![vec![None; vars.names.len()]];
let mut projected: Option<Projected> = None;
let mut pipeline_group_columns: Option<Vec<String>> = None;
for op in plan {
match op {
PlanOp::ScanLabel { var, label } => {
rows = scan_label(view, &vars, &rows, var, label.as_deref())?;
}
PlanOp::ScanKey { var, key, label } => {
rows = scan_key(view, &vars, &rows, var, key, label.as_deref(), params)?;
}
PlanOp::IndexScan {
var,
label,
field,
value,
} => {
rows = scan_index(
view,
&vars,
&rows,
var,
label.as_deref(),
field,
value,
params,
)?;
}
PlanOp::IndexIntersect {
var,
label,
equalities,
} => {
rows = scan_intersect(
view,
&vars,
&rows,
var,
label.as_deref(),
equalities,
params,
)?;
}
PlanOp::LookupProps { var, props } => {
rows = retain_node(view, &vars, &rows, var, None, props, params)?;
}
PlanOp::JoinBound { var, label, props } => {
rows = retain_node(view, &vars, &rows, var, label.as_deref(), props, params)?;
}
PlanOp::Expand { .. } => {
rows = exec_expand(view, &vars, &rows, op, params)?;
}
PlanOp::VarExpand {
from,
rel_var,
etypes,
dir,
to,
min,
max,
} => {
rows = exec_var_expand(
view, &vars, &rows, from, rel_var, etypes, *dir, to, *min, *max,
)?;
}
PlanOp::ShortestPath {
from,
rel_var,
etypes,
dir,
to,
max_hops,
} => {
rows = exec_shortest_path(
view, &vars, &rows, from, rel_var, etypes, *dir, to, *max_hops,
)?;
}
PlanOp::Filter { expr } => {
rows = exec_filter(view, &vars, &rows, expr, params)?;
}
PlanOp::Project { items } => {
projected = Some(exec_project(view, &vars, &rows, items, params)?);
}
PlanOp::Distinct => {
if let Some(table) = projected.as_mut() {
exec_distinct(table)?;
} else {
return Err("DISTINCT requires a Project".into());
}
}
PlanOp::OrderBy { items } => {
if let Some(table) = projected.as_mut() {
exec_order_by(table, items)?;
} else {
exec_order_by_rows(&vars, &mut rows, items, view);
}
}
PlanOp::Skip(ls) => {
let n = resolve_ls(ls, params)?;
if let Some(table) = projected.as_mut() {
apply_skip(&mut table.rows, n);
} else {
apply_skip(&mut rows, n);
}
}
PlanOp::Limit(ls) => {
let n = resolve_ls(ls, params)?;
if let Some(table) = projected.as_mut() {
apply_limit(&mut table.rows, n);
} else {
apply_limit(&mut rows, n);
}
}
PlanOp::GroupAggregate { keys, aggs } => {
let mut grp_groups: HashMap<GroupKey, GroupEntry> = HashMap::new();
let mut grp_key_order: Vec<GroupKey> = Vec::new();
let mut grp_cells: HashMap<GroupKey, Vec<Option<Cell>>> = HashMap::new();
for row in &rows {
let mut gk: GroupKey = Vec::with_capacity(keys.len());
let mut display_vals: Vec<Option<Value>> = Vec::with_capacity(keys.len());
for (_, item) in keys {
let val = project_item(view, &vars, row, item, params)?;
gk.push(val.as_ref().and_then(group_key_normalize));
display_vals.push(val);
}
if !grp_groups.contains_key(&gk) {
if grp_groups.len() >= max_groups() {
return Err(group_cap_err());
}
grp_key_order.push(gk.clone());
grp_cells.insert(gk.clone(), key_source_cells(&vars, row, keys));
grp_groups.insert(
gk.clone(),
(
display_vals,
aggs.iter().map(|(f, a, _)| AggAcc::for_arg(f, a)).collect(),
),
);
}
let (_, accs) = grp_groups.get_mut(&gk).unwrap();
for (acc, (func, arg, _)) in accs.iter_mut().zip(aggs.iter()) {
update_acc(view, &vars, row, func, arg, acc)?;
}
}
if keys.is_empty() && grp_key_order.is_empty() {
let empty_key: GroupKey = vec![];
grp_key_order.push(empty_key.clone());
grp_groups.insert(
empty_key,
(
vec![],
aggs.iter().map(|(f, a, _)| AggAcc::for_arg(f, a)).collect(),
),
);
}
if is_pipeline {
rows = group_result_to_rows(
keys,
aggs,
grp_key_order,
&mut grp_groups,
&mut grp_cells,
&vars,
);
projected = None;
let mut cols: Vec<String> = keys.iter().map(|(c, _)| c.clone()).collect();
cols.extend(aggs.iter().map(|(_, _, c)| c.clone()));
pipeline_group_columns = Some(cols);
} else {
projected = Some(build_group_projected(
keys,
aggs,
grp_key_order,
&mut grp_groups,
));
pipeline_group_columns = None;
}
}
PlanOp::With {
items,
where_expr,
order_by,
skip,
limit,
} => {
let row_len = vars.names.len();
let mut new_rows: Vec<Row> = Vec::with_capacity(rows.len());
for row in &rows {
let mut new_row: Row = vec![None; row_len];
for item in items {
let col = column_name(item);
let Some(dst_slot) = vars.slot(&col) else {
continue;
};
match &item.value {
RetVal::Var(v) => {
if let Some(src_slot) = vars.slot(v) {
new_row[dst_slot] = row.get(src_slot).cloned().flatten();
}
}
RetVal::Prop { var, field } => {
let val = resolve_prop(view, &vars, row, var, field)?;
new_row[dst_slot] = val.map(Cell::Scalar);
}
RetVal::Agg { .. } => {} RetVal::FuncCall { name, args } => {
let val = eval_func(name, args, view, &vars, row, params)?;
new_row[dst_slot] = val.map(Cell::Scalar);
}
RetVal::ScalarExpr(op) => {
let val = resolve_operand(view, &vars, row, op, params)?;
new_row[dst_slot] = val.map(Cell::Scalar);
}
}
}
new_rows.push(new_row);
}
rows = new_rows;
if let Some(expr) = where_expr {
rows = exec_filter(view, &vars, &rows, expr, params)?;
}
if !order_by.is_empty() {
exec_order_by_rows(&vars, &mut rows, order_by, view);
}
if let Some(ls) = skip {
let n = resolve_ls(ls, params)?;
apply_skip(&mut rows, n);
}
if let Some(ls) = limit {
let n = resolve_ls(ls, params)?;
apply_limit(&mut rows, n);
}
}
PlanOp::Unwind { expr, alias } => {
let alias_slot = vars
.slot(alias)
.ok_or_else(|| format!("UNWIND alias `{alias}` not in VarTable"))?;
let cap = max_intermediate_rows();
let mut new_rows: Vec<Row> = Vec::new();
for row in &rows {
let list_val: Option<Value> = match expr {
UnwindExpr::Lit(vals) => Some(Value::List(vals.clone())),
UnwindExpr::Prop { var, field } => {
resolve_prop(view, &vars, row, var, field)?
}
UnwindExpr::Var(name) => {
let slot = vars
.slot(name)
.ok_or_else(|| format!("UNWIND variable `{name}` is not bound"))?;
match row.get(slot).and_then(|c| c.as_ref()) {
Some(Cell::Scalar(v)) => Some(v.clone()),
Some(Cell::Node(_) | Cell::Rel(_) | Cell::Path(_)) => {
return Err(format!(
"UNWIND requires a list; `{name}` is bound to a graph element"
));
}
None => None,
}
}
};
match list_val {
None => {} Some(Value::List(items_list)) => {
for item_val in items_list {
if new_rows.len() >= cap {
return Err(row_cap_err(cap));
}
let mut new_row = row.clone();
new_row[alias_slot] = Some(Cell::Scalar(item_val));
new_rows.push(new_row);
}
}
Some(other) => {
let type_name = match &other {
Value::Int(_) => "Int",
Value::Float(_) => "Float",
Value::Str(_) => "Str",
Value::Bool(_) => "Bool",
Value::List(_) => unreachable!(),
Value::Map(_) => "Map",
};
return Err(format!(
"UNWIND requires a list; got {type_name} value for `{alias}`"
));
}
}
}
rows = new_rows;
}
PlanOp::Aggregate { func, arg, column } => {
let ctx = AggStreamCtx {
view,
vars: &vars,
params,
func,
arg,
};
let mut acc = AggAcc::for_arg(func, arg);
for row in &rows {
agg_stream(&ctx, &[], row, &mut acc)?;
}
let value = acc.finish();
let mut rs = ResultSet::new(vec![column.clone()]);
rs.push_row(vec![value]);
return Ok(rs);
}
PlanOp::LeftOuterApply {
inner,
optional_vars,
} => {
let cap = max_intermediate_rows();
let mut new_rows: Vec<Row> = Vec::new();
for outer_row in &rows {
let inner_seed: Vec<Row> = vec![outer_row.clone()];
let inner_result =
exec_left_outer_inner(view, &vars, inner_seed, inner, params)?;
if inner_result.is_empty() {
let mut null_row = outer_row.clone();
for v in optional_vars {
if let Some(slot) = vars.slot(v) {
null_row[slot] = None;
}
}
if new_rows.len() >= cap {
return Err(row_cap_err(cap));
}
new_rows.push(null_row);
} else {
for r in inner_result {
if new_rows.len() >= cap {
return Err(row_cap_err(cap));
}
new_rows.push(r);
}
}
}
rows = new_rows;
}
}
}
Ok(match projected {
Some(table) => finish(table),
None => {
if let Some(cols) = pipeline_group_columns {
let mut rs = ResultSet::new(cols.clone());
for row in rows {
let vals: Vec<Option<Value>> = cols
.iter()
.map(|col| {
vars.slot(col)
.and_then(|s| row.get(s))
.and_then(|c| c.as_ref())
.and_then(|c| match c {
Cell::Scalar(v) => Some(v.clone()),
Cell::Node(id) => {
view.ids.key_of(*id).map(|k| Value::Str(k.to_owned()))
}
Cell::Path(h) => Some(Value::Int(*h as i64)),
Cell::Rel(_) => None,
})
})
.collect();
rs.push_row(vals);
}
rs
} else {
ResultSet::new(vec![])
}
}
})
}
fn exec_left_outer_inner(
view: &GraphView,
vars: &VarTable,
mut rows: Vec<Row>,
inner: &[PlanOp],
params: &Params,
) -> Result<Vec<Row>, String> {
for op in inner {
match op {
PlanOp::ScanLabel { var, label } => {
rows = scan_label(view, vars, &rows, var, label.as_deref())?;
}
PlanOp::ScanKey { var, key, label } => {
rows = scan_key(view, vars, &rows, var, key, label.as_deref(), params)?;
}
PlanOp::LookupProps { var, props } => {
rows = retain_node(view, vars, &rows, var, None, props, params)?;
}
PlanOp::JoinBound { var, label, props } => {
rows = retain_node(view, vars, &rows, var, label.as_deref(), props, params)?;
}
PlanOp::Expand { .. } => {
rows = exec_expand(view, vars, &rows, op, params)?;
}
PlanOp::Filter { expr } => {
rows = exec_filter(view, vars, &rows, expr, params)?;
}
other => {
return Err(format!(
"unsupported op inside OPTIONAL MATCH inner plan: {other:?}"
));
}
}
}
Ok(rows)
}
fn finish(table: Projected) -> ResultSet {
let mut rs = ResultSet::new(table.columns);
for row in table.rows {
rs.push_row(row);
}
rs
}
fn check_params(plan: &[PlanOp], params: &Params) -> Result<(), String> {
let mut names = Vec::new();
let mut seen = BTreeSet::new();
collect_params_from_ops(plan, &mut names, &mut seen)?;
for name in names {
if !params.0.contains_key(&name) {
return Err(format!("missing parameter `{name}`"));
}
}
Ok(())
}
fn collect_params_from_ops(
plan: &[PlanOp],
names: &mut Vec<String>,
seen: &mut BTreeSet<String>,
) -> Result<(), String> {
for op in plan {
match op {
PlanOp::ScanKey { key, .. } => collect_operand(key, names, seen),
PlanOp::LookupProps { props, .. }
| PlanOp::JoinBound { props, .. }
| PlanOp::Expand {
to_props: props, ..
} => {
for (_, operand) in props {
collect_operand(operand, names, seen);
}
}
PlanOp::Filter { expr } => collect_expr(expr, names, seen, 0)?,
PlanOp::LeftOuterApply { inner, .. } => {
collect_params_from_ops(inner, names, seen)?;
}
PlanOp::Project { items } => {
for item in items {
collect_ret_item_params(item, names, seen);
}
}
PlanOp::With {
items, where_expr, ..
} => {
for item in items {
collect_ret_item_params(item, names, seen);
}
if let Some(expr) = where_expr {
let _ = collect_expr(expr, names, seen, 0);
}
}
PlanOp::GroupAggregate { keys, aggs } => {
for (_, item) in keys {
collect_ret_item_params(item, names, seen);
}
for (_, arg, _) in aggs {
match arg {
AggArg::Star => {}
AggArg::Var(_) => {}
AggArg::Prop { .. } => {}
AggArg::Distinct(_) => {}
}
}
}
PlanOp::Skip(LimitSkip::Param(n)) | PlanOp::Limit(LimitSkip::Param(n))
if seen.insert(n.clone()) =>
{
names.push(n.clone());
}
_ => {}
}
}
Ok(())
}
fn collect_ret_item_params(item: &RetItem, names: &mut Vec<String>, seen: &mut BTreeSet<String>) {
match &item.value {
RetVal::FuncCall { args, .. } => {
for arg in args {
collect_operand(arg, names, seen);
}
}
RetVal::ScalarExpr(op) => {
collect_operand(op, names, seen);
}
RetVal::Prop { .. } | RetVal::Var(_) | RetVal::Agg { .. } => {}
}
}
fn collect_operand(op: &Operand, names: &mut Vec<String>, seen: &mut BTreeSet<String>) {
match op {
Operand::Param(n) => {
if seen.insert(n.clone()) {
names.push(n.clone());
}
}
Operand::FuncCall { args, .. } => {
for arg in args {
collect_operand(arg, names, seen);
}
}
Operand::BinArith { left, right, .. } => {
collect_operand(left, names, seen);
collect_operand(right, names, seen);
}
_ => {}
}
}
fn collect_expr(
expr: &Expr,
names: &mut Vec<String>,
seen: &mut BTreeSet<String>,
depth: u32,
) -> Result<(), String> {
if depth > 256 {
return Err("expression nesting too deep".into());
}
match expr {
Expr::And(lhs, rhs) | Expr::Or(lhs, rhs) => {
collect_expr(lhs, names, seen, depth + 1)?;
collect_expr(rhs, names, seen, depth + 1)
}
Expr::Not(inner) => collect_expr(inner, names, seen, depth + 1),
Expr::Cmp { lhs, rhs, .. } => {
collect_operand(lhs, names, seen);
collect_operand(rhs, names, seen);
Ok(())
}
Expr::Truthy(op) => {
collect_operand(op, names, seen);
Ok(())
}
Expr::IsNull(op) | Expr::IsNotNull(op) => {
collect_operand(op, names, seen);
Ok(())
}
Expr::In { expr, list } => {
collect_operand(expr, names, seen);
for item in list {
collect_operand(item, names, seen);
}
Ok(())
}
}
}
fn collect_vars(plan: &[PlanOp]) -> VarTable {
let mut vars = VarTable { names: Vec::new() };
for op in plan {
match op {
PlanOp::ScanLabel { var, .. } => {
vars.intern(var);
}
PlanOp::ScanKey { var, key, .. } => {
vars.intern(var);
intern_operand(&mut vars, key);
}
PlanOp::IndexScan { var, value, .. } => {
vars.intern(var);
intern_operand(&mut vars, value);
}
PlanOp::IndexIntersect {
var, equalities, ..
} => {
vars.intern(var);
for (_, operand) in equalities {
intern_operand(&mut vars, operand);
}
}
PlanOp::LookupProps { var, props } | PlanOp::JoinBound { var, props, .. } => {
vars.intern(var);
for (_, operand) in props {
intern_operand(&mut vars, operand);
}
}
PlanOp::Expand {
from,
rel_var,
to,
to_props,
..
} => {
vars.intern(from);
vars.intern(to);
if let Some(r) = rel_var {
vars.intern(r);
}
for (_, operand) in to_props {
intern_operand(&mut vars, operand);
}
}
PlanOp::VarExpand {
from, rel_var, to, ..
} => {
vars.intern(from);
vars.intern(to);
if let Some(r) = rel_var {
vars.intern(r);
}
}
PlanOp::ShortestPath {
from, rel_var, to, ..
} => {
vars.intern(from);
vars.intern(to);
if let Some(r) = rel_var {
vars.intern(r);
}
}
PlanOp::Filter { expr } => intern_expr(&mut vars, expr),
PlanOp::Project { items } => {
for item in items {
match &item.value {
RetVal::Var(name) | RetVal::Prop { var: name, .. } => {
vars.intern(name);
}
RetVal::Agg { .. } => {} RetVal::FuncCall { args, .. } => {
for arg in args {
intern_operand(&mut vars, arg);
}
}
RetVal::ScalarExpr(op) => {
intern_operand(&mut vars, op);
}
}
}
}
PlanOp::Aggregate { arg, .. } => intern_agg_arg(&mut vars, arg),
PlanOp::GroupAggregate { keys, aggs } => {
for (col, item) in keys {
match &item.value {
RetVal::Var(name) | RetVal::Prop { var: name, .. } => {
vars.intern(name);
}
RetVal::Agg { .. } => {}
RetVal::FuncCall { args, .. } => {
for arg in args {
intern_operand(&mut vars, arg);
}
}
RetVal::ScalarExpr(op) => {
intern_operand(&mut vars, op);
}
}
vars.intern(col);
}
for (_, arg, col) in aggs {
intern_agg_arg(&mut vars, arg);
vars.intern(col);
}
}
PlanOp::With {
items,
where_expr,
order_by,
..
} => {
for item in items {
match &item.value {
RetVal::Var(name) | RetVal::Prop { var: name, .. } => {
vars.intern(name);
}
RetVal::Agg { .. } => {}
RetVal::FuncCall { args, .. } => {
for arg in args {
intern_operand(&mut vars, arg);
}
}
RetVal::ScalarExpr(op) => {
intern_operand(&mut vars, op);
}
}
if let Some(alias) = &item.alias {
vars.intern(alias);
} else if let Some(col) = ret_val_label(&item.value) {
vars.intern(&col);
}
}
if let Some(expr) = where_expr {
intern_expr(&mut vars, expr);
}
for oi in order_by {
match &oi.target {
OrderTarget::Alias(name) | OrderTarget::Var(name) => {
vars.intern(name);
}
OrderTarget::Prop { var, .. } => {
vars.intern(var);
}
}
}
}
PlanOp::Unwind { expr, alias } => {
vars.intern(alias);
match expr {
UnwindExpr::Prop { var, .. } => {
vars.intern(var);
}
UnwindExpr::Var(name) => {
vars.intern(name);
}
UnwindExpr::Lit(_) => {}
}
}
PlanOp::LeftOuterApply {
inner,
optional_vars,
} => {
for op in inner {
match op {
PlanOp::ScanLabel { var, .. } => {
vars.intern(var);
}
PlanOp::ScanKey { var, key, .. } => {
vars.intern(var);
intern_operand(&mut vars, key);
}
PlanOp::IndexScan { var, value, .. } => {
vars.intern(var);
intern_operand(&mut vars, value);
}
PlanOp::IndexIntersect {
var, equalities, ..
} => {
vars.intern(var);
for (_, operand) in equalities {
intern_operand(&mut vars, operand);
}
}
PlanOp::Expand {
from, rel_var, to, ..
} => {
vars.intern(from);
vars.intern(to);
if let Some(r) = rel_var {
vars.intern(r);
}
}
PlanOp::JoinBound { var, .. } | PlanOp::LookupProps { var, .. } => {
vars.intern(var);
}
PlanOp::Filter { expr } => intern_expr(&mut vars, expr),
_ => {}
}
}
for v in optional_vars {
vars.intern(v);
}
}
_ => {}
}
}
vars
}
fn intern_operand(vars: &mut VarTable, operand: &Operand) {
match operand {
Operand::Prop { var, .. } | Operand::Var(var) => {
vars.intern(var);
}
Operand::Lit(_) | Operand::Param(_) => {}
Operand::BinArith { left, right, .. } => {
intern_operand(vars, left);
intern_operand(vars, right);
}
Operand::FuncCall { args, .. } => {
for arg in args {
intern_operand(vars, arg);
}
}
Operand::Case { branches, default } => {
for (cond, value) in branches {
intern_expr(vars, cond);
intern_operand(vars, value);
}
if let Some(d) = default {
intern_operand(vars, d);
}
}
Operand::Index { base, index } => {
intern_operand(vars, base);
intern_operand(vars, index);
}
}
}
fn intern_agg_arg(vars: &mut VarTable, arg: &AggArg) {
match arg {
AggArg::Star => {}
AggArg::Var(v) => {
vars.intern(v);
}
AggArg::Prop { var, .. } => {
vars.intern(var);
}
AggArg::Distinct(inner) => intern_agg_arg(vars, inner),
}
}
fn agg_arg_label(arg: &AggArg) -> String {
match arg {
AggArg::Star => "*".to_string(),
AggArg::Var(v) => v.clone(),
AggArg::Prop { var, field } => format!("{var}.{field}"),
AggArg::Distinct(inner) => format!("DISTINCT {}", agg_arg_label(inner)),
}
}
fn intern_expr(vars: &mut VarTable, expr: &Expr) {
match expr {
Expr::And(lhs, rhs) | Expr::Or(lhs, rhs) => {
intern_expr(vars, lhs);
intern_expr(vars, rhs);
}
Expr::Not(inner) => intern_expr(vars, inner),
Expr::Cmp { lhs, rhs, .. } => {
intern_operand(vars, lhs);
intern_operand(vars, rhs);
}
Expr::Truthy(op) => intern_operand(vars, op),
Expr::IsNull(op) | Expr::IsNotNull(op) => intern_operand(vars, op),
Expr::In { expr, list } => {
intern_operand(vars, expr);
for item in list {
intern_operand(vars, item);
}
}
}
}
fn scan_ids(view: &GraphView, label: Option<&str>) -> Vec<u32> {
let ids = match label {
Some(label) => view.nodes_with_label(label),
None => (0..view.ids.len() as u32)
.filter(|&id| view.label_of(id).is_some())
.collect(),
};
if view.mask.is_some() {
ids.into_iter().filter(|&id| view.visible(id)).collect()
} else {
ids
}
}
fn resolve_scan_key_id(
view: &GraphView,
vars: &VarTable,
row: &Row,
key: &Operand,
label: Option<&str>,
params: &Params,
) -> Result<Option<u32>, String> {
#[cfg(test)]
SCAN_KEY_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let Some(val) = resolve_operand(view, vars, row, key, params)? else {
return Ok(None);
};
let Value::Str(s) = val else {
return Ok(None);
};
let Some(id) = view.node_id(&s) else {
return Ok(None);
};
if !view.visible(id) {
return Ok(None);
}
if let Some(want) = label {
match view.label_of(id) {
Some(got) if got == want => {}
_ => return Ok(None),
}
}
Ok(Some(id))
}
fn scan_key(
view: &GraphView,
vars: &VarTable,
rows: &[Row],
var: &str,
key: &Operand,
label: Option<&str>,
params: &Params,
) -> Result<Vec<Row>, String> {
let slot = vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
let cap = max_intermediate_rows();
let mut out = Vec::new();
for row in rows {
let Some(id) = resolve_scan_key_id(view, vars, row, key, label, params)? else {
continue;
};
if out.len() >= cap {
return Err(row_cap_err(cap));
}
let mut next = row.clone();
next[slot] = Some(Cell::Node(id));
out.push(next);
}
Ok(out)
}
fn scan_label(
view: &GraphView,
vars: &VarTable,
rows: &[Row],
var: &str,
label: Option<&str>,
) -> Result<Vec<Row>, String> {
let ids = scan_ids(view, label);
let slot = vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
let cap = max_intermediate_rows();
let mut out = Vec::with_capacity(rows.len().saturating_mul(ids.len()).min(cap));
for row in rows {
for &id in &ids {
if out.len() >= cap {
return Err(row_cap_err(cap));
}
let mut next = row.clone();
next[slot] = Some(Cell::Node(id));
out.push(next);
}
}
Ok(out)
}
#[allow(clippy::too_many_arguments)]
fn index_scan_ids(
view: &GraphView,
vars: &VarTable,
row: &Row,
label: Option<&str>,
field: &str,
value: &Operand,
params: &Params,
) -> Result<Vec<u32>, String> {
if is_identity_eq_field(field) {
return identity_eq_ids(view, vars, row, label, field, value, params);
}
let resolved = resolve_operand(view, vars, row, value, params)?;
if let (Some(label_str), Some(val)) = (label, resolved.as_ref()) {
if let Some(ids) = view.nodes_with_prop(label_str, field, val) {
#[cfg(test)]
INDEX_SCAN_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
return Ok(ids);
}
}
let props = [(field.to_string(), value.clone())];
let mut out = Vec::new();
for id in scan_ids(view, label) {
if node_matches(view, vars, row, id, None, &props, params)? {
out.push(id);
}
}
Ok(out)
}
#[allow(clippy::too_many_arguments)]
fn identity_eq_ids(
view: &GraphView,
vars: &VarTable,
row: &Row,
label: Option<&str>,
field: &str,
value: &Operand,
params: &Params,
) -> Result<Vec<u32>, String> {
let resolved = resolve_operand(view, vars, row, value, params)?;
let mut out = Vec::new();
let mut seen = HashSet::new();
if let Some(id) = resolve_scan_key_id(view, vars, row, value, label, params)? {
let props = [(field.to_string(), value.clone())];
if node_matches(view, vars, row, id, None, &props, params)? {
seen.insert(id);
out.push(id);
}
}
if let Some(val) = resolved.as_ref() {
for id in stored_identity_hits(view, label, field, val) {
if seen.insert(id) {
out.push(id);
}
}
}
Ok(out)
}
fn stored_identity_hits(
view: &GraphView,
label: Option<&str>,
field: &str,
val: &Value,
) -> Vec<u32> {
if let Some(label_str) = label {
if let Some(ids) = view.nodes_with_prop(label_str, field, val) {
return ids;
}
}
let mut out = Vec::new();
for id in scan_ids(view, label) {
if let Some(got) = view.prop(id, field).map(|vr| vr.into_value()) {
if values_equal(&got, val) {
out.push(id);
}
}
}
out
}
#[allow(clippy::too_many_arguments)]
fn scan_index(
view: &GraphView,
vars: &VarTable,
rows: &[Row],
var: &str,
label: Option<&str>,
field: &str,
value: &Operand,
params: &Params,
) -> Result<Vec<Row>, String> {
let Some(first) = rows.first() else {
return Ok(Vec::new());
};
let ids = index_scan_ids(view, vars, first, label, field, value, params)?;
let slot = vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
let cap = max_intermediate_rows();
let mut out = Vec::new();
for row in rows {
for &id in &ids {
if out.len() >= cap {
return Err(row_cap_err(cap));
}
let mut next = row.clone();
next[slot] = Some(Cell::Node(id));
out.push(next);
}
}
Ok(out)
}
#[allow(clippy::too_many_arguments)]
fn index_intersect_ids(
view: &GraphView,
vars: &VarTable,
row: &Row,
label: Option<&str>,
equalities: &[(String, Operand)],
params: &Params,
) -> Result<Vec<u32>, String> {
let mut resolved: Vec<(String, Option<Value>)> = Vec::with_capacity(equalities.len());
for (field, operand) in equalities {
let val = resolve_operand(view, vars, row, operand, params)?;
resolved.push((field.clone(), val));
}
let mut indexed_lists: Vec<Vec<u32>> = Vec::new();
let mut unindexed_props: Vec<(String, Operand)> = Vec::new();
for ((field, val_opt), (_, operand)) in resolved.iter().zip(equalities.iter()) {
if let (Some(label_str), Some(val)) = (label, val_opt.as_ref()) {
if let Some(ids) = view.nodes_with_prop(label_str, field, val) {
indexed_lists.push(ids);
continue;
}
}
unindexed_props.push((field.clone(), operand.clone()));
}
if indexed_lists.is_empty() {
let mut out = Vec::new();
for id in scan_ids(view, label) {
if node_matches(view, vars, row, id, None, equalities, params)? {
out.push(id);
}
}
return Ok(out);
}
#[cfg(test)]
INDEX_INTERSECT_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
indexed_lists.sort_unstable_by_key(|v| v.len());
let mut result = indexed_lists.remove(0);
for other in indexed_lists {
let mut merged = Vec::new();
let (mut i, mut j) = (0, 0);
while i < result.len() && j < other.len() {
match result[i].cmp(&other[j]) {
std::cmp::Ordering::Equal => {
merged.push(result[i]);
i += 1;
j += 1;
}
std::cmp::Ordering::Less => i += 1,
std::cmp::Ordering::Greater => j += 1,
}
}
result = merged;
}
if !unindexed_props.is_empty() {
let mut filtered = Vec::with_capacity(result.len());
for id in result {
if node_matches(view, vars, row, id, None, &unindexed_props, params)? {
filtered.push(id);
}
}
result = filtered;
}
Ok(result)
}
#[allow(clippy::too_many_arguments)]
fn scan_intersect(
view: &GraphView,
vars: &VarTable,
rows: &[Row],
var: &str,
label: Option<&str>,
equalities: &[(String, Operand)],
params: &Params,
) -> Result<Vec<Row>, String> {
let Some(first) = rows.first() else {
return Ok(Vec::new());
};
let ids = index_intersect_ids(view, vars, first, label, equalities, params)?;
let slot = vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
let cap = max_intermediate_rows();
let mut out = Vec::new();
for row in rows {
for &id in &ids {
if out.len() >= cap {
return Err(row_cap_err(cap));
}
let mut next = row.clone();
next[slot] = Some(Cell::Node(id));
out.push(next);
}
}
Ok(out)
}
fn require_cell<'a>(row: &'a Row, vars: &VarTable, var: &str) -> Result<&'a Cell, String> {
let slot = vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
row.get(slot)
.and_then(|c| c.as_ref())
.ok_or_else(|| format!("unbound variable `{var}`"))
}
fn require_node(row: &Row, vars: &VarTable, var: &str) -> Result<u32, String> {
match require_cell(row, vars, var)? {
Cell::Node(id) => Ok(*id),
Cell::Rel(_) => Err(format!("variable `{var}` is not a node")),
Cell::Path(_) => Err(format!("variable `{var}` is a path, not a node")),
Cell::Scalar(_) => Err(format!("variable `{var}` is a scalar value, not a node")),
}
}
const SCALAR_FUNCS: &[&str] = &[
"toLower",
"toUpper",
"size",
"coalesce",
"type",
"abs",
"round",
"textMatches",
"contains",
"startsWith",
"endsWith",
"toInteger",
"toFloat",
"toString",
"decay",
"key",
"id",
"labels",
];
fn eval_string_predicate(
name: &str,
args: &[Operand],
view: &GraphView,
vars: &VarTable,
row: &Row,
params: &Params,
f: impl Fn(&str, &str) -> bool,
) -> Result<Option<Value>, String> {
if args.len() != 2 {
return Err(format!(
"{name}() requires exactly 2 arguments, got {}",
args.len()
));
}
let a = resolve_operand(view, vars, row, &args[0], params)?;
let b = resolve_operand(view, vars, row, &args[1], params)?;
match (a, b) {
(Some(Value::Str(s)), Some(Value::Str(sub))) => Ok(Some(Value::Bool(f(&s, &sub)))),
(None, _) | (_, None) => Ok(None),
_ => Ok(None),
}
}
fn eval_func(
name: &str,
args: &[Operand],
view: &GraphView,
vars: &VarTable,
row: &Row,
params: &Params,
) -> Result<Option<Value>, String> {
let norm = name.to_ascii_lowercase();
match norm.as_str() {
"tolower" => {
if args.len() != 1 {
return Err(format!(
"toLower() requires exactly 1 argument, got {}",
args.len()
));
}
let v = resolve_operand(view, vars, row, &args[0], params)?;
Ok(v.map(|val| match val {
Value::Str(s) => Value::Str(s.to_ascii_lowercase()),
other => other, }))
}
"toupper" => {
if args.len() != 1 {
return Err(format!(
"toUpper() requires exactly 1 argument, got {}",
args.len()
));
}
let v = resolve_operand(view, vars, row, &args[0], params)?;
Ok(v.map(|val| match val {
Value::Str(s) => Value::Str(s.to_ascii_uppercase()),
other => other,
}))
}
"size" => {
if args.len() != 1 {
return Err(format!(
"size() requires exactly 1 argument, got {}",
args.len()
));
}
let v = resolve_operand(view, vars, row, &args[0], params)?;
match v {
None => Ok(None), Some(Value::Str(s)) => Ok(Some(Value::Int(s.len() as i64))),
Some(Value::List(items)) => Ok(Some(Value::Int(items.len() as i64))),
Some(_) => Ok(None), }
}
"coalesce" => {
for arg in args {
if let Some(v) = resolve_operand(view, vars, row, arg, params)? {
return Ok(Some(v));
}
}
Ok(None)
}
"type" => {
if args.len() != 1 {
return Err(format!(
"type() requires exactly 1 argument, got {}",
args.len()
));
}
let Operand::Var(var_name) = &args[0] else {
return Err(
"type() argument must be a relationship variable (e.g. type(r))".to_string(),
);
};
let slot = vars
.slot(var_name)
.ok_or_else(|| format!("unbound variable `{var_name}` in type()"))?;
match row.get(slot).and_then(|c| c.as_ref()) {
Some(Cell::Rel(e)) => {
let etype = view.syms.resolve(e.etype).unwrap_or("").to_owned();
Ok(Some(Value::Str(etype)))
}
Some(Cell::Node(_)) => Err(format!(
"type() argument `{var_name}` is a node, not a relationship"
)),
Some(Cell::Scalar(_) | Cell::Path(_)) => Err(format!(
"type() argument `{var_name}` is not a relationship"
)),
None => Ok(None), }
}
"key" | "id" => {
let fname = if norm == "id" { "id" } else { "key" };
if args.len() != 1 {
return Err(format!(
"{fname}() requires exactly 1 argument, got {}",
args.len()
));
}
let Operand::Var(var_name) = &args[0] else {
return Err(format!(
"{fname}() argument must be a node variable (e.g. {fname}(n))"
));
};
let slot = vars
.slot(var_name)
.ok_or_else(|| format!("unbound variable `{var_name}` in {fname}()"))?;
match row.get(slot).and_then(|c| c.as_ref()) {
Some(Cell::Node(id)) => Ok(Some(Value::Str(view.key_of(*id).to_owned()))),
Some(Cell::Rel(_)) => Err(format!(
"{fname}() argument `{var_name}` is a relationship, not a node"
)),
Some(Cell::Scalar(_) | Cell::Path(_)) => {
Err(format!("{fname}() argument `{var_name}` is not a node"))
}
None => Ok(None), }
}
"labels" => {
if args.len() != 1 {
return Err(format!(
"labels() requires exactly 1 argument, got {}",
args.len()
));
}
let Operand::Var(var_name) = &args[0] else {
return Err(
"labels() argument must be a node variable (e.g. labels(n))".to_string()
);
};
let slot = vars
.slot(var_name)
.ok_or_else(|| format!("unbound variable `{var_name}` in labels()"))?;
match row.get(slot).and_then(|c| c.as_ref()) {
Some(Cell::Node(id)) => Ok(Some(Value::List(
view.label_of(*id)
.map(|l| vec![Value::Str(l.to_owned())])
.unwrap_or_default(),
))),
Some(Cell::Rel(_)) => Err(format!(
"labels() argument `{var_name}` is a relationship, not a node"
)),
Some(Cell::Scalar(_) | Cell::Path(_)) => {
Err(format!("labels() argument `{var_name}` is not a node"))
}
None => Ok(None), }
}
"abs" => {
if args.len() != 1 {
return Err(format!(
"abs() requires exactly 1 argument, got {}",
args.len()
));
}
let v = resolve_operand(view, vars, row, &args[0], params)?;
match v {
None => Ok(None),
Some(Value::Int(n)) => Ok(Some(Value::Int(n.abs()))),
Some(Value::Float(f)) => Ok(Some(Value::Float(f.abs()))),
Some(_) => Ok(None), }
}
"round" => {
if args.len() != 1 {
return Err(format!(
"round() requires exactly 1 argument, got {}",
args.len()
));
}
let v = resolve_operand(view, vars, row, &args[0], params)?;
match v {
None => Ok(None),
Some(Value::Int(n)) => Ok(Some(Value::Int(n))), Some(Value::Float(f)) => Ok(Some(Value::Float(f.round()))),
Some(_) => Ok(None), }
}
"textmatches" => {
if args.len() != 2 {
return Err(format!(
"textMatches() requires exactly 2 arguments (field_value, query), got {}",
args.len()
));
}
let field_val = resolve_operand(view, vars, row, &args[0], params)?;
let query_val = resolve_operand(view, vars, row, &args[1], params)?;
match (field_val, query_val) {
(None, _) | (_, None) => Ok(None),
(Some(Value::Str(s)), Some(Value::Str(q))) => {
Ok(Some(Value::Bool(core_storage::fulltext::eval_query_str(
&s, &q,
))))
}
(Some(Value::List(items)), Some(Value::Str(q))) => Ok(Some(Value::Bool(
core_storage::fulltext::eval_query_str_list(&items, &q),
))),
_ => Ok(Some(Value::Bool(false))), }
}
"contains" => eval_string_predicate("contains", args, view, vars, row, params, |s, sub| {
s.contains(sub)
}),
"startswith" => {
eval_string_predicate("startsWith", args, view, vars, row, params, |s, p| {
s.starts_with(p)
})
}
"endswith" => eval_string_predicate("endsWith", args, view, vars, row, params, |s, p| {
s.ends_with(p)
}),
"tointeger" => {
if args.len() != 1 {
return Err(format!(
"toInteger() requires exactly 1 argument, got {}",
args.len()
));
}
let v = resolve_operand(view, vars, row, &args[0], params)?;
Ok(match v {
None => None,
Some(Value::Int(n)) => Some(Value::Int(n)),
Some(Value::Float(f)) => Some(Value::Int(f.trunc() as i64)),
Some(Value::Str(s)) => s
.trim()
.parse::<i64>()
.ok()
.or_else(|| s.trim().parse::<f64>().ok().map(|f| f.trunc() as i64))
.map(Value::Int),
Some(_) => None,
})
}
"tofloat" => {
if args.len() != 1 {
return Err(format!(
"toFloat() requires exactly 1 argument, got {}",
args.len()
));
}
let v = resolve_operand(view, vars, row, &args[0], params)?;
Ok(match v {
None => None,
Some(Value::Float(f)) => Some(Value::Float(f)),
Some(Value::Int(n)) => Some(Value::Float(n as f64)),
Some(Value::Str(s)) => s.trim().parse::<f64>().ok().map(Value::Float),
Some(_) => None,
})
}
"tostring" => {
if args.len() != 1 {
return Err(format!(
"toString() requires exactly 1 argument, got {}",
args.len()
));
}
let v = resolve_operand(view, vars, row, &args[0], params)?;
Ok(match v {
None => None,
Some(Value::Str(s)) => Some(Value::Str(s)),
Some(Value::Int(n)) => Some(Value::Str(n.to_string())),
Some(Value::Float(f)) => Some(Value::Str(f.to_string())),
Some(Value::Bool(b)) => Some(Value::Str(b.to_string())),
Some(_) => None, })
}
"decay" => {
if args.len() != 3 {
return Err(format!(
"decay() requires exactly 3 arguments, got {}",
args.len()
));
}
let base = resolve_operand(view, vars, row, &args[0], params)?;
let age = resolve_operand(view, vars, row, &args[1], params)?;
let halflife = resolve_operand(view, vars, row, &args[2], params)?;
match (base, age, halflife) {
(None, _, _) | (_, None, _) | (_, _, None) => Ok(None),
(Some(b), Some(a), Some(h)) => {
let b = numeric_val(&b)
.ok_or_else(|| "decay() requires numeric arguments".to_string())?;
let a = numeric_val(&a)
.ok_or_else(|| "decay() requires numeric arguments".to_string())?;
let h = numeric_val(&h)
.ok_or_else(|| "decay() requires numeric arguments".to_string())?;
if h <= 0.0 {
return Err("decay() requires halflife > 0".to_string());
}
Ok(Some(Value::Float(b * 0.5f64.powf(a / h))))
}
}
}
_ => Err(format!(
"unknown function `{name}`; supported: {}",
SCALAR_FUNCS.join(", ")
)),
}
}
fn resolve_operand(
view: &GraphView,
vars: &VarTable,
row: &Row,
operand: &Operand,
params: &Params,
) -> Result<Option<Value>, String> {
match operand {
Operand::Lit(v) => Ok(Some(v.clone())),
Operand::Param(name) => match params.0.get(name) {
Some(v) => Ok(Some(v.clone())),
None => Err(format!("missing parameter `{name}`")),
},
Operand::Prop { var, field } => resolve_prop(view, vars, row, var, field),
Operand::Var(name) => {
match vars
.slot(name)
.and_then(|s| row.get(s))
.and_then(|c| c.as_ref())
{
Some(Cell::Scalar(v)) => Ok(Some(v.clone())),
Some(Cell::Node(id)) => match view.ids.key_of(*id) {
Some(key) => Ok(Some(Value::Str(key.to_owned()))),
None => Ok(None),
},
Some(Cell::Path(hops)) => Ok(Some(Value::Int(*hops as i64))),
Some(Cell::Rel(_)) => Err(format!("variable `{name}` is a relationship")),
None => Ok(None),
}
}
Operand::FuncCall { name, args } => eval_func(name, args, view, vars, row, params),
Operand::Index { base, index } => {
let base_val = resolve_operand(view, vars, row, base, params)?;
let idx_val = resolve_operand(view, vars, row, index, params)?;
Ok(crate::value_ops::index_list(base_val, idx_val))
}
Operand::Case { branches, default } => {
for (cond, value) in branches {
if eval_expr(view, vars, row, cond, params, 0)? {
return resolve_operand(view, vars, row, value, params);
}
}
match default {
Some(d) => resolve_operand(view, vars, row, d, params),
None => Ok(None),
}
}
Operand::BinArith { op, left, right } => {
use super::ast::ArithOp;
let lv = resolve_operand(view, vars, row, left, params)?;
let rv = resolve_operand(view, vars, row, right, params)?;
match (lv, rv) {
(None, _) | (_, None) => Ok(None), (Some(Value::Int(a)), Some(Value::Int(b))) => {
let result = match op {
ArithOp::Sub => a.saturating_sub(b),
ArithOp::Mul => a.saturating_mul(b),
ArithOp::Add => a.saturating_add(b),
ArithOp::Div => {
if b == 0 {
return Err("division by zero".into());
}
a.checked_div(b).unwrap_or(i64::MAX)
}
};
Ok(Some(Value::Int(result)))
}
(Some(lv), Some(rv)) => {
let a = match &lv {
Value::Float(f) => *f,
Value::Int(i) => *i as f64,
_ => return Err(format!("arithmetic operand must be numeric, got {lv:?}")),
};
let b = match &rv {
Value::Float(f) => *f,
Value::Int(i) => *i as f64,
_ => return Err(format!("arithmetic operand must be numeric, got {rv:?}")),
};
let result = match op {
ArithOp::Sub => a - b,
ArithOp::Mul => a * b,
ArithOp::Add => a + b,
ArithOp::Div => {
if b == 0.0 {
return Err("division by zero".into());
}
a / b
}
};
Ok(Some(Value::Float(result)))
}
}
}
}
}
fn is_identity_field(field: &str) -> bool {
field == "key" || field == "id" || field == "label"
}
fn is_identity_eq_field(field: &str) -> bool {
field == "key" || field == "id"
}
fn node_identity_prop(view: &GraphView, id: u32, field: &str) -> Option<Value> {
match field {
"key" | "id" => view.ids.key_of(id).map(|k| Value::Str(k.to_owned())),
"label" => view.label_of(id).map(|l| Value::Str(l.to_owned())),
_ => None,
}
}
fn resolve_prop(
view: &GraphView,
vars: &VarTable,
row: &Row,
var: &str,
field: &str,
) -> Result<Option<Value>, String> {
let slot = vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
let cell = match row.get(slot).and_then(|c| c.as_ref()) {
Some(c) => c,
None => return Ok(None),
};
match cell {
Cell::Node(id) => Ok(view
.prop(*id, field)
.map(|vr| vr.into_value())
.or_else(|| node_identity_prop(view, *id, field))),
Cell::Rel(e) => Ok(view.edge_props.get(e.etype, e.src, e.dst, field)),
Cell::Path(hops) => {
if field == "length" {
Ok(Some(Value::Int(*hops as i64)))
} else {
Ok(None)
}
}
Cell::Scalar(_) => Ok(None),
}
}
fn node_matches(
view: &GraphView,
vars: &VarTable,
row: &Row,
id: u32,
label: Option<&str>,
props: &[(String, Operand)],
params: &Params,
) -> Result<bool, String> {
if let Some(want) = label {
match view.label_of(id) {
Some(got) if got == want => {}
_ => return Ok(false),
}
}
for (field, operand) in props {
let Some(expected) = resolve_operand(view, vars, row, operand, params)? else {
return Ok(false);
};
let got = view
.prop(id, field)
.map(|vr| vr.into_value())
.or_else(|| node_identity_prop(view, id, field));
match got {
Some(got) if values_equal(&got, &expected) => {}
_ => return Ok(false),
}
}
Ok(true)
}
fn retain_node(
view: &GraphView,
vars: &VarTable,
rows: &[Row],
var: &str,
label: Option<&str>,
props: &[(String, Operand)],
params: &Params,
) -> Result<Vec<Row>, String> {
let mut out = Vec::with_capacity(rows.len());
for row in rows {
let id = require_node(row, vars, var)?;
if node_matches(view, vars, row, id, label, props, params)? {
out.push(row.clone());
}
}
Ok(out)
}
fn map_dir(dir: RelDir) -> Dir {
match dir {
RelDir::Right => Dir::Out,
RelDir::Left => Dir::In,
RelDir::Undirected => Dir::Both,
}
}
fn neighbor(from: u32, e: &EdgeRef, dir: RelDir) -> u32 {
match dir {
RelDir::Right => e.dst,
RelDir::Left => e.src,
RelDir::Undirected => {
if e.src == from {
e.dst
} else {
e.src
}
}
}
}
fn row_has_edge(row: &Row, e: &EdgeRef) -> bool {
row.iter()
.any(|c| matches!(c, Some(Cell::Rel(existing)) if existing == e))
}
fn resolve_etypes(view: &GraphView, etypes: &[String]) -> Option<Vec<u32>> {
if etypes.is_empty() {
None } else {
Some(etypes.iter().filter_map(|n| view.syms.get(n)).collect())
}
}
fn exec_expand(
view: &GraphView,
vars: &VarTable,
rows: &[Row],
op: &PlanOp,
params: &Params,
) -> Result<Vec<Row>, String> {
let PlanOp::Expand {
from,
rel_var,
etypes,
dir,
to,
to_label,
to_props,
} = op
else {
return Err("internal: expected Expand".into());
};
let etypes = resolve_etypes(view, etypes);
let exp_dir = map_dir(*dir);
let to_slot = vars
.slot(to)
.ok_or_else(|| format!("unbound variable `{to}`"))?;
let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
let cap = max_intermediate_rows();
let mut out = Vec::with_capacity(rows.len().saturating_mul(2).min(cap));
for row in rows {
let from_id = require_node(row, vars, from)?;
let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
Some(Cell::Node(id)) => Some(*id),
Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
return Err(format!("variable `{to}` is not a node"))
}
None => None,
};
for e in expand(view, from_id, etypes.as_deref(), exp_dir) {
if row_has_edge(row, &e) {
continue;
}
let nbr = neighbor(from_id, &e, *dir);
if !view.visible(nbr) {
continue;
}
if let Some(want) = bound_to {
if nbr != want {
continue;
}
}
if !node_matches(view, vars, row, nbr, to_label.as_deref(), to_props, params)? {
continue;
}
if out.len() >= cap {
return Err(row_cap_err(cap));
}
let mut next = row.clone();
if let Some(slot) = rel_slot {
next[slot] = Some(Cell::Rel(e));
}
if bound_to.is_none() {
next[to_slot] = Some(Cell::Node(nbr));
}
out.push(next);
#[cfg(test)]
record_expand_row();
}
}
Ok(out)
}
#[allow(clippy::too_many_arguments)]
fn exec_var_expand(
view: &GraphView,
vars: &VarTable,
rows: &[Row],
from: &str,
rel_var: &Option<String>,
etypes: &[String],
dir: RelDir,
to: &str,
min: u8,
max: u8,
) -> Result<Vec<Row>, String> {
let etypes = resolve_etypes(view, etypes);
let exp_dir = map_dir(dir);
let to_slot = vars
.slot(to)
.ok_or_else(|| format!("unbound variable `{to}`"))?;
let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
let cap = max_intermediate_rows();
let mut out: Vec<Row> = Vec::new();
for row in rows {
let from_id = require_node(row, vars, from)?;
let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
Some(Cell::Node(id)) => Some(*id),
Some(_) => return Err(format!("variable `{to}` is not a node")),
None => None,
};
struct PathState {
node: u32,
edges: Vec<EdgeRef>,
}
let mut frontier: Vec<PathState> = vec![PathState {
node: from_id,
edges: Vec::new(),
}];
let mut frontier_count: usize = 0;
for depth in 1u8..=max {
let mut next_frontier: Vec<PathState> = Vec::new();
for state in &frontier {
for e in expand(view, state.node, etypes.as_deref(), exp_dir) {
if state.edges.contains(&e) {
continue;
}
let nbr = neighbor(state.node, &e, dir);
if !view.visible(nbr) {
continue;
}
if depth >= min {
let dest_matches = match bound_to {
Some(want) => nbr == want,
None => true,
};
if dest_matches {
if out.len() >= cap {
return Err(row_cap_err(cap));
}
let mut next = row.clone();
if let Some(slot) = rel_slot {
next[slot] = Some(Cell::Path(depth));
}
next[to_slot] = Some(Cell::Node(nbr));
out.push(next);
}
}
if depth < max {
frontier_count += 1;
if frontier_count >= cap {
return Err(row_cap_err(cap));
}
let mut new_edges = state.edges.clone();
new_edges.push(e);
next_frontier.push(PathState {
node: nbr,
edges: new_edges,
});
}
}
}
frontier = next_frontier;
if frontier.is_empty() {
break;
}
}
}
Ok(out)
}
#[allow(clippy::too_many_arguments)]
fn exec_shortest_path(
view: &GraphView,
vars: &VarTable,
rows: &[Row],
from: &str,
rel_var: &Option<String>,
etypes: &[String],
dir: RelDir,
to: &str,
max_hops: u8,
) -> Result<Vec<Row>, String> {
let etypes = resolve_etypes(view, etypes);
let exp_dir = map_dir(dir);
let rel_slot = rel_var.as_ref().and_then(|rv| vars.slot(rv));
let mut out: Vec<Row> = Vec::new();
for row in rows {
let from_id = require_node(row, vars, from)?;
let to_id = require_node(row, vars, to)?;
let mut visited = std::collections::BTreeSet::new();
visited.insert(from_id);
let mut frontier: Vec<u32> = vec![from_id];
'bfs: for depth in 1u8..=max_hops {
let mut next_frontier: Vec<u32> = Vec::new();
for &node in &frontier {
for e in expand(view, node, etypes.as_deref(), exp_dir) {
let nbr = neighbor(node, &e, dir);
if !view.visible(nbr) {
continue;
}
if nbr == to_id {
let mut next = row.clone();
if let Some(slot) = rel_slot {
next[slot] = Some(Cell::Path(depth));
}
out.push(next);
break 'bfs;
}
if !visited.contains(&nbr) {
visited.insert(nbr);
next_frontier.push(nbr);
}
}
}
frontier = next_frontier;
if frontier.is_empty() {
break;
}
}
}
Ok(out)
}
fn exec_filter(
view: &GraphView,
vars: &VarTable,
rows: &[Row],
expr: &Expr,
params: &Params,
) -> Result<Vec<Row>, String> {
let mut out = Vec::with_capacity(rows.len());
for row in rows {
if eval_expr(view, vars, row, expr, params, 0)? {
out.push(row.clone());
}
}
Ok(out)
}
struct PullCtx<'a> {
view: &'a GraphView<'a>,
vars: &'a VarTable,
project_items: &'a [RetItem],
params: &'a Params<'a>,
bound: usize,
}
fn execute_pull(
view: &GraphView,
plan: &[PlanOp],
params: &Params,
bound: usize,
) -> Result<ResultSet, String> {
let proj_pos = match plan
.iter()
.position(|op| matches!(op, PlanOp::Project { .. }))
{
Some(p) => p,
None => return Ok(ResultSet::new(vec![])),
};
let producers = &plan[..proj_pos];
let project_items = match &plan[proj_pos] {
PlanOp::Project { items } => items,
_ => unreachable!(),
};
let columns: Vec<String> = project_items.iter().map(column_name).collect();
let vars = collect_vars(plan);
let ctx = PullCtx {
view,
vars: &vars,
project_items,
params,
bound,
};
let mut initial_row: Row = vec![None; vars.names.len()];
let mut result_rows: Vec<Vec<Option<Value>>> = Vec::with_capacity(bound);
pull_rows(&ctx, producers, &mut initial_row, &mut result_rows)?;
let skip_n = plan[proj_pos + 1..]
.iter()
.find_map(|op| match op {
PlanOp::Skip(ls) => Some(resolve_ls(ls, params)),
_ => None,
})
.transpose()?
.unwrap_or(0);
let skip_n = usize::try_from(skip_n).unwrap_or(usize::MAX);
let mut rs = ResultSet::new(columns);
for row in result_rows.into_iter().skip(skip_n) {
rs.push_row(row);
}
Ok(rs)
}
enum AggAcc {
Count(u64),
Sum {
val: f64,
has_value: bool,
},
Avg {
sum: f64,
n: u64,
},
Min(Option<Value>),
Max(Option<Value>),
Collect(Vec<Value>),
Distinct {
seen: HashSet<Option<ValueKey>>,
inner: Box<AggAcc>,
},
}
impl AggAcc {
fn for_arg(func: &AggFunc, arg: &AggArg) -> Self {
match arg {
AggArg::Distinct(_) => AggAcc::Distinct {
seen: HashSet::new(),
inner: Box::new(AggAcc::new(func)),
},
_ => AggAcc::new(func),
}
}
fn new(func: &AggFunc) -> Self {
match func {
AggFunc::Count => AggAcc::Count(0),
AggFunc::Sum => AggAcc::Sum {
val: 0.0,
has_value: false,
},
AggFunc::Avg => AggAcc::Avg { sum: 0.0, n: 0 },
AggFunc::Min => AggAcc::Min(None),
AggFunc::Max => AggAcc::Max(None),
AggFunc::Collect => AggAcc::Collect(Vec::new()),
}
}
fn finish(self) -> Option<Value> {
match self {
AggAcc::Count(n) => Some(Value::Int(i64::try_from(n).unwrap_or(i64::MAX))),
AggAcc::Sum { val, has_value } => {
if has_value {
Some(Value::Float(val))
} else {
None
}
}
AggAcc::Avg { sum, n } => {
if n > 0 {
Some(Value::Float(sum / n as f64))
} else {
None
}
}
AggAcc::Min(v) => v,
AggAcc::Max(v) => v,
AggAcc::Collect(items) => Some(Value::List(items)),
AggAcc::Distinct { inner, .. } => inner.finish(),
}
}
}
fn numeric_val(v: &Value) -> Option<f64> {
match v {
Value::Int(n) => Some(*n as f64),
Value::Float(f) => Some(*f),
_ => None,
}
}
struct AggStreamCtx<'a> {
view: &'a GraphView<'a>,
vars: &'a VarTable,
params: &'a Params<'a>,
func: &'a AggFunc,
arg: &'a AggArg,
}
fn execute_aggregate(
view: &GraphView,
plan: &[PlanOp],
params: &Params,
) -> Result<ResultSet, String> {
let agg_pos = match plan
.iter()
.position(|op| matches!(op, PlanOp::Aggregate { .. }))
{
Some(p) => p,
None => return Ok(ResultSet::new(vec![])),
};
let producers = &plan[..agg_pos];
let (func, arg, column) = match &plan[agg_pos] {
PlanOp::Aggregate { func, arg, column } => (func, arg, column),
_ => unreachable!(),
};
let vars = collect_vars(plan);
let ctx = AggStreamCtx {
view,
vars: &vars,
params,
func,
arg,
};
let initial_row: Row = vec![None; vars.names.len()];
let mut acc = AggAcc::for_arg(func, arg);
agg_stream(&ctx, producers, &initial_row, &mut acc)?;
let value = acc.finish();
let mut rs = ResultSet::new(vec![column.clone()]);
rs.push_row(vec![value]);
Ok(rs)
}
fn distinct_value(
view: &GraphView,
vars: &VarTable,
row: &Row,
arg: &AggArg,
) -> Result<Option<Value>, String> {
match arg {
AggArg::Star => Ok(None),
AggArg::Var(v) => {
let Some(slot) = vars.slot(v) else {
return Ok(None);
};
Ok(match row.get(slot).and_then(|c| c.as_ref()) {
Some(Cell::Node(id)) => view.ids.key_of(*id).map(|k| Value::Str(k.to_owned())),
Some(Cell::Scalar(val)) => Some(val.clone()),
Some(Cell::Path(hops)) => Some(Value::Int(*hops as i64)),
Some(Cell::Rel(e)) => Some(Value::Str(format!(
"{}\u{1}{}\u{1}{}",
e.etype, e.src, e.dst
))),
None => None,
})
}
AggArg::Prop { var, field } => resolve_prop(view, vars, row, var, field),
AggArg::Distinct(inner) => distinct_value(view, vars, row, inner),
}
}
fn update_acc(
view: &GraphView,
vars: &VarTable,
row: &Row,
func: &AggFunc,
arg: &AggArg,
acc: &mut AggAcc,
) -> Result<(), String> {
if let AggArg::Distinct(inner_arg) = arg {
let AggAcc::Distinct { seen, inner } = acc else {
return Ok(());
};
let val = distinct_value(view, vars, row, inner_arg)?;
let Some(val) = val else {
return Ok(());
};
if !seen.insert(group_key_normalize(&val)) {
return Ok(());
}
return update_acc(view, vars, row, func, inner_arg, inner);
}
match (func, arg) {
(AggFunc::Count, AggArg::Star) => {
if let AggAcc::Count(n) = acc {
*n += 1;
}
}
(AggFunc::Count, AggArg::Var(v)) => {
let slot = vars.slot(v);
let is_bound = slot
.and_then(|s| row.get(s))
.and_then(|c| c.as_ref())
.is_some();
if is_bound {
if let AggAcc::Count(n) = acc {
*n += 1;
}
}
}
(AggFunc::Count, AggArg::Prop { var, field }) => {
let val = resolve_prop(view, vars, row, var, field)?;
if val.is_some() {
if let AggAcc::Count(n) = acc {
*n += 1;
}
}
}
(AggFunc::Sum, AggArg::Prop { var, field }) => {
if let Some(v) = resolve_prop(view, vars, row, var, field)? {
if let Some(num) = numeric_val(&v) {
if let AggAcc::Sum { val, has_value } = acc {
*val += num;
*has_value = true;
}
}
}
}
(AggFunc::Avg, AggArg::Prop { var, field }) => {
if let Some(v) = resolve_prop(view, vars, row, var, field)? {
if let Some(num) = numeric_val(&v) {
if let AggAcc::Avg { sum, n } = acc {
*sum += num;
*n += 1;
}
}
}
}
(AggFunc::Min, AggArg::Prop { var, field }) => {
if let Some(v) = resolve_prop(view, vars, row, var, field)? {
if numeric_val(&v).is_some() {
if let AggAcc::Min(current) = acc {
*current = Some(match current.take() {
None => v,
Some(prev) => {
if cmp_optional(Some(&prev), Some(&v), false)
== std::cmp::Ordering::Greater
{
v
} else {
prev
}
}
});
}
}
}
}
(AggFunc::Max, AggArg::Prop { var, field }) => {
if let Some(v) = resolve_prop(view, vars, row, var, field)? {
if numeric_val(&v).is_some() {
if let AggAcc::Max(current) = acc {
*current = Some(match current.take() {
None => v,
Some(prev) => {
if cmp_optional(Some(&prev), Some(&v), true)
== std::cmp::Ordering::Greater
{
v
} else {
prev
}
}
});
}
}
}
}
(AggFunc::Collect, AggArg::Var(v)) => {
let val = vars
.slot(v)
.and_then(|s| row.get(s))
.and_then(|c| c.as_ref())
.and_then(|cell| match cell {
Cell::Scalar(x) => Some(x.clone()),
Cell::Node(id) => view.ids.key_of(*id).map(|k| Value::Str(k.to_owned())),
Cell::Path(h) => Some(Value::Int(*h as i64)),
Cell::Rel(_) => None,
});
if let Some(val) = val {
if let AggAcc::Collect(items) = acc {
items.push(val);
}
}
}
(AggFunc::Collect, AggArg::Prop { var, field }) => {
if let Some(val) = resolve_prop(view, vars, row, var, field)? {
if let AggAcc::Collect(items) = acc {
items.push(val);
}
}
}
_ => {}
}
Ok(())
}
fn agg_stream(
ctx: &AggStreamCtx<'_>,
ops: &[PlanOp],
row: &Row,
acc: &mut AggAcc,
) -> Result<(), String> {
let (op, rest) = match ops.split_first() {
Some(pair) => pair,
None => {
update_acc(ctx.view, ctx.vars, row, ctx.func, ctx.arg, acc)?;
return Ok(());
}
};
match op {
PlanOp::ScanLabel { var, label } => {
let ids = scan_ids(ctx.view, label.as_deref());
let slot = ctx
.vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
for &id in &ids {
let mut next = row.clone();
next[slot] = Some(Cell::Node(id));
agg_stream(ctx, rest, &next, acc)?;
}
}
PlanOp::ScanKey { var, key, label } => {
let slot = ctx
.vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
if let Some(id) =
resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
{
let mut next = row.clone();
next[slot] = Some(Cell::Node(id));
agg_stream(ctx, rest, &next, acc)?;
}
}
PlanOp::IndexScan {
var,
label,
field,
value,
} => {
let ids = index_scan_ids(
ctx.view,
ctx.vars,
row,
label.as_deref(),
field,
value,
ctx.params,
)?;
let slot = ctx
.vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
for &id in &ids {
let mut next = row.clone();
next[slot] = Some(Cell::Node(id));
agg_stream(ctx, rest, &next, acc)?;
}
}
PlanOp::IndexIntersect {
var,
label,
equalities,
} => {
let ids = index_intersect_ids(
ctx.view,
ctx.vars,
row,
label.as_deref(),
equalities,
ctx.params,
)?;
let slot = ctx
.vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
for id in ids {
let mut next = row.clone();
next[slot] = Some(Cell::Node(id));
agg_stream(ctx, rest, &next, acc)?;
}
}
PlanOp::Expand {
from,
rel_var,
etypes,
dir,
to,
to_label,
to_props,
} => {
let etypes = resolve_etypes(ctx.view, etypes);
let exp_dir = map_dir(*dir);
let to_slot = ctx
.vars
.slot(to)
.ok_or_else(|| format!("unbound variable `{to}`"))?;
let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
let from_id = require_node(row, ctx.vars, from)?;
let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
Some(Cell::Node(id)) => Some(*id),
Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
return Err(format!("variable `{to}` is not a node"))
}
None => None,
};
for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
if row_has_edge(row, &e) {
continue;
}
let nbr = neighbor(from_id, &e, *dir);
if !ctx.view.visible(nbr) {
continue;
}
if let Some(want) = bound_to {
if nbr != want {
continue;
}
}
if !node_matches(
ctx.view,
ctx.vars,
row,
nbr,
to_label.as_deref(),
to_props,
ctx.params,
)? {
continue;
}
let mut next = row.clone();
if let Some(slot) = rel_slot {
next[slot] = Some(Cell::Rel(e));
}
if bound_to.is_none() {
next[to_slot] = Some(Cell::Node(nbr));
}
agg_stream(ctx, rest, &next, acc)?;
}
}
PlanOp::Filter { expr } => {
if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
agg_stream(ctx, rest, row, acc)?;
}
}
PlanOp::LookupProps { var, props } => {
let id = require_node(row, ctx.vars, var)?;
if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
agg_stream(ctx, rest, row, acc)?;
}
}
PlanOp::JoinBound { var, label, props } => {
let id = require_node(row, ctx.vars, var)?;
if node_matches(
ctx.view,
ctx.vars,
row,
id,
label.as_deref(),
props,
ctx.params,
)? {
agg_stream(ctx, rest, row, acc)?;
}
}
PlanOp::VarExpand {
from,
rel_var,
etypes,
dir,
to,
min,
max,
} => {
let new_rows = exec_var_expand(
ctx.view,
ctx.vars,
std::slice::from_ref(row),
from,
rel_var,
etypes,
*dir,
to,
*min,
*max,
)?;
for nr in &new_rows {
agg_stream(ctx, rest, nr, acc)?;
}
}
PlanOp::ShortestPath {
from,
rel_var,
etypes,
dir,
to,
max_hops,
} => {
let new_rows = exec_shortest_path(
ctx.view,
ctx.vars,
std::slice::from_ref(row),
from,
rel_var,
etypes,
*dir,
to,
*max_hops,
)?;
for nr in &new_rows {
agg_stream(ctx, rest, nr, acc)?;
}
}
PlanOp::Project { .. } => {
return Err(
"agg executor: Project in producer slice — plan is structurally malformed"
.to_string(),
);
}
PlanOp::Distinct => {
return Err(
"agg executor: Distinct in producer slice — plan is structurally malformed"
.to_string(),
);
}
PlanOp::OrderBy { .. } => {
return Err(
"agg executor: OrderBy in producer slice — structurally malformed".to_string(),
);
}
PlanOp::Skip(_) => {
return Err(
"agg executor: Skip in producer slice — structurally malformed".to_string(),
);
}
PlanOp::Limit(_) => {
return Err(
"agg executor: Limit in producer slice — structurally malformed".to_string(),
);
}
PlanOp::Aggregate { .. } => {
return Err(
"agg executor: nested Aggregate in producer slice — structurally malformed"
.to_string(),
);
}
PlanOp::GroupAggregate { .. } => {
return Err(
"agg executor: GroupAggregate in producer slice — structurally malformed"
.to_string(),
);
}
PlanOp::With { .. } => {
return Err(
"agg executor: With in producer slice — structurally malformed".to_string(),
);
}
PlanOp::Unwind { .. } => {
return Err(
"agg executor: Unwind in producer slice — structurally malformed".to_string(),
);
}
PlanOp::LeftOuterApply { .. } => {
return Err(
"agg executor: LeftOuterApply in producer slice — structurally malformed"
.to_string(),
);
}
}
Ok(())
}
struct GroupStreamCtx<'a> {
view: &'a GraphView<'a>,
vars: &'a VarTable,
params: &'a Params<'a>,
keys: &'a [(String, RetItem)],
aggs: &'a [(AggFunc, AggArg, String)],
}
fn build_group_projected(
keys: &[(String, RetItem)],
aggs: &[(AggFunc, AggArg, String)],
key_order: Vec<GroupKey>,
groups: &mut HashMap<GroupKey, GroupEntry>,
) -> Projected {
let columns: Vec<String> = keys
.iter()
.map(|(col, _)| col.clone())
.chain(aggs.iter().map(|(_, _, col)| col.clone()))
.collect();
let mut rows: Vec<Vec<Option<Value>>> = Vec::with_capacity(key_order.len());
for gk in key_order {
let (display_keys, accs) = groups.remove(&gk).unwrap_or_default();
let mut row: Vec<Option<Value>> = Vec::with_capacity(columns.len());
for display_val in display_keys {
row.push(display_val);
}
for acc in accs {
row.push(acc.finish());
}
rows.push(row);
}
Projected { columns, rows }
}
fn key_source_cells(vars: &VarTable, row: &Row, keys: &[(String, RetItem)]) -> Vec<Option<Cell>> {
keys.iter()
.map(|(_, item)| match &item.value {
RetVal::Var(v) => vars
.slot(v)
.and_then(|s| row.get(s))
.and_then(|c| c.clone()),
_ => None,
})
.collect()
}
fn group_result_to_rows(
keys: &[(String, RetItem)],
aggs: &[(AggFunc, AggArg, String)],
key_order: Vec<GroupKey>,
groups: &mut HashMap<GroupKey, GroupEntry>,
key_cells: &mut HashMap<GroupKey, Vec<Option<Cell>>>,
vars: &VarTable,
) -> Vec<Row> {
let row_len = vars.names.len();
let mut out: Vec<Row> = Vec::with_capacity(key_order.len());
for gk in key_order {
let (display_keys, accs) = groups.remove(&gk).unwrap_or_default();
let mut cells = key_cells.remove(&gk).unwrap_or_default();
cells.resize(keys.len(), None);
let mut row: Row = vec![None; row_len];
for (((col, _), val), cell) in keys.iter().zip(display_keys).zip(cells) {
if let Some(slot) = vars.slot(col) {
row[slot] = cell.or_else(|| val.map(Cell::Scalar));
}
}
for ((_, _, col), acc) in aggs.iter().zip(accs) {
if let Some(slot) = vars.slot(col) {
row[slot] = acc.finish().map(Cell::Scalar);
}
}
out.push(row);
}
out
}
fn exec_order_by_rows(vars: &VarTable, rows: &mut Vec<Row>, items: &[OrderItem], view: &GraphView) {
let mut key_table: Vec<Vec<Option<Value>>> = Vec::with_capacity(rows.len());
for row in rows.iter() {
let mut row_key: Vec<Option<Value>> = Vec::with_capacity(items.len());
for item in items {
let val = match &item.target {
OrderTarget::Alias(name) | OrderTarget::Var(name) => vars
.slot(name)
.and_then(|s| row.get(s))
.and_then(|c| c.as_ref())
.and_then(|c| match c {
Cell::Scalar(v) => Some(v.clone()),
Cell::Node(id) => view.ids.key_of(*id).map(|k| Value::Str(k.to_owned())),
Cell::Path(hops) => Some(Value::Int(*hops as i64)),
Cell::Rel(_) => None,
}),
OrderTarget::Prop { var, field } => vars
.slot(var)
.and_then(|s| row.get(s))
.and_then(|c| c.as_ref())
.and_then(|c| match c {
Cell::Node(id) => view
.prop(*id, field)
.map(|vr| vr.into_value())
.or_else(|| node_identity_prop(view, *id, field)),
Cell::Rel(e) => view.edge_props.get(e.etype, e.src, e.dst, field),
_ => None,
}),
};
row_key.push(val);
}
key_table.push(row_key);
}
let mut indices: Vec<usize> = (0..rows.len()).collect();
indices.sort_by(|&a, &b| {
for (ki, item) in items.iter().enumerate() {
let c = cmp_optional(
key_table[a].get(ki).and_then(|x| x.as_ref()),
key_table[b].get(ki).and_then(|x| x.as_ref()),
item.descending,
);
if c != std::cmp::Ordering::Equal {
return c;
}
}
std::cmp::Ordering::Equal
});
let sorted: Vec<Row> = indices.into_iter().map(|i| rows[i].clone()).collect();
*rows = sorted;
}
fn execute_group_aggregate(
view: &GraphView,
plan: &[PlanOp],
params: &Params,
) -> Result<ResultSet, String> {
let gagg_pos = plan
.iter()
.position(|op| matches!(op, PlanOp::GroupAggregate { .. }))
.ok_or_else(|| "internal: GroupAggregate op not found in plan".to_string())?;
let producers = &plan[..gagg_pos];
let (keys, aggs) = match &plan[gagg_pos] {
PlanOp::GroupAggregate { keys, aggs } => (keys, aggs),
_ => unreachable!(),
};
let tail = &plan[gagg_pos + 1..];
let vars = collect_vars(plan);
let initial_row: Row = vec![None; vars.names.len()];
let mut groups: HashMap<GroupKey, GroupEntry> = HashMap::new();
let mut key_order: Vec<GroupKey> = Vec::new();
let ctx = GroupStreamCtx {
view,
vars: &vars,
params,
keys,
aggs,
};
group_stream(&ctx, producers, &initial_row, &mut groups, &mut key_order)?;
if keys.is_empty() && key_order.is_empty() {
let empty_key: GroupKey = vec![];
key_order.push(empty_key.clone());
groups.insert(
empty_key,
(
vec![],
aggs.iter().map(|(f, a, _)| AggAcc::for_arg(f, a)).collect(),
),
);
}
let mut projected = build_group_projected(keys, aggs, key_order, &mut groups);
for op in tail {
match op {
PlanOp::OrderBy { items } => exec_order_by(&mut projected, items)?,
PlanOp::Skip(ls) => {
let n = resolve_ls(ls, params)?;
apply_skip(&mut projected.rows, n);
}
PlanOp::Limit(ls) => {
let n = resolve_ls(ls, params)?;
apply_limit(&mut projected.rows, n);
}
_ => {} }
}
Ok(finish(projected))
}
fn group_stream(
ctx: &GroupStreamCtx<'_>,
ops: &[PlanOp],
row: &Row,
groups: &mut HashMap<GroupKey, GroupEntry>,
key_order: &mut Vec<GroupKey>,
) -> Result<(), String> {
let (op, rest) = match ops.split_first() {
Some(pair) => pair,
None => {
let mut gk: GroupKey = Vec::with_capacity(ctx.keys.len());
let mut display_vals: Vec<Option<Value>> = Vec::with_capacity(ctx.keys.len());
for (_, item) in ctx.keys {
let val = project_item(ctx.view, ctx.vars, row, item, ctx.params)?;
gk.push(val.as_ref().and_then(group_key_normalize));
display_vals.push(val);
}
if !groups.contains_key(&gk) {
if groups.len() >= max_groups() {
return Err(group_cap_err());
}
key_order.push(gk.clone());
let init: Vec<AggAcc> = ctx
.aggs
.iter()
.map(|(f, a, _)| AggAcc::for_arg(f, a))
.collect();
groups.insert(gk.clone(), (display_vals, init));
}
let (_, accs) = groups.get_mut(&gk).unwrap();
for (acc, (func, arg, _)) in accs.iter_mut().zip(ctx.aggs.iter()) {
update_acc(ctx.view, ctx.vars, row, func, arg, acc)?;
}
return Ok(());
}
};
match op {
PlanOp::ScanLabel { var, label } => {
let ids = scan_ids(ctx.view, label.as_deref());
let slot = ctx
.vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
for &id in &ids {
let mut next = row.clone();
next[slot] = Some(Cell::Node(id));
group_stream(ctx, rest, &next, groups, key_order)?;
}
}
PlanOp::ScanKey { var, key, label } => {
let slot = ctx
.vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
if let Some(id) =
resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
{
let mut next = row.clone();
next[slot] = Some(Cell::Node(id));
group_stream(ctx, rest, &next, groups, key_order)?;
}
}
PlanOp::IndexScan {
var,
label,
field,
value,
} => {
let ids = index_scan_ids(
ctx.view,
ctx.vars,
row,
label.as_deref(),
field,
value,
ctx.params,
)?;
let slot = ctx
.vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
for &id in &ids {
let mut next = row.clone();
next[slot] = Some(Cell::Node(id));
group_stream(ctx, rest, &next, groups, key_order)?;
}
}
PlanOp::IndexIntersect {
var,
label,
equalities,
} => {
let ids = index_intersect_ids(
ctx.view,
ctx.vars,
row,
label.as_deref(),
equalities,
ctx.params,
)?;
let slot = ctx
.vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
for id in ids {
let mut next = row.clone();
next[slot] = Some(Cell::Node(id));
group_stream(ctx, rest, &next, groups, key_order)?;
}
}
PlanOp::Expand {
from,
rel_var,
etypes,
dir,
to,
to_label,
to_props,
} => {
let etypes = resolve_etypes(ctx.view, etypes);
let exp_dir = map_dir(*dir);
let to_slot = ctx
.vars
.slot(to)
.ok_or_else(|| format!("unbound variable `{to}`"))?;
let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
let from_id = require_node(row, ctx.vars, from)?;
let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
Some(Cell::Node(id)) => Some(*id),
Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
return Err(format!("variable `{to}` is not a node"))
}
None => None,
};
for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
if row_has_edge(row, &e) {
continue;
}
let nbr = neighbor(from_id, &e, *dir);
if !ctx.view.visible(nbr) {
continue;
}
if let Some(want) = bound_to {
if nbr != want {
continue;
}
}
if !node_matches(
ctx.view,
ctx.vars,
row,
nbr,
to_label.as_deref(),
to_props,
ctx.params,
)? {
continue;
}
let mut next = row.clone();
if let Some(slot) = rel_slot {
next[slot] = Some(Cell::Rel(e));
}
if bound_to.is_none() {
next[to_slot] = Some(Cell::Node(nbr));
}
group_stream(ctx, rest, &next, groups, key_order)?;
}
}
PlanOp::Filter { expr } => {
if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
group_stream(ctx, rest, row, groups, key_order)?;
}
}
PlanOp::LookupProps { var, props } => {
let id = require_node(row, ctx.vars, var)?;
if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
group_stream(ctx, rest, row, groups, key_order)?;
}
}
PlanOp::JoinBound { var, label, props } => {
let id = require_node(row, ctx.vars, var)?;
if node_matches(
ctx.view,
ctx.vars,
row,
id,
label.as_deref(),
props,
ctx.params,
)? {
group_stream(ctx, rest, row, groups, key_order)?;
}
}
PlanOp::VarExpand {
from,
rel_var,
etypes,
dir,
to,
min,
max,
} => {
let new_rows = exec_var_expand(
ctx.view,
ctx.vars,
std::slice::from_ref(row),
from,
rel_var,
etypes,
*dir,
to,
*min,
*max,
)?;
for nr in &new_rows {
group_stream(ctx, rest, nr, groups, key_order)?;
}
}
PlanOp::ShortestPath {
from,
rel_var,
etypes,
dir,
to,
max_hops,
} => {
let new_rows = exec_shortest_path(
ctx.view,
ctx.vars,
std::slice::from_ref(row),
from,
rel_var,
etypes,
*dir,
to,
*max_hops,
)?;
for nr in &new_rows {
group_stream(ctx, rest, nr, groups, key_order)?;
}
}
PlanOp::Project { .. } => {
return Err(
"group executor: Project in producer slice — structurally malformed".to_string(),
);
}
PlanOp::Distinct => {
return Err(
"group executor: Distinct in producer slice — structurally malformed".to_string(),
);
}
PlanOp::OrderBy { .. } => {
return Err(
"group executor: OrderBy in producer slice — structurally malformed".to_string(),
);
}
PlanOp::Skip(_) => {
return Err(
"group executor: Skip in producer slice — structurally malformed".to_string(),
);
}
PlanOp::Limit(_) => {
return Err(
"group executor: Limit in producer slice — structurally malformed".to_string(),
);
}
PlanOp::Aggregate { .. } => {
return Err(
"group executor: Aggregate in producer slice — structurally malformed".to_string(),
);
}
PlanOp::GroupAggregate { .. } => {
return Err(
"group executor: nested GroupAggregate in producer slice — structurally malformed"
.to_string(),
);
}
PlanOp::With { .. } => {
return Err(
"group executor: With in producer slice — structurally malformed".to_string(),
);
}
PlanOp::Unwind { .. } => {
return Err(
"group executor: Unwind in producer slice — structurally malformed".to_string(),
);
}
PlanOp::LeftOuterApply { .. } => {
return Err(
"group executor: LeftOuterApply in producer slice — structurally malformed"
.to_string(),
);
}
}
Ok(())
}
fn pull_rows(
ctx: &PullCtx<'_>,
ops: &[PlanOp],
row: &mut Row,
result: &mut Vec<Vec<Option<Value>>>,
) -> Result<(), String> {
if result.len() >= ctx.bound {
return Ok(());
}
let (op, rest) = match ops.split_first() {
Some(pair) => pair,
None => {
let mut cells = Vec::with_capacity(ctx.project_items.len());
for item in ctx.project_items {
cells.push(project_item(ctx.view, ctx.vars, row, item, ctx.params)?);
}
result.push(cells);
return Ok(());
}
};
match op {
PlanOp::ScanLabel { var, label } => {
let slot = ctx
.vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
let want_sym = label.as_deref().and_then(|l| ctx.view.syms.get(l));
if label.is_some() && want_sym.is_none() {
return Ok(());
}
let prev = row[slot].clone();
let fused_filter = rest.first().and_then(|next_op| {
if let PlanOp::Filter {
expr:
Expr::Cmp {
lhs:
Operand::Prop {
var: ref fv,
field: ref f,
},
op: ref cmp_op_ref,
rhs: Operand::Lit(ref lit),
},
} = *next_op
{
if fv == var {
return Some((f.as_str(), cmp_op_ref, lit));
}
}
None
});
let fused_filter = fused_filter.filter(|(f, _, _)| !is_identity_field(f));
if let Some((field, cmp_op_ref, lit)) = fused_filter {
#[cfg(test)]
FUSED_SCAN_FIRES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let col = ctx.view.props.column(field);
let rest_after_filter = &rest[1..];
for (i, &sym) in ctx.view.labels.iter().enumerate() {
if result.len() >= ctx.bound {
break;
}
if sym == u32::MAX {
continue;
}
if let Some(ws) = want_sym {
if sym != ws {
continue;
}
}
let id = i as u32;
if !ctx.view.visible(id) {
continue;
}
if let Some(v) = col.get(id) {
if eval_cmp(cmp_op_ref, v, lit) {
row[slot] = Some(Cell::Node(id));
pull_rows(ctx, rest_after_filter, row, result)?;
}
}
}
} else {
for (i, &sym) in ctx.view.labels.iter().enumerate() {
if result.len() >= ctx.bound {
break;
}
if sym == u32::MAX {
continue;
}
if let Some(ws) = want_sym {
if sym != ws {
continue;
}
}
let id = i as u32;
if !ctx.view.visible(id) {
continue;
}
row[slot] = Some(Cell::Node(id));
pull_rows(ctx, rest, row, result)?;
}
}
row[slot] = prev;
}
PlanOp::ScanKey { var, key, label } => {
let slot = ctx
.vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
if let Some(id) =
resolve_scan_key_id(ctx.view, ctx.vars, row, key, label.as_deref(), ctx.params)?
{
let prev = row[slot].clone();
row[slot] = Some(Cell::Node(id));
pull_rows(ctx, rest, row, result)?;
row[slot] = prev;
}
}
PlanOp::IndexScan {
var,
label,
field,
value,
} => {
let slot = ctx
.vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
let ids = index_scan_ids(
ctx.view,
ctx.vars,
row,
label.as_deref(),
field,
value,
ctx.params,
)?;
let prev = row[slot].clone();
for id in ids {
if result.len() >= ctx.bound {
break;
}
row[slot] = Some(Cell::Node(id));
pull_rows(ctx, rest, row, result)?;
}
row[slot] = prev;
}
PlanOp::IndexIntersect {
var,
label,
equalities,
} => {
let slot = ctx
.vars
.slot(var)
.ok_or_else(|| format!("unbound variable `{var}`"))?;
let ids = index_intersect_ids(
ctx.view,
ctx.vars,
row,
label.as_deref(),
equalities,
ctx.params,
)?;
let prev = row[slot].clone();
for id in ids {
if result.len() >= ctx.bound {
break;
}
row[slot] = Some(Cell::Node(id));
pull_rows(ctx, rest, row, result)?;
}
row[slot] = prev;
}
PlanOp::Expand {
from,
rel_var,
etypes,
dir,
to,
to_label,
to_props,
} => {
let etypes = resolve_etypes(ctx.view, etypes);
let exp_dir = map_dir(*dir);
let to_slot = ctx
.vars
.slot(to)
.ok_or_else(|| format!("unbound variable `{to}`"))?;
let rel_slot = rel_var.as_ref().and_then(|rv| ctx.vars.slot(rv));
let from_id = require_node(row, ctx.vars, from)?;
let bound_to = match row.get(to_slot).and_then(|c| c.as_ref()) {
Some(Cell::Node(id)) => Some(*id),
Some(Cell::Rel(_) | Cell::Path(_) | Cell::Scalar(_)) => {
return Err(format!("variable `{to}` is not a node"))
}
None => None,
};
for e in expand(ctx.view, from_id, etypes.as_deref(), exp_dir) {
if result.len() >= ctx.bound {
break;
}
if row_has_edge(row, &e) {
continue;
}
let nbr = neighbor(from_id, &e, *dir);
if !ctx.view.visible(nbr) {
continue;
}
if let Some(want) = bound_to {
if nbr != want {
continue;
}
}
if !node_matches(
ctx.view,
ctx.vars,
row,
nbr,
to_label.as_deref(),
to_props,
ctx.params,
)? {
continue;
}
let mut next = row.clone();
if let Some(slot) = rel_slot {
next[slot] = Some(Cell::Rel(e));
}
if bound_to.is_none() {
next[to_slot] = Some(Cell::Node(nbr));
}
#[cfg(test)]
record_expand_row();
pull_rows(ctx, rest, &mut next, result)?;
}
}
PlanOp::Filter { expr } => {
if eval_expr(ctx.view, ctx.vars, row, expr, ctx.params, 0)? {
pull_rows(ctx, rest, row, result)?;
}
}
PlanOp::LookupProps { var, props } => {
let id = require_node(row, ctx.vars, var)?;
if node_matches(ctx.view, ctx.vars, row, id, None, props, ctx.params)? {
pull_rows(ctx, rest, row, result)?;
}
}
PlanOp::JoinBound { var, label, props } => {
let id = require_node(row, ctx.vars, var)?;
if node_matches(
ctx.view,
ctx.vars,
row,
id,
label.as_deref(),
props,
ctx.params,
)? {
pull_rows(ctx, rest, row, result)?;
}
}
PlanOp::Project { .. } => {
return Err(
"pull executor: Project reached pull_rows — plan is structurally malformed"
.to_string(),
);
}
PlanOp::Distinct => {
return Err(
"pull executor: Distinct reached pull_rows — DISTINCT queries must use \
the staged path (row_bound returns None)"
.to_string(),
);
}
PlanOp::OrderBy { .. } => {
return Err(
"pull executor: OrderBy reached pull_rows — queries with ORDER BY \
must use the staged path (row_bound returns None)"
.to_string(),
);
}
PlanOp::Skip(_) => {
return Err(
"pull executor: Skip reached pull_rows — Skip must appear after Project"
.to_string(),
);
}
PlanOp::Limit(_) => {
return Err(
"pull executor: Limit reached pull_rows — Limit must appear after Project"
.to_string(),
);
}
PlanOp::Aggregate { .. } => {
return Err(
"pull executor: Aggregate reached pull_rows — aggregate plans must use \
the execute_aggregate path (routed before pull in execute_inner)"
.to_string(),
);
}
PlanOp::VarExpand { .. } => {
return Err(
"pull executor: VarExpand reached pull_rows — variable-length path \
plans must use the staged path (row_bound returns None)"
.to_string(),
);
}
PlanOp::ShortestPath { .. } => {
return Err(
"pull executor: ShortestPath reached pull_rows — shortestPath plans \
must use the staged path (row_bound returns None)"
.to_string(),
);
}
PlanOp::GroupAggregate { .. } => {
return Err(
"pull executor: GroupAggregate reached pull_rows — grouped aggregate plans \
must use the execute_group_aggregate path (routed before pull in execute_inner)"
.to_string(),
);
}
PlanOp::With { .. } => {
return Err(
"pull executor: With reached pull_rows — pipeline plans must use the staged path \
(row_bound returns None for plans containing With)"
.to_string(),
);
}
PlanOp::Unwind { .. } => {
return Err(
"pull executor: Unwind reached pull_rows — pipeline plans must use the staged path \
(row_bound returns None for plans containing Unwind)"
.to_string(),
);
}
PlanOp::LeftOuterApply { .. } => {
return Err(
"pull executor: LeftOuterApply reached pull_rows — OPTIONAL MATCH plans must use \
the staged path (row_bound returns None for plans containing LeftOuterApply)"
.to_string(),
);
}
}
Ok(())
}
fn eval_expr(
view: &GraphView,
vars: &VarTable,
row: &Row,
expr: &Expr,
params: &Params,
depth: u32,
) -> Result<bool, String> {
if depth > 256 {
return Err("expression nesting too deep".into());
}
match expr {
Expr::And(lhs, rhs) => {
let l = eval_expr(view, vars, row, lhs, params, depth + 1)?;
let r = eval_expr(view, vars, row, rhs, params, depth + 1)?;
Ok(l && r)
}
Expr::Or(lhs, rhs) => {
let l = eval_expr(view, vars, row, lhs, params, depth + 1)?;
let r = eval_expr(view, vars, row, rhs, params, depth + 1)?;
Ok(l || r)
}
Expr::Not(inner) => Ok(!eval_expr(view, vars, row, inner, params, depth + 1)?),
Expr::Cmp { lhs, op, rhs } => {
let l = resolve_operand(view, vars, row, lhs, params)?;
let r = resolve_operand(view, vars, row, rhs, params)?;
match (l, r) {
(Some(a), Some(b)) => Ok(eval_cmp(op, &a, &b)),
_ => Ok(false),
}
}
Expr::Truthy(op) => {
let val = resolve_operand(view, vars, row, op, params)?;
Ok(match val {
None => false,
Some(Value::Bool(b)) => b,
Some(Value::Int(n)) => n != 0,
Some(Value::Float(f)) => f != 0.0,
Some(Value::Str(s)) => !s.is_empty(),
Some(Value::List(v)) => !v.is_empty(),
Some(Value::Map(m)) => !m.is_empty(),
})
}
Expr::IsNull(op) => {
let val = resolve_operand(view, vars, row, op, params)?;
Ok(val.is_none())
}
Expr::IsNotNull(op) => {
let val = resolve_operand(view, vars, row, op, params)?;
Ok(val.is_some())
}
Expr::In { expr, list } => eval_in(view, vars, row, expr, list, params),
}
}
fn eval_in(
view: &GraphView,
vars: &VarTable,
row: &Row,
expr: &Operand,
list: &[Operand],
params: &Params,
) -> Result<bool, String> {
let Some(needle) = resolve_operand(view, vars, row, expr, params)? else {
return Ok(false);
};
for item_op in list {
match resolve_operand(view, vars, row, item_op, params)? {
None => {}
Some(Value::List(items)) => {
for item in items {
if crate::filter::eval_cmp(&crate::filter::CmpOp::Eq, &needle, &item) {
return Ok(true);
}
}
}
Some(item) if crate::filter::eval_cmp(&crate::filter::CmpOp::Eq, &needle, &item) => {
return Ok(true);
}
Some(_) => {}
}
}
Ok(false)
}
fn exec_distinct(table: &mut Projected) -> Result<(), String> {
let cap = max_intermediate_rows();
let mut seen: BTreeSet<Vec<Option<ValueKey>>> = BTreeSet::new();
let mut out = Vec::with_capacity(table.rows.len().min(cap));
for row in table.rows.drain(..) {
let key: Vec<Option<ValueKey>> = row
.iter()
.map(|cell| cell.as_ref().and_then(group_key_normalize))
.collect();
if seen.insert(key) {
if out.len() >= cap {
return Err(row_cap_err(cap));
}
out.push(row);
}
}
table.rows = out;
Ok(())
}
fn column_name(item: &RetItem) -> String {
if let Some(alias) = &item.alias {
return alias.clone();
}
ret_val_label(&item.value).unwrap_or_else(|| match &item.value {
RetVal::Agg { func, arg } => {
let f = match func {
AggFunc::Count => "COUNT",
AggFunc::Sum => "SUM",
AggFunc::Avg => "AVG",
AggFunc::Min => "MIN",
AggFunc::Max => "MAX",
AggFunc::Collect => "COLLECT",
};
format!("{f}({})", agg_arg_label(arg))
}
_ => unreachable!("ret_val_label names every non-aggregate item"),
})
}
fn exec_project(
view: &GraphView,
vars: &VarTable,
rows: &[Row],
items: &[RetItem],
params: &Params,
) -> Result<Projected, String> {
let columns: Vec<String> = items.iter().map(column_name).collect();
let mut out_rows = Vec::with_capacity(rows.len());
for row in rows {
let mut cells = Vec::with_capacity(items.len());
for item in items {
cells.push(project_item(view, vars, row, item, params)?);
}
out_rows.push(cells);
}
Ok(Projected {
columns,
rows: out_rows,
})
}
fn project_item(
view: &GraphView,
vars: &VarTable,
row: &Row,
item: &RetItem,
params: &Params,
) -> Result<Option<Value>, String> {
match &item.value {
RetVal::Var(v) => {
let slot = vars.slot(v).ok_or_else(|| format!("unbound variable `{v}`"))?;
match row.get(slot).and_then(|c| c.as_ref()) {
None => Ok(None),
Some(Cell::Node(id)) => match view.ids.key_of(*id) {
Some(key) => Ok(Some(Value::Str(key.to_owned()))),
None => Err(format!("unknown node id {id}")),
},
Some(Cell::Scalar(val)) => Ok(Some(val.clone())),
Some(Cell::Rel(_)) => Err(format!(
"variable `{v}` is a relationship; return its properties ({v}.field) instead"
)),
Some(Cell::Path(hops)) => Ok(Some(Value::Int(*hops as i64))),
}
}
RetVal::Prop { var, field } => resolve_prop(view, vars, row, var, field),
RetVal::Agg { .. } => Err(
"project_item: Agg variant reached exec_project — aggregate plans must not contain Project"
.to_string(),
),
RetVal::FuncCall { name, args } => eval_func(name, args, view, vars, row, params),
RetVal::ScalarExpr(op) => resolve_operand(view, vars, row, op, params),
}
}
fn order_column(item: &OrderItem) -> String {
match &item.target {
OrderTarget::Alias(name) | OrderTarget::Var(name) => name.clone(),
OrderTarget::Prop { var, field } => format!("{var}.{field}"),
}
}
fn exec_order_by(table: &mut Projected, items: &[OrderItem]) -> Result<(), String> {
let mut keys = Vec::with_capacity(items.len());
for item in items {
let name = order_column(item);
let idx = table
.columns
.iter()
.position(|c| c == &name)
.ok_or_else(|| format!("ORDER BY target `{name}` is not a projected column"))?;
keys.push((idx, item.descending));
}
table.rows.sort_by(|a, b| {
for &(idx, desc) in &keys {
let c = cmp_optional(
a.get(idx).and_then(|x| x.as_ref()),
b.get(idx).and_then(|x| x.as_ref()),
desc,
);
if c != std::cmp::Ordering::Equal {
return c;
}
}
std::cmp::Ordering::Equal
});
Ok(())
}
fn resolve_ls(ls: &LimitSkip, params: &Params) -> Result<u64, String> {
match ls {
LimitSkip::Exact(n) => Ok(*n),
LimitSkip::Param(name) => {
let val = params
.0
.get(name)
.ok_or_else(|| format!("missing parameter `{name}` (used in LIMIT/SKIP)"))?;
match val {
Value::Int(i) if *i >= 0 => Ok(*i as u64),
Value::Int(i) => Err(format!(
"LIMIT/SKIP parameter `{name}` must be a non-negative integer, got {i}"
)),
other => Err(format!(
"LIMIT/SKIP parameter `{name}` must be an integer, got {other:?}"
)),
}
}
}
}
fn apply_skip<T>(rows: &mut Vec<T>, n: u64) {
let n = usize::try_from(n).unwrap_or(usize::MAX);
if n >= rows.len() {
rows.clear();
} else {
rows.drain(0..n);
}
}
fn apply_limit<T>(rows: &mut Vec<T>, n: u64) {
let n = usize::try_from(n).unwrap_or(usize::MAX);
rows.truncate(n);
}
#[cfg(test)]
mod tests {
use super::{execute, resolve_operand, Params, Row, VarTable};
use crate::cypher::ast::{
ArithOp, LimitSkip, Operand, OrderItem, OrderTarget, RetItem, RetVal,
};
use crate::cypher::plan::{plan, PlanOp};
use crate::cypher::{lex, parse, RelDir};
use crate::result::ResultSet;
use crate::view::GraphView;
use core_storage::v8::seam::{ColumnsView, EdgePropsView, TopologyView};
use core_storage::{ColumnStore, EdgeProps, IdMap, Interner, Topology, Value};
use proptest::prelude::*;
use std::collections::BTreeMap;
struct Fx {
ids: IdMap,
syms: Interner,
labels: Vec<u32>,
props: ColumnStore,
topo: Topology,
eprops: EdgeProps,
}
impl Fx {
fn new() -> Self {
Fx {
ids: IdMap::new(),
syms: Interner::new(),
labels: vec![],
props: ColumnStore::new(),
topo: Topology::new(),
eprops: EdgeProps::new(),
}
}
fn add(&mut self, label: &str, key: &str, props: Vec<(&str, Value)>) -> u32 {
let id = self.ids.get_or_insert(key);
let sym = self.syms.intern(label);
self.labels.resize(id as usize + 1, u32::MAX);
self.labels[id as usize] = sym;
for (f, v) in props {
self.props.set(id, f, v);
}
id
}
fn edge(&mut self, etype: &str, src: u32, dst: u32, props: Vec<(&str, Value)>) {
let et = self.syms.intern(etype);
self.topo.add_edge(et, src, dst);
for (f, v) in props {
self.eprops.set(et, src, dst, f, v);
}
}
fn view(&self) -> GraphView<'_> {
GraphView {
ids: &self.ids,
syms: &self.syms,
labels: &self.labels,
props: ColumnsView::owned(&self.props),
topo: TopologyView::owned(&self.topo),
edge_props: EdgePropsView::owned(&self.eprops),
mask: None,
prop_index: None,
}
}
fn view_indexed<'a>(
&'a self,
index: &'a core_storage::property_index::PropertyIndex,
) -> GraphView<'a> {
GraphView {
prop_index: Some(index),
..self.view()
}
}
}
fn compile(src: &str) -> Vec<PlanOp> {
plan(&parse(&lex(src).expect("lex")).expect("parse")).expect("plan")
}
fn run(
view: &GraphView,
src: &str,
params: &BTreeMap<String, Value>,
) -> Result<ResultSet, String> {
execute(view, &compile(src), &Params(params))
}
fn s(v: &str) -> Value {
Value::Str(v.into())
}
fn f(v: f64) -> Value {
Value::Float(v)
}
fn i(v: i64) -> Value {
Value::Int(v)
}
fn rows_of(rs: &ResultSet) -> Vec<Vec<Option<Value>>> {
(0..rs.len()).map(|i| rs.row(i).to_vec()).collect()
}
fn col(rs: &ResultSet, name: &str) -> Vec<Option<Value>> {
(0..rs.len()).map(|i| rs.get(i, name).cloned()).collect()
}
fn hop_graph() -> Fx {
let mut fx = Fx::new();
let ada = fx.add("Person", "ada", vec![]);
let bob = fx.add("Person", "bob", vec![]);
let cam = fx.add("Person", "cam", vec![]);
let acme = fx.add("Company", "acme", vec![]);
fx.edge("KNOWS", ada, bob, vec![]);
fx.edge("KNOWS", ada, cam, vec![]);
fx.edge("KNOWS", bob, cam, vec![]);
fx.edge("LIKES", ada, acme, vec![]);
fx
}
fn undirected_graph() -> Fx {
let mut fx = Fx::new();
let a = fx.add("N", "a", vec![]);
let b = fx.add("N", "b", vec![]);
fx.edge("T", a, b, vec![("w", i(42))]);
fx
}
fn triangle() -> Fx {
let mut fx = Fx::new();
let a = fx.add("N", "a", vec![]);
let b = fx.add("N", "b", vec![]);
let c = fx.add("N", "c", vec![]);
fx.edge("T", a, b, vec![("eid", i(1))]);
fx.edge("T", b, c, vec![("eid", i(2))]);
fx.edge("T", c, a, vec![("eid", i(3))]);
fx
}
fn single_edge() -> (Fx, u32, u32) {
let mut fx = Fx::new();
let a = fx.add("N", "a", vec![]);
let b = fx.add("N", "b", vec![]);
fx.edge("T", a, b, vec![]);
(fx, a, b)
}
fn dogfood_graph() -> Fx {
let mut fx = Fx::new();
let t1 = fx.add("Talent", "t1", vec![("id", s("t1"))]);
let acme = fx.add("Company", "acme", vec![]);
let beta = fx.add("Company", "beta", vec![]);
let gamma = fx.add("Company", "gamma", vec![]);
let delta = fx.add("Company", "delta", vec![]);
let echo = fx.add("Company", "echo", vec![]);
let foxtrot = fx.add("Company", "foxtrot", vec![]);
let zeta = fx.add("Company", "zeta", vec![]);
fx.edge("INDUSTRY_ALIGNMENT", acme, t1, vec![("score", f(0.9))]);
fx.edge("SPECIALTY_MATCH", acme, t1, vec![("score", f(0.8))]);
fx.edge("INDUSTRY_ALIGNMENT", beta, t1, vec![("score", f(0.6))]);
fx.edge("SPECIALTY_MATCH", beta, t1, vec![("score", f(0.7))]);
fx.edge("INDUSTRY_ALIGNMENT", gamma, t1, vec![("score", f(0.4))]);
fx.edge("SPECIALTY_MATCH", gamma, t1, vec![("score", f(0.9))]);
fx.edge("INDUSTRY_ALIGNMENT", delta, t1, vec![("score", f(0.8))]);
fx.edge("SPECIALTY_MATCH", delta, t1, vec![("score", f(0.3))]);
fx.edge("INDUSTRY_ALIGNMENT", echo, t1, vec![("score", f(0.5))]);
fx.edge("SPECIALTY_MATCH", echo, t1, vec![("score", f(0.5))]);
fx.edge("INDUSTRY_ALIGNMENT", foxtrot, t1, vec![("score", f(0.95))]);
fx.edge("INDUSTRY_ALIGNMENT", zeta, t1, vec![("score", f(0.9))]);
fx.edge("SPECIALTY_MATCH", zeta, t1, vec![("score", f(0.6))]);
fx
}
const DOGFOOD: &str = "\
MATCH (t:Talent {id: $tid}) \
MATCH (c:Company)-[i:INDUSTRY_ALIGNMENT]->(t) \
MATCH (c)-[s:SPECIALTY_MATCH]->(t) \
WHERE i.score >= 0.5 AND s.score >= 0.5 \
RETURN c, i.score AS industry, s.score AS specialty \
ORDER BY industry DESC, specialty DESC \
LIMIT 10";
fn tid_params() -> BTreeMap<String, Value> {
let mut p = BTreeMap::new();
p.insert("tid".into(), s("t1"));
p
}
#[test]
fn single_hop_match_label_and_etype_filters() {
let fx = hop_graph();
let v = fx.view();
let rs = run(
&v,
"MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b",
&BTreeMap::new(),
)
.expect("single-hop");
assert_eq!(rs.columns(), &["a".to_string(), "b".to_string()]);
assert_eq!(
rows_of(&rs),
vec![
vec![Some(s("ada")), Some(s("bob"))],
vec![Some(s("ada")), Some(s("cam"))],
vec![Some(s("bob")), Some(s("cam"))],
]
);
let likes = run(
&v,
"MATCH (a:Person)-[:LIKES]->(b:Company) RETURN a, b",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rows_of(&likes), vec![vec![Some(s("ada")), Some(s("acme"))]]);
let no_combo = run(
&v,
"MATCH (a:Person)-[:KNOWS]->(b:Company) RETURN a, b",
&BTreeMap::new(),
)
.unwrap();
assert!(no_combo.is_empty());
}
#[test]
fn undirected_match_finds_both_orientations_and_binds_true_triple() {
let fx = undirected_graph();
let v = fx.view();
let rs =
run(&v, "MATCH (x)-[r:T]-(y) RETURN x, y, r.w", &BTreeMap::new()).expect("undirected");
assert_eq!(
rows_of(&rs),
vec![
vec![Some(s("a")), Some(s("b")), Some(i(42))],
vec![Some(s("b")), Some(s("a")), Some(i(42))],
]
);
let left = run(
&v,
"MATCH (y)<-[r:T]-(x) RETURN x, y, r.w",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
rows_of(&left),
vec![vec![Some(s("a")), Some(s("b")), Some(i(42))]]
);
}
#[test]
fn relationship_uniqueness_triangle_and_two_hop_cycle() {
let tri = triangle();
let v = tri.view();
let rs = run(
&v,
"MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z, r1.eid, r2.eid",
&BTreeMap::new(),
)
.expect("triangle 2-hop");
assert_eq!(
rows_of(&rs),
vec![
vec![
Some(s("a")),
Some(s("b")),
Some(s("c")),
Some(i(1)),
Some(i(2))
],
vec![
Some(s("b")),
Some(s("c")),
Some(s("a")),
Some(i(2)),
Some(i(3))
],
vec![
Some(s("c")),
Some(s("a")),
Some(s("b")),
Some(i(3)),
Some(i(1))
],
]
);
for row in rows_of(&rs) {
assert_ne!(row[3], row[4], "r1 must never bind the same edge as r2");
}
let (mut one, a, b) = single_edge();
let v = one.view();
let cycle = run(
&v,
"MATCH (x)-[:T]->(y)-[:T]->(x) RETURN x",
&BTreeMap::new(),
)
.expect("2-hop cycle");
assert!(
cycle.is_empty(),
"single directed edge cannot close a 2-hop cycle"
);
let undirected_cycle = run(&v, "MATCH (x)-[:T]-(y)-[:T]-(x) RETURN x", &BTreeMap::new())
.expect("undirected uniqueness");
assert!(
undirected_cycle.is_empty(),
"relationship uniqueness must reject walking the same triple back"
);
one.edge("T", b, a, vec![]);
let v = one.view();
let with_recip = run(
&v,
"MATCH (x)-[:T]->(y)-[:T]->(x) RETURN x",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(col(&with_recip, "x"), vec![Some(s("a")), Some(s("b"))]);
}
#[test]
fn multi_match_join_bound_and_bound_destination_expand() {
let fx = dogfood_graph();
let v = fx.view();
let rs = run(
&v,
"MATCH (t:Talent {id: $tid}) \
MATCH (c:Company)-[i:INDUSTRY_ALIGNMENT]->(t) \
MATCH (c)-[s:SPECIALTY_MATCH]->(t) \
RETURN c",
&tid_params(),
)
.expect("join + bound dest");
assert_eq!(
col(&rs, "c"),
vec![
Some(s("acme")),
Some(s("beta")),
Some(s("gamma")),
Some(s("delta")),
Some(s("echo")),
Some(s("zeta")),
]
);
let keep = run(&v, "MATCH (c:Company) MATCH (c) RETURN c", &BTreeMap::new()).unwrap();
assert_eq!(keep.len(), 7);
let drop = run(
&v,
"MATCH (c:Company) MATCH (c:Talent) RETURN c",
&BTreeMap::new(),
)
.unwrap();
assert!(drop.is_empty());
}
#[test]
fn scan_key_exec_does_not_use_label_scan() {
let mut fx = Fx::new();
fx.add("Person", "p1", vec![]);
fx.add("Person", "p2", vec![]);
fx.add("Person", "p3", vec![]);
fx.add("Company", "c1", vec![]);
let v = fx.view();
let params = BTreeMap::new();
let fires_before = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
let rs = run(&v, "MATCH (n:Person {id: 'p2'}) RETURN n", ¶ms).expect("scan key");
let fires_after = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
assert!(
fires_after > fires_before,
"SCAN_KEY_FIRES must increment; before={fires_before} after={fires_after}"
);
assert_eq!(rows_of(&rs), vec![vec![Some(s("p2"))]]);
let miss = run(&v, "MATCH (n:Person {id: 'nope'}) RETURN n", ¶ms).unwrap();
assert!(
miss.is_empty(),
"missing key must be zero rows, not an error"
);
let wrong = run(&v, "MATCH (n:Person {id: 'c1'}) RETURN n", ¶ms).unwrap();
assert!(
wrong.is_empty(),
"wrong label must be zero rows, not an error"
);
}
#[test]
fn rel_var_edge_prop_filter() {
let mut fx = Fx::new();
let a = fx.add("N", "a", vec![]);
let b = fx.add("N", "b", vec![]);
let c = fx.add("N", "c", vec![]);
fx.edge("T", a, b, vec![("w", f(0.7))]);
fx.edge("T", a, c, vec![("w", f(0.3))]);
let v = fx.view();
let rs = run(
&v,
"MATCH (x)-[r:T]->(y) WHERE r.w >= 0.5 RETURN y, r.w",
&BTreeMap::new(),
)
.expect("edge-prop filter");
assert_eq!(rows_of(&rs), vec![vec![Some(s("b")), Some(f(0.7))]]);
let fail = run(
&v,
"MATCH (x)-[r:T]->(y) WHERE r.w >= 0.8 RETURN y",
&BTreeMap::new(),
)
.unwrap();
assert!(fail.is_empty());
}
#[test]
fn params_present_resolve_missing_is_err_before_rows() {
let fx = dogfood_graph();
let v = fx.view();
let hit =
run(&v, "MATCH (t:Talent {id: $tid}) RETURN t", &tid_params()).expect("present param");
assert_eq!(col(&hit, "t"), vec![Some(s("t1"))]);
let err = run(
&v,
"MATCH (t:NoSuchLabel {id: $tid}) RETURN t",
&BTreeMap::new(),
)
.expect_err("missing param must be Err, not Ok(empty)");
assert!(
err.contains("tid")
&& (err.contains("param") || err.contains("Param") || err.contains("missing")),
"missing-param error must name the parameter, got: {err}"
);
let err = run(
&v,
"MATCH (t:Talent) WHERE t.id = $tid RETURN t",
&BTreeMap::new(),
)
.expect_err("missing WHERE param");
assert!(err.contains("tid"), "got: {err}");
}
#[test]
fn order_by_none_last_then_skip_limit() {
let mut fx = Fx::new();
fx.add("Person", "ada", vec![("age", i(30))]);
fx.add("Person", "bob", vec![]); fx.add("Person", "cam", vec![("age", i(10))]);
fx.add("Person", "dan", vec![("age", i(20))]);
let v = fx.view();
let asc = run(
&v,
"MATCH (p:Person) RETURN p, p.age AS age ORDER BY age",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
rows_of(&asc),
vec![
vec![Some(s("cam")), Some(i(10))],
vec![Some(s("dan")), Some(i(20))],
vec![Some(s("ada")), Some(i(30))],
vec![Some(s("bob")), None],
]
);
let desc = run(
&v,
"MATCH (p:Person) RETURN p, p.age AS age ORDER BY age DESC",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
rows_of(&desc),
vec![
vec![Some(s("ada")), Some(i(30))],
vec![Some(s("dan")), Some(i(20))],
vec![Some(s("cam")), Some(i(10))],
vec![Some(s("bob")), None],
]
);
let skip_lim = run(
&v,
"MATCH (p:Person) RETURN p, p.age AS age ORDER BY age SKIP 1 LIMIT 2",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
rows_of(&skip_lim),
vec![
vec![Some(s("dan")), Some(i(20))],
vec![Some(s("ada")), Some(i(30))],
]
);
let desc_sl = run(
&v,
"MATCH (p:Person) RETURN p, p.age AS age ORDER BY age DESC SKIP 1 LIMIT 2",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
rows_of(&desc_sl),
vec![
vec![Some(s("dan")), Some(i(20))],
vec![Some(s("cam")), Some(i(10))],
]
);
}
#[test]
fn unknown_label_and_etype_are_ok_empty() {
let fx = hop_graph();
let v = fx.view();
let lab = run(&v, "MATCH (x:Nope) RETURN x", &BTreeMap::new()).expect("unknown label");
assert!(lab.is_empty());
let et = run(
&v,
"MATCH (a)-[:NO_SUCH_ETYPE]->(b) RETURN a",
&BTreeMap::new(),
)
.expect("unknown etype");
assert!(et.is_empty());
}
#[test]
fn execute_is_deterministic() {
let fx = dogfood_graph();
let v = fx.view();
let p = tid_params();
let a = run(&v, DOGFOOD, &p).expect("first");
let b = run(&v, DOGFOOD, &p).expect("second");
assert_eq!(a, b);
let hop = hop_graph();
let hv = hop.view();
let q = "MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b";
assert_eq!(run(&hv, q, &BTreeMap::new()), run(&hv, q, &BTreeMap::new()));
}
#[test]
fn dogfood_pipeline_exact_rows() {
let fx = dogfood_graph();
let v = fx.view();
let rs = run(&v, DOGFOOD, &tid_params()).expect("dogfood");
assert_eq!(
rs.columns(),
&[
"c".to_string(),
"industry".to_string(),
"specialty".to_string()
]
);
assert_eq!(
rows_of(&rs),
vec![
vec![Some(s("acme")), Some(f(0.9)), Some(f(0.8))],
vec![Some(s("zeta")), Some(f(0.9)), Some(f(0.6))],
vec![Some(s("beta")), Some(f(0.6)), Some(f(0.7))],
vec![Some(s("echo")), Some(f(0.5)), Some(f(0.5))],
]
);
}
#[test]
fn unknown_var_in_op_is_err_not_panic() {
let fx = hop_graph();
let v = fx.view();
let plan = vec![
PlanOp::ScanLabel {
var: "a".into(),
label: None,
},
PlanOp::Expand {
from: "zzz".into(),
rel_var: Some("r".into()),
etypes: vec![],
dir: RelDir::Right,
to: "b".into(),
to_label: None,
to_props: vec![],
},
PlanOp::Project {
items: vec![RetItem {
value: RetVal::Var("a".into()),
alias: None,
}],
},
];
let params = BTreeMap::new();
let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
execute(&v, &plan, &Params(¶ms))
}));
assert!(caught.is_ok(), "execute panicked on unknown var");
let err = caught.unwrap().expect_err("unknown var must be Err");
assert!(
err.contains("zzz") && err.to_ascii_lowercase().contains("unbound"),
"got: {err}"
);
let join = vec![
PlanOp::JoinBound {
var: "ghost".into(),
label: None,
props: vec![],
},
PlanOp::Project {
items: vec![RetItem {
value: RetVal::Var("ghost".into()),
alias: None,
}],
},
];
let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
execute(&v, &join, &Params(¶ms))
}));
assert!(caught.is_ok(), "execute panicked on JoinBound unknown var");
assert!(caught.unwrap().is_err());
}
#[test]
fn dest_props_on_bound_expand_are_applied() {
let fx = dogfood_graph();
let v = fx.view();
let rs = run(
&v,
"MATCH (t:Talent {id: $tid}) \
MATCH (c:Company)-[r:INDUSTRY_ALIGNMENT]->(t:Talent {id: $tid}) \
RETURN c",
&tid_params(),
)
.unwrap();
assert_eq!(
col(&rs, "c"),
vec![
Some(s("acme")),
Some(s("beta")),
Some(s("gamma")),
Some(s("delta")),
Some(s("echo")),
Some(s("foxtrot")),
Some(s("zeta")),
]
);
let miss = run(
&v,
"MATCH (t:Talent {id: $tid}) \
MATCH (c:Company)-[r:INDUSTRY_ALIGNMENT]->(t {id: 'nope'}) \
RETURN c",
&tid_params(),
)
.unwrap();
assert!(miss.is_empty());
}
#[test]
fn missing_node_prop_projects_none_and_pattern_misses() {
let mut fx = Fx::new();
fx.add("Person", "ada", vec![("age", i(30))]);
fx.add("Person", "bob", vec![]);
let v = fx.view();
let rs = run(&v, "MATCH (p:Person) RETURN p.age", &BTreeMap::new()).unwrap();
assert_eq!(col(&rs, "p.age"), vec![Some(i(30)), None]);
let pat = run(&v, "MATCH (p:Person {age: 30}) RETURN p", &BTreeMap::new()).unwrap();
assert_eq!(col(&pat, "p"), vec![Some(s("ada"))]);
}
#[test]
fn unlabeled_match_does_not_project_sentinel_ghost_key() {
let mut fx = Fx::new();
fx.add("Person", "ada", vec![]);
fx.ids.get_or_insert("ghost");
fx.labels.resize(fx.ids.len(), u32::MAX);
fx.add("Person", "bob", vec![]);
let v = fx.view();
let rs = run(&v, "MATCH (n) RETURN n", &BTreeMap::new()).expect("unlabeled scan");
assert_eq!(col(&rs, "n"), vec![Some(s("ada")), Some(s("bob"))]);
assert!(
!rows_of(&rs)
.iter()
.any(|row| row.iter().any(|c| *c == Some(s("ghost")))),
"sentinel slot must not project a ghost key"
);
}
#[test]
fn execute_never_panics_on_hostile_plans() {
let fx = hop_graph();
let v = fx.view();
let params = BTreeMap::new();
let hostile = vec![
vec![],
vec![PlanOp::Project { items: vec![] }],
vec![PlanOp::OrderBy {
items: vec![OrderItem {
target: OrderTarget::Alias("nope".into()),
descending: false,
}],
}],
vec![PlanOp::Filter {
expr: crate::cypher::ast::Expr::Cmp {
lhs: Operand::Prop {
var: "missing".into(),
field: "x".into(),
},
op: crate::filter::CmpOp::Eq,
rhs: Operand::Lit(i(1)),
},
}],
vec![
PlanOp::ScanLabel {
var: "a".into(),
label: None,
},
PlanOp::LookupProps {
var: "zzz".into(),
props: vec![("k".into(), Operand::Lit(i(1)))],
},
],
vec![
PlanOp::Skip(LimitSkip::Exact(99)),
PlanOp::Limit(LimitSkip::Exact(0)),
],
];
for plan in hostile {
let caught = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
execute(&v, &plan, &Params(¶ms))
}));
assert!(caught.is_ok(), "execute panicked on hostile plan: {plan:?}");
}
}
#[test]
fn unlabeled_and_labeled_scans_skip_tombstoned_ids() {
let mut fx = Fx::new();
let ada = fx.add("Person", "ada", vec![]);
let bob = fx.add("Person", "bob", vec![]);
fx.edge("KNOWS", ada, bob, vec![]);
fx.ids.delete("ada");
fx.labels[ada as usize] = u32::MAX;
let knows = fx.syms.get("KNOWS").unwrap();
fx.topo.remove_edge(knows, ada, bob);
let v = fx.view();
let labeled = run(&v, "MATCH (p:Person) RETURN p", &BTreeMap::new()).unwrap();
assert_eq!(col(&labeled, "p"), vec![Some(s("bob"))]);
let unlabeled = run(&v, "MATCH (n) RETURN n", &BTreeMap::new()).unwrap();
assert_eq!(col(&unlabeled, "n"), vec![Some(s("bob"))]);
let hop = run(&v, "MATCH (x)-[:KNOWS]->(y) RETURN x, y", &BTreeMap::new()).unwrap();
assert!(
hop.is_empty(),
"expand cannot yield edges to a deleted node once topology is swept"
);
}
proptest! {
#[test]
fn prop_bounded_equals_unbounded_slice(
n_nodes in 2u32..10u32,
edge_pairs in proptest::collection::vec(
(any::<u32>(), any::<u32>()), 0..20usize
),
recip_pairs in proptest::collection::vec(
(any::<u32>(), any::<u32>()), 0..8usize
),
n_hops in 1u32..4u32, use_filter in any::<bool>(),
threshold in 0i64..8i64, limit in 1u64..10u64,
skip in 0u64..4u64,
) {
let mut fx = Fx::new();
let mut node_ids = Vec::new();
for idx in 0..n_nodes {
let id = fx.add("N", &format!("n{idx}"), vec![("v", i(idx as i64 % 8))]);
node_ids.push(id);
}
let n = node_ids.len();
for (si, di) in &edge_pairs {
let si = (*si as usize) % n;
let di = (*di as usize) % n;
if si != di {
fx.edge("T", node_ids[si], node_ids[di], vec![]);
}
}
for (si, di) in &recip_pairs {
let si = (*si as usize) % n;
let di = (*di as usize) % n;
if si != di {
fx.edge("T", node_ids[si], node_ids[di], vec![]);
fx.edge("T", node_ids[di], node_ids[si], vec![]);
}
}
let v = fx.view();
let params = BTreeMap::new();
let var_names = ["a", "b", "c", "d"];
let hop_count = n_hops as usize;
let mut pattern = format!("({}:N)", var_names[0]);
for h in 0..hop_count {
pattern.push_str(&format!("-[:T]->({}", var_names[h + 1]));
if h + 1 == hop_count {
pattern.push_str(":N)");
} else {
pattern.push(')');
}
}
let last_var = var_names[hop_count];
let where_clause = if use_filter {
format!(" WHERE {last_var}.v > {threshold}")
} else {
String::new()
};
let ret_vars: Vec<&str> = var_names[..=hop_count].to_vec();
let ret_clause = ret_vars.join(", ");
let full_q = format!("MATCH {pattern}{where_clause} RETURN {ret_clause}");
let bounded_q = format!(
"MATCH {pattern}{where_clause} RETURN {ret_clause} SKIP {skip} LIMIT {limit}"
);
let full_plan = compile(&full_q);
let unbounded = super::with_max_intermediate_rows(100_000, || {
super::execute_unbounded(&v, &full_plan, &Params(¶ms))
});
let unbounded = match unbounded {
Ok(rs) => rs,
Err(_) => return Ok(()),
};
let total = unbounded.len();
let full_rows = rows_of(&unbounded);
let bounded = super::with_max_intermediate_rows(100_000, || {
run(&v, &bounded_q, ¶ms)
}).expect("bounded must not error");
let s = (skip as usize).min(total);
let e = (skip as usize + limit as usize).min(total);
prop_assert_eq!(
rows_of(&bounded),
full_rows[s..e].to_vec(),
"hops={} filter={} threshold={} SKIP {} LIMIT {}: \
bounded != unbounded[{}..{}]",
hop_count, use_filter, threshold, skip, limit, s, e
);
}
}
proptest! {
#[test]
fn prop_scan_filter_fused_equals_unbounded_slice(
n_nodes in 0u32..20u32,
prop_mask in any::<u32>(),
float_mask in any::<u32>(),
threshold in -1i64..8i64,
op_idx in 0u32..6u32,
skip in 0u64..5u64,
limit in 1u64..8u64,
) {
let op_str = match op_idx {
0 => "=",
1 => "<>",
2 => "<",
3 => "<=",
4 => ">",
_ => ">=",
};
let mut fx = Fx::new();
for idx in 0..n_nodes {
let has_prop = (prop_mask >> (idx % 32)) & 1 == 1;
let use_float = (float_mask >> (idx % 32)) & 1 == 1;
let props: Vec<(&str, Value)> = if has_prop {
if use_float {
vec![("v", f(idx as f64 % 7.0))]
} else {
vec![("v", i(idx as i64 % 7))]
}
} else {
vec![]
};
fx.add("N", &format!("n{idx}"), props);
}
let v = fx.view();
let params = BTreeMap::new();
let full_q = format!("MATCH (n:N) WHERE n.v {op_str} {threshold} RETURN n");
let bounded_q = format!(
"MATCH (n:N) WHERE n.v {op_str} {threshold} RETURN n SKIP {skip} LIMIT {limit}"
);
let full_plan = compile(&full_q);
let unbounded = super::execute_unbounded(&v, &full_plan, &Params(¶ms));
let unbounded = match unbounded {
Ok(rs) => rs,
Err(_) => return Ok(()),
};
let total = unbounded.len();
let full_rows = rows_of(&unbounded);
let fires_before =
super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
let bounded = run(&v, &bounded_q, ¶ms).expect("fused bounded must not error");
let fires_after =
super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
if n_nodes > 0 && op_idx != 0 {
prop_assert!(
fires_after > fires_before,
"fused arm did NOT fire for op={} threshold={} n_nodes={}: \
counter before={} after={}",
op_str,
threshold,
n_nodes,
fires_before,
fires_after
);
}
let s = (skip as usize).min(total);
let e = (skip as usize + limit as usize).min(total);
prop_assert_eq!(
rows_of(&bounded),
full_rows[s..e].to_vec(),
"fused path: op={} threshold={} n_nodes={} SKIP {} LIMIT {}: \
bounded != unbounded[{}..{}]",
op_str, threshold, n_nodes, skip, limit, s, e
);
let compound_q = format!(
"MATCH (n:N) WHERE n.v {} {} AND n.v >= -999 RETURN n SKIP {} LIMIT {}",
op_str, threshold, skip, limit
);
let fires_before_compound =
super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
let compound_bounded =
run(&v, &compound_q, ¶ms).expect("compound-AND bounded must not error");
let fires_after_compound =
super::FUSED_SCAN_FIRES.load(std::sync::atomic::Ordering::Relaxed);
prop_assert_eq!(
fires_after_compound,
fires_before_compound,
"fused arm fired for compound-AND shape (should use generic path): \
op={} threshold={} n_nodes={}",
op_str,
threshold,
n_nodes
);
prop_assert_eq!(
rows_of(&compound_bounded),
full_rows[s..e].to_vec(),
"compound-AND fallback: op={} threshold={} n_nodes={} SKIP {} LIMIT {}: \
result differs from fused",
op_str, threshold, n_nodes, skip, limit
);
}
}
#[test]
fn dense_hop1_with_filter_survives_pull() {
const LEAVES: usize = 120; const CAP: usize = 100;
let mut fx = Fx::new();
let src = fx.add("Src", "src", vec![]);
for idx in 0..LEAVES {
let leaf = fx.add("Leaf", &format!("l{idx}"), vec![("v", i(idx as i64))]);
fx.edge("T", src, leaf, vec![]);
}
let v = fx.view();
let params = BTreeMap::new();
let staged_err = super::with_max_intermediate_rows(CAP, || {
super::execute_unbounded(
&v,
&compile("MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v >= 110 RETURN l, l.v"),
&Params(¶ms),
)
});
assert!(
staged_err.is_err(),
"staged path must error on 120 leaves with cap={CAP}"
);
assert!(
staged_err
.unwrap_err()
.contains("intermediate result exceeds"),
"wrong error message"
);
let ok = super::with_max_intermediate_rows(CAP, || {
run(
&v,
"MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v >= 110 RETURN l, l.v LIMIT 5",
¶ms,
)
});
let rs = ok.expect("pull-based must survive despite dense hop-1 exceeding cap");
assert_eq!(rs.len(), 5, "LIMIT 5 must return exactly 5 rows");
let vs: Vec<i64> = (0..rs.len())
.filter_map(|i| match rs.get(i, "l.v") {
Some(Value::Int(n)) => Some(*n),
_ => None,
})
.collect();
assert_eq!(vs.len(), 5, "all projected rows must have v");
for v_val in &vs {
assert!(*v_val >= 110, "filter must hold: v={v_val} is not >= 110");
}
let (pull_result, pull_produced) = super::with_expand_counter(|| {
super::with_max_intermediate_rows(1_000_000, || {
run(
&v,
"MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v < 10 RETURN l LIMIT 5",
¶ms,
)
})
});
pull_result.expect("pull must succeed without cap");
assert!(
pull_produced <= 5,
"pull expand count {pull_produced} should be ≤ 5 (stops after 5 passing leaves)"
);
let (staged_result, staged_produced) = super::with_expand_counter(|| {
super::with_max_intermediate_rows(1_000_000, || {
super::execute_unbounded(
&v,
&compile("MATCH (s:Src)-[:T]->(l:Leaf) WHERE l.v < 10 RETURN l"),
&Params(¶ms),
)
})
});
staged_result.expect("staged must succeed with 1M cap");
assert_eq!(
staged_produced, LEAVES,
"staged must expand all {LEAVES} leaves"
);
assert!(
staged_produced >= pull_produced * 10,
"staged ({staged_produced}) must be ≥ 10× pull ({pull_produced})"
);
}
#[test]
fn harness_shape_two_hop_dense_survives_pull() {
const N_TALENT: usize = 70;
const N_COMPANY: usize = 20;
const N_INDUSTRY: usize = 3;
const CAP: usize = 100;
const LIMIT: usize = 10;
let mut fx = Fx::new();
let mut talent_ids: Vec<u32> = Vec::new();
let mut talent_industry: Vec<usize> = Vec::new();
for i in 0..N_TALENT {
let ind = i % N_INDUSTRY;
let id = fx.add(
"Talent",
&format!("t{i}"),
vec![("industry", s(&ind.to_string()))],
);
talent_ids.push(id);
talent_industry.push(ind);
}
let mut company_ids: Vec<u32> = Vec::new();
let mut company_industry: Vec<usize> = Vec::new();
for i in 0..N_COMPANY {
let ind = i % N_INDUSTRY;
let id = fx.add(
"Company",
&format!("c{i}"),
vec![("industry", s(&ind.to_string()))],
);
company_ids.push(id);
company_industry.push(ind);
}
for (ti, &tid) in talent_ids.iter().enumerate() {
for (ci, &cid) in company_ids.iter().enumerate() {
if talent_industry[ti] == company_industry[ci] {
fx.edge("INDUSTRY_ALIGNMENT", tid, cid, vec![]);
}
}
}
let v = fx.view();
let params = BTreeMap::new();
let query = format!(
"MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company)\
<-[:INDUSTRY_ALIGNMENT]-(t2:Talent) RETURN t, c, t2 LIMIT {LIMIT}"
);
const UNBOUNDED_Q: &str = "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company)\
<-[:INDUSTRY_ALIGNMENT]-(t2:Talent) RETURN t, c, t2";
let staged_err = super::with_max_intermediate_rows(CAP, || {
super::execute_unbounded(&v, &compile(UNBOUNDED_Q), &Params(¶ms))
});
assert!(
staged_err.is_err(),
"staged must error with cap={CAP} on harness-shape graph"
);
assert!(
staged_err
.unwrap_err()
.contains("intermediate result exceeds"),
"wrong error"
);
let ok = super::with_max_intermediate_rows(CAP, || run(&v, &query, ¶ms));
let rs = ok.expect("pull-based must complete on harness-shape with LIMIT 10");
assert_eq!(rs.len(), LIMIT, "must return exactly {LIMIT} rows");
for i in 0..rs.len() {
let row = rs.row(i);
assert_eq!(row.len(), 3, "each row must have 3 columns (t, c, t2)");
assert!(row.iter().all(|c| c.is_some()), "all cells must be Some");
}
}
#[test]
fn intermediate_row_cap_errors_on_scan_and_expand() {
let cap_msg = |n: usize| {
format!(
"intermediate result exceeds {n} rows; add a LIMIT or constrain patterns with shared variables"
)
};
let mut scan_fx = Fx::new();
scan_fx.add("N", "a", vec![]);
scan_fx.add("N", "b", vec![]);
scan_fx.add("N", "c", vec![]);
let sv = scan_fx.view();
let scan_err = super::with_max_intermediate_rows(2, || {
run(&sv, "MATCH (n:N) RETURN n", &BTreeMap::new())
})
.expect_err("3-row scan must exceed cap 2");
assert_eq!(scan_err, cap_msg(2));
let mut exp_fx = Fx::new();
let src = exp_fx.add("Src", "s", vec![]);
let d1 = exp_fx.add("Dst", "d1", vec![]);
let d2 = exp_fx.add("Dst", "d2", vec![]);
exp_fx.edge("T", src, d1, vec![]);
exp_fx.edge("T", src, d2, vec![]);
let ev = exp_fx.view();
let exp_err = super::with_max_intermediate_rows(1, || {
run(&ev, "MATCH (x:Src)-[:T]->(y) RETURN x, y", &BTreeMap::new())
})
.expect_err("2-row expand must exceed cap 1");
assert_eq!(exp_err, cap_msg(1));
}
#[test]
fn bounded_matches_unbounded_slice_various_limits() {
let fx = hop_graph();
let v = fx.view();
let params = BTreeMap::new();
let full_plan = compile("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b");
let full_rs = super::execute_unbounded(&v, &full_plan, &Params(¶ms)).unwrap();
let full_rows = rows_of(&full_rs);
assert_eq!(full_rows.len(), 3, "hop_graph has exactly 3 KNOWS paths");
for limit in [1u64, 2, 3, 10] {
let q = format!("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b LIMIT {limit}");
let rs = run(&v, &q, ¶ms).unwrap();
let expected_len = (limit as usize).min(full_rows.len());
assert_eq!(
rs.len(),
expected_len,
"LIMIT {limit}: expected {expected_len} rows, got {}",
rs.len()
);
assert_eq!(
rows_of(&rs),
full_rows[..expected_len],
"LIMIT {limit}: rows differ from reference slice"
);
}
let skip_rs = run(
&v,
"MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b SKIP 1 LIMIT 2",
¶ms,
)
.unwrap();
assert_eq!(rows_of(&skip_rs), full_rows[1..3]);
let mut fx2 = Fx::new();
fx2.add("N", "a", vec![("v", i(1))]);
fx2.add("N", "b", vec![("v", i(2))]);
fx2.add("N", "c", vec![("v", i(3))]);
let v2 = fx2.view();
let filter_full = super::execute_unbounded(
&v2,
&compile("MATCH (n:N) WHERE n.v > 1 RETURN n"),
&Params(¶ms),
)
.unwrap();
assert_eq!(filter_full.len(), 2);
let filter_lim = run(&v2, "MATCH (n:N) WHERE n.v > 1 RETURN n LIMIT 1", ¶ms).unwrap();
assert_eq!(filter_lim.len(), 1, "LIMIT 1 on filter query");
assert_eq!(rows_of(&filter_lim), rows_of(&filter_full)[..1]);
let tri = triangle();
let tv = tri.view();
let tri_full = super::execute_unbounded(
&tv,
&compile("MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z"),
&Params(¶ms),
)
.unwrap();
assert_eq!(tri_full.len(), 3, "triangle has 3 unique two-hop paths");
for limit in [1u64, 2, 3, 5] {
let q = format!("MATCH (x)-[r1:T]->(y)-[r2:T]->(z) RETURN x, y, z LIMIT {limit}");
let rs = run(&tv, &q, ¶ms).unwrap();
let expected_len = (limit as usize).min(3);
assert_eq!(
rs.len(),
expected_len,
"triangle LIMIT {limit}: got {} rows",
rs.len()
);
assert_eq!(
rows_of(&rs),
rows_of(&tri_full)[..expected_len],
"triangle LIMIT {limit}: rows differ"
);
}
}
#[test]
fn expand_terminates_early_with_row_bound() {
const LEAVES: usize = 500;
let mut fx = Fx::new();
let hub = fx.add("Hub", "hub", vec![]);
for i in 0..LEAVES {
let leaf = fx.add("Leaf", &format!("leaf-{i}"), vec![]);
fx.edge("T", hub, leaf, vec![]);
}
let v = fx.view();
let params = BTreeMap::new();
let full_plan = compile("MATCH (h:Hub)-[:T]->(x:Leaf) RETURN x");
let (bounded_result, bounded_produced) = super::with_expand_counter(|| {
run(&v, "MATCH (h:Hub)-[:T]->(x:Leaf) RETURN x LIMIT 5", ¶ms)
});
let bounded_rs = bounded_result.unwrap();
assert_eq!(bounded_rs.len(), 5, "LIMIT 5 must return exactly 5 rows");
assert!(
bounded_produced <= 5,
"bounded: exec_expand emitted {bounded_produced} rows, expected ≤ 5"
);
let (unbounded_result, unbounded_produced) = super::with_expand_counter(|| {
super::execute_unbounded(&v, &full_plan, &Params(¶ms))
});
let unbounded_rs = unbounded_result.unwrap();
assert_eq!(
unbounded_rs.len(),
LEAVES,
"unbounded must return all {LEAVES} rows"
);
assert_eq!(
unbounded_produced, LEAVES,
"unbounded: exec_expand must emit all {LEAVES} rows"
);
assert!(
unbounded_produced >= bounded_produced * 100,
"unbounded ({unbounded_produced}) must be ≥ 100× bounded ({bounded_produced})"
);
}
#[test]
fn bounded_query_survives_low_intermediate_row_cap() {
let mut fx = Fx::new();
let src = fx.add("Src", "s", vec![]);
for i in 0..20usize {
let dst = fx.add("Dst", &format!("d{i}"), vec![]);
fx.edge("T", src, dst, vec![]);
}
let v = fx.view();
let params = BTreeMap::new();
let full_plan = compile("MATCH (s:Src)-[:T]->(d:Dst) RETURN d");
let cap_err = super::with_max_intermediate_rows(10, || {
super::execute_unbounded(&v, &full_plan, &Params(¶ms))
});
assert!(
cap_err.is_err(),
"unbounded must hit the intermediate-row cap"
);
assert!(
cap_err.unwrap_err().contains("intermediate result exceeds"),
"cap error message must be the budget message"
);
let ok = super::with_max_intermediate_rows(10, || {
run(&v, "MATCH (s:Src)-[:T]->(d:Dst) RETURN d LIMIT 5", ¶ms)
});
assert_eq!(
ok.unwrap().len(),
5,
"bounded (LIMIT 5) must complete with 5 rows, not a cap error"
);
let at_cap = super::with_max_intermediate_rows(10, || {
run(&v, "MATCH (s:Src)-[:T]->(d:Dst) RETURN d LIMIT 10", ¶ms)
});
assert_eq!(
at_cap.unwrap().len(),
10,
"bounded at LIMIT==cap must complete with 10 rows"
);
}
#[test]
fn count_star_returns_total_node_count() {
let mut fx = Fx::new();
fx.add("Person", "ada", vec![]);
fx.add("Person", "bob", vec![]);
fx.add("Person", "cam", vec![]);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(&v, "MATCH (n:Person) RETURN COUNT(*)", ¶ms).expect("COUNT(*)");
assert_eq!(rs.columns(), &["COUNT(*)".to_string()]);
assert_eq!(rs.len(), 1);
assert_eq!(rs.row(0), &[Some(i(3))]);
let rs_empty = run(&v, "MATCH (n:Ghost) RETURN COUNT(*)", ¶ms).expect("COUNT(*) empty");
assert_eq!(rs_empty.row(0), &[Some(i(0))]);
}
#[test]
fn count_star_alias_sets_column_name() {
let mut fx = Fx::new();
fx.add("N", "a", vec![]);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(&v, "MATCH (n:N) RETURN COUNT(*) AS total", ¶ms).expect("COUNT AS");
assert_eq!(rs.columns(), &["total".to_string()]);
assert_eq!(rs.row(0), &[Some(i(1))]);
}
#[test]
fn count_var_skips_null_node_bindings() {
let mut fx = Fx::new();
fx.add("N", "a", vec![]);
fx.add("N", "b", vec![]);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(&v, "MATCH (n:N) RETURN COUNT(n)", ¶ms).expect("COUNT(n)");
assert_eq!(rs.columns(), &["COUNT(n)".to_string()]);
assert_eq!(rs.row(0), &[Some(i(2))]);
}
#[test]
fn sum_numeric_prop_ignores_null_and_non_numeric() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("v", i(10))]);
fx.add("N", "b", vec![("v", i(20))]);
fx.add("N", "c", vec![]); let v = fx.view();
let params = BTreeMap::new();
let rs = run(&v, "MATCH (n:N) RETURN SUM(n.v)", ¶ms).expect("SUM");
assert_eq!(rs.columns(), &["SUM(n.v)".to_string()]);
assert_eq!(rs.row(0), &[Some(f(30.0))]);
let rs_null = run(&v, "MATCH (n:N) RETURN SUM(n.missing)", ¶ms).expect("SUM null");
assert_eq!(rs_null.row(0), &[None]);
}
#[test]
fn avg_numeric_prop() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("v", i(10))]);
fx.add("N", "b", vec![("v", i(30))]);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(&v, "MATCH (n:N) RETURN AVG(n.v) AS avg_v", ¶ms).expect("AVG");
assert_eq!(rs.columns(), &["avg_v".to_string()]);
assert_eq!(rs.row(0), &[Some(f(20.0))]);
let rs_empty = run(&v, "MATCH (n:Ghost) RETURN AVG(n.v)", ¶ms).expect("AVG empty");
assert_eq!(rs_empty.row(0), &[None]);
}
#[test]
fn min_max_numeric_prop() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("v", i(5))]);
fx.add("N", "b", vec![("v", i(1))]);
fx.add("N", "c", vec![("v", i(9))]);
fx.add("N", "d", vec![]); let v = fx.view();
let params = BTreeMap::new();
let min_rs = run(&v, "MATCH (n:N) RETURN MIN(n.v)", ¶ms).expect("MIN");
assert_eq!(min_rs.row(0), &[Some(i(1))]);
let max_rs = run(&v, "MATCH (n:N) RETURN MAX(n.v)", ¶ms).expect("MAX");
assert_eq!(max_rs.row(0), &[Some(i(9))]);
}
#[test]
fn min_max_mixed_int_float_props() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("v", i(3))]);
fx.add("N", "b", vec![("v", f(1.5))]);
fx.add("N", "c", vec![("v", i(7))]);
fx.add("N", "d", vec![("v", f(2.0))]);
let v = fx.view();
let params = BTreeMap::new();
let min_rs = run(&v, "MATCH (n:N) RETURN MIN(n.v)", ¶ms).expect("MIN mixed");
assert_eq!(min_rs.row(0), &[Some(f(1.5))]);
let max_rs = run(&v, "MATCH (n:N) RETURN MAX(n.v)", ¶ms).expect("MAX mixed");
assert_eq!(max_rs.row(0), &[Some(i(7))]);
}
#[test]
fn aggregate_limit_skip_order_by_are_no_ops() {
let mut fx = Fx::new();
fx.add("N", "a", vec![]);
fx.add("N", "b", vec![]);
fx.add("N", "c", vec![]);
let v = fx.view();
let params = BTreeMap::new();
let rs_lim5 =
run(&v, "MATCH (n:N) RETURN COUNT(*) LIMIT 5", ¶ms).expect("COUNT(*) LIMIT 5");
assert_eq!(
rs_lim5.len(),
1,
"aggregate with LIMIT 5 must still return 1 row"
);
assert_eq!(rs_lim5.row(0), &[Some(i(3))]);
let rs_lim0 =
run(&v, "MATCH (n:N) RETURN COUNT(*) LIMIT 0", ¶ms).expect("COUNT(*) LIMIT 0");
assert_eq!(
rs_lim0.len(),
1,
"aggregate with LIMIT 0 must still return 1 row"
);
assert_eq!(rs_lim0.row(0), &[Some(i(3))]);
let rs_skip =
run(&v, "MATCH (n:N) RETURN COUNT(*) SKIP 100", ¶ms).expect("COUNT(*) SKIP 100");
assert_eq!(
rs_skip.len(),
1,
"aggregate with large SKIP must still return 1 row"
);
let rs_ord = plan_src("MATCH (n:N) RETURN COUNT(*) ORDER BY n");
assert!(
rs_ord.is_ok(),
"COUNT(*) ORDER BY should plan without error (ORDER BY dropped)"
);
let plan_ops = rs_ord.unwrap();
assert!(
!plan_ops
.iter()
.any(|op| matches!(op, crate::cypher::plan::PlanOp::OrderBy { .. })),
"aggregate plan must not contain OrderBy"
);
}
#[test]
fn count_star_no_budget_cap_applies() {
let mut fx = Fx::new();
let src = fx.add("Src", "s", vec![]);
for i in 0..30usize {
let dst = fx.add("Dst", &format!("d{i}"), vec![]);
fx.edge("T", src, dst, vec![]);
}
let v = fx.view();
let params = BTreeMap::new();
let cap_err =
super::with_max_intermediate_rows(10, || run(&v, "MATCH (n:Dst) RETURN n", ¶ms));
assert!(
cap_err.is_err(),
"staged path must error on 30 nodes with cap=10"
);
let agg_ok = super::with_max_intermediate_rows(10, || {
run(&v, "MATCH (n:Dst) RETURN COUNT(*)", ¶ms)
})
.expect("aggregate must not hit the intermediate-row cap");
assert_eq!(
agg_ok.row(0),
&[Some(i(30))],
"COUNT(*) must count all 30 nodes regardless of cap"
);
}
fn plan_src(src: &str) -> Result<Vec<crate::cypher::plan::PlanOp>, String> {
use crate::cypher::{lex, parse, plan};
let toks = lex(src).map_err(|e| format!("lex: {e}"))?;
let ast = parse(&toks).map_err(|e| format!("parse: {e}"))?;
plan(&ast).map_err(|e| format!("plan: {e}"))
}
#[test]
fn grouped_aggregation_plan_routing() {
use crate::cypher::plan::PlanOp;
let ops = plan_src("MATCH (a:N) RETURN a, COUNT(*)")
.expect("grouped aggregation must now succeed");
assert!(
ops.iter()
.any(|op| matches!(op, PlanOp::GroupAggregate { .. })),
"grouped aggregation plan must contain GroupAggregate op, got: {ops:?}"
);
let ops2 = plan_src("MATCH (a:N) RETURN COUNT(*), COUNT(a)")
.expect("multi-aggregate must now succeed");
assert!(
ops2.iter()
.any(|op| matches!(op, PlanOp::GroupAggregate { .. })),
"multi-aggregate plan must contain GroupAggregate op, got: {ops2:?}"
);
let err3 = plan_src("MATCH (a:N) RETURN SUM(*)").expect_err("SUM(*) must be plan error");
assert!(
err3.to_ascii_lowercase().contains("sum") || err3.to_ascii_lowercase().contains("*"),
"error must mention SUM or *, got: {err3}"
);
}
#[test]
fn grouped_single_key_count() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("t", s("X"))]);
fx.add("N", "b", vec![("t", s("X"))]);
fx.add("N", "c", vec![("t", s("Y"))]);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
.expect("single-key grouped COUNT must succeed");
assert_eq!(
rs.columns(),
&["n.t".to_string(), "cnt".to_string()],
"columns must match RETURN clause"
);
assert_eq!(rs.len(), 2, "must produce exactly 2 groups (X and Y)");
let find = |label: &Value| (0..rs.len()).find(|&i| rs.row(i)[0].as_ref() == Some(label));
let xi = find(&s("X")).expect("group X must exist");
let yi = find(&s("Y")).expect("group Y must exist");
assert_eq!(rs.row(xi)[1], Some(i(2)), "X group count must be 2");
assert_eq!(rs.row(yi)[1], Some(i(1)), "Y group count must be 1");
}
#[test]
fn grouped_two_keys_sum_avg() {
let mut fx = Fx::new();
fx.add(
"N",
"a",
vec![("cat", s("A")), ("sub", s("1")), ("v", i(10))],
);
fx.add(
"N",
"b",
vec![("cat", s("A")), ("sub", s("1")), ("v", i(20))],
);
fx.add(
"N",
"c",
vec![("cat", s("A")), ("sub", s("2")), ("v", i(5))],
);
fx.add(
"N",
"d",
vec![("cat", s("B")), ("sub", s("1")), ("v", i(100))],
);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(
&v,
"MATCH (n:N) RETURN n.cat, n.sub, SUM(n.v) AS total, AVG(n.v) AS avg_v",
¶ms,
)
.expect("two-key SUM + AVG must succeed");
assert_eq!(
rs.columns(),
&[
"n.cat".to_string(),
"n.sub".to_string(),
"total".to_string(),
"avg_v".to_string()
]
);
assert_eq!(rs.len(), 3, "must produce 3 groups: (A,1), (A,2), (B,1)");
let find = |cat: &Value, sub: &Value| {
(0..rs.len())
.find(|&i| rs.row(i)[0].as_ref() == Some(cat) && rs.row(i)[1].as_ref() == Some(sub))
};
let a1 = find(&s("A"), &s("1")).expect("group (A,1) must exist");
assert_eq!(rs.row(a1)[2], Some(f(30.0)), "(A,1) SUM must be 30.0");
assert_eq!(rs.row(a1)[3], Some(f(15.0)), "(A,1) AVG must be 15.0");
let a2 = find(&s("A"), &s("2")).expect("group (A,2) must exist");
assert_eq!(rs.row(a2)[2], Some(f(5.0)), "(A,2) SUM must be 5.0");
let b1 = find(&s("B"), &s("1")).expect("group (B,1) must exist");
assert_eq!(rs.row(b1)[2], Some(f(100.0)), "(B,1) SUM must be 100.0");
assert_eq!(rs.row(b1)[3], Some(f(100.0)), "(B,1) AVG must be 100.0");
}
#[test]
fn grouped_order_by_count_desc_limit() {
let mut fx = Fx::new();
fx.add("N", "a1", vec![("cat", s("A"))]);
fx.add("N", "a2", vec![("cat", s("A"))]);
fx.add("N", "a3", vec![("cat", s("A"))]);
fx.add("N", "b1", vec![("cat", s("B"))]);
fx.add("N", "b2", vec![("cat", s("B"))]);
fx.add("N", "c1", vec![("cat", s("C"))]);
fx.add("N", "d1", vec![("cat", s("D"))]);
fx.add("N", "d2", vec![("cat", s("D"))]);
fx.add("N", "d3", vec![("cat", s("D"))]);
fx.add("N", "d4", vec![("cat", s("D"))]);
fx.add("N", "e1", vec![("cat", s("E"))]);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(
&v,
"MATCH (n:N) RETURN n.cat, COUNT(*) AS cnt ORDER BY cnt DESC LIMIT 3",
¶ms,
)
.expect("ORDER BY count DESC LIMIT 3 must succeed");
assert_eq!(rs.len(), 3, "LIMIT 3 must return exactly 3 groups");
assert_eq!(rs.row(0)[1], Some(i(4)), "row 0 must be count 4");
assert_eq!(rs.row(0)[0], Some(s("D")), "row 0 must be category D");
assert_eq!(rs.row(1)[1], Some(i(3)), "row 1 must be count 3");
assert_eq!(rs.row(1)[0], Some(s("A")), "row 1 must be category A");
assert_eq!(rs.row(2)[1], Some(i(2)), "row 2 must be count 2");
assert_eq!(rs.row(2)[0], Some(s("B")), "row 2 must be category B");
let plan_ops =
plan_src("MATCH (n:N) RETURN n.cat, COUNT(*) AS cnt ORDER BY cnt DESC LIMIT 3")
.expect("plan must succeed");
assert_eq!(
crate::cypher::plan::row_bound(&plan_ops),
None,
"GroupAggregate plan with LIMIT must have row_bound = None"
);
}
#[test]
fn grouped_empty_input_yields_zero_groups() {
let fx = Fx::new(); let v = fx.view();
let params = BTreeMap::new();
let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
.expect("grouped aggregate on empty graph must succeed");
assert_eq!(rs.len(), 0, "empty input must yield zero groups");
assert_eq!(
rs.columns(),
&["n.t".to_string(), "cnt".to_string()],
"columns must still be present even with zero rows"
);
}
#[test]
fn grouped_null_key_groups_together() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("t", s("X"))]);
fx.add("N", "b", vec![]); fx.add("N", "c", vec![]); fx.add("N", "d", vec![("t", s("Y"))]);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
.expect("null-key grouped aggregate must succeed");
assert_eq!(rs.len(), 3, "must produce 3 groups: X, null, Y");
let null_row = (0..rs.len())
.find(|&i| rs.row(i)[0].is_none())
.expect("null group must be present");
assert_eq!(
rs.row(null_row)[1],
Some(i(2)),
"null group must count 2 rows (b and c)"
);
let x_row = (0..rs.len())
.find(|&i| rs.row(i)[0] == Some(s("X")))
.expect("X group must exist");
assert_eq!(rs.row(x_row)[1], Some(i(1)));
let y_row = (0..rs.len())
.find(|&i| rs.row(i)[0] == Some(s("Y")))
.expect("Y group must exist");
assert_eq!(rs.row(y_row)[1], Some(i(1)));
}
#[test]
fn grouped_cap_error_on_high_cardinality() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("t", s("A"))]);
fx.add("N", "b", vec![("t", s("B"))]);
fx.add("N", "c", vec![("t", s("C"))]);
let v = fx.view();
let params = BTreeMap::new();
let err = super::with_max_groups(2, || {
run(&v, "MATCH (n:N) RETURN n.t, COUNT(*) AS cnt", ¶ms)
})
.expect_err("must error when group count exceeds cap");
assert!(
err.to_ascii_lowercase().contains("group count"),
"error must mention group count, got: {err}"
);
}
#[test]
fn multi_aggregate_no_keys() {
let mut fx = Fx::new();
fx.add("N", "a", vec![]);
fx.add("N", "b", vec![]);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
.expect("multi-aggregate no keys must succeed");
assert_eq!(rs.len(), 1, "must produce exactly one result row");
assert_eq!(rs.row(0)[0], Some(i(2)), "COUNT(*) must be 2");
assert_eq!(rs.row(0)[1], Some(i(2)), "COUNT(n) must be 2");
}
#[test]
fn aggregate_over_hop_counts_edges() {
let fx = hop_graph();
let v = fx.view();
let params = BTreeMap::new();
let rs = run(
&v,
"MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN COUNT(*)",
¶ms,
)
.expect("COUNT(*) on hop graph");
assert_eq!(rs.row(0), &[Some(i(3))]);
}
#[test]
fn multi_aggregate_no_keys_empty_graph() {
let fx = Fx::new(); let v = fx.view();
let params = BTreeMap::new();
let rs = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
.expect("empty-graph multi-agg must succeed");
assert_eq!(rs.len(), 1, "must produce exactly 1 row on empty input");
assert_eq!(
rs.row(0)[0],
Some(i(0)),
"COUNT(*) on empty graph must be 0"
);
assert_eq!(
rs.row(0)[1],
Some(i(0)),
"COUNT(n) on empty graph must be 0"
);
}
#[test]
fn fast_path_and_group_path_agree_on_empty_input() {
let fx = Fx::new();
let v = fx.view();
let params = BTreeMap::new();
let fast = run(&v, "MATCH (n:N) RETURN COUNT(*)", ¶ms)
.expect("fast-path COUNT on empty graph");
assert_eq!(fast.len(), 1, "fast path: 1 row on empty input");
let fast_count = fast.row(0)[0].clone();
let grouped = run(&v, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
.expect("group-path COUNT on empty graph");
assert_eq!(grouped.len(), 1, "group path: 1 row on empty input");
let group_count = grouped.row(0)[0].clone();
assert_eq!(
fast_count, group_count,
"fast path and group path must agree on COUNT(*) for empty input"
);
let mut fx2 = Fx::new();
fx2.add("N", "x", vec![]);
fx2.add("N", "y", vec![]);
let v2 = fx2.view();
let fast2 = run(&v2, "MATCH (n:N) RETURN COUNT(*)", ¶ms)
.expect("fast-path COUNT on 2-node graph");
assert_eq!(fast2.row(0)[0], Some(i(2)), "fast path: COUNT(*) = 2");
let grouped2 = run(&v2, "MATCH (n:N) RETURN COUNT(*), COUNT(n)", ¶ms)
.expect("group-path COUNT on 2-node graph");
assert_eq!(grouped2.row(0)[0], Some(i(2)), "group path: COUNT(*) = 2");
assert_eq!(grouped2.row(0)[1], Some(i(2)), "group path: COUNT(n) = 2");
}
#[test]
fn grouped_int_float_key_unification() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("score", i(1))]); fx.add("N", "b", vec![("score", f(1.0))]);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(&v, "MATCH (n:N) RETURN n.score, COUNT(*) AS cnt", ¶ms)
.expect("int/float unification must succeed");
assert_eq!(
rs.len(),
1,
"Int(1) and Float(1.0) must group together into 1 group"
);
assert_eq!(
rs.row(0)[0],
Some(i(1)),
"key column must display Int(1) (first-seen)"
);
assert_eq!(rs.row(0)[1], Some(i(2)), "unified group must have count=2");
}
#[test]
fn distinct_collapses_duplicate_projected_values() {
let mut fx = Fx::new();
fx.add("N", "a1", vec![("city", s("Austin"))]);
fx.add("N", "a2", vec![("city", s("Austin"))]);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(&v, "MATCH (n:N) RETURN DISTINCT n.city", ¶ms).unwrap();
assert_eq!(rs.len(), 1);
assert_eq!(rs.row(0)[0], Some(s("Austin")));
}
#[test]
fn distinct_int_float_unify() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("score", i(1))]);
fx.add("N", "b", vec![("score", f(1.0))]);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(&v, "MATCH (n:N) RETURN DISTINCT n.score", ¶ms).unwrap();
assert_eq!(
rs.len(),
1,
"Int(1) and Float(1.0) must DISTINCT as one row"
);
assert_eq!(rs.row(0)[0], Some(i(1)), "first-seen Int(1) wins display");
}
#[test]
fn distinct_caps_at_intermediate_row_budget() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("city", s("Austin"))]);
fx.add("N", "b", vec![("city", s("Paris"))]);
let v = fx.view();
let params = BTreeMap::new();
let err = super::with_max_intermediate_rows(1, || {
run(&v, "MATCH (n:N) RETURN DISTINCT n.city", ¶ms)
})
.expect_err("two distinct cities must exceed cap=1");
assert!(
err.contains("1") || err.contains("row"),
"cap error must mention the budget, got: {err}"
);
}
#[test]
fn where_in_list_filters_rows() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("city", s("Austin"))]);
fx.add("N", "p", vec![("city", s("Paris"))]);
fx.add("N", "l", vec![("city", s("London"))]);
let v = fx.view();
let mut params = BTreeMap::new();
params.insert("c".into(), s("Paris"));
let rs = run(
&v,
"MATCH (n:N) WHERE n.city IN ['Austin', $c] RETURN n.city",
¶ms,
)
.unwrap();
assert_eq!(rs.len(), 2);
}
#[test]
fn pure_int_keys_display_as_int() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("age", i(10))]);
fx.add("N", "b", vec![("age", i(20))]);
fx.add("N", "c", vec![("age", i(10))]);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(
&v,
"MATCH (n:N) RETURN n.age, COUNT(*) AS cnt ORDER BY n.age",
¶ms,
)
.expect("pure-Int group keys must succeed");
assert_eq!(rs.len(), 2, "must have 2 groups: age=10 and age=20");
assert_eq!(
rs.row(0)[0],
Some(i(10)),
"age=10 key column must display as Int(10)"
);
assert_eq!(rs.row(0)[1], Some(i(2)), "age=10 group has 2 nodes");
assert_eq!(
rs.row(1)[0],
Some(i(20)),
"age=20 key column must display as Int(20)"
);
assert_eq!(rs.row(1)[1], Some(i(1)), "age=20 group has 1 node");
}
#[test]
fn var_expand_group_aggregate_staged_path() {
let mut fx = Fx::new();
let a = fx.add("N", "a", vec![("key", s("a"))]);
let b = fx.add("N", "b", vec![("key", s("b"))]);
let c = fx.add("N", "c", vec![("key", s("c"))]);
fx.edge("T", a, b, vec![]);
fx.edge("T", b, c, vec![]);
let v = fx.view();
let params = BTreeMap::new();
let rs = run(
&v,
"MATCH (x:N)-[*1..2]->(y:N) RETURN y.key, COUNT(*) AS cnt ORDER BY y.key",
¶ms,
)
.expect("VarExpand + GroupAggregate must succeed");
assert_eq!(rs.len(), 2, "must have 2 destination groups: b and c");
assert_eq!(rs.row(0)[0], Some(s("b")), "first group key must be 'b'");
assert_eq!(rs.row(0)[1], Some(i(1)), "b is reached via 1 path");
assert_eq!(rs.row(1)[0], Some(s("c")), "second group key must be 'c'");
assert_eq!(
rs.row(1)[1],
Some(i(2)),
"c is reached via 2 paths (1-hop and 2-hop)"
);
}
#[test]
fn pull_rows_var_expand_arm_returns_named_err() {
let fx = Fx::new();
let view = fx.view();
let vars = super::VarTable {
names: vec!["a".into(), "b".into()],
};
let project_items: Vec<crate::cypher::ast::RetItem> = vec![];
let empty_params = BTreeMap::new();
let params = super::Params(&empty_params);
let ctx = super::PullCtx {
view: &view,
vars: &vars,
project_items: &project_items,
params: ¶ms,
bound: 100,
};
let ops = vec![PlanOp::VarExpand {
from: "a".into(),
rel_var: None,
etypes: vec![],
dir: crate::cypher::RelDir::Right,
to: "b".into(),
min: 1,
max: 3,
}];
let mut row = vec![None; vars.names.len()];
let mut result = Vec::new();
let err = super::pull_rows(&ctx, &ops, &mut row, &mut result)
.expect_err("VarExpand must Err in pull_rows");
assert!(
err.contains("VarExpand") && err.contains("pull executor"),
"error must name VarExpand and pull executor, got: {err}"
);
}
#[test]
fn pull_rows_shortest_path_arm_returns_named_err() {
let fx = Fx::new();
let view = fx.view();
let vars = super::VarTable {
names: vec!["a".into(), "b".into()],
};
let project_items: Vec<crate::cypher::ast::RetItem> = vec![];
let empty_params = BTreeMap::new();
let params = super::Params(&empty_params);
let ctx = super::PullCtx {
view: &view,
vars: &vars,
project_items: &project_items,
params: ¶ms,
bound: 100,
};
let ops = vec![PlanOp::ShortestPath {
from: "a".into(),
rel_var: None,
etypes: vec![],
dir: crate::cypher::RelDir::Right,
to: "b".into(),
max_hops: 5,
}];
let mut row = vec![None; vars.names.len()];
let mut result = Vec::new();
let err = super::pull_rows(&ctx, &ops, &mut row, &mut result)
.expect_err("ShortestPath must Err in pull_rows");
assert!(
err.contains("ShortestPath") && err.contains("pull executor"),
"error must name ShortestPath and pull executor, got: {err}"
);
}
fn city_graph() -> Fx {
let mut fx = Fx::new();
fx.add(
"Person",
"alice",
vec![("city", s("Boston")), ("age", i(30))],
);
fx.add("Person", "bob", vec![("city", s("Boston")), ("age", i(25))]);
fx.add(
"Person",
"carol",
vec![("city", s("Austin")), ("age", i(35))],
);
fx.add(
"Person",
"dave",
vec![("city", s("Austin")), ("age", i(28))],
);
fx.add("Person", "eve", vec![("city", s("Boston")), ("age", i(22))]);
fx
}
#[test]
fn with_aggregate_having_filters_groups() {
let fx = city_graph();
let view = fx.view();
let params = BTreeMap::new();
let rs = run(
&view,
"MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt WHERE cnt > 2 RETURN city, cnt",
¶ms,
)
.expect("WITH HAVING query must succeed");
assert_eq!(rs.len(), 1, "only Boston group has cnt > 2");
assert_eq!(rs.get(0, "city"), Some(&s("Boston")));
assert_eq!(rs.get(0, "cnt"), Some(&i(3)));
}
#[test]
fn with_order_limit_then_return() {
let fx = city_graph();
let view = fx.view();
let params = BTreeMap::new();
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)
.expect("WITH ORDER LIMIT must succeed");
assert_eq!(rs.len(), 2, "LIMIT 2");
let ages: Vec<Option<Value>> = (0..rs.len()).map(|i| rs.get(i, "age").cloned()).collect();
assert_eq!(ages[0], Some(i(35)), "oldest person first");
assert_eq!(ages[1], Some(i(30)), "second oldest");
}
#[test]
fn chained_with_stages() {
let fx = city_graph();
let view = fx.view();
let params = BTreeMap::new();
let rs = run(&view,
"MATCH (p:Person) WITH p.city AS city, p.age AS age WITH city, age WHERE age > 25 RETURN city, age",
¶ms).expect("chained WITH must succeed");
assert_eq!(rs.len(), 3, "3 people with age > 25");
}
#[test]
fn with_then_match_reentry() {
let mut fx = Fx::new();
let alice = fx.add("Person", "alice", vec![]);
let bob = fx.add("Person", "bob", vec![]);
let corp = fx.add("Company", "acme", vec![]);
fx.edge("WORKS_AT", alice, corp, vec![("years", i(5))]);
fx.edge("WORKS_AT", bob, corp, vec![("years", i(3))]);
let view = fx.view();
let params = BTreeMap::new();
let rs = run(&view,
"MATCH (p:Person)-[r:WORKS_AT]->(c:Company) WITH p, c MATCH (c)-[r2:WORKS_AT]-(colleague:Person) RETURN p, colleague",
¶ms).expect("WITH MATCH re-entry must succeed");
assert!(rs.len() >= 2, "cross join via company: got {}", rs.len());
}
#[test]
fn unwind_literal_list_produces_rows() {
let fx = city_graph();
let view = fx.view();
let params = BTreeMap::new();
let rs = run(
&view,
"MATCH (p:Person) WHERE p.city = 'Boston' UNWIND [1, 2, 3] AS x RETURN p, x",
¶ms,
)
.expect("UNWIND literal must succeed");
assert_eq!(rs.len(), 9, "UNWIND [1,2,3] × 3 Boston people = 9 rows");
let xs: Vec<Option<Value>> = (0..rs.len())
.map(|row_i| rs.get(row_i, "x").cloned())
.collect();
assert!(xs.contains(&Some(i(1))));
assert!(xs.contains(&Some(i(2))));
assert!(xs.contains(&Some(i(3))));
}
#[test]
fn unwind_list_property() {
let mut fx = Fx::new();
fx.add(
"Tag",
"post1",
vec![("tags", Value::List(vec![s("rust"), s("graph")]))],
);
fx.add("Tag", "post2", vec![("tags", Value::List(vec![s("db")]))]);
let view = fx.view();
let params = BTreeMap::new();
let rs = run(
&view,
"MATCH (p:Tag) UNWIND p.tags AS tag RETURN p, tag",
¶ms,
)
.expect("UNWIND property must succeed");
assert_eq!(rs.len(), 3, "2+1 tag elements");
}
#[test]
fn unwind_empty_list_yields_zero_rows() {
let fx = city_graph();
let view = fx.view();
let params = BTreeMap::new();
let rs = run(
&view,
"MATCH (p:Person) WHERE p.city = 'Boston' UNWIND [] AS x RETURN x",
¶ms,
)
.expect("UNWIND [] must succeed with 0 rows");
assert_eq!(rs.len(), 0, "UNWIND [] should produce 0 rows");
}
#[test]
fn unwind_non_list_is_named_error() {
let fx = city_graph();
let view = fx.view();
let params = BTreeMap::new();
let err = run(
&view,
"MATCH (p:Person) WHERE p.city = 'Boston' UNWIND p.city AS x RETURN x",
¶ms,
)
.expect_err("UNWIND non-list must error");
assert!(
err.contains("UNWIND") && err.contains("list"),
"error must mention UNWIND and list: {err}"
);
}
#[test]
fn unwind_cross_product_trips_budget() {
let mut fx = Fx::new();
for i in 0..10 {
fx.add("N", &format!("n{i}"), vec![]);
}
let view = fx.view();
let params = BTreeMap::new();
let err = super::with_max_intermediate_rows(5, || {
run(
&view,
"MATCH (n:N) UNWIND [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] AS x RETURN n, x",
¶ms,
)
})
.expect_err("must hit budget");
assert!(
err.contains("intermediate result exceeds"),
"error must mention intermediate result limit: {err}"
);
}
#[test]
fn routing_with_plan_uses_staged_path() {
use crate::cypher::plan::row_bound;
let ops_with = compile("MATCH (a) WITH a RETURN a");
assert_eq!(
row_bound(&ops_with),
None,
"WITH plan must have row_bound=None"
);
let ops_unwind = compile("MATCH (a) UNWIND [1, 2] AS x RETURN a");
assert_eq!(
row_bound(&ops_unwind),
None,
"UNWIND plan must have row_bound=None"
);
let ops_plain = compile("MATCH (a) RETURN a LIMIT 5");
assert!(
row_bound(&ops_plain).is_some(),
"plain LIMIT plan must have a row bound"
);
}
#[test]
fn unwind_null_property_yields_zero_rows_for_that_node() {
let mut fx = Fx::new();
fx.add(
"Post",
"post1",
vec![("tags", Value::List(vec![s("rust"), s("graph")]))],
);
fx.add("Post", "post2", vec![("other", Value::Str("hello".into()))]);
let view = fx.view();
let params = BTreeMap::new();
let rs = run(
&view,
"MATCH (p:Post) UNWIND p.tags AS tag RETURN tag",
¶ms,
)
.expect("UNWIND null property must succeed (not error)");
assert_eq!(
rs.len(),
2,
"null property must yield 0 rows; list property yields N rows"
);
let tag0 = rs.get(0, "tag");
let tag1 = rs.get(1, "tag");
let tags = [tag0, tag1];
assert!(tags.contains(&Some(&s("rust"))), "rust tag present");
assert!(tags.contains(&Some(&s("graph"))), "graph tag present");
}
#[test]
fn where_before_unwind_filters_nodes() {
let mut fx = Fx::new();
fx.add(
"P",
"a",
vec![
("group", Value::Str("keep".into())),
("items", Value::List(vec![Value::Int(1), Value::Int(2)])),
],
);
fx.add(
"P",
"b",
vec![
("group", Value::Str("drop".into())),
("items", Value::List(vec![Value::Int(3), Value::Int(4)])),
],
);
let view = fx.view();
let params = BTreeMap::new();
let rs = run(
&view,
"MATCH (n:P) WHERE n.group = 'keep' UNWIND n.items AS x RETURN x",
¶ms,
)
.expect("WHERE before UNWIND must succeed");
assert_eq!(rs.len(), 2, "only matching node expands; 2 items");
}
#[test]
fn where_after_unwind_filters_expanded_rows() {
let mut fx = Fx::new();
fx.add(
"N",
"n1",
vec![(
"vals",
Value::List(vec![Value::Int(1), Value::Int(3), Value::Int(5)]),
)],
);
let view = fx.view();
let params = BTreeMap::new();
let rs = run(
&view,
"MATCH (n:N) UNWIND n.vals AS x WHERE x > 2 RETURN x",
¶ms,
)
.expect("WHERE after UNWIND must succeed");
assert_eq!(rs.len(), 2, "x=3 and x=5 pass; x=1 filtered out");
let v0 = rs.get(0, "x").cloned();
let v1 = rs.get(1, "x").cloned();
let vals = [v0, v1];
assert!(vals.contains(&Some(Value::Int(3))), "x=3 present");
assert!(vals.contains(&Some(Value::Int(5))), "x=5 present");
}
#[test]
fn aggregate_with_order_by_unknown_alias_is_named_error() {
use crate::cypher::{lex, parser::parse, plan::plan};
let src = "MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt ORDER BY cntt DESC RETURN city, cnt";
let plan_result = plan(&parse(&lex(src).expect("lex")).expect("parse"));
let err = plan_result
.expect_err("typo'd ORDER BY alias in aggregate WITH must be a named plan error");
assert!(
err.contains("cntt") || err.contains("unbound"),
"error must mention the unknown alias: {err}"
);
}
#[test]
fn non_aggregate_with_order_by_unknown_alias_is_named_error() {
use crate::cypher::{lex, parser::parse, plan::plan};
let src = "MATCH (p:Person) WITH p, p.age AS age ORDER BY nope DESC RETURN p.city AS city";
let plan_result = plan(&parse(&lex(src).expect("lex")).expect("parse"));
let err = plan_result
.expect_err("typo'd ORDER BY alias in non-aggregate WITH must be a named plan error");
assert!(
err.contains("nope") || err.contains("unbound"),
"error must mention the unknown alias: {err}"
);
}
#[test]
fn post_unwind_where_references_pre_match_variable() {
let mut fx = Fx::new();
fx.add(
"N",
"n1",
vec![
("threshold", Value::Int(3)),
(
"xs",
Value::List(vec![
Value::Int(1),
Value::Int(2),
Value::Int(3),
Value::Int(4),
Value::Int(5),
]),
),
],
);
fx.add(
"N",
"n2",
vec![
("threshold", Value::Int(10)),
("xs", Value::List(vec![Value::Int(1), Value::Int(2)])),
],
);
let view = fx.view();
let params = BTreeMap::new();
let rs = run(
&view,
"MATCH (n:N) UNWIND n.xs AS x WHERE x > n.threshold RETURN x",
¶ms,
)
.expect("post-UNWIND WHERE referencing MATCH variable must succeed");
assert_eq!(rs.len(), 2, "exactly 2 rows: x=4 and x=5 from n1");
let v0 = rs.get(0, "x").cloned();
let v1 = rs.get(1, "x").cloned();
let got = [v0, v1];
assert!(got.contains(&Some(Value::Int(4))), "x=4 present");
assert!(got.contains(&Some(Value::Int(5))), "x=5 present");
}
#[test]
fn with_stage_aggregate_order_by_descending() {
let fx = city_graph();
let view = fx.view();
let params = BTreeMap::new();
let rs = run(
&view,
"MATCH (p:Person) WITH p.city AS city, COUNT(*) AS cnt ORDER BY cnt DESC RETURN city, cnt",
¶ms,
)
.expect("aggregate WITH ORDER BY must succeed");
assert_eq!(rs.len(), 2, "two city groups");
assert_eq!(
rs.get(0, "cnt"),
Some(&Value::Int(3)),
"first row must be the group with cnt=3 (Boston)"
);
assert_eq!(
rs.get(1, "cnt"),
Some(&Value::Int(2)),
"second row must be the group with cnt=2 (Austin)"
);
}
#[test]
fn unwind_then_with_composition() {
let mut fx = Fx::new();
fx.add(
"Doc",
"d1",
vec![(
"scores",
Value::List(vec![Value::Int(10), Value::Int(20), Value::Int(30)]),
)],
);
fx.add(
"Doc",
"d2",
vec![("scores", Value::List(vec![Value::Int(5), Value::Int(15)]))],
);
let view = fx.view();
let params = BTreeMap::new();
let rs = run(
&view,
"MATCH (n:Doc) UNWIND n.scores AS x WITH x RETURN x",
¶ms,
)
.expect("UNWIND then WITH pass-through must succeed");
assert_eq!(rs.len(), 5, "3 + 2 = 5 expanded rows carried through WITH");
let rs2 = run(
&view,
"MATCH (n:Doc) UNWIND n.scores AS x WITH COUNT(*) AS total RETURN total",
¶ms,
)
.expect("UNWIND then aggregate WITH must succeed");
assert_eq!(rs2.len(), 1, "single aggregate row");
assert_eq!(
rs2.get(0, "total"),
Some(&Value::Int(5)),
"total must be 5 (3+2 expanded rows)"
);
}
#[test]
fn binarith_div_min_over_neg1_does_not_panic() {
let fx = Fx::new();
let view = fx.view();
let vars = VarTable { names: vec![] };
let row: Row = vec![];
let params = BTreeMap::new();
let operand = Operand::BinArith {
op: ArithOp::Div,
left: Box::new(Operand::Lit(Value::Int(i64::MIN))),
right: Box::new(Operand::Lit(Value::Int(-1))),
};
let result = resolve_operand(&view, &vars, &row, &operand, &Params(¶ms));
assert!(
result.is_ok(),
"overflow division must not return Err: {result:?}"
);
assert_eq!(
result.unwrap(),
Some(Value::Int(i64::MAX)),
"i64::MIN / -1 must saturate to i64::MAX, not panic"
);
}
#[test]
fn binarith_missing_param_caught_at_preflight() {
let fx = Fx::new();
let view = fx.view();
let err = run(
&view,
"MATCH (n:X) RETURN abs($missing - 1)",
&BTreeMap::new(),
)
.expect_err("missing param inside BinArith must be caught at preflight");
assert!(
err.contains("missing"),
"error must name the param 'missing', got: {err}"
);
}
#[test]
fn is_null_filters_absent_prop() {
let mut fx = Fx::new();
fx.add("Person", "alice", vec![("age", Value::Int(30))]);
fx.add("Person", "bob", vec![]); let view = fx.view();
let rs = run(
&view,
"MATCH (n:Person) WHERE n.age IS NULL RETURN n",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.len(), 1, "only bob lacks age");
assert_eq!(rs.get(0, "n"), Some(&s("bob")));
}
#[test]
fn is_not_null_filters_present_prop() {
let mut fx = Fx::new();
fx.add("Person", "alice", vec![("age", Value::Int(30))]);
fx.add("Person", "bob", vec![]); let view = fx.view();
let rs = run(
&view,
"MATCH (n:Person) WHERE n.age IS NOT NULL RETURN n",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.len(), 1, "only alice has age");
assert_eq!(rs.get(0, "n"), Some(&s("alice")));
}
#[test]
fn optional_match_is_null_anti_join() {
let mut fx = Fx::new();
let alice = fx.add("Person", "alice", vec![]);
let bob = fx.add("Person", "bob", vec![]);
let carol = fx.add("Person", "carol", vec![]);
fx.edge("KNOWS", alice, carol, vec![]); fx.edge("KNOWS", carol, bob, vec![]); let view = fx.view();
let rs = run(
&view,
"MATCH (a:Person) OPTIONAL MATCH (a)-[:KNOWS]->(b) WITH a, b WHERE b IS NULL RETURN a",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.len(), 1, "only bob has no outgoing KNOWS edge");
assert_eq!(rs.get(0, "a"), Some(&s("bob")));
}
#[test]
fn is_null_combined_with_and_exec() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("x", Value::Int(1))]);
fx.add("N", "b", vec![("x", Value::Int(2)), ("y", Value::Int(9))]);
fx.add("N", "c", vec![]); let view = fx.view();
let rs = run(
&view,
"MATCH (n:N) WHERE n.y IS NULL AND n.x IS NOT NULL RETURN n",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.len(), 1);
assert_eq!(rs.get(0, "n"), Some(&s("a")));
}
#[test]
fn arithmetic_add_in_return() {
let mut fx = Fx::new();
fx.add("Person", "alice", vec![("age", Value::Int(29))]);
let view = fx.view();
let rs = run(
&view,
"MATCH (n:Person) RETURN n.age + 1 AS adjusted",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.len(), 1);
assert_eq!(rs.get(0, "adjusted"), Some(&Value::Int(30)));
}
#[test]
fn arithmetic_precedence_parens_pin() {
let mut fx = Fx::new();
fx.add("N", "x", vec![]);
let view = fx.view();
let rs = run(
&view,
"MATCH (n:N) RETURN (1 + 2) * 3 AS r",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.len(), 1);
assert_eq!(rs.get(0, "r"), Some(&Value::Int(9)));
}
#[test]
fn arithmetic_precedence_mul_over_add_pin() {
let mut fx = Fx::new();
fx.add("N", "x", vec![]);
let view = fx.view();
let rs = run(&view, "MATCH (n:N) RETURN 1 + 2 * 3 AS r", &BTreeMap::new()).unwrap();
assert_eq!(rs.len(), 1);
assert_eq!(rs.get(0, "r"), Some(&Value::Int(7)));
}
#[test]
fn arithmetic_div_by_zero_returns_error() {
let mut fx = Fx::new();
fx.add("N", "x", vec![]);
let view = fx.view();
let err = run(
&view,
"MATCH (n:N) WHERE 1 / 0 > 0 RETURN n",
&BTreeMap::new(),
)
.expect_err("division by zero must error");
assert!(
err.contains("division by zero"),
"error must mention division by zero, got: {err}"
);
}
#[test]
fn arithmetic_null_propagates() {
let mut fx = Fx::new();
fx.add("N", "x", vec![]); let view = fx.view();
let rs = run(
&view,
"MATCH (n:N) WHERE n.val + 1 > 0 RETURN n",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.len(), 0, "null arithmetic must not match");
}
#[test]
fn case_when_expression_in_return() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("id", s("a")), ("age", i(20))]);
fx.add("N", "b", vec![("id", s("b")), ("age", i(65))]);
let v = fx.view();
let rs = run(
&v,
"MATCH (n:N) RETURN n.id AS id, \
CASE WHEN n.age >= 65 THEN 'senior' ELSE 'other' END AS band",
&BTreeMap::new(),
)
.unwrap();
let band_of = |who: &str| {
(0..rs.len())
.find(|&i| rs.get(i, "id") == Some(&s(who)))
.and_then(|i| rs.get(i, "band").cloned())
};
assert_eq!(band_of("a"), Some(s("other")));
assert_eq!(band_of("b"), Some(s("senior")));
}
#[test]
fn case_when_no_else_yields_null() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("age", i(20))]);
let v = fx.view();
let rs = run(
&v,
"MATCH (n:N) RETURN CASE WHEN n.age >= 65 THEN 'senior' END AS band",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.get(0, "band"), None);
}
#[test]
fn multi_relationship_type_pattern_matches_either() {
let mut fx = Fx::new();
let a = fx.add("N", "a", vec![("id", s("a"))]);
let b = fx.add("N", "b", vec![("id", s("b"))]);
let c = fx.add("N", "c", vec![("id", s("c"))]);
let d = fx.add("N", "d", vec![("id", s("d"))]);
fx.edge("KNOWS", a, b, vec![]);
fx.edge("LIKES", a, c, vec![]);
fx.edge("HATES", a, d, vec![]);
let v = fx.view();
let rs = run(
&v,
"MATCH (a:N {id: 'a'})-[r:KNOWS|:LIKES]->(x) RETURN x",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.len(), 2, "KNOWS|LIKES reaches exactly b and c");
}
#[test]
fn collect_grouped_gathers_values_per_group() {
let mut fx = Fx::new();
fx.add("P", "a", vec![("city", s("austin")), ("name", s("Ann"))]);
fx.add("P", "b", vec![("city", s("austin")), ("name", s("Bob"))]);
fx.add("P", "c", vec![("city", s("boston")), ("name", s("Cy"))]);
let v = fx.view();
let rs = run(
&v,
"MATCH (n:P) RETURN n.city AS city, collect(n.name) AS names",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.len(), 2, "two city groups");
let austin = (0..rs.len())
.find(|&i| rs.get(i, "city") == Some(&s("austin")))
.expect("austin group present");
assert_eq!(
rs.get(austin, "names"),
Some(&Value::List(vec![s("Ann"), s("Bob")]))
);
}
#[test]
fn collect_ungrouped_gathers_all_into_one_list() {
let mut fx = Fx::new();
fx.add("P", "a", vec![("name", s("Ann"))]);
fx.add("P", "b", vec![("name", s("Bob"))]);
let v = fx.view();
let rs = run(
&v,
"MATCH (n:P) RETURN collect(n.name) AS names",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.len(), 1);
assert_eq!(
rs.get(0, "names"),
Some(&Value::List(vec![s("Ann"), s("Bob")]))
);
}
#[test]
fn string_predicate_functions_in_where() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("email", s("alice@acme.com"))]);
fx.add("N", "b", vec![("email", s("bob@other.org"))]);
let v = fx.view();
let p = BTreeMap::new();
assert_eq!(
run(
&v,
"MATCH (n:N) WHERE endsWith(n.email, '.com') RETURN n",
&p
)
.unwrap()
.len(),
1
);
assert_eq!(
run(
&v,
"MATCH (n:N) WHERE startsWith(n.email, 'bob') RETURN n",
&p
)
.unwrap()
.len(),
1
);
assert_eq!(
run(
&v,
"MATCH (n:N) WHERE contains(n.email, 'acme') RETURN n",
&p
)
.unwrap()
.len(),
1
);
}
#[test]
fn coercion_functions_in_return() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("s", s("42")), ("n", i(7)), ("g", f(3.9))]);
let v = fx.view();
let rs = run(
&v,
"MATCH (n:N) RETURN toInteger(n.s) AS ti, toFloat(n.n) AS tf, toString(n.g) AS ts",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.get(0, "ti"), Some(&Value::Int(42)));
assert_eq!(rs.get(0, "tf"), Some(&Value::Float(7.0)));
assert_eq!(rs.get(0, "ts"), Some(&Value::Str("3.9".into())));
}
#[test]
fn to_integer_unparseable_string_is_null() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("s", s("not-a-number"))]);
let v = fx.view();
let rs = run(
&v,
"MATCH (n:N) RETURN toInteger(n.s) AS ti",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.get(0, "ti"), None);
}
#[test]
fn index_scan_uses_index_and_matches_fallback() {
use core_storage::property_index::PropertyIndex;
use std::sync::atomic::Ordering;
let mut fx = Fx::new();
let a = fx.add("Person", "a", vec![("city", s("austin"))]);
let _b = fx.add("Person", "b", vec![("city", s("boston"))]);
let c = fx.add("Person", "c", vec![("city", s("austin"))]);
let mut pi = PropertyIndex::new();
pi.enable("Person", "city");
pi.set("Person", "city", a, &s("austin"));
pi.set("Person", "city", 1, &s("boston"));
pi.set("Person", "city", c, &s("austin"));
let q = "MATCH (n:Person {city: 'austin'}) RETURN n";
let before = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
let after = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
assert!(after > before, "IndexScan must take the indexed path");
assert_eq!(indexed.len(), 2);
let before2 = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
let after2 = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
assert_eq!(after2, before2, "fallback must not touch the index counter");
assert_eq!(
fallback.len(),
indexed.len(),
"fallback matches indexed result"
);
}
#[test]
fn arithmetic_in_where_comparison() {
let mut fx = Fx::new();
fx.add("Person", "alice", vec![("age", Value::Int(5))]); fx.add("Person", "bob", vec![("age", Value::Int(4))]); let view = fx.view();
let rs = run(
&view,
"MATCH (n:Person) WHERE n.age + 1 > 5 RETURN n",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.len(), 1);
assert_eq!(rs.get(0, "n"), Some(&s("alice")));
}
#[test]
fn where_fold_indexed_matches_fallback() {
use core_storage::property_index::PropertyIndex;
use std::sync::atomic::Ordering;
let mut fx = Fx::new();
let a = fx.add("Person", "alice", vec![("city", s("austin"))]);
let b = fx.add("Person", "bob", vec![("city", s("boston"))]);
let c = fx.add("Person", "carol", vec![("city", s("austin"))]);
let mut pi = PropertyIndex::new();
pi.enable("Person", "city");
pi.set("Person", "city", a, &s("austin"));
pi.set("Person", "city", b, &s("boston"));
pi.set("Person", "city", c, &s("austin"));
let q = "MATCH (n:Person) WHERE n.city = 'austin' RETURN n";
let before = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
let after = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
assert!(
after > before,
"WHERE equality fold must take the indexed path"
);
assert_eq!(indexed.len(), 2);
let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
assert_eq!(
rows_of(&fallback),
rows_of(&indexed),
"fallback must return identical rows, not just the same count"
);
}
#[test]
fn where_fold_miss_returns_empty() {
let mut fx = Fx::new();
fx.add("Person", "alice", vec![("city", s("austin"))]);
let q = "MATCH (n:Person) WHERE n.city = 'berlin' RETURN n";
let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
assert_eq!(rs.len(), 0);
}
#[test]
fn where_fold_param_uses_index() {
use core_storage::property_index::PropertyIndex;
use std::sync::atomic::Ordering;
let mut fx = Fx::new();
let a = fx.add("Person", "alice", vec![("city", s("austin"))]);
let b = fx.add("Person", "bob", vec![("city", s("boston"))]);
let mut pi = PropertyIndex::new();
pi.enable("Person", "city");
pi.set("Person", "city", a, &s("austin"));
pi.set("Person", "city", b, &s("boston"));
let q = "MATCH (n:Person) WHERE n.city = $c RETURN n";
let mut params = BTreeMap::new();
params.insert("c".to_string(), s("austin"));
let before = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
let rs = run(&fx.view_indexed(&pi), q, ¶ms).unwrap();
let after = super::INDEX_SCAN_FIRES.load(Ordering::Relaxed);
assert!(after > before, "$param WHERE equality must use index");
assert_eq!(rs.len(), 1);
}
#[test]
fn where_fold_residual_filter_applied() {
let mut fx = Fx::new();
fx.add(
"Person",
"young-austin",
vec![("city", s("austin")), ("age", Value::Int(20))],
);
fx.add(
"Person",
"old-austin",
vec![("city", s("austin")), ("age", Value::Int(40))],
);
fx.add(
"Person",
"boston",
vec![("city", s("boston")), ("age", Value::Int(20))],
);
let q = "MATCH (n:Person) WHERE n.city = 'austin' AND n.age > 30 RETURN n";
let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
assert_eq!(rs.len(), 1, "only old-austin should match city+age filter");
assert_eq!(rs.get(0, "n"), Some(&s("old-austin")));
}
#[test]
fn where_fold_unindexed_field_fallback() {
let mut fx = Fx::new();
fx.add("Person", "a", vec![("notindexed", s("x"))]);
fx.add("Person", "b", vec![("notindexed", s("y"))]);
let rs = run(
&fx.view(),
"MATCH (n:Person) WHERE n.notindexed = 'x' RETURN n",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.len(), 1);
assert_eq!(rs.get(0, "n"), Some(&s("a")));
}
#[test]
fn index_intersect_both_indexed_fires() {
use core_storage::property_index::PropertyIndex;
use std::sync::atomic::Ordering;
let mut fx = Fx::new();
let a = fx.add(
"Person",
"alice",
vec![("city", s("austin")), ("age", Value::Int(30))],
);
let b = fx.add(
"Person",
"bob",
vec![("city", s("austin")), ("age", Value::Int(25))],
);
let c = fx.add(
"Person",
"carol",
vec![("city", s("boston")), ("age", Value::Int(30))],
);
let mut pi = PropertyIndex::new();
pi.enable("Person", "city");
pi.enable("Person", "age");
pi.set("Person", "city", a, &s("austin"));
pi.set("Person", "city", b, &s("austin"));
pi.set("Person", "city", c, &s("boston"));
pi.set("Person", "age", a, &Value::Int(30));
pi.set("Person", "age", b, &Value::Int(25));
pi.set("Person", "age", c, &Value::Int(30));
let q = "MATCH (n:Person) WHERE n.city = 'austin' AND n.age = 30 RETURN n";
let before = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
let after = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
assert!(
after > before,
"IndexIntersect must advance counter on indexed path"
);
assert_eq!(indexed.len(), 1);
assert_eq!(indexed.get(0, "n"), Some(&s("alice")));
let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
assert_eq!(
rows_of(&fallback),
rows_of(&indexed),
"fallback must return identical rows"
);
}
#[test]
fn index_intersect_one_indexed_one_not() {
use core_storage::property_index::PropertyIndex;
use std::sync::atomic::Ordering;
let mut fx = Fx::new();
let a = fx.add(
"Person",
"alice",
vec![("city", s("austin")), ("role", s("eng"))],
);
let b = fx.add(
"Person",
"bob",
vec![("city", s("austin")), ("role", s("mgr"))],
);
let _c = fx.add(
"Person",
"carol",
vec![("city", s("boston")), ("role", s("eng"))],
);
let mut pi = PropertyIndex::new();
pi.enable("Person", "city");
pi.set("Person", "city", a, &s("austin"));
pi.set("Person", "city", b, &s("austin"));
let q = "MATCH (n:Person) WHERE n.city = 'austin' AND n.role = 'eng' RETURN n";
let before = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
let after = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
assert!(
after > before,
"IndexIntersect must fire when at least one field is indexed"
);
assert_eq!(indexed.len(), 1);
assert_eq!(indexed.get(0, "n"), Some(&s("alice")));
let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
assert_eq!(
rows_of(&fallback),
rows_of(&indexed),
"fallback must return identical rows"
);
}
#[test]
fn index_intersect_no_indexed_fallback() {
let mut fx = Fx::new();
fx.add("Person", "alice", vec![("x", s("1")), ("y", s("a"))]);
fx.add("Person", "bob", vec![("x", s("1")), ("y", s("b"))]);
fx.add("Person", "carol", vec![("x", s("2")), ("y", s("a"))]);
let q = "MATCH (n:Person) WHERE n.x = '1' AND n.y = 'a' RETURN n";
let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
assert_eq!(rs.len(), 1);
assert_eq!(rs.get(0, "n"), Some(&s("alice")));
}
#[test]
fn index_intersect_empty_intersection() {
use core_storage::property_index::PropertyIndex;
let mut fx = Fx::new();
let a = fx.add(
"Person",
"alice",
vec![("city", s("austin")), ("age", Value::Int(30))],
);
let b = fx.add(
"Person",
"bob",
vec![("city", s("boston")), ("age", Value::Int(25))],
);
let mut pi = PropertyIndex::new();
pi.enable("Person", "city");
pi.enable("Person", "age");
pi.set("Person", "city", a, &s("austin"));
pi.set("Person", "city", b, &s("boston"));
pi.set("Person", "age", a, &Value::Int(30));
pi.set("Person", "age", b, &Value::Int(25));
let q = "MATCH (n:Person) WHERE n.city = 'boston' AND n.age = 30 RETURN n";
let rs = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
assert_eq!(rs.len(), 0);
}
#[test]
fn index_intersect_with_params() {
use core_storage::property_index::PropertyIndex;
use std::sync::atomic::Ordering;
let mut fx = Fx::new();
let a = fx.add(
"Person",
"alice",
vec![("city", s("austin")), ("age", Value::Int(30))],
);
let b = fx.add(
"Person",
"bob",
vec![("city", s("boston")), ("age", Value::Int(30))],
);
let mut pi = PropertyIndex::new();
pi.enable("Person", "city");
pi.enable("Person", "age");
pi.set("Person", "city", a, &s("austin"));
pi.set("Person", "city", b, &s("boston"));
pi.set("Person", "age", a, &Value::Int(30));
pi.set("Person", "age", b, &Value::Int(30));
let q = "MATCH (n:Person) WHERE n.city = $c AND n.age = $a RETURN n";
let mut params = BTreeMap::new();
params.insert("c".to_string(), s("austin"));
params.insert("a".to_string(), Value::Int(30));
let before = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
let indexed = run(&fx.view_indexed(&pi), q, ¶ms).unwrap();
let after = super::INDEX_INTERSECT_FIRES.load(Ordering::Relaxed);
assert!(after > before, "$param intersect must fire indexed path");
assert_eq!(indexed.len(), 1);
assert_eq!(indexed.get(0, "n"), Some(&s("alice")));
let fallback = run(&fx.view(), q, ¶ms).unwrap();
assert_eq!(
rows_of(&fallback),
rows_of(&indexed),
"fallback must return identical rows"
);
}
#[test]
fn index_intersect_three_fields() {
use core_storage::property_index::PropertyIndex;
let mut fx = Fx::new();
let a = fx.add(
"Person",
"alice",
vec![
("city", s("austin")),
("age", Value::Int(30)),
("role", s("eng")),
],
);
let b = fx.add(
"Person",
"bob",
vec![
("city", s("austin")),
("age", Value::Int(30)),
("role", s("mgr")),
],
);
let mut pi = PropertyIndex::new();
pi.enable("Person", "city");
pi.enable("Person", "age");
pi.set("Person", "city", a, &s("austin"));
pi.set("Person", "city", b, &s("austin"));
pi.set("Person", "age", a, &Value::Int(30));
pi.set("Person", "age", b, &Value::Int(30));
let q =
"MATCH (n:Person) WHERE n.city = 'austin' AND n.age = 30 AND n.role = 'eng' RETURN n";
let indexed = run(&fx.view_indexed(&pi), q, &BTreeMap::new()).unwrap();
assert_eq!(indexed.len(), 1);
assert_eq!(indexed.get(0, "n"), Some(&s("alice")));
let fallback = run(&fx.view(), q, &BTreeMap::new()).unwrap();
assert_eq!(
rows_of(&fallback),
rows_of(&indexed),
"fallback must return identical rows"
);
}
fn assoc_graph() -> Fx {
let mut fx = Fx::new();
let t1 = fx.add(
"Talent",
"t1",
vec![
("status", s("published")),
("years_of_experience", i(12)),
(
"specialties",
Value::List(vec![s("hospitality"), s("retail")]),
),
("location", Value::List(vec![f(40.71), f(-74.01)])),
],
);
let t2 = fx.add(
"Talent",
"t2",
vec![
("status", s("published")),
("years_of_experience", i(11)),
("location", Value::List(vec![f(41.88), f(-87.63)])),
],
);
let t3 = fx.add(
"Talent",
"t3",
vec![("status", s("published")), ("years_of_experience", i(3))],
);
let c1 = fx.add("Company", "c1", vec![("name", s("Acme Design Works"))]);
let c2 = fx.add("Company", "c2", vec![("name", s("Beta Studio"))]);
let c3 = fx.add("Company", "c3", vec![("name", s("Gamma Works"))]);
for (t, c) in [(t1, c1), (t2, c1), (t3, c3)] {
fx.edge("INDUSTRY_ALIGNMENT", t, c, vec![]);
fx.edge("SPECIALTY_MATCH", t, c, vec![]);
fx.edge("LOCATION_FIT", t, c, vec![]);
}
fx.edge("INDUSTRY_ALIGNMENT", t1, c2, vec![]);
fx.edge("SPECIALTY_MATCH", t1, c2, vec![]);
fx
}
#[test]
fn node_key_reads_as_a_property() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (n:Company) RETURN n.key",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
rows_of(&rs),
vec![
vec![Some(s("c1"))],
vec![Some(s("c2"))],
vec![Some(s("c3"))]
]
);
}
#[test]
fn node_key_survives_a_with_aggregation() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company) \
WITH c, count(*) AS n WHERE n >= 1 RETURN c.key, key(c), n",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
rows_of(&rs),
vec![
vec![Some(s("c1")), Some(s("c1")), Some(i(2))],
vec![Some(s("c2")), Some(s("c2")), Some(i(1))],
vec![Some(s("c3")), Some(s("c3")), Some(i(1))],
]
);
}
#[test]
fn stored_key_property_wins_over_node_key() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("key", s("stored"))]);
let rs = run(&fx.view(), "MATCH (n:N) RETURN n.key", &BTreeMap::new()).unwrap();
assert_eq!(rows_of(&rs), vec![vec![Some(s("stored"))]]);
}
#[test]
fn n_id_falls_back_to_key_when_unstored() {
let mut fx = Fx::new();
fx.add("Person", "alice", vec![]);
let rs = run(
&fx.view(),
"MATCH (n:Person) WHERE n.id = 'alice' RETURN n.id, id(n), n.key",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
rows_of(&rs),
vec![vec![Some(s("alice")), Some(s("alice")), Some(s("alice"))]]
);
}
#[test]
fn stored_id_property_wins_over_key() {
let mut fx = Fx::new();
fx.add("N", "k", vec![("id", s("other"))]);
fx.add("N", "fallback", vec![]);
let v = fx.view();
let by_key = run(
&v,
"MATCH (n:N) WHERE n.id = 'fallback' RETURN key(n) AS k",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
col(&by_key, "k"),
vec![Some(s("fallback"))],
"a node with no stored id is found by its key, and the stored-id \
node is not swept in with it"
);
let miss = run(
&v,
"MATCH (n:N) WHERE n.id = 'k' RETURN n.id",
&BTreeMap::new(),
)
.unwrap();
assert!(
miss.is_empty(),
"stored id must win: WHERE n.id = key is empty"
);
let hit = run(
&v,
"MATCH (n:N) WHERE n.id = 'other' RETURN n.id",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rows_of(&hit), vec![vec![Some(s("other"))]]);
let projected = run(&v, "MATCH (n:N) RETURN n.id AS i", &BTreeMap::new()).unwrap();
let mut got: Vec<String> = col(&projected, "i")
.into_iter()
.map(|v| match v {
Some(Value::Str(s)) => s,
other => panic!("expected a string, got {other:?}"),
})
.collect();
got.sort();
assert_eq!(
got,
vec!["fallback".to_string(), "other".to_string()],
"projection is stored-wins per node: the stored id for one, the \
key fallback for the other"
);
}
#[test]
fn id_function_rejects_non_node() {
let (fx, _, _) = single_edge();
let v = fx.view();
let key_err = run(&v, "MATCH (a)-[r:T]->(b) RETURN key(r)", &BTreeMap::new())
.expect_err("key() on a relationship must error");
let id_err = run(&v, "MATCH (a)-[r:T]->(b) RETURN id(r)", &BTreeMap::new())
.expect_err("id() on a relationship must error");
assert!(
key_err.contains("not a node"),
"key() error class: {key_err}"
);
assert!(id_err.contains("not a node"), "id() error class: {id_err}");
assert!(
id_err.contains("id()"),
"id() error must name itself: {id_err}"
);
}
#[test]
fn where_n_id_eq_literal_uses_scan_key() {
let mut fx = Fx::new();
fx.add("Person", "alice", vec![]);
let fires_before = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
let rs = run(
&fx.view(),
"MATCH (n:Person) WHERE n.id = 'alice' RETURN n",
&BTreeMap::new(),
)
.unwrap();
let fires_after = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
assert!(
fires_after > fires_before,
"SCAN_KEY_FIRES must increment; before={fires_before} after={fires_after}"
);
assert_eq!(rows_of(&rs), vec![vec![Some(s("alice"))]]);
}
#[test]
fn where_n_key_eq_param_uses_scan_key() {
let mut fx = Fx::new();
fx.add("Person", "alice", vec![]);
let mut params = BTreeMap::new();
params.insert("k".to_string(), s("alice"));
let fires_before = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
let rs = run(
&fx.view(),
"MATCH (n:Person) WHERE n.key = $k RETURN n",
¶ms,
)
.unwrap();
let fires_after = super::SCAN_KEY_FIRES.load(std::sync::atomic::Ordering::Relaxed);
assert!(
fires_after > fires_before,
"SCAN_KEY_FIRES must increment; before={fires_before} after={fires_after}"
);
assert_eq!(rows_of(&rs), vec![vec![Some(s("alice"))]]);
}
#[test]
fn where_key_func_ignores_a_stored_key_property() {
let mut fx = Fx::new();
fx.add("N", "a", vec![("key", s("K"))]);
fx.add("N", "K", vec![]);
let rs = run(
&fx.view(),
"MATCH (n:N) WHERE key(n) = 'K' RETURN key(n) AS k",
&BTreeMap::new(),
)
.unwrap();
let got: Vec<Option<Value>> = col(&rs, "k");
assert_eq!(
got,
vec![Some(s("K"))],
"key(n) is the id-map key: only the node keyed K matches, not the \
node whose stored `key` property is K"
);
}
#[test]
fn where_id_func_ignores_a_stored_id_property() {
let mut fx = Fx::new();
fx.add("N", "k", vec![("id", s("other"))]);
let rs = run(
&fx.view(),
"MATCH (n:N) WHERE id(n) = 'k' RETURN id(n) AS k",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
col(&rs, "k"),
vec![Some(s("k"))],
"id(n) is the id-map key: a stored `id` property must not suppress \
the match"
);
}
#[test]
fn where_n_id_respects_stored_wins() {
let mut fx = Fx::new();
fx.add("N", "k", vec![("id", s("other"))]);
fx.add("N", "other", vec![]);
let rs = run(
&fx.view(),
"MATCH (n:N) WHERE n.id = 'other' RETURN key(n) AS k",
&BTreeMap::new(),
)
.unwrap();
let mut got: Vec<String> = col(&rs, "k")
.into_iter()
.map(|v| match v {
Some(Value::Str(s)) => s,
other => panic!("expected a string key, got {other:?}"),
})
.collect();
got.sort();
assert_eq!(
got,
vec!["k".to_string(), "other".to_string()],
"stored-wins union: the stored-id node and the key-fallback node"
);
}
#[test]
fn labels_and_label_read_the_node_label() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (n:Company) RETURN labels(n), n.label LIMIT 1",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
rows_of(&rs),
vec![vec![
Some(Value::List(vec![s("Company")])),
Some(s("Company"))
]]
);
}
#[test]
fn unknown_function_is_a_named_error() {
let fx = assoc_graph();
let err = run(
&fx.view(),
"MATCH (n:Company) RETURN nodes(n)",
&BTreeMap::new(),
)
.expect_err("unknown function must error");
assert!(err.contains("unknown function `nodes`"), "{err}");
assert!(
err.contains("labels"),
"error must list what is supported: {err}"
);
}
#[test]
fn infix_string_predicates_filter() {
let fx = assoc_graph();
let p = BTreeMap::new();
for (q, want) in [
(
"MATCH (c:Company) WHERE c.name STARTS WITH 'Acme' RETURN c.key",
vec!["c1"],
),
(
"MATCH (c:Company) WHERE c.name ENDS WITH 'Works' RETURN c.key",
vec!["c1", "c3"],
),
(
"MATCH (c:Company) WHERE c.name CONTAINS 'Studio' RETURN c.key",
vec!["c2"],
),
(
"MATCH (c:Company) WHERE NOT c.name CONTAINS 'Works' RETURN c.key",
vec!["c2"],
),
] {
let rs = run(&fx.view(), q, &p).unwrap_or_else(|e| panic!("{q}: {e}"));
let got: Vec<String> = (0..rs.len())
.map(|r| match rs.row(r)[0].clone() {
Some(Value::Str(k)) => k,
other => panic!("{q}: {other:?}"),
})
.collect();
assert_eq!(got, want, "{q}");
}
}
#[test]
fn infix_string_predicate_on_missing_property_is_false() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent) WHERE t.name STARTS WITH 'x' RETURN t.key",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rows_of(&rs), Vec::<Vec<Option<Value>>>::new());
}
#[test]
fn starts_without_with_is_a_named_error() {
let err = parse(&lex("MATCH (c:Company) WHERE c.name STARTS 'Acme' RETURN c").unwrap())
.expect_err("must not parse");
assert!(err.contains("expected WITH after STARTS"), "{err}");
}
#[test]
fn list_subscript_reads_one_element() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent) WHERE t.key = 't1' \
RETURN t.location[0] AS lat, t.location[1] AS lon, \
t.location[-1] AS last, t.location[7] AS oob, t.status[0] AS notalist",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
rows_of(&rs),
vec![vec![
Some(f(40.71)),
Some(f(-74.01)),
Some(f(-74.01)),
None,
None
]]
);
}
#[test]
fn node_key_works_in_every_position() {
let fx = assoc_graph();
let p = BTreeMap::new();
for (q, want) in [
(
"MATCH (c:Company) WHERE c.key = 'c2' RETURN c.key",
vec!["c2"],
),
(
"MATCH (c:Company) WHERE c.key <> 'c1' RETURN c.key LIMIT 1",
vec!["c2"],
),
(
"MATCH (c:Company) WITH c ORDER BY c.key DESC RETURN c.key LIMIT 1",
vec!["c3"],
),
("MATCH (c:Company {key: 'c3'}) RETURN c.key", vec!["c3"]),
(
"MATCH (c:Company) WHERE c.label = 'Company' RETURN c.key LIMIT 1",
vec!["c1"],
),
] {
let rs = run(&fx.view(), q, &p).unwrap_or_else(|e| panic!("{q}: {e}"));
let got: Vec<String> = (0..rs.len())
.map(|r| match rs.row(r)[0].clone() {
Some(Value::Str(k)) => k,
other => panic!("{q}: {other:?}"),
})
.collect();
assert_eq!(got, want, "{q}");
}
}
#[test]
fn list_subscript_filters() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent) WHERE t.location[0] > 41.0 RETURN t.key",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rows_of(&rs), vec![vec![Some(s("t2"))]]);
}
#[test]
fn count_distinct_counts_each_binding_once() {
let fx = assoc_graph();
let q = "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT|:SPECIALTY_MATCH]->(c:Company) \
WITH c, count(t) AS raw, count(DISTINCT t) AS uniq WHERE raw >= 1 \
RETURN c.key, raw, uniq";
let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
assert_eq!(
rows_of(&rs),
vec![
vec![Some(s("c1")), Some(i(4)), Some(i(2))],
vec![Some(s("c2")), Some(i(2)), Some(i(1))],
vec![Some(s("c3")), Some(i(2)), Some(i(1))],
]
);
}
#[test]
fn count_distinct_on_a_relationship_counts_edges() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (a)-[r]->(b) RETURN count(r) AS raw, count(DISTINCT r) AS uniq",
&BTreeMap::new(),
)
.unwrap();
let row = &rows_of(&rs)[0];
assert_eq!(
row[0], row[1],
"no pair is joined twice by one type: {row:?}"
);
assert_ne!(row[1], Some(i(0)), "a graph full of edges counts them");
}
#[test]
fn count_distinct_on_a_relationship_survives_an_alternation() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent)-[r:INDUSTRY_ALIGNMENT|:SPECIALTY_MATCH]->(c:Company) \
RETURN count(r) AS raw, count(DISTINCT r) AS uniq",
&BTreeMap::new(),
)
.unwrap();
let row = &rows_of(&rs)[0];
assert_eq!(row[0], row[1], "every row bound a different edge: {row:?}");
}
#[test]
fn collect_distinct_dedupes() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT|:SPECIALTY_MATCH]->(c:Company) \
WHERE c.key = 'c1' WITH collect(DISTINCT t.status) AS st RETURN st",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
rows_of(&rs),
vec![vec![Some(Value::List(vec![s("published")]))]]
);
}
#[test]
fn count_distinct_star_is_rejected() {
let err =
parse(&lex("MATCH (n) RETURN count(DISTINCT *)").unwrap()).expect_err("must not parse");
assert!(
err.contains("DISTINCT * is not a valid aggregate argument"),
"{err}"
);
}
#[test]
fn distinct_is_still_usable_as_a_variable_name() {
let q = parse(&lex("MATCH (distinct) RETURN count(distinct)").unwrap()).unwrap();
assert_eq!(
q.returns[0].value,
RetVal::Agg {
func: crate::cypher::ast::AggFunc::Count,
arg: crate::cypher::ast::AggArg::Var("distinct".into()),
}
);
}
#[test]
fn comma_patterns_intersect_on_shared_variables() {
let fx = assoc_graph();
let q = "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company), \
(t)-[:SPECIALTY_MATCH]->(c), \
(t)-[:LOCATION_FIT]->(c) \
WHERE t.status = 'published' AND t.years_of_experience >= 10 \
WITH c, count(DISTINCT t) AS n WHERE n >= 2 \
RETURN c.key, n ORDER BY n DESC";
let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
assert_eq!(rows_of(&rs), vec![vec![Some(s("c1")), Some(i(2))]]);
}
#[test]
fn comma_patterns_exclude_partial_matches() {
let fx = assoc_graph();
let q = "MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company), \
(t)-[:SPECIALTY_MATCH]->(c), \
(t)-[:LOCATION_FIT]->(c) \
WHERE t.years_of_experience >= 10 \
WITH c, count(DISTINCT t) AS n RETURN c.key, n";
let rs = run(&fx.view(), q, &BTreeMap::new()).unwrap();
assert_eq!(rows_of(&rs), vec![vec![Some(s("c1")), Some(i(2))]]);
}
#[test]
fn comma_patterns_equal_separate_match_clauses() {
let commas =
parse(&lex("MATCH (a:A)-[:X]->(b:B), (a)-[:Y]->(b) RETURN a.key").unwrap()).unwrap();
let clauses =
parse(&lex("MATCH (a:A)-[:X]->(b:B) MATCH (a)-[:Y]->(b) RETURN a.key").unwrap())
.unwrap();
assert_eq!(commas.matches, clauses.matches);
}
#[test]
fn aggregate_with_projects_without_a_having_clause() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company) \
WITH c, count(t) AS n RETURN c.name, n * 2 AS dbl",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.columns(), ["c.name", "dbl"]);
assert_eq!(
rows_of(&rs),
vec![
vec![Some(s("Acme Design Works")), Some(i(4))],
vec![Some(s("Beta Studio")), Some(i(2))],
vec![Some(s("Gamma Works")), Some(i(2))],
]
);
}
#[test]
fn aggregate_with_names_its_projected_columns() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company) \
WITH c, count(t) AS n RETURN key(c), n",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.columns(), ["key(c)", "n"]);
assert_eq!(
rows_of(&rs),
vec![
vec![Some(s("c1")), Some(i(2))],
vec![Some(s("c2")), Some(i(1))],
vec![Some(s("c3")), Some(i(1))],
]
);
}
#[test]
fn aggregate_with_returning_the_bare_node_keeps_the_key() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent)-[:INDUSTRY_ALIGNMENT]->(c:Company) \
WITH c, count(t) AS n RETURN c",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.columns(), ["c"]);
assert_eq!(
rows_of(&rs),
vec![
vec![Some(s("c1"))],
vec![Some(s("c2"))],
vec![Some(s("c3"))]
]
);
}
#[test]
fn a_plain_aggregate_is_unchanged() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (c:Company) RETURN c.name, count(*)",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.columns(), ["c.name", "COUNT(*)"]);
assert_eq!(rs.len(), 3);
}
#[test]
fn two_subscripts_of_one_list_are_two_named_columns() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent) RETURN t.location[0], t.location[1]",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.columns(), ["t.location[0]", "t.location[1]"]);
assert_eq!(
rows_of(&rs),
vec![
vec![Some(f(40.71)), Some(f(-74.01))],
vec![Some(f(41.88)), Some(f(-87.63))],
vec![None, None],
]
);
}
#[test]
fn a_subscript_carries_through_a_with_stage() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent) WHERE t.location[0] > 41.0 \
WITH t, t.location[0] AS lat, t.location[1] \
RETURN key(t), lat, t.location[1]",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.columns(), ["key(t)", "lat", "t.location[1]"]);
assert_eq!(
rows_of(&rs),
vec![vec![Some(s("t2")), Some(f(41.88)), Some(f(-87.63))]]
);
}
#[test]
fn subscript_column_names_cover_nested_and_computed_forms() {
use crate::cypher::ast::operand_label;
let prop = || Operand::Prop {
var: "t".into(),
field: "location".into(),
};
let at = |idx: Operand| Operand::Index {
base: Box::new(prop()),
index: Box::new(idx),
};
assert_eq!(operand_label(&at(Operand::Lit(i(0)))), "t.location[0]");
assert_eq!(operand_label(&at(Operand::Lit(i(-1)))), "t.location[-1]");
assert_eq!(
operand_label(&at(Operand::Param("k".into()))),
"t.location[$k]"
);
assert_eq!(
operand_label(&at(Operand::Var("j".into()))),
"t.location[j]"
);
assert_eq!(
operand_label(&Operand::Index {
base: Box::new(at(Operand::Lit(i(0)))),
index: Box::new(Operand::Lit(i(1))),
}),
"t.location[0][1]"
);
}
#[test]
fn a_with_alias_is_visible_to_its_where() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent) WITH t, t.years_of_experience AS x WHERE x > 11 \
RETURN key(t), x",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.columns(), ["key(t)", "x"]);
assert_eq!(rows_of(&rs), vec![vec![Some(s("t1")), Some(i(12))]]);
}
#[test]
fn a_with_alias_is_visible_to_where_and_order_by_together() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent) WITH t, t.years_of_experience AS x WHERE x > 3 \
ORDER BY x DESC RETURN key(t), x",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.columns(), ["key(t)", "x"]);
assert_eq!(
rows_of(&rs),
vec![
vec![Some(s("t1")), Some(i(12))],
vec![Some(s("t2")), Some(i(11))],
]
);
}
#[test]
fn a_with_carrying_only_a_variable_still_filters_on_it() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (t:Talent) WHERE t.status = 'published' \
WITH t WHERE t.years_of_experience > 11 RETURN key(t)",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.columns(), ["key(t)"]);
assert_eq!(rows_of(&rs), vec![vec![Some(s("t1"))]]);
}
#[test]
fn a_with_alias_projects_under_its_new_name() {
let fx = assoc_graph();
let rs = run(
&fx.view(),
"MATCH (c:Company) WITH c, c.name AS nm RETURN nm",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(rs.columns(), ["nm"]);
assert_eq!(
rows_of(&rs),
vec![
vec![Some(s("Acme Design Works"))],
vec![Some(s("Beta Studio"))],
vec![Some(s("Gamma Works"))],
]
);
}
#[test]
fn every_with_alias_shape_is_in_scope_for_the_where() {
let fx = assoc_graph();
let p = BTreeMap::new();
for (q, want) in [
(
"MATCH (t:Talent) WITH t, t.years_of_experience + 1 AS x WHERE x > 12 \
RETURN key(t)",
vec!["t1"],
),
(
"MATCH (t:Talent) WITH t, t.location[0] AS lat WHERE lat > 41.0 RETURN key(t)",
vec!["t2"],
),
(
"MATCH (t:Talent) WITH t AS u WHERE u.years_of_experience > 11 RETURN key(u)",
vec!["t1"],
),
(
"MATCH (t:Talent) WITH t.key AS k WHERE k = 't2' RETURN k",
vec!["t2"],
),
] {
let rs = run(&fx.view(), q, &p).unwrap_or_else(|e| panic!("{q}: {e}"));
let got: Vec<String> = (0..rs.len())
.map(|r| match rs.row(r)[0].clone() {
Some(Value::Str(k)) => k,
other => panic!("{q}: {other:?}"),
})
.collect();
assert_eq!(got, want, "{q}");
}
}
#[test]
fn an_unknown_name_in_a_with_where_is_still_an_error() {
let err = plan(
&parse(
&lex(
"MATCH (t:Talent) WITH t, t.years_of_experience AS x WHERE nope > 1 \
RETURN key(t)",
)
.unwrap(),
)
.unwrap(),
)
.expect_err("must not plan");
assert_eq!(err, "unbound variable `nope` in WHERE");
}
#[test]
fn comma_patterns_without_shared_vars_are_a_product() {
let mut fx = Fx::new();
fx.add("A", "a1", vec![]);
fx.add("A", "a2", vec![]);
fx.add("B", "b1", vec![]);
let rs = run(
&fx.view(),
"MATCH (a:A), (b:B) RETURN a.key, b.key",
&BTreeMap::new(),
)
.unwrap();
assert_eq!(
rows_of(&rs),
vec![
vec![Some(s("a1")), Some(s("b1"))],
vec![Some(s("a2")), Some(s("b1"))],
]
);
}
}