marsdb_query/executor.rs
1use std::collections::{BTreeMap, HashMap, HashSet};
2
3use marsdb_graph::{AdjEntry, Direction, EdgeId, GraphStore, NodeId, PropertyValue, Txn, WriteTransaction};
4
5use crate::aggregate::{property_value_hash_key, value_hash_key, AggAcc, HashKey};
6use crate::ast::{
7 is_aggregate_name, CompareOp, Expr, Literal, MergeClause, NodePattern, Pattern, PropAccess, QueryClause,
8 QueryPart, RelDirection, RemoveItem, ReturnExpr, ReturnItem, SetItem, SortDir, Statement, Tail, UnwindClause,
9 UnwindSource, WithClause, WithExpr,
10};
11use crate::error::QueryError;
12use crate::ir::{ExpandDirection, LogicalPlan};
13use crate::planner::{build_match_plan, pattern_all_vars, pattern_new_vars};
14use crate::result::QueryResult;
15use crate::value::{PathElem, Value};
16
17/// Hidden key used to correlate `OPTIONAL MATCH` results back to the outer
18/// row that seeded them — never visible to user Cypher (not a valid
19/// identifier prefix a parsed pattern could ever produce).
20const OPTIONAL_SEED_IDX_KEY: &str = "__seed_idx";
21
22/// Hidden key tagging whether a `MERGE`d row came from the create-path or
23/// the match-path, consumed (and stripped) by `apply_merge_set` before the
24/// row becomes visible to the rest of the query.
25const MERGE_CREATED_KEY: &str = "__merge_created";
26
27#[derive(Debug, Clone)]
28enum Binding {
29 Node(NodeId),
30 Edge(EdgeId),
31 /// A scalar carried through a `WITH` projection (e.g. `WITH message.id
32 /// AS messageId`) — no graph identity, just a value along for the ride
33 /// to the next `QueryPart`/the final `Tail`.
34 Value(PropertyValue),
35 /// A `collect()` result carried through a `WITH` projection. Separate
36 /// from `Binding::Value` because `PropertyValue` (storage-layer) has no
37 /// list variant — lists are a query-layer-only concept, never
38 /// persisted — so a materialized `collect()` has nowhere else to live
39 /// between one `QueryPart` and the next. Elements are already-resolved
40 /// `Value`s, not `Binding`s — `UNWIND` restores graph identity on the
41 /// way back out via `value_to_binding_restore`, a separate step from
42 /// how this is stored here.
43 List(Vec<Value>),
44 /// A named path (`p = (a)-->(b)`) or `shortestPath()` result — see
45 /// `assemble_path`/`eval_shortest_path`. `PathBinding` (not `Binding`
46 /// again) because a path element only ever needs graph identity
47 /// (`NodeId`/`EdgeId`), never any of `Binding`'s other cases — using
48 /// `Binding` itself here would make "a path containing a path" a type
49 /// state nothing ever produces or handles.
50 Path(Vec<PathBinding>),
51}
52
53/// One element of a `Binding::Path`, alternating node/edge/node/.../node
54/// — the row-carried counterpart to `Value::Path`'s `PathElem` (which
55/// carries full `Node`/`Edge` records instead of just their ids, the same
56/// "keep identity in the row, resolve to a full record only when
57/// materializing for display" split every other `Binding`/`Value` pair
58/// already uses).
59#[derive(Debug, Clone)]
60enum PathBinding {
61 Node(NodeId),
62 Edge(EdgeId),
63}
64
65type BindingRow = HashMap<String, Binding>;
66
67/// Safety cap on unbounded variable-length traversal (`[:TYPE*0..]`) depth.
68/// Hitting it errors rather than silently truncating — see `VarExpand`
69/// evaluation. Node-visited-set BFS (not relationship-uniqueness) is used
70/// throughout, which is only correct because the graphs this targets
71/// (LDBC's REPLY_OF-style reply chains) form a forest, not a general
72/// cyclic graph — not safe to reuse as-is for a variable-length pattern
73/// over a cyclic relationship type without revisiting that assumption.
74const VAR_EXPAND_DEPTH_CAP: u32 = 30;
75
76pub struct Executor<'a> {
77 store: &'a GraphStore,
78}
79
80impl<'a> Executor<'a> {
81 pub fn new(store: &'a GraphStore) -> Self {
82 Self { store }
83 }
84
85 /// Dispatches on whether `stmt` ever mutates anything. A read-only
86 /// statement (`MATCH ... RETURN`, `is_read_only` below) runs inside a
87 /// `ReadTransaction` — a consistent snapshot that doesn't contend for
88 /// redb's single-writer lock, so concurrent readers run in parallel
89 /// instead of queueing behind each other. Everything else runs inside
90 /// a `WriteTransaction`, committed or aborted as a whole — the
91 /// crash-safety boundary from the plan (one statement = one commit).
92 /// Every graph access below this point must go through the `*_in_txn`
93 /// GraphStore methods, never the standalone `self.store.*` methods,
94 /// which open (and would deadlock trying to re-open) their own
95 /// transaction.
96 pub fn execute(&self, stmt: &Statement) -> Result<QueryResult, QueryError> {
97 if is_read_only(stmt) {
98 let read_txn = self.store.begin_read()?;
99 let Statement::Match {
100 clauses,
101 tail,
102 order_by,
103 limit,
104 } = stmt
105 else {
106 unreachable!("is_read_only only returns true for Statement::Match")
107 };
108 // No explicit commit/abort — a ReadTransaction is a pure
109 // snapshot view with nothing to roll back; it releases on drop.
110 return self.execute_match(Txn::Read(&read_txn), clauses, tail, order_by, *limit);
111 }
112 let write_txn = self.store.begin_write()?;
113 let outcome = match stmt {
114 Statement::Create(patterns) => self.execute_create(&write_txn, patterns),
115 Statement::Match {
116 clauses,
117 tail,
118 order_by,
119 limit,
120 } => self.execute_match(Txn::Write(&write_txn), clauses, tail, order_by, *limit),
121 };
122 match outcome {
123 Ok(result) => {
124 GraphStore::commit(write_txn)?;
125 Ok(result)
126 }
127 Err(e) => {
128 // Best-effort rollback; the original error is what matters.
129 let _ = GraphStore::abort(write_txn);
130 Err(e)
131 }
132 }
133 }
134
135 fn execute_create(&self, write_txn: &WriteTransaction, patterns: &[Pattern]) -> Result<QueryResult, QueryError> {
136 // A standalone CREATE is a MATCH...CREATE tail run against a
137 // single empty row -- `resolve_or_create_node` below never finds
138 // any variable already bound in an empty `BindingRow`, so every
139 // node token is fresh, exactly like standalone CREATE always was.
140 self.materialize_create(write_txn, patterns, &[BindingRow::new()])
141 }
142
143 /// Runs CREATE patterns once per row in `rows`. Shared by a
144 /// standalone `CREATE` statement (`execute_create`, a single empty
145 /// row) and a `MATCH ... CREATE` tail (`execute_match`, rows carry
146 /// bindings from the preceding MATCH/WITH). The only real difference
147 /// between the two is what `resolve_or_create_node` finds already
148 /// bound in a row -- nothing for standalone CREATE, real nodes for a
149 /// MATCH...CREATE tail, which is what lets the tail form add an edge
150 /// between two nodes that already exist.
151 fn materialize_create(
152 &self,
153 write_txn: &WriteTransaction,
154 patterns: &[Pattern],
155 rows: &[BindingRow],
156 ) -> Result<QueryResult, QueryError> {
157 for row in rows {
158 // A variable bound earlier in this same CREATE (an earlier hop,
159 // or an earlier comma-separated pattern) must be visible to
160 // later tokens naming it again -- e.g. a self-loop `(a)-[:R]->(a)`
161 // -- so track newly-created bindings in a local, per-row copy
162 // instead of just consulting the original incoming `row`.
163 let mut row = row.clone();
164 for pattern in patterns {
165 let mut prev_id = self.resolve_or_create_node(write_txn, &pattern.start, &row)?;
166 if let Some(var) = &pattern.start.var {
167 row.insert(var.clone(), Binding::Node(prev_id));
168 }
169 for (rel, node) in &pattern.hops {
170 if rel.hop_range.is_some() {
171 return Err(QueryError::Parse(
172 "CREATE doesn't support variable-length relationship patterns (e.g. [:TYPE*1..3])".into(),
173 ));
174 }
175 let node_id = self.resolve_or_create_node(write_txn, node, &row)?;
176 if let Some(var) = &node.var {
177 row.insert(var.clone(), Binding::Node(node_id));
178 }
179
180 let rel_label = rel.rel_type.clone().unwrap_or_else(|| "REL".to_string());
181 let rel_props = literal_props_to_values(&rel.props);
182 let (src, dst) = match rel.direction {
183 RelDirection::Right => (prev_id, node_id),
184 RelDirection::Left => (node_id, prev_id),
185 RelDirection::Either => {
186 return Err(QueryError::Parse(
187 "CREATE requires a directed relationship (-> or <-), not an undirected pattern".into(),
188 ))
189 }
190 };
191 GraphStore::create_edge_in_txn(write_txn, &rel_label, src, dst, rel_props)?;
192 prev_id = node_id;
193 }
194 }
195 }
196 Ok(QueryResult {
197 columns: vec![],
198 rows: vec![],
199 })
200 }
201
202 /// A node pattern token reuses an existing binding iff it names a
203 /// variable already bound in `row` (from a preceding MATCH/WITH) --
204 /// restating labels/props on that token is rejected with a clear
205 /// error rather than silently ignored, since silently dropping
206 /// user-written labels/props would be a correctness trap. Anything
207 /// else (no variable, or a variable not yet bound in this row)
208 /// creates a brand-new node, exactly like standalone CREATE always
209 /// has for every node token.
210 fn resolve_or_create_node(
211 &self,
212 write_txn: &WriteTransaction,
213 node: &NodePattern,
214 row: &BindingRow,
215 ) -> Result<NodeId, QueryError> {
216 if let Some(var) = &node.var {
217 if let Some(binding) = row.get(var) {
218 let Binding::Node(id) = binding else {
219 return Err(QueryError::Parse(format!(
220 "'{var}' is not a node — can't use it as a CREATE pattern endpoint"
221 )));
222 };
223 if !node.labels.is_empty() || !node.props.is_empty() {
224 return Err(QueryError::Parse(format!(
225 "'{var}' is already bound — CREATE can't add labels/properties to an existing node"
226 )));
227 }
228 return Ok(*id);
229 }
230 }
231 let labels: Vec<&str> = node.labels.iter().map(String::as_str).collect();
232 let props = literal_props_to_values(&node.props);
233 Ok(GraphStore::create_node_in_txn(write_txn, &labels, props)?)
234 }
235
236 /// Runs `MERGE` once per row in `rows` (`clause.pattern.hops.len() <=
237 /// 1`, enforced at parse time — whole-pattern atomicity across
238 /// multiple simultaneously-unbound hops isn't attempted in v1: which
239 /// hop's "not found" should trigger creation of what, in what order,
240 /// gets genuinely hard to reason about correctly for longer chains).
241 fn eval_merge(
242 &self,
243 write_txn: &WriteTransaction,
244 clause: &MergeClause,
245 rows: &[BindingRow],
246 ) -> Result<Vec<BindingRow>, QueryError> {
247 let mut out = Vec::new();
248 for row in rows {
249 out.extend(self.merge_one_row(write_txn, clause, row)?);
250 }
251 self.apply_merge_set(write_txn, clause, &mut out)?;
252 Ok(out)
253 }
254
255 fn merge_one_row(
256 &self,
257 write_txn: &WriteTransaction,
258 clause: &MergeClause,
259 row: &BindingRow,
260 ) -> Result<Vec<BindingRow>, QueryError> {
261 // Validate every token before doing any graph work (search or
262 // create) — an unconstrained node pattern that isn't already bound
263 // would otherwise let the search below silently "match" every
264 // node in the graph (AllNodesScan, no Filter), which is a
265 // wrong-answer footgun, not a helpful default.
266 require_mergeable(&clause.pattern.start, row)?;
267 for (rel, node) in &clause.pattern.hops {
268 if rel.hop_range.is_some() {
269 return Err(QueryError::Parse(
270 "MERGE doesn't support variable-length relationship patterns (e.g. [:TYPE*1..3])".into(),
271 ));
272 }
273 require_mergeable(node, row)?;
274 }
275
276 // Try the pattern as an ordinary MATCH first. Whatever's already
277 // bound in `row` (e.g. `a` from a preceding MATCH) becomes a Seed,
278 // not a fresh scan — build_match_plan already knows how to do
279 // this, the same mechanism every ordinary MATCH clause uses. For a
280 // one-hop pattern this already searches the *connected*
281 // sub-pattern (Expand from the resolved source, Filter by the
282 // target's own constraints), not each node independently — which
283 // is exactly the correctness property MERGE needs and gets for
284 // free by reusing this instead of inventing bespoke search logic.
285 let carried_vars: HashSet<String> = row.keys().cloned().collect();
286 let plan = build_match_plan(&clause.pattern, &None, &carried_vars)?;
287 let found = self.eval_plan(Txn::Write(write_txn), &plan, std::slice::from_ref(row))?;
288 if !found.is_empty() {
289 return Ok(found.into_iter().map(|r| tag_merge_created(r, false)).collect());
290 }
291
292 // Nothing found — create exactly one new instance. Reuses
293 // resolve_or_create_node, the same "reuse if the token's var is
294 // already bound in the row, else create fresh" logic
295 // Tail::Create/materialize_create already use.
296 let mut new_row = row.clone();
297 let start_id = self.resolve_or_create_node(write_txn, &clause.pattern.start, &new_row)?;
298 if let Some(var) = &clause.pattern.start.var {
299 new_row.insert(var.clone(), Binding::Node(start_id));
300 }
301 // At most one hop (enforced at parse time) -- a plain `if let`,
302 // not a loop, so there's no dangling "previous node" state to
303 // thread once a 2nd+ hop is ever supported.
304 if let Some((rel, node)) = clause.pattern.hops.first() {
305 let node_id = self.resolve_or_create_node(write_txn, node, &new_row)?;
306 if let Some(var) = &node.var {
307 new_row.insert(var.clone(), Binding::Node(node_id));
308 }
309 let rel_label = rel.rel_type.clone().unwrap_or_else(|| "REL".to_string());
310 let rel_props = literal_props_to_values(&rel.props);
311 let (src, dst) = match rel.direction {
312 RelDirection::Right => (start_id, node_id),
313 RelDirection::Left => (node_id, start_id),
314 RelDirection::Either => {
315 return Err(QueryError::Parse(
316 "MERGE requires a directed relationship (-> or <-), not an undirected pattern".into(),
317 ))
318 }
319 };
320 let edge_id = GraphStore::create_edge_in_txn(write_txn, &rel_label, src, dst, rel_props)?;
321 if let Some(var) = &rel.var {
322 new_row.insert(var.clone(), Binding::Edge(edge_id));
323 }
324 }
325 Ok(vec![tag_merge_created(new_row, true)])
326 }
327
328 /// Applies `ON CREATE SET`/`ON MATCH SET` to the right rows (matching
329 /// real Cypher semantics exactly: `ON CREATE` fires whenever anything
330 /// in the pattern was newly created, `ON MATCH` only when the whole
331 /// pattern already existed as-is — the single per-row
332 /// `MERGE_CREATED_KEY` tag is the correct model for this, not a
333 /// simplification of it — see `eval_optional_part`'s
334 /// `OPTIONAL_SEED_IDX_KEY` for the same hidden-tag precedent), then
335 /// strips the tag before the rows become visible to the rest of the
336 /// query.
337 fn apply_merge_set(
338 &self,
339 write_txn: &WriteTransaction,
340 clause: &MergeClause,
341 rows: &mut Vec<BindingRow>,
342 ) -> Result<(), QueryError> {
343 for row in rows.iter_mut() {
344 let created = match row.remove(MERGE_CREATED_KEY) {
345 Some(Binding::Value(PropertyValue::Bool(b))) => b,
346 other => unreachable!("{MERGE_CREATED_KEY} tagged internally as Binding::Value(Bool), got {other:?}"),
347 };
348 let items = if created { &clause.on_create } else { &clause.on_match };
349 for item in items {
350 apply_set_item(write_txn, row, item)?;
351 }
352 }
353 Ok(())
354 }
355
356 fn execute_match(
357 &self,
358 txn: Txn,
359 clauses: &[QueryClause],
360 tail: &Option<Tail>,
361 order_by: &Option<Vec<(ReturnExpr, SortDir)>>,
362 limit: Option<i64>,
363 ) -> Result<QueryResult, QueryError> {
364 // Threads bindings through each MATCH/UNWIND/WITH clause.
365 // `carried_vars` tells the planner which of the next MATCH clause's
366 // pattern variables are already bound (-> LogicalPlan::Seed) rather
367 // than fresh (-> a scan). Starts empty: the first clause never has
368 // anything carried into it.
369 let mut carried_vars: HashSet<String> = HashSet::new();
370 let mut current_rows: Vec<BindingRow> = vec![BindingRow::new()];
371 // LIMIT push-down: when the *entire* statement is nothing but one
372 // un-filtered, non-optional, single-node MATCH (no hops, no WHERE,
373 // no WITH, at most the one label a NodeByLabelScan already narrows
374 // by) feeding straight into a LIMIT with no ORDER BY, the scan
375 // itself never needs to look past the first `limit` nodes -- there
376 // is *nothing* downstream (no Filter/Expand/aggregation) that
377 // could still drop a row, so capping the raw storage scan can't
378 // change the result. Every more complex shape falls through to the
379 // general path below unchanged, which doesn't short-circuit --
380 // this executor materializes a `Vec<BindingRow>` at every step
381 // rather than pulling lazily, so pushing LIMIT further (past a
382 // Filter, an Expand, more than one clause, ...) would need a real
383 // streaming executor to stay correct, not just a deeper check here.
384 // A DISTINCT RETURN can also drop rows -- capping the raw scan at
385 // `limit` before dedup could return fewer than `limit` *distinct*
386 // rows even when more exist past what got scanned, so this shape
387 // is excluded the same way a WHERE/Filter already is.
388 let scan_limit_shortcut = order_by.is_none()
389 && limit.is_some()
390 && !matches!(tail, Some(Tail::Return(_, true)))
391 && matches!(clauses, [QueryClause::Match(part)] if
392 !part.shortest_path
393 && part.path_var.is_none()
394 && !part.optional
395 && part.with.is_none()
396 && part.pattern.hops.is_empty()
397 && part.where_clause.is_none()
398 && part.pattern.start.labels.len() <= 1
399 && part.pattern.start.props.is_empty()
400 && part.pattern.start.var.is_some());
401 if scan_limit_shortcut {
402 let [QueryClause::Match(part)] = clauses else {
403 unreachable!("scan_limit_shortcut's own matches! already checked this shape");
404 };
405 let var = part.pattern.start.var.as_deref().expect("checked by scan_limit_shortcut");
406 let label = part.pattern.start.labels.first().map(String::as_str);
407 let limit_usize = limit.expect("checked by scan_limit_shortcut").max(0) as usize;
408 current_rows = self.scan(txn, var, label, ¤t_rows, Some(limit_usize))?;
409 } else {
410 for clause in clauses {
411 match clause {
412 QueryClause::Match(part) => {
413 current_rows = if part.shortest_path {
414 // Not a LogicalPlan/eval_plan traversal at all —
415 // see eval_shortest_path's docs.
416 self.eval_shortest_path(txn, part, ¤t_rows)?
417 } else if let Some(path_var) = &part.path_var {
418 let (named_pattern, synthesized) = name_pattern_for_path(&part.pattern);
419 let plan = build_match_plan(&named_pattern, &part.where_clause, &carried_vars)?;
420 let mut rows = if part.optional {
421 let new_vars = pattern_new_vars(&named_pattern, &carried_vars);
422 self.eval_optional_part(txn, &plan, ¤t_rows, &new_vars)?
423 } else {
424 self.eval_plan(txn, &plan, ¤t_rows)?
425 };
426 for row in &mut rows {
427 let path_binding = assemble_path(&named_pattern, row);
428 for key in &synthesized {
429 row.remove(key);
430 }
431 row.insert(path_var.clone(), path_binding);
432 }
433 rows
434 } else {
435 let plan = build_match_plan(&part.pattern, &part.where_clause, &carried_vars)?;
436 if part.optional {
437 let new_vars = pattern_new_vars(&part.pattern, &carried_vars);
438 self.eval_optional_part(txn, &plan, ¤t_rows, &new_vars)?
439 } else {
440 self.eval_plan(txn, &plan, ¤t_rows)?
441 }
442 };
443 let mut new_vars = pattern_all_vars(&part.pattern);
444 if let Some(path_var) = &part.path_var {
445 new_vars.insert(path_var.clone());
446 }
447 current_rows = self.apply_with_or_carry(txn, &part.with, current_rows, new_vars, &mut carried_vars)?;
448 }
449 QueryClause::Unwind(u) => {
450 current_rows = self.eval_unwind(txn, u, ¤t_rows)?;
451 current_rows = self.apply_with_or_carry(
452 txn,
453 &u.with,
454 current_rows,
455 HashSet::from([u.var.clone()]),
456 &mut carried_vars,
457 )?;
458 }
459 QueryClause::Merge(m) => {
460 // MERGE always needs real `.insert`-capable write
461 // access, whether or not the rest of the statement
462 // would otherwise be read-only (e.g. `MERGE (n) RETURN
463 // n`) — see `is_read_only`, which already accounts for
464 // this by checking `clauses` too, so `txn` is
465 // guaranteed to be `Txn::Write` here.
466 let write_txn = require_write_txn(txn);
467 current_rows = self.eval_merge(write_txn, m, ¤t_rows)?;
468 current_rows = self.apply_with_or_carry(
469 txn,
470 &m.with,
471 current_rows,
472 pattern_all_vars(&m.pattern),
473 &mut carried_vars,
474 )?;
475 }
476 }
477 }
478 }
479 // ORDER BY must see every matching row before LIMIT truncates —
480 // sort, then take N, not the other way around. Only pre-truncate
481 // (the v1 "doesn't short-circuit" path) when there's no ORDER BY to
482 // invalidate it; DELETE/SET+LIMIT keep their "stop after N
483 // bindings" behavior since they have no ORDER BY position in the
484 // grammar. RETURN DISTINCT is excluded too, same reasoning as
485 // ORDER BY: DISTINCT can still drop rows *after* this point, so
486 // pre-truncating the raw input here could return fewer than
487 // `limit` distinct rows even when more exist -- its LIMIT gets
488 // applied after dedup instead, below.
489 let distinct_return = matches!(tail, Some(Tail::Return(_, true)));
490 if order_by.is_none() && !distinct_return {
491 if let Some(count) = limit {
492 current_rows.truncate(count.max(0) as usize);
493 }
494 }
495 // Delete/Set need real `.insert`/`.remove`-capable write access,
496 // not just `Txn`'s read-only `get`/`iter` — but they're only ever
497 // reached via `Executor::execute`'s write-dispatch path (see
498 // `is_read_only`), which always opens a `WriteTransaction`, so
499 // `txn` is guaranteed to be `Txn::Write` here.
500 // A non-aggregating RETURN's ORDER BY can reference either a
501 // RETURN-introduced alias (`RETURN friend.id AS friendId ORDER BY
502 // friendId`) or a variable still in scope that isn't returned at
503 // all (`RETURN n.num AS prop ORDER BY n.num` — `n` itself never
504 // appears in the RETURN list) — real Cypher allows both. Sorting
505 // needs both the pre-projection bindings *and* the post-projection
506 // output columns available at once, so it happens after
507 // `materialize_return`, against a combined view of the two (see
508 // `apply_order_by_with_scope`) rather than either alone. The
509 // aggregating case can't use pre-projection bindings at all
510 // (grouping has already collapsed the per-row bindings by then), so
511 // it keeps sorting the post-projection output alone via
512 // `apply_order_by`, further down.
513 let mut order_by_pre_applied = false;
514 let mut result = match tail {
515 // A missing tail only ever occurs with a MERGE clause and
516 // nothing after it — a pure write, same empty result shape
517 // standalone CREATE already returns (not one blank row per
518 // `current_rows`, which a synthetic `Tail::Return(vec![])`
519 // would produce instead).
520 None => QueryResult { columns: vec![], rows: vec![] },
521 Some(Tail::Return(items, distinct)) => {
522 let projected = self.materialize_return(txn, items, ¤t_rows, *distinct)?;
523 if let Some(ob) = order_by {
524 // DISTINCT (like aggregation) can drop rows, breaking
525 // the 1:1 correspondence `apply_order_by_with_scope`
526 // needs between `current_rows` and the projected
527 // output -- ORDER BY after DISTINCT can only sort the
528 // post-projection, post-dedup result, same as the
529 // aggregating case just below.
530 if !has_aggregate(items) && !distinct {
531 order_by_pre_applied = true;
532 self.apply_order_by_with_scope(txn, ¤t_rows, projected, ob, limit)?
533 } else {
534 projected
535 }
536 } else {
537 projected
538 }
539 }
540 Some(Tail::Delete(vars)) => {
541 self.materialize_delete(require_write_txn(txn), vars, ¤t_rows, false)?
542 }
543 Some(Tail::DetachDelete(vars)) => {
544 self.materialize_delete(require_write_txn(txn), vars, ¤t_rows, true)?
545 }
546 Some(Tail::Set(items)) => self.materialize_set(require_write_txn(txn), items, ¤t_rows)?,
547 Some(Tail::Remove(items)) => self.materialize_remove(require_write_txn(txn), items, ¤t_rows)?,
548 Some(Tail::Create(patterns)) => {
549 self.materialize_create(require_write_txn(txn), patterns, ¤t_rows)?
550 }
551 };
552 if let Some(order_by) = order_by {
553 if !order_by_pre_applied {
554 result.rows = apply_order_by(result.rows, &result.columns, order_by, limit)?;
555 }
556 } else if distinct_return {
557 // The pre-truncate above was skipped for exactly this case --
558 // apply LIMIT now, after materialize_return's dedup, instead.
559 if let Some(count) = limit {
560 result.rows.truncate(count.max(0) as usize);
561 }
562 }
563 Ok(result)
564 }
565
566 /// Applies a clause's optional trailing `WITH` (shared by both
567 /// `QueryClause::Match` and `QueryClause::Unwind`, which can each end
568 /// in one — see `QueryClause`'s docs), or, with no `WITH`, grows
569 /// `carried_vars` by `new_vars` so the next clause shares this one's
570 /// binding scope — same "no WITH means stay in scope" rule `OPTIONAL
571 /// MATCH` already gets, now uniform across clause kinds.
572 fn apply_with_or_carry(
573 &self,
574 txn: Txn,
575 with: &Option<WithClause>,
576 rows: Vec<BindingRow>,
577 new_vars: HashSet<String>,
578 carried_vars: &mut HashSet<String>,
579 ) -> Result<Vec<BindingRow>, QueryError> {
580 let Some(with) = with else {
581 carried_vars.extend(new_vars);
582 return Ok(rows);
583 };
584 let mut rows = self.materialize_with(txn, with, &rows)?;
585 if let Some(with_order_by) = &with.order_by {
586 rows = self.apply_order_by_bindings(txn, rows, with_order_by, with.limit)?;
587 } else if let Some(with_limit) = with.limit {
588 rows.truncate(with_limit.max(0) as usize);
589 }
590 *carried_vars = with.items.iter().enumerate().map(with_item_output_name).collect();
591 Ok(rows)
592 }
593
594 /// `UNWIND`'s fan-out. Not a graph traversal — like `WITH`, handled
595 /// directly here rather than through a `LogicalPlan`/`eval_plan` (see
596 /// `UnwindClause`'s docs). Cross-joins each input row against every
597 /// element of that row's resolved list, then applies the clause's own
598 /// `WHERE`.
599 fn eval_unwind(&self, txn: Txn, clause: &UnwindClause, rows: &[BindingRow]) -> Result<Vec<BindingRow>, QueryError> {
600 let mut out = Vec::new();
601 for row in rows {
602 let elements: Vec<Binding> = match &clause.source {
603 UnwindSource::Var(name) => {
604 let binding = row.get(name).ok_or_else(|| QueryError::UnboundVariable(name.clone()))?;
605 let Binding::List(items) = binding else {
606 return Err(QueryError::Parse(format!(
607 "'{name}' isn't a list — UNWIND needs a list (e.g. from collect())"
608 )));
609 };
610 items.iter().map(value_to_binding_restore).collect()
611 }
612 UnwindSource::List(literals) => {
613 literals.iter().map(|lit| Binding::Value(literal_to_value(lit))).collect()
614 }
615 };
616 for element in elements {
617 let mut new_row = row.clone();
618 new_row.insert(clause.var.clone(), element);
619 out.push(new_row);
620 }
621 }
622 if let Some(where_clause) = &clause.where_clause {
623 let mut filtered = Vec::with_capacity(out.len());
624 for row in out {
625 if self.eval_with_expr(txn, where_clause, &row)? == Some(true) {
626 filtered.push(row);
627 }
628 }
629 out = filtered;
630 }
631 Ok(out)
632 }
633
634 /// `shortestPath((a)-[:TYPE*..N]-(b))` — a real parent-pointer BFS
635 /// between two already-bound endpoints, not a `LogicalPlan`/
636 /// `VarExpand` traversal (which only tracks final position plus a
637 /// visited set, not the hop-by-hop chain a path needs to reconstruct).
638 /// BFS visits in non-decreasing depth order, so the first time `b` is
639 /// reached is *a* shortest path — stop there and reconstruct via
640 /// parent pointers, rather than enumerating every path up to some
641 /// bound the way `VarExpand` does.
642 ///
643 /// Both endpoints must already be bound by a preceding clause (e.g.
644 /// `MATCH (a:Person{name:'Alice'}), (b:Person{name:'Bob'}) MATCH p =
645 /// shortestPath((a)-[:KNOWS*]-(b)) RETURN p` — parser-enforced, see
646 /// `parser::validate_shortest_path_pattern`) — v1 doesn't attempt to
647 /// resolve a fresh/scanned endpoint here the way ordinary MATCH does,
648 /// since "shortest path to *any* node matching these constraints" is a
649 /// different, more ambiguous question than "shortest path between
650 /// these two specific nodes."
651 ///
652 /// Every input row always survives (unlike an ordinary pattern match,
653 /// which can produce zero rows for a non-match) — an unreachable pair
654 /// binds the path variable to `Null`, same as `OPTIONAL MATCH`'s
655 /// null-padding, rather than dropping the row. `part.optional` is
656 /// therefore a no-op here, not separately handled. Exceeding the
657 /// safety depth cap on an unbounded (`*..`) search also resolves to
658 /// `Null`, not an error — unlike `VarExpand`'s cap (which errors,
659 /// because truncating there would silently produce an *incomplete
660 /// set* of paths, a wrong-answer risk), `shortestPath()` is only ever
661 /// answering "is there a path within the searched horizon," which is
662 /// a well-defined answer either way.
663 fn eval_shortest_path(
664 &self,
665 txn: Txn,
666 part: &QueryPart,
667 rows: &[BindingRow],
668 ) -> Result<Vec<BindingRow>, QueryError> {
669 let Some(path_var) = &part.path_var else {
670 // Nothing names the result, so there's nothing to bind and no
671 // filtering effect (see this function's docs) — pure no-op.
672 return Ok(rows.to_vec());
673 };
674 let start_var = part.pattern.start.var.as_deref().expect(
675 "shortestPath()'s start node always has a var — validated at parse time by \
676 validate_shortest_path_pattern",
677 );
678 let (rel, end_node) = &part.pattern.hops[0];
679 let end_var = end_node.var.as_deref().expect(
680 "shortestPath()'s end node always has a var — validated at parse time by \
681 validate_shortest_path_pattern",
682 );
683 let (min_hops, max_hops) = rel.hop_range.expect(
684 "shortestPath()'s relationship is always variable-length — validated at parse time by \
685 validate_shortest_path_pattern",
686 );
687 let direction = match rel.direction {
688 RelDirection::Right => ExpandDirection::Out,
689 RelDirection::Left => ExpandDirection::In,
690 RelDirection::Either => ExpandDirection::Either,
691 };
692 let rel_label = rel.rel_type.as_deref();
693
694 let mut out = Vec::with_capacity(rows.len());
695 for row in rows {
696 let start_id = require_bound_node(row, start_var)?;
697 let end_id = require_bound_node(row, end_var)?;
698 let path = self.shortest_path_between(txn, start_id, end_id, direction, rel_label, min_hops, max_hops)?;
699 let mut new_row = row.clone();
700 let binding = match path {
701 Some(elems) => Binding::Path(elems),
702 None => Binding::Value(PropertyValue::Null),
703 };
704 new_row.insert(path_var.clone(), binding);
705 out.push(new_row);
706 }
707 if let Some(where_clause) = &part.where_clause {
708 let mut filtered = Vec::with_capacity(out.len());
709 for row in out {
710 if self.eval_expr(txn, where_clause, &row)? == Some(true) {
711 filtered.push(row);
712 }
713 }
714 out = filtered;
715 }
716 Ok(out)
717 }
718
719 /// The BFS itself. `min_hops` is only ever 0 or 1 (`validate_shortest_
720 /// path_pattern` rejects anything higher) — deliberately: a plain
721 /// visited-set BFS can't correctly answer "shortest path of at least N
722 /// hops" for N > 1 (a node first reached at a too-early depth would
723 /// need to stay revisitable for a later, longer route to it, which a
724 /// visited-set structurally can't represent) without a different
725 /// (node, depth)-keyed algorithm. Rejecting the case outright at parse
726 /// time is safer than silently answering it wrong.
727 fn shortest_path_between(
728 &self,
729 txn: Txn,
730 start: NodeId,
731 end: NodeId,
732 direction: ExpandDirection,
733 rel_label: Option<&str>,
734 min_hops: u32,
735 max_hops: Option<u32>,
736 ) -> Result<Option<Vec<PathBinding>>, QueryError> {
737 if start == end && min_hops == 0 {
738 return Ok(Some(vec![PathBinding::Node(start)]));
739 }
740 let cap = max_hops.unwrap_or(VAR_EXPAND_DEPTH_CAP);
741 let mut parent: HashMap<NodeId, (NodeId, EdgeId)> = HashMap::new();
742 let mut visited: HashSet<NodeId> = HashSet::new();
743 visited.insert(start);
744 let mut frontier = vec![start];
745 let mut depth = 0u32;
746 while depth < cap && !frontier.is_empty() {
747 depth += 1;
748 let mut next_frontier = Vec::new();
749 for node in frontier {
750 for entry in neighbors_for_direction(txn, node, direction, rel_label)? {
751 if entry.other == end {
752 parent.insert(entry.other, (node, entry.edge_id));
753 return Ok(Some(reconstruct_path(&parent, start, end)));
754 }
755 if visited.insert(entry.other) {
756 parent.insert(entry.other, (node, entry.edge_id));
757 next_frontier.push(entry.other);
758 }
759 }
760 }
761 frontier = next_frontier;
762 }
763 Ok(None)
764 }
765
766 /// Projects `rows` through a `WITH` clause. Unlike `materialize_return`
767 /// (which resolves everything down to display `Value`s), a bare
768 /// variable reference (`WITH message`) must keep its graph identity
769 /// (`Binding::Node`/`Edge`) so the next `QueryPart` can keep
770 /// traversing from it — only computed expressions collapse to a
771 /// scalar `Binding::Value`.
772 fn materialize_with(
773 &self,
774 txn: Txn,
775 with: &WithClause,
776 rows: &[BindingRow],
777 ) -> Result<Vec<BindingRow>, QueryError> {
778 let mut out = if !has_aggregate(&with.items) {
779 let mut out = Vec::with_capacity(rows.len());
780 for row in rows {
781 let mut new_row = BindingRow::new();
782 for (i, item) in with.items.iter().enumerate() {
783 let name = with_item_output_name((i, item));
784 let binding = self.item_binding(txn, &item.expr, row)?;
785 new_row.insert(name, binding);
786 }
787 out.push(new_row);
788 }
789 out
790 } else {
791 validate_return_items(&with.items)?;
792 let grouped = self.resolve_grouped_rows(txn, &with.items, rows)?;
793 grouped
794 .into_iter()
795 .map(|bindings| {
796 with.items
797 .iter()
798 .enumerate()
799 .zip(bindings)
800 .map(|((i, item), b)| (with_item_output_name((i, item)), b))
801 .collect()
802 })
803 .collect()
804 };
805 if let Some(where_clause) = &with.where_clause {
806 let mut filtered = Vec::with_capacity(out.len());
807 for row in out {
808 if self.eval_with_expr(txn, where_clause, &row)? == Some(true) {
809 filtered.push(row);
810 }
811 }
812 out = filtered;
813 }
814 Ok(out)
815 }
816
817 /// The `Binding` one WITH/RETURN item evaluates to for one input row. A
818 /// bare `Var` keeps its graph identity (`Binding::Node`/`Edge`) so a
819 /// later `QueryPart` can keep traversing from it; anything else
820 /// (computed expressions) collapses to `Binding::Value`. Shared by the
821 /// non-aggregating `materialize_with` path and grouping-key evaluation.
822 fn item_binding(&self, txn: Txn, expr: &ReturnExpr, row: &BindingRow) -> Result<Binding, QueryError> {
823 match expr {
824 ReturnExpr::Var(v) => row.get(v).cloned().ok_or_else(|| QueryError::UnboundVariable(v.clone())),
825 other => {
826 let value = self.eval_return_expr(txn, other, row)?;
827 Ok(Binding::Value(value_to_property_value(&value)))
828 }
829 }
830 }
831
832 /// Same sort as `apply_order_by`, but over `BindingRow`s (a `WITH`
833 /// clause's own ORDER BY, which must run before that row set becomes
834 /// the seed for the next `QueryPart` — sorting/limiting a WITH changes
835 /// *which* rows continue, not just their presentation order).
836 fn apply_order_by_bindings(
837 &self,
838 txn: Txn,
839 rows: Vec<BindingRow>,
840 order_by: &[(ReturnExpr, SortDir)],
841 limit: Option<i64>,
842 ) -> Result<Vec<BindingRow>, QueryError> {
843 let mut keyed: Vec<(Vec<Value>, BindingRow)> = Vec::with_capacity(rows.len());
844 for row in rows {
845 let value_map = self.binding_row_to_value_map(txn, &row)?;
846 let keys = order_by
847 .iter()
848 .map(|(expr, _)| eval_projected_expr(expr, &value_map))
849 .collect::<Result<Vec<_>, _>>()?;
850 keyed.push((keys, row));
851 }
852 Ok(top_k_by(keyed, order_by, limit).into_iter().map(|(_, row)| row).collect())
853 }
854
855 /// Sorts an already-`materialize_return`d result for a non-aggregating
856 /// `RETURN`, evaluating each ORDER BY expression against *both* the
857 /// pre-projection `BindingRow` it came from and its own projected
858 /// output columns overlaid on top — real Cypher allows ORDER BY to
859 /// reference either a RETURN alias or a still-in-scope variable that
860 /// wasn't returned at all, so neither view alone is enough (see the
861 /// call site in `execute_match`). `binding_rows` and `result.rows` are
862 /// the same length and pairwise correspond — `materialize_return`'s
863 /// non-aggregating path preserves row order 1:1 with its input.
864 fn apply_order_by_with_scope(
865 &self,
866 txn: Txn,
867 binding_rows: &[BindingRow],
868 result: QueryResult,
869 order_by: &[(ReturnExpr, SortDir)],
870 limit: Option<i64>,
871 ) -> Result<QueryResult, QueryError> {
872 let QueryResult { columns, rows } = result;
873 let mut keyed: Vec<(Vec<Value>, Vec<Value>)> = Vec::with_capacity(rows.len());
874 for (binding_row, row) in binding_rows.iter().zip(rows) {
875 let mut value_map = self.binding_row_to_value_map(txn, binding_row)?;
876 for (col, val) in columns.iter().zip(&row) {
877 value_map.insert(col.clone(), val.clone());
878 }
879 let keys = order_by
880 .iter()
881 .map(|(expr, _)| eval_projected_expr(expr, &value_map))
882 .collect::<Result<Vec<_>, _>>()?;
883 keyed.push((keys, row));
884 }
885 let rows = top_k_by(keyed, order_by, limit).into_iter().map(|(_, row)| row).collect();
886 Ok(QueryResult { columns, rows })
887 }
888
889 fn binding_row_to_value_map(
890 &self,
891 txn: Txn,
892 row: &BindingRow,
893 ) -> Result<HashMap<String, Value>, QueryError> {
894 let mut map = HashMap::with_capacity(row.len());
895 for (k, binding) in row {
896 map.insert(k.clone(), self.binding_to_value(txn, binding)?);
897 }
898 Ok(map)
899 }
900
901 /// Resolves a `Binding` to its display `Value` — a `Node`/`Edge`
902 /// binding fetches the full current record, a scalar `Value` binding
903 /// passes through (collapsing a stored `PropertyValue::Null` to
904 /// `Value::Null`, same as everywhere else null is represented).
905 fn binding_to_value(&self, txn: Txn, b: &Binding) -> Result<Value, QueryError> {
906 Ok(match b {
907 Binding::Node(id) => Value::Node(
908 GraphStore::get_node_in_txn(txn, *id)?
909 .expect("bound node exists within this statement's transaction"),
910 ),
911 Binding::Edge(id) => Value::Edge(
912 GraphStore::get_edge_in_txn(txn, *id)?
913 .expect("bound edge exists within this statement's transaction"),
914 ),
915 Binding::Value(PropertyValue::Null) => Value::Null,
916 Binding::Value(pv) => Value::Property(pv.clone()),
917 Binding::List(items) => Value::List(items.clone()),
918 Binding::Path(elems) => Value::Path(self.resolve_path_elems(txn, elems)?),
919 })
920 }
921
922 /// `binding_to_value`'s per-element helper for `Binding::Path` — fetches
923 /// each element's full current record, same "keep just the id in the
924 /// row, resolve to a full record only when materializing for display"
925 /// split `Binding::Node`/`Edge` already use above.
926 fn resolve_path_elems(&self, txn: Txn, elems: &[PathBinding]) -> Result<Vec<PathElem>, QueryError> {
927 elems
928 .iter()
929 .map(|e| {
930 Ok(match e {
931 PathBinding::Node(id) => PathElem::Node(
932 GraphStore::get_node_in_txn(txn, *id)?
933 .expect("bound node exists within this statement's transaction"),
934 ),
935 PathBinding::Edge(id) => PathElem::Edge(
936 GraphStore::get_edge_in_txn(txn, *id)?
937 .expect("bound edge exists within this statement's transaction"),
938 ),
939 })
940 })
941 .collect()
942 }
943
944 /// Folds `rows` into groups keyed by every non-aggregate item's per-row
945 /// `Binding` (via `item_binding`), then finishes each aggregate item's
946 /// accumulator per group. Returns one `Vec<Binding>` per output group,
947 /// column-aligned with `items`. Shared by `materialize_with` and
948 /// `materialize_return` — both already take the same `rows: &[BindingRow]`
949 /// input type, so the grouping core stays in `Binding`-space (preserving
950 /// graph identity for bare-var grouping keys) and each caller does its
951 /// own thin final conversion.
952 ///
953 /// Grouping-key lookup is a hash-map lookup (`group_index`, keyed by
954 /// `binding_hash_key`'s output — `Binding`/`PropertyValue` don't
955 /// derive `Eq`/`Hash` themselves, `PropertyValue::Float` can't, so
956 /// `HashKey` stands in for them; see its docs) into `groups`, which
957 /// stays a plain `Vec` for insertion-order-stable output when there's
958 /// no ORDER BY. O(1) average per row, not the O(rows × groups) linear
959 /// scan this used to be — see BENCHMARKS.md for the measured
960 /// before/after.
961 ///
962 /// Callers must call `validate_return_items` first — this function
963 /// assumes every aggregate `Call` item has already been checked to
964 /// have exactly one argument.
965 fn resolve_grouped_rows(
966 &self,
967 txn: Txn,
968 items: &[ReturnItem],
969 rows: &[BindingRow],
970 ) -> Result<Vec<Vec<Binding>>, QueryError> {
971 struct Group {
972 // Aligned to `items`: `Some` at a non-aggregate item's index,
973 // `None` at an aggregate item's index (both vecs below are
974 // index-aligned to `items` the same way, so exactly one of
975 // `key_bindings[i]`/`accs[i]` is populated per `i`).
976 key_bindings: Vec<Option<Binding>>,
977 accs: Vec<Option<AggAcc>>,
978 row_count: i64,
979 }
980 fn fresh_accs(items: &[ReturnItem]) -> Vec<Option<AggAcc>> {
981 items
982 .iter()
983 .map(|item| match &item.expr {
984 ReturnExpr::Call { name, distinct, .. } if is_aggregate_name(name) => {
985 Some(AggAcc::identity(name, *distinct))
986 }
987 _ => None,
988 })
989 .collect()
990 }
991
992 // Groups live in `groups` (insertion order, for stable output when
993 // there's no ORDER BY) with `group_index` as a hash-based lookup
994 // into it, keyed by a hashable stand-in for `key_bindings` (see
995 // `HashKey` — `Binding`/`PropertyValue` don't derive `Eq`/`Hash`
996 // themselves, `PropertyValue::Float` can't). O(1) average lookup
997 // per row instead of the O(groups) linear scan this replaced —
998 // see BENCHMARKS.md for the measured before/after.
999 let mut groups: Vec<Group> = Vec::new();
1000 let mut group_index: HashMap<Vec<Option<HashKey>>, usize> = HashMap::new();
1001 for row in rows {
1002 let mut key_bindings = Vec::with_capacity(items.len());
1003 for item in items {
1004 key_bindings.push(if is_top_level_aggregate(&item.expr) {
1005 None
1006 } else {
1007 Some(self.item_binding(txn, &item.expr, row)?)
1008 });
1009 }
1010 let hash_key: Vec<Option<HashKey>> = key_bindings
1011 .iter()
1012 .map(|b| b.as_ref().map(binding_hash_key).transpose())
1013 .collect::<Result<Vec<_>, _>>()?;
1014 let group_idx = *group_index.entry(hash_key).or_insert_with(|| {
1015 groups.push(Group {
1016 key_bindings: key_bindings.clone(),
1017 accs: fresh_accs(items),
1018 row_count: 0,
1019 });
1020 groups.len() - 1
1021 });
1022 let group = &mut groups[group_idx];
1023 group.row_count += 1;
1024 for (i, item) in items.iter().enumerate() {
1025 let ReturnExpr::Call { args, .. } = &item.expr else { continue };
1026 if !is_top_level_aggregate(&item.expr) {
1027 continue;
1028 }
1029 // Standard Cypher null-skipping: a null argument (e.g. an
1030 // unmatched OPTIONAL MATCH variable) contributes to
1031 // neither the accumulator nor its DISTINCT dedup set —
1032 // this is what makes `count(x)` exclude a null-padded row
1033 // while `count(*)` (tracked via `row_count`, not an
1034 // accumulator at all) includes it.
1035 let value = self.eval_return_expr(txn, &args[0], row)?;
1036 if !matches!(value, Value::Null) {
1037 if let Some(acc) = &mut group.accs[i] {
1038 acc.fold(&value)?;
1039 }
1040 }
1041 }
1042 }
1043
1044 // Global aggregate over an empty result set (no grouping-key items
1045 // at all, and no rows to seed a group from) still produces exactly
1046 // one output row — `count`/`count(*)` -> 0, `sum` -> 0,
1047 // `avg`/`min`/`max` -> Null, `collect` -> [] — via the same
1048 // fresh-accumulator `finish()` path a normal empty-contribution
1049 // group already uses below, not a separate code path.
1050 let no_key_items = items.iter().all(|item| is_top_level_aggregate(&item.expr));
1051 if groups.is_empty() && no_key_items {
1052 groups.push(Group {
1053 key_bindings: vec![None; items.len()],
1054 accs: fresh_accs(items),
1055 row_count: 0,
1056 });
1057 }
1058
1059 let mut out = Vec::with_capacity(groups.len());
1060 for mut group in groups {
1061 let mut row_out = Vec::with_capacity(items.len());
1062 for (i, item) in items.iter().enumerate() {
1063 let binding = if matches!(item.expr, ReturnExpr::CountStar) {
1064 Binding::Value(PropertyValue::Int(group.row_count))
1065 } else if is_top_level_aggregate(&item.expr) {
1066 let value = group.accs[i]
1067 .take()
1068 .expect("aggregate item must have an accumulator")
1069 .finish();
1070 value_to_binding(value)
1071 } else {
1072 group.key_bindings[i].clone().expect("non-aggregate item must have a key binding")
1073 };
1074 row_out.push(binding);
1075 }
1076 out.push(row_out);
1077 }
1078 Ok(out)
1079 }
1080
1081 /// WITH's HAVING-equivalent — evaluated against the already-projected/
1082 /// grouped row, same as ORDER BY. Never pushed into the planner (see
1083 /// `WithExpr`'s docs).
1084 /// `Option<bool>` — `None` is Cypher's "unknown" (see `compare()`'s
1085 /// docs), propagated through `AND`/`OR`/`NOT` via `and3`/`or3`/`map`
1086 /// instead of collapsing to `false` partway through. Every call site
1087 /// filters a row by checking `== Some(true)` — unknown behaves like
1088 /// `false` for filtering purposes, but *only* at that final step, not
1089 /// internally, since `AND`/`OR`'s truth tables need to tell "false"
1090 /// and "unknown" apart to combine correctly.
1091 fn eval_with_expr(&self, txn: Txn, expr: &WithExpr, row: &BindingRow) -> Result<Option<bool>, QueryError> {
1092 Ok(match expr {
1093 WithExpr::And(l, r) => and3(self.eval_with_expr(txn, l, row)?, self.eval_with_expr(txn, r, row)?),
1094 WithExpr::Or(l, r) => or3(self.eval_with_expr(txn, l, row)?, self.eval_with_expr(txn, r, row)?),
1095 WithExpr::Not(e) => self.eval_with_expr(txn, e, row)?.map(|b| !b),
1096 WithExpr::Compare(lhs, op, lit) => {
1097 let value = self.eval_return_expr(txn, lhs, row)?;
1098 compare_value(&value, *op, lit)
1099 }
1100 })
1101 }
1102
1103 /// Evaluates an `OPTIONAL MATCH` part with left-outer-join semantics:
1104 /// every outer row survives, whether or not the optional pattern
1105 /// matched anything for it. Must wrap the *whole* subplan rather than
1106 /// null-padding inside `Expand`/`VarExpand` themselves — baking it in
1107 /// there would turn every default (non-optional) `Expand` into a
1108 /// left-outer-join too (breaking existing inner-join semantics), and
1109 /// would mis-handle multi-hop optional patterns: IS7's optional
1110 /// pattern is 2 hops, and per-hop null-padding would emit one
1111 /// null-padded row per *hop-1* match even when hop 2 also matched,
1112 /// instead of collapsing to exactly one row per outer row that had
1113 /// zero end-to-end matches.
1114 ///
1115 /// Implementation: tag each outer row with its index, evaluate the
1116 /// subplan once over the whole tagged batch (a single seed, not one
1117 /// call per row), group results back by that index, then for any
1118 /// outer index with zero results, emit the outer row unchanged plus
1119 /// `Null` for every variable the optional pattern would have newly
1120 /// introduced.
1121 fn eval_optional_part(
1122 &self,
1123 txn: Txn,
1124 plan: &LogicalPlan,
1125 outer_rows: &[BindingRow],
1126 new_vars: &HashSet<String>,
1127 ) -> Result<Vec<BindingRow>, QueryError> {
1128 let tagged: Vec<BindingRow> = outer_rows
1129 .iter()
1130 .enumerate()
1131 .map(|(i, row)| {
1132 let mut r = row.clone();
1133 r.insert(OPTIONAL_SEED_IDX_KEY.to_string(), Binding::Value(PropertyValue::Int(i as i64)));
1134 r
1135 })
1136 .collect();
1137 let results = self.eval_plan(txn, plan, &tagged)?;
1138 let mut by_idx: HashMap<i64, Vec<BindingRow>> = HashMap::new();
1139 for mut row in results {
1140 let idx = match row.remove(OPTIONAL_SEED_IDX_KEY) {
1141 Some(Binding::Value(PropertyValue::Int(i))) => i,
1142 other => unreachable!("__seed_idx tagged internally as Binding::Value(Int), got {other:?}"),
1143 };
1144 by_idx.entry(idx).or_default().push(row);
1145 }
1146 let mut out = Vec::with_capacity(outer_rows.len());
1147 for (i, outer_row) in outer_rows.iter().enumerate() {
1148 match by_idx.remove(&(i as i64)) {
1149 Some(matches) => out.extend(matches),
1150 None => {
1151 let mut padded = outer_row.clone();
1152 for var in new_vars {
1153 padded.insert(var.clone(), Binding::Value(PropertyValue::Null));
1154 }
1155 out.push(padded);
1156 }
1157 }
1158 }
1159 Ok(out)
1160 }
1161
1162 fn eval_plan(
1163 &self,
1164 txn: Txn,
1165 plan: &LogicalPlan,
1166 seed: &[BindingRow],
1167 ) -> Result<Vec<BindingRow>, QueryError> {
1168 match plan {
1169 LogicalPlan::Seed { var } => {
1170 debug_assert!(
1171 seed.first().is_none_or(|row| row.contains_key(var)),
1172 "Seed{{var: {var:?}}} planned for a var not present in the carried-forward rows"
1173 );
1174 Ok(seed.to_vec())
1175 }
1176 LogicalPlan::AllNodesScan { var } => self.scan(txn, var, None, seed, None),
1177 LogicalPlan::NodeByLabelScan { var, label } => self.scan(txn, var, Some(label), seed, None),
1178 LogicalPlan::Expand {
1179 input,
1180 from_var,
1181 to_var,
1182 rel_var,
1183 rel_label,
1184 direction,
1185 } => {
1186 let base_rows = self.eval_plan(txn, input, seed)?;
1187 let mut out = Vec::new();
1188 for row in base_rows {
1189 let from_id = match row.get(from_var) {
1190 Some(Binding::Node(id)) => *id,
1191 // A null `from_var` (padded by an outer, already-
1192 // resolved `OPTIONAL MATCH` that didn't match) has
1193 // no neighbors, same as any other traversal from
1194 // null -- contributes zero rows, not an error. A
1195 // truly missing/wrong-typed binding still is one.
1196 Some(Binding::Value(PropertyValue::Null)) => continue,
1197 _ => return Err(QueryError::UnboundVariable(from_var.clone())),
1198 };
1199 let entries = neighbors_for_direction(txn, from_id, *direction, rel_label.as_deref())?;
1200 for entry in entries {
1201 let mut new_row = row.clone();
1202 new_row.insert(to_var.clone(), Binding::Node(entry.other));
1203 if let Some(rv) = rel_var {
1204 new_row.insert(rv.clone(), Binding::Edge(entry.edge_id));
1205 }
1206 out.push(new_row);
1207 }
1208 }
1209 Ok(out)
1210 }
1211 LogicalPlan::VarExpand {
1212 input,
1213 from_var,
1214 to_var,
1215 rel_label,
1216 direction,
1217 min_hops,
1218 max_hops,
1219 } => {
1220 let base_rows = self.eval_plan(txn, input, seed)?;
1221 let mut out = Vec::new();
1222 let unbounded = max_hops.is_none();
1223 let effective_max = max_hops.unwrap_or(VAR_EXPAND_DEPTH_CAP);
1224 for row in base_rows {
1225 let start_id = match row.get(from_var) {
1226 Some(Binding::Node(id)) => *id,
1227 // Same null-propagation as `Expand` above.
1228 Some(Binding::Value(PropertyValue::Null)) => continue,
1229 _ => return Err(QueryError::UnboundVariable(from_var.clone())),
1230 };
1231 let mut visited = HashSet::new();
1232 visited.insert(start_id);
1233 if *min_hops == 0 {
1234 let mut new_row = row.clone();
1235 new_row.insert(to_var.clone(), Binding::Node(start_id));
1236 out.push(new_row);
1237 }
1238 let mut frontier = vec![start_id];
1239 let mut depth = 0u32;
1240 while depth < effective_max && !frontier.is_empty() {
1241 depth += 1;
1242 let mut next_frontier = Vec::new();
1243 for node in frontier {
1244 let entries = neighbors_for_direction(txn, node, *direction, rel_label.as_deref())?;
1245 for entry in entries {
1246 if visited.insert(entry.other) {
1247 next_frontier.push(entry.other);
1248 if depth >= *min_hops {
1249 let mut new_row = row.clone();
1250 new_row.insert(to_var.clone(), Binding::Node(entry.other));
1251 out.push(new_row);
1252 }
1253 }
1254 }
1255 }
1256 frontier = next_frontier;
1257 if depth == effective_max && unbounded && !frontier.is_empty() {
1258 // Unbounded (`*N..`) traversal hit the safety
1259 // cap with more still reachable — error rather
1260 // than silently truncate results, which would
1261 // be a wrong-answer failure mode for a
1262 // correctness-benchmark tool.
1263 return Err(QueryError::Parse(format!(
1264 "variable-length traversal exceeded the safety depth cap ({VAR_EXPAND_DEPTH_CAP} \
1265 hops) — likely a cyclic graph or unexpectedly large fanout; narrow the pattern or \
1266 add an explicit upper bound (e.g. *0..10)"
1267 )));
1268 }
1269 }
1270 }
1271 Ok(out)
1272 }
1273 LogicalPlan::Filter { input, predicate } => {
1274 let rows = self.eval_plan(txn, input, seed)?;
1275 let mut out = Vec::with_capacity(rows.len());
1276 for row in rows {
1277 if self.eval_expr(txn, predicate, &row)? == Some(true) {
1278 out.push(row);
1279 }
1280 }
1281 Ok(out)
1282 }
1283 }
1284 }
1285
1286 /// Cross-joins the scan against `seed` — for the first `QueryPart` in a
1287 /// statement, `seed` is always exactly one empty row (see
1288 /// `execute_match`), so this reduces to "one row per scanned node,"
1289 /// the same as before this scan ever needed a `seed` parameter at
1290 /// all. It matters for a later `QueryPart` (after a `WITH` boundary)
1291 /// whose pattern doesn't chain from an already-bound variable — e.g.
1292 /// `MATCH (a) WITH a MATCH (b) ...` — real Cypher's cross-join
1293 /// semantics require every carried-forward binding (`a`) to survive
1294 /// alongside every row this scan produces (`b`), not get silently
1295 /// dropped. This is a real cost, not just a correctness fix: a scan
1296 /// against N carried rows does N × (scanned rows) work, same as any
1297 /// cross join.
1298 /// `row_limit` bounds the underlying storage scan itself (see
1299 /// `GraphStore::all_nodes_limited_in_txn`) -- only ever `Some` from the
1300 /// dedicated shortcut in `execute_match` for a plan that's *just* this
1301 /// one scan feeding straight into `LIMIT`, nothing else (no `Filter`,
1302 /// no `Expand`, no `ORDER BY`). Every other caller (the general
1303 /// `eval_plan` recursion) passes `None`, since capping the raw scan is
1304 /// only safe when nothing downstream could still drop a row.
1305 fn scan(
1306 &self,
1307 txn: Txn,
1308 var: &str,
1309 label: Option<&str>,
1310 seed: &[BindingRow],
1311 row_limit: Option<usize>,
1312 ) -> Result<Vec<BindingRow>, QueryError> {
1313 let nodes = match row_limit {
1314 Some(limit) => GraphStore::all_nodes_limited_in_txn(txn, label, limit)?,
1315 None => GraphStore::all_nodes_in_txn(txn, label)?,
1316 };
1317 let mut out = Vec::with_capacity(seed.len() * nodes.len());
1318 for base_row in seed {
1319 for n in &nodes {
1320 let mut row = base_row.clone();
1321 row.insert(var.to_string(), Binding::Node(n.id));
1322 out.push(row);
1323 }
1324 }
1325 Ok(out)
1326 }
1327
1328 /// `Option<bool>` — see `eval_with_expr`'s docs, same reasoning.
1329 /// `HasLabel`/`VarEq` never produce "unknown" (they operate on real
1330 /// bound node/edge identity, not a possibly-null property), so they
1331 /// always return `Some`.
1332 fn eval_expr(&self, txn: Txn, expr: &Expr, row: &BindingRow) -> Result<Option<bool>, QueryError> {
1333 Ok(match expr {
1334 Expr::And(l, r) => and3(self.eval_expr(txn, l, row)?, self.eval_expr(txn, r, row)?),
1335 Expr::Or(l, r) => or3(self.eval_expr(txn, l, row)?, self.eval_expr(txn, r, row)?),
1336 Expr::Not(e) => self.eval_expr(txn, e, row)?.map(|b| !b),
1337 Expr::Compare(pa, op, lit) => {
1338 let prop_value = self.lookup_prop(txn, pa, row)?;
1339 compare(&prop_value, *op, lit)
1340 }
1341 Expr::HasLabel(var, label) => {
1342 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
1343 let Binding::Node(id) = binding else {
1344 return Err(QueryError::UnboundVariable(var.clone()));
1345 };
1346 let node = GraphStore::get_node_in_txn(txn, *id)?;
1347 Some(node.is_some_and(|n| n.labels.iter().any(|l| l == label)))
1348 }
1349 Expr::VarEq(a, b) => {
1350 let ba = row.get(a).ok_or_else(|| QueryError::UnboundVariable(a.clone()))?;
1351 let bb = row.get(b).ok_or_else(|| QueryError::UnboundVariable(b.clone()))?;
1352 Some(match (ba, bb) {
1353 (Binding::Node(x), Binding::Node(y)) => x == y,
1354 (Binding::Edge(x), Binding::Edge(y)) => x == y,
1355 // A null-padded `Binding::Value` (from an earlier
1356 // OPTIONAL MATCH that didn't match) can't equal a
1357 // real node/edge, and comparing across binding kinds
1358 // (a node vs an edge) is never meaningful here — the
1359 // planner only ever synthesizes VarEq between two
1360 // occurrences of the same pattern variable, which are
1361 // always the same kind when both are real.
1362 _ => false,
1363 })
1364 }
1365 })
1366 }
1367
1368 fn lookup_prop(
1369 &self,
1370 txn: Txn,
1371 pa: &PropAccess,
1372 row: &BindingRow,
1373 ) -> Result<Option<PropertyValue>, QueryError> {
1374 let binding = row
1375 .get(&pa.var)
1376 .ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
1377 match binding {
1378 Binding::Node(id) => {
1379 let node = GraphStore::get_node_in_txn(txn, *id)?;
1380 Ok(node.and_then(|n| n.props.get(&pa.prop).cloned()))
1381 }
1382 Binding::Edge(id) => {
1383 let edge = GraphStore::get_edge_in_txn(txn, *id)?;
1384 Ok(edge.and_then(|e| e.props.get(&pa.prop).cloned()))
1385 }
1386 // A WITH-projected scalar (or list/path) has no `.prop` to
1387 // access — e.g. `WITH message.id AS messageId` then
1388 // `messageId.foo` isn't meaningful. Treat as absent rather
1389 // than erroring, consistent with how a missing property
1390 // already behaves.
1391 Binding::Value(_) | Binding::List(_) | Binding::Path(_) => Ok(None),
1392 }
1393 }
1394
1395 fn materialize_return(
1396 &self,
1397 txn: Txn,
1398 items: &[ReturnItem],
1399 rows: &[BindingRow],
1400 distinct: bool,
1401 ) -> Result<QueryResult, QueryError> {
1402 let columns = items
1403 .iter()
1404 .enumerate()
1405 .map(|(i, item)| item.alias.clone().unwrap_or_else(|| default_column_name(&item.expr, i)))
1406 .collect();
1407 let mut out_rows = if !has_aggregate(items) {
1408 let mut out_rows = Vec::with_capacity(rows.len());
1409 for row in rows {
1410 let mut out_row = Vec::with_capacity(items.len());
1411 for item in items {
1412 out_row.push(self.eval_return_expr(txn, &item.expr, row)?);
1413 }
1414 out_rows.push(out_row);
1415 }
1416 out_rows
1417 } else {
1418 validate_return_items(items)?;
1419 let grouped = self.resolve_grouped_rows(txn, items, rows)?;
1420 grouped
1421 .into_iter()
1422 .map(|bindings| {
1423 bindings
1424 .iter()
1425 .map(|b| self.binding_to_value(txn, b))
1426 .collect::<Result<Vec<_>, _>>()
1427 })
1428 .collect::<Result<Vec<_>, _>>()?
1429 };
1430 if distinct {
1431 out_rows = dedup_rows(out_rows)?;
1432 }
1433 Ok(QueryResult {
1434 columns,
1435 rows: out_rows,
1436 })
1437 }
1438
1439 fn eval_return_expr(
1440 &self,
1441 txn: Txn,
1442 expr: &ReturnExpr,
1443 row: &BindingRow,
1444 ) -> Result<Value, QueryError> {
1445 match expr {
1446 ReturnExpr::Var(var) => {
1447 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
1448 self.binding_to_value(txn, binding)
1449 }
1450 ReturnExpr::Prop(pa) => {
1451 let value = self.lookup_prop(txn, pa, row)?;
1452 Ok(match value {
1453 // Collapse "prop missing" and "prop stored as null" into
1454 // one null representation — see Value::Null docs.
1455 Some(PropertyValue::Null) | None => Value::Null,
1456 Some(pv) => Value::Property(pv),
1457 })
1458 }
1459 ReturnExpr::Lit(lit) => Ok(match lit {
1460 Literal::Null => Value::Null,
1461 other => Value::Literal(other.clone()),
1462 }),
1463 ReturnExpr::Call { name, args, .. } => {
1464 // Reaching here with an aggregate name means an aggregate
1465 // call slipped past `validate_return_items` (which only
1466 // allows one at a return item's top level) — grouping
1467 // itself never calls `eval_return_expr` on the aggregate
1468 // wrapper, only on each aggregate's own argument
1469 // subexpression (see `resolve_grouped_rows`), so this is
1470 // an internal-consistency error, not a normal user path.
1471 if is_aggregate_name(name) {
1472 return Err(QueryError::Parse(format!(
1473 "aggregate function '{name}' can only be used as a return item's top-level expression"
1474 )));
1475 }
1476 let arg_values = args
1477 .iter()
1478 .map(|a| self.eval_return_expr(txn, a, row))
1479 .collect::<Result<Vec<_>, _>>()?;
1480 call_builtin(name, &arg_values)
1481 }
1482 ReturnExpr::CountStar => Err(QueryError::Parse(
1483 "count(*) can only be used as a return item's top-level expression".into(),
1484 )),
1485 ReturnExpr::Case { test, whens, else_ } => {
1486 let test_value = match test {
1487 Some(t) => Some(self.eval_return_expr(txn, t, row)?),
1488 None => None,
1489 };
1490 for (when, then) in whens {
1491 let when_value = self.eval_return_expr(txn, when, row)?;
1492 // Deliberately reuses the same Null == Null -> true
1493 // convention as `compare()` below, not standard
1494 // three-valued NULL logic — IS7's `CASE r WHEN null
1495 // THEN false ELSE true END` depends on this exact
1496 // semantics to detect an OPTIONAL MATCH non-match.
1497 let matched = match &test_value {
1498 Some(tv) => value_eq(tv, &when_value),
1499 None => matches!(when_value, Value::Literal(Literal::Bool(true))),
1500 };
1501 if matched {
1502 return self.eval_return_expr(txn, then, row);
1503 }
1504 }
1505 match else_ {
1506 Some(e) => self.eval_return_expr(txn, e, row),
1507 None => Ok(Value::Null),
1508 }
1509 }
1510 }
1511 }
1512
1513 fn materialize_delete(
1514 &self,
1515 write_txn: &WriteTransaction,
1516 vars: &[String],
1517 rows: &[BindingRow],
1518 detach: bool,
1519 ) -> Result<QueryResult, QueryError> {
1520 let mut deleted_nodes = HashSet::new();
1521 let mut deleted_edges = HashSet::new();
1522 for row in rows {
1523 for var in vars {
1524 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
1525 match binding {
1526 Binding::Node(id) => {
1527 if deleted_nodes.insert(*id) {
1528 GraphStore::delete_node_in_txn(write_txn, *id, detach)?;
1529 }
1530 }
1531 Binding::Edge(id) => {
1532 if deleted_edges.insert(*id) {
1533 GraphStore::delete_edge_in_txn(write_txn, *id)?;
1534 }
1535 }
1536 // A null binding is a real, legal DELETE target -- an
1537 // `OPTIONAL MATCH` that didn't match pads its variables
1538 // with null, and deleting that null is specified as a
1539 // silent no-op, not an error (real Cypher: "deleting
1540 // null does nothing").
1541 Binding::Value(PropertyValue::Null) => {}
1542 Binding::Value(_) | Binding::List(_) | Binding::Path(_) => {
1543 return Err(QueryError::UnboundVariable(format!(
1544 "'{var}' is a WITH-projected scalar, not a node/edge — DELETE needs a graph binding"
1545 )))
1546 }
1547 }
1548 }
1549 }
1550 Ok(QueryResult {
1551 columns: vec![],
1552 rows: vec![],
1553 })
1554 }
1555
1556 fn materialize_set(
1557 &self,
1558 write_txn: &WriteTransaction,
1559 items: &[SetItem],
1560 rows: &[BindingRow],
1561 ) -> Result<QueryResult, QueryError> {
1562 for row in rows {
1563 for item in items {
1564 apply_set_item(write_txn, row, item)?;
1565 }
1566 }
1567 Ok(QueryResult {
1568 columns: vec![],
1569 rows: vec![],
1570 })
1571 }
1572
1573 fn materialize_remove(
1574 &self,
1575 write_txn: &WriteTransaction,
1576 items: &[RemoveItem],
1577 rows: &[BindingRow],
1578 ) -> Result<QueryResult, QueryError> {
1579 for row in rows {
1580 for item in items {
1581 apply_remove_item(write_txn, row, item)?;
1582 }
1583 }
1584 Ok(QueryResult {
1585 columns: vec![],
1586 rows: vec![],
1587 })
1588 }
1589}
1590
1591fn apply_set_item(write_txn: &WriteTransaction, row: &BindingRow, item: &SetItem) -> Result<(), QueryError> {
1592 match item {
1593 SetItem::Prop(pa, lit) => {
1594 let binding = row.get(&pa.var).ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
1595 let value = literal_to_value(lit);
1596 match binding {
1597 Binding::Node(id) => {
1598 GraphStore::set_node_prop_in_txn(write_txn, *id, &pa.prop, value)?;
1599 }
1600 Binding::Edge(id) => {
1601 GraphStore::set_edge_prop_in_txn(write_txn, *id, &pa.prop, value)?;
1602 }
1603 Binding::Value(_) | Binding::List(_) | Binding::Path(_) => {
1604 return Err(QueryError::UnboundVariable(format!(
1605 "'{}' is a WITH-projected scalar, not a node/edge — SET needs a graph binding",
1606 pa.var
1607 )))
1608 }
1609 }
1610 }
1611 SetItem::Labels(var, labels) => {
1612 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
1613 let Binding::Node(id) = binding else {
1614 return Err(QueryError::UnboundVariable(format!(
1615 "'{var}' isn't a node — SET can only add labels to a node"
1616 )));
1617 };
1618 for label in labels {
1619 GraphStore::add_node_label_in_txn(write_txn, *id, label)?;
1620 }
1621 }
1622 }
1623 Ok(())
1624}
1625
1626fn apply_remove_item(write_txn: &WriteTransaction, row: &BindingRow, item: &RemoveItem) -> Result<(), QueryError> {
1627 match item {
1628 RemoveItem::Prop(pa) => {
1629 let binding = row.get(&pa.var).ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
1630 match binding {
1631 Binding::Node(id) => {
1632 GraphStore::remove_node_prop_in_txn(write_txn, *id, &pa.prop)?;
1633 }
1634 Binding::Edge(id) => {
1635 GraphStore::remove_edge_prop_in_txn(write_txn, *id, &pa.prop)?;
1636 }
1637 Binding::Value(_) | Binding::List(_) | Binding::Path(_) => {
1638 return Err(QueryError::UnboundVariable(format!(
1639 "'{}' is a WITH-projected scalar, not a node/edge — REMOVE needs a graph binding",
1640 pa.var
1641 )))
1642 }
1643 }
1644 }
1645 RemoveItem::Labels(var, labels) => {
1646 let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
1647 let Binding::Node(id) = binding else {
1648 return Err(QueryError::UnboundVariable(format!(
1649 "'{var}' isn't a node — REMOVE can only remove labels from a node"
1650 )));
1651 };
1652 for label in labels {
1653 GraphStore::remove_node_label_in_txn(write_txn, *id, label)?;
1654 }
1655 }
1656 }
1657 Ok(())
1658}
1659
1660/// A statement never mutates anything iff it's a `MATCH ... RETURN` with no
1661/// `DELETE`/`DETACH DELETE`/`SET` tail *and* no `MERGE` clause anywhere in
1662/// it (`MERGE (n) RETURN n` has a `Tail::Return`, but still writes whenever
1663/// it has to create — checking `tail` alone here would be a real bug, not
1664/// just an incomplete check: it would send a MERGE-that-creates through a
1665/// `ReadTransaction`, which has no `.insert`). `Statement::Create` and
1666/// every other `Tail` variant always write. Confirmed by tracing every
1667/// function reachable from pattern/WHERE/WITH evaluation: none of them
1668/// ever call a table-mutating `*_in_txn` method for a `Tail::Return`
1669/// statement with no `MERGE` clause (a label-filtered scan looks up an
1670/// existing label id, it never allocates one — allocation only happens in
1671/// `create_node_in_txn`/`create_edge_in_txn`). `Executor::execute` uses
1672/// this to decide whether to open a `ReadTransaction` (no contention with
1673/// concurrent readers or a concurrent writer) or a `WriteTransaction`.
1674fn is_read_only(stmt: &Statement) -> bool {
1675 let Statement::Match { tail: Some(Tail::Return(_, _)), clauses, .. } = stmt else {
1676 return false;
1677 };
1678 !clauses.iter().any(|c| matches!(c, QueryClause::Merge(_)))
1679}
1680
1681/// Recovers the real `&WriteTransaction` from a `Txn` for the two
1682/// `execute_match` tail arms (`DELETE`/`SET`) that need `.insert`/
1683/// `.remove`, not just `Txn`'s read-only `get`/`iter`. Panics if given
1684/// `Txn::Read` — which can't happen: `Tail::Delete`/`DetachDelete`/`Set`
1685/// make `is_read_only` return `false`, so `Executor::execute` always opens
1686/// a `WriteTransaction` (and thus `Txn::Write`) before reaching this path.
1687fn require_write_txn(txn: Txn<'_>) -> &WriteTransaction {
1688 let Txn::Write(write_txn) = txn else {
1689 unreachable!(
1690 "materialize_delete/materialize_set only reached via the write-dispatch path in \
1691 Executor::execute — is_read_only(stmt) is false for any statement with a Delete/ \
1692 DetachDelete/Set tail, so execute always opens a WriteTransaction for these"
1693 )
1694 };
1695 write_txn
1696}
1697
1698fn default_column_name(expr: &ReturnExpr, idx: usize) -> String {
1699 match expr {
1700 ReturnExpr::Var(v) => v.clone(),
1701 ReturnExpr::Prop(pa) => format!("{}.{}", pa.var, pa.prop),
1702 ReturnExpr::Lit(_) => format!("col{idx}"),
1703 ReturnExpr::Call { name, .. } => format!("{name}(...)"),
1704 ReturnExpr::CountStar => "count(*)".to_string(),
1705 ReturnExpr::Case { .. } => format!("case{idx}"),
1706 }
1707}
1708
1709/// The name a `WITH`/`RETURN` item is known by afterward — its alias, or
1710/// a name derived from the expression (its bare var name, `col{i}`, etc).
1711fn with_item_output_name((i, item): (usize, &ReturnItem)) -> String {
1712 item.alias.clone().unwrap_or_else(|| default_column_name(&item.expr, i))
1713}
1714
1715/// True iff `expr` is itself an aggregate call — `count(*)`, or a `Call`
1716/// whose name is in `is_aggregate_name`'s fixed set. Does NOT look inside
1717/// `expr` for a nested aggregate — see `contains_aggregate` for that.
1718fn is_top_level_aggregate(expr: &ReturnExpr) -> bool {
1719 match expr {
1720 ReturnExpr::CountStar => true,
1721 ReturnExpr::Call { name, .. } => is_aggregate_name(name),
1722 _ => false,
1723 }
1724}
1725
1726/// True iff `expr` contains an aggregate call anywhere inside it, at any
1727/// depth — used to reject an aggregate nested inside another aggregate's
1728/// argument, or inside a non-aggregate expression's `CASE`/`Call`
1729/// arguments (an aggregate must be a return item's *entire* top-level
1730/// expression — see `validate_return_items`).
1731fn contains_aggregate(expr: &ReturnExpr) -> bool {
1732 match expr {
1733 ReturnExpr::CountStar => true,
1734 ReturnExpr::Call { name, args, .. } => is_aggregate_name(name) || args.iter().any(contains_aggregate),
1735 ReturnExpr::Case { test, whens, else_ } => {
1736 test.as_deref().is_some_and(contains_aggregate)
1737 || whens.iter().any(|(w, t)| contains_aggregate(w) || contains_aggregate(t))
1738 || else_.as_deref().is_some_and(contains_aggregate)
1739 }
1740 ReturnExpr::Var(_) | ReturnExpr::Prop(_) | ReturnExpr::Lit(_) => false,
1741 }
1742}
1743
1744/// True iff any item's top-level expression is an aggregate call —
1745/// `materialize_with`/`materialize_return` dispatch to the grouping path
1746/// iff this is true, otherwise the existing row-at-a-time path runs
1747/// completely unchanged (zero perf/behavior impact on non-aggregating
1748/// queries).
1749fn has_aggregate(items: &[ReturnItem]) -> bool {
1750 items.iter().any(|item| is_top_level_aggregate(&item.expr))
1751}
1752
1753/// Validates a RETURN/WITH item list before any row is processed: every
1754/// aggregate call has exactly one argument (`count(*)`, the zero-argument
1755/// form, is `CountStar`, a separate variant — never reaches the `Call`
1756/// arm here), no aggregate's own argument contains a nested aggregate
1757/// call, and no non-aggregate item's expression contains an aggregate
1758/// call anywhere inside it (aggregates must be a return item's entire
1759/// top-level expression — justified by there being no arithmetic
1760/// operators anywhere in this engine yet, so `count(n) * 2`-style
1761/// composition is already impossible, and nothing in the target query set
1762/// needs an aggregate nested inside a `CASE` branch).
1763fn validate_return_items(items: &[ReturnItem]) -> Result<(), QueryError> {
1764 for item in items {
1765 match &item.expr {
1766 ReturnExpr::CountStar => {}
1767 ReturnExpr::Call { name, args, .. } if is_aggregate_name(name) => {
1768 if args.len() != 1 {
1769 return Err(QueryError::Parse(format!(
1770 "{name}() takes exactly one argument (use count(*) for a row count with no argument)"
1771 )));
1772 }
1773 if contains_aggregate(&args[0]) {
1774 return Err(QueryError::Parse(format!(
1775 "aggregate function '{name}' can't take another aggregate as an argument"
1776 )));
1777 }
1778 }
1779 other => {
1780 if contains_aggregate(other) {
1781 return Err(QueryError::Parse(
1782 "an aggregate function must be a return item's entire expression, not nested inside \
1783 another expression"
1784 .into(),
1785 ));
1786 }
1787 }
1788 }
1789 }
1790 Ok(())
1791}
1792
1793/// Grouping-key hashing — deliberately at the `Binding` level (`NodeId`/
1794/// `EdgeId`/`PropertyValue`), not `Value`: cheaper (no `GraphStore` fetch
1795/// just to compute) and the correct semantics (two `Binding::Node`s are
1796/// the same group iff the same node **identity**, not equal-by-struct-
1797/// contents). `Binding::List`'s elements are `Value`s already, so those
1798/// delegate to `value_hash_key` directly.
1799fn binding_hash_key(b: &Binding) -> Result<HashKey, QueryError> {
1800 Ok(match b {
1801 Binding::Node(id) => HashKey::Node(*id),
1802 Binding::Edge(id) => HashKey::Edge(*id),
1803 Binding::Value(pv) => property_value_hash_key(pv),
1804 Binding::List(items) => HashKey::List(items.iter().map(value_hash_key).collect::<Result<Vec<_>, _>>()?),
1805 // Explicit error, not a silent hash-by-something-arbitrary —
1806 // grouping/collecting by a captured path isn't a case any real
1807 // usage needs, and this codebase's stance is to reject an
1808 // untested shape rather than guess at its semantics.
1809 Binding::Path(_) => {
1810 return Err(QueryError::Parse(
1811 "grouping or collecting by a path (e.g. a named-path/shortestPath() variable) isn't supported"
1812 .into(),
1813 ))
1814 }
1815 })
1816}
1817
1818/// Converts a finished `AggAcc::finish()` result to the `Binding` it's
1819/// carried as through a `WITH` boundary — `collect()`'s `Value::List`
1820/// needs `Binding::List` (no list variant in `PropertyValue`, the
1821/// storage-layer type `Binding::Value` wraps), everything else collapses
1822/// to `Binding::Value` same as any other computed WITH item.
1823fn value_to_binding(v: Value) -> Binding {
1824 match v {
1825 Value::List(items) => Binding::List(items),
1826 other => Binding::Value(value_to_property_value(&other)),
1827 }
1828}
1829
1830/// `UNWIND`'s counterpart to `value_to_binding` — restores graph identity
1831/// from a `collect()`'d element instead of collapsing it. `Value::Node`/
1832/// `Edge` carry their full `id`, so this isn't lossy the way carrying only
1833/// a display value would be: a `MATCH` after the `UNWIND` can keep
1834/// traversing from the restored `Binding::Node`/`Edge`, exactly as if it
1835/// had been bound by a fresh scan/expand. See `Binding::List`'s docs,
1836/// which anticipated this exact restoration.
1837fn value_to_binding_restore(v: &Value) -> Binding {
1838 match v {
1839 Value::Node(n) => Binding::Node(n.id),
1840 Value::Edge(e) => Binding::Edge(e.id),
1841 Value::Property(pv) => Binding::Value(pv.clone()),
1842 Value::Literal(lit) => Binding::Value(literal_to_value(lit)),
1843 Value::List(items) => Binding::List(items.clone()),
1844 Value::Path(elems) => Binding::Path(elems.iter().map(path_elem_to_binding).collect()),
1845 Value::Null => Binding::Value(PropertyValue::Null),
1846 }
1847}
1848
1849fn path_elem_to_binding(elem: &PathElem) -> PathBinding {
1850 match elem {
1851 PathElem::Node(n) => PathBinding::Node(n.id),
1852 PathElem::Edge(e) => PathBinding::Edge(e.id),
1853 }
1854}
1855
1856/// When a path is being captured, every hop's rel/node needs a trackable
1857/// binding even if the user left it anonymous — `Expand` only inserts a
1858/// `rel_var` into the row `if let Some(rv) = rel_var`, silently dropping
1859/// anonymous rels, which is fine for ordinary matching but loses exactly
1860/// the information path assembly needs. Returns a clone of `pattern` with
1861/// every position named (synthesizing `__path_elemN` for anything
1862/// anonymous), plus the set of names that were synthesized so
1863/// `execute_match` can strip them from the row again after `assemble_path`
1864/// runs — they were never something the user could reference. Only this
1865/// renamed clone is used for plan-building/OPTIONAL-MATCH null-padding
1866/// bookkeeping *within this one clause*; `carried_vars` (what's exposed to
1867/// later clauses) is still computed from the original `part.pattern`
1868/// elsewhere, so synthesized names never leak past this function's caller.
1869fn name_pattern_for_path(pattern: &Pattern) -> (Pattern, HashSet<String>) {
1870 fn fresh(counter: &mut usize, synthesized: &mut HashSet<String>) -> String {
1871 *counter += 1;
1872 let name = format!("__path_elem{counter}");
1873 synthesized.insert(name.clone());
1874 name
1875 }
1876 let mut counter = 0usize;
1877 let mut synthesized = HashSet::new();
1878 let mut start = pattern.start.clone();
1879 if start.var.is_none() {
1880 start.var = Some(fresh(&mut counter, &mut synthesized));
1881 }
1882 let hops = pattern
1883 .hops
1884 .iter()
1885 .map(|(rel, node)| {
1886 let mut rel = rel.clone();
1887 if rel.var.is_none() {
1888 rel.var = Some(fresh(&mut counter, &mut synthesized));
1889 }
1890 let mut node = node.clone();
1891 if node.var.is_none() {
1892 node.var = Some(fresh(&mut counter, &mut synthesized));
1893 }
1894 (rel, node)
1895 })
1896 .collect();
1897 (Pattern { start, hops }, synthesized)
1898}
1899
1900/// Assembles a `Binding::Path` from `pattern`'s (fully-named, via
1901/// `name_pattern_for_path`) start/hop variables, in pattern order. Falls
1902/// back to `Binding::Value(Null)` — never errors — if any position isn't a
1903/// real node/edge binding, which only happens when this row came from
1904/// `OPTIONAL MATCH` null-padding (every position `name_pattern_for_path`
1905/// named is guaranteed present in the row either way, as a real binding or
1906/// as `Binding::Value(Null)`, so "missing key" isn't a case this needs to
1907/// handle) — same "no match survives as Null, not a dropped row" outcome
1908/// `OPTIONAL MATCH` already gives every other variable.
1909fn assemble_path(pattern: &Pattern, row: &BindingRow) -> Binding {
1910 let Some(start_id) = path_node_id(pattern.start.var.as_deref(), row) else {
1911 return Binding::Value(PropertyValue::Null);
1912 };
1913 let mut elems = vec![PathBinding::Node(start_id)];
1914 for (rel, node) in &pattern.hops {
1915 let Some(edge_id) = path_edge_id(rel.var.as_deref(), row) else {
1916 return Binding::Value(PropertyValue::Null);
1917 };
1918 let Some(node_id) = path_node_id(node.var.as_deref(), row) else {
1919 return Binding::Value(PropertyValue::Null);
1920 };
1921 elems.push(PathBinding::Edge(edge_id));
1922 elems.push(PathBinding::Node(node_id));
1923 }
1924 Binding::Path(elems)
1925}
1926
1927fn path_node_id(var: Option<&str>, row: &BindingRow) -> Option<NodeId> {
1928 match var.and_then(|v| row.get(v)) {
1929 Some(Binding::Node(id)) => Some(*id),
1930 _ => None,
1931 }
1932}
1933
1934fn path_edge_id(var: Option<&str>, row: &BindingRow) -> Option<EdgeId> {
1935 match var.and_then(|v| row.get(v)) {
1936 Some(Binding::Edge(id)) => Some(*id),
1937 _ => None,
1938 }
1939}
1940
1941fn require_bound_node(row: &BindingRow, var: &str) -> Result<NodeId, QueryError> {
1942 match row.get(var) {
1943 Some(Binding::Node(id)) => Ok(*id),
1944 _ => Err(QueryError::UnboundVariable(format!(
1945 "'{var}' must already be bound to a node before shortestPath() — match it in a preceding MATCH"
1946 ))),
1947 }
1948}
1949
1950/// Walks `parent` (populated by `shortest_path_between`'s BFS) backward
1951/// from `end` to `start`, then reverses — `parent` only ever needs to
1952/// answer "how did BFS first reach this node," not support any other
1953/// traversal, so a plain `HashMap` (not a `LogicalPlan`/adjacency
1954/// structure) is enough.
1955fn reconstruct_path(parent: &HashMap<NodeId, (NodeId, EdgeId)>, start: NodeId, end: NodeId) -> Vec<PathBinding> {
1956 let mut hops = Vec::new();
1957 let mut current = end;
1958 while current != start {
1959 let (prev, edge_id) = parent[¤t];
1960 hops.push((edge_id, current));
1961 current = prev;
1962 }
1963 hops.reverse();
1964 let mut elems = vec![PathBinding::Node(start)];
1965 for (edge_id, node) in hops {
1966 elems.push(PathBinding::Edge(edge_id));
1967 elems.push(PathBinding::Node(node));
1968 }
1969 elems
1970}
1971
1972/// `WithExpr::Compare`'s value-vs-literal comparison — reuses `compare()`
1973/// (below) by reducing a `Value` down to the `Option<PropertyValue>` shape
1974/// it expects; `Node`/`Edge`/`List` have no meaningful comparison against
1975/// a `Literal` and fall back to "absent", same as a missing property does.
1976fn compare_value(value: &Value, op: CompareOp, lit: &Literal) -> Option<bool> {
1977 let prop = match value {
1978 Value::Null => None,
1979 Value::Property(pv) => Some(pv.clone()),
1980 Value::Literal(l) => Some(literal_to_value(l)),
1981 Value::Node(_) | Value::Edge(_) | Value::List(_) | Value::Path(_) => None,
1982 };
1983 compare(&prop, op, lit)
1984}
1985
1986/// Coerces a materialized `Value` down to a `PropertyValue` for storing in
1987/// `Binding::Value` — used by `item_binding` for a computed (non-bare-var)
1988/// WITH/RETURN item. `Value::Node`/`Edge` can't occur here in practice (no
1989/// non-aggregate `ReturnExpr` form produces one except `Var`, which takes
1990/// the bare-variable path instead). `Value::List` can't occur here either
1991/// — `collect()` only ever appears in an aggregating item list, which
1992/// `has_aggregate` routes to `resolve_grouped_rows`/`Binding::List`
1993/// instead of through `item_binding` at all. Both fall back to `Null`
1994/// rather than needing a fallible signature for an unreachable case.
1995fn value_to_property_value(v: &Value) -> PropertyValue {
1996 match v {
1997 Value::Null => PropertyValue::Null,
1998 Value::Property(pv) => pv.clone(),
1999 Value::Literal(lit) => literal_to_value(lit),
2000 Value::Node(_) | Value::Edge(_) | Value::List(_) | Value::Path(_) => PropertyValue::Null,
2001 }
2002}
2003
2004fn literal_to_value(lit: &Literal) -> PropertyValue {
2005 match lit {
2006 Literal::Int(i) => PropertyValue::Int(*i),
2007 Literal::Float(f) => PropertyValue::Float(*f),
2008 Literal::String(s) => PropertyValue::String(s.clone()),
2009 Literal::Bool(b) => PropertyValue::Bool(*b),
2010 Literal::Null => PropertyValue::Null,
2011 Literal::Param(name) => {
2012 unreachable!("param ${name} must be substituted before execution — see params::substitute_params")
2013 }
2014 }
2015}
2016
2017fn literal_props_to_values(props: &[(String, Literal)]) -> BTreeMap<String, PropertyValue> {
2018 props.iter().map(|(k, v)| (k.clone(), literal_to_value(v))).collect()
2019}
2020
2021fn tag_merge_created(mut row: BindingRow, created: bool) -> BindingRow {
2022 row.insert(MERGE_CREATED_KEY.to_string(), Binding::Value(PropertyValue::Bool(created)));
2023 row
2024}
2025
2026/// Rejects a `MERGE` pattern token that's neither already bound in `row`
2027/// nor constrained by any label/property — matching or creating it would
2028/// mean guessing at "any node," which this codebase's "error on an
2029/// ambiguous shape" stance treats as a mistake to catch (not a silent
2030/// "match/create arbitrarily" default). Called before any graph work, not
2031/// just before the create-fallback branch — an unconstrained, unbound
2032/// token would otherwise let `eval_merge`'s search phase silently "match"
2033/// every node in the graph (`AllNodesScan`, no `Filter`) instead of
2034/// erroring.
2035fn require_mergeable(node: &NodePattern, row: &BindingRow) -> Result<(), QueryError> {
2036 let already_bound = node.var.as_ref().is_some_and(|v| row.contains_key(v));
2037 if !already_bound && node.labels.is_empty() && node.props.is_empty() {
2038 return Err(QueryError::Parse(
2039 "MERGE requires a label or property to match/create by — an unconstrained node pattern is ambiguous"
2040 .into(),
2041 ));
2042 }
2043 Ok(())
2044}
2045
2046/// `Either` (undirected `-[r:TYPE]-`) has no single storage-level call —
2047/// query both directions and dedupe by `edge_id` (a self-loop would
2048/// otherwise appear twice, once from each direction's adjacency table).
2049fn neighbors_for_direction(
2050 txn: Txn,
2051 node: NodeId,
2052 direction: ExpandDirection,
2053 rel_label: Option<&str>,
2054) -> Result<Vec<AdjEntry>, QueryError> {
2055 Ok(match direction {
2056 ExpandDirection::Out => GraphStore::neighbors_in_txn(txn, node, Direction::Out, rel_label)?,
2057 ExpandDirection::In => GraphStore::neighbors_in_txn(txn, node, Direction::In, rel_label)?,
2058 ExpandDirection::Either => {
2059 let mut out = GraphStore::neighbors_in_txn(txn, node, Direction::Out, rel_label)?;
2060 let inbound = GraphStore::neighbors_in_txn(txn, node, Direction::In, rel_label)?;
2061 let seen: HashSet<EdgeId> = out.iter().map(|e| e.edge_id).collect();
2062 out.extend(inbound.into_iter().filter(|e| !seen.contains(&e.edge_id)));
2063 out
2064 }
2065 })
2066}
2067
2068/// Three-valued: `None` is Cypher's "unknown", not `false` -- any
2069/// comparison touching a null (a missing property, or a literal `null` on
2070/// either side) is unknown, always, regardless of operator -- including
2071/// `Eq` (`x = null` is unknown, never true, same as real Cypher; it is
2072/// *not* how `x`'s own missing-ness is tested -- there's no `IS NULL`
2073/// operator yet). Callers combine this with `and3`/`or3`/`Option::map`
2074/// (for `NOT`) rather than unwrapping early, so unknown propagates
2075/// correctly through `AND`/`OR`/`NOT` instead of collapsing to `false`.
2076fn compare(prop: &Option<PropertyValue>, op: CompareOp, lit: &Literal) -> Option<bool> {
2077 let Some(prop) = prop else { return None };
2078 if matches!(prop, PropertyValue::Null) || matches!(lit, Literal::Null) {
2079 return None;
2080 }
2081 Some(match (prop, lit) {
2082 (PropertyValue::Int(a), Literal::Int(b)) => cmp_ord(op, *a, *b),
2083 (PropertyValue::Int(a), Literal::Float(b)) => cmp_f64(op, *a as f64, *b),
2084 (PropertyValue::Float(a), Literal::Float(b)) => cmp_f64(op, *a, *b),
2085 (PropertyValue::Float(a), Literal::Int(b)) => cmp_f64(op, *a, *b as f64),
2086 (PropertyValue::String(a), Literal::String(b)) => match op {
2087 CompareOp::StartsWith => a.starts_with(b.as_str()),
2088 CompareOp::EndsWith => a.ends_with(b.as_str()),
2089 CompareOp::Contains => a.contains(b.as_str()),
2090 _ => cmp_ord(op, a.as_str(), b.as_str()),
2091 },
2092 (PropertyValue::Bool(a), Literal::Bool(b)) => match op {
2093 CompareOp::Eq => a == b,
2094 CompareOp::Ne => a != b,
2095 _ => false,
2096 },
2097 _ => false,
2098 })
2099}
2100
2101/// `None`/`None` (both unknown) combines to unknown, matching Cypher's
2102/// `AND` truth table -- `false` wins over `unknown` (`false AND unknown =
2103/// false`), but `true AND unknown = unknown`, not `true`.
2104fn and3(a: Option<bool>, b: Option<bool>) -> Option<bool> {
2105 match (a, b) {
2106 (Some(false), _) | (_, Some(false)) => Some(false),
2107 (Some(true), Some(true)) => Some(true),
2108 _ => None,
2109 }
2110}
2111
2112/// Mirrors `and3` for `OR` -- `true` wins over `unknown`.
2113fn or3(a: Option<bool>, b: Option<bool>) -> Option<bool> {
2114 match (a, b) {
2115 (Some(true), _) | (_, Some(true)) => Some(true),
2116 (Some(false), Some(false)) => Some(false),
2117 _ => None,
2118 }
2119}
2120
2121fn cmp_f64(op: CompareOp, a: f64, b: f64) -> bool {
2122 match op {
2123 CompareOp::Eq => a == b,
2124 CompareOp::Ne => a != b,
2125 CompareOp::Lt => a < b,
2126 CompareOp::Le => a <= b,
2127 CompareOp::Gt => a > b,
2128 CompareOp::Ge => a >= b,
2129 // Only meaningful for String/String, handled separately in
2130 // `compare()` before reaching here -- a numeric operand with one
2131 // of these ops is a type mismatch, same as any other.
2132 CompareOp::StartsWith | CompareOp::EndsWith | CompareOp::Contains => false,
2133 }
2134}
2135
2136fn cmp_ord<T: PartialOrd>(op: CompareOp, a: T, b: T) -> bool {
2137 match op {
2138 CompareOp::Eq => a == b,
2139 CompareOp::Ne => a != b,
2140 CompareOp::Lt => a < b,
2141 CompareOp::Le => a <= b,
2142 CompareOp::Gt => a > b,
2143 CompareOp::Ge => a >= b,
2144 CompareOp::StartsWith | CompareOp::EndsWith | CompareOp::Contains => false,
2145 }
2146}
2147
2148/// Value equality for CASE's WHEN-comparison (and, elsewhere, DISTINCT
2149/// dedup within an aggregate). Null == Null -> true here deliberately,
2150/// unlike `compare()`'s three-valued `WHERE`-filter semantics -- CASE and
2151/// DISTINCT need a definite yes/no ("is this the same value as a value
2152/// already collected", "does this WHEN branch match") rather than
2153/// "unknown", so plain equality is the correct, separate choice here, not
2154/// an oversight. `Node`/`Edge` compare by id (graph identity), not
2155/// full-struct contents — cheaper, and the correct semantics regardless
2156/// (two bindings are "the same node" iff the same node, not iff their
2157/// label/prop snapshots happen to match).
2158pub(crate) fn value_eq(a: &Value, b: &Value) -> bool {
2159 match (a, b) {
2160 (Value::Null, Value::Null) => true,
2161 (Value::Null, _) | (_, Value::Null) => false,
2162 (Value::Property(pa), Value::Property(pb)) => pa == pb,
2163 (Value::Literal(la), Value::Literal(lb)) => la == lb,
2164 (Value::Property(pa), Value::Literal(lb)) => *pa == literal_to_value(lb),
2165 (Value::Literal(la), Value::Property(pb)) => literal_to_value(la) == *pb,
2166 (Value::Node(na), Value::Node(nb)) => na.id == nb.id,
2167 (Value::Edge(ea), Value::Edge(eb)) => ea.id == eb.id,
2168 (Value::List(la), Value::List(lb)) => la.len() == lb.len() && la.iter().zip(lb).all(|(x, y)| value_eq(x, y)),
2169 _ => false,
2170 }
2171}
2172
2173fn call_builtin(name: &str, args: &[Value]) -> Result<Value, QueryError> {
2174 match name.to_ascii_lowercase().as_str() {
2175 "coalesce" => Ok(args
2176 .iter()
2177 .find(|v| !matches!(v, Value::Null))
2178 .cloned()
2179 .unwrap_or(Value::Null)),
2180 "tointeger" => Ok(args.first().map(to_integer).unwrap_or(Value::Null)),
2181 // The dominant real-world use of shortestPath() is measuring it
2182 // (degrees-of-separation queries), not returning/rendering the
2183 // raw path object — path elements alternate node/edge/.../node,
2184 // so edge count is (elements.len() - 1) / 2.
2185 "length" => Ok(match args.first() {
2186 Some(Value::Path(elems)) => Value::Property(PropertyValue::Int(((elems.len().max(1) - 1) / 2) as i64)),
2187 Some(Value::Null) | None => Value::Null,
2188 Some(other) => {
2189 return Err(QueryError::Parse(format!("length() expects a path, got {other:?}")))
2190 }
2191 }),
2192 other => Err(QueryError::Parse(format!("unknown function: {other}"))),
2193 }
2194}
2195
2196fn to_integer(v: &Value) -> Value {
2197 let as_str_parse = |s: &str| match s.trim().parse::<i64>() {
2198 Ok(i) => Value::Property(PropertyValue::Int(i)),
2199 Err(_) => Value::Null,
2200 };
2201 match v {
2202 Value::Property(PropertyValue::Int(i)) => Value::Property(PropertyValue::Int(*i)),
2203 Value::Property(PropertyValue::Float(f)) => Value::Property(PropertyValue::Int(*f as i64)),
2204 Value::Property(PropertyValue::String(s)) => as_str_parse(s),
2205 Value::Literal(Literal::Int(i)) => Value::Property(PropertyValue::Int(*i)),
2206 Value::Literal(Literal::Float(f)) => Value::Property(PropertyValue::Int(*f as i64)),
2207 Value::Literal(Literal::String(s)) => as_str_parse(s),
2208 _ => Value::Null,
2209 }
2210}
2211
2212/// Sorts `rows` (already-projected `RETURN`/`WITH` output, `columns`
2213/// aligned by index) by `order_by`, which evaluates against the projected
2214/// column names — never the raw pattern `BindingRow` — since every ORDER BY
2215/// key in practice is a RETURN/WITH alias, not a bare pattern variable.
2216fn apply_order_by(
2217 rows: Vec<Vec<Value>>,
2218 columns: &[String],
2219 order_by: &[(ReturnExpr, SortDir)],
2220 limit: Option<i64>,
2221) -> Result<Vec<Vec<Value>>, QueryError> {
2222 // An ORDER BY expression that repeats a returned expression verbatim
2223 // (`RETURN n.name, count(*) AS foo ORDER BY n.name`) names a real
2224 // output column by its default name -- match it directly by position
2225 // rather than re-evaluating the expression, which would need bindings
2226 // (e.g. `n`) that only the pre-aggregation rows had and are gone by
2227 // this post-projection point.
2228 let order_by_col: Vec<Option<usize>> = order_by
2229 .iter()
2230 .map(|(expr, _)| columns.iter().position(|c| *c == default_column_name(expr, 0)))
2231 .collect();
2232 let mut keyed: Vec<(Vec<Value>, Vec<Value>)> = Vec::with_capacity(rows.len());
2233 for row in rows {
2234 let row_map: HashMap<String, Value> = columns.iter().cloned().zip(row.iter().cloned()).collect();
2235 let keys = order_by
2236 .iter()
2237 .zip(&order_by_col)
2238 .map(|((expr, _), col)| match col {
2239 Some(i) => Ok(row[*i].clone()),
2240 None => eval_projected_expr(expr, &row_map),
2241 })
2242 .collect::<Result<Vec<_>, _>>()?;
2243 keyed.push((keys, row));
2244 }
2245 Ok(top_k_by(keyed, order_by, limit).into_iter().map(|(_, row)| row).collect())
2246}
2247
2248/// Same expression shape as `eval_return_expr`, but resolves `Var`/`Prop`
2249/// against already-projected output columns instead of the graph-bound
2250/// `BindingRow` — no `WriteTransaction`/`GraphStore` access needed, since a
2251/// projected `Value::Node`/`Value::Edge` already carries its full record
2252/// (including props) from when it was first materialized.
2253fn eval_projected_expr(expr: &ReturnExpr, row: &HashMap<String, Value>) -> Result<Value, QueryError> {
2254 match expr {
2255 ReturnExpr::Var(name) => row
2256 .get(name)
2257 .cloned()
2258 .ok_or_else(|| QueryError::UnboundVariable(name.clone())),
2259 ReturnExpr::Prop(pa) => {
2260 let base = row
2261 .get(&pa.var)
2262 .ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
2263 let pv = match base {
2264 Value::Node(n) => n.props.get(&pa.prop).cloned(),
2265 Value::Edge(e) => e.props.get(&pa.prop).cloned(),
2266 _ => None,
2267 };
2268 Ok(match pv {
2269 Some(PropertyValue::Null) | None => Value::Null,
2270 Some(v) => Value::Property(v),
2271 })
2272 }
2273 ReturnExpr::Lit(lit) => Ok(match lit {
2274 Literal::Null => Value::Null,
2275 other => Value::Literal(other.clone()),
2276 }),
2277 ReturnExpr::Call { name, args, .. } => {
2278 // Same internal-consistency stance as `eval_return_expr`'s
2279 // `Call` arm: by the time ORDER BY runs, aggregation has
2280 // already resolved into ordinary named output columns
2281 // (referenced here via `Var`), so a raw aggregate `Call`
2282 // reaching this point means it wasn't top-level as
2283 // `validate_return_items` requires.
2284 if is_aggregate_name(name) {
2285 return Err(QueryError::Parse(format!(
2286 "aggregate function '{name}' can only be used as a return item's top-level expression"
2287 )));
2288 }
2289 let arg_values = args
2290 .iter()
2291 .map(|a| eval_projected_expr(a, row))
2292 .collect::<Result<Vec<_>, _>>()?;
2293 call_builtin(name, &arg_values)
2294 }
2295 ReturnExpr::CountStar => Err(QueryError::Parse(
2296 "count(*) can only be used as a return item's top-level expression".into(),
2297 )),
2298 ReturnExpr::Case { test, whens, else_ } => {
2299 let test_value = match test {
2300 Some(t) => Some(eval_projected_expr(t, row)?),
2301 None => None,
2302 };
2303 for (when, then) in whens {
2304 let when_value = eval_projected_expr(when, row)?;
2305 let matched = match &test_value {
2306 Some(tv) => value_eq(tv, &when_value),
2307 None => matches!(when_value, Value::Literal(Literal::Bool(true))),
2308 };
2309 if matched {
2310 return eval_projected_expr(then, row);
2311 }
2312 }
2313 match else_ {
2314 Some(e) => eval_projected_expr(e, row),
2315 None => Ok(Value::Null),
2316 }
2317 }
2318 }
2319}
2320
2321/// `RETURN DISTINCT`'s result-set-level dedup -- structural equality of
2322/// the whole row (same `HashKey` machinery `DISTINCT` inside an aggregate
2323/// call and `resolve_grouped_rows`' grouping already use, not `value_eq`'s
2324/// definite-equality-only comparison, since a `HashSet` needs `Hash` too).
2325/// Keeps the first occurrence of each distinct row, preserving order --
2326/// what every other DB's `DISTINCT` does, and what a human reading the
2327/// query would expect.
2328fn dedup_rows(rows: Vec<Vec<Value>>) -> Result<Vec<Vec<Value>>, QueryError> {
2329 let mut seen: HashSet<Vec<HashKey>> = HashSet::with_capacity(rows.len());
2330 let mut out = Vec::with_capacity(rows.len());
2331 for row in rows {
2332 let key = row.iter().map(value_hash_key).collect::<Result<Vec<_>, _>>()?;
2333 if seen.insert(key) {
2334 out.push(row);
2335 }
2336 }
2337 Ok(out)
2338}
2339
2340/// Sorts `keyed` (each entry paired with its precomputed per-column sort
2341/// keys) by `order_by`'s directions, keeping only the first `limit` items
2342/// when one is given and smaller than the row count. When it is, uses
2343/// `select_nth_unstable_by` to partition around the k-th smallest element
2344/// (O(n) average) and sorts only that k-sized prefix (O(k log k)), instead
2345/// of a full O(n log n) sort of every row just to immediately discard all
2346/// but the first few -- the "ORDER BY + LIMIT -> TOP-K" rewrite real query
2347/// engines apply. Shared by all three ORDER BY sites (`WITH`'s own,
2348/// non-aggregating `RETURN`'s, and aggregating `RETURN`'s), which otherwise
2349/// each build the identical `keyed`-then-sort shape around a different row
2350/// type.
2351fn top_k_by<T>(
2352 mut keyed: Vec<(Vec<Value>, T)>,
2353 order_by: &[(ReturnExpr, SortDir)],
2354 limit: Option<i64>,
2355) -> Vec<(Vec<Value>, T)> {
2356 let cmp = |a: &(Vec<Value>, T), b: &(Vec<Value>, T)| -> std::cmp::Ordering {
2357 for (i, (_, dir)) in order_by.iter().enumerate() {
2358 let ord = compare_with_dir(&a.0[i], &b.0[i], *dir);
2359 if ord != std::cmp::Ordering::Equal {
2360 return ord;
2361 }
2362 }
2363 std::cmp::Ordering::Equal
2364 };
2365 match limit {
2366 Some(n) => {
2367 let k = n.max(0) as usize;
2368 if k == 0 {
2369 keyed.clear();
2370 } else if k < keyed.len() {
2371 keyed.select_nth_unstable_by(k - 1, cmp);
2372 keyed.truncate(k);
2373 keyed.sort_by(cmp);
2374 } else {
2375 keyed.sort_by(cmp);
2376 }
2377 }
2378 None => keyed.sort_by(cmp),
2379 }
2380 keyed
2381}
2382
2383/// NULLs sort last regardless of ASC/DESC (matches Neo4j's documented
2384/// behavior) — only non-null comparisons get reversed for DESC.
2385fn compare_with_dir(a: &Value, b: &Value, dir: SortDir) -> std::cmp::Ordering {
2386 use std::cmp::Ordering;
2387 let a_null = matches!(a, Value::Null);
2388 let b_null = matches!(b, Value::Null);
2389 match (a_null, b_null) {
2390 (true, true) => return Ordering::Equal,
2391 (true, false) => return Ordering::Greater,
2392 (false, true) => return Ordering::Less,
2393 (false, false) => {}
2394 }
2395 let ord = compare_non_null(a, b);
2396 if dir == SortDir::Desc {
2397 ord.reverse()
2398 } else {
2399 ord
2400 }
2401}
2402
2403fn compare_non_null(a: &Value, b: &Value) -> std::cmp::Ordering {
2404 use std::cmp::Ordering;
2405 let pa = value_to_comparable(a);
2406 let pb = value_to_comparable(b);
2407 match (pa, pb) {
2408 (Some(PropertyValue::Int(x)), Some(PropertyValue::Int(y))) => x.cmp(&y),
2409 (Some(PropertyValue::Int(x)), Some(PropertyValue::Float(y))) => {
2410 (x as f64).partial_cmp(&y).unwrap_or(Ordering::Equal)
2411 }
2412 (Some(PropertyValue::Float(x)), Some(PropertyValue::Int(y))) => {
2413 x.partial_cmp(&(y as f64)).unwrap_or(Ordering::Equal)
2414 }
2415 (Some(PropertyValue::Float(x)), Some(PropertyValue::Float(y))) => x.partial_cmp(&y).unwrap_or(Ordering::Equal),
2416 (Some(PropertyValue::String(x)), Some(PropertyValue::String(y))) => x.cmp(&y),
2417 (Some(PropertyValue::Bool(x)), Some(PropertyValue::Bool(y))) => x.cmp(&y),
2418 _ => Ordering::Equal,
2419 }
2420}
2421
2422fn value_to_comparable(v: &Value) -> Option<PropertyValue> {
2423 match v {
2424 Value::Property(pv) => Some(pv.clone()),
2425 Value::Literal(lit) => Some(literal_to_value(lit)),
2426 _ => None,
2427 }
2428}
2429
2430/// Ordering for `min`/`max` aggregate folding — `None` for values with no
2431/// natural order (`Node`/`Edge`/`List`, or a `Null`, which `AggAcc::fold`
2432/// never passes here anyway since null contributions are skipped before
2433/// folding). The caller turns `None` into a clear error rather than an
2434/// arbitrary "always equal" fallback — unlike ORDER BY's
2435/// `compare_non_null`, which tolerates that for presentation ordering
2436/// (see its docs), silently treating two nodes as "equal" inside an
2437/// aggregate would be a wrong-answer failure mode, not just an
2438/// unhelpful sort order.
2439pub(crate) fn comparable_ordering(a: &Value, b: &Value) -> Option<std::cmp::Ordering> {
2440 use std::cmp::Ordering;
2441 let pa = value_to_comparable(a)?;
2442 let pb = value_to_comparable(b)?;
2443 Some(match (pa, pb) {
2444 (PropertyValue::Int(x), PropertyValue::Int(y)) => x.cmp(&y),
2445 (PropertyValue::Int(x), PropertyValue::Float(y)) => (x as f64).partial_cmp(&y).unwrap_or(Ordering::Equal),
2446 (PropertyValue::Float(x), PropertyValue::Int(y)) => x.partial_cmp(&(y as f64)).unwrap_or(Ordering::Equal),
2447 (PropertyValue::Float(x), PropertyValue::Float(y)) => x.partial_cmp(&y).unwrap_or(Ordering::Equal),
2448 (PropertyValue::String(x), PropertyValue::String(y)) => x.cmp(&y),
2449 (PropertyValue::Bool(x), PropertyValue::Bool(y)) => x.cmp(&y),
2450 _ => return None,
2451 })
2452}