use std::cell::{Cell, RefCell};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::rc::Rc;
use std::sync::{
atomic::{AtomicBool, Ordering as AtomicOrdering},
Arc,
};
use std::time::{Duration, Instant};
use marsdb_graph::{
AdjEntry, Direction, Edge, EdgeId, GraphStore, Node, NodeId, PropertyValue, Txn,
TzId as GraphTzId, WriteTransaction,
};
use crate::aggregate::{property_value_hash_key, value_hash_key, AggAcc, HashKey};
use crate::ast::{
is_aggregate_name, is_percentile_name, ArithOp, CallClause, CallYield, CompareOp, Expr,
Literal, MergeClause, NodePattern, Pattern, PropAccess, QuantifierKind, QueryClause, QueryPart,
RelDirection, RemoveItem, ReturnExpr, ReturnItem, ReturnTail, SetItem, SortDir, Statement,
Tail, UnwindClause, WithClause, WithExpr,
};
use crate::error::QueryError;
use crate::ir::{ExpandDirection, IndexSeekValue, LogicalPlan};
use crate::parse_helpers::validate_named_path_pattern;
use crate::planner::{
apply_index_seeks, build_match_plan, pattern_all_vars, pattern_new_vars, plan_reversed_pattern,
};
use crate::procedure::{ProcedureProvider, ProcedureSignature};
use crate::result::QueryResult;
use crate::temporal;
use crate::value::{PathElem, Value};
mod arith;
mod scalar_fns;
mod temporal_fns;
mod value_cmp;
use arith::*;
use scalar_fns::*;
pub(crate) use temporal_fns::tz_from_graph;
use temporal_fns::*;
pub(crate) use value_cmp::comparable_ordering;
use value_cmp::*;
const OPTIONAL_SEED_IDX_KEY: &str = "__seed_idx";
const MERGE_CREATED_KEY: &str = "__merge_created";
#[derive(Debug, Clone, Default)]
pub struct CancellationToken(Arc<AtomicBool>);
impl CancellationToken {
pub fn new() -> Self {
Self::default()
}
pub fn cancel(&self) {
self.0.store(true, AtomicOrdering::Release);
}
pub fn is_cancelled(&self) -> bool {
self.0.load(AtomicOrdering::Acquire)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ExecutionOutcome {
Success,
SyntaxError,
SemanticError,
TypeError,
GraphError,
UnboundVariable,
MissingParameter,
Cancelled,
Timeout,
ResourceLimit,
}
impl ExecutionOutcome {
pub fn from_error(error: &QueryError) -> Self {
match error {
QueryError::Syntax(_) => Self::SyntaxError,
QueryError::Semantic(_) => Self::SemanticError,
QueryError::Type(_) => Self::TypeError,
QueryError::Graph(_) => Self::GraphError,
QueryError::UnboundVariable(_) => Self::UnboundVariable,
QueryError::MissingParam(_) => Self::MissingParameter,
QueryError::Cancelled => Self::Cancelled,
QueryError::Timeout => Self::Timeout,
QueryError::ResourceLimit(_) => Self::ResourceLimit,
}
}
}
#[derive(Debug, Clone)]
pub struct ExecutionEvent {
pub elapsed: Duration,
pub statement_read_only: Option<bool>,
pub result_rows: Option<usize>,
pub relationship_expansions: u64,
pub outcome: ExecutionOutcome,
}
#[derive(Clone)]
pub struct ExecutionObserver(Arc<dyn Fn(&ExecutionEvent) + Send + Sync>);
impl ExecutionObserver {
pub fn new(callback: impl Fn(&ExecutionEvent) + Send + Sync + 'static) -> Self {
Self(Arc::new(callback))
}
pub fn observe(&self, event: &ExecutionEvent) {
let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| (self.0)(event)));
}
}
impl std::fmt::Debug for ExecutionObserver {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("ExecutionObserver(..)")
}
}
#[derive(Debug, Clone, Default)]
pub struct ExecutionOptions {
pub max_intermediate_rows: Option<usize>,
pub max_result_rows: Option<usize>,
pub max_relationship_expansions: Option<u64>,
pub timeout: Option<Duration>,
pub cancellation_token: Option<CancellationToken>,
pub observer: Option<ExecutionObserver>,
pub procedures: Option<crate::procedure::Procedures>,
pub params: HashMap<String, PropertyValue>,
}
struct ExecutionGuard<'a> {
options: &'a ExecutionOptions,
deadline: Option<Instant>,
relationship_expansions: Cell<u64>,
deleted_edge_types: RefCell<HashMap<EdgeId, String>>,
}
impl<'a> ExecutionGuard<'a> {
fn new(options: &'a ExecutionOptions) -> Self {
Self {
options,
deadline: options
.timeout
.and_then(|timeout| Instant::now().checked_add(timeout)),
relationship_expansions: Cell::new(0),
deleted_edge_types: RefCell::new(HashMap::new()),
}
}
fn record_deleted_edge_type(&self, id: EdgeId, label: String) {
self.deleted_edge_types.borrow_mut().insert(id, label);
}
fn deleted_edge_type(&self, id: EdgeId) -> Option<String> {
self.deleted_edge_types.borrow().get(&id).cloned()
}
fn procedure_provider(&self) -> Option<&dyn ProcedureProvider> {
self.options.procedures.as_ref().map(|p| p.0.as_ref())
}
fn checkpoint(&self) -> Result<(), QueryError> {
if self
.options
.cancellation_token
.as_ref()
.is_some_and(CancellationToken::is_cancelled)
{
return Err(QueryError::Cancelled);
}
if self
.deadline
.is_some_and(|deadline| Instant::now() >= deadline)
{
return Err(QueryError::Timeout);
}
Ok(())
}
fn check_intermediate_rows(&self, rows: usize) -> Result<(), QueryError> {
self.checkpoint()?;
if self
.options
.max_intermediate_rows
.is_some_and(|limit| rows > limit)
{
return Err(QueryError::ResourceLimit(format!(
"intermediate row count {rows} exceeds configured maximum {}",
self.options.max_intermediate_rows.unwrap()
)));
}
Ok(())
}
fn check_result_rows(&self, rows: usize) -> Result<(), QueryError> {
self.checkpoint()?;
if self
.options
.max_result_rows
.is_some_and(|limit| rows > limit)
{
return Err(QueryError::ResourceLimit(format!(
"result row count {rows} exceeds configured maximum {}",
self.options.max_result_rows.unwrap()
)));
}
Ok(())
}
fn relationship_expansion(&self) -> Result<(), QueryError> {
self.checkpoint()?;
let count = self
.relationship_expansions
.get()
.checked_add(1)
.ok_or_else(|| {
QueryError::ResourceLimit("relationship expansion counter overflow".into())
})?;
self.relationship_expansions.set(count);
if self
.options
.max_relationship_expansions
.is_some_and(|limit| count > limit)
{
return Err(QueryError::ResourceLimit(format!(
"relationship expansion count {count} exceeds configured maximum {}",
self.options.max_relationship_expansions.unwrap()
)));
}
Ok(())
}
}
#[derive(Debug, Clone)]
enum Binding {
Node(NodeId),
Edge(EdgeId),
Value(PropertyValue),
List(Vec<Value>),
Map(BTreeMap<String, Value>),
Path(Vec<PathBinding>),
}
#[derive(Debug, Clone)]
enum PathBinding {
Node(NodeId),
Edge(EdgeId),
}
struct ShortestPathSpec<'a> {
direction: ExpandDirection,
rel_labels: &'a [String],
min_hops: u32,
max_hops: Option<u32>,
}
struct VarExpandSpec<'a> {
from_var: &'a str,
to_var: &'a str,
rel_labels: &'a [String],
direction: ExpandDirection,
min_hops: u32,
max_hops: Option<u32>,
exclude_edge_vars: &'a [String],
exclude_edge_sets: &'a [String],
exclude_edge_var: &'a str,
path_segment_var: Option<&'a str>,
rel_list_var: Option<&'a str>,
rel_props: &'a [(String, ReturnExpr)],
}
struct MatchRelListSpec<'a> {
from_var: &'a str,
to_var: &'a str,
rel_list_var: &'a str,
rel_labels: &'a [String],
direction: ExpandDirection,
min_hops: u32,
max_hops: Option<u32>,
}
struct PatternComprehensionSpec<'a> {
path_var: &'a Option<String>,
pattern: &'a Pattern,
where_clause: &'a Option<Box<Expr>>,
projection: &'a ReturnExpr,
}
struct IndexSeekSpec<'a> {
var: &'a str,
label: &'a str,
prop: &'a str,
value: &'a IndexSeekValue,
}
struct GroupFinishCtx<'a> {
items: &'a [ReturnItem],
key_bindings: &'a [Option<Binding>],
}
struct ResultModifiers<'a> {
order_by: &'a Option<Vec<(ReturnExpr, SortDir)>>,
skip: Option<i64>,
limit: Option<i64>,
}
type BindingRow = HashMap<String, Binding>;
type FastCountResult = (Vec<BindingRow>, HashSet<String>);
type RowStream<'a> = Box<dyn Iterator<Item = Result<BindingRow, QueryError>> + 'a>;
const VAR_EXPAND_DEPTH_CAP: u32 = 30;
pub struct Executor<'a> {
store: &'a GraphStore,
now: Cell<Option<temporal::NowSnapshot>>,
node_cache: RefCell<HashMap<NodeId, Rc<Node>>>,
node_cache_enabled: Cell<bool>,
prop_id_memo: RefCell<HashMap<String, Option<u32>>>,
}
impl<'a> Executor<'a> {
pub fn new(store: &'a GraphStore) -> Self {
Self {
store,
now: Cell::new(None),
node_cache: RefCell::new(HashMap::new()),
node_cache_enabled: Cell::new(false),
prop_id_memo: RefCell::new(HashMap::new()),
}
}
fn get_node_cached(&self, txn: Txn, id: NodeId) -> Result<Option<Rc<Node>>, QueryError> {
if self.node_cache_enabled.get() {
if let Some(cached) = self.node_cache.borrow().get(&id) {
return Ok(Some(Rc::clone(cached)));
}
}
let node = GraphStore::get_node_in_txn(txn, id)?.map(Rc::new);
if self.node_cache_enabled.get() {
if let Some(n) = &node {
self.node_cache.borrow_mut().insert(id, Rc::clone(n));
}
}
Ok(node)
}
fn now_snapshot(&self) -> temporal::NowSnapshot {
if let Some(n) = self.now.get() {
return n;
}
let n = temporal::capture_now();
self.now.set(Some(n));
n
}
pub fn execute(&self, stmt: &Statement) -> Result<QueryResult, QueryError> {
self.execute_with_options(stmt, &ExecutionOptions::default())
}
pub fn execute_with_options(
&self,
stmt: &Statement,
options: &ExecutionOptions,
) -> Result<QueryResult, QueryError> {
let started = Instant::now();
let guard = ExecutionGuard::new(options);
let result = self.execute_with_guard(stmt, &guard);
Self::notify_observer(options, stmt, started, &guard, &result);
result
}
fn execute_with_guard(
&self,
stmt: &Statement,
guard: &ExecutionGuard<'_>,
) -> Result<QueryResult, QueryError> {
crate::semantic::validate_statement(stmt)?;
guard.checkpoint()?;
self.node_cache.borrow_mut().clear();
self.prop_id_memo.borrow_mut().clear();
self.node_cache_enabled.set(is_read_only(stmt));
if let Statement::Explain(inner) = stmt {
return self.execute_explain(inner);
}
if is_read_only(stmt) {
let read_txn = self.store.begin_read()?;
return match stmt {
Statement::Union { parts, all } => {
self.materialize_union(Txn::Read(&read_txn), parts, *all, guard)
}
Statement::Match {
clauses,
tail,
order_by,
skip,
limit,
} => {
let skip = self.resolve_skip_limit(
Txn::Read(&read_txn),
skip.as_deref(),
"SKIP",
guard,
)?;
let limit = self.resolve_skip_limit(
Txn::Read(&read_txn),
limit.as_deref(),
"LIMIT",
guard,
)?;
self.execute_match(
Txn::Read(&read_txn),
clauses,
tail,
ResultModifiers {
order_by,
skip,
limit,
},
guard,
)
}
_ => unreachable!("is_read_only only returns true for Statement::Match/Union"),
};
}
let write_txn = self.store.begin_write()?;
let outcome = self.execute_in_write_transaction_validated(stmt, &write_txn, guard);
match outcome {
Ok(result) => {
GraphStore::commit(write_txn)?;
Ok(result)
}
Err(e) => {
let _ = GraphStore::abort(write_txn);
Err(e)
}
}
}
pub fn execute_in_write_transaction(
&self,
stmt: &Statement,
write_txn: &WriteTransaction,
) -> Result<QueryResult, QueryError> {
self.execute_in_write_transaction_with_options(
stmt,
write_txn,
&ExecutionOptions::default(),
)
}
pub fn execute_in_write_transaction_with_options(
&self,
stmt: &Statement,
write_txn: &WriteTransaction,
options: &ExecutionOptions,
) -> Result<QueryResult, QueryError> {
let started = Instant::now();
let guard = ExecutionGuard::new(options);
let result = self.execute_in_write_transaction_with_guard(stmt, write_txn, &guard);
Self::notify_observer(options, stmt, started, &guard, &result);
result
}
fn execute_in_write_transaction_with_guard(
&self,
stmt: &Statement,
write_txn: &WriteTransaction,
guard: &ExecutionGuard<'_>,
) -> Result<QueryResult, QueryError> {
crate::semantic::validate_statement(stmt)?;
guard.checkpoint()?;
self.node_cache.borrow_mut().clear();
self.prop_id_memo.borrow_mut().clear();
self.node_cache_enabled.set(is_read_only(stmt));
if let Statement::Explain(inner) = stmt {
return self.execute_explain(inner);
}
self.execute_in_write_transaction_validated(stmt, write_txn, guard)
}
fn execute_explain(&self, inner: &Statement) -> Result<QueryResult, QueryError> {
let read_txn = self.store.begin_read()?;
let lines = crate::explain::explain_statement(inner, Txn::Read(&read_txn))?;
Ok(QueryResult {
columns: vec!["plan".to_string()],
rows: lines
.into_iter()
.map(|line| vec![Value::Literal(Literal::String(line))])
.collect(),
})
}
fn notify_observer(
options: &ExecutionOptions,
stmt: &Statement,
started: Instant,
guard: &ExecutionGuard<'_>,
result: &Result<QueryResult, QueryError>,
) {
let Some(observer) = &options.observer else {
return;
};
let (result_rows, outcome) = match result {
Ok(result) => (Some(result.rows.len()), ExecutionOutcome::Success),
Err(error) => (None, ExecutionOutcome::from_error(error)),
};
observer.observe(&ExecutionEvent {
elapsed: started.elapsed(),
statement_read_only: Some(is_read_only(stmt)),
result_rows,
relationship_expansions: guard.relationship_expansions.get(),
outcome,
});
}
fn execute_in_write_transaction_validated(
&self,
stmt: &Statement,
write_txn: &WriteTransaction,
guard: &ExecutionGuard<'_>,
) -> Result<QueryResult, QueryError> {
match stmt {
Statement::Create(patterns) => {
guard.checkpoint()?;
self.execute_create(write_txn, patterns, guard)
}
Statement::CreateIndex {
label,
prop,
unique,
} => {
guard.checkpoint()?;
GraphStore::create_index_in_txn(write_txn, label, prop, *unique)?;
Ok(QueryResult {
columns: vec![],
rows: vec![],
})
}
Statement::Match {
clauses,
tail,
order_by,
skip,
limit,
} => {
let skip =
self.resolve_skip_limit(Txn::Write(write_txn), skip.as_deref(), "SKIP", guard)?;
let limit = self.resolve_skip_limit(
Txn::Write(write_txn),
limit.as_deref(),
"LIMIT",
guard,
)?;
self.execute_match(
Txn::Write(write_txn),
clauses,
tail,
ResultModifiers {
order_by,
skip,
limit,
},
guard,
)
}
Statement::Explain(inner) => {
self.execute_explain(inner)
}
Statement::Union { parts, all } => {
self.materialize_union(Txn::Write(write_txn), parts, *all, guard)
}
Statement::StandaloneCall(call) => {
self.eval_standalone_call(Txn::Write(write_txn), call, guard)
}
}
}
fn eval_call_clause(
&self,
txn: Txn,
call: &CallClause,
current_rows: &[BindingRow],
guard: &ExecutionGuard<'_>,
) -> Result<Vec<BindingRow>, QueryError> {
let mut out = Vec::new();
for row in current_rows {
guard.checkpoint()?;
let (sig, proc_rows) = self.call_procedure(txn, call, row, guard)?;
let Some(yield_items) = &call.yield_items else {
out.push(row.clone());
continue;
};
let names: Vec<String> = match yield_items {
CallYield::Star => sig.outputs.clone(),
CallYield::Items(items, _) => items
.iter()
.map(|(name, alias)| alias.clone().unwrap_or_else(|| name.clone()))
.collect(),
};
for proc_row in &proc_rows {
let projected = project_call_row(&sig, proc_row, yield_items)?;
let mut new_row = row.clone();
for (name, value) in names.iter().zip(&projected) {
new_row.insert(name.clone(), value_to_binding_restore(value));
}
if let CallYield::Items(_, Some(where_expr)) = yield_items {
if self.eval_expr(txn, where_expr, &new_row, guard)? != Some(true) {
continue;
}
}
out.push(new_row);
guard.check_intermediate_rows(out.len())?;
}
}
Ok(out)
}
fn eval_standalone_call(
&self,
txn: Txn,
call: &CallClause,
guard: &ExecutionGuard<'_>,
) -> Result<QueryResult, QueryError> {
let empty_row = BindingRow::new();
let (sig, proc_rows) = self.call_procedure(txn, call, &empty_row, guard)?;
let yield_items = call.yield_items.clone().unwrap_or(CallYield::Star);
let columns: Vec<String> = match &yield_items {
CallYield::Star => sig.outputs.clone(),
CallYield::Items(items, _) => items
.iter()
.map(|(name, alias)| alias.clone().unwrap_or_else(|| name.clone()))
.collect(),
};
let mut rows = Vec::with_capacity(proc_rows.len());
for proc_row in &proc_rows {
rows.push(project_call_row(&sig, proc_row, &yield_items)?);
}
if let CallYield::Items(_, Some(where_expr)) = &yield_items {
let mut filtered = Vec::with_capacity(rows.len());
for row_values in &rows {
let mut binding_row = BindingRow::new();
for (col, v) in columns.iter().zip(row_values) {
binding_row.insert(col.clone(), value_to_binding_restore(v));
}
if self.eval_expr(txn, where_expr, &binding_row, guard)? == Some(true) {
filtered.push(row_values.clone());
}
}
rows = filtered;
}
Ok(QueryResult { columns, rows })
}
fn call_procedure(
&self,
txn: Txn,
call: &CallClause,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<(ProcedureSignature, Vec<Vec<Value>>), QueryError> {
let provider = guard.procedure_provider().ok_or_else(|| {
QueryError::Semantic(format!(
"procedure '{}' not found -- no procedure provider is configured",
call.name
))
})?;
let sig = provider
.signature(&call.name)
.ok_or_else(|| QueryError::Semantic(format!("procedure '{}' not found", call.name)))?;
let args = self.eval_call_args(txn, call, &sig, row, guard)?;
let rows = provider.call(&call.name, &args)?;
Ok((sig, rows))
}
fn eval_call_args(
&self,
txn: Txn,
call: &CallClause,
sig: &ProcedureSignature,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<Vec<Value>, QueryError> {
let values: Vec<Value> = match &call.args {
Some(args) => {
if args.len() != sig.inputs.len() {
return Err(QueryError::Semantic(format!(
"'{}' expects {} argument(s), got {}",
call.name,
sig.inputs.len(),
args.len()
)));
}
args.iter()
.map(|a| self.eval_return_expr(txn, a, row, guard))
.collect::<Result<_, _>>()?
}
None => sig
.inputs
.iter()
.map(|input_name| {
guard
.options
.params
.get(input_name)
.cloned()
.map(property_value_to_value)
.ok_or_else(|| QueryError::MissingParam(input_name.clone()))
})
.collect::<Result<_, _>>()?,
};
for (value, (input_name, declared_type)) in
values.iter().zip(sig.inputs.iter().zip(&sig.input_types))
{
if !value_matches_declared_type(value, declared_type) {
return Err(QueryError::Type(format!(
"'{}' argument '{input_name}' expects {declared_type}, got {value:?}",
call.name
)));
}
}
Ok(values)
}
fn execute_create(
&self,
write_txn: &WriteTransaction,
patterns: &[Pattern],
guard: &ExecutionGuard<'_>,
) -> Result<QueryResult, QueryError> {
self.materialize_create(write_txn, patterns, &[BindingRow::new()], guard)?;
Ok(QueryResult {
columns: vec![],
rows: vec![],
})
}
fn materialize_create(
&self,
write_txn: &WriteTransaction,
patterns: &[Pattern],
rows: &[BindingRow],
guard: &ExecutionGuard<'_>,
) -> Result<Vec<BindingRow>, QueryError> {
let mut out = Vec::with_capacity(rows.len());
for row in rows {
let mut row = row.clone();
for pattern in patterns {
let mut prev_id =
self.resolve_or_create_node(write_txn, &pattern.start, &row, guard)?;
if let Some(var) = &pattern.start.var {
row.insert(var.clone(), Binding::Node(prev_id));
}
for (rel, node) in &pattern.hops {
if rel.hop_range.is_some() {
return Err(QueryError::Semantic(
"CREATE doesn't support variable-length relationship patterns (e.g. [:TYPE*1..3])".into(),
));
}
let node_id = self.resolve_or_create_node(write_txn, node, &row, guard)?;
if let Some(var) = &node.var {
row.insert(var.clone(), Binding::Node(node_id));
}
let rel_label = rel.rel_types.first().cloned().expect(
"CREATE relationship has exactly one type -- checked by \
semantic::bind_create_pattern",
);
let rel_props =
self.eval_props_to_values(Txn::Write(write_txn), &rel.props, &row, guard)?;
let (src, dst) = match rel.direction {
RelDirection::Right => (prev_id, node_id),
RelDirection::Left => (node_id, prev_id),
RelDirection::Either => {
return Err(QueryError::Semantic(
"CREATE requires a directed relationship (-> or <-), not an undirected pattern".into(),
))
}
};
let edge_id =
GraphStore::create_edge_in_txn(write_txn, &rel_label, src, dst, rel_props)?;
if let Some(var) = &rel.var {
row.insert(var.clone(), Binding::Edge(edge_id));
}
prev_id = node_id;
}
}
out.push(row);
}
Ok(out)
}
fn resolve_or_create_node(
&self,
write_txn: &WriteTransaction,
node: &NodePattern,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<NodeId, QueryError> {
if let Some(var) = &node.var {
if let Some(binding) = row.get(var) {
let Binding::Node(id) = binding else {
return Err(QueryError::Type(format!(
"'{var}' is not a node — can't use it as a CREATE pattern endpoint"
)));
};
return Ok(*id);
}
}
let labels: Vec<&str> = node.labels.iter().map(String::as_str).collect();
let props = self.eval_props_to_values(Txn::Write(write_txn), &node.props, row, guard)?;
Ok(GraphStore::create_node_in_txn(write_txn, &labels, props)?)
}
fn eval_props_to_values(
&self,
txn: Txn,
props: &[(String, ReturnExpr)],
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<BTreeMap<String, PropertyValue>, QueryError> {
props
.iter()
.filter_map(|(k, expr)| {
let value = match self.eval_return_expr(txn, expr, row, guard) {
Ok(v) => v,
Err(e) => return Some(Err(e)),
};
if matches!(value, Value::Null) {
return None;
}
let pv = match value_to_storable_property(&value).ok_or_else(|| {
QueryError::Type(format!(
"property '{k}' can't be stored -- MarsDB's node/edge properties are limited to null/\
bool/int/float/string/date/duration; a list/map/node/edge/path value (got {value:?}) \
isn't storable, matching PropertyValue's real, deliberately fixed set of variants (see \
its doc comment)"
))
}) {
Ok(pv) => pv,
Err(e) => return Some(Err(e)),
};
Some(Ok((k.clone(), pv)))
})
.collect()
}
fn eval_merge(
&self,
write_txn: &WriteTransaction,
clause: &MergeClause,
rows: &[BindingRow],
guard: &ExecutionGuard<'_>,
) -> Result<Vec<BindingRow>, QueryError> {
let mut out = Vec::new();
for row in rows {
guard.checkpoint()?;
out.extend(self.merge_one_row(write_txn, clause, row, guard)?);
guard.check_intermediate_rows(out.len())?;
}
self.apply_merge_set(write_txn, clause, &mut out, guard)?;
Ok(out)
}
fn merge_pattern_has_null_property(
&self,
txn: Txn,
clause: &MergeClause,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<bool, QueryError> {
let any_null = |props: &[(String, ReturnExpr)]| -> Result<bool, QueryError> {
for (_, expr) in props {
if matches!(self.eval_return_expr(txn, expr, row, guard)?, Value::Null) {
return Ok(true);
}
}
Ok(false)
};
if any_null(&clause.pattern.start.props)? {
return Ok(true);
}
for (rel, node) in &clause.pattern.hops {
if any_null(&rel.props)? || any_null(&node.props)? {
return Ok(true);
}
}
Ok(false)
}
fn merge_one_row(
&self,
write_txn: &WriteTransaction,
clause: &MergeClause,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<Vec<BindingRow>, QueryError> {
for (rel, _node) in &clause.pattern.hops {
if rel.hop_range.is_some() {
return Err(QueryError::Semantic(
"MERGE doesn't support variable-length relationship patterns (e.g. [:TYPE*1..3])".into(),
));
}
}
let (pattern, synthesized) = if clause.path_var.is_some() {
name_pattern_for_path(&clause.pattern)
} else {
(clause.pattern.clone(), HashSet::new())
};
let pattern = &pattern;
if self.merge_pattern_has_null_property(Txn::Write(write_txn), clause, row, guard)? {
return Err(QueryError::Semantic(
"MERGE pattern property is null — a MERGE's own {...} properties can never be \
null (searching for null never matches anything, but storing null is the same \
as not storing the property at all)"
.into(),
));
}
let carried_vars: HashSet<String> = row.keys().cloned().collect();
let plan = apply_index_seeks(
build_match_plan(pattern, &None, &carried_vars)?,
Txn::Write(write_txn),
)?;
let found = self.eval_plan(
Txn::Write(write_txn),
&plan,
std::slice::from_ref(row),
guard,
)?;
if !found.is_empty() {
return Ok(found
.into_iter()
.map(|mut r| {
if let Some(path_var) = &clause.path_var {
let path_binding = assemble_path(pattern, &r);
for key in &synthesized {
r.remove(key);
}
r.insert(path_var.clone(), path_binding);
}
tag_merge_created(r, false)
})
.collect());
}
let mut new_row = row.clone();
let start_id = self.resolve_or_create_node(write_txn, &pattern.start, &new_row, guard)?;
if let Some(var) = &pattern.start.var {
new_row.insert(var.clone(), Binding::Node(start_id));
}
if let Some((rel, node)) = pattern.hops.first() {
let node_id = self.resolve_or_create_node(write_txn, node, &new_row, guard)?;
if let Some(var) = &node.var {
new_row.insert(var.clone(), Binding::Node(node_id));
}
let rel_label = rel.rel_types.first().cloned().expect(
"MERGE relationship has exactly one type -- checked by semantic::bind_merge",
);
let rel_props =
self.eval_props_to_values(Txn::Write(write_txn), &rel.props, &new_row, guard)?;
let (src, dst) = match rel.direction {
RelDirection::Right | RelDirection::Either => (start_id, node_id),
RelDirection::Left => (node_id, start_id),
};
let edge_id =
GraphStore::create_edge_in_txn(write_txn, &rel_label, src, dst, rel_props)?;
if let Some(var) = &rel.var {
new_row.insert(var.clone(), Binding::Edge(edge_id));
}
}
if let Some(path_var) = &clause.path_var {
let path_binding = assemble_path(pattern, &new_row);
for key in &synthesized {
new_row.remove(key);
}
new_row.insert(path_var.clone(), path_binding);
}
Ok(vec![tag_merge_created(new_row, true)])
}
fn apply_merge_set(
&self,
write_txn: &WriteTransaction,
clause: &MergeClause,
rows: &mut [BindingRow],
guard: &ExecutionGuard<'_>,
) -> Result<(), QueryError> {
for row in rows.iter_mut() {
let created = match row.remove(MERGE_CREATED_KEY) {
Some(Binding::Value(PropertyValue::Bool(b))) => b,
other => unreachable!(
"{MERGE_CREATED_KEY} tagged internally as Binding::Value(Bool), got {other:?}"
),
};
let items = if created {
&clause.on_create
} else {
&clause.on_match
};
for item in items {
self.apply_set_item(Txn::Write(write_txn), write_txn, row, item, guard)?;
}
}
Ok(())
}
fn execute_match(
&self,
txn: Txn,
clauses: &[QueryClause],
tail: &Option<Tail>,
modifiers: ResultModifiers<'_>,
guard: &ExecutionGuard<'_>,
) -> Result<QueryResult, QueryError> {
self.execute_match_seeded(txn, clauses, tail, modifiers, None, guard)
}
fn execute_match_seeded(
&self,
txn: Txn,
clauses: &[QueryClause],
tail: &Option<Tail>,
modifiers: ResultModifiers<'_>,
seed: Option<&BindingRow>,
guard: &ExecutionGuard<'_>,
) -> Result<QueryResult, QueryError> {
let ResultModifiers {
order_by,
skip,
limit,
} = modifiers;
let mut carried_vars: HashSet<String> = match seed {
Some(row) => row.keys().cloned().collect(),
None => HashSet::new(),
};
let mut current_rows: Vec<BindingRow> = vec![seed.cloned().unwrap_or_default()];
let final_stream_limit = match (order_by, limit, tail) {
(None, Some(limit), Some(Tail::Return(items, false))) if !has_aggregate(items) => {
Some(skip.unwrap_or(0).max(0) as usize + limit.max(0) as usize)
}
_ => None,
};
for (clause_index, clause) in clauses.iter().enumerate() {
let is_final_clause = clause_index + 1 == clauses.len();
match clause {
QueryClause::Match(part) => {
let plan_limit = is_final_clause
.then_some(final_stream_limit)
.flatten()
.filter(|_| !part.shortest_path && !part.optional && part.with.is_none());
current_rows = if part.shortest_path {
self.eval_shortest_path(txn, part, ¤t_rows, guard)?
} else if let Some(path_var) = &part.path_var {
let (named_pattern, synthesized) = name_pattern_for_path(&part.pattern);
let defer_where = !part.optional && part.where_clause.is_some();
let plan_where = if defer_where {
&None
} else {
&part.where_clause
};
let plan = apply_index_seeks(
build_match_plan(&named_pattern, plan_where, &carried_vars)?,
txn,
)?;
let mut rows = if part.optional {
let new_vars = pattern_new_vars(&named_pattern, &carried_vars);
self.eval_optional_part(txn, &plan, ¤t_rows, &new_vars, guard)?
} else {
let limit = plan_limit.filter(|_| !defer_where);
self.eval_plan_with_limit(txn, &plan, ¤t_rows, guard, limit)?
};
for row in &mut rows {
let path_binding = assemble_path(&named_pattern, row);
for key in &synthesized {
row.remove(key);
}
row.insert(path_var.clone(), path_binding);
}
if defer_where {
let where_clause = part
.where_clause
.as_ref()
.expect("defer_where implies where_clause is Some");
let mut filtered = Vec::with_capacity(rows.len());
for row in rows {
if self.eval_expr(txn, where_clause, &row, guard)? == Some(true) {
filtered.push(row);
}
}
rows = filtered;
}
rows
} else {
let reversed = plan_reversed_pattern(
&part.pattern,
&part.where_clause,
&carried_vars,
txn,
)?;
let pattern = reversed.as_ref().unwrap_or(&part.pattern);
let plan = apply_index_seeks(
build_match_plan(pattern, &part.where_clause, &carried_vars)?,
txn,
)?;
if part.optional {
let new_vars = pattern_new_vars(&part.pattern, &carried_vars);
self.eval_optional_part(txn, &plan, ¤t_rows, &new_vars, guard)?
} else {
let tail_hint = if is_final_clause {
match (order_by, limit, tail) {
(
Some(keys),
Some(tail_limit),
Some(Tail::Return(items, false)),
) if keys.len() == 1 && !has_aggregate(items) => {
let (key, dir) = &keys[0];
Some((
key,
*dir,
skip.unwrap_or(0).max(0) as usize
+ tail_limit.max(0) as usize,
))
}
_ => None,
}
} else {
None
};
if let Some((rows, out_names)) = self.try_fast_expand_expand_count(
txn,
&plan,
&part.with,
¤t_rows,
tail_hint,
guard,
)? {
current_rows = rows;
carried_vars = out_names;
continue;
}
self.eval_plan_with_limit(txn, &plan, ¤t_rows, guard, plan_limit)?
}
};
let mut new_vars = pattern_all_vars(&part.pattern);
if let Some(path_var) = &part.path_var {
new_vars.insert(path_var.clone());
}
current_rows = self.apply_with_or_carry(
txn,
&part.with,
current_rows,
new_vars,
&mut carried_vars,
guard,
)?;
}
QueryClause::Unwind(u) => {
current_rows = self.eval_unwind(txn, u, ¤t_rows, guard)?;
current_rows = self.apply_with_or_carry(
txn,
&u.with,
current_rows,
HashSet::from([u.var.clone()]),
&mut carried_vars,
guard,
)?;
}
QueryClause::Call(call) => {
current_rows = self.eval_call_clause(txn, call, ¤t_rows, guard)?;
let new_vars: HashSet<String> = match &call.yield_items {
Some(CallYield::Items(items, _)) => items
.iter()
.map(|(name, alias)| alias.clone().unwrap_or_else(|| name.clone()))
.collect(),
Some(CallYield::Star) | None => HashSet::new(),
};
current_rows = self.apply_with_or_carry(
txn,
&call.with,
current_rows,
new_vars,
&mut carried_vars,
guard,
)?;
}
QueryClause::Merge(m) => {
let write_txn = require_write_txn(txn);
current_rows = self.eval_merge(write_txn, m, ¤t_rows, guard)?;
let mut new_vars = pattern_all_vars(&m.pattern);
if let Some(path_var) = &m.path_var {
new_vars.insert(path_var.clone());
}
current_rows = self.apply_with_or_carry(
txn,
&m.with,
current_rows,
new_vars,
&mut carried_vars,
guard,
)?;
}
QueryClause::With(with) => {
current_rows = self.apply_with_or_carry(
txn,
&Some(with.clone()),
current_rows,
HashSet::new(),
&mut carried_vars,
guard,
)?;
}
QueryClause::Set(items) => {
let write_txn = require_write_txn(txn);
for row in ¤t_rows {
for item in items {
self.apply_set_item(txn, write_txn, row, item, guard)?;
}
}
}
QueryClause::Delete { items, detach } => {
let write_txn = require_write_txn(txn);
self.delete_targets(txn, write_txn, items, ¤t_rows, *detach, guard)?;
}
QueryClause::Remove(items) => {
let write_txn = require_write_txn(txn);
for row in ¤t_rows {
for item in items {
apply_remove_item(write_txn, row, item)?;
}
}
}
QueryClause::Create(patterns) => {
let write_txn = require_write_txn(txn);
current_rows =
self.materialize_create(write_txn, patterns, ¤t_rows, guard)?;
carried_vars.extend(patterns.iter().flat_map(pattern_all_vars));
}
}
guard.check_intermediate_rows(current_rows.len())?;
}
let distinct_return = tail_is_distinct_return(tail);
if order_by.is_none() && !distinct_return {
let skip_n = skip.unwrap_or(0).max(0) as usize;
if skip_n > 0 {
current_rows.drain(0..skip_n.min(current_rows.len()));
}
if let Some(count) = limit {
current_rows.truncate(count.max(0) as usize);
}
}
let mut order_by_pre_applied = false;
let mut result = match tail {
None => QueryResult {
columns: vec![],
rows: vec![],
},
Some(Tail::Return(items, distinct)) => {
if let Some(ob) = order_by {
if !has_aggregate(items) && !distinct {
let projected =
self.materialize_return(txn, items, ¤t_rows, *distinct, guard)?;
order_by_pre_applied = true;
self.apply_order_by_with_scope(
txn,
¤t_rows,
projected,
ob,
skip,
limit,
)?
} else if !distinct {
order_by_pre_applied = true;
self.materialize_aggregating_return_with_order(
txn,
items,
¤t_rows,
ob,
(skip, limit),
guard,
)?
} else {
self.materialize_return(txn, items, ¤t_rows, *distinct, guard)?
}
} else {
self.materialize_return(txn, items, ¤t_rows, *distinct, guard)?
}
}
Some(Tail::ReturnStar(distinct)) => {
let items = return_star_items(carried_vars.iter().cloned())?;
let projected =
self.materialize_return(txn, &items, ¤t_rows, *distinct, guard)?;
if let Some(ob) = order_by {
if !distinct {
order_by_pre_applied = true;
self.apply_order_by_with_scope(
txn,
¤t_rows,
projected,
ob,
skip,
limit,
)?
} else {
projected
}
} else {
projected
}
}
Some(Tail::Delete(vars, ret)) => {
self.materialize_delete(txn, vars, ¤t_rows, false, ret, guard)?
}
Some(Tail::DetachDelete(vars, ret)) => {
self.materialize_delete(txn, vars, ¤t_rows, true, ret, guard)?
}
Some(Tail::Set(items, ret)) => {
self.materialize_set(txn, items, ¤t_rows, ret, guard)?
}
Some(Tail::Remove(items, ret)) => {
self.materialize_remove(txn, items, ¤t_rows, ret, guard)?
}
Some(Tail::Create(patterns, ret)) => {
let updated_rows = self.materialize_create(
require_write_txn(txn),
patterns,
¤t_rows,
guard,
)?;
match ret {
Some(rt) => {
self.materialize_return(txn, &rt.items, &updated_rows, rt.distinct, guard)?
}
None => QueryResult {
columns: vec![],
rows: vec![],
},
}
}
};
if let Some(order_by) = order_by {
if !order_by_pre_applied {
let tail_items: Option<&[ReturnItem]> = match tail {
Some(Tail::Return(items, _)) => Some(items),
_ => None,
};
result.rows = apply_order_by(
result.rows,
&result.columns,
order_by,
tail_items,
skip,
limit,
)?;
}
} else if distinct_return {
let skip_n = skip.unwrap_or(0).max(0) as usize;
if skip_n > 0 {
result.rows.drain(0..skip_n.min(result.rows.len()));
}
if let Some(count) = limit {
result.rows.truncate(count.max(0) as usize);
}
}
guard.check_result_rows(result.rows.len())?;
Ok(result)
}
fn apply_with_or_carry(
&self,
txn: Txn,
with: &Option<WithClause>,
rows: Vec<BindingRow>,
new_vars: HashSet<String>,
carried_vars: &mut HashSet<String>,
guard: &ExecutionGuard<'_>,
) -> Result<Vec<BindingRow>, QueryError> {
let Some(with) = with else {
carried_vars.extend(new_vars);
return Ok(rows);
};
let with_owned;
let with: &WithClause = if with.star {
let star_items = with_star_items(carried_vars.union(&new_vars).cloned());
let mut owned = with.clone();
let mut items = star_items;
items.extend(owned.items);
owned.items = items;
with_owned = owned;
&with_owned
} else {
with
};
let with_skip = self.resolve_skip_limit(txn, with.skip.as_ref(), "SKIP", guard)?;
let with_limit = self.resolve_skip_limit(txn, with.limit.as_ref(), "LIMIT", guard)?;
let rows = if let Some(with_order_by) = with
.order_by
.as_ref()
.filter(|_| has_aggregate(&with.items))
{
self.materialize_aggregating_with_with_order(
txn,
&with.items,
&rows,
with_order_by,
(with_skip, with_limit),
guard,
)?
} else {
let pre_with_rows = (with.order_by.is_some() && !with.distinct).then(|| rows.clone());
let mut rows = self.materialize_with(txn, with, &rows, guard)?;
if let Some(with_order_by) = &with.order_by {
rows = self.apply_order_by_bindings(
txn,
rows,
pre_with_rows.as_deref(),
&with.items,
with_order_by,
(with_skip, with_limit),
)?;
} else {
let skip_n = with_skip.unwrap_or(0).max(0) as usize;
if skip_n > 0 {
rows.drain(0..skip_n.min(rows.len()));
}
if let Some(with_limit) = with_limit {
rows.truncate(with_limit.max(0) as usize);
}
}
rows
};
*carried_vars = with
.items
.iter()
.enumerate()
.map(with_item_output_name)
.collect();
Ok(rows)
}
fn eval_unwind(
&self,
txn: Txn,
clause: &UnwindClause,
rows: &[BindingRow],
guard: &ExecutionGuard<'_>,
) -> Result<Vec<BindingRow>, QueryError> {
let mut out = Vec::new();
for row in rows {
let source_value = self.eval_return_expr(txn, &clause.source.0, row, guard)?;
let elements: Vec<Binding> = match source_value {
Value::List(items) => items.iter().map(value_to_binding_restore).collect(),
Value::Null => Vec::new(),
other => {
return Err(QueryError::Type(format!(
"UNWIND needs a list, got {other:?}"
)))
}
};
for element in elements {
let mut new_row = row.clone();
new_row.insert(clause.var.clone(), element);
out.push(new_row);
}
}
if let Some(where_clause) = &clause.where_clause {
let mut filtered = Vec::with_capacity(out.len());
for row in out {
if self.eval_with_expr(txn, where_clause, &row, guard)? == Some(true) {
filtered.push(row);
}
}
out = filtered;
}
Ok(out)
}
fn eval_shortest_path(
&self,
txn: Txn,
part: &QueryPart,
rows: &[BindingRow],
guard: &ExecutionGuard<'_>,
) -> Result<Vec<BindingRow>, QueryError> {
let Some(path_var) = &part.path_var else {
return Ok(rows.to_vec());
};
let start_var = part.pattern.start.var.as_deref().expect(
"shortestPath()'s start node always has a var — validated at parse time by \
validate_shortest_path_pattern",
);
let (rel, end_node) = &part.pattern.hops[0];
let end_var = end_node.var.as_deref().expect(
"shortestPath()'s end node always has a var — validated at parse time by \
validate_shortest_path_pattern",
);
let (min_hops, max_hops) = rel.hop_range.expect(
"shortestPath()'s relationship is always variable-length — validated at parse time by \
validate_shortest_path_pattern",
);
let direction = match rel.direction {
RelDirection::Right => ExpandDirection::Out,
RelDirection::Left => ExpandDirection::In,
RelDirection::Either => ExpandDirection::Either,
};
let rel_labels = &rel.rel_types;
let mut out = Vec::with_capacity(rows.len());
for row in rows {
let start_id = require_bound_node(row, start_var)?;
let end_id = require_bound_node(row, end_var)?;
let path = self.shortest_path_between(
txn,
start_id,
end_id,
ShortestPathSpec {
direction,
rel_labels,
min_hops,
max_hops,
},
)?;
let mut new_row = row.clone();
let binding = match path {
Some(elems) => Binding::Path(elems),
None => Binding::Value(PropertyValue::Null),
};
new_row.insert(path_var.clone(), binding);
out.push(new_row);
}
if let Some(where_clause) = &part.where_clause {
let mut filtered = Vec::with_capacity(out.len());
for row in out {
if self.eval_expr(txn, where_clause, &row, guard)? == Some(true) {
filtered.push(row);
}
}
out = filtered;
}
Ok(out)
}
fn shortest_path_between(
&self,
txn: Txn,
start: NodeId,
end: NodeId,
spec: ShortestPathSpec<'_>,
) -> Result<Option<Vec<PathBinding>>, QueryError> {
if start == end && spec.min_hops == 0 {
return Ok(Some(vec![PathBinding::Node(start)]));
}
let cap = spec.max_hops.unwrap_or(VAR_EXPAND_DEPTH_CAP);
let mut parent: HashMap<NodeId, (NodeId, EdgeId)> = HashMap::new();
let mut visited: HashSet<NodeId> = HashSet::new();
visited.insert(start);
let mut frontier = vec![start];
let mut depth = 0u32;
while depth < cap && !frontier.is_empty() {
depth += 1;
let mut next_frontier = Vec::new();
for node in frontier {
for entry in neighbors_for_direction(txn, node, spec.direction, spec.rel_labels)? {
if entry.other == end {
parent.insert(entry.other, (node, entry.edge_id));
return Ok(Some(reconstruct_path(&parent, start, end)));
}
if visited.insert(entry.other) {
parent.insert(entry.other, (node, entry.edge_id));
next_frontier.push(entry.other);
}
}
}
frontier = next_frontier;
}
Ok(None)
}
fn materialize_with(
&self,
txn: Txn,
with: &WithClause,
rows: &[BindingRow],
guard: &ExecutionGuard<'_>,
) -> Result<Vec<BindingRow>, QueryError> {
let is_aggregating = has_aggregate(&with.items);
let mut out = if !is_aggregating {
let mut out = Vec::with_capacity(rows.len());
for row in rows {
let mut new_row = BindingRow::new();
for (i, item) in with.items.iter().enumerate() {
let name = with_item_output_name((i, item));
let binding = self.item_binding(txn, &item.expr, row, guard)?;
new_row.insert(name, binding);
}
out.push(new_row);
}
out
} else {
validate_return_items(&with.items)?;
let grouped = self.resolve_grouped_rows(txn, &with.items, rows, guard)?;
grouped
.into_iter()
.map(|bindings| {
with.items
.iter()
.enumerate()
.zip(bindings)
.map(|((i, item), b)| (with_item_output_name((i, item)), b))
.collect()
})
.collect()
};
if let Some(where_clause) = &with.where_clause {
let mut filtered = Vec::with_capacity(out.len());
if is_aggregating {
for row in out {
if self.eval_with_expr(txn, where_clause, &row, guard)? == Some(true) {
filtered.push(row);
}
}
} else {
for (row, new_row) in rows.iter().zip(out) {
let mut merged = row.clone();
merged.extend(new_row.iter().map(|(k, v)| (k.clone(), v.clone())));
if self.eval_with_expr(txn, where_clause, &merged, guard)? == Some(true) {
filtered.push(new_row);
}
}
}
out = filtered;
}
if with.distinct {
out = dedup_binding_rows(&with.items, out)?;
}
Ok(out)
}
fn materialize_aggregating_with_with_order(
&self,
txn: Txn,
with_items: &[ReturnItem],
rows: &[BindingRow],
order_by: &[(ReturnExpr, SortDir)],
skip_limit: (Option<i64>, Option<i64>),
guard: &ExecutionGuard<'_>,
) -> Result<Vec<BindingRow>, QueryError> {
let (skip, limit) = skip_limit;
enum OrderKeySource {
RealColumn(usize),
Extra(usize),
}
let mut extra_exprs: Vec<ReturnExpr> = Vec::new();
let order_by_source: Vec<OrderKeySource> = order_by
.iter()
.map(|(expr, _)| {
match with_items
.iter()
.enumerate()
.position(|(i, it)| item_matches_leaf(expr, i, it))
{
Some(i) => OrderKeySource::RealColumn(i),
None => {
let idx = extra_exprs.len();
extra_exprs.push(expr.clone());
OrderKeySource::Extra(idx)
}
}
})
.collect();
let extended_items: Vec<ReturnItem> = with_items
.iter()
.cloned()
.chain(
extra_exprs
.into_iter()
.map(|expr| ReturnItem { expr, alias: None }),
)
.collect();
validate_return_items(&extended_items)?;
let grouped = self.resolve_grouped_rows(txn, &extended_items, rows, guard)?;
let real_len = with_items.len();
let mut keyed: Vec<(Vec<Value>, BindingRow)> = Vec::with_capacity(grouped.len());
for bindings in grouped {
let (real, extra) = bindings.split_at(real_len);
let real_values: Vec<Value> = real
.iter()
.map(|b| self.binding_to_value(txn, b))
.collect::<Result<Vec<_>, _>>()?;
let extra_values: Vec<Value> = extra
.iter()
.map(|b| self.binding_to_value(txn, b))
.collect::<Result<Vec<_>, _>>()?;
let keys: Vec<Value> = order_by_source
.iter()
.map(|src| match src {
OrderKeySource::RealColumn(i) => real_values[*i].clone(),
OrderKeySource::Extra(k) => extra_values[*k].clone(),
})
.collect();
let real_row: BindingRow = with_items
.iter()
.enumerate()
.zip(real)
.map(|((i, item), binding)| (with_item_output_name((i, item)), binding.clone()))
.collect();
keyed.push((keys, real_row));
}
Ok(top_k_by(keyed, order_by, skip, limit)
.into_iter()
.map(|(_, row)| row)
.collect())
}
fn item_binding(
&self,
txn: Txn,
expr: &ReturnExpr,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<Binding, QueryError> {
match expr {
ReturnExpr::Var(v) => row
.get(v)
.cloned()
.ok_or_else(|| QueryError::UnboundVariable(v.clone())),
other => {
let value = self.eval_return_expr(txn, other, row, guard)?;
Ok(match value {
Value::Node(n) => Binding::Node(n.id),
Value::Edge(e) => Binding::Edge(e.id),
Value::List(items) => Binding::List(items),
Value::Map(m) => Binding::Map(m),
other => Binding::Value(value_to_property_value(&other)),
})
}
}
}
fn apply_order_by_bindings(
&self,
txn: Txn,
rows: Vec<BindingRow>,
pre_with_rows: Option<&[BindingRow]>,
with_items: &[ReturnItem],
order_by: &[(ReturnExpr, SortDir)],
skip_limit: (Option<i64>, Option<i64>),
) -> Result<Vec<BindingRow>, QueryError> {
let (skip, limit) = skip_limit;
let order_by_output: Vec<Option<String>> = order_by
.iter()
.map(|(expr, _)| {
with_items
.iter()
.enumerate()
.find(|(_, item)| item.expr == *expr)
.map(with_item_output_name)
})
.collect();
let mut keyed: Vec<(Vec<Value>, BindingRow)> = Vec::with_capacity(rows.len());
for (i, row) in rows.into_iter().enumerate() {
let mut value_map = self.binding_row_to_value_map(txn, &row)?;
if let Some(pre) = pre_with_rows {
for (k, v) in self.binding_row_to_value_map(txn, &pre[i])? {
value_map.entry(k).or_insert(v);
}
}
let keys = order_by
.iter()
.zip(&order_by_output)
.map(|((expr, _), output_name)| match output_name {
Some(name) => Ok(value_map.get(name).cloned().unwrap_or(Value::Null)),
None => eval_projected_expr(expr, &value_map),
})
.collect::<Result<Vec<_>, _>>()?;
keyed.push((keys, row));
}
Ok(top_k_by(keyed, order_by, skip, limit)
.into_iter()
.map(|(_, row)| row)
.collect())
}
fn apply_order_by_with_scope(
&self,
txn: Txn,
binding_rows: &[BindingRow],
result: QueryResult,
order_by: &[(ReturnExpr, SortDir)],
skip: Option<i64>,
limit: Option<i64>,
) -> Result<QueryResult, QueryError> {
let QueryResult { columns, rows } = result;
let mut keyed: Vec<(Vec<Value>, Vec<Value>)> = Vec::with_capacity(rows.len());
for (binding_row, row) in binding_rows.iter().zip(rows) {
let mut value_map = self.binding_row_to_value_map(txn, binding_row)?;
for (col, val) in columns.iter().zip(&row) {
value_map.insert(col.clone(), val.clone());
}
let keys = order_by
.iter()
.map(|(expr, _)| eval_projected_expr(expr, &value_map))
.collect::<Result<Vec<_>, _>>()?;
keyed.push((keys, row));
}
let rows = top_k_by(keyed, order_by, skip, limit)
.into_iter()
.map(|(_, row)| row)
.collect();
Ok(QueryResult { columns, rows })
}
fn binding_row_to_value_map(
&self,
txn: Txn,
row: &BindingRow,
) -> Result<HashMap<String, Value>, QueryError> {
let mut map = HashMap::with_capacity(row.len());
for (k, binding) in row {
map.insert(k.clone(), self.binding_to_value(txn, binding)?);
}
Ok(map)
}
fn binding_to_value(&self, txn: Txn, b: &Binding) -> Result<Value, QueryError> {
Ok(match b {
Binding::Node(id) => {
Value::Node((*deleted_entity_access(self.get_node_cached(txn, *id)?)?).clone())
}
Binding::Edge(id) => Value::Edge(deleted_entity_access(GraphStore::get_edge_in_txn(
txn, *id,
)?)?),
Binding::Value(PropertyValue::Null) => Value::Null,
Binding::Value(pv) => property_value_to_value(pv.clone()),
Binding::List(items) => Value::List(items.clone()),
Binding::Map(m) => Value::Map(m.clone()),
Binding::Path(elems) => Value::Path(self.resolve_path_elems(txn, elems)?),
})
}
fn start_or_end_node(
&self,
txn: Txn,
which: &str,
arg: Option<&Value>,
) -> Result<Value, QueryError> {
match arg {
None | Some(Value::Null) => Ok(Value::Null),
Some(Value::Edge(e)) => {
let id = if which == "startnode" { e.src } else { e.dst };
let node = deleted_entity_access(self.get_node_cached(txn, id)?)?;
Ok(Value::Node((*node).clone()))
}
Some(other) => Err(QueryError::Type(format!(
"{which}() expects a relationship, got {other:?}"
))),
}
}
fn eval_type_call(
&self,
txn: Txn,
arg_expr: Option<&ReturnExpr>,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<Value, QueryError> {
let Some(arg_expr) = arg_expr else {
return type_builtin(None);
};
match self.eval_return_expr(txn, arg_expr, row, guard) {
Ok(v) => type_builtin(Some(&v)),
Err(err) => {
if let ReturnExpr::Var(v) = arg_expr {
if let Some(Binding::Edge(id)) = row.get(v) {
if let Some(label) = guard.deleted_edge_type(*id) {
return Ok(Value::Property(PropertyValue::String(label)));
}
}
}
Err(err)
}
}
}
fn resolve_path_elems(
&self,
txn: Txn,
elems: &[PathBinding],
) -> Result<Vec<PathElem>, QueryError> {
elems
.iter()
.map(|e| {
Ok(match e {
PathBinding::Node(id) => PathElem::Node(
(*deleted_entity_access(self.get_node_cached(txn, *id)?)?).clone(),
),
PathBinding::Edge(id) => PathElem::Edge(deleted_entity_access(
GraphStore::get_edge_in_txn(txn, *id)?,
)?),
})
})
.collect()
}
fn resolve_grouped_rows(
&self,
txn: Txn,
items: &[ReturnItem],
rows: &[BindingRow],
guard: &ExecutionGuard<'_>,
) -> Result<Vec<Vec<Binding>>, QueryError> {
struct Group {
key_bindings: Vec<Option<Binding>>,
accs: Vec<Vec<AggAcc>>,
row_count: i64,
}
fn fresh_accs(items: &[ReturnItem]) -> Vec<Vec<AggAcc>> {
items
.iter()
.map(|item| {
let mut nodes = Vec::new();
collect_agg_nodes(&item.expr, &mut nodes);
nodes
.into_iter()
.map(|node| match node {
ReturnExpr::CountStar => AggAcc::identity("count", false),
ReturnExpr::Call { name, distinct, .. } => {
AggAcc::identity(name, *distinct)
}
_ => unreachable!(
"collect_agg_nodes only ever collects CountStar/aggregate Call nodes"
),
})
.collect()
})
.collect()
}
let item_agg_nodes: Vec<Vec<&ReturnExpr>> = items
.iter()
.map(|item| {
let mut nodes = Vec::new();
collect_agg_nodes(&item.expr, &mut nodes);
nodes
})
.collect();
let mut groups: Vec<Group> = Vec::new();
let mut group_index: HashMap<Vec<Option<HashKey>>, usize> = HashMap::new();
for row in rows {
let mut key_bindings = Vec::with_capacity(items.len());
for item in items {
key_bindings.push(if contains_aggregate(&item.expr) {
None
} else {
Some(self.item_binding(txn, &item.expr, row, guard)?)
});
}
let hash_key: Vec<Option<HashKey>> = key_bindings
.iter()
.map(|b| b.as_ref().map(binding_hash_key).transpose())
.collect::<Result<Vec<_>, _>>()?;
let group_idx = *group_index.entry(hash_key).or_insert_with(|| {
groups.push(Group {
key_bindings: key_bindings.clone(),
accs: fresh_accs(items),
row_count: 0,
});
groups.len() - 1
});
let group = &mut groups[group_idx];
group.row_count += 1;
for (i, nodes) in item_agg_nodes.iter().enumerate() {
for (k, node) in nodes.iter().enumerate() {
match node {
ReturnExpr::CountStar => {
group.accs[i][k].fold(&Value::Literal(Literal::Bool(true)))?;
}
ReturnExpr::Call { name, args, .. } => {
let value = self.eval_return_expr(txn, &args[0], row, guard)?;
if is_percentile_name(name) {
let percentile =
self.eval_return_expr(txn, &args[1], row, guard)?;
if !matches!(value, Value::Null) {
group.accs[i][k].fold_percentile(&value, &percentile)?;
}
} else if !matches!(value, Value::Null) {
group.accs[i][k].fold(&value)?;
}
}
_ => unreachable!(
"collect_agg_nodes only ever collects CountStar/aggregate Call nodes"
),
}
}
}
}
let no_key_items = items.iter().all(|item| contains_aggregate(&item.expr));
if groups.is_empty() && no_key_items {
groups.push(Group {
key_bindings: vec![None; items.len()],
accs: fresh_accs(items),
row_count: 0,
});
}
let mut out = Vec::with_capacity(groups.len());
for mut group in groups {
let ctx = GroupFinishCtx {
items,
key_bindings: &group.key_bindings,
};
let mut row_out = Vec::with_capacity(items.len());
for (i, item) in items.iter().enumerate() {
let binding = match &group.key_bindings[i] {
Some(b) => b.clone(),
None => {
let mut accs = std::mem::take(&mut group.accs[i]).into_iter();
let mut subst = HashMap::new();
let rewritten = self
.rewrite_composed_item(txn, &item.expr, &ctx, &mut accs, &mut subst)?;
value_to_binding(eval_projected_expr(&rewritten, &subst)?)
}
};
row_out.push(binding);
}
out.push(row_out);
}
Ok(out)
}
fn rewrite_composed_item(
&self,
txn: Txn,
expr: &ReturnExpr,
ctx: &GroupFinishCtx<'_>,
accs: &mut std::vec::IntoIter<AggAcc>,
subst: &mut HashMap<String, Value>,
) -> Result<ReturnExpr, QueryError> {
if matches!(expr, ReturnExpr::CountStar)
|| matches!(expr, ReturnExpr::Call { name, .. } if is_aggregate_name(name))
{
let value = accs
.next()
.expect("accs is aligned with this same expr's collect_agg_nodes traversal order")
.finish();
let slot = format!("__slot{}", subst.len());
subst.insert(slot.clone(), value);
return Ok(ReturnExpr::Var(slot));
}
if matches!(expr, ReturnExpr::Var(_) | ReturnExpr::Prop(_)) {
let j = ctx
.items
.iter()
.enumerate()
.position(|(i, it)| item_matches_leaf(expr, i, it) && !contains_aggregate(&it.expr))
.expect(
"validate_return_items already checked this leaf matches a grouping-key item",
);
let binding = ctx.key_bindings[j]
.clone()
.expect("a non-aggregating item always has a key binding");
let value = self.binding_to_value(txn, &binding)?;
let slot = format!("__slot{}", subst.len());
subst.insert(slot.clone(), value);
return Ok(ReturnExpr::Var(slot));
}
Ok(match expr {
ReturnExpr::Lit(lit) => ReturnExpr::Lit(lit.clone()),
ReturnExpr::Call {
name,
args,
distinct,
} => ReturnExpr::Call {
name: name.clone(),
distinct: *distinct,
args: args
.iter()
.map(|a| self.rewrite_composed_item(txn, a, ctx, accs, subst))
.collect::<Result<_, _>>()?,
},
ReturnExpr::Case { test, whens, else_ } => ReturnExpr::Case {
test: test
.as_deref()
.map(|t| self.rewrite_composed_item(txn, t, ctx, accs, subst))
.transpose()?
.map(Box::new),
whens: whens
.iter()
.map(|(w, t)| {
Ok::<_, QueryError>((
self.rewrite_composed_item(txn, w, ctx, accs, subst)?,
self.rewrite_composed_item(txn, t, ctx, accs, subst)?,
))
})
.collect::<Result<_, _>>()?,
else_: else_
.as_deref()
.map(|e| self.rewrite_composed_item(txn, e, ctx, accs, subst))
.transpose()?
.map(Box::new),
},
ReturnExpr::Arith(l, op, r) => ReturnExpr::Arith(
Box::new(self.rewrite_composed_item(txn, l, ctx, accs, subst)?),
*op,
Box::new(self.rewrite_composed_item(txn, r, ctx, accs, subst)?),
),
ReturnExpr::Neg(e) => ReturnExpr::Neg(Box::new(
self.rewrite_composed_item(txn, e, ctx, accs, subst)?,
)),
ReturnExpr::ListLit(list_items) => ReturnExpr::ListLit(
list_items
.iter()
.map(|i| self.rewrite_composed_item(txn, i, ctx, accs, subst))
.collect::<Result<_, _>>()?,
),
ReturnExpr::Index(base, index) => ReturnExpr::Index(
Box::new(self.rewrite_composed_item(txn, base, ctx, accs, subst)?),
Box::new(self.rewrite_composed_item(txn, index, ctx, accs, subst)?),
),
ReturnExpr::PropOf(base, prop) => ReturnExpr::PropOf(
Box::new(self.rewrite_composed_item(txn, base, ctx, accs, subst)?),
prop.clone(),
),
ReturnExpr::Slice(base, start, end) => ReturnExpr::Slice(
Box::new(self.rewrite_composed_item(txn, base, ctx, accs, subst)?),
start
.as_deref()
.map(|s| self.rewrite_composed_item(txn, s, ctx, accs, subst))
.transpose()?
.map(Box::new),
end.as_deref()
.map(|e| self.rewrite_composed_item(txn, e, ctx, accs, subst))
.transpose()?
.map(Box::new),
),
ReturnExpr::ListComp {
var,
source,
where_clause,
project,
} => ReturnExpr::ListComp {
var: var.clone(),
source: Box::new(self.rewrite_composed_item(txn, source, ctx, accs, subst)?),
where_clause: where_clause.clone(),
project: project.clone(),
},
ReturnExpr::Quantifier {
kind,
var,
source,
where_clause,
} => ReturnExpr::Quantifier {
kind: *kind,
var: var.clone(),
source: Box::new(self.rewrite_composed_item(txn, source, ctx, accs, subst)?),
where_clause: where_clause.clone(),
},
ReturnExpr::MapLit(entries) => ReturnExpr::MapLit(
entries
.iter()
.map(|(k, v)| {
Ok::<_, QueryError>((
k.clone(),
self.rewrite_composed_item(txn, v, ctx, accs, subst)?,
))
})
.collect::<Result<_, _>>()?,
),
ReturnExpr::And(l, r) => ReturnExpr::And(
Box::new(self.rewrite_composed_item(txn, l, ctx, accs, subst)?),
Box::new(self.rewrite_composed_item(txn, r, ctx, accs, subst)?),
),
ReturnExpr::Or(l, r) => ReturnExpr::Or(
Box::new(self.rewrite_composed_item(txn, l, ctx, accs, subst)?),
Box::new(self.rewrite_composed_item(txn, r, ctx, accs, subst)?),
),
ReturnExpr::Xor(l, r) => ReturnExpr::Xor(
Box::new(self.rewrite_composed_item(txn, l, ctx, accs, subst)?),
Box::new(self.rewrite_composed_item(txn, r, ctx, accs, subst)?),
),
ReturnExpr::Not(e) => ReturnExpr::Not(Box::new(
self.rewrite_composed_item(txn, e, ctx, accs, subst)?,
)),
ReturnExpr::Compare(l, op, r) => ReturnExpr::Compare(
Box::new(self.rewrite_composed_item(txn, l, ctx, accs, subst)?),
*op,
Box::new(self.rewrite_composed_item(txn, r, ctx, accs, subst)?),
),
ReturnExpr::IsNull(e) => ReturnExpr::IsNull(Box::new(
self.rewrite_composed_item(txn, e, ctx, accs, subst)?,
)),
ReturnExpr::In(needle, haystack) => ReturnExpr::In(
Box::new(self.rewrite_composed_item(txn, needle, ctx, accs, subst)?),
Box::new(self.rewrite_composed_item(txn, haystack, ctx, accs, subst)?),
),
ReturnExpr::HasLabel(v, l) => ReturnExpr::HasLabel(v.clone(), l.clone()),
ReturnExpr::PatternPredicate(p) => ReturnExpr::PatternPredicate(p.clone()),
ReturnExpr::PatternComprehension { .. } => expr.clone(),
ReturnExpr::ExistsPattern { .. } => expr.clone(),
ReturnExpr::ExistsSubquery(_) => expr.clone(),
ReturnExpr::Var(_) | ReturnExpr::Prop(_) | ReturnExpr::CountStar => {
unreachable!("handled above, before this match")
}
})
}
fn eval_with_expr(
&self,
txn: Txn,
expr: &WithExpr,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<Option<bool>, QueryError> {
Ok(match expr {
WithExpr::And(l, r) => and3(
self.eval_with_expr(txn, l, row, guard)?,
self.eval_with_expr(txn, r, row, guard)?,
),
WithExpr::Or(l, r) => or3(
self.eval_with_expr(txn, l, row, guard)?,
self.eval_with_expr(txn, r, row, guard)?,
),
WithExpr::Not(e) => self.eval_with_expr(txn, e, row, guard)?.map(|b| !b),
WithExpr::Compare(lhs, op, rhs) => {
let lv = self.eval_return_expr(txn, lhs, row, guard)?;
let rv = self.eval_return_expr(txn, rhs, row, guard)?;
compare_values(&lv, *op, &rv)
}
WithExpr::IsNull(e) => Some(matches!(
self.eval_return_expr(txn, e, row, guard)?,
Value::Null
)),
WithExpr::Bare(ReturnExpr::PatternPredicate(pattern)) => {
Some(self.eval_pattern_predicate_exists(txn, pattern, row, guard)?)
}
WithExpr::Bare(e) => self.eval_return_expr_bool3(txn, e, row, guard)?,
})
}
fn eval_pattern_predicate_exists(
&self,
txn: Txn,
pattern: &Pattern,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<bool, QueryError> {
let carried_vars: HashSet<String> = row.keys().cloned().collect();
let plan = apply_index_seeks(build_match_plan(pattern, &None, &carried_vars)?, txn)?;
let found =
self.eval_plan_with_limit(txn, &plan, std::slice::from_ref(row), guard, Some(1))?;
Ok(!found.is_empty())
}
fn eval_exists_subquery(
&self,
txn: Txn,
stmt: &Statement,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<bool, QueryError> {
let Statement::Match {
clauses,
tail,
order_by,
skip,
limit,
} = stmt
else {
unreachable!(
"semantic::validate_statement only allows Statement::Match inside exists {{}}"
)
};
let skip = self.resolve_skip_limit(txn, skip.as_deref(), "SKIP", guard)?;
let limit = self.resolve_skip_limit(txn, limit.as_deref(), "LIMIT", guard)?;
let result = self.execute_match_seeded(
txn,
clauses,
tail,
ResultModifiers {
order_by,
skip,
limit,
},
Some(row),
guard,
)?;
Ok(!result.rows.is_empty())
}
fn eval_optional_part(
&self,
txn: Txn,
plan: &LogicalPlan,
outer_rows: &[BindingRow],
new_vars: &HashSet<String>,
guard: &ExecutionGuard<'_>,
) -> Result<Vec<BindingRow>, QueryError> {
let tagged: Vec<BindingRow> = outer_rows
.iter()
.enumerate()
.map(|(i, row)| {
let mut r = row.clone();
r.insert(
OPTIONAL_SEED_IDX_KEY.to_string(),
Binding::Value(PropertyValue::Int(i as i64)),
);
r
})
.collect();
guard.check_intermediate_rows(tagged.len())?;
let results = self.eval_plan(txn, plan, &tagged, guard)?;
let mut by_idx: HashMap<i64, Vec<BindingRow>> = HashMap::new();
for mut row in results {
let idx = match row.remove(OPTIONAL_SEED_IDX_KEY) {
Some(Binding::Value(PropertyValue::Int(i))) => i,
other => unreachable!(
"__seed_idx tagged internally as Binding::Value(Int), got {other:?}"
),
};
by_idx.entry(idx).or_default().push(row);
}
let mut out = Vec::with_capacity(outer_rows.len());
for (i, outer_row) in outer_rows.iter().enumerate() {
match by_idx.remove(&(i as i64)) {
Some(matches) => out.extend(matches),
None => {
let mut padded = outer_row.clone();
for var in new_vars {
padded.insert(var.clone(), Binding::Value(PropertyValue::Null));
}
out.push(padded);
}
}
guard.check_intermediate_rows(out.len())?;
}
Ok(out)
}
fn eval_plan(
&self,
txn: Txn,
plan: &LogicalPlan,
seed: &[BindingRow],
guard: &ExecutionGuard<'_>,
) -> Result<Vec<BindingRow>, QueryError> {
self.eval_plan_with_limit(txn, plan, seed, guard, None)
}
fn eval_plan_with_limit(
&self,
txn: Txn,
plan: &LogicalPlan,
seed: &[BindingRow],
guard: &ExecutionGuard<'_>,
limit: Option<usize>,
) -> Result<Vec<BindingRow>, QueryError> {
let stream = self.stream_plan(txn, plan, seed, guard, limit);
match limit {
Some(limit) => stream.take(limit).collect(),
None => stream.collect(),
}
}
fn stream_plan<'s>(
&'s self,
txn: Txn<'s>,
plan: &'s LogicalPlan,
seed: &'s [BindingRow],
guard: &'s ExecutionGuard<'_>,
scan_limit: Option<usize>,
) -> RowStream<'s> {
match plan {
LogicalPlan::Seed { var } => {
debug_assert!(
seed.first().is_none_or(|row| row.contains_key(var)),
"Seed{{var: {var:?}}} planned for a var not present in the carried-forward rows"
);
Self::count_stream(Box::new(seed.iter().cloned().map(Ok)), guard)
}
LogicalPlan::AllNodesScan { var } => {
self.stream_scan(txn, var, None, seed, guard, scan_limit)
}
LogicalPlan::NodeByLabelScan { var, label } => {
self.stream_scan(txn, var, Some(label), seed, guard, scan_limit)
}
LogicalPlan::IndexSeek {
var,
label,
prop,
value,
} => self.stream_index_seek(
txn,
IndexSeekSpec {
var,
label,
prop,
value,
},
seed,
guard,
scan_limit,
),
LogicalPlan::Expand {
input,
from_var,
to_var,
rel_var,
rel_labels,
direction,
} => {
let input = self.stream_plan(txn, input, seed, guard, None);
let stream = input.flat_map(move |res| -> RowStream<'s> {
let row = match res {
Ok(row) => row,
Err(error) => return Box::new(std::iter::once(Err(error))),
};
let from_id = match row.get(from_var) {
Some(Binding::Node(id)) => *id,
Some(Binding::Value(PropertyValue::Null)) => {
return Box::new(std::iter::empty())
}
_ => {
return Box::new(std::iter::once(Err(QueryError::UnboundVariable(
from_var.clone(),
))))
}
};
match neighbors_for_direction(txn, from_id, *direction, rel_labels) {
Ok(entries) => Box::new(entries.into_iter().map(move |entry| {
guard.relationship_expansion()?;
let mut new_row = row.clone();
new_row.insert(to_var.clone(), Binding::Node(entry.other));
if let Some(rel_var) = rel_var {
new_row.insert(rel_var.clone(), Binding::Edge(entry.edge_id));
}
Ok(new_row)
})),
Err(error) => Box::new(std::iter::once(Err(error))),
}
});
Self::count_stream(Box::new(stream), guard)
}
LogicalPlan::VarExpand {
input,
from_var,
to_var,
rel_labels,
direction,
min_hops,
max_hops,
exclude_edge_vars,
exclude_edge_sets,
exclude_edge_var,
path_segment_var,
rel_list_var,
rel_props,
} => {
let input = self.stream_plan(txn, input, seed, guard, None);
let stream = input.flat_map(move |res| {
let rows = res.and_then(|row| {
self.expand_variable_row(
txn,
row,
VarExpandSpec {
from_var,
to_var,
rel_labels,
direction: *direction,
min_hops: *min_hops,
max_hops: *max_hops,
exclude_edge_vars,
exclude_edge_sets,
exclude_edge_var,
path_segment_var: path_segment_var.as_deref(),
rel_list_var: rel_list_var.as_deref(),
rel_props,
},
guard,
)
});
match rows {
Ok(rows) => Box::new(rows.into_iter().map(Ok)) as RowStream<'s>,
Err(error) => Box::new(std::iter::once(Err(error))),
}
});
Self::count_stream(Box::new(stream), guard)
}
LogicalPlan::MatchRelList {
input,
from_var,
to_var,
rel_list_var,
rel_labels,
direction,
min_hops,
max_hops,
} => {
let input = self.stream_plan(txn, input, seed, guard, None);
let stream = input.filter_map(move |res| {
let row = match res {
Ok(row) => row,
Err(error) => return Some(Err(error)),
};
self.match_bound_rel_list_row(
row,
MatchRelListSpec {
from_var,
to_var,
rel_list_var,
rel_labels,
direction: *direction,
min_hops: *min_hops,
max_hops: *max_hops,
},
)
.transpose()
});
Self::count_stream(Box::new(stream), guard)
}
LogicalPlan::Filter { input, predicate } => {
let input = self.stream_plan(txn, input, seed, guard, None);
let stream = input.filter_map(move |res| {
let row = match res {
Ok(row) => row,
Err(error) => return Some(Err(error)),
};
if let Err(error) = guard.checkpoint() {
return Some(Err(error));
}
match self.eval_expr(txn, predicate, &row, guard) {
Ok(Some(true)) => Some(Ok(row)),
Ok(_) => None,
Err(error) => Some(Err(error)),
}
});
Self::count_stream(Box::new(stream), guard)
}
}
}
fn count_stream<'s>(mut stream: RowStream<'s>, guard: &'s ExecutionGuard<'_>) -> RowStream<'s> {
let mut produced = 0usize;
let mut done = false;
Box::new(std::iter::from_fn(move || {
if done {
return None;
}
let item = stream.next()?;
if item.is_ok() {
produced = match produced.checked_add(1) {
Some(produced) => produced,
None => {
done = true;
return Some(Err(QueryError::ResourceLimit(
"stream row counter overflow".into(),
)));
}
};
if let Err(error) = guard.check_intermediate_rows(produced) {
done = true;
return Some(Err(error));
}
} else {
done = true;
}
Some(item)
}))
}
fn try_fast_expand_expand_count(
&self,
txn: Txn,
plan: &LogicalPlan,
with: &Option<WithClause>,
current_rows: &[BindingRow],
tail_hint: Option<(&ReturnExpr, SortDir, usize)>,
guard: &ExecutionGuard<'_>,
) -> Result<Option<FastCountResult>, QueryError> {
if current_rows.len() != 1 || !current_rows[0].is_empty() {
return Ok(None);
}
let Some(with) = with else { return Ok(None) };
if with.star || with.distinct || with.where_clause.is_some() || with.items.len() < 2 {
return Ok(None);
}
fn peel<'p>(mut plan: &'p LogicalPlan, preds: &mut Vec<&'p Expr>) -> &'p LogicalPlan {
while let LogicalPlan::Filter { input, predicate } = plan {
push_conjunct_refs(predicate, preds);
plan = input;
}
plan
}
fn push_conjunct_refs<'p>(expr: &'p Expr, out: &mut Vec<&'p Expr>) {
if let Expr::And(l, r) = expr {
push_conjunct_refs(l, out);
push_conjunct_refs(r, out);
} else {
out.push(expr);
}
}
struct Stage<'p> {
from: &'p str,
to: &'p str,
rel_var: Option<&'p str>,
label: Option<&'p str>,
dir: Direction,
preds: Vec<&'p Expr>,
}
let mut stages: Vec<Stage<'_>> = Vec::new();
let mut cursor = plan;
let leaf = loop {
let mut preds = Vec::new();
match peel(cursor, &mut preds) {
LogicalPlan::Expand {
input,
from_var,
to_var,
rel_var,
rel_labels,
direction,
} if stages.len() < 2 => {
let (Some(dir), Some(label)) =
(fast_direction(*direction), fast_label(rel_labels))
else {
return Ok(None);
};
stages.push(Stage {
from: from_var,
to: to_var,
rel_var: rel_var.as_deref(),
label,
dir,
preds,
});
cursor = input;
}
_ => {
if stages.is_empty() || plan_contains_expansion(cursor) {
return Ok(None);
}
break cursor;
}
}
};
stages.reverse(); if stages.len() == 2 && stages[1].from != stages[0].to {
return Ok(None);
}
let final_to = stages.last().expect("at least one stage").to;
let origin = stages[0].from;
let mid_var = (stages.len() == 2).then(|| stages[0].to);
enum OutCol<'p> {
Group,
Count,
Collect(&'p str), }
let mut cols: Vec<OutCol<'_>> = Vec::with_capacity(with.items.len());
let mut group_seen = false;
let mut group_by_origin = false;
let mut count_seen = false;
for item in &with.items {
match &item.expr {
ReturnExpr::Var(v) if v == final_to && !group_seen => {
group_seen = true;
cols.push(OutCol::Group);
}
ReturnExpr::Var(v) if v == origin && !group_seen => {
group_seen = true;
group_by_origin = true;
cols.push(OutCol::Group);
}
ReturnExpr::CountStar if !count_seen => {
count_seen = true;
cols.push(OutCol::Count);
}
ReturnExpr::Call {
name,
args,
distinct: false,
} if name.eq_ignore_ascii_case("collect") => {
let [ReturnExpr::Prop(pa)] = args.as_slice() else {
return Ok(None);
};
let Some(mid) = mid_var else { return Ok(None) };
if pa.var != mid {
return Ok(None);
}
cols.push(OutCol::Collect(&pa.prop));
}
_ => return Ok(None),
}
}
if !group_seen {
return Ok(None);
}
let names: Vec<String> = with
.items
.iter()
.enumerate()
.map(with_item_output_name)
.collect();
let count_name = cols
.iter()
.position(|c| matches!(c, OutCol::Count))
.map(|i| names[i].as_str());
let mut pre_keep: Option<usize> = None;
let count_sort: Option<SortDir> = match &with.order_by {
None => {
match tail_hint {
Some((key, dir, keep)) if with.skip.is_none() && with.limit.is_none() => {
let matches_count = match key {
ReturnExpr::Var(v) => count_name == Some(v.as_str()),
ReturnExpr::CountStar => count_seen,
_ => false,
};
if matches_count {
pre_keep = Some(keep);
Some(dir)
} else {
None
}
}
_ => None,
}
}
Some(keys) => {
let [(key, dir)] = keys.as_slice() else {
return Ok(None);
};
let matches_count = match key {
ReturnExpr::Var(v) => count_name == Some(v.as_str()),
ReturnExpr::CountStar => count_seen,
_ => false,
};
if !matches_count {
return Ok(None);
}
Some(*dir)
}
};
let mut stage_label_filters: Vec<Vec<&str>> = vec![Vec::new(); stages.len()];
let mut isomorphism = false;
for (i, stage) in stages.iter().enumerate() {
for pred in &stage.preds {
match pred {
Expr::HasLabel(v, l) if v == stage.to => stage_label_filters[i].push(l),
Expr::Not(inner) if i == 1 => {
match (&**inner, stages[0].rel_var, stage.rel_var) {
(Expr::VarEq(x, y), Some(r1), Some(r2))
if (x == r1 && y == r2) || (x == r2 && y == r1) =>
{
isomorphism = true;
}
_ => return Ok(None),
}
}
_ => return Ok(None),
}
}
}
let skip = self.resolve_skip_limit(txn, with.skip.as_ref(), "SKIP", guard)?;
let limit = self.resolve_skip_limit(txn, with.limit.as_ref(), "LIMIT", guard)?;
let label_set = |label: &str| -> Result<std::collections::HashSet<u64>, QueryError> {
Ok(
GraphStore::all_node_ids_limited_in_txn(txn, Some(label), usize::MAX)?
.into_iter()
.map(|n| n.0)
.collect(),
)
};
let stage_sets: Vec<Vec<std::collections::HashSet<u64>>> = stage_label_filters
.iter()
.map(|labels| labels.iter().map(|l| label_set(l)).collect())
.collect::<Result<_, _>>()?;
let collect_prop_ids: Vec<Option<u32>> = cols
.iter()
.map(|c| match c {
OutCol::Collect(prop) => self.prop_id_for(txn, prop),
_ => Ok(None),
})
.collect::<Result<_, _>>()?;
let mut seeds = Vec::new();
let mut leaf_preds = Vec::new();
let leaf_base = peel(leaf, &mut leaf_preds);
let leaf_candidates: Option<Vec<NodeId>> = match leaf_base {
LogicalPlan::AllNodesScan { var } if var == stages[0].from => Some(
GraphStore::all_node_ids_limited_in_txn(txn, None, usize::MAX)?,
),
LogicalPlan::NodeByLabelScan { var, label } if var == stages[0].from => Some(
GraphStore::all_node_ids_limited_in_txn(txn, Some(label), usize::MAX)?,
),
LogicalPlan::IndexSeek {
var,
label,
prop,
value: crate::ir::IndexSeekValue::Fixed(value),
} if var == stages[0].from => {
Some(GraphStore::lookup_by_index_in_txn(txn, label, prop, value)?)
}
_ => None,
};
match leaf_candidates {
Some(candidates) => {
let simple: Option<Vec<(&PropAccess, CompareOp, &Literal)>> = leaf_preds
.iter()
.map(|pred| match pred {
Expr::Compare(pa, op, lit) if pa.var == stages[0].from => {
Some((pa, *op, lit))
}
_ => None,
})
.collect();
if let Some(simple) = simple {
let pred_ids: Vec<Option<u32>> = simple
.iter()
.map(|(pa, _, _)| self.prop_id_for(txn, &pa.prop))
.collect::<Result<_, _>>()?;
let mut read_prop = GraphStore::node_prop_reader(txn)?;
'cand: for id in candidates {
guard.checkpoint()?;
for ((_, op, lit), prop_id) in simple.iter().zip(&pred_ids) {
let value = match prop_id {
Some(pid) => read_prop(id, *pid)?.flatten(),
None => None,
};
if compare(&value, *op, lit) != Some(true) {
continue 'cand;
}
}
seeds.push(id);
}
} else {
let mut probe = BindingRow::new();
for id in candidates {
guard.checkpoint()?;
probe.insert(stages[0].from.to_string(), Binding::Node(id));
let mut pass = true;
for pred in &leaf_preds {
if self.eval_expr(txn, pred, &probe, guard)? != Some(true) {
pass = false;
break;
}
}
if pass {
seeds.push(id);
}
}
}
}
None => {
for row in self.eval_plan(txn, leaf, current_rows, guard)? {
match row.get(stages[0].from) {
Some(Binding::Node(id)) => seeds.push(*id),
_ => return Ok(None),
}
}
}
}
struct Group {
count: i64,
collects: Vec<Vec<Value>>,
}
let n_collects = cols
.iter()
.filter(|c| matches!(c, OutCol::Collect(_)))
.count();
let mut order: Vec<u64> = Vec::new();
let mut groups: HashMap<u64, Group> = HashMap::new();
let mut mid_prop_memo: HashMap<(u64, u32), Option<Value>> = HashMap::new();
let mut mid_values: Vec<Option<Value>> = vec![None; n_collects];
let one_hop = stages.len() == 1;
for &s in &seeds {
guard.checkpoint()?;
for e1 in GraphStore::neighbors_in_txn(txn, s, stages[0].dir, stages[0].label)? {
guard.relationship_expansion()?;
if !stage_sets[0].iter().all(|set| set.contains(&e1.other.0)) {
continue;
}
if one_hop {
let key = if group_by_origin { s.0 } else { e1.other.0 };
let group = groups.entry(key).or_insert_with(|| {
order.push(key);
Group {
count: 0,
collects: vec![Vec::new(); n_collects],
}
});
group.count += 1;
continue;
}
let mut ci = 0usize;
for (col, prop_id) in cols.iter().zip(&collect_prop_ids) {
if let OutCol::Collect(_) = col {
mid_values[ci] = match prop_id {
Some(pid) => mid_prop_memo
.entry((e1.other.0, *pid))
.or_insert_with(|| {
GraphStore::get_node_prop_in_txn(txn, e1.other, *pid)
.ok()
.flatten()
.flatten()
.map(property_value_to_value)
})
.clone(),
None => None, };
ci += 1;
}
}
guard.checkpoint()?;
for e2 in
GraphStore::neighbors_in_txn(txn, e1.other, stages[1].dir, stages[1].label)?
{
guard.relationship_expansion()?;
if isomorphism && e2.edge_id == e1.edge_id {
continue;
}
if !stage_sets[1].iter().all(|set| set.contains(&e2.other.0)) {
continue;
}
let key = if group_by_origin { s.0 } else { e2.other.0 };
let group = groups.entry(key).or_insert_with(|| {
order.push(key);
Group {
count: 0,
collects: vec![Vec::new(); n_collects],
}
});
group.count += 1;
for (ci, value) in mid_values.iter().enumerate() {
if let Some(v) = value {
group.collects[ci].push(v.clone());
}
}
}
}
}
let mut grouped: Vec<(u64, Group)> = order
.into_iter()
.map(|id| {
let group = groups.remove(&id).expect("group recorded in order");
(id, group)
})
.collect();
match count_sort {
Some(SortDir::Asc) => grouped.sort_by_key(|(_, g)| g.count),
Some(SortDir::Desc) => grouped.sort_by_key(|(_, g)| std::cmp::Reverse(g.count)),
None => {}
}
if let Some(keep) = pre_keep {
grouped.truncate(keep);
}
let skip_n = skip.unwrap_or(0).max(0) as usize;
if skip_n > 0 {
grouped.drain(0..skip_n.min(grouped.len()));
}
if let Some(limit) = limit {
grouped.truncate(limit.max(0) as usize);
}
let rows: Vec<BindingRow> = grouped
.into_iter()
.map(|(id, group)| {
let mut row = BindingRow::new();
let mut collects = group.collects.into_iter();
for (col, name) in cols.iter().zip(&names) {
let binding = match col {
OutCol::Group => Binding::Node(NodeId(id)),
OutCol::Count => Binding::Value(PropertyValue::Int(group.count)),
OutCol::Collect(_) => {
Binding::List(collects.next().expect("one list per collect column"))
}
};
row.insert(name.clone(), binding);
}
row
})
.collect();
if std::env::var("MARSDB_FAST_DEBUG").is_ok() {
eprintln!(
"[fast-path FIRED] stages={} groups={}",
stages.len(),
rows.len()
);
}
Ok(Some((rows, names.into_iter().collect())))
}
fn stream_scan<'s>(
&'s self,
txn: Txn<'s>,
var: &'s str,
label: Option<&'s str>,
seed: &'s [BindingRow],
guard: &'s ExecutionGuard<'_>,
row_limit: Option<usize>,
) -> RowStream<'s> {
let mut initialized = false;
let mut node_ids = Vec::new();
let mut seed_index = 0usize;
let mut node_index = 0usize;
let mut done = false;
let stream = std::iter::from_fn(move || {
if done || seed.is_empty() {
return None;
}
if !initialized {
initialized = true;
let budget_node_limit = guard.options.max_intermediate_rows.map(|max_rows| {
max_rows
.checked_div(seed.len())
.unwrap_or(0)
.saturating_add(1)
});
let storage_limit = match (row_limit, budget_node_limit) {
(Some(a), Some(b)) => Some(a.min(b)),
(Some(a), None) => Some(a),
(None, Some(b)) => Some(b),
(None, None) => None,
};
let storage_limit = storage_limit.unwrap_or(usize::MAX);
match GraphStore::all_node_ids_limited_in_txn(txn, label, storage_limit) {
Ok(ids) => node_ids = ids,
Err(error) => {
done = true;
return Some(Err(error.into()));
}
}
}
if node_ids.is_empty() || seed_index >= seed.len() {
return None;
}
if let Err(error) = guard.checkpoint() {
done = true;
return Some(Err(error));
}
let mut row = seed[seed_index].clone();
row.insert(var.to_string(), Binding::Node(node_ids[node_index]));
node_index += 1;
if node_index == node_ids.len() {
node_index = 0;
seed_index += 1;
}
Some(Ok(row))
});
Self::count_stream(Box::new(stream), guard)
}
fn stream_index_seek<'s>(
&'s self,
txn: Txn<'s>,
spec: IndexSeekSpec<'s>,
seed: &'s [BindingRow],
guard: &'s ExecutionGuard<'_>,
row_limit: Option<usize>,
) -> RowStream<'s> {
let budget_node_limit = guard.options.max_intermediate_rows.map(|max_rows| {
max_rows
.checked_div(seed.len().max(1))
.unwrap_or(0)
.saturating_add(1)
});
let storage_limit = match (row_limit, budget_node_limit) {
(Some(a), Some(b)) => Some(a.min(b)),
(Some(a), None) => Some(a),
(None, Some(b)) => Some(b),
(None, None) => None,
};
let lookup = move |value: &PropertyValue| -> Result<Vec<NodeId>, QueryError> {
match storage_limit {
Some(limit) => GraphStore::lookup_by_index_limited_in_txn(
txn, spec.label, spec.prop, value, limit,
)
.map_err(Into::into),
None => GraphStore::lookup_by_index_in_txn(txn, spec.label, spec.prop, value)
.map_err(Into::into),
}
};
match spec.value {
IndexSeekValue::Fixed(value) => {
let mut node_ids: Option<Vec<NodeId>> = None;
let mut seed_index = 0usize;
let mut node_index = 0usize;
let mut done = false;
let stream = std::iter::from_fn(move || {
if done || seed.is_empty() {
return None;
}
let ids = match &node_ids {
Some(ids) => ids,
None => match lookup(value) {
Ok(ids) => node_ids.insert(ids),
Err(error) => {
done = true;
return Some(Err(error));
}
},
};
if ids.is_empty() || seed_index >= seed.len() {
return None;
}
if let Err(error) = guard.checkpoint() {
done = true;
return Some(Err(error));
}
let mut row = seed[seed_index].clone();
row.insert(spec.var.to_string(), Binding::Node(ids[node_index]));
node_index += 1;
if node_index == ids.len() {
node_index = 0;
seed_index += 1;
}
Some(Ok(row))
});
Self::count_stream(Box::new(stream), guard)
}
IndexSeekValue::RowExpr(expr) => {
let mut node_ids: Vec<NodeId> = Vec::new();
let mut seed_index = 0usize;
let mut node_index = 0usize;
let mut done = false;
let stream = std::iter::from_fn(move || loop {
if done || seed_index >= seed.len() {
return None;
}
if node_index == 0 {
let evaluated =
match self.eval_return_expr(txn, expr, &seed[seed_index], guard) {
Ok(v) => v,
Err(error) => {
done = true;
return Some(Err(error));
}
};
let value = value_to_property_value(&evaluated);
if matches!(value, PropertyValue::Null) {
seed_index += 1;
continue;
}
node_ids = match lookup(&value) {
Ok(ids) => ids,
Err(error) => {
done = true;
return Some(Err(error));
}
};
if node_ids.is_empty() {
seed_index += 1;
continue;
}
}
if let Err(error) = guard.checkpoint() {
done = true;
return Some(Err(error));
}
let mut row = seed[seed_index].clone();
row.insert(spec.var.to_string(), Binding::Node(node_ids[node_index]));
node_index += 1;
if node_index == node_ids.len() {
node_index = 0;
seed_index += 1;
}
return Some(Ok(row));
});
Self::count_stream(Box::new(stream), guard)
}
}
}
fn expand_variable_row(
&self,
txn: Txn,
row: BindingRow,
spec: VarExpandSpec<'_>,
guard: &ExecutionGuard<'_>,
) -> Result<Vec<BindingRow>, QueryError> {
let start_id = match row.get(spec.from_var) {
Some(Binding::Node(id)) => *id,
Some(Binding::Value(PropertyValue::Null)) => return Ok(Vec::new()),
_ => return Err(QueryError::UnboundVariable(spec.from_var.to_string())),
};
let mut out = Vec::new();
if spec.min_hops == 0 {
let mut new_row = row.clone();
new_row.insert(spec.to_var.to_string(), Binding::Node(start_id));
if let Some(path_segment_var) = spec.path_segment_var {
new_row.insert(path_segment_var.to_string(), Binding::Path(Vec::new()));
}
if let Some(rel_list_var) = spec.rel_list_var {
new_row.insert(rel_list_var.to_string(), Binding::List(Vec::new()));
}
new_row.insert(spec.exclude_edge_var.to_string(), Binding::Path(Vec::new()));
out.push(new_row);
}
let rel_props = spec
.rel_props
.iter()
.map(|(key, expr)| {
let value = self.eval_return_expr(txn, expr, &row, guard)?;
Ok::<_, QueryError>((key.as_str(), value_to_property_value(&value)))
})
.collect::<Result<Vec<_>, _>>()?;
let unbounded = spec.max_hops.is_none();
let effective_max = spec.max_hops.unwrap_or(VAR_EXPAND_DEPTH_CAP);
let seed_used_edges: HashSet<EdgeId> = spec
.exclude_edge_vars
.iter()
.filter_map(|v| match row.get(v) {
Some(Binding::Edge(id)) => Some(*id),
_ => None,
})
.chain(spec.exclude_edge_sets.iter().flat_map(|v| {
match row.get(v) {
Some(Binding::Path(segment)) => segment
.iter()
.filter_map(|p| match p {
PathBinding::Edge(id) => Some(*id),
PathBinding::Node(_) => None,
})
.collect::<Vec<_>>(),
_ => Vec::new(),
}
}))
.collect();
let mut frontier = vec![(start_id, seed_used_edges, Vec::<PathBinding>::new())];
let mut depth = 0u32;
while depth < effective_max && !frontier.is_empty() {
depth += 1;
let mut next_frontier = Vec::new();
for (node, used_edges, segment) in frontier {
for entry in neighbors_for_direction(txn, node, spec.direction, spec.rel_labels)? {
guard.relationship_expansion()?;
if used_edges.contains(&entry.edge_id) {
continue;
}
if !rel_props.is_empty() {
let edge = deleted_entity_access(GraphStore::get_edge_in_txn(
txn,
entry.edge_id,
)?)?;
let matches = rel_props
.iter()
.all(|(key, expected)| edge.props.get(*key) == Some(expected));
if !matches {
continue;
}
}
let mut next_used_edges = used_edges.clone();
next_used_edges.insert(entry.edge_id);
let mut next_segment = segment.clone();
next_segment.push(PathBinding::Edge(entry.edge_id));
next_segment.push(PathBinding::Node(entry.other));
next_frontier.push((entry.other, next_used_edges, next_segment.clone()));
guard.check_intermediate_rows(next_frontier.len())?;
if depth >= spec.min_hops {
let mut new_row = row.clone();
new_row.insert(spec.to_var.to_string(), Binding::Node(entry.other));
if let Some(path_segment_var) = spec.path_segment_var {
new_row.insert(
path_segment_var.to_string(),
Binding::Path(next_segment.clone()),
);
}
if let Some(rel_list_var) = spec.rel_list_var {
let edges = segment_edges_to_list(txn, &next_segment)?;
new_row.insert(rel_list_var.to_string(), edges);
}
new_row.insert(
spec.exclude_edge_var.to_string(),
Binding::Path(next_segment.clone()),
);
out.push(new_row);
guard.check_intermediate_rows(out.len())?;
}
}
}
frontier = next_frontier;
if depth == effective_max && unbounded && !frontier.is_empty() {
return Err(QueryError::ResourceLimit(format!(
"variable-length traversal exceeded the safety depth cap ({VAR_EXPAND_DEPTH_CAP} \
hops) — likely a cyclic graph or unexpectedly large fanout; narrow the pattern or \
add an explicit upper bound (e.g. *0..10)"
)));
}
}
Ok(out)
}
fn match_bound_rel_list_row(
&self,
row: BindingRow,
spec: MatchRelListSpec<'_>,
) -> Result<Option<BindingRow>, QueryError> {
let start_id = match row.get(spec.from_var) {
Some(Binding::Node(id)) => *id,
Some(Binding::Value(PropertyValue::Null)) => return Ok(None),
_ => return Err(QueryError::UnboundVariable(spec.from_var.to_string())),
};
let edges: Vec<&Edge> = match row.get(spec.rel_list_var) {
Some(Binding::List(items)) => items
.iter()
.map(|v| match v {
Value::Edge(e) => Ok(e),
other => Err(QueryError::Type(format!(
"'{}' must be a list of relationships, found {other:?} in it",
spec.rel_list_var
))),
})
.collect::<Result<_, _>>()?,
Some(Binding::Value(PropertyValue::Null)) => return Ok(None),
_ => return Err(QueryError::UnboundVariable(spec.rel_list_var.to_string())),
};
let hops = edges.len() as u32;
if hops < spec.min_hops || spec.max_hops.is_some_and(|max| hops > max) {
return Ok(None);
}
if !spec.rel_labels.is_empty() && edges.iter().any(|e| !spec.rel_labels.contains(&e.label))
{
return Ok(None);
}
let mut current = start_id;
for edge in &edges {
let next = match spec.direction {
ExpandDirection::Out if edge.src == current => edge.dst,
ExpandDirection::In if edge.dst == current => edge.src,
ExpandDirection::Either if edge.src == current => edge.dst,
ExpandDirection::Either if edge.dst == current => edge.src,
_ => return Ok(None),
};
current = next;
}
let mut new_row = row.clone();
new_row.insert(spec.to_var.to_string(), Binding::Node(current));
Ok(Some(new_row))
}
fn eval_expr(
&self,
txn: Txn,
expr: &Expr,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<Option<bool>, QueryError> {
Ok(match expr {
Expr::And(l, r) => and3(
self.eval_expr(txn, l, row, guard)?,
self.eval_expr(txn, r, row, guard)?,
),
Expr::Or(l, r) => or3(
self.eval_expr(txn, l, row, guard)?,
self.eval_expr(txn, r, row, guard)?,
),
Expr::Not(e) => self.eval_expr(txn, e, row, guard)?.map(|b| !b),
Expr::Compare(pa, op, lit) => {
let prop_value = self.lookup_prop(txn, pa, row)?;
compare(&prop_value, *op, lit)
}
Expr::PropCompare(left, op, right) => {
let a = self.lookup_prop(txn, left, row)?;
let b = self.lookup_prop(txn, right, row)?;
compare_property_pair_opt(&a, *op, &b)
}
Expr::IsNull(pa) => Some(matches!(
self.lookup_prop(txn, pa, row)?,
None | Some(PropertyValue::Null)
)),
Expr::HasLabel(var, label) => {
let binding = row
.get(var)
.ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
let Binding::Node(id) = binding else {
return Err(QueryError::UnboundVariable(var.clone()));
};
let node = self.get_node_cached(txn, *id)?;
Some(node.is_some_and(|n| n.labels.iter().any(|l| l == label)))
}
Expr::VarEq(a, b) => {
let ba = row
.get(a)
.ok_or_else(|| QueryError::UnboundVariable(a.clone()))?;
let bb = row
.get(b)
.ok_or_else(|| QueryError::UnboundVariable(b.clone()))?;
Some(match (ba, bb) {
(Binding::Node(x), Binding::Node(y)) => x == y,
(Binding::Edge(x), Binding::Edge(y)) => x == y,
_ => false,
})
}
Expr::GeneralCompare(lhs, op, rhs) => {
let lv = self.eval_return_expr(txn, lhs, row, guard)?;
let rv = self.eval_return_expr(txn, rhs, row, guard)?;
compare_values(&lv, *op, &rv)
}
Expr::GeneralIsNull(e) => Some(matches!(
self.eval_return_expr(txn, e, row, guard)?,
Value::Null
)),
Expr::GeneralBare(e) => self.eval_return_expr_bool3(txn, e, row, guard)?,
Expr::Pattern(pattern) => {
Some(self.eval_pattern_predicate_exists(txn, pattern, row, guard)?)
}
Expr::Exists {
pattern,
where_clause,
} => {
let carried_vars: HashSet<String> = row.keys().cloned().collect();
let wc: Option<Expr> = where_clause.as_deref().cloned();
let plan = apply_index_seeks(build_match_plan(pattern, &wc, &carried_vars)?, txn)?;
let found = self.eval_plan_with_limit(
txn,
&plan,
std::slice::from_ref(row),
guard,
Some(1),
)?;
Some(!found.is_empty())
}
Expr::ExistsSubquery(stmt) => Some(self.eval_exists_subquery(txn, stmt, row, guard)?),
Expr::EdgeNotInSet {
edge_var,
edge_set_var,
} => {
let Some(Binding::Edge(edge_id)) = row.get(edge_var) else {
return Err(QueryError::UnboundVariable(edge_var.clone()));
};
let Some(Binding::Path(segment)) = row.get(edge_set_var) else {
return Err(QueryError::UnboundVariable(edge_set_var.clone()));
};
Some(
!segment
.iter()
.any(|elem| matches!(elem, PathBinding::Edge(id) if id == edge_id)),
)
}
})
}
fn prop_id_for(&self, txn: Txn, name: &str) -> Result<Option<u32>, QueryError> {
if self.node_cache_enabled.get() {
if let Some(cached) = self.prop_id_memo.borrow().get(name) {
return Ok(*cached);
}
}
let id = GraphStore::lookup_prop_id_in_txn(txn, name)?;
if self.node_cache_enabled.get() {
self.prop_id_memo.borrow_mut().insert(name.to_string(), id);
}
Ok(id)
}
fn lookup_prop(
&self,
txn: Txn,
pa: &PropAccess,
row: &BindingRow,
) -> Result<Option<PropertyValue>, QueryError> {
let binding = row
.get(&pa.var)
.ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
match binding {
Binding::Node(id) => {
if self.node_cache_enabled.get() {
if let Some(cached) = self.node_cache.borrow().get(id) {
return Ok(cached.props.get(&pa.prop).cloned());
}
}
match self.prop_id_for(txn, &pa.prop)? {
Some(prop_id) => Ok(deleted_entity_access(GraphStore::get_node_prop_in_txn(
txn, *id, prop_id,
)?)?),
None => {
deleted_entity_access(
GraphStore::node_exists_in_txn(txn, *id)?.then_some(()),
)?;
Ok(None)
}
}
}
Binding::Edge(id) => match self.prop_id_for(txn, &pa.prop)? {
Some(prop_id) => Ok(deleted_entity_access(GraphStore::get_edge_prop_in_txn(
txn, *id, prop_id,
)?)?),
None => {
deleted_entity_access(GraphStore::edge_exists_in_txn(txn, *id)?.then_some(()))?;
Ok(None)
}
},
Binding::Value(_) | Binding::List(_) | Binding::Map(_) => Ok(None),
Binding::Path(_) => Err(QueryError::Type(format!(
"'{}' is a path — property access requires a node, relationship, or map",
pa.var
))),
}
}
fn lookup_prop_value(
&self,
txn: Txn,
pa: &PropAccess,
row: &BindingRow,
) -> Result<Value, QueryError> {
match row.get(&pa.var) {
Some(Binding::Map(m)) => Ok(m.get(&pa.prop).cloned().unwrap_or(Value::Null)),
Some(Binding::Value(PropertyValue::Null)) => Ok(Value::Null),
Some(Binding::Value(pv)) => match temporal_component(pv, &pa.prop) {
Some(component) => Ok(Value::Property(component)),
None if is_temporal_property_value(pv) => Ok(Value::Null),
None => Err(QueryError::Type(format!(
"'{}' can't have properties accessed on it -- property access requires a \
node, relationship, map, or temporal value",
pa.var
))),
},
Some(Binding::List(_)) => Err(QueryError::Type(format!(
"'{}' can't have properties accessed on it -- property access requires a node, \
relationship, map, or temporal value, not a list",
pa.var
))),
Some(_) => Ok(match self.lookup_prop(txn, pa, row)? {
Some(PropertyValue::Null) | None => Value::Null,
Some(pv) => property_value_to_value(pv),
}),
None => Err(QueryError::UnboundVariable(pa.var.clone())),
}
}
fn materialize_return(
&self,
txn: Txn,
items: &[ReturnItem],
rows: &[BindingRow],
distinct: bool,
guard: &ExecutionGuard<'_>,
) -> Result<QueryResult, QueryError> {
let columns = items
.iter()
.enumerate()
.map(|(i, item)| {
item.alias
.clone()
.unwrap_or_else(|| default_column_name(&item.expr, i))
})
.collect();
let mut out_rows = if !has_aggregate(items) {
let mut out_rows = Vec::with_capacity(rows.len());
for row in rows {
let mut out_row = Vec::with_capacity(items.len());
for item in items {
out_row.push(self.eval_return_expr(txn, &item.expr, row, guard)?);
}
out_rows.push(out_row);
}
out_rows
} else {
validate_return_items(items)?;
let grouped = self.resolve_grouped_rows(txn, items, rows, guard)?;
grouped
.into_iter()
.map(|bindings| {
bindings
.iter()
.map(|b| self.binding_to_value(txn, b))
.collect::<Result<Vec<_>, _>>()
})
.collect::<Result<Vec<_>, _>>()?
};
if distinct {
out_rows = dedup_rows(out_rows)?;
}
Ok(QueryResult {
columns,
rows: out_rows,
})
}
fn materialize_aggregating_return_with_order(
&self,
txn: Txn,
items: &[ReturnItem],
rows: &[BindingRow],
order_by: &[(ReturnExpr, SortDir)],
skip_limit: (Option<i64>, Option<i64>),
guard: &ExecutionGuard<'_>,
) -> Result<QueryResult, QueryError> {
let (skip, limit) = skip_limit;
enum OrderKeySource {
RealColumn(usize),
Extra(usize),
}
let mut extra_exprs: Vec<ReturnExpr> = Vec::new();
let order_by_source: Vec<OrderKeySource> = order_by
.iter()
.map(|(expr, _)| {
match items
.iter()
.enumerate()
.position(|(i, it)| item_matches_leaf(expr, i, it))
{
Some(i) => OrderKeySource::RealColumn(i),
None => {
let idx = extra_exprs.len();
extra_exprs.push(expr.clone());
OrderKeySource::Extra(idx)
}
}
})
.collect();
let extended_items: Vec<ReturnItem> = items
.iter()
.cloned()
.chain(
extra_exprs
.into_iter()
.map(|expr| ReturnItem { expr, alias: None }),
)
.collect();
validate_return_items(&extended_items)?;
let grouped = self.resolve_grouped_rows(txn, &extended_items, rows, guard)?;
let columns: Vec<String> = items
.iter()
.enumerate()
.map(|(i, item)| {
item.alias
.clone()
.unwrap_or_else(|| default_column_name(&item.expr, i))
})
.collect();
let real_len = items.len();
let mut keyed: Vec<(Vec<Value>, Vec<Value>)> = Vec::with_capacity(grouped.len());
for bindings in grouped {
let values: Vec<Value> = bindings
.iter()
.map(|b| self.binding_to_value(txn, b))
.collect::<Result<Vec<_>, _>>()?;
let (real, extra) = values.split_at(real_len);
let keys: Vec<Value> = order_by_source
.iter()
.map(|src| match src {
OrderKeySource::RealColumn(i) => real[*i].clone(),
OrderKeySource::Extra(k) => extra[*k].clone(),
})
.collect();
keyed.push((keys, real.to_vec()));
}
let rows = top_k_by(keyed, order_by, skip, limit)
.into_iter()
.map(|(_, row)| row)
.collect();
Ok(QueryResult { columns, rows })
}
fn resolve_skip_limit(
&self,
txn: Txn,
expr: Option<&ReturnExpr>,
clause: &str,
guard: &ExecutionGuard<'_>,
) -> Result<Option<i64>, QueryError> {
let Some(expr) = expr else {
return Ok(None);
};
let value = self.eval_return_expr(txn, expr, &BindingRow::new(), guard)?;
let n = match value {
Value::Literal(Literal::Int(n)) | Value::Property(PropertyValue::Int(n)) => n,
_ => {
return Err(QueryError::Semantic(format!(
"{clause} must evaluate to an integer"
)));
}
};
if n < 0 {
return Err(QueryError::Semantic(format!("{clause} can't be negative")));
}
Ok(Some(n))
}
fn eval_return_expr(
&self,
txn: Txn,
expr: &ReturnExpr,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<Value, QueryError> {
match expr {
ReturnExpr::Var(var) => {
let binding = row
.get(var)
.ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
self.binding_to_value(txn, binding)
}
ReturnExpr::Prop(pa) => self.lookup_prop_value(txn, pa, row),
ReturnExpr::PropOf(base, prop) => {
let v = self.eval_return_expr(txn, base, row, guard)?;
property_of_value(&v, prop)
}
ReturnExpr::Lit(lit) => Ok(match lit {
Literal::Null => Value::Null,
other => Value::Literal(other.clone()),
}),
ReturnExpr::Call { name, args, .. } => {
if is_aggregate_name(name) {
return Err(QueryError::Semantic(format!(
"aggregate function '{name}' can only be used as a return item's top-level expression"
)));
}
let lower = name.to_ascii_lowercase();
if lower == "type" {
return self.eval_type_call(txn, args.first(), row, guard);
}
let arg_values = args
.iter()
.map(|a| self.eval_return_expr(txn, a, row, guard))
.collect::<Result<Vec<_>, _>>()?;
if lower == "startnode" || lower == "endnode" {
return self.start_or_end_node(txn, &lower, arg_values.first());
}
call_builtin(name, &arg_values, self.now_snapshot())
}
ReturnExpr::CountStar => Err(QueryError::Semantic(
"count(*) can only be used as a return item's top-level expression".into(),
)),
ReturnExpr::Case { test, whens, else_ } => {
let test_value = match test {
Some(t) => Some(self.eval_return_expr(txn, t, row, guard)?),
None => None,
};
for (when, then) in whens {
let when_value = self.eval_return_expr(txn, when, row, guard)?;
let matched = match &test_value {
Some(tv) => value_eq(tv, &when_value),
None => matches!(when_value, Value::Literal(Literal::Bool(true))),
};
if matched {
return self.eval_return_expr(txn, then, row, guard);
}
}
match else_ {
Some(e) => self.eval_return_expr(txn, e, row, guard),
None => Ok(Value::Null),
}
}
ReturnExpr::Arith(l, op, r) => {
let lv = self.eval_return_expr(txn, l, row, guard)?;
let rv = self.eval_return_expr(txn, r, row, guard)?;
apply_arith(*op, &lv, &rv)
}
ReturnExpr::Neg(e) => {
let v = self.eval_return_expr(txn, e, row, guard)?;
apply_neg(&v)
}
ReturnExpr::ListLit(items) => Ok(Value::List(
items
.iter()
.map(|item| self.eval_return_expr(txn, item, row, guard))
.collect::<Result<Vec<_>, _>>()?,
)),
ReturnExpr::Index(base, index) => {
let base_v = self.eval_return_expr(txn, base, row, guard)?;
let index_v = self.eval_return_expr(txn, index, row, guard)?;
apply_index(&base_v, &index_v)
}
ReturnExpr::Slice(base, start, end) => {
let base_v = self.eval_return_expr(txn, base, row, guard)?;
let start_v = start
.as_deref()
.map(|s| self.eval_return_expr(txn, s, row, guard))
.transpose()?;
let end_v = end
.as_deref()
.map(|e| self.eval_return_expr(txn, e, row, guard))
.transpose()?;
apply_slice(&base_v, start_v.as_ref(), end_v.as_ref())
}
ReturnExpr::ListComp {
var,
source,
where_clause,
project,
} => {
let source_v = self.eval_return_expr(txn, source, row, guard)?;
let items = match source_v {
Value::List(items) => items,
Value::Null => return Ok(Value::Null),
other => {
return Err(QueryError::Type(format!(
"list comprehension source must be a list, got {other:?}"
)))
}
};
let mut result = Vec::with_capacity(items.len());
for item in items {
let mut scoped_row = row.clone();
scoped_row.insert(var.clone(), value_to_binding_restore(&item));
let keep = match where_clause {
Some(w) => {
self.eval_return_expr_bool3(txn, w, &scoped_row, guard)? == Some(true)
}
None => true,
};
if !keep {
continue;
}
result.push(match project {
Some(p) => self.eval_return_expr(txn, p, &scoped_row, guard)?,
None => item,
});
}
Ok(Value::List(result))
}
ReturnExpr::Quantifier {
kind,
var,
source,
where_clause,
} => {
let source_v = self.eval_return_expr(txn, source, row, guard)?;
let items = match source_v {
Value::List(items) => items,
Value::Null => return Ok(Value::Null),
other => {
return Err(QueryError::Type(format!(
"quantifier source must be a list, got {other:?}"
)))
}
};
let mut preds = Vec::with_capacity(items.len());
for item in &items {
let mut scoped_row = row.clone();
scoped_row.insert(var.clone(), value_to_binding_restore(item));
preds.push(match where_clause {
Some(w) => self.eval_return_expr_bool3(txn, w, &scoped_row, guard)?,
None => item_truthy(item),
});
}
Ok(match eval_quantifier(*kind, &preds) {
Some(b) => Value::Literal(Literal::Bool(b)),
None => Value::Null,
})
}
ReturnExpr::MapLit(entries) => {
let mut map = BTreeMap::new();
for (k, v) in entries {
map.insert(k.clone(), self.eval_return_expr(txn, v, row, guard)?);
}
Ok(Value::Map(map))
}
ReturnExpr::And(l, r) => Ok(bool3_to_value(and3(
self.eval_return_expr_bool3(txn, l, row, guard)?,
self.eval_return_expr_bool3(txn, r, row, guard)?,
))),
ReturnExpr::Or(l, r) => Ok(bool3_to_value(or3(
self.eval_return_expr_bool3(txn, l, row, guard)?,
self.eval_return_expr_bool3(txn, r, row, guard)?,
))),
ReturnExpr::Xor(l, r) => Ok(bool3_to_value(xor3(
self.eval_return_expr_bool3(txn, l, row, guard)?,
self.eval_return_expr_bool3(txn, r, row, guard)?,
))),
ReturnExpr::Not(e) => Ok(bool3_to_value(
self.eval_return_expr_bool3(txn, e, row, guard)?.map(|b| !b),
)),
ReturnExpr::Compare(l, op, r) => {
let lv = self.eval_return_expr(txn, l, row, guard)?;
let rv = self.eval_return_expr(txn, r, row, guard)?;
Ok(bool3_to_value(compare_values(&lv, *op, &rv)))
}
ReturnExpr::IsNull(e) => {
let v = self.eval_return_expr(txn, e, row, guard)?;
Ok(Value::Literal(Literal::Bool(matches!(v, Value::Null))))
}
ReturnExpr::In(needle, haystack) => {
let nv = self.eval_return_expr(txn, needle, row, guard)?;
let hv = self.eval_return_expr(txn, haystack, row, guard)?;
Ok(bool3_to_value(list_membership_ternary(&nv, &hv)?))
}
ReturnExpr::HasLabel(var, labels) => {
let binding = row
.get(var)
.ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
match binding {
Binding::Node(id) => {
let node = deleted_entity_access(self.get_node_cached(txn, *id)?)?;
Ok(Value::Literal(Literal::Bool(
labels.iter().all(|l| node.labels.contains(l)),
)))
}
Binding::Edge(id) => {
let edge = deleted_entity_access(GraphStore::get_edge_in_txn(txn, *id)?)?;
Ok(Value::Literal(Literal::Bool(
labels.iter().all(|l| edge.label == *l),
)))
}
Binding::Value(PropertyValue::Null) => Ok(Value::Null),
other => Err(QueryError::Type(format!(
"'{var}' isn't a node or relationship — (n:Label) needs one, got {other:?}"
))),
}
}
ReturnExpr::PatternPredicate(_) => Err(QueryError::Semantic(
"a pattern predicate (`(n)-->()` etc) can only be used inside WHERE".into(),
)),
ReturnExpr::PatternComprehension {
path_var,
pattern,
where_clause,
projection,
} => self.eval_pattern_comprehension(
txn,
PatternComprehensionSpec {
path_var,
pattern,
where_clause,
projection,
},
row,
guard,
),
ReturnExpr::ExistsPattern { .. } | ReturnExpr::ExistsSubquery(_) => Err(
QueryError::Semantic("an exists {} subquery can only be used inside WHERE".into()),
),
}
}
fn eval_pattern_comprehension(
&self,
txn: Txn,
spec: PatternComprehensionSpec<'_>,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<Value, QueryError> {
let PatternComprehensionSpec {
path_var,
pattern,
where_clause,
projection,
} = spec;
if path_var.is_some() {
validate_named_path_pattern(pattern)?;
}
let carried_vars: HashSet<String> = row.keys().cloned().collect();
let (named_pattern, synthesized) = match path_var {
Some(_) => name_pattern_for_path(pattern),
None => (pattern.clone(), HashSet::new()),
};
let wc: Option<Expr> = where_clause.as_deref().cloned();
let plan = apply_index_seeks(build_match_plan(&named_pattern, &wc, &carried_vars)?, txn)?;
let rows = self.eval_plan_with_limit(txn, &plan, std::slice::from_ref(row), guard, None)?;
let mut out = Vec::with_capacity(rows.len());
for mut r in rows {
if let Some(pv) = path_var {
let path_binding = assemble_path(&named_pattern, &r);
for key in &synthesized {
r.remove(key);
}
r.insert(pv.clone(), path_binding);
}
out.push(self.eval_return_expr(txn, projection, &r, guard)?);
}
Ok(Value::List(out))
}
fn eval_return_expr_bool3(
&self,
txn: Txn,
expr: &ReturnExpr,
row: &BindingRow,
guard: &ExecutionGuard<'_>,
) -> Result<Option<bool>, QueryError> {
value_to_bool3(&self.eval_return_expr(txn, expr, row, guard)?)
}
fn delete_targets(
&self,
txn: Txn,
write_txn: &WriteTransaction,
targets: &[ReturnExpr],
rows: &[BindingRow],
detach: bool,
guard: &ExecutionGuard<'_>,
) -> Result<(), QueryError> {
let mut deleted_edges = HashSet::new();
let mut pending_nodes = HashSet::new();
for row in rows {
for target in targets {
if let ReturnExpr::Var(name) = target {
let binding = row
.get(name)
.ok_or_else(|| QueryError::UnboundVariable(name.clone()))?;
delete_binding(
txn,
binding,
write_txn,
&mut deleted_edges,
&mut pending_nodes,
guard,
)?;
} else {
let value = self.eval_return_expr(txn, target, row, guard)?;
delete_value(
&value,
write_txn,
&mut deleted_edges,
&mut pending_nodes,
guard,
)?;
}
}
}
for id in pending_nodes {
GraphStore::delete_node_in_txn(write_txn, id, detach)?;
}
Ok(())
}
fn materialize_delete(
&self,
txn: Txn,
targets: &[ReturnExpr],
rows: &[BindingRow],
detach: bool,
ret: &Option<ReturnTail>,
guard: &ExecutionGuard<'_>,
) -> Result<QueryResult, QueryError> {
let write_txn = require_write_txn(txn);
self.delete_targets(txn, write_txn, targets, rows, detach, guard)?;
let result = match ret {
Some(rt) => self.materialize_return(txn, &rt.items, rows, rt.distinct, guard)?,
None => QueryResult {
columns: vec![],
rows: vec![],
},
};
Ok(result)
}
fn materialize_set(
&self,
txn: Txn,
items: &[SetItem],
rows: &[BindingRow],
ret: &Option<ReturnTail>,
guard: &ExecutionGuard<'_>,
) -> Result<QueryResult, QueryError> {
let write_txn = require_write_txn(txn);
for row in rows {
for item in items {
self.apply_set_item(txn, write_txn, row, item, guard)?;
}
}
match ret {
Some(rt) => self.materialize_return(txn, &rt.items, rows, rt.distinct, guard),
None => Ok(QueryResult {
columns: vec![],
rows: vec![],
}),
}
}
fn materialize_remove(
&self,
txn: Txn,
items: &[RemoveItem],
rows: &[BindingRow],
ret: &Option<ReturnTail>,
guard: &ExecutionGuard<'_>,
) -> Result<QueryResult, QueryError> {
let write_txn = require_write_txn(txn);
for row in rows {
for item in items {
apply_remove_item(write_txn, row, item)?;
}
}
match ret {
Some(rt) => self.materialize_return(txn, &rt.items, rows, rt.distinct, guard),
None => Ok(QueryResult {
columns: vec![],
rows: vec![],
}),
}
}
fn materialize_union(
&self,
txn: Txn,
parts: &[Statement],
all: bool,
guard: &ExecutionGuard<'_>,
) -> Result<QueryResult, QueryError> {
let mut combined: Option<QueryResult> = None;
for part in parts {
let Statement::Match {
clauses,
tail,
order_by,
skip,
limit,
} = part
else {
unreachable!(
"union_stmt parts are always Statement::Match -- see parser::parse_union_stmt"
)
};
let skip = self.resolve_skip_limit(txn, skip.as_deref(), "SKIP", guard)?;
let limit = self.resolve_skip_limit(txn, limit.as_deref(), "LIMIT", guard)?;
let result = self.execute_match(
txn,
clauses,
tail,
ResultModifiers {
order_by,
skip,
limit,
},
guard,
)?;
combined = Some(match combined {
None => result,
Some(mut acc) => {
if acc.columns != result.columns {
return Err(QueryError::Semantic(format!(
"UNION requires every part to return the same columns -- got {:?} \
and {:?}",
acc.columns, result.columns
)));
}
acc.rows.extend(result.rows);
acc
}
});
guard.check_intermediate_rows(combined.as_ref().map(|r| r.rows.len()).unwrap_or(0))?;
}
let mut result = combined.expect("union_stmt grammar guarantees at least 2 parts");
if !all {
result.rows = dedup_rows(result.rows)?;
}
Ok(result)
}
fn apply_set_item(
&self,
txn: Txn,
write_txn: &WriteTransaction,
row: &BindingRow,
item: &SetItem,
guard: &ExecutionGuard<'_>,
) -> Result<(), QueryError> {
match item {
SetItem::Prop(pa, expr) => {
let binding = row
.get(&pa.var)
.ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
if matches!(binding, Binding::Value(PropertyValue::Null)) {
return Ok(());
}
let node_id = if let Binding::Node(id) = binding {
Some(*id)
} else {
None
};
let edge_id = if let Binding::Edge(id) = binding {
Some(*id)
} else {
None
};
if node_id.is_none() && edge_id.is_none() {
return Err(QueryError::UnboundVariable(format!(
"'{}' is a WITH-projected scalar, not a node/edge — SET needs a graph binding",
pa.var
)));
}
let value = self.eval_return_expr(txn, expr, row, guard)?;
if matches!(value, Value::Null) {
if let Some(id) = node_id {
GraphStore::remove_node_prop_in_txn(write_txn, id, &pa.prop)?;
}
if let Some(id) = edge_id {
GraphStore::remove_edge_prop_in_txn(write_txn, id, &pa.prop)?;
}
} else {
let pv = value_to_storable_property(&value).ok_or_else(|| {
QueryError::Type(format!(
"property '{}' can't be stored -- MarsDB's node/edge properties are limited \
to null/bool/int/float/string/date/duration; a list/map/node/edge/path value \
(got {value:?}) isn't storable",
pa.prop
))
})?;
if let Some(id) = node_id {
GraphStore::set_node_prop_in_txn(write_txn, id, &pa.prop, pv.clone())?;
}
if let Some(id) = edge_id {
GraphStore::set_edge_prop_in_txn(write_txn, id, &pa.prop, pv)?;
}
}
}
SetItem::Labels(var, labels) => {
let binding = row
.get(var)
.ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
match binding {
Binding::Node(id) => {
for label in labels {
GraphStore::add_node_label_in_txn(write_txn, *id, label)?;
}
}
Binding::Value(PropertyValue::Null) => {}
_ => {
return Err(QueryError::UnboundVariable(format!(
"'{var}' isn't a node — SET can only add labels to a node"
)))
}
}
}
SetItem::MapAssign { var, value, merge } => {
let binding = row
.get(var)
.ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
if matches!(binding, Binding::Value(PropertyValue::Null)) {
return Ok(());
}
let node_id = if let Binding::Node(id) = binding {
Some(*id)
} else {
None
};
let edge_id = if let Binding::Edge(id) = binding {
Some(*id)
} else {
None
};
if node_id.is_none() && edge_id.is_none() {
return Err(QueryError::UnboundVariable(format!(
"'{var}' is a WITH-projected scalar, not a node/edge — SET needs a graph binding"
)));
}
let map_value = self.eval_return_expr(txn, value, row, guard)?;
let entries = match map_value {
Value::Map(entries) => entries,
Value::Node(n) => n
.props
.into_iter()
.map(|(k, v)| (k, property_value_to_value(v)))
.collect(),
Value::Edge(e) => e
.props
.into_iter()
.map(|(k, v)| (k, property_value_to_value(v)))
.collect(),
other => {
return Err(QueryError::Type(format!(
"SET {var} = ...{} needs a map, node, or relationship, got {other:?}",
if *merge { " (+=)" } else { "" }
)))
}
};
if !merge {
let existing_keys: Vec<String> = if let Some(id) = node_id {
deleted_entity_access(GraphStore::get_node_in_txn(txn, id)?)?
.props
.into_keys()
.collect()
} else {
deleted_entity_access(GraphStore::get_edge_in_txn(
txn,
edge_id.expect("node_id or edge_id is Some, checked above"),
)?)?
.props
.into_keys()
.collect()
};
for key in existing_keys {
if let Some(id) = node_id {
GraphStore::remove_node_prop_in_txn(write_txn, id, &key)?;
}
if let Some(id) = edge_id {
GraphStore::remove_edge_prop_in_txn(write_txn, id, &key)?;
}
}
}
for (key, entry_value) in entries {
if matches!(entry_value, Value::Null) {
if let Some(id) = node_id {
GraphStore::remove_node_prop_in_txn(write_txn, id, &key)?;
}
if let Some(id) = edge_id {
GraphStore::remove_edge_prop_in_txn(write_txn, id, &key)?;
}
continue;
}
let pv = value_to_storable_property(&entry_value).ok_or_else(|| {
QueryError::Type(format!(
"property '{key}' can't be stored -- MarsDB's node/edge properties are \
limited to null/bool/int/float/string/date/duration/list; a map/node/\
edge/path value (got {entry_value:?}) isn't storable"
))
})?;
if let Some(id) = node_id {
GraphStore::set_node_prop_in_txn(write_txn, id, &key, pv.clone())?;
}
if let Some(id) = edge_id {
GraphStore::set_edge_prop_in_txn(write_txn, id, &key, pv)?;
}
}
}
}
Ok(())
}
}
fn record_and_delete_edge(
txn: Txn,
write_txn: &WriteTransaction,
id: EdgeId,
guard: &ExecutionGuard<'_>,
) -> Result<(), QueryError> {
if let Some(edge) = GraphStore::get_edge_in_txn(txn, id)? {
guard.record_deleted_edge_type(id, edge.label);
}
GraphStore::delete_edge_in_txn(write_txn, id)?;
Ok(())
}
fn delete_binding(
txn: Txn,
binding: &Binding,
write_txn: &WriteTransaction,
deleted_edges: &mut HashSet<EdgeId>,
pending_nodes: &mut HashSet<NodeId>,
guard: &ExecutionGuard<'_>,
) -> Result<(), QueryError> {
match binding {
Binding::Node(id) => {
pending_nodes.insert(*id);
}
Binding::Edge(id) => {
if deleted_edges.insert(*id) {
record_and_delete_edge(txn, write_txn, *id, guard)?;
}
}
Binding::Path(elems) => {
for elem in elems {
if let PathBinding::Edge(id) = elem {
if deleted_edges.insert(*id) {
record_and_delete_edge(txn, write_txn, *id, guard)?;
}
}
}
for elem in elems {
if let PathBinding::Node(id) = elem {
pending_nodes.insert(*id);
}
}
}
Binding::Value(PropertyValue::Null) => {}
Binding::Value(_) | Binding::List(_) | Binding::Map(_) => {
return Err(QueryError::Type(
"DELETE needs a node, relationship, or path, not a scalar/list/map".into(),
))
}
}
Ok(())
}
fn delete_value(
value: &Value,
write_txn: &WriteTransaction,
deleted_edges: &mut HashSet<EdgeId>,
pending_nodes: &mut HashSet<NodeId>,
guard: &ExecutionGuard<'_>,
) -> Result<(), QueryError> {
match value {
Value::Node(n) => {
pending_nodes.insert(n.id);
}
Value::Edge(e) => {
if deleted_edges.insert(e.id) {
guard.record_deleted_edge_type(e.id, e.label.clone());
GraphStore::delete_edge_in_txn(write_txn, e.id)?;
}
}
Value::Path(elems) => {
for elem in elems {
if let PathElem::Edge(e) = elem {
if deleted_edges.insert(e.id) {
guard.record_deleted_edge_type(e.id, e.label.clone());
GraphStore::delete_edge_in_txn(write_txn, e.id)?;
}
}
}
for elem in elems {
if let PathElem::Node(n) = elem {
pending_nodes.insert(n.id);
}
}
}
Value::Null => {}
other => {
return Err(QueryError::Type(format!(
"DELETE needs a node, relationship, or path, got {other:?}"
)))
}
}
Ok(())
}
fn apply_remove_item(
write_txn: &WriteTransaction,
row: &BindingRow,
item: &RemoveItem,
) -> Result<(), QueryError> {
match item {
RemoveItem::Prop(pa) => {
let binding = row
.get(&pa.var)
.ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
match binding {
Binding::Node(id) => {
GraphStore::remove_node_prop_in_txn(write_txn, *id, &pa.prop)?;
}
Binding::Edge(id) => {
GraphStore::remove_edge_prop_in_txn(write_txn, *id, &pa.prop)?;
}
Binding::Value(PropertyValue::Null) => {}
Binding::Value(_) | Binding::List(_) | Binding::Map(_) | Binding::Path(_) => {
return Err(QueryError::UnboundVariable(format!(
"'{}' is a WITH-projected scalar, not a node/edge — REMOVE needs a graph binding",
pa.var
)))
}
}
}
RemoveItem::Labels(var, labels) => {
let binding = row
.get(var)
.ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
match binding {
Binding::Node(id) => {
for label in labels {
GraphStore::remove_node_label_in_txn(write_txn, *id, label)?;
}
}
Binding::Value(PropertyValue::Null) => {}
_ => {
return Err(QueryError::UnboundVariable(format!(
"'{var}' isn't a node — REMOVE can only remove labels from a node"
)))
}
}
}
}
Ok(())
}
fn tail_is_distinct_return(tail: &Option<Tail>) -> bool {
match tail {
Some(Tail::Return(_, distinct)) | Some(Tail::ReturnStar(distinct)) => *distinct,
Some(Tail::Delete(_, ret))
| Some(Tail::DetachDelete(_, ret))
| Some(Tail::Set(_, ret))
| Some(Tail::Remove(_, ret))
| Some(Tail::Create(_, ret)) => ret.as_ref().is_some_and(|rt| rt.distinct),
None => false,
}
}
pub fn is_read_only(stmt: &Statement) -> bool {
if let Statement::Union { parts, .. } = stmt {
return parts.iter().all(is_read_only);
}
let Statement::Match {
tail: Some(Tail::Return(_, _)) | Some(Tail::ReturnStar(_)),
clauses,
..
} = stmt
else {
return false;
};
!clauses.iter().any(|c| {
matches!(
c,
QueryClause::Merge(_)
| QueryClause::Set(_)
| QueryClause::Delete { .. }
| QueryClause::Remove(_)
| QueryClause::Create(_)
| QueryClause::Call(_)
)
})
}
fn require_write_txn(txn: Txn<'_>) -> &WriteTransaction {
let Txn::Write(write_txn) = txn else {
unreachable!(
"materialize_delete/materialize_set/QueryClause::Set only reached via the \
write-dispatch path in Executor::execute — is_read_only(stmt) is false for any \
statement with one of these, so execute always opens a WriteTransaction for them"
)
};
write_txn
}
fn default_column_name(expr: &ReturnExpr, idx: usize) -> String {
match expr {
ReturnExpr::Var(v) => v.clone(),
ReturnExpr::Prop(pa) => format!("{}.{}", pa.var, pa.prop),
ReturnExpr::Lit(_) => format!("col{idx}"),
ReturnExpr::Call { name, .. } => format!("{name}(...)"),
ReturnExpr::CountStar => "count(*)".to_string(),
ReturnExpr::Case { .. } => format!("case{idx}"),
ReturnExpr::Arith(..) | ReturnExpr::Neg(..) => format!("col{idx}"),
ReturnExpr::ListLit(..)
| ReturnExpr::Index(..)
| ReturnExpr::PropOf(..)
| ReturnExpr::Slice(..)
| ReturnExpr::ListComp { .. }
| ReturnExpr::Quantifier { .. }
| ReturnExpr::MapLit(..)
| ReturnExpr::And(..)
| ReturnExpr::Or(..)
| ReturnExpr::Xor(..)
| ReturnExpr::Not(..)
| ReturnExpr::Compare(..)
| ReturnExpr::IsNull(..)
| ReturnExpr::In(..)
| ReturnExpr::HasLabel(..)
| ReturnExpr::PatternPredicate(..)
| ReturnExpr::PatternComprehension { .. }
| ReturnExpr::ExistsPattern { .. }
| ReturnExpr::ExistsSubquery(_) => format!("col{idx}"),
}
}
pub(crate) fn with_item_output_name((i, item): (usize, &ReturnItem)) -> String {
item.alias
.clone()
.unwrap_or_else(|| default_column_name(&item.expr, i))
}
fn collect_agg_nodes<'a>(expr: &'a ReturnExpr, out: &mut Vec<&'a ReturnExpr>) {
match expr {
ReturnExpr::CountStar => out.push(expr),
ReturnExpr::Call { name, args, .. } => {
if is_aggregate_name(name) {
out.push(expr);
} else {
for arg in args {
collect_agg_nodes(arg, out);
}
}
}
ReturnExpr::Case { test, whens, else_ } => {
if let Some(t) = test.as_deref() {
collect_agg_nodes(t, out);
}
for (w, t) in whens {
collect_agg_nodes(w, out);
collect_agg_nodes(t, out);
}
if let Some(e) = else_.as_deref() {
collect_agg_nodes(e, out);
}
}
ReturnExpr::Arith(l, _, r) => {
collect_agg_nodes(l, out);
collect_agg_nodes(r, out);
}
ReturnExpr::Neg(e) => collect_agg_nodes(e, out),
ReturnExpr::ListLit(items) => {
for item in items {
collect_agg_nodes(item, out);
}
}
ReturnExpr::Index(base, index) => {
collect_agg_nodes(base, out);
collect_agg_nodes(index, out);
}
ReturnExpr::PropOf(base, _) => collect_agg_nodes(base, out),
ReturnExpr::Slice(base, start, end) => {
collect_agg_nodes(base, out);
if let Some(s) = start.as_deref() {
collect_agg_nodes(s, out);
}
if let Some(e) = end.as_deref() {
collect_agg_nodes(e, out);
}
}
ReturnExpr::ListComp {
source, project, ..
} => {
collect_agg_nodes(source, out);
if let Some(p) = project.as_deref() {
collect_agg_nodes(p, out);
}
}
ReturnExpr::Quantifier { source, .. } => collect_agg_nodes(source, out),
ReturnExpr::MapLit(entries) => {
for (_, v) in entries {
collect_agg_nodes(v, out);
}
}
ReturnExpr::And(l, r) | ReturnExpr::Or(l, r) | ReturnExpr::Xor(l, r) => {
collect_agg_nodes(l, out);
collect_agg_nodes(r, out);
}
ReturnExpr::Not(e) => collect_agg_nodes(e, out),
ReturnExpr::Compare(l, _, r) => {
collect_agg_nodes(l, out);
collect_agg_nodes(r, out);
}
ReturnExpr::IsNull(e) => collect_agg_nodes(e, out),
ReturnExpr::In(needle, haystack) => {
collect_agg_nodes(needle, out);
collect_agg_nodes(haystack, out);
}
ReturnExpr::Var(_)
| ReturnExpr::Prop(_)
| ReturnExpr::Lit(_)
| ReturnExpr::HasLabel(..)
| ReturnExpr::PatternPredicate(..)
| ReturnExpr::PatternComprehension { .. }
| ReturnExpr::ExistsPattern { .. }
| ReturnExpr::ExistsSubquery(_) => {}
}
}
pub(crate) fn contains_aggregate(expr: &ReturnExpr) -> bool {
match expr {
ReturnExpr::CountStar => true,
ReturnExpr::Call { name, args, .. } => {
is_aggregate_name(name) || args.iter().any(contains_aggregate)
}
ReturnExpr::Case { test, whens, else_ } => {
test.as_deref().is_some_and(contains_aggregate)
|| whens
.iter()
.any(|(w, t)| contains_aggregate(w) || contains_aggregate(t))
|| else_.as_deref().is_some_and(contains_aggregate)
}
ReturnExpr::Arith(l, _, r) => contains_aggregate(l) || contains_aggregate(r),
ReturnExpr::Neg(e) => contains_aggregate(e),
ReturnExpr::ListLit(items) => items.iter().any(contains_aggregate),
ReturnExpr::Index(base, index) => contains_aggregate(base) || contains_aggregate(index),
ReturnExpr::PropOf(base, _) => contains_aggregate(base),
ReturnExpr::Slice(base, start, end) => {
contains_aggregate(base)
|| start.as_deref().is_some_and(contains_aggregate)
|| end.as_deref().is_some_and(contains_aggregate)
}
ReturnExpr::ListComp {
source, project, ..
} => contains_aggregate(source) || project.as_deref().is_some_and(contains_aggregate),
ReturnExpr::Quantifier { source, .. } => contains_aggregate(source),
ReturnExpr::MapLit(entries) => entries.iter().any(|(_, v)| contains_aggregate(v)),
ReturnExpr::And(l, r) | ReturnExpr::Or(l, r) | ReturnExpr::Xor(l, r) => {
contains_aggregate(l) || contains_aggregate(r)
}
ReturnExpr::Not(e) => contains_aggregate(e),
ReturnExpr::Compare(l, _, r) => contains_aggregate(l) || contains_aggregate(r),
ReturnExpr::IsNull(e) => contains_aggregate(e),
ReturnExpr::In(needle, haystack) => {
contains_aggregate(needle) || contains_aggregate(haystack)
}
ReturnExpr::Var(_)
| ReturnExpr::Prop(_)
| ReturnExpr::Lit(_)
| ReturnExpr::HasLabel(..)
| ReturnExpr::PatternPredicate(..)
| ReturnExpr::PatternComprehension { .. }
| ReturnExpr::ExistsPattern { .. }
| ReturnExpr::ExistsSubquery(_) => false,
}
}
fn fast_direction(dir: ExpandDirection) -> Option<Direction> {
match dir {
ExpandDirection::Out => Some(Direction::Out),
ExpandDirection::In => Some(Direction::In),
ExpandDirection::Either => None,
}
}
#[allow(clippy::option_option)]
fn fast_label(labels: &[String]) -> Option<Option<&str>> {
match labels {
[] => Some(None),
[one] => Some(Some(one.as_str())),
_ => None,
}
}
fn plan_contains_expansion(plan: &LogicalPlan) -> bool {
match plan {
LogicalPlan::Expand { .. }
| LogicalPlan::VarExpand { .. }
| LogicalPlan::MatchRelList { .. }
| LogicalPlan::Seed { .. } => true,
LogicalPlan::Filter { input, .. } => plan_contains_expansion(input),
LogicalPlan::AllNodesScan { .. }
| LogicalPlan::NodeByLabelScan { .. }
| LogicalPlan::IndexSeek { .. } => false,
}
}
pub(crate) fn has_aggregate(items: &[ReturnItem]) -> bool {
items.iter().any(|item| contains_aggregate(&item.expr))
}
fn contains_rand_call(expr: &ReturnExpr) -> bool {
match expr {
ReturnExpr::Call { name, args, .. } => {
name.eq_ignore_ascii_case("rand") || args.iter().any(contains_rand_call)
}
ReturnExpr::Case { test, whens, else_ } => {
test.as_deref().is_some_and(contains_rand_call)
|| whens
.iter()
.any(|(w, t)| contains_rand_call(w) || contains_rand_call(t))
|| else_.as_deref().is_some_and(contains_rand_call)
}
ReturnExpr::Arith(l, _, r) => contains_rand_call(l) || contains_rand_call(r),
ReturnExpr::Neg(e) => contains_rand_call(e),
ReturnExpr::ListLit(items) => items.iter().any(contains_rand_call),
ReturnExpr::Index(base, index) => contains_rand_call(base) || contains_rand_call(index),
ReturnExpr::PropOf(base, _) => contains_rand_call(base),
ReturnExpr::Slice(base, start, end) => {
contains_rand_call(base)
|| start.as_deref().is_some_and(contains_rand_call)
|| end.as_deref().is_some_and(contains_rand_call)
}
ReturnExpr::ListComp {
source, project, ..
} => contains_rand_call(source) || project.as_deref().is_some_and(contains_rand_call),
ReturnExpr::Quantifier { source, .. } => contains_rand_call(source),
ReturnExpr::MapLit(entries) => entries.iter().any(|(_, v)| contains_rand_call(v)),
ReturnExpr::And(l, r) | ReturnExpr::Or(l, r) | ReturnExpr::Xor(l, r) => {
contains_rand_call(l) || contains_rand_call(r)
}
ReturnExpr::Not(e) => contains_rand_call(e),
ReturnExpr::Compare(l, _, r) => contains_rand_call(l) || contains_rand_call(r),
ReturnExpr::IsNull(e) => contains_rand_call(e),
ReturnExpr::In(needle, haystack) => {
contains_rand_call(needle) || contains_rand_call(haystack)
}
ReturnExpr::CountStar
| ReturnExpr::Var(_)
| ReturnExpr::Prop(_)
| ReturnExpr::Lit(_)
| ReturnExpr::HasLabel(..)
| ReturnExpr::PatternPredicate(..)
| ReturnExpr::PatternComprehension { .. }
| ReturnExpr::ExistsPattern { .. }
| ReturnExpr::ExistsSubquery(_) => false,
}
}
pub(crate) fn return_star_items(
names: impl Iterator<Item = String>,
) -> Result<Vec<ReturnItem>, QueryError> {
let names: Vec<String> = names.collect();
if names.is_empty() {
return Err(QueryError::Semantic(
"RETURN * needs at least one variable in scope".into(),
));
}
Ok(star_items(names))
}
pub(crate) fn with_star_items(names: impl Iterator<Item = String>) -> Vec<ReturnItem> {
star_items(names.collect())
}
fn star_items(mut names: Vec<String>) -> Vec<ReturnItem> {
names.sort();
names
.into_iter()
.map(|name| ReturnItem {
expr: ReturnExpr::Var(name),
alias: None,
})
.collect()
}
pub(crate) fn validate_return_items(items: &[ReturnItem]) -> Result<(), QueryError> {
for item in items {
if contains_aggregate(&item.expr) {
validate_composed_expr(&item.expr, items)?;
}
}
Ok(())
}
pub(crate) fn item_matches_leaf(expr: &ReturnExpr, index: usize, item: &ReturnItem) -> bool {
item.expr == *expr
|| matches!(expr, ReturnExpr::Var(name) if *name == with_item_output_name((index, item)))
}
pub(crate) fn validate_composed_expr(
expr: &ReturnExpr,
items: &[ReturnItem],
) -> Result<(), QueryError> {
if matches!(expr, ReturnExpr::CountStar) {
return Ok(());
}
if let ReturnExpr::Call { name, args, .. } = expr {
if is_aggregate_name(name) {
let expected_args = if is_percentile_name(name) { 2 } else { 1 };
if args.len() != expected_args {
return Err(QueryError::Semantic(if expected_args == 2 {
format!("{name}() takes exactly two arguments (the value, then the percentile)")
} else {
format!(
"{name}() takes exactly one argument (use count(*) for a row count with no argument)"
)
}));
}
for arg in args {
if contains_aggregate(arg) {
return Err(QueryError::Semantic(format!(
"aggregate function '{name}' can't take another aggregate as an argument"
)));
}
if contains_rand_call(arg) {
return Err(QueryError::Semantic(format!(
"aggregate function '{name}' can't take a non-deterministic expression \
(e.g. rand()) as an argument"
)));
}
}
return Ok(());
}
}
if matches!(expr, ReturnExpr::Var(_) | ReturnExpr::Prop(_)) {
let is_grouping_key = items
.iter()
.enumerate()
.any(|(i, it)| item_matches_leaf(expr, i, it) && !contains_aggregate(&it.expr));
return if is_grouping_key {
Ok(())
} else {
Err(QueryError::Semantic(format!(
"{expr:?} is used alongside an aggregate function but isn't itself one of this \
RETURN/WITH's own items -- once any item aggregates, every other value used \
with it must be listed as its own explicit grouping key"
)))
};
}
match expr {
ReturnExpr::Case { test, whens, else_ } => {
if let Some(t) = test.as_deref() {
validate_composed_expr(t, items)?;
}
for (w, t) in whens {
validate_composed_expr(w, items)?;
validate_composed_expr(t, items)?;
}
if let Some(e) = else_.as_deref() {
validate_composed_expr(e, items)?;
}
}
ReturnExpr::Call { args, .. } => {
for arg in args {
validate_composed_expr(arg, items)?;
}
}
ReturnExpr::Arith(l, _, r) => {
validate_composed_expr(l, items)?;
validate_composed_expr(r, items)?;
}
ReturnExpr::Neg(e) => validate_composed_expr(e, items)?,
ReturnExpr::ListLit(list_items) => {
for item in list_items {
validate_composed_expr(item, items)?;
}
}
ReturnExpr::Index(base, index) => {
validate_composed_expr(base, items)?;
validate_composed_expr(index, items)?;
}
ReturnExpr::PropOf(base, _) => validate_composed_expr(base, items)?,
ReturnExpr::Slice(base, start, end) => {
validate_composed_expr(base, items)?;
if let Some(s) = start.as_deref() {
validate_composed_expr(s, items)?;
}
if let Some(e) = end.as_deref() {
validate_composed_expr(e, items)?;
}
}
ReturnExpr::ListComp {
source,
project,
where_clause,
..
} => {
if project.as_deref().is_some_and(contains_aggregate) {
return Err(QueryError::Semantic(
"an aggregate function can't be used inside a list comprehension's projection"
.into(),
));
}
validate_composed_expr(source, items)?;
let _ = where_clause;
}
ReturnExpr::Quantifier { source, .. } => validate_composed_expr(source, items)?,
ReturnExpr::MapLit(entries) => {
for (_, v) in entries {
validate_composed_expr(v, items)?;
}
}
ReturnExpr::And(l, r) | ReturnExpr::Or(l, r) | ReturnExpr::Xor(l, r) => {
validate_composed_expr(l, items)?;
validate_composed_expr(r, items)?;
}
ReturnExpr::Not(e) => validate_composed_expr(e, items)?,
ReturnExpr::Compare(l, _, r) => {
validate_composed_expr(l, items)?;
validate_composed_expr(r, items)?;
}
ReturnExpr::IsNull(e) => validate_composed_expr(e, items)?,
ReturnExpr::In(needle, haystack) => {
validate_composed_expr(needle, items)?;
validate_composed_expr(haystack, items)?;
}
ReturnExpr::CountStar
| ReturnExpr::Var(_)
| ReturnExpr::Prop(_)
| ReturnExpr::Lit(_)
| ReturnExpr::HasLabel(..)
| ReturnExpr::PatternPredicate(..)
| ReturnExpr::PatternComprehension { .. }
| ReturnExpr::ExistsPattern { .. }
| ReturnExpr::ExistsSubquery(_) => {}
}
Ok(())
}
pub(crate) fn validate_order_by_composed_expr(
expr: &ReturnExpr,
items: &[ReturnItem],
) -> Result<(), QueryError> {
validate_composed_expr(expr, items)?;
let mut agg_nodes = Vec::new();
collect_agg_nodes(expr, &mut agg_nodes);
for node in agg_nodes {
let matches_item = items
.iter()
.enumerate()
.any(|(i, it)| item_matches_leaf(node, i, it));
if !matches_item {
return Err(QueryError::Semantic(
"ORDER BY aggregate does not match any RETURN/WITH item".into(),
));
}
}
Ok(())
}
fn binding_hash_key(b: &Binding) -> Result<HashKey, QueryError> {
Ok(match b {
Binding::Node(id) => HashKey::Node(*id),
Binding::Edge(id) => HashKey::Edge(*id),
Binding::Value(pv) => property_value_hash_key(pv),
Binding::List(items) => HashKey::List(
items
.iter()
.map(value_hash_key)
.collect::<Result<Vec<_>, _>>()?,
),
Binding::Path(elems) => HashKey::List(
elems
.iter()
.map(|e| match e {
PathBinding::Node(id) => HashKey::Node(*id),
PathBinding::Edge(id) => HashKey::Edge(*id),
})
.collect(),
),
Binding::Map(m) => HashKey::List(
m.iter()
.map(|(k, v)| -> Result<HashKey, QueryError> {
Ok(HashKey::List(vec![
HashKey::Str(k.clone()),
value_hash_key(v)?,
]))
})
.collect::<Result<Vec<_>, _>>()?,
),
})
}
fn project_call_row(
sig: &ProcedureSignature,
proc_row: &[Value],
yield_items: &CallYield,
) -> Result<Vec<Value>, QueryError> {
match yield_items {
CallYield::Star => Ok(proc_row.to_vec()),
CallYield::Items(items, _) => items
.iter()
.map(|(name, _)| {
let idx = sig.outputs.iter().position(|o| o == name).ok_or_else(|| {
QueryError::Semantic(format!(
"'{name}' isn't a declared output of this procedure"
))
})?;
Ok(proc_row[idx].clone())
})
.collect(),
}
}
fn value_matches_declared_type(value: &Value, declared: &str) -> bool {
if matches!(value, Value::Null) {
return true;
}
let is_int = matches!(
value,
Value::Literal(Literal::Int(_)) | Value::Property(PropertyValue::Int(_))
);
let is_float = matches!(
value,
Value::Literal(Literal::Float(_)) | Value::Property(PropertyValue::Float(_))
);
match declared.trim_end_matches('?').to_ascii_uppercase().as_str() {
"INTEGER" => is_int,
"FLOAT" | "NUMBER" => is_int || is_float,
"STRING" => matches!(
value,
Value::Literal(Literal::String(_)) | Value::Property(PropertyValue::String(_))
),
"BOOLEAN" => matches!(
value,
Value::Literal(Literal::Bool(_)) | Value::Property(PropertyValue::Bool(_))
),
_ => true,
}
}
fn value_to_binding(v: Value) -> Binding {
match v {
Value::List(items) => Binding::List(items),
Value::Map(m) => Binding::Map(m),
other => Binding::Value(value_to_property_value(&other)),
}
}
fn value_to_binding_restore(v: &Value) -> Binding {
match v {
Value::Node(n) => Binding::Node(n.id),
Value::Edge(e) => Binding::Edge(e.id),
Value::Property(pv) => Binding::Value(pv.clone()),
Value::Literal(lit) => Binding::Value(literal_to_value(lit)),
Value::List(items) => Binding::List(items.clone()),
Value::Map(m) => Binding::Map(m.clone()),
Value::Path(elems) => Binding::Path(elems.iter().map(path_elem_to_binding).collect()),
Value::Null => Binding::Value(PropertyValue::Null),
}
}
fn path_elem_to_binding(elem: &PathElem) -> PathBinding {
match elem {
PathElem::Node(n) => PathBinding::Node(n.id),
PathElem::Edge(e) => PathBinding::Edge(e.id),
}
}
fn name_pattern_for_path(pattern: &Pattern) -> (Pattern, HashSet<String>) {
fn fresh(counter: &mut usize, synthesized: &mut HashSet<String>) -> String {
*counter += 1;
let name = format!("__path_elem{counter}");
synthesized.insert(name.clone());
name
}
let mut counter = 0usize;
let mut synthesized = HashSet::new();
let mut start = pattern.start.clone();
if start.var.is_none() {
start.var = Some(fresh(&mut counter, &mut synthesized));
}
let hops = pattern
.hops
.iter()
.map(|(rel, node)| {
let mut rel = rel.clone();
if rel.hop_range.is_some() {
rel.rel_list_var = rel.var.take();
rel.var = Some(fresh(&mut counter, &mut synthesized));
rel.capture_path_segment = true;
} else if rel.var.is_none() {
rel.var = Some(fresh(&mut counter, &mut synthesized));
}
let mut node = node.clone();
if node.var.is_none() {
node.var = Some(fresh(&mut counter, &mut synthesized));
}
(rel, node)
})
.collect();
(Pattern { start, hops }, synthesized)
}
fn assemble_path(pattern: &Pattern, row: &BindingRow) -> Binding {
let Some(start_id) = path_node_id(pattern.start.var.as_deref(), row) else {
return Binding::Value(PropertyValue::Null);
};
let mut elems = vec![PathBinding::Node(start_id)];
for (rel, node) in &pattern.hops {
if rel.capture_path_segment {
let Some(Binding::Path(segment)) = rel.var.as_deref().and_then(|v| row.get(v)) else {
return Binding::Value(PropertyValue::Null);
};
elems.extend(segment.iter().cloned());
continue;
}
let Some(edge_id) = path_edge_id(rel.var.as_deref(), row) else {
return Binding::Value(PropertyValue::Null);
};
let Some(node_id) = path_node_id(node.var.as_deref(), row) else {
return Binding::Value(PropertyValue::Null);
};
elems.push(PathBinding::Edge(edge_id));
elems.push(PathBinding::Node(node_id));
}
Binding::Path(elems)
}
fn segment_edges_to_list(txn: Txn, segment: &[PathBinding]) -> Result<Binding, QueryError> {
let edges = segment
.iter()
.filter_map(|elem| match elem {
PathBinding::Edge(id) => Some(*id),
PathBinding::Node(_) => None,
})
.map(|id| {
let edge = deleted_entity_access(GraphStore::get_edge_in_txn(txn, id)?)?;
Ok(Value::Edge(edge))
})
.collect::<Result<Vec<_>, QueryError>>()?;
Ok(Binding::List(edges))
}
fn path_node_id(var: Option<&str>, row: &BindingRow) -> Option<NodeId> {
match var.and_then(|v| row.get(v)) {
Some(Binding::Node(id)) => Some(*id),
_ => None,
}
}
fn path_edge_id(var: Option<&str>, row: &BindingRow) -> Option<EdgeId> {
match var.and_then(|v| row.get(v)) {
Some(Binding::Edge(id)) => Some(*id),
_ => None,
}
}
fn require_bound_node(row: &BindingRow, var: &str) -> Result<NodeId, QueryError> {
match row.get(var) {
Some(Binding::Node(id)) => Ok(*id),
_ => Err(QueryError::UnboundVariable(format!(
"'{var}' must already be bound to a node before shortestPath() — match it in a preceding MATCH"
))),
}
}
fn reconstruct_path(
parent: &HashMap<NodeId, (NodeId, EdgeId)>,
start: NodeId,
end: NodeId,
) -> Vec<PathBinding> {
let mut hops = Vec::new();
let mut current = end;
while current != start {
let (prev, edge_id) = parent[¤t];
hops.push((edge_id, current));
current = prev;
}
hops.reverse();
let mut elems = vec![PathBinding::Node(start)];
for (edge_id, node) in hops {
elems.push(PathBinding::Edge(edge_id));
elems.push(PathBinding::Node(node));
}
elems
}
fn value_to_property_value(v: &Value) -> PropertyValue {
match v {
Value::Null => PropertyValue::Null,
Value::Property(pv) => pv.clone(),
Value::Literal(lit) => literal_to_value(lit),
Value::List(items) => {
PropertyValue::List(items.iter().map(value_to_property_value).collect())
}
Value::Node(_) | Value::Edge(_) | Value::Map(_) | Value::Path(_) => PropertyValue::Null,
}
}
fn value_to_storable_property(v: &Value) -> Option<PropertyValue> {
match v {
Value::Null => Some(PropertyValue::Null),
Value::Property(pv) => Some(pv.clone()),
Value::Literal(lit) => Some(literal_to_value(lit)),
Value::List(items) => Some(PropertyValue::List(
items
.iter()
.map(value_to_storable_property)
.collect::<Option<Vec<_>>>()?,
)),
Value::Node(_) | Value::Edge(_) | Value::Map(_) | Value::Path(_) => None,
}
}
fn property_value_to_value(pv: PropertyValue) -> Value {
match pv {
PropertyValue::Null => Value::Null,
PropertyValue::List(items) => {
Value::List(items.into_iter().map(property_value_to_value).collect())
}
other => Value::Property(other),
}
}
fn deleted_entity_access<T>(record: Option<T>) -> Result<T, QueryError> {
record.ok_or_else(|| {
QueryError::UnboundVariable(
"refers to a node/relationship that no longer exists — it was deleted earlier in this statement".into(),
)
})
}
pub(crate) fn literal_to_value(lit: &Literal) -> PropertyValue {
match lit {
Literal::Int(i) => PropertyValue::Int(*i),
Literal::Float(f) => PropertyValue::Float(*f),
Literal::String(s) => PropertyValue::String(s.clone()),
Literal::Bool(b) => PropertyValue::Bool(*b),
Literal::Null => PropertyValue::Null,
Literal::Param(name) => {
unreachable!("param ${name} must be substituted before execution — see params::substitute_params")
}
}
}
fn tag_merge_created(mut row: BindingRow, created: bool) -> BindingRow {
row.insert(
MERGE_CREATED_KEY.to_string(),
Binding::Value(PropertyValue::Bool(created)),
);
row
}
fn neighbors_for_direction(
txn: Txn,
node: NodeId,
direction: ExpandDirection,
rel_labels: &[String],
) -> Result<Vec<AdjEntry>, QueryError> {
let dirs: &[Direction] = match direction {
ExpandDirection::Out => &[Direction::Out],
ExpandDirection::In => &[Direction::In],
ExpandDirection::Either => &[Direction::Out, Direction::In],
};
let mut out = Vec::new();
let mut seen: HashSet<EdgeId> = HashSet::new();
let label_filters: Vec<Option<&str>> = if rel_labels.is_empty() {
vec![None]
} else {
rel_labels.iter().map(|l| Some(l.as_str())).collect()
};
for label in label_filters {
for &dir in dirs {
for entry in GraphStore::neighbors_in_txn(txn, node, dir, label)? {
if seen.insert(entry.edge_id) {
out.push(entry);
}
}
}
}
Ok(out)
}
fn property_of_value(v: &Value, prop: &str) -> Result<Value, QueryError> {
match v {
Value::Node(n) => Ok(n
.props
.get(prop)
.cloned()
.map(property_value_to_value)
.unwrap_or(Value::Null)),
Value::Edge(e) => Ok(e
.props
.get(prop)
.cloned()
.map(property_value_to_value)
.unwrap_or(Value::Null)),
Value::Map(m) => Ok(m.get(prop).cloned().unwrap_or(Value::Null)),
Value::Null => Ok(Value::Null),
Value::Property(PropertyValue::Null) => Ok(Value::Null),
Value::Property(pv) => match temporal_component(pv, prop) {
Some(component) => Ok(Value::Property(component)),
None if is_temporal_property_value(pv) => Ok(Value::Null),
None => Err(QueryError::Type(
"property access requires a node, relationship, map, or temporal value".into(),
)),
},
Value::List(_) | Value::Path(_) => Err(QueryError::Type(
"property access requires a node, relationship, map, or temporal value, not a list \
or path"
.into(),
)),
Value::Literal(_) => Err(QueryError::Type(
"property access requires a node, relationship, map, or temporal value".into(),
)),
}
}