use std::cell::{Cell, RefCell};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::{
atomic::{AtomicBool, Ordering as AtomicOrdering},
Arc,
};
use std::time::{Duration, Instant};
use marsdb_graph::{
AdjEntry, Direction, Edge, EdgeId, GraphStore, 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};
use crate::procedure::{ProcedureProvider, ProcedureSignature};
use crate::result::QueryResult;
use crate::temporal;
use crate::value::{PathElem, Value};
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 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>>,
}
impl<'a> Executor<'a> {
pub fn new(store: &'a GraphStore) -> Self {
Self {
store,
now: Cell::new(None),
}
}
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()?;
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()?;
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 plan = apply_index_seeks(
build_match_plan(&part.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 {
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(GraphStore::get_node_in_txn(
txn, *id,
)?)?),
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(GraphStore::get_node_in_txn(txn, id)?)?;
Ok(Value::Node(node))
}
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(
GraphStore::get_node_in_txn(txn, *id)?,
)?),
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 mut input = self.stream_plan(txn, input, seed, guard, None);
let mut current: Option<(BindingRow, std::vec::IntoIter<AdjEntry>)> = None;
let mut done = false;
let stream = std::iter::from_fn(move || loop {
if done {
return None;
}
if let Some((row, entries)) = &mut current {
if let Some(entry) = entries.next() {
if let Err(error) = guard.relationship_expansion() {
done = true;
return Some(Err(error));
}
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));
}
return Some(Ok(new_row));
}
current = None;
}
let row = match input.next()? {
Ok(row) => row,
Err(error) => {
done = true;
return Some(Err(error));
}
};
let from_id = match row.get(from_var) {
Some(Binding::Node(id)) => *id,
Some(Binding::Value(PropertyValue::Null)) => continue,
_ => {
done = true;
return Some(Err(QueryError::UnboundVariable(from_var.clone())));
}
};
match neighbors_for_direction(txn, from_id, *direction, rel_labels) {
Ok(entries) => current = Some((row, entries.into_iter())),
Err(error) => {
done = true;
return Some(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 mut input = self.stream_plan(txn, input, seed, guard, None);
let mut pending = Vec::new().into_iter();
let mut done = false;
let stream = std::iter::from_fn(move || loop {
if done {
return None;
}
if let Some(row) = pending.next() {
return Some(Ok(row));
}
let row = match input.next()? {
Ok(row) => row,
Err(error) => {
done = true;
return Some(Err(error));
}
};
match 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,
) {
Ok(rows) => pending = rows.into_iter(),
Err(error) => {
done = true;
return Some(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 mut input = self.stream_plan(txn, input, seed, guard, None);
let mut done = false;
let stream = std::iter::from_fn(move || loop {
if done {
return None;
}
let row = match input.next()? {
Ok(row) => row,
Err(error) => {
done = true;
return Some(Err(error));
}
};
match 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,
},
) {
Ok(Some(row)) => return Some(Ok(row)),
Ok(None) => continue,
Err(error) => {
done = true;
return Some(Err(error));
}
}
});
Self::count_stream(Box::new(stream), guard)
}
LogicalPlan::Filter { input, predicate } => {
let mut input = self.stream_plan(txn, input, seed, guard, None);
let mut done = false;
let stream = std::iter::from_fn(move || loop {
if done {
return None;
}
let row = match input.next()? {
Ok(row) => row,
Err(error) => {
done = true;
return Some(Err(error));
}
};
if let Err(error) = guard.checkpoint() {
done = true;
return Some(Err(error));
}
match self.eval_expr(txn, predicate, &row, guard) {
Ok(Some(true)) => return Some(Ok(row)),
Ok(_) => continue,
Err(error) => {
done = true;
return 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 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 = GraphStore::get_node_in_txn(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 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) => {
let node = deleted_entity_access(GraphStore::get_node_in_txn(txn, *id)?)?;
Ok(node.props.get(&pa.prop).cloned())
}
Binding::Edge(id) => {
let edge = deleted_entity_access(GraphStore::get_edge_in_txn(txn, *id)?)?;
Ok(edge.props.get(&pa.prop).cloned())
}
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(GraphStore::get_node_in_txn(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,
}
}
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 compare(prop: &Option<PropertyValue>, op: CompareOp, lit: &Literal) -> Option<bool> {
let Some(prop) = prop else { return None };
if matches!(prop, PropertyValue::Null) || matches!(lit, Literal::Null) {
return None;
}
compare_property_pair(prop, op, &literal_to_value(lit))
}
fn compare_property_pair_opt(
a: &Option<PropertyValue>,
op: CompareOp,
b: &Option<PropertyValue>,
) -> Option<bool> {
let (Some(a), Some(b)) = (a, b) else {
return None;
};
if matches!(a, PropertyValue::Null) || matches!(b, PropertyValue::Null) {
return None;
}
compare_property_pair(a, op, b)
}
fn compare_property_pair(a: &PropertyValue, op: CompareOp, b: &PropertyValue) -> Option<bool> {
match (a, b) {
(PropertyValue::Int(a), PropertyValue::Int(b)) => Some(cmp_ord(op, *a, *b)),
(PropertyValue::Int(a), PropertyValue::Float(b)) => Some(cmp_f64(op, *a as f64, *b)),
(PropertyValue::Float(a), PropertyValue::Float(b)) => Some(cmp_f64(op, *a, *b)),
(PropertyValue::Float(a), PropertyValue::Int(b)) => Some(cmp_f64(op, *a, *b as f64)),
(PropertyValue::String(a), PropertyValue::String(b)) => Some(match op {
CompareOp::StartsWith => a.starts_with(b.as_str()),
CompareOp::EndsWith => a.ends_with(b.as_str()),
CompareOp::Contains => a.contains(b.as_str()),
_ => cmp_ord(op, a.as_str(), b.as_str()),
}),
(PropertyValue::Bool(a), PropertyValue::Bool(b)) => Some(cmp_ord(op, *a, *b)),
(PropertyValue::Date(a), PropertyValue::Date(b)) => Some(cmp_ord(op, *a, *b)),
(PropertyValue::LocalTime(a), PropertyValue::LocalTime(b)) => Some(cmp_ord(op, *a, *b)),
(
PropertyValue::Time {
nanos_of_day: na,
offset_seconds: oa,
},
PropertyValue::Time {
nanos_of_day: nb,
offset_seconds: ob,
},
) => Some(cmp_ord(
op,
na - *oa as i64 * 1_000_000_000,
nb - *ob as i64 * 1_000_000_000,
)),
(
PropertyValue::LocalDateTime {
epoch_seconds: sa,
nanos: na,
},
PropertyValue::LocalDateTime {
epoch_seconds: sb,
nanos: nb,
},
) => Some(cmp_ord(op, (*sa, *na), (*sb, *nb))),
(
PropertyValue::DateTime {
epoch_seconds: sa,
nanos: na,
..
},
PropertyValue::DateTime {
epoch_seconds: sb,
nanos: nb,
..
},
) => Some(cmp_ord(op, (*sa, *na), (*sb, *nb))),
(PropertyValue::Duration { .. }, PropertyValue::Duration { .. }) => match op {
CompareOp::Eq => Some(a == b),
CompareOp::Ne => Some(a != b),
_ => None,
},
_ => match op {
CompareOp::Eq => Some(false),
CompareOp::Ne => Some(true),
CompareOp::StartsWith
| CompareOp::EndsWith
| CompareOp::Contains
| CompareOp::Lt
| CompareOp::Le
| CompareOp::Gt
| CompareOp::Ge => None,
},
}
}
fn value_to_bool3(v: &Value) -> Result<Option<bool>, QueryError> {
match v {
Value::Null => Ok(None),
Value::Literal(Literal::Bool(b)) | Value::Property(PropertyValue::Bool(b)) => Ok(Some(*b)),
other => Err(QueryError::Type(format!(
"expected a boolean, got {other:?}"
))),
}
}
fn bool3_to_value(b: Option<bool>) -> Value {
match b {
Some(b) => Value::Literal(Literal::Bool(b)),
None => Value::Null,
}
}
fn and3(a: Option<bool>, b: Option<bool>) -> Option<bool> {
match (a, b) {
(Some(false), _) | (_, Some(false)) => Some(false),
(Some(true), Some(true)) => Some(true),
_ => None,
}
}
fn or3(a: Option<bool>, b: Option<bool>) -> Option<bool> {
match (a, b) {
(Some(true), _) | (_, Some(true)) => Some(true),
(Some(false), Some(false)) => Some(false),
_ => None,
}
}
fn xor3(a: Option<bool>, b: Option<bool>) -> Option<bool> {
match (a, b) {
(Some(a), Some(b)) => Some(a != b),
_ => None,
}
}
fn cmp_f64(op: CompareOp, a: f64, b: f64) -> bool {
match op {
CompareOp::Eq => a == b,
CompareOp::Ne => a != b,
CompareOp::Lt => a < b,
CompareOp::Le => a <= b,
CompareOp::Gt => a > b,
CompareOp::Ge => a >= b,
CompareOp::StartsWith | CompareOp::EndsWith | CompareOp::Contains => false,
}
}
fn cmp_ord<T: PartialOrd>(op: CompareOp, a: T, b: T) -> bool {
match op {
CompareOp::Eq => a == b,
CompareOp::Ne => a != b,
CompareOp::Lt => a < b,
CompareOp::Le => a <= b,
CompareOp::Gt => a > b,
CompareOp::Ge => a >= b,
CompareOp::StartsWith | CompareOp::EndsWith | CompareOp::Contains => false,
}
}
pub(crate) fn value_eq(a: &Value, b: &Value) -> bool {
match (a, b) {
(Value::Null, Value::Null) => true,
(Value::Null, _) | (_, Value::Null) => false,
(Value::Property(pa), Value::Property(pb)) => property_value_eq(pa, pb),
(Value::Literal(la), Value::Literal(lb)) => la == lb,
(Value::Property(pa), Value::Literal(lb)) => *pa == literal_to_value(lb),
(Value::Literal(la), Value::Property(pb)) => literal_to_value(la) == *pb,
(Value::Node(na), Value::Node(nb)) => na.id == nb.id,
(Value::Edge(ea), Value::Edge(eb)) => ea.id == eb.id,
(Value::List(la), Value::List(lb)) => {
la.len() == lb.len() && la.iter().zip(lb).all(|(x, y)| value_eq(x, y))
}
(Value::Path(pa), Value::Path(pb)) => {
pa.len() == pb.len()
&& pa.iter().zip(pb).all(|(x, y)| match (x, y) {
(PathElem::Node(na), PathElem::Node(nb)) => na.id == nb.id,
(PathElem::Edge(ea), PathElem::Edge(eb)) => ea.id == eb.id,
_ => false,
})
}
_ => false,
}
}
fn property_value_eq(a: &PropertyValue, b: &PropertyValue) -> bool {
match (a, b) {
(
PropertyValue::Time {
nanos_of_day: na,
offset_seconds: oa,
},
PropertyValue::Time {
nanos_of_day: nb,
offset_seconds: ob,
},
) => na - *oa as i64 * 1_000_000_000 == nb - *ob as i64 * 1_000_000_000,
(
PropertyValue::DateTime {
epoch_seconds: sa,
nanos: na,
..
},
PropertyValue::DateTime {
epoch_seconds: sb,
nanos: nb,
..
},
) => sa == sb && na == nb,
_ => a == b,
}
}
enum ArithNum {
Int(i64),
Float(f64),
}
fn as_arith_num(v: &Value) -> Option<ArithNum> {
match v {
Value::Property(PropertyValue::Int(i)) | Value::Literal(Literal::Int(i)) => {
Some(ArithNum::Int(*i))
}
Value::Property(PropertyValue::Float(f)) | Value::Literal(Literal::Float(f)) => {
Some(ArithNum::Float(*f))
}
_ => None,
}
}
fn require_int_arg(v: Option<&Value>, fn_name: &str) -> Result<i64, QueryError> {
match v {
Some(Value::Property(PropertyValue::Int(i))) | Some(Value::Literal(Literal::Int(i))) => {
Ok(*i)
}
other => Err(QueryError::Type(format!(
"{fn_name}() expects an integer argument, got {other:?}"
))),
}
}
fn as_arith_str(v: &Value) -> Option<&str> {
match v {
Value::Property(PropertyValue::String(s)) | Value::Literal(Literal::String(s)) => {
Some(s.as_str())
}
_ => None,
}
}
fn apply_neg(v: &Value) -> Result<Value, QueryError> {
if matches!(v, Value::Null) {
return Ok(Value::Null);
}
Ok(match as_arith_num(v) {
Some(ArithNum::Int(i)) => {
Value::Property(PropertyValue::Int(i.checked_neg().ok_or_else(|| {
QueryError::Type("integer arithmetic overflow".into())
})?))
}
Some(ArithNum::Float(f)) => Value::Property(PropertyValue::Float(-f)),
None => {
return Err(QueryError::Type(format!(
"unary minus needs a number -- got {v:?}"
)))
}
})
}
fn apply_arith(op: ArithOp, a: &Value, b: &Value) -> Result<Value, QueryError> {
if matches!(a, Value::Null) || matches!(b, Value::Null) {
return Ok(Value::Null);
}
if op == ArithOp::Add {
match (a, b) {
(Value::List(xs), Value::List(ys)) => {
let mut combined = xs.clone();
combined.extend(ys.iter().cloned());
return Ok(Value::List(combined));
}
(Value::List(xs), scalar) => {
let mut combined = xs.clone();
combined.push(scalar.clone());
return Ok(Value::List(combined));
}
(scalar, Value::List(ys)) => {
let mut combined = vec![scalar.clone()];
combined.extend(ys.iter().cloned());
return Ok(Value::List(combined));
}
_ => {}
}
if let (Some(sa), Some(sb)) = (as_arith_str(a), as_arith_str(b)) {
return Ok(Value::Property(PropertyValue::String(format!("{sa}{sb}"))));
}
}
if let Some(result) = apply_temporal_arith(op, a, b)? {
return Ok(result);
}
let (Some(na), Some(nb)) = (as_arith_num(a), as_arith_num(b)) else {
return Err(QueryError::Type(format!(
"arithmetic needs two numbers (or, for +, two strings) -- got {a:?} and {b:?}"
)));
};
if op == ArithOp::Pow {
let to_f64 = |n: ArithNum| match n {
ArithNum::Int(i) => i as f64,
ArithNum::Float(f) => f,
};
return Ok(Value::Property(PropertyValue::Float(
to_f64(na).powf(to_f64(nb)),
)));
}
Ok(match (na, nb) {
(ArithNum::Int(x), ArithNum::Int(y)) => {
if matches!(op, ArithOp::Div | ArithOp::Mod) && y == 0 {
return Err(QueryError::Type("division by zero".into()));
}
let value = match op {
ArithOp::Add => x.checked_add(y),
ArithOp::Sub => x.checked_sub(y),
ArithOp::Mul => x.checked_mul(y),
ArithOp::Div => x.checked_div(y),
ArithOp::Mod => x.checked_rem(y),
ArithOp::Pow => unreachable!("handled above"),
}
.ok_or_else(|| QueryError::Type("integer arithmetic overflow".into()))?;
Value::Property(PropertyValue::Int(value))
}
(x, y) => {
let x = match x {
ArithNum::Int(i) => i as f64,
ArithNum::Float(f) => f,
};
let y = match y {
ArithNum::Int(i) => i as f64,
ArithNum::Float(f) => f,
};
Value::Property(PropertyValue::Float(match op {
ArithOp::Add => x + y,
ArithOp::Sub => x - y,
ArithOp::Mul => x * y,
ArithOp::Div => x / y,
ArithOp::Mod => x % y,
ArithOp::Pow => unreachable!("handled above"),
}))
}
})
}
fn as_date(v: &Value) -> Option<i32> {
match v {
Value::Property(PropertyValue::Date(d)) => Some(*d),
_ => None,
}
}
fn as_duration(v: &Value) -> Option<temporal::DurationParts> {
match v {
Value::Property(PropertyValue::Duration {
months,
days,
seconds,
nanos,
}) => Some((*months, *days, *seconds, *nanos)),
_ => None,
}
}
fn duration_value((months, days, seconds, nanos): temporal::DurationParts) -> Value {
Value::Property(PropertyValue::Duration {
months,
days,
seconds,
nanos,
})
}
fn as_local_time(v: &Value) -> Option<i64> {
match v {
Value::Property(PropertyValue::LocalTime(n)) => Some(*n),
_ => None,
}
}
fn as_time(v: &Value) -> Option<(i64, i32)> {
match v {
Value::Property(PropertyValue::Time {
nanos_of_day,
offset_seconds,
}) => Some((*nanos_of_day, *offset_seconds)),
_ => None,
}
}
fn as_local_date_time(v: &Value) -> Option<(i64, i32)> {
match v {
Value::Property(PropertyValue::LocalDateTime {
epoch_seconds,
nanos,
}) => Some((*epoch_seconds, *nanos)),
_ => None,
}
}
fn as_date_time(v: &Value) -> Option<(i64, i32, temporal::TzId)> {
match v {
Value::Property(PropertyValue::DateTime {
epoch_seconds,
nanos,
zone,
}) => Some((*epoch_seconds, *nanos, tz_from_graph(zone))),
_ => None,
}
}
fn tz_from_graph(zone: &GraphTzId) -> temporal::TzId {
match zone {
GraphTzId::Offset(o) => temporal::TzId::Offset(*o),
GraphTzId::Named(name) => temporal::TzId::Named(name.clone()),
}
}
fn tz_to_graph(zone: temporal::TzId) -> GraphTzId {
match zone {
temporal::TzId::Offset(o) => GraphTzId::Offset(o),
temporal::TzId::Named(name) => GraphTzId::Named(name),
}
}
fn apply_temporal_arith(op: ArithOp, a: &Value, b: &Value) -> Result<Option<Value>, QueryError> {
let date_plus_duration =
|d: i32, dur: temporal::DurationParts, negate: bool| -> Result<Value, QueryError> {
let (months, days, seconds, nanos) = dur;
temporal::add_duration_to_date(d, months, days, seconds, nanos, negate)
.map(|d| Value::Property(PropertyValue::Date(d)))
.ok_or_else(|| {
QueryError::Type("date +/- duration produced an out-of-range date".into())
})
};
let local_time_plus_duration = |t: i64, dur: temporal::DurationParts, negate: bool| -> Value {
let (_, _, seconds, nanos) = dur;
Value::Property(PropertyValue::LocalTime(temporal::add_duration_to_time(
t, seconds, nanos, negate,
)))
};
let time_plus_duration =
|(t, offset): (i64, i32), dur: temporal::DurationParts, negate: bool| -> Value {
let (_, _, seconds, nanos) = dur;
Value::Property(PropertyValue::Time {
nanos_of_day: temporal::add_duration_to_time(t, seconds, nanos, negate),
offset_seconds: offset,
})
};
let local_date_time_plus_duration = |(epoch_seconds, existing_nanos): (i64, i32),
dur: temporal::DurationParts,
negate: bool|
-> Result<Value, QueryError> {
let (months, days, seconds, nanos) = dur;
temporal::add_duration_to_local_date_time(
epoch_seconds,
existing_nanos,
months,
days,
seconds,
nanos,
negate,
)
.map(|(epoch_seconds, nanos)| {
Value::Property(PropertyValue::LocalDateTime {
epoch_seconds,
nanos,
})
})
.ok_or_else(|| {
QueryError::Type("local date-time +/- duration produced an out-of-range value".into())
})
};
let date_time_plus_duration =
|(epoch_seconds, existing_nanos, zone): (i64, i32, temporal::TzId),
dur: temporal::DurationParts,
negate: bool|
-> Result<Value, QueryError> {
let (months, days, seconds, nanos) = dur;
let offset_seconds = temporal::resolve_offset(&zone, epoch_seconds);
temporal::add_duration_to_local_date_time(
epoch_seconds + offset_seconds as i64,
existing_nanos,
months,
days,
seconds,
nanos,
negate,
)
.map(|(local_epoch_seconds, nanos)| {
Value::Property(PropertyValue::DateTime {
epoch_seconds: local_epoch_seconds - offset_seconds as i64,
nanos,
zone: tz_to_graph(zone),
})
})
.ok_or_else(|| {
QueryError::Type("date-time +/- duration produced an out-of-range value".into())
})
};
Ok(match op {
ArithOp::Add => {
if let (Some(d), Some(dur)) = (as_date(a), as_duration(b)) {
Some(date_plus_duration(d, dur, false)?)
} else if let (Some(dur), Some(d)) = (as_duration(a), as_date(b)) {
Some(date_plus_duration(d, dur, false)?)
} else if let (Some(t), Some(dur)) = (as_local_time(a), as_duration(b)) {
Some(local_time_plus_duration(t, dur, false))
} else if let (Some(dur), Some(t)) = (as_duration(a), as_local_time(b)) {
Some(local_time_plus_duration(t, dur, false))
} else if let (Some(t), Some(dur)) = (as_time(a), as_duration(b)) {
Some(time_plus_duration(t, dur, false))
} else if let (Some(dur), Some(t)) = (as_duration(a), as_time(b)) {
Some(time_plus_duration(t, dur, false))
} else if let (Some(dt), Some(dur)) = (as_local_date_time(a), as_duration(b)) {
Some(local_date_time_plus_duration(dt, dur, false)?)
} else if let (Some(dur), Some(dt)) = (as_duration(a), as_local_date_time(b)) {
Some(local_date_time_plus_duration(dt, dur, false)?)
} else if let (Some(dt), Some(dur)) = (as_date_time(a), as_duration(b)) {
Some(date_time_plus_duration(dt, dur, false)?)
} else if let (Some(dur), Some(dt)) = (as_duration(a), as_date_time(b)) {
Some(date_time_plus_duration(dt, dur, false)?)
} else if let (Some(x), Some(y)) = (as_duration(a), as_duration(b)) {
Some(duration_value(temporal::add_duration(x, y).ok_or_else(
|| QueryError::Type("duration addition overflow".into()),
)?))
} else {
None
}
}
ArithOp::Sub => {
if let (Some(d), Some(dur)) = (as_date(a), as_duration(b)) {
Some(date_plus_duration(d, dur, true)?)
} else if let (Some(t), Some(dur)) = (as_local_time(a), as_duration(b)) {
Some(local_time_plus_duration(t, dur, true))
} else if let (Some(t), Some(dur)) = (as_time(a), as_duration(b)) {
Some(time_plus_duration(t, dur, true))
} else if let (Some(dt), Some(dur)) = (as_local_date_time(a), as_duration(b)) {
Some(local_date_time_plus_duration(dt, dur, true)?)
} else if let (Some(dt), Some(dur)) = (as_date_time(a), as_duration(b)) {
Some(date_time_plus_duration(dt, dur, true)?)
} else if let (Some(x), Some(y)) = (as_duration(a), as_duration(b)) {
Some(duration_value(temporal::sub_duration(x, y).ok_or_else(
|| QueryError::Type("duration subtraction overflow".into()),
)?))
} else {
None
}
}
ArithOp::Mul => {
if let (Some(dur), Some(f)) = (as_duration(a), value_as_f64(b)) {
Some(duration_value(temporal::scale_duration(dur, f)))
} else if let (Some(f), Some(dur)) = (value_as_f64(a), as_duration(b)) {
Some(duration_value(temporal::scale_duration(dur, f)))
} else {
None
}
}
ArithOp::Div => {
if let (Some(dur), Some(f)) = (as_duration(a), value_as_f64(b)) {
if f == 0.0 {
return Err(QueryError::Type("division by zero".into()));
}
Some(duration_value(temporal::scale_duration(dur, 1.0 / f)))
} else {
None
}
}
ArithOp::Mod => None,
ArithOp::Pow => None,
})
}
fn apply_index(list: &Value, index: &Value) -> Result<Value, QueryError> {
if matches!(list, Value::Null) || matches!(index, Value::Null) {
return Ok(Value::Null);
}
if let Value::Map(entries) = list {
let Some(key) = as_arith_str(index) else {
return Err(QueryError::Type(format!(
"a map index must be a string, got {index:?}"
)));
};
return Ok(entries.get(key).cloned().unwrap_or(Value::Null));
}
if matches!(list, Value::Node(_) | Value::Edge(_) | Value::Property(_)) {
let Some(key) = as_arith_str(index) else {
return Err(QueryError::Type(format!(
"a property index must be a string, got {index:?}"
)));
};
return property_of_value(list, key);
}
let Value::List(items) = list else {
return Err(QueryError::Type(format!(
"[] indexing needs a list or map, got {list:?}"
)));
};
let Some(ArithNum::Int(i)) = as_arith_num(index) else {
return Err(QueryError::Type(format!(
"a list index must be an integer, got {index:?}"
)));
};
let len = items.len() as i64;
let i = if i < 0 { i + len } else { i };
if i < 0 || i >= len {
return Ok(Value::Null);
}
Ok(items[i as usize].clone())
}
fn apply_slice(
list: &Value,
start: Option<&Value>,
end: Option<&Value>,
) -> Result<Value, QueryError> {
if matches!(list, Value::Null) {
return Ok(Value::Null);
}
let Value::List(items) = list else {
return Err(QueryError::Type(format!(
"[..] slicing needs a list, got {list:?}"
)));
};
let len = items.len() as i64;
let clamp = |i: i64| -> i64 {
let i = if i < 0 { i + len } else { i };
i.clamp(0, len)
};
let bound_index = |v: Option<&Value>, default: i64| -> Result<Option<i64>, QueryError> {
match v {
None => Ok(Some(default)),
Some(Value::Null) => Ok(None),
Some(other) => match as_arith_num(other) {
Some(ArithNum::Int(i)) => Ok(Some(clamp(i))),
_ => Err(QueryError::Type(format!(
"a slice bound must be an integer, got {other:?}"
))),
},
}
};
let (Some(start_idx), Some(end_idx)) = (bound_index(start, 0)?, bound_index(end, len)?) else {
return Ok(Value::Null);
};
if start_idx >= end_idx {
return Ok(Value::List(Vec::new()));
}
Ok(Value::List(
items[start_idx as usize..end_idx as usize].to_vec(),
))
}
fn call_builtin(
name: &str,
args: &[Value],
now: temporal::NowSnapshot,
) -> Result<Value, QueryError> {
match name.to_ascii_lowercase().as_str() {
"coalesce" => Ok(args
.iter()
.find(|v| !matches!(v, Value::Null))
.cloned()
.unwrap_or(Value::Null)),
"tointeger" => match args.first() {
Some(v) => to_integer(v),
None => Ok(Value::Null),
},
"tostring" => match args.first() {
Some(v) => to_string_value(v),
None => Ok(Value::Null),
},
"date" => date_builtin(args, now),
"date.transaction" | "date.statement" | "date.realtime" => Ok(now_or_null(args, || {
Value::Property(PropertyValue::Date(now.epoch_day))
})),
"duration" => duration_builtin(args),
"localtime" => local_time_builtin(args, now),
"localtime.transaction" | "localtime.statement" | "localtime.realtime" => {
Ok(now_or_null(args, || {
Value::Property(PropertyValue::LocalTime(now.nanos_of_day))
}))
}
"time" => time_builtin(args, now),
"time.transaction" | "time.statement" | "time.realtime" => {
Ok(now_or_null(args, || {
Value::Property(PropertyValue::Time {
nanos_of_day: now.nanos_of_day,
offset_seconds: 0,
})
}))
}
"localdatetime" => local_date_time_builtin(args, now),
"localdatetime.transaction" | "localdatetime.statement" | "localdatetime.realtime" => {
Ok(now_or_null(args, || {
Value::Property(PropertyValue::LocalDateTime {
epoch_seconds: now.epoch_seconds,
nanos: now.nanos,
})
}))
}
"datetime" => date_time_builtin(args, now),
"datetime.transaction" | "datetime.statement" | "datetime.realtime" => {
Ok(now_or_null(args, || {
Value::Property(PropertyValue::DateTime {
epoch_seconds: now.epoch_seconds,
nanos: now.nanos,
zone: GraphTzId::Offset(0),
})
}))
}
"datetime.fromepoch" => {
let seconds = require_int_arg(args.first(), "datetime.fromepoch")?;
let nanos = require_int_arg(args.get(1), "datetime.fromepoch")?;
Ok(Value::Property(PropertyValue::DateTime {
epoch_seconds: seconds,
nanos: nanos as i32,
zone: GraphTzId::Offset(0),
}))
}
"datetime.fromepochmillis" => {
let millis = require_int_arg(args.first(), "datetime.fromepochmillis")?;
Ok(Value::Property(PropertyValue::DateTime {
epoch_seconds: millis.div_euclid(1000),
nanos: (millis.rem_euclid(1000) * 1_000_000) as i32,
zone: GraphTzId::Offset(0),
}))
}
"duration.between" => {
duration_between_builtin("duration.between", args, temporal::duration_between)
}
"duration.inmonths" => {
duration_between_builtin("duration.inMonths", args, temporal::duration_in_months)
}
"duration.indays" => {
duration_between_builtin("duration.inDays", args, temporal::duration_in_days)
}
"duration.inseconds" => {
duration_between_builtin("duration.inSeconds", args, temporal::duration_in_seconds)
}
"date.truncate" => date_truncate_builtin(args),
"localtime.truncate" => local_time_truncate_builtin(args),
"time.truncate" => time_truncate_builtin(args),
"localdatetime.truncate" => local_date_time_truncate_builtin(args),
"datetime.truncate" => date_time_truncate_builtin(args),
"length" => Ok(match args.first() {
Some(Value::Path(elems)) => {
Value::Property(PropertyValue::Int(((elems.len().max(1) - 1) / 2) as i64))
}
Some(Value::Null) | None => Value::Null,
Some(other) => {
return Err(QueryError::Type(format!(
"length() expects a path, got {other:?}"
)))
}
}),
"keys" => keys_builtin(args.first()),
"labels" => labels_builtin(args.first()),
"type" => type_builtin(args.first()),
"properties" => properties_builtin(args.first()),
"id" => id_builtin(args.first()),
"size" => size_builtin(args.first()),
"nodes" => nodes_builtin(args.first()),
"relationships" => relationships_builtin(args.first()),
"head" => list_edge_builtin(args.first(), "head", |items| items.first().cloned()),
"last" => list_edge_builtin(args.first(), "last", |items| items.last().cloned()),
"tail" => match args.first() {
Some(Value::List(items)) => Ok(Value::List(
items.iter().skip(1).cloned().collect::<Vec<_>>(),
)),
Some(Value::Null) | None => Ok(Value::Null),
Some(other) => Err(QueryError::Type(format!(
"tail() expects a list, got {other:?}"
))),
},
"range" => range_builtin(args),
"exists" => Ok(Value::Literal(Literal::Bool(!matches!(
args.first(),
None | Some(Value::Null)
)))),
"toupper" | "upper" => string_transform(args.first(), "toUpper", str::to_uppercase),
"tolower" | "lower" => string_transform(args.first(), "toLower", str::to_lowercase),
"trim" => string_transform(args.first(), "trim", |s| s.trim().to_string()),
"ltrim" => string_transform(args.first(), "ltrim", |s| s.trim_start().to_string()),
"rtrim" => string_transform(args.first(), "rtrim", |s| s.trim_end().to_string()),
"reverse" => reverse_builtin(args.first()),
"replace" => replace_builtin(args),
"split" => split_builtin(args),
"substring" => substring_builtin(args),
"left" => left_right_builtin(args, true),
"right" => left_right_builtin(args, false),
"tofloat" => match args.first() {
Some(v) => to_float(v),
None => Ok(Value::Null),
},
"toboolean" => match args.first() {
Some(v) => to_boolean(v),
None => Ok(Value::Null),
},
"abs" => match args.first() {
Some(Value::Property(PropertyValue::Int(i)))
| Some(Value::Literal(Literal::Int(i))) => {
Ok(Value::Property(PropertyValue::Int(i.abs())))
}
Some(Value::Null) | None => Ok(Value::Null),
Some(other) => match value_as_f64(other) {
Some(f) => Ok(Value::Property(PropertyValue::Float(f.abs()))),
None => Err(QueryError::Type(format!(
"abs() expects a number, got {other:?}"
))),
},
},
"ceil" => float_math_fn(args.first(), "ceil", f64::ceil),
"floor" => float_math_fn(args.first(), "floor", f64::floor),
"round" => float_math_fn(args.first(), "round", f64::round),
"sqrt" => float_math_fn(args.first(), "sqrt", f64::sqrt),
"sign" => match args.first() {
Some(Value::Null) | None => Ok(Value::Null),
Some(other) => match value_as_f64(other) {
Some(f) => Ok(Value::Property(PropertyValue::Int(if f > 0.0 {
1
} else if f < 0.0 {
-1
} else {
0
}))),
None => Err(QueryError::Type(format!(
"sign() expects a number, got {other:?}"
))),
},
},
"rand" => Ok(Value::Property(PropertyValue::Float(rand_f64()))),
other => Err(QueryError::Semantic(format!("unknown function: {other}"))),
}
}
fn rand_f64() -> f64 {
use std::collections::hash_map::RandomState;
use std::hash::{BuildHasher, Hasher};
use std::sync::atomic::{AtomicU64, Ordering};
static COUNTER: AtomicU64 = AtomicU64::new(0);
let mut hasher = RandomState::new().build_hasher();
hasher.write_u64(COUNTER.fetch_add(1, Ordering::Relaxed));
let bits = hasher.finish();
(bits >> 11) as f64 / (1u64 << 53) as f64
}
fn keys_builtin(arg: Option<&Value>) -> Result<Value, QueryError> {
Ok(match arg {
Some(Value::Node(n)) => Value::List(
n.props
.keys()
.map(|k| Value::Property(PropertyValue::String(k.clone())))
.collect(),
),
Some(Value::Edge(e)) => Value::List(
e.props
.keys()
.map(|k| Value::Property(PropertyValue::String(k.clone())))
.collect(),
),
Some(Value::Map(m)) => Value::List(
m.keys()
.map(|k| Value::Property(PropertyValue::String(k.clone())))
.collect(),
),
Some(Value::Null) | None => Value::Null,
Some(other) => {
return Err(QueryError::Type(format!(
"keys() expects a node, relationship, or map, got {other:?}"
)))
}
})
}
fn labels_builtin(arg: Option<&Value>) -> Result<Value, QueryError> {
Ok(match arg {
Some(Value::Node(n)) => Value::List(
n.labels
.iter()
.map(|l| Value::Property(PropertyValue::String(l.clone())))
.collect(),
),
Some(Value::Null) | None => Value::Null,
Some(other) => {
return Err(QueryError::Type(format!(
"labels() expects a node, got {other:?}"
)))
}
})
}
fn type_builtin(arg: Option<&Value>) -> Result<Value, QueryError> {
Ok(match arg {
Some(Value::Edge(e)) => Value::Property(PropertyValue::String(e.label.clone())),
Some(Value::Null) | None => Value::Null,
Some(other) => {
return Err(QueryError::Type(format!(
"type() expects a relationship, got {other:?}"
)))
}
})
}
fn properties_builtin(arg: Option<&Value>) -> Result<Value, QueryError> {
Ok(match arg {
Some(Value::Node(n)) => Value::Map(
n.props
.iter()
.map(|(k, v)| (k.clone(), property_value_to_value(v.clone())))
.collect(),
),
Some(Value::Edge(e)) => Value::Map(
e.props
.iter()
.map(|(k, v)| (k.clone(), property_value_to_value(v.clone())))
.collect(),
),
Some(Value::Map(m)) => Value::Map(m.clone()),
Some(Value::Null) | None => Value::Null,
Some(other) => {
return Err(QueryError::Type(format!(
"properties() expects a node, relationship, or map, got {other:?}"
)))
}
})
}
fn id_builtin(arg: Option<&Value>) -> Result<Value, QueryError> {
Ok(match arg {
Some(Value::Node(n)) => Value::Property(PropertyValue::Int(n.id.0 as i64)),
Some(Value::Edge(e)) => Value::Property(PropertyValue::Int(e.id.0 as i64)),
Some(Value::Null) | None => Value::Null,
Some(other) => {
return Err(QueryError::Type(format!(
"id() expects a node or relationship, got {other:?}"
)))
}
})
}
fn size_builtin(arg: Option<&Value>) -> Result<Value, QueryError> {
Ok(match arg {
Some(Value::List(items)) => Value::Property(PropertyValue::Int(items.len() as i64)),
Some(Value::Null) | None => Value::Null,
Some(other) => match as_arith_str(other) {
Some(s) => Value::Property(PropertyValue::Int(s.chars().count() as i64)),
None => {
return Err(QueryError::Type(format!(
"size() expects a list or string, got {other:?}"
)))
}
},
})
}
fn nodes_builtin(arg: Option<&Value>) -> Result<Value, QueryError> {
Ok(match arg {
Some(Value::Path(elems)) => Value::List(
elems
.iter()
.filter_map(|e| match e {
PathElem::Node(n) => Some(Value::Node(n.clone())),
PathElem::Edge(_) => None,
})
.collect(),
),
Some(Value::Null) | None => Value::Null,
Some(other) => {
return Err(QueryError::Type(format!(
"nodes() expects a path, got {other:?}"
)))
}
})
}
fn relationships_builtin(arg: Option<&Value>) -> Result<Value, QueryError> {
Ok(match arg {
Some(Value::Path(elems)) => Value::List(
elems
.iter()
.filter_map(|e| match e {
PathElem::Edge(e) => Some(Value::Edge(e.clone())),
PathElem::Node(_) => None,
})
.collect(),
),
Some(Value::Null) | None => Value::Null,
Some(other) => {
return Err(QueryError::Type(format!(
"relationships() expects a path, got {other:?}"
)))
}
})
}
fn list_edge_builtin(
arg: Option<&Value>,
fn_name: &str,
pick: impl Fn(&[Value]) -> Option<Value>,
) -> Result<Value, QueryError> {
Ok(match arg {
Some(Value::List(items)) => pick(items).unwrap_or(Value::Null),
Some(Value::Null) | None => Value::Null,
Some(other) => {
return Err(QueryError::Type(format!(
"{fn_name}() expects a list, got {other:?}"
)))
}
})
}
fn range_builtin(args: &[Value]) -> Result<Value, QueryError> {
let int_arg = |v: &Value, which: &str| -> Result<i64, QueryError> {
value_as_i64(v).ok_or_else(|| {
QueryError::Type(format!("range()'s {which} must be an integer, got {v:?}"))
})
};
let start = int_arg(
args.first()
.ok_or_else(|| QueryError::Semantic("range() requires at least 2 arguments".into()))?,
"start",
)?;
let end = int_arg(
args.get(1)
.ok_or_else(|| QueryError::Semantic("range() requires at least 2 arguments".into()))?,
"end",
)?;
let step = match args.get(2) {
Some(v) => int_arg(v, "step")?,
None => 1,
};
if step == 0 {
return Err(QueryError::Type("range()'s step can't be 0".into()));
}
let mut out = Vec::new();
let mut i = start;
if step > 0 {
while i <= end {
out.push(Value::Property(PropertyValue::Int(i)));
i += step;
}
} else {
while i >= end {
out.push(Value::Property(PropertyValue::Int(i)));
i += step;
}
}
Ok(Value::List(out))
}
fn string_transform(
arg: Option<&Value>,
fn_name: &str,
f: impl FnOnce(&str) -> String,
) -> Result<Value, QueryError> {
Ok(match arg {
Some(Value::Null) | None => Value::Null,
Some(other) => match as_arith_str(other) {
Some(s) => Value::Property(PropertyValue::String(f(s))),
None => {
return Err(QueryError::Type(format!(
"{fn_name}() expects a string, got {other:?}"
)))
}
},
})
}
fn reverse_builtin(arg: Option<&Value>) -> Result<Value, QueryError> {
Ok(match arg {
Some(Value::Null) | None => Value::Null,
Some(Value::List(items)) => Value::List(items.iter().rev().cloned().collect()),
Some(other) => match as_arith_str(other) {
Some(s) => Value::Property(PropertyValue::String(s.chars().rev().collect())),
None => {
return Err(QueryError::Type(format!(
"reverse() expects a string or list, got {other:?}"
)))
}
},
})
}
fn replace_str_arg<'a>(v: &'a Value, which: &str) -> Result<&'a str, QueryError> {
as_arith_str(v)
.ok_or_else(|| QueryError::Type(format!("replace()'s {which} must be a string, got {v:?}")))
}
fn replace_builtin(args: &[Value]) -> Result<Value, QueryError> {
if args.iter().any(|v| matches!(v, Value::Null)) {
return Ok(Value::Null);
}
let original = replace_str_arg(
args.first()
.ok_or_else(|| QueryError::Semantic("replace() requires 3 arguments".into()))?,
"original",
)?;
let search = replace_str_arg(
args.get(1)
.ok_or_else(|| QueryError::Semantic("replace() requires 3 arguments".into()))?,
"search",
)?;
let replacement = replace_str_arg(
args.get(2)
.ok_or_else(|| QueryError::Semantic("replace() requires 3 arguments".into()))?,
"replacement",
)?;
Ok(Value::Property(PropertyValue::String(
original.replace(search, replacement),
)))
}
fn split_builtin(args: &[Value]) -> Result<Value, QueryError> {
if args.iter().any(|v| matches!(v, Value::Null)) {
return Ok(Value::Null);
}
let s = args
.first()
.and_then(as_arith_str)
.ok_or_else(|| QueryError::Type("split()'s first argument must be a string".into()))?;
let delim = args
.get(1)
.and_then(as_arith_str)
.ok_or_else(|| QueryError::Type("split()'s second argument must be a string".into()))?;
let parts = if delim.is_empty() {
s.split("").filter(|p| !p.is_empty()).collect::<Vec<_>>()
} else {
s.split(delim).collect::<Vec<_>>()
};
Ok(Value::List(
parts
.into_iter()
.map(|p| Value::Property(PropertyValue::String(p.to_string())))
.collect(),
))
}
fn substring_builtin(args: &[Value]) -> Result<Value, QueryError> {
if matches!(args.first(), Some(Value::Null)) {
return Ok(Value::Null);
}
let s = args
.first()
.and_then(as_arith_str)
.ok_or_else(|| QueryError::Type("substring()'s first argument must be a string".into()))?;
let chars: Vec<char> = s.chars().collect();
let start = args
.get(1)
.and_then(value_as_i64)
.ok_or_else(|| QueryError::Type("substring()'s start must be an integer".into()))?
.max(0) as usize;
let start = start.min(chars.len());
let end = match args.get(2) {
Some(v) => {
let len = value_as_i64(v)
.ok_or_else(|| QueryError::Type("substring()'s length must be an integer".into()))?
.max(0) as usize;
(start + len).min(chars.len())
}
None => chars.len(),
};
Ok(Value::Property(PropertyValue::String(
chars[start..end].iter().collect(),
)))
}
fn left_right_builtin(args: &[Value], from_left: bool) -> Result<Value, QueryError> {
if matches!(args.first(), Some(Value::Null)) {
return Ok(Value::Null);
}
let fn_name = if from_left { "left" } else { "right" };
let s = args.first().and_then(as_arith_str).ok_or_else(|| {
QueryError::Type(format!("{fn_name}()'s first argument must be a string"))
})?;
let n = args
.get(1)
.and_then(value_as_i64)
.ok_or_else(|| {
QueryError::Type(format!("{fn_name}()'s second argument must be an integer"))
})?
.max(0) as usize;
let chars: Vec<char> = s.chars().collect();
let n = n.min(chars.len());
let slice = if from_left {
&chars[..n]
} else {
&chars[chars.len() - n..]
};
Ok(Value::Property(PropertyValue::String(
slice.iter().collect(),
)))
}
fn float_math_fn(
arg: Option<&Value>,
fn_name: &str,
f: impl FnOnce(f64) -> f64,
) -> Result<Value, QueryError> {
Ok(match arg {
Some(Value::Null) | None => Value::Null,
Some(other) => match value_as_f64(other) {
Some(x) => Value::Property(PropertyValue::Float(f(x))),
None => {
return Err(QueryError::Type(format!(
"{fn_name}() expects a number, got {other:?}"
)))
}
},
})
}
fn to_float(v: &Value) -> Result<Value, QueryError> {
Ok(match v {
Value::Property(PropertyValue::Int(i)) => Value::Property(PropertyValue::Float(*i as f64)),
Value::Property(PropertyValue::Float(f)) => Value::Property(PropertyValue::Float(*f)),
Value::Literal(Literal::Int(i)) => Value::Property(PropertyValue::Float(*i as f64)),
Value::Literal(Literal::Float(f)) => Value::Property(PropertyValue::Float(*f)),
Value::Property(PropertyValue::String(s)) | Value::Literal(Literal::String(s)) => {
match s.trim().parse::<f64>() {
Ok(f) => Value::Property(PropertyValue::Float(f)),
Err(_) => Value::Null,
}
}
Value::Property(PropertyValue::Null) | Value::Literal(Literal::Null) | Value::Null => {
Value::Null
}
Value::Literal(Literal::Param(name)) => {
unreachable!("param ${name} must be substituted before execution — see params::substitute_params")
}
other => {
return Err(QueryError::Type(format!(
"toFloat() cannot convert {other:?} to a float"
)))
}
})
}
fn to_boolean(v: &Value) -> Result<Value, QueryError> {
Ok(match v {
Value::Property(PropertyValue::Bool(b)) | Value::Literal(Literal::Bool(b)) => {
Value::Literal(Literal::Bool(*b))
}
Value::Property(PropertyValue::String(s)) | Value::Literal(Literal::String(s)) => {
match s.trim().to_ascii_lowercase().as_str() {
"true" => Value::Literal(Literal::Bool(true)),
"false" => Value::Literal(Literal::Bool(false)),
_ => Value::Null,
}
}
Value::Property(PropertyValue::Null) | Value::Literal(Literal::Null) | Value::Null => {
Value::Null
}
Value::Literal(Literal::Param(name)) => {
unreachable!("param ${name} must be substituted before execution — see params::substitute_params")
}
other => {
return Err(QueryError::Type(format!(
"toBoolean() cannot convert {other:?} to a boolean"
)))
}
})
}
fn item_truthy(v: &Value) -> Option<bool> {
match v {
Value::Null => None,
Value::Literal(Literal::Bool(b)) | Value::Property(PropertyValue::Bool(b)) => Some(*b),
_ => Some(false),
}
}
fn eval_quantifier(kind: QuantifierKind, preds: &[Option<bool>]) -> Option<bool> {
let true_count = preds.iter().filter(|p| **p == Some(true)).count();
let any_false = preds.contains(&Some(false));
let any_null = preds.iter().any(|p| p.is_none());
match kind {
QuantifierKind::Any => {
if true_count > 0 {
Some(true)
} else if any_null {
None
} else {
Some(false)
}
}
QuantifierKind::None => {
if true_count > 0 {
Some(false)
} else if any_null {
None
} else {
Some(true)
}
}
QuantifierKind::All => {
if any_false {
Some(false)
} else if any_null {
None
} else {
Some(true)
}
}
QuantifierKind::Single => {
if true_count >= 2 {
Some(false)
} else if any_null {
None
} else {
Some(true_count == 1)
}
}
}
}
fn to_integer(v: &Value) -> Result<Value, QueryError> {
let as_str_parse = |s: &str| match s.trim().parse::<i64>() {
Ok(i) => Value::Property(PropertyValue::Int(i)),
Err(_) => match s.trim().parse::<f64>() {
Ok(f) => Value::Property(PropertyValue::Int(f as i64)),
Err(_) => Value::Null,
},
};
Ok(match v {
Value::Property(PropertyValue::Int(i)) => Value::Property(PropertyValue::Int(*i)),
Value::Property(PropertyValue::Float(f)) => Value::Property(PropertyValue::Int(*f as i64)),
Value::Property(PropertyValue::String(s)) => as_str_parse(s),
Value::Literal(Literal::Int(i)) => Value::Property(PropertyValue::Int(*i)),
Value::Literal(Literal::Float(f)) => Value::Property(PropertyValue::Int(*f as i64)),
Value::Literal(Literal::String(s)) => as_str_parse(s),
Value::Property(PropertyValue::Bool(_) | PropertyValue::Null)
| Value::Literal(Literal::Bool(_) | Literal::Null)
| Value::Null => Value::Null,
Value::Literal(Literal::Param(name)) => {
unreachable!("param ${name} must be substituted before execution — see params::substitute_params")
}
Value::Property(
PropertyValue::Date(_)
| PropertyValue::Duration { .. }
| PropertyValue::LocalTime(_)
| PropertyValue::Time { .. }
| PropertyValue::LocalDateTime { .. }
| PropertyValue::DateTime { .. }
| PropertyValue::List(_)
| PropertyValue::Map(_),
)
| Value::Node(_)
| Value::Edge(_)
| Value::List(_)
| Value::Map(_)
| Value::Path(_) => {
return Err(QueryError::Type(format!(
"toInteger() cannot convert {v:?} to an integer"
)))
}
})
}
fn to_string_value(v: &Value) -> Result<Value, QueryError> {
let s = match v {
Value::Property(PropertyValue::String(s)) | Value::Literal(Literal::String(s)) => s.clone(),
Value::Property(PropertyValue::Int(i)) | Value::Literal(Literal::Int(i)) => i.to_string(),
Value::Property(PropertyValue::Float(f)) | Value::Literal(Literal::Float(f)) => {
f.to_string()
}
Value::Property(PropertyValue::Bool(b)) | Value::Literal(Literal::Bool(b)) => b.to_string(),
Value::Property(PropertyValue::Date(d)) => temporal::format_date(*d),
Value::Property(PropertyValue::Duration {
months,
days,
seconds,
nanos,
}) => temporal::format_duration(*months, *days, *seconds, *nanos),
Value::Property(PropertyValue::LocalTime(nanos_of_day)) => {
temporal::format_local_time(*nanos_of_day)
}
Value::Property(PropertyValue::Time {
nanos_of_day,
offset_seconds,
}) => temporal::format_time(*nanos_of_day, *offset_seconds),
Value::Property(PropertyValue::LocalDateTime {
epoch_seconds,
nanos,
}) => temporal::format_local_date_time(*epoch_seconds, *nanos),
Value::Property(PropertyValue::DateTime {
epoch_seconds,
nanos,
zone,
}) => temporal::format_date_time(*epoch_seconds, *nanos, &tz_from_graph(zone)),
Value::Property(PropertyValue::Null) | Value::Literal(Literal::Null) | Value::Null => {
return Ok(Value::Null);
}
Value::Literal(Literal::Param(name)) => {
unreachable!("param ${name} must be substituted before execution — see params::substitute_params")
}
Value::Property(PropertyValue::List(_) | PropertyValue::Map(_))
| Value::Node(_)
| Value::Edge(_)
| Value::List(_)
| Value::Map(_)
| Value::Path(_) => {
return Err(QueryError::Type(format!(
"toString() cannot convert {v:?} to a string"
)))
}
};
Ok(Value::Property(PropertyValue::String(s)))
}
fn now_or_null(args: &[Value], now_value: impl FnOnce() -> Value) -> Value {
if matches!(args.first(), Some(Value::Null)) {
Value::Null
} else {
now_value()
}
}
fn date_builtin(args: &[Value], now: temporal::NowSnapshot) -> Result<Value, QueryError> {
if args.len() > 1 {
return Err(QueryError::Semantic(format!(
"date() expects zero or one argument, got {}",
args.len()
)));
}
let Some(arg) = args.first() else {
return Ok(Value::Property(PropertyValue::Date(now.epoch_day)));
};
if matches!(arg, Value::Null) {
return Ok(Value::Null);
}
if let Value::Property(PropertyValue::Date(d)) = arg {
return Ok(Value::Property(PropertyValue::Date(*d)));
}
if matches!(
arg,
Value::Property(PropertyValue::LocalDateTime { .. } | PropertyValue::DateTime { .. })
) {
let epoch_day = extract_date_base_epoch_day("date() argument", arg)?;
return Ok(Value::Property(PropertyValue::Date(epoch_day)));
}
if let Some(s) = as_arith_str(arg) {
let d = temporal::parse_date(s).ok_or_else(|| {
QueryError::Type(format!(
"'{s}' isn't a date string MarsDB can parse -- only the calendar forms YYYY-MM-DD/YYYYMMDD/\
YYYY-MM/YYYYMM/YYYY, week-date forms YYYY-Www[-D]/YYYYWww[D], and ordinal-date forms \
YYYY-DDD/YYYYDDD are supported"
))
})?;
return Ok(Value::Property(PropertyValue::Date(d)));
}
if let Value::Map(m) = arg {
return Ok(Value::Property(PropertyValue::Date(date_from_map(m)?)));
}
Err(QueryError::Type(format!(
"date() doesn't support this argument: {arg:?}"
)))
}
fn extract_date_base_epoch_day(key: &str, v: &Value) -> Result<i32, QueryError> {
match v {
Value::Property(PropertyValue::Date(d)) => Ok(*d),
Value::Property(PropertyValue::LocalDateTime { epoch_seconds, .. }) => {
Ok(temporal::split_epoch_seconds(*epoch_seconds).0)
}
Value::Property(PropertyValue::DateTime {
epoch_seconds,
zone,
..
}) => {
let offset_seconds = temporal::resolve_offset(&tz_from_graph(zone), *epoch_seconds);
Ok(temporal::split_epoch_seconds(epoch_seconds + offset_seconds as i64).0)
}
other => Err(QueryError::Type(format!(
"'{key}' must be a Date, LocalDateTime, or DateTime, got {other:?}"
))),
}
}
type ClockBase = (i64, i64, i64, i64, Option<(temporal::TzId, i32)>);
fn extract_time_base(key: &str, v: &Value) -> Result<ClockBase, QueryError> {
let hms_nanos = |nanos_of_day: i64| {
(
temporal::local_time_component(nanos_of_day, "hour").unwrap(),
temporal::local_time_component(nanos_of_day, "minute").unwrap(),
temporal::local_time_component(nanos_of_day, "second").unwrap(),
temporal::local_time_component(nanos_of_day, "nanosecond").unwrap(),
)
};
match v {
Value::Property(PropertyValue::LocalTime(n)) => {
let (h, m, s, ns) = hms_nanos(*n);
Ok((h, m, s, ns, None))
}
Value::Property(PropertyValue::Time {
nanos_of_day,
offset_seconds,
}) => {
let (h, m, s, ns) = hms_nanos(*nanos_of_day);
Ok((
h,
m,
s,
ns,
Some((temporal::TzId::Offset(*offset_seconds), *offset_seconds)),
))
}
Value::Property(PropertyValue::LocalDateTime {
epoch_seconds,
nanos,
}) => {
let (_, nanos_of_day) = temporal::split_epoch_seconds(*epoch_seconds);
let (h, m, s, _) = hms_nanos(nanos_of_day);
Ok((h, m, s, *nanos as i64, None))
}
Value::Property(PropertyValue::DateTime {
epoch_seconds,
nanos,
zone,
}) => {
let tz = tz_from_graph(zone);
let offset_seconds = temporal::resolve_offset(&tz, *epoch_seconds);
let local = epoch_seconds + offset_seconds as i64;
let (_, nanos_of_day) = temporal::split_epoch_seconds(local);
let (h, m, s, _) = hms_nanos(nanos_of_day);
Ok((h, m, s, *nanos as i64, Some((tz, offset_seconds))))
}
other => Err(QueryError::Type(format!(
"'{key}' must be a LocalTime, Time, LocalDateTime, or DateTime, got {other:?}"
))),
}
}
const DATE_ALLOWED_KEYS: &[&str] = &[
"year",
"month",
"day",
"week",
"dayOfWeek",
"ordinalDay",
"quarter",
"dayOfQuarter",
"date",
];
fn date_from_map(m: &BTreeMap<String, Value>) -> Result<i32, QueryError> {
let (year, month, day) = calendar_fields_from_map("date", m, DATE_ALLOWED_KEYS)?;
temporal::epoch_day_from_ymd(year, month, day).ok_or_else(|| {
QueryError::Type(format!(
"{year:04}-{month:02}-{day:02} isn't a valid calendar date"
))
})
}
fn calendar_fields_from_map(
caller: &str,
m: &BTreeMap<String, Value>,
allowed: &[&str],
) -> Result<(i32, u32, u32), QueryError> {
if let Some(bad) = m.keys().find(|k| !allowed.contains(&k.as_str())) {
return Err(QueryError::Type(format!(
"{caller}({{...}}) key '{bad}' isn't a recognized field"
)));
}
let int_field = |key: &str, value: &Value| {
value_as_i64(value).ok_or_else(|| {
QueryError::Type(format!("{caller}({{...}})'s '{key}' must be an integer"))
})
};
let base_epoch_day = m
.get("date")
.map(|v| ("date", v))
.or_else(|| m.get("datetime").map(|v| ("datetime", v)))
.map(|(k, v)| extract_date_base_epoch_day(k, v))
.transpose()?;
let epoch_day_from_component =
|prop: &str| base_epoch_day.map(|ed| temporal::date_component(ed, prop).unwrap());
if m.contains_key("week") || m.contains_key("dayOfWeek") {
let week_year = match m.get("year") {
Some(v) => i32::try_from(int_field("year", v)?).map_err(|_| {
QueryError::Type(format!("{caller}({{...}})'s 'year' is out of range"))
})?,
None => i32::try_from(epoch_day_from_component("weekYear").ok_or_else(|| {
QueryError::Type(format!("{caller}({{...}}) requires a 'year' key"))
})?)
.unwrap(),
};
let week = match m.get("week") {
Some(v) => u32::try_from(int_field("week", v)?).map_err(|_| {
QueryError::Type(format!("{caller}({{...}})'s 'week' is out of range"))
})?,
None => u32::try_from(epoch_day_from_component("week").ok_or_else(|| {
QueryError::Type(format!("{caller}({{...}}) requires a 'week' key"))
})?)
.unwrap(),
};
let day_of_week = match m.get("dayOfWeek") {
Some(v) => int_field("dayOfWeek", v)?,
None => epoch_day_from_component("dayOfWeek").unwrap_or(1),
};
let epoch_day = temporal::epoch_day_from_week_fields(week_year, week, day_of_week)
.ok_or_else(|| {
QueryError::Type(format!(
"{caller}({{...}}) has an out-of-range week-date field"
))
})?;
return Ok((
temporal::date_component(epoch_day, "year").unwrap() as i32,
temporal::date_component(epoch_day, "month").unwrap() as u32,
temporal::date_component(epoch_day, "day").unwrap() as u32,
));
}
if m.contains_key("ordinalDay") {
let year = match m.get("year") {
Some(v) => i32::try_from(int_field("year", v)?).map_err(|_| {
QueryError::Type(format!("{caller}({{...}})'s 'year' is out of range"))
})?,
None => i32::try_from(epoch_day_from_component("year").ok_or_else(|| {
QueryError::Type(format!("{caller}({{...}}) requires a 'year' key"))
})?)
.unwrap(),
};
let ordinal_raw = int_field("ordinalDay", m.get("ordinalDay").unwrap())?;
let ordinal_day = u32::try_from(ordinal_raw).map_err(|_| {
QueryError::Type(format!("{caller}({{...}})'s 'ordinalDay' is out of range"))
})?;
let epoch_day =
temporal::epoch_day_from_ordinal_fields(year, ordinal_day).ok_or_else(|| {
QueryError::Type(format!(
"{caller}({{...}}) has an out-of-range ordinalDay field"
))
})?;
return Ok((
year,
temporal::date_component(epoch_day, "month").unwrap() as u32,
temporal::date_component(epoch_day, "day").unwrap() as u32,
));
}
if m.contains_key("quarter") || m.contains_key("dayOfQuarter") {
let year = match m.get("year") {
Some(v) => i32::try_from(int_field("year", v)?).map_err(|_| {
QueryError::Type(format!("{caller}({{...}})'s 'year' is out of range"))
})?,
None => i32::try_from(epoch_day_from_component("year").ok_or_else(|| {
QueryError::Type(format!("{caller}({{...}}) requires a 'year' key"))
})?)
.unwrap(),
};
let quarter = match m.get("quarter") {
Some(v) => u32::try_from(int_field("quarter", v)?).map_err(|_| {
QueryError::Type(format!("{caller}({{...}})'s 'quarter' is out of range"))
})?,
None => u32::try_from(epoch_day_from_component("quarter").ok_or_else(|| {
QueryError::Type(format!("{caller}({{...}}) requires a 'quarter' key"))
})?)
.unwrap(),
};
let day_of_quarter = match m.get("dayOfQuarter") {
Some(v) => int_field("dayOfQuarter", v)?,
None => epoch_day_from_component("dayOfQuarter").unwrap_or(1),
};
let epoch_day = temporal::epoch_day_from_quarter_fields(year, quarter, day_of_quarter)
.ok_or_else(|| {
QueryError::Type(format!(
"{caller}({{...}}) has an out-of-range quarter-date field"
))
})?;
return Ok((
year,
temporal::date_component(epoch_day, "month").unwrap() as u32,
temporal::date_component(epoch_day, "day").unwrap() as u32,
));
}
let year_raw = match m.get("year") {
Some(v) => int_field("year", v)?,
None => epoch_day_from_component("year")
.ok_or_else(|| QueryError::Type(format!("{caller}({{...}}) requires a 'year' key")))?,
};
let year = i32::try_from(year_raw).map_err(|_| {
QueryError::Type(format!(
"{caller}({{...}})'s 'year' is out of range: {year_raw}"
))
})?;
let month_raw = match m.get("month") {
Some(v) => int_field("month", v)?,
None => epoch_day_from_component("month").unwrap_or(1),
};
let month = u32::try_from(month_raw).map_err(|_| {
QueryError::Type(format!(
"{caller}({{...}})'s 'month' is out of range: {month_raw}"
))
})?;
let day_raw = match m.get("day") {
Some(v) => int_field("day", v)?,
None => epoch_day_from_component("day").unwrap_or(1),
};
let day = u32::try_from(day_raw).map_err(|_| {
QueryError::Type(format!(
"{caller}({{...}})'s 'day' is out of range: {day_raw}"
))
})?;
Ok((year, month, day))
}
fn duration_builtin(args: &[Value]) -> Result<Value, QueryError> {
if args.len() != 1 {
return Err(QueryError::Semantic(format!(
"duration() expects exactly one argument, got {}",
args.len()
)));
}
let arg = &args[0];
if matches!(arg, Value::Null) {
return Ok(Value::Null);
}
let (months, days, seconds, nanos) = if let Some(s) = as_arith_str(arg) {
temporal::parse_duration(s).ok_or_else(|| {
QueryError::Type(format!(
"'{s}' isn't a duration string MarsDB can parse -- only ISO-8601 'PnYnMnWnDTnHnMnS' text is \
supported, not the alternate combined date-time duration syntax"
))
})?
} else if let Value::Map(m) = arg {
temporal::normalize_duration(duration_fields_from_map(m)?)
} else {
return Err(QueryError::Type(format!(
"duration() doesn't support this argument: {arg:?}"
)));
};
Ok(Value::Property(PropertyValue::Duration {
months,
days,
seconds,
nanos,
}))
}
fn duration_fields_from_map(
m: &BTreeMap<String, Value>,
) -> Result<temporal::DurationFields, QueryError> {
const ALLOWED: &[&str] = &[
"years",
"months",
"weeks",
"days",
"hours",
"minutes",
"seconds",
"milliseconds",
"microseconds",
"nanoseconds",
];
if let Some(bad) = m.keys().find(|k| !ALLOWED.contains(&k.as_str())) {
return Err(QueryError::Type(format!(
"duration({{...}}) key '{bad}' isn't a recognized duration unit"
)));
}
let field = |key: &str| -> Result<f64, QueryError> {
match m.get(key) {
None => Ok(0.0),
Some(v) => value_as_f64(v).ok_or_else(|| {
QueryError::Type(format!("duration({{...}})'s '{key}' must be a number"))
}),
}
};
Ok(temporal::DurationFields {
years: field("years")?,
months: field("months")?,
weeks: field("weeks")?,
days: field("days")?,
hours: field("hours")?,
minutes: field("minutes")?,
seconds: field("seconds")?,
milliseconds: field("milliseconds")?,
microseconds: field("microseconds")?,
nanoseconds: field("nanoseconds")?,
})
}
fn sub_second_nanos_from_map(
base_fraction_ns: i64,
m: &BTreeMap<String, Value>,
) -> Result<i64, QueryError> {
let base_ms = base_fraction_ns / 1_000_000;
let base_us = (base_fraction_ns / 1_000) % 1000;
let base_ns = base_fraction_ns % 1000;
let ms = int_field(m, "millisecond", base_ms)?;
let us = int_field(m, "microsecond", base_us)?;
let ns = int_field(m, "nanosecond", base_ns)?;
Ok(ms * 1_000_000 + us * 1_000 + ns)
}
fn int_field(m: &BTreeMap<String, Value>, key: &str, default: i64) -> Result<i64, QueryError> {
match m.get(key) {
None => Ok(default),
Some(v) => {
value_as_i64(v).ok_or_else(|| QueryError::Type(format!("'{key}' must be an integer")))
}
}
}
fn clock_fields_from_map(
m: &BTreeMap<String, Value>,
epoch_day: Option<i32>,
) -> Result<ClockBase, QueryError> {
let (base_h, base_m, base_s, base_ns, base_zone) = if let Some(v) = m.get("time") {
extract_time_base("time", v)?
} else if let Some(v) = m.get("datetime") {
extract_time_base("datetime", v)?
} else {
(0, 0, 0, 0, None)
};
let has_explicit_timezone = m.contains_key("timezone");
let effective_zone = match m.get("timezone") {
Some(v) => Some(timezone_value_to_tzid(v)?),
None => base_zone.as_ref().map(|(tz, _)| tz.clone()),
};
let base_nanos_of_day =
base_h * 3_600_000_000_000 + base_m * 60_000_000_000 + base_s * 1_000_000_000 + base_ns;
let (base_h, base_m, base_s, base_ns, effective_offset) = if has_explicit_timezone {
let from_offset = match base_zone.as_ref() {
Some((temporal::TzId::Offset(o), _)) => Some(*o),
Some((zone @ temporal::TzId::Named(_), resolved)) => Some(match epoch_day {
Some(ed) => temporal::resolve_offset(
zone,
temporal::combine_epoch_day_and_nanos_of_day(ed, base_nanos_of_day),
),
None => *resolved,
}),
None => None,
};
let to_offset = match (from_offset, effective_zone.as_ref(), epoch_day) {
(Some(_), Some(temporal::TzId::Offset(to)), _) => Some(*to),
(Some(from), Some(zone @ temporal::TzId::Named(_)), Some(ed)) => {
let approx_epoch_seconds =
temporal::combine_epoch_day_and_nanos_of_day(ed, base_nanos_of_day)
- from as i64;
Some(temporal::resolve_offset(zone, approx_epoch_seconds))
}
_ => None,
};
match (from_offset, to_offset) {
(Some(from), Some(to)) if from != to => {
let shifted = (base_nanos_of_day + (to - from) as i64 * 1_000_000_000)
.rem_euclid(86_400_000_000_000);
(
shifted / 3_600_000_000_000,
(shifted / 60_000_000_000) % 60,
(shifted / 1_000_000_000) % 60,
shifted % 1_000_000_000,
to_offset.unwrap_or(0),
)
}
_ => (base_h, base_m, base_s, base_ns, to_offset.unwrap_or(0)),
}
} else {
(
base_h,
base_m,
base_s,
base_ns,
base_zone.as_ref().map_or(0, |(_, o)| *o),
)
};
Ok((
int_field(m, "hour", base_h)?,
int_field(m, "minute", base_m)?,
int_field(m, "second", base_s)?,
sub_second_nanos_from_map(base_ns, m)?,
effective_zone.map(|z| (z, effective_offset)),
))
}
fn local_time_builtin(args: &[Value], now: temporal::NowSnapshot) -> Result<Value, QueryError> {
if args.len() > 1 {
return Err(QueryError::Semantic(format!(
"localtime() expects zero or one argument, got {}",
args.len()
)));
}
let Some(arg) = args.first() else {
return Ok(Value::Property(PropertyValue::LocalTime(now.nanos_of_day)));
};
if matches!(arg, Value::Null) {
return Ok(Value::Null);
}
if let Value::Property(PropertyValue::LocalTime(t)) = arg {
return Ok(Value::Property(PropertyValue::LocalTime(*t)));
}
if matches!(
arg,
Value::Property(
PropertyValue::Time { .. }
| PropertyValue::LocalDateTime { .. }
| PropertyValue::DateTime { .. }
)
) {
let (hour, minute, second, nanos, _) = extract_time_base("localtime() argument", arg)?;
let t = temporal::local_time_nanos_from_fields(hour, minute, second, nanos).ok_or_else(
|| QueryError::Type("localtime() argument has an out-of-range field".into()),
)?;
return Ok(Value::Property(PropertyValue::LocalTime(t)));
}
if let Some(s) = as_arith_str(arg) {
let t = temporal::parse_local_time(s).ok_or_else(|| {
QueryError::Type(format!("'{s}' isn't a local time string MarsDB can parse"))
})?;
return Ok(Value::Property(PropertyValue::LocalTime(t)));
}
if let Value::Map(m) = arg {
const ALLOWED: &[&str] = &[
"hour",
"minute",
"second",
"millisecond",
"microsecond",
"nanosecond",
"time",
];
if let Some(bad) = m.keys().find(|k| !ALLOWED.contains(&k.as_str())) {
return Err(QueryError::Type(format!(
"localtime({{...}}) key '{bad}' isn't a recognized field"
)));
}
let (hour, minute, second, nanos, _) = clock_fields_from_map(m, None)?;
let t = temporal::local_time_nanos_from_fields(hour, minute, second, nanos)
.ok_or_else(|| QueryError::Type("localtime({...}) has an out-of-range field".into()))?;
return Ok(Value::Property(PropertyValue::LocalTime(t)));
}
Err(QueryError::Type(format!(
"localtime() doesn't support this argument: {arg:?}"
)))
}
fn time_builtin(args: &[Value], now: temporal::NowSnapshot) -> Result<Value, QueryError> {
if args.len() > 1 {
return Err(QueryError::Semantic(format!(
"time() expects zero or one argument, got {}",
args.len()
)));
}
let Some(arg) = args.first() else {
return Ok(Value::Property(PropertyValue::Time {
nanos_of_day: now.nanos_of_day,
offset_seconds: 0,
}));
};
if matches!(arg, Value::Null) {
return Ok(Value::Null);
}
if let Value::Property(PropertyValue::Time {
nanos_of_day,
offset_seconds,
}) = arg
{
return Ok(Value::Property(PropertyValue::Time {
nanos_of_day: *nanos_of_day,
offset_seconds: *offset_seconds,
}));
}
if matches!(
arg,
Value::Property(
PropertyValue::LocalTime(_)
| PropertyValue::LocalDateTime { .. }
| PropertyValue::DateTime { .. }
)
) {
let (hour, minute, second, nanos, zone) = extract_time_base("time() argument", arg)?;
let nanos_of_day = temporal::local_time_nanos_from_fields(hour, minute, second, nanos)
.ok_or_else(|| QueryError::Type("time() argument has an out-of-range field".into()))?;
return Ok(Value::Property(PropertyValue::Time {
nanos_of_day,
offset_seconds: zone.map_or(0, |(_, o)| o),
}));
}
if let Some(s) = as_arith_str(arg) {
if s.contains('[') {
return Err(QueryError::Type(
"time('...'): named timezones (e.g. '[Europe/Stockholm]') aren't supported, only a fixed UTC \
offset like '+01:00'"
.into(),
));
}
let (nanos_of_day, offset_seconds) = temporal::parse_time(s).ok_or_else(|| {
QueryError::Type(format!("'{s}' isn't a time string MarsDB can parse"))
})?;
return Ok(Value::Property(PropertyValue::Time {
nanos_of_day,
offset_seconds,
}));
}
if let Value::Map(m) = arg {
const ALLOWED: &[&str] = &[
"hour",
"minute",
"second",
"millisecond",
"microsecond",
"nanosecond",
"timezone",
"time",
];
if let Some(bad) = m.keys().find(|k| !ALLOWED.contains(&k.as_str())) {
return Err(QueryError::Type(format!(
"time({{...}}) key '{bad}' isn't a recognized field"
)));
}
let (hour, minute, second, nanos, zone) = clock_fields_from_map(m, None)?;
let offset_seconds = match zone {
None => 0,
Some((_, o)) if !m.contains_key("timezone") => o,
Some((temporal::TzId::Offset(o), _)) => o,
Some((temporal::TzId::Named(name), _)) => {
return Err(QueryError::Type(format!(
"'timezone': '{name}' looks like a named timezone (e.g. 'Europe/Stockholm') -- TIME has \
no calendar date to resolve a named zone's DST-dependent offset against, only a fixed \
UTC offset like '+01:00' is supported"
)));
}
};
let nanos_of_day = temporal::local_time_nanos_from_fields(hour, minute, second, nanos)
.ok_or_else(|| QueryError::Type("time({...}) has an out-of-range field".into()))?;
return Ok(Value::Property(PropertyValue::Time {
nanos_of_day,
offset_seconds,
}));
}
Err(QueryError::Type(format!(
"time() doesn't support this argument: {arg:?}"
)))
}
fn timezone_value_to_tzid(v: &Value) -> Result<temporal::TzId, QueryError> {
let s = as_arith_str(v).ok_or_else(|| {
QueryError::Type(
"'timezone' must be a string offset or IANA zone name, e.g. '+01:00' or \
'Europe/Stockholm'"
.into(),
)
})?;
if let Some(offset) = temporal::parse_offset_seconds(s) {
return Ok(temporal::TzId::Offset(offset));
}
if temporal::parse_timezone_name(s).is_some() {
return Ok(temporal::TzId::Named(s.to_string()));
}
Err(QueryError::Type(format!(
"'timezone': '{s}' isn't a valid UTC offset or a recognized IANA zone name"
)))
}
fn local_date_time_builtin(
args: &[Value],
now: temporal::NowSnapshot,
) -> Result<Value, QueryError> {
if args.len() > 1 {
return Err(QueryError::Semantic(format!(
"localdatetime() expects zero or one argument, got {}",
args.len()
)));
}
let Some(arg) = args.first() else {
return Ok(Value::Property(PropertyValue::LocalDateTime {
epoch_seconds: now.epoch_seconds,
nanos: now.nanos,
}));
};
if matches!(arg, Value::Null) {
return Ok(Value::Null);
}
if let Value::Property(PropertyValue::LocalDateTime {
epoch_seconds,
nanos,
}) = arg
{
return Ok(Value::Property(PropertyValue::LocalDateTime {
epoch_seconds: *epoch_seconds,
nanos: *nanos,
}));
}
if matches!(arg, Value::Property(PropertyValue::DateTime { .. })) {
let epoch_day = extract_date_base_epoch_day("localdatetime() argument", arg)?;
let year = temporal::date_component(epoch_day, "year").unwrap() as i32;
let month = temporal::date_component(epoch_day, "month").unwrap() as u32;
let day = temporal::date_component(epoch_day, "day").unwrap() as u32;
let (hour, minute, second, nanos, _) = extract_time_base("localdatetime() argument", arg)?;
let (epoch_seconds, nanos) =
temporal::local_date_time_from_fields(temporal::CalendarDateTime {
year,
month,
day,
hour,
minute,
second,
nanos,
})
.ok_or_else(|| {
QueryError::Type("localdatetime() argument has an out-of-range field".into())
})?;
return Ok(Value::Property(PropertyValue::LocalDateTime {
epoch_seconds,
nanos,
}));
}
if let Some(s) = as_arith_str(arg) {
let (epoch_seconds, nanos) = temporal::parse_local_date_time(s).ok_or_else(|| {
QueryError::Type(format!(
"'{s}' isn't a local date-time string MarsDB can parse"
))
})?;
return Ok(Value::Property(PropertyValue::LocalDateTime {
epoch_seconds,
nanos,
}));
}
if let Value::Map(m) = arg {
let (year, month, day) =
calendar_fields_from_map("localdatetime", m, DATE_TIME_ALLOWED_KEYS)?;
let (hour, minute, second, nanos, _) = clock_fields_from_map(m, None)?;
let (epoch_seconds, nanos) =
temporal::local_date_time_from_fields(temporal::CalendarDateTime {
year,
month,
day,
hour,
minute,
second,
nanos,
})
.ok_or_else(|| {
QueryError::Type("localdatetime({...}) has an out-of-range field".into())
})?;
return Ok(Value::Property(PropertyValue::LocalDateTime {
epoch_seconds,
nanos,
}));
}
Err(QueryError::Type(format!(
"localdatetime() doesn't support this argument: {arg:?}"
)))
}
const DATE_TIME_ALLOWED_KEYS: &[&str] = &[
"year",
"month",
"day",
"week",
"dayOfWeek",
"ordinalDay",
"quarter",
"dayOfQuarter",
"hour",
"minute",
"second",
"millisecond",
"microsecond",
"nanosecond",
"timezone",
"date",
"time",
"datetime",
];
fn date_time_builtin(args: &[Value], now: temporal::NowSnapshot) -> Result<Value, QueryError> {
if args.len() > 1 {
return Err(QueryError::Semantic(format!(
"datetime() expects zero or one argument, got {}",
args.len()
)));
}
let Some(arg) = args.first() else {
return Ok(Value::Property(PropertyValue::DateTime {
epoch_seconds: now.epoch_seconds,
nanos: now.nanos,
zone: GraphTzId::Offset(0),
}));
};
if matches!(arg, Value::Null) {
return Ok(Value::Null);
}
if let Value::Property(PropertyValue::DateTime {
epoch_seconds,
nanos,
zone,
}) = arg
{
return Ok(Value::Property(PropertyValue::DateTime {
epoch_seconds: *epoch_seconds,
nanos: *nanos,
zone: zone.clone(),
}));
}
if let Value::Property(PropertyValue::LocalDateTime {
epoch_seconds,
nanos,
}) = arg
{
return Ok(Value::Property(PropertyValue::DateTime {
epoch_seconds: *epoch_seconds,
nanos: *nanos,
zone: GraphTzId::Offset(0),
}));
}
if let Some(s) = as_arith_str(arg) {
let (epoch_seconds, nanos, zone) = temporal::parse_date_time(s).ok_or_else(|| {
QueryError::Type(format!("'{s}' isn't a date-time string MarsDB can parse"))
})?;
return Ok(Value::Property(PropertyValue::DateTime {
epoch_seconds,
nanos,
zone: tz_to_graph(zone),
}));
}
if let Value::Map(m) = arg {
let (year, month, day) = calendar_fields_from_map("datetime", m, DATE_TIME_ALLOWED_KEYS)?;
let epoch_day = temporal::epoch_day_from_ymd(year, month, day);
let (hour, minute, second, nanos, zone) = clock_fields_from_map(m, epoch_day)?;
let zone = zone.map_or(temporal::TzId::Offset(0), |(z, _)| z);
let (epoch_seconds, nanos) = temporal::date_time_from_fields(
temporal::CalendarDateTime {
year,
month,
day,
hour,
minute,
second,
nanos,
},
&zone,
)
.ok_or_else(|| QueryError::Type("datetime({...}) has an out-of-range field".into()))?;
return Ok(Value::Property(PropertyValue::DateTime {
epoch_seconds,
nanos,
zone: tz_to_graph(zone),
}));
}
Err(QueryError::Type(format!(
"datetime() doesn't support this argument: {arg:?}"
)))
}
fn between_operand(name: &str, v: &Value) -> Result<BetweenOperand, QueryError> {
match v {
Value::Property(PropertyValue::Date(d)) => Ok((Some(*d), None, None)),
Value::Property(PropertyValue::LocalTime(n)) => Ok((None, Some(*n), None)),
Value::Property(PropertyValue::Time {
nanos_of_day,
offset_seconds,
}) => Ok((
None,
Some(*nanos_of_day),
Some(temporal::TzId::Offset(*offset_seconds)),
)),
Value::Property(PropertyValue::LocalDateTime {
epoch_seconds,
nanos,
}) => {
let (d, n) = temporal::split_epoch_seconds(*epoch_seconds);
Ok((Some(d), Some(n + *nanos as i64), None))
}
Value::Property(PropertyValue::DateTime {
epoch_seconds,
nanos,
zone,
}) => {
let tz = tz_from_graph(zone);
let offset_seconds = temporal::resolve_offset(&tz, *epoch_seconds);
let local = epoch_seconds + offset_seconds as i64;
let (d, n) = temporal::split_epoch_seconds(local);
Ok((Some(d), Some(n + *nanos as i64), Some(tz)))
}
other => Err(QueryError::Type(format!(
"{name}() needs a Date, LocalTime, Time, LocalDateTime, or DateTime, got {other:?}"
))),
}
}
type BetweenOperand = (Option<i32>, Option<i64>, Option<temporal::TzId>);
type BetweenFn = fn(
Option<i32>,
Option<i64>,
Option<&temporal::TzId>,
Option<i32>,
Option<i64>,
Option<&temporal::TzId>,
) -> temporal::DurationParts;
fn duration_between_builtin(name: &str, args: &[Value], f: BetweenFn) -> Result<Value, QueryError> {
if args.len() != 2 {
return Err(QueryError::Semantic(format!(
"{name}() expects exactly two arguments, got {}",
args.len()
)));
}
if matches!(args[0], Value::Null) || matches!(args[1], Value::Null) {
return Ok(Value::Null);
}
let (a_date, a_time, a_zone) = between_operand(name, &args[0])?;
let (b_date, b_time, b_zone) = between_operand(name, &args[1])?;
Ok(duration_value(f(
a_date,
a_time,
a_zone.as_ref(),
b_date,
b_time,
b_zone.as_ref(),
)))
}
type TruncateArgs<'a> = (&'a str, &'a Value, Option<&'a BTreeMap<String, Value>>);
fn parse_truncate_args<'a>(name: &str, args: &'a [Value]) -> Result<TruncateArgs<'a>, QueryError> {
if args.len() < 2 || args.len() > 3 {
return Err(QueryError::Semantic(format!(
"{name}() expects 2 or 3 arguments, got {}",
args.len()
)));
}
let unit = as_arith_str(&args[0]).ok_or_else(|| {
QueryError::Type(format!("{name}()'s first argument must be a unit string"))
})?;
let map = match args.get(2) {
None | Some(Value::Null) => None,
Some(Value::Map(m)) => Some(m),
Some(other) => {
return Err(QueryError::Type(format!(
"{name}()'s third argument must be a map, got {other:?}"
)))
}
};
Ok((unit, &args[1], map))
}
fn apply_date_overrides(
base_epoch_day: i32,
map: Option<&BTreeMap<String, Value>>,
) -> Result<i32, QueryError> {
let base_y = temporal::date_component(base_epoch_day, "year").unwrap();
let base_m = temporal::date_component(base_epoch_day, "month").unwrap();
let base_d = temporal::date_component(base_epoch_day, "day").unwrap();
let Some(m) = map else {
return Ok(base_epoch_day);
};
let year_raw = int_field(m, "year", base_y)?;
let year = i32::try_from(year_raw)
.map_err(|_| QueryError::Type(format!("'year' is out of range: {year_raw}")))?;
let month_raw = int_field(m, "month", base_m)?;
let month = u32::try_from(month_raw)
.map_err(|_| QueryError::Type(format!("'month' is out of range: {month_raw}")))?;
let day_raw = int_field(m, "day", base_d)?;
let day = u32::try_from(day_raw)
.map_err(|_| QueryError::Type(format!("'day' is out of range: {day_raw}")))?;
let result = temporal::epoch_day_from_ymd(year, month, day).ok_or_else(|| {
QueryError::Type(format!(
"{year:04}-{month:02}-{day:02} isn't a valid calendar date"
))
})?;
match m.get("dayOfWeek") {
None => Ok(result),
Some(v) => {
let dow = value_as_i64(v)
.ok_or_else(|| QueryError::Type("'dayOfWeek' must be an integer".into()))?;
temporal::set_iso_weekday(result, dow).ok_or_else(|| {
QueryError::Type(format!(
"'dayOfWeek' must be 1..7 (Monday..Sunday), got {dow}"
))
})
}
}
}
fn apply_time_overrides(
base_nanos_of_day: i64,
map: Option<&BTreeMap<String, Value>>,
) -> Result<i64, QueryError> {
let base_h = temporal::local_time_component(base_nanos_of_day, "hour").unwrap();
let base_min = temporal::local_time_component(base_nanos_of_day, "minute").unwrap();
let base_s = temporal::local_time_component(base_nanos_of_day, "second").unwrap();
let base_ns = temporal::local_time_component(base_nanos_of_day, "nanosecond").unwrap();
let Some(m) = map else {
return Ok(base_nanos_of_day);
};
let nanos = sub_second_nanos_from_map(base_ns, m)?;
let hour = int_field(m, "hour", base_h)?;
let minute = int_field(m, "minute", base_min)?;
let second = int_field(m, "second", base_s)?;
temporal::local_time_nanos_from_fields(hour, minute, second, nanos)
.ok_or_else(|| QueryError::Type("truncate(...)'s map has an out-of-range field".into()))
}
fn validate_truncate_map_keys(
name: &str,
map: Option<&BTreeMap<String, Value>>,
allowed: &[&str],
) -> Result<(), QueryError> {
let Some(m) = map else { return Ok(()) };
if let Some(bad) = m.keys().find(|k| !allowed.contains(&k.as_str())) {
return Err(QueryError::Type(format!(
"{name}(...)'s map has an unrecognized field '{bad}'"
)));
}
Ok(())
}
fn date_truncate_builtin(args: &[Value]) -> Result<Value, QueryError> {
let (unit, other, map) = parse_truncate_args("date.truncate", args)?;
validate_truncate_map_keys("date.truncate", map, &["year", "month", "day", "dayOfWeek"])?;
if matches!(other, Value::Null) {
return Ok(Value::Null);
}
let (base_date, _, _) = between_operand("date.truncate", other)?;
let base_date = base_date.ok_or_else(|| {
QueryError::Type(
"date.truncate() needs a value with a calendar date (Date, LocalDateTime, or DateTime)"
.into(),
)
})?;
let truncated = temporal::truncate_date_unit(base_date, unit).ok_or_else(|| {
QueryError::Type(format!(
"date.truncate(): '{unit}' isn't a recognized date unit"
))
})?;
Ok(Value::Property(PropertyValue::Date(apply_date_overrides(
truncated, map,
)?)))
}
const TIME_TRUNCATE_MAP_KEYS: &[&str] = &[
"hour",
"minute",
"second",
"millisecond",
"microsecond",
"nanosecond",
];
fn local_time_truncate_builtin(args: &[Value]) -> Result<Value, QueryError> {
let (unit, other, map) = parse_truncate_args("localtime.truncate", args)?;
validate_truncate_map_keys("localtime.truncate", map, TIME_TRUNCATE_MAP_KEYS)?;
if matches!(other, Value::Null) {
return Ok(Value::Null);
}
let (_, base_time, _) = between_operand("localtime.truncate", other)?;
let base_time = base_time.ok_or_else(|| {
QueryError::Type(
"localtime.truncate() needs a value with a time-of-day (LocalTime, Time, \
LocalDateTime, or DateTime)"
.into(),
)
})?;
let truncated = temporal::truncate_time_unit(base_time, unit).ok_or_else(|| {
QueryError::Type(format!(
"localtime.truncate(): '{unit}' isn't a recognized time unit"
))
})?;
Ok(Value::Property(PropertyValue::LocalTime(
apply_time_overrides(truncated, map)?,
)))
}
fn time_truncate_builtin(args: &[Value]) -> Result<Value, QueryError> {
let (unit, other, map) = parse_truncate_args("time.truncate", args)?;
validate_truncate_map_keys(
"time.truncate",
map,
&[
"hour",
"minute",
"second",
"millisecond",
"microsecond",
"nanosecond",
"timezone",
],
)?;
if matches!(other, Value::Null) {
return Ok(Value::Null);
}
let (_, base_time, base_offset) = between_operand("time.truncate", other)?;
let base_time = base_time.ok_or_else(|| {
QueryError::Type(
"time.truncate() needs a value with a time-of-day (LocalTime, Time, LocalDateTime, \
or DateTime)"
.into(),
)
})?;
let truncated = temporal::truncate_time_unit(base_time, unit).ok_or_else(|| {
QueryError::Type(format!(
"time.truncate(): '{unit}' isn't a recognized time unit"
))
})?;
let nanos_of_day = apply_time_overrides(truncated, map)?;
let offset_seconds = match map.and_then(|m| m.get("timezone")) {
Some(v) => match timezone_value_to_tzid(v)? {
temporal::TzId::Offset(o) => o,
temporal::TzId::Named(name) => {
return Err(QueryError::Type(format!(
"'timezone': '{name}' looks like a named timezone (e.g. 'Europe/Stockholm') -- TIME has \
no calendar date to resolve a named zone's DST-dependent offset against, only a fixed \
UTC offset like '+01:00' is supported"
)));
}
},
None => match base_offset {
Some(temporal::TzId::Offset(o)) => o,
_ => 0,
},
};
Ok(Value::Property(PropertyValue::Time {
nanos_of_day,
offset_seconds,
}))
}
fn truncate_date_time(base_date: i32, base_time: i64, unit: &str) -> Option<(i32, i64)> {
if let Some(d) = temporal::truncate_date_unit(base_date, unit) {
Some((d, 0))
} else {
temporal::truncate_time_unit(base_time, unit).map(|t| (base_date, t))
}
}
fn local_date_time_truncate_builtin(args: &[Value]) -> Result<Value, QueryError> {
let (unit, other, map) = parse_truncate_args("localdatetime.truncate", args)?;
validate_truncate_map_keys(
"localdatetime.truncate",
map,
&[
"year",
"month",
"day",
"dayOfWeek",
"hour",
"minute",
"second",
"millisecond",
"microsecond",
"nanosecond",
],
)?;
if matches!(other, Value::Null) {
return Ok(Value::Null);
}
let (base_date, base_time, _) = between_operand("localdatetime.truncate", other)?;
let base_date = base_date.ok_or_else(|| {
QueryError::Type(
"localdatetime.truncate() needs a value with a calendar date (Date, LocalDateTime, \
or DateTime)"
.into(),
)
})?;
let (trunc_date, trunc_time) = truncate_date_time(base_date, base_time.unwrap_or(0), unit)
.ok_or_else(|| {
QueryError::Type(format!(
"localdatetime.truncate(): '{unit}' isn't a recognized unit"
))
})?;
let final_date = apply_date_overrides(trunc_date, map)?;
let final_time = apply_time_overrides(trunc_time, map)?;
let (epoch_seconds, nanos) = temporal::combine_date_and_time(final_date, final_time);
Ok(Value::Property(PropertyValue::LocalDateTime {
epoch_seconds,
nanos,
}))
}
fn date_time_truncate_builtin(args: &[Value]) -> Result<Value, QueryError> {
let (unit, other, map) = parse_truncate_args("datetime.truncate", args)?;
validate_truncate_map_keys(
"datetime.truncate",
map,
&[
"year",
"month",
"day",
"dayOfWeek",
"hour",
"minute",
"second",
"millisecond",
"microsecond",
"nanosecond",
"timezone",
],
)?;
if matches!(other, Value::Null) {
return Ok(Value::Null);
}
let (base_date, base_time, base_offset) = between_operand("datetime.truncate", other)?;
let base_date = base_date.ok_or_else(|| {
QueryError::Type(
"datetime.truncate() needs a value with a calendar date (Date, LocalDateTime, or \
DateTime)"
.into(),
)
})?;
let (trunc_date, trunc_time) = truncate_date_time(base_date, base_time.unwrap_or(0), unit)
.ok_or_else(|| {
QueryError::Type(format!(
"datetime.truncate(): '{unit}' isn't a recognized unit"
))
})?;
let final_date = apply_date_overrides(trunc_date, map)?;
let final_time = apply_time_overrides(trunc_time, map)?;
let zone = match map.and_then(|m| m.get("timezone")) {
Some(v) => timezone_value_to_tzid(v)?,
None => base_offset.unwrap_or(temporal::TzId::Offset(0)),
};
let calendar = temporal::CalendarDateTime {
year: temporal::date_component(final_date, "year").unwrap() as i32,
month: temporal::date_component(final_date, "month").unwrap() as u32,
day: temporal::date_component(final_date, "day").unwrap() as u32,
hour: temporal::local_time_component(final_time, "hour").unwrap(),
minute: temporal::local_time_component(final_time, "minute").unwrap(),
second: temporal::local_time_component(final_time, "second").unwrap(),
nanos: temporal::local_time_component(final_time, "nanosecond").unwrap(),
};
let (epoch_seconds, nanos) =
temporal::date_time_from_fields(calendar, &zone).ok_or_else(|| {
QueryError::Type("datetime.truncate() produced an out-of-range value".into())
})?;
Ok(Value::Property(PropertyValue::DateTime {
epoch_seconds,
nanos,
zone: tz_to_graph(zone),
}))
}
fn value_as_i64(v: &Value) -> Option<i64> {
match v {
Value::Property(PropertyValue::Int(i)) | Value::Literal(Literal::Int(i)) => Some(*i),
_ => None,
}
}
fn value_as_f64(v: &Value) -> Option<f64> {
match as_arith_num(v)? {
ArithNum::Int(i) => Some(i as f64),
ArithNum::Float(f) => Some(f),
}
}
fn is_temporal_property_value(pv: &PropertyValue) -> bool {
matches!(
pv,
PropertyValue::Date(_)
| PropertyValue::Duration { .. }
| PropertyValue::LocalTime(_)
| PropertyValue::Time { .. }
| PropertyValue::LocalDateTime { .. }
| PropertyValue::DateTime { .. }
)
}
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(),
)),
}
}
fn temporal_component(pv: &PropertyValue, prop: &str) -> Option<PropertyValue> {
match pv {
PropertyValue::Date(d) => temporal::date_component(*d, prop).map(PropertyValue::Int),
PropertyValue::Duration {
months,
days,
seconds,
nanos,
} => temporal::duration_component(*months, *days, *seconds, *nanos, prop)
.map(PropertyValue::Int),
PropertyValue::LocalTime(nanos_of_day) => {
temporal::local_time_component(*nanos_of_day, prop).map(PropertyValue::Int)
}
PropertyValue::Time {
nanos_of_day,
offset_seconds,
} => time_component(*nanos_of_day, *offset_seconds, prop),
PropertyValue::LocalDateTime {
epoch_seconds,
nanos,
} => date_time_component(*epoch_seconds, *nanos, None, prop),
PropertyValue::DateTime {
epoch_seconds,
nanos,
zone,
} => date_time_component(*epoch_seconds, *nanos, Some(&tz_from_graph(zone)), prop),
_ => None,
}
}
fn time_component(nanos_of_day: i64, offset_seconds: i32, prop: &str) -> Option<PropertyValue> {
match prop {
"timezone" | "offset" => Some(PropertyValue::String(temporal::format_offset(
offset_seconds,
))),
"offsetSeconds" => Some(PropertyValue::Int(offset_seconds as i64)),
"offsetMinutes" => Some(PropertyValue::Int(offset_seconds as i64 / 60)),
_ => temporal::local_time_component(nanos_of_day, prop).map(PropertyValue::Int),
}
}
fn date_time_component(
epoch_seconds: i64,
nanos: i32,
zone: Option<&temporal::TzId>,
prop: &str,
) -> Option<PropertyValue> {
if let Some(zone) = zone {
let offset_seconds = temporal::resolve_offset(zone, epoch_seconds);
match prop {
"timezone" => {
let text = match zone {
temporal::TzId::Named(name) => name.clone(),
temporal::TzId::Offset(_) => temporal::format_offset(offset_seconds),
};
return Some(PropertyValue::String(text));
}
"offset" => {
return Some(PropertyValue::String(temporal::format_offset(
offset_seconds,
)))
}
"offsetSeconds" => return Some(PropertyValue::Int(offset_seconds as i64)),
"offsetMinutes" => return Some(PropertyValue::Int(offset_seconds as i64 / 60)),
"epochSeconds" => return Some(PropertyValue::Int(epoch_seconds)),
"epochMillis" => {
return Some(PropertyValue::Int(
temporal::epoch_seconds_and_millis(epoch_seconds, nanos).1,
))
}
_ => {}
}
}
let offset_seconds = zone.map_or(0, |z| temporal::resolve_offset(z, epoch_seconds));
let local_epoch_seconds = epoch_seconds + offset_seconds as i64;
temporal::date_time_calendar_component(local_epoch_seconds, prop)
.or_else(|| temporal::date_time_clock_component(local_epoch_seconds, nanos, prop))
.map(PropertyValue::Int)
}
fn apply_order_by(
rows: Vec<Vec<Value>>,
columns: &[String],
order_by: &[(ReturnExpr, SortDir)],
items: Option<&[ReturnItem]>,
skip: Option<i64>,
limit: Option<i64>,
) -> Result<Vec<Vec<Value>>, QueryError> {
let order_by_col: Vec<Option<usize>> = order_by
.iter()
.map(|(expr, _)| {
columns
.iter()
.position(|c| *c == default_column_name(expr, 0))
.or_else(|| {
items.and_then(|items| items.iter().position(|item| item.expr == *expr))
})
})
.collect();
let mut keyed: Vec<(Vec<Value>, Vec<Value>)> = Vec::with_capacity(rows.len());
for row in rows {
let row_map: HashMap<String, Value> =
columns.iter().cloned().zip(row.iter().cloned()).collect();
let keys = order_by
.iter()
.zip(&order_by_col)
.map(|((expr, _), col)| match col {
Some(i) => Ok(row[*i].clone()),
None => eval_projected_expr(expr, &row_map),
})
.collect::<Result<Vec<_>, _>>()?;
keyed.push((keys, row));
}
Ok(top_k_by(keyed, order_by, skip, limit)
.into_iter()
.map(|(_, row)| row)
.collect())
}
fn eval_projected_expr(
expr: &ReturnExpr,
row: &HashMap<String, Value>,
) -> Result<Value, QueryError> {
match expr {
ReturnExpr::Var(name) => row
.get(name)
.cloned()
.ok_or_else(|| QueryError::UnboundVariable(name.clone())),
ReturnExpr::Prop(pa) => {
let base = row
.get(&pa.var)
.ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
match base {
Value::Map(m) => Ok(m.get(&pa.prop).cloned().unwrap_or(Value::Null)),
Value::Node(n) => Ok(match n.props.get(&pa.prop).cloned() {
Some(PropertyValue::Null) | None => Value::Null,
Some(v) => property_value_to_value(v),
}),
Value::Edge(e) => Ok(match e.props.get(&pa.prop).cloned() {
Some(PropertyValue::Null) | None => Value::Null,
Some(v) => property_value_to_value(v),
}),
Value::Property(pv) => Ok(match temporal_component(pv, &pa.prop) {
Some(component) => Value::Property(component),
None => Value::Null,
}),
_ => Ok(Value::Null),
}
}
ReturnExpr::PropOf(base, prop) => {
let v = eval_projected_expr(base, row)?;
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 arg_values = args
.iter()
.map(|a| eval_projected_expr(a, row))
.collect::<Result<Vec<_>, _>>()?;
call_builtin(name, &arg_values, temporal::capture_now())
}
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(eval_projected_expr(t, row)?),
None => None,
};
for (when, then) in whens {
let when_value = eval_projected_expr(when, row)?;
let matched = match &test_value {
Some(tv) => value_eq(tv, &when_value),
None => matches!(when_value, Value::Literal(Literal::Bool(true))),
};
if matched {
return eval_projected_expr(then, row);
}
}
match else_ {
Some(e) => eval_projected_expr(e, row),
None => Ok(Value::Null),
}
}
ReturnExpr::Arith(l, op, r) => {
let lv = eval_projected_expr(l, row)?;
let rv = eval_projected_expr(r, row)?;
apply_arith(*op, &lv, &rv)
}
ReturnExpr::Neg(e) => {
let v = eval_projected_expr(e, row)?;
apply_neg(&v)
}
ReturnExpr::ListLit(items) => Ok(Value::List(
items
.iter()
.map(|item| eval_projected_expr(item, row))
.collect::<Result<Vec<_>, _>>()?,
)),
ReturnExpr::Index(base, index) => {
let base_v = eval_projected_expr(base, row)?;
let index_v = eval_projected_expr(index, row)?;
apply_index(&base_v, &index_v)
}
ReturnExpr::Slice(base, start, end) => {
let base_v = eval_projected_expr(base, row)?;
let start_v = start
.as_deref()
.map(|s| eval_projected_expr(s, row))
.transpose()?;
let end_v = end
.as_deref()
.map(|e| eval_projected_expr(e, row))
.transpose()?;
apply_slice(&base_v, start_v.as_ref(), end_v.as_ref())
}
ReturnExpr::ListComp {
var,
source,
where_clause,
project,
} => {
let source_v = eval_projected_expr(source, row)?;
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(), item.clone());
let keep = match where_clause {
Some(w) => value_to_bool3(&eval_projected_expr(w, &scoped_row)?)? == Some(true),
None => true,
};
if !keep {
continue;
}
result.push(match project {
Some(p) => eval_projected_expr(p, &scoped_row)?,
None => item,
});
}
Ok(Value::List(result))
}
ReturnExpr::Quantifier {
kind,
var,
source,
where_clause,
} => {
let source_v = eval_projected_expr(source, row)?;
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(), item.clone());
preds.push(match where_clause {
Some(w) => value_to_bool3(&eval_projected_expr(w, &scoped_row)?)?,
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(), eval_projected_expr(v, row)?);
}
Ok(Value::Map(map))
}
ReturnExpr::And(l, r) => Ok(bool3_to_value(and3(
value_to_bool3(&eval_projected_expr(l, row)?)?,
value_to_bool3(&eval_projected_expr(r, row)?)?,
))),
ReturnExpr::Or(l, r) => Ok(bool3_to_value(or3(
value_to_bool3(&eval_projected_expr(l, row)?)?,
value_to_bool3(&eval_projected_expr(r, row)?)?,
))),
ReturnExpr::Xor(l, r) => Ok(bool3_to_value(xor3(
value_to_bool3(&eval_projected_expr(l, row)?)?,
value_to_bool3(&eval_projected_expr(r, row)?)?,
))),
ReturnExpr::Not(e) => Ok(bool3_to_value(
value_to_bool3(&eval_projected_expr(e, row)?)?.map(|b| !b),
)),
ReturnExpr::Compare(l, op, r) => {
let lv = eval_projected_expr(l, row)?;
let rv = eval_projected_expr(r, row)?;
Ok(bool3_to_value(compare_values(&lv, *op, &rv)))
}
ReturnExpr::IsNull(e) => Ok(Value::Literal(Literal::Bool(matches!(
eval_projected_expr(e, row)?,
Value::Null
)))),
ReturnExpr::In(needle, haystack) => {
let nv = eval_projected_expr(needle, row)?;
let hv = eval_projected_expr(haystack, row)?;
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 {
Value::Node(n) => Ok(Value::Literal(Literal::Bool(
labels.iter().all(|l| n.labels.contains(l)),
))),
Value::Null => Ok(Value::Null),
other => Err(QueryError::Type(format!(
"'{var}' isn't a node — (n:Label) needs a node binding, got {other:?}"
))),
}
}
ReturnExpr::PatternPredicate(_) => Err(QueryError::Semantic(
"a pattern predicate (`(n)-->()` etc) can only be used inside WHERE".into(),
)),
ReturnExpr::PatternComprehension { .. } => Err(QueryError::Semantic(
"a pattern comprehension can only be used in RETURN/WITH position, or as an ORDER BY \
key that repeats one of their items verbatim"
.into(),
)),
ReturnExpr::ExistsPattern { .. } | ReturnExpr::ExistsSubquery(_) => Err(
QueryError::Semantic("an exists {} subquery can only be used inside WHERE".into()),
),
}
}
fn dedup_rows(rows: Vec<Vec<Value>>) -> Result<Vec<Vec<Value>>, QueryError> {
let mut seen: HashSet<Vec<HashKey>> = HashSet::with_capacity(rows.len());
let mut out = Vec::with_capacity(rows.len());
for row in rows {
let key = row
.iter()
.map(value_hash_key)
.collect::<Result<Vec<_>, _>>()?;
if seen.insert(key) {
out.push(row);
}
}
Ok(out)
}
fn dedup_binding_rows(
items: &[ReturnItem],
rows: Vec<BindingRow>,
) -> Result<Vec<BindingRow>, QueryError> {
let names: Vec<String> = items
.iter()
.enumerate()
.map(with_item_output_name)
.collect();
let mut seen: HashSet<Vec<HashKey>> = HashSet::with_capacity(rows.len());
let mut out = Vec::with_capacity(rows.len());
for row in rows {
let key = names
.iter()
.map(|name| {
binding_hash_key(row.get(name).unwrap_or_else(|| {
panic!("DISTINCT row missing its own projected column '{name}'")
}))
})
.collect::<Result<Vec<_>, _>>()?;
if seen.insert(key) {
out.push(row);
}
}
Ok(out)
}
fn top_k_by<T>(
mut keyed: Vec<(Vec<Value>, T)>,
order_by: &[(ReturnExpr, SortDir)],
skip: Option<i64>,
limit: Option<i64>,
) -> Vec<(Vec<Value>, T)> {
let cmp = |a: &(Vec<Value>, T), b: &(Vec<Value>, T)| -> std::cmp::Ordering {
for (i, (_, dir)) in order_by.iter().enumerate() {
let ord = compare_with_dir(&a.0[i], &b.0[i], *dir);
if ord != std::cmp::Ordering::Equal {
return ord;
}
}
std::cmp::Ordering::Equal
};
let skip_n = skip.unwrap_or(0).max(0) as usize;
match limit {
Some(n) => {
let k = skip_n + n.max(0) as usize;
if k == 0 {
keyed.clear();
} else if k < keyed.len() {
keyed.select_nth_unstable_by(k - 1, cmp);
keyed.truncate(k);
keyed.sort_by(cmp);
} else {
keyed.sort_by(cmp);
}
}
None => keyed.sort_by(cmp),
}
if skip_n > 0 {
keyed.drain(0..skip_n.min(keyed.len()));
}
keyed
}
fn compare_with_dir(a: &Value, b: &Value, dir: SortDir) -> std::cmp::Ordering {
let ord = compare_non_null(a, b);
if dir == SortDir::Desc {
ord.reverse()
} else {
ord
}
}
fn cmp_f64_nan_greatest(x: f64, y: f64) -> std::cmp::Ordering {
use std::cmp::Ordering;
match (x.is_nan(), y.is_nan()) {
(true, true) => Ordering::Equal,
(true, false) => Ordering::Greater,
(false, true) => Ordering::Less,
(false, false) => x.partial_cmp(&y).unwrap_or(Ordering::Equal),
}
}
fn compare_non_null(a: &Value, b: &Value) -> std::cmp::Ordering {
use std::cmp::Ordering;
if let (Value::List(_), Value::List(_)) = (a, b) {
return list_cmp_asc(a, b);
}
let pa = value_to_comparable(a);
let pb = value_to_comparable(b);
match (pa, pb) {
(Some(PropertyValue::Int(x)), Some(PropertyValue::Int(y))) => x.cmp(&y),
(Some(PropertyValue::Int(x)), Some(PropertyValue::Float(y))) => {
cmp_f64_nan_greatest(x as f64, y)
}
(Some(PropertyValue::Float(x)), Some(PropertyValue::Int(y))) => {
cmp_f64_nan_greatest(x, y as f64)
}
(Some(PropertyValue::Float(x)), Some(PropertyValue::Float(y))) => {
cmp_f64_nan_greatest(x, y)
}
(Some(PropertyValue::String(x)), Some(PropertyValue::String(y))) => x.cmp(&y),
(Some(PropertyValue::Bool(x)), Some(PropertyValue::Bool(y))) => x.cmp(&y),
(Some(PropertyValue::Date(x)), Some(PropertyValue::Date(y))) => x.cmp(&y),
(Some(PropertyValue::LocalTime(x)), Some(PropertyValue::LocalTime(y))) => x.cmp(&y),
(
Some(PropertyValue::Time {
nanos_of_day: x,
offset_seconds: ox,
}),
Some(PropertyValue::Time {
nanos_of_day: y,
offset_seconds: oy,
}),
) => (x - ox as i64 * 1_000_000_000).cmp(&(y - oy as i64 * 1_000_000_000)),
(
Some(PropertyValue::LocalDateTime {
epoch_seconds: xs,
nanos: xn,
}),
Some(PropertyValue::LocalDateTime {
epoch_seconds: ys,
nanos: yn,
}),
) => (xs, xn).cmp(&(ys, yn)),
(
Some(PropertyValue::DateTime {
epoch_seconds: xs,
nanos: xn,
..
}),
Some(PropertyValue::DateTime {
epoch_seconds: ys,
nanos: yn,
..
}),
) => (xs, xn).cmp(&(ys, yn)),
_ => match (type_rank(a), type_rank(b)) {
(Some(ra), Some(rb)) if ra != rb => ra.cmp(&rb),
_ => Ordering::Equal,
},
}
}
fn type_rank(v: &Value) -> Option<u8> {
match v {
Value::Map(_) => Some(0),
Value::Node(_) => Some(1),
Value::Edge(_) => Some(2),
Value::List(_) => Some(3),
Value::Path(_) => Some(4),
Value::Literal(Literal::String(_)) | Value::Property(PropertyValue::String(_)) => Some(5),
Value::Literal(Literal::Bool(_)) | Value::Property(PropertyValue::Bool(_)) => Some(6),
Value::Literal(Literal::Int(_))
| Value::Property(PropertyValue::Int(_))
| Value::Literal(Literal::Float(_))
| Value::Property(PropertyValue::Float(_)) => Some(7),
Value::Property(PropertyValue::Date(_)) => Some(8),
Value::Property(PropertyValue::LocalTime(_)) => Some(9),
Value::Property(PropertyValue::Time { .. }) => Some(10),
Value::Property(PropertyValue::LocalDateTime { .. }) => Some(11),
Value::Property(PropertyValue::DateTime { .. }) => Some(12),
Value::Null | Value::Literal(Literal::Null) | Value::Property(PropertyValue::Null) => {
Some(13)
}
_ => None,
}
}
fn list_cmp_asc(a: &Value, b: &Value) -> std::cmp::Ordering {
use std::cmp::Ordering;
let a_null = matches!(a, Value::Null);
let b_null = matches!(b, Value::Null);
match (a_null, b_null) {
(true, true) => return Ordering::Equal,
(true, false) => return Ordering::Greater,
(false, true) => return Ordering::Less,
(false, false) => {}
}
if let (Value::List(xs), Value::List(ys)) = (a, b) {
for (x, y) in xs.iter().zip(ys) {
match list_cmp_asc(x, y) {
Ordering::Equal => continue,
other => return other,
}
}
return xs.len().cmp(&ys.len());
}
compare_non_null(a, b)
}
fn value_to_comparable(v: &Value) -> Option<PropertyValue> {
match v {
Value::Property(pv) => Some(pv.clone()),
Value::Literal(lit) => Some(literal_to_value(lit)),
_ => None,
}
}
pub(crate) fn comparable_ordering(a: &Value, b: &Value) -> Option<std::cmp::Ordering> {
if let (Value::List(_), Value::List(_)) = (a, b) {
return Some(list_cmp_asc(a, b));
}
let (pa, pb) = match (value_to_comparable(a), value_to_comparable(b)) {
(Some(pa), Some(pb)) => (pa, pb),
_ => {
return match (type_rank(a), type_rank(b)) {
(Some(ra), Some(rb)) if ra != rb => Some(ra.cmp(&rb)),
_ => None,
};
}
};
Some(match (pa, pb) {
(PropertyValue::Int(x), PropertyValue::Int(y)) => x.cmp(&y),
(PropertyValue::Int(x), PropertyValue::Float(y)) => cmp_f64_nan_greatest(x as f64, y),
(PropertyValue::Float(x), PropertyValue::Int(y)) => cmp_f64_nan_greatest(x, y as f64),
(PropertyValue::Float(x), PropertyValue::Float(y)) => cmp_f64_nan_greatest(x, y),
(PropertyValue::String(x), PropertyValue::String(y)) => x.cmp(&y),
(PropertyValue::Bool(x), PropertyValue::Bool(y)) => x.cmp(&y),
(PropertyValue::Date(x), PropertyValue::Date(y)) => x.cmp(&y),
(PropertyValue::LocalTime(x), PropertyValue::LocalTime(y)) => x.cmp(&y),
(
PropertyValue::Time {
nanos_of_day: x,
offset_seconds: ox,
},
PropertyValue::Time {
nanos_of_day: y,
offset_seconds: oy,
},
) => (x - ox as i64 * 1_000_000_000).cmp(&(y - oy as i64 * 1_000_000_000)),
(
PropertyValue::LocalDateTime {
epoch_seconds: xs,
nanos: xn,
},
PropertyValue::LocalDateTime {
epoch_seconds: ys,
nanos: yn,
},
) => (xs, xn).cmp(&(ys, yn)),
(
PropertyValue::DateTime {
epoch_seconds: xs,
nanos: xn,
..
},
PropertyValue::DateTime {
epoch_seconds: ys,
nanos: yn,
..
},
) => (xs, xn).cmp(&(ys, yn)),
_ => return None,
})
}
fn compare_values(a: &Value, op: CompareOp, b: &Value) -> Option<bool> {
if matches!(a, Value::Null) || matches!(b, Value::Null) {
return None;
}
match op {
CompareOp::Eq => value_equal_ternary(a, b),
CompareOp::Ne => value_equal_ternary(a, b).map(|eq| !eq),
CompareOp::Lt => ordered_compare(a, b, |o| o == std::cmp::Ordering::Less),
CompareOp::Le => ordered_compare(a, b, |o| o != std::cmp::Ordering::Greater),
CompareOp::Gt => ordered_compare(a, b, |o| o == std::cmp::Ordering::Greater),
CompareOp::Ge => ordered_compare(a, b, |o| o != std::cmp::Ordering::Less),
CompareOp::StartsWith | CompareOp::EndsWith | CompareOp::Contains => {
let (Some(s), Some(p)) = (as_arith_str(a), as_arith_str(b)) else {
return None;
};
Some(match op {
CompareOp::StartsWith => s.starts_with(p),
CompareOp::EndsWith => s.ends_with(p),
CompareOp::Contains => s.contains(p),
_ => unreachable!("only StartsWith/EndsWith/Contains reach this arm"),
})
}
}
}
fn ordered_compare(
a: &Value,
b: &Value,
pred: impl Fn(std::cmp::Ordering) -> bool,
) -> Option<bool> {
if let (Some(x), Some(y)) = (value_as_f64(a), value_as_f64(b)) {
return Some(x.partial_cmp(&y).map(pred).unwrap_or(false));
}
value_partial_cmp(a, b).map(pred)
}
fn value_partial_cmp(a: &Value, b: &Value) -> Option<std::cmp::Ordering> {
use std::cmp::Ordering;
if matches!(a, Value::Null) || matches!(b, Value::Null) {
return None;
}
if let (Value::List(xs), Value::List(ys)) = (a, b) {
for (x, y) in xs.iter().zip(ys) {
match value_partial_cmp(x, y) {
Some(Ordering::Equal) => continue,
other => return other,
}
}
return Some(xs.len().cmp(&ys.len()));
}
if value_to_comparable(a).is_none() || value_to_comparable(b).is_none() {
return None;
}
comparable_ordering(a, b)
}
fn value_equal_ternary(a: &Value, b: &Value) -> Option<bool> {
match (a, b) {
(Value::Null, _) | (_, Value::Null) => None,
(Value::List(xs), Value::List(ys)) => {
if xs.len() != ys.len() {
return Some(false);
}
fold_ternary_eq(xs.iter().zip(ys).map(|(x, y)| value_equal_ternary(x, y)))
}
(Value::Map(x), Value::Map(y)) => {
if !x.keys().eq(y.keys()) {
return Some(false);
}
fold_ternary_eq(x.iter().map(|(k, xv)| value_equal_ternary(xv, &y[k])))
}
_ => Some(values_equal_numeric_aware(a, b)),
}
}
fn list_membership_ternary(needle: &Value, haystack: &Value) -> Result<Option<bool>, QueryError> {
match haystack {
Value::Null => Ok(None),
Value::List(items) => {
let mut saw_unknown = false;
for item in items {
match value_equal_ternary(needle, item) {
Some(true) => return Ok(Some(true)),
Some(false) => {}
None => saw_unknown = true,
}
}
Ok(if saw_unknown { None } else { Some(false) })
}
other => Err(QueryError::Type(format!(
"IN requires a list on the right-hand side, got {other:?}"
))),
}
}
fn fold_ternary_eq(mut results: impl Iterator<Item = Option<bool>>) -> Option<bool> {
let mut saw_unknown = false;
for r in results.by_ref() {
match r {
Some(false) => return Some(false),
Some(true) => {}
None => saw_unknown = true,
}
}
if saw_unknown {
None
} else {
Some(true)
}
}
fn values_equal_numeric_aware(a: &Value, b: &Value) -> bool {
match (as_arith_num(a), as_arith_num(b)) {
(Some(ArithNum::Int(x)), Some(ArithNum::Int(y))) => x == y,
(Some(ArithNum::Int(x)), Some(ArithNum::Float(y)))
| (Some(ArithNum::Float(y)), Some(ArithNum::Int(x))) => x as f64 == y,
(Some(ArithNum::Float(x)), Some(ArithNum::Float(y))) => x == y,
_ => value_eq(a, b),
}
}