Skip to main content

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, Pattern, PropAccess, QueryPart, RelDirection, ReturnExpr,
8    ReturnItem, SortDir, Statement, Tail, WithClause, WithExpr,
9};
10use crate::error::QueryError;
11use crate::ir::{ExpandDirection, LogicalPlan};
12use crate::planner::{build_match_plan, pattern_all_vars, pattern_new_vars};
13use crate::result::QueryResult;
14use crate::value::Value;
15
16/// Hidden key used to correlate `OPTIONAL MATCH` results back to the outer
17/// row that seeded them — never visible to user Cypher (not a valid
18/// identifier prefix a parsed pattern could ever produce).
19const OPTIONAL_SEED_IDX_KEY: &str = "__seed_idx";
20
21#[derive(Debug, Clone)]
22enum Binding {
23    Node(NodeId),
24    Edge(EdgeId),
25    /// A scalar carried through a `WITH` projection (e.g. `WITH message.id
26    /// AS messageId`) — no graph identity, just a value along for the ride
27    /// to the next `QueryPart`/the final `Tail`.
28    Value(PropertyValue),
29    /// A `collect()` result carried through a `WITH` projection. Separate
30    /// from `Binding::Value` because `PropertyValue` (storage-layer) has no
31    /// list variant — lists are a query-layer-only concept, never
32    /// persisted — so a materialized `collect()` has nowhere else to live
33    /// between one `QueryPart` and the next. Elements are already-resolved
34    /// `Value`s, not `Binding`s: there's no `UNWIND` yet to pull one back
35    /// out with restored graph identity.
36    List(Vec<Value>),
37}
38
39type BindingRow = HashMap<String, Binding>;
40
41/// Safety cap on unbounded variable-length traversal (`[:TYPE*0..]`) depth.
42/// Hitting it errors rather than silently truncating — see `VarExpand`
43/// evaluation. Node-visited-set BFS (not relationship-uniqueness) is used
44/// throughout, which is only correct because the graphs this targets
45/// (LDBC's REPLY_OF-style reply chains) form a forest, not a general
46/// cyclic graph — not safe to reuse as-is for a variable-length pattern
47/// over a cyclic relationship type without revisiting that assumption.
48const VAR_EXPAND_DEPTH_CAP: u32 = 30;
49
50pub struct Executor<'a> {
51    store: &'a GraphStore,
52}
53
54impl<'a> Executor<'a> {
55    pub fn new(store: &'a GraphStore) -> Self {
56        Self { store }
57    }
58
59    /// Dispatches on whether `stmt` ever mutates anything. A read-only
60    /// statement (`MATCH ... RETURN`, `is_read_only` below) runs inside a
61    /// `ReadTransaction` — a consistent snapshot that doesn't contend for
62    /// redb's single-writer lock, so concurrent readers run in parallel
63    /// instead of queueing behind each other. Everything else runs inside
64    /// a `WriteTransaction`, committed or aborted as a whole — the
65    /// crash-safety boundary from the plan (one statement = one commit).
66    /// Every graph access below this point must go through the `*_in_txn`
67    /// GraphStore methods, never the standalone `self.store.*` methods,
68    /// which open (and would deadlock trying to re-open) their own
69    /// transaction.
70    pub fn execute(&self, stmt: &Statement) -> Result<QueryResult, QueryError> {
71        if is_read_only(stmt) {
72            let read_txn = self.store.begin_read()?;
73            let Statement::Match {
74                parts,
75                tail,
76                order_by,
77                limit,
78            } = stmt
79            else {
80                unreachable!("is_read_only only returns true for Statement::Match")
81            };
82            // No explicit commit/abort — a ReadTransaction is a pure
83            // snapshot view with nothing to roll back; it releases on drop.
84            return self.execute_match(Txn::Read(&read_txn), parts, tail, order_by, *limit);
85        }
86        let write_txn = self.store.begin_write()?;
87        let outcome = match stmt {
88            Statement::Create(patterns) => self.execute_create(&write_txn, patterns),
89            Statement::Match {
90                parts,
91                tail,
92                order_by,
93                limit,
94            } => self.execute_match(Txn::Write(&write_txn), parts, tail, order_by, *limit),
95        };
96        match outcome {
97            Ok(result) => {
98                GraphStore::commit(write_txn)?;
99                Ok(result)
100            }
101            Err(e) => {
102                // Best-effort rollback; the original error is what matters.
103                let _ = GraphStore::abort(write_txn);
104                Err(e)
105            }
106        }
107    }
108
109    fn execute_create(&self, write_txn: &WriteTransaction, patterns: &[Pattern]) -> Result<QueryResult, QueryError> {
110        for pattern in patterns {
111            let start_labels = pattern_labels(&pattern.start.labels);
112            let start_props = literal_props_to_values(&pattern.start.props);
113            let mut prev_id = GraphStore::create_node_in_txn(write_txn, &start_labels, start_props)?;
114
115            for (rel, node) in &pattern.hops {
116                if rel.hop_range.is_some() {
117                    return Err(QueryError::Parse(
118                        "CREATE doesn't support variable-length relationship patterns (e.g. [:TYPE*1..3])".into(),
119                    ));
120                }
121                let labels = pattern_labels(&node.labels);
122                let props = literal_props_to_values(&node.props);
123                let node_id = GraphStore::create_node_in_txn(write_txn, &labels, props)?;
124
125                let rel_label = rel.rel_type.clone().unwrap_or_else(|| "REL".to_string());
126                let rel_props = literal_props_to_values(&rel.props);
127                let (src, dst) = match rel.direction {
128                    RelDirection::Right => (prev_id, node_id),
129                    RelDirection::Left => (node_id, prev_id),
130                    RelDirection::Either => {
131                        return Err(QueryError::Parse(
132                            "CREATE requires a directed relationship (-> or <-), not an undirected pattern".into(),
133                        ))
134                    }
135                };
136                GraphStore::create_edge_in_txn(write_txn, &rel_label, src, dst, rel_props)?;
137                prev_id = node_id;
138            }
139        }
140        Ok(QueryResult {
141            columns: vec![],
142            rows: vec![],
143        })
144    }
145
146    fn execute_match(
147        &self,
148        txn: Txn,
149        parts: &[QueryPart],
150        tail: &Tail,
151        order_by: &Option<Vec<(ReturnExpr, SortDir)>>,
152        limit: Option<i64>,
153    ) -> Result<QueryResult, QueryError> {
154        // Threads bindings through each MATCH/WITH segment. `carried_vars`
155        // tells the planner which of the next part's pattern variables are
156        // already bound (-> LogicalPlan::Seed) rather than fresh
157        // (-> a scan). Starts empty: the first part never has anything
158        // carried into it.
159        let mut carried_vars: HashSet<String> = HashSet::new();
160        let mut current_rows: Vec<BindingRow> = vec![BindingRow::new()];
161        for part in parts {
162            let plan = build_match_plan(&part.pattern, &part.where_clause, &carried_vars)?;
163            current_rows = if part.optional {
164                let new_vars = pattern_new_vars(&part.pattern, &carried_vars);
165                self.eval_optional_part(txn, &plan, &current_rows, &new_vars)?
166            } else {
167                self.eval_plan(txn, &plan, &current_rows)?
168            };
169            if let Some(with) = &part.with {
170                current_rows = self.materialize_with(txn, with, &current_rows)?;
171                if let Some(with_order_by) = &with.order_by {
172                    current_rows = self.apply_order_by_bindings(txn, current_rows, with_order_by)?;
173                }
174                if let Some(with_limit) = with.limit {
175                    current_rows.truncate(with_limit.max(0) as usize);
176                }
177                carried_vars = with.items.iter().enumerate().map(with_item_output_name).collect();
178            } else {
179                // No WITH: real Cypher shares one binding scope across
180                // MATCH/OPTIONAL MATCH clauses that aren't WITH-separated
181                // — every var this part bound stays in scope for whatever
182                // comes next, on top of what was already carried in.
183                carried_vars.extend(pattern_all_vars(&part.pattern));
184            }
185        }
186        // ORDER BY must see every matching row before LIMIT truncates —
187        // sort, then take N, not the other way around. Only pre-truncate
188        // (the v1 "doesn't short-circuit" path) when there's no ORDER BY to
189        // invalidate it; DELETE/SET+LIMIT keep their "stop after N
190        // bindings" behavior since they have no ORDER BY position in the
191        // grammar.
192        if order_by.is_none() {
193            if let Some(count) = limit {
194                current_rows.truncate(count.max(0) as usize);
195            }
196        }
197        // Delete/Set need real `.insert`/`.remove`-capable write access,
198        // not just `Txn`'s read-only `get`/`iter` — but they're only ever
199        // reached via `Executor::execute`'s write-dispatch path (see
200        // `is_read_only`), which always opens a `WriteTransaction`, so
201        // `txn` is guaranteed to be `Txn::Write` here.
202        let mut result = match tail {
203            Tail::Return(items) => self.materialize_return(txn, items, &current_rows)?,
204            Tail::Delete(vars) => {
205                self.materialize_delete(require_write_txn(txn), vars, &current_rows, false)?
206            }
207            Tail::DetachDelete(vars) => {
208                self.materialize_delete(require_write_txn(txn), vars, &current_rows, true)?
209            }
210            Tail::Set(items) => self.materialize_set(require_write_txn(txn), items, &current_rows)?,
211        };
212        if let Some(order_by) = order_by {
213            result.rows = apply_order_by(result.rows, &result.columns, order_by)?;
214            if let Some(count) = limit {
215                result.rows.truncate(count.max(0) as usize);
216            }
217        }
218        Ok(result)
219    }
220
221    /// Projects `rows` through a `WITH` clause. Unlike `materialize_return`
222    /// (which resolves everything down to display `Value`s), a bare
223    /// variable reference (`WITH message`) must keep its graph identity
224    /// (`Binding::Node`/`Edge`) so the next `QueryPart` can keep
225    /// traversing from it — only computed expressions collapse to a
226    /// scalar `Binding::Value`.
227    fn materialize_with(
228        &self,
229        txn: Txn,
230        with: &WithClause,
231        rows: &[BindingRow],
232    ) -> Result<Vec<BindingRow>, QueryError> {
233        let mut out = if !has_aggregate(&with.items) {
234            let mut out = Vec::with_capacity(rows.len());
235            for row in rows {
236                let mut new_row = BindingRow::new();
237                for (i, item) in with.items.iter().enumerate() {
238                    let name = with_item_output_name((i, item));
239                    let binding = self.item_binding(txn, &item.expr, row)?;
240                    new_row.insert(name, binding);
241                }
242                out.push(new_row);
243            }
244            out
245        } else {
246            validate_return_items(&with.items)?;
247            let grouped = self.resolve_grouped_rows(txn, &with.items, rows)?;
248            grouped
249                .into_iter()
250                .map(|bindings| {
251                    with.items
252                        .iter()
253                        .enumerate()
254                        .zip(bindings)
255                        .map(|((i, item), b)| (with_item_output_name((i, item)), b))
256                        .collect()
257                })
258                .collect()
259        };
260        if let Some(where_clause) = &with.where_clause {
261            let mut filtered = Vec::with_capacity(out.len());
262            for row in out {
263                if self.eval_with_expr(txn, where_clause, &row)? {
264                    filtered.push(row);
265                }
266            }
267            out = filtered;
268        }
269        Ok(out)
270    }
271
272    /// The `Binding` one WITH/RETURN item evaluates to for one input row. A
273    /// bare `Var` keeps its graph identity (`Binding::Node`/`Edge`) so a
274    /// later `QueryPart` can keep traversing from it; anything else
275    /// (computed expressions) collapses to `Binding::Value`. Shared by the
276    /// non-aggregating `materialize_with` path and grouping-key evaluation.
277    fn item_binding(&self, txn: Txn, expr: &ReturnExpr, row: &BindingRow) -> Result<Binding, QueryError> {
278        match expr {
279            ReturnExpr::Var(v) => row.get(v).cloned().ok_or_else(|| QueryError::UnboundVariable(v.clone())),
280            other => {
281                let value = self.eval_return_expr(txn, other, row)?;
282                Ok(Binding::Value(value_to_property_value(&value)))
283            }
284        }
285    }
286
287    /// Same sort as `apply_order_by`, but over `BindingRow`s (a `WITH`
288    /// clause's own ORDER BY, which must run before that row set becomes
289    /// the seed for the next `QueryPart` — sorting/limiting a WITH changes
290    /// *which* rows continue, not just their presentation order).
291    fn apply_order_by_bindings(
292        &self,
293        txn: Txn,
294        rows: Vec<BindingRow>,
295        order_by: &[(ReturnExpr, SortDir)],
296    ) -> Result<Vec<BindingRow>, QueryError> {
297        let mut keyed: Vec<(Vec<Value>, BindingRow)> = Vec::with_capacity(rows.len());
298        for row in rows {
299            let value_map = self.binding_row_to_value_map(txn, &row)?;
300            let keys = order_by
301                .iter()
302                .map(|(expr, _)| eval_projected_expr(expr, &value_map))
303                .collect::<Result<Vec<_>, _>>()?;
304            keyed.push((keys, row));
305        }
306        keyed.sort_by(|(ka, _), (kb, _)| {
307            for (i, (_, dir)) in order_by.iter().enumerate() {
308                let ord = compare_with_dir(&ka[i], &kb[i], *dir);
309                if ord != std::cmp::Ordering::Equal {
310                    return ord;
311                }
312            }
313            std::cmp::Ordering::Equal
314        });
315        Ok(keyed.into_iter().map(|(_, row)| row).collect())
316    }
317
318    fn binding_row_to_value_map(
319        &self,
320        txn: Txn,
321        row: &BindingRow,
322    ) -> Result<HashMap<String, Value>, QueryError> {
323        let mut map = HashMap::with_capacity(row.len());
324        for (k, binding) in row {
325            map.insert(k.clone(), self.binding_to_value(txn, binding)?);
326        }
327        Ok(map)
328    }
329
330    /// Resolves a `Binding` to its display `Value` — a `Node`/`Edge`
331    /// binding fetches the full current record, a scalar `Value` binding
332    /// passes through (collapsing a stored `PropertyValue::Null` to
333    /// `Value::Null`, same as everywhere else null is represented).
334    fn binding_to_value(&self, txn: Txn, b: &Binding) -> Result<Value, QueryError> {
335        Ok(match b {
336            Binding::Node(id) => Value::Node(
337                GraphStore::get_node_in_txn(txn, *id)?
338                    .expect("bound node exists within this statement's transaction"),
339            ),
340            Binding::Edge(id) => Value::Edge(
341                GraphStore::get_edge_in_txn(txn, *id)?
342                    .expect("bound edge exists within this statement's transaction"),
343            ),
344            Binding::Value(PropertyValue::Null) => Value::Null,
345            Binding::Value(pv) => Value::Property(pv.clone()),
346            Binding::List(items) => Value::List(items.clone()),
347        })
348    }
349
350    /// Folds `rows` into groups keyed by every non-aggregate item's per-row
351    /// `Binding` (via `item_binding`), then finishes each aggregate item's
352    /// accumulator per group. Returns one `Vec<Binding>` per output group,
353    /// column-aligned with `items`. Shared by `materialize_with` and
354    /// `materialize_return` — both already take the same `rows: &[BindingRow]`
355    /// input type, so the grouping core stays in `Binding`-space (preserving
356    /// graph identity for bare-var grouping keys) and each caller does its
357    /// own thin final conversion.
358    ///
359    /// Grouping-key lookup is a hash-map lookup (`group_index`, keyed by
360    /// `binding_hash_key`'s output — `Binding`/`PropertyValue` don't
361    /// derive `Eq`/`Hash` themselves, `PropertyValue::Float` can't, so
362    /// `HashKey` stands in for them; see its docs) into `groups`, which
363    /// stays a plain `Vec` for insertion-order-stable output when there's
364    /// no ORDER BY. O(1) average per row, not the O(rows × groups) linear
365    /// scan this used to be — see BENCHMARKS.md for the measured
366    /// before/after.
367    ///
368    /// Callers must call `validate_return_items` first — this function
369    /// assumes every aggregate `Call` item has already been checked to
370    /// have exactly one argument.
371    fn resolve_grouped_rows(
372        &self,
373        txn: Txn,
374        items: &[ReturnItem],
375        rows: &[BindingRow],
376    ) -> Result<Vec<Vec<Binding>>, QueryError> {
377        struct Group {
378            // Aligned to `items`: `Some` at a non-aggregate item's index,
379            // `None` at an aggregate item's index (both vecs below are
380            // index-aligned to `items` the same way, so exactly one of
381            // `key_bindings[i]`/`accs[i]` is populated per `i`).
382            key_bindings: Vec<Option<Binding>>,
383            accs: Vec<Option<AggAcc>>,
384            row_count: i64,
385        }
386        fn fresh_accs(items: &[ReturnItem]) -> Vec<Option<AggAcc>> {
387            items
388                .iter()
389                .map(|item| match &item.expr {
390                    ReturnExpr::Call { name, distinct, .. } if is_aggregate_name(name) => {
391                        Some(AggAcc::identity(name, *distinct))
392                    }
393                    _ => None,
394                })
395                .collect()
396        }
397
398        // Groups live in `groups` (insertion order, for stable output when
399        // there's no ORDER BY) with `group_index` as a hash-based lookup
400        // into it, keyed by a hashable stand-in for `key_bindings` (see
401        // `HashKey` — `Binding`/`PropertyValue` don't derive `Eq`/`Hash`
402        // themselves, `PropertyValue::Float` can't). O(1) average lookup
403        // per row instead of the O(groups) linear scan this replaced —
404        // see BENCHMARKS.md for the measured before/after.
405        let mut groups: Vec<Group> = Vec::new();
406        let mut group_index: HashMap<Vec<Option<HashKey>>, usize> = HashMap::new();
407        for row in rows {
408            let mut key_bindings = Vec::with_capacity(items.len());
409            for item in items {
410                key_bindings.push(if is_top_level_aggregate(&item.expr) {
411                    None
412                } else {
413                    Some(self.item_binding(txn, &item.expr, row)?)
414                });
415            }
416            let hash_key: Vec<Option<HashKey>> =
417                key_bindings.iter().map(|b| b.as_ref().map(binding_hash_key)).collect();
418            let group_idx = *group_index.entry(hash_key).or_insert_with(|| {
419                groups.push(Group {
420                    key_bindings: key_bindings.clone(),
421                    accs: fresh_accs(items),
422                    row_count: 0,
423                });
424                groups.len() - 1
425            });
426            let group = &mut groups[group_idx];
427            group.row_count += 1;
428            for (i, item) in items.iter().enumerate() {
429                let ReturnExpr::Call { args, .. } = &item.expr else { continue };
430                if !is_top_level_aggregate(&item.expr) {
431                    continue;
432                }
433                // Standard Cypher null-skipping: a null argument (e.g. an
434                // unmatched OPTIONAL MATCH variable) contributes to
435                // neither the accumulator nor its DISTINCT dedup set —
436                // this is what makes `count(x)` exclude a null-padded row
437                // while `count(*)` (tracked via `row_count`, not an
438                // accumulator at all) includes it.
439                let value = self.eval_return_expr(txn, &args[0], row)?;
440                if !matches!(value, Value::Null) {
441                    if let Some(acc) = &mut group.accs[i] {
442                        acc.fold(&value)?;
443                    }
444                }
445            }
446        }
447
448        // Global aggregate over an empty result set (no grouping-key items
449        // at all, and no rows to seed a group from) still produces exactly
450        // one output row — `count`/`count(*)` -> 0, `sum` -> 0,
451        // `avg`/`min`/`max` -> Null, `collect` -> [] — via the same
452        // fresh-accumulator `finish()` path a normal empty-contribution
453        // group already uses below, not a separate code path.
454        let no_key_items = items.iter().all(|item| is_top_level_aggregate(&item.expr));
455        if groups.is_empty() && no_key_items {
456            groups.push(Group {
457                key_bindings: vec![None; items.len()],
458                accs: fresh_accs(items),
459                row_count: 0,
460            });
461        }
462
463        let mut out = Vec::with_capacity(groups.len());
464        for mut group in groups {
465            let mut row_out = Vec::with_capacity(items.len());
466            for (i, item) in items.iter().enumerate() {
467                let binding = if matches!(item.expr, ReturnExpr::CountStar) {
468                    Binding::Value(PropertyValue::Int(group.row_count))
469                } else if is_top_level_aggregate(&item.expr) {
470                    let value = group.accs[i]
471                        .take()
472                        .expect("aggregate item must have an accumulator")
473                        .finish();
474                    value_to_binding(value)
475                } else {
476                    group.key_bindings[i].clone().expect("non-aggregate item must have a key binding")
477                };
478                row_out.push(binding);
479            }
480            out.push(row_out);
481        }
482        Ok(out)
483    }
484
485    /// WITH's HAVING-equivalent — evaluated against the already-projected/
486    /// grouped row, same as ORDER BY. Never pushed into the planner (see
487    /// `WithExpr`'s docs).
488    fn eval_with_expr(&self, txn: Txn, expr: &WithExpr, row: &BindingRow) -> Result<bool, QueryError> {
489        Ok(match expr {
490            WithExpr::And(l, r) => self.eval_with_expr(txn, l, row)? && self.eval_with_expr(txn, r, row)?,
491            WithExpr::Or(l, r) => self.eval_with_expr(txn, l, row)? || self.eval_with_expr(txn, r, row)?,
492            WithExpr::Not(e) => !self.eval_with_expr(txn, e, row)?,
493            WithExpr::Compare(lhs, op, lit) => {
494                let value = self.eval_return_expr(txn, lhs, row)?;
495                compare_value(&value, *op, lit)
496            }
497        })
498    }
499
500    /// Evaluates an `OPTIONAL MATCH` part with left-outer-join semantics:
501    /// every outer row survives, whether or not the optional pattern
502    /// matched anything for it. Must wrap the *whole* subplan rather than
503    /// null-padding inside `Expand`/`VarExpand` themselves — baking it in
504    /// there would turn every default (non-optional) `Expand` into a
505    /// left-outer-join too (breaking existing inner-join semantics), and
506    /// would mis-handle multi-hop optional patterns: IS7's optional
507    /// pattern is 2 hops, and per-hop null-padding would emit one
508    /// null-padded row per *hop-1* match even when hop 2 also matched,
509    /// instead of collapsing to exactly one row per outer row that had
510    /// zero end-to-end matches.
511    ///
512    /// Implementation: tag each outer row with its index, evaluate the
513    /// subplan once over the whole tagged batch (a single seed, not one
514    /// call per row), group results back by that index, then for any
515    /// outer index with zero results, emit the outer row unchanged plus
516    /// `Null` for every variable the optional pattern would have newly
517    /// introduced.
518    fn eval_optional_part(
519        &self,
520        txn: Txn,
521        plan: &LogicalPlan,
522        outer_rows: &[BindingRow],
523        new_vars: &HashSet<String>,
524    ) -> Result<Vec<BindingRow>, QueryError> {
525        let tagged: Vec<BindingRow> = outer_rows
526            .iter()
527            .enumerate()
528            .map(|(i, row)| {
529                let mut r = row.clone();
530                r.insert(OPTIONAL_SEED_IDX_KEY.to_string(), Binding::Value(PropertyValue::Int(i as i64)));
531                r
532            })
533            .collect();
534        let results = self.eval_plan(txn, plan, &tagged)?;
535        let mut by_idx: HashMap<i64, Vec<BindingRow>> = HashMap::new();
536        for mut row in results {
537            let idx = match row.remove(OPTIONAL_SEED_IDX_KEY) {
538                Some(Binding::Value(PropertyValue::Int(i))) => i,
539                other => unreachable!("__seed_idx tagged internally as Binding::Value(Int), got {other:?}"),
540            };
541            by_idx.entry(idx).or_default().push(row);
542        }
543        let mut out = Vec::with_capacity(outer_rows.len());
544        for (i, outer_row) in outer_rows.iter().enumerate() {
545            match by_idx.remove(&(i as i64)) {
546                Some(matches) => out.extend(matches),
547                None => {
548                    let mut padded = outer_row.clone();
549                    for var in new_vars {
550                        padded.insert(var.clone(), Binding::Value(PropertyValue::Null));
551                    }
552                    out.push(padded);
553                }
554            }
555        }
556        Ok(out)
557    }
558
559    fn eval_plan(
560        &self,
561        txn: Txn,
562        plan: &LogicalPlan,
563        seed: &[BindingRow],
564    ) -> Result<Vec<BindingRow>, QueryError> {
565        match plan {
566            LogicalPlan::Seed { var } => {
567                debug_assert!(
568                    seed.first().is_none_or(|row| row.contains_key(var)),
569                    "Seed{{var: {var:?}}} planned for a var not present in the carried-forward rows"
570                );
571                Ok(seed.to_vec())
572            }
573            LogicalPlan::AllNodesScan { var } => self.scan(txn, var, None),
574            LogicalPlan::NodeByLabelScan { var, label } => self.scan(txn, var, Some(label)),
575            LogicalPlan::Expand {
576                input,
577                from_var,
578                to_var,
579                rel_var,
580                rel_label,
581                direction,
582            } => {
583                let base_rows = self.eval_plan(txn, input, seed)?;
584                let mut out = Vec::new();
585                for row in base_rows {
586                    let Some(Binding::Node(from_id)) = row.get(from_var).cloned() else {
587                        return Err(QueryError::UnboundVariable(from_var.clone()));
588                    };
589                    let entries = neighbors_for_direction(txn, from_id, *direction, rel_label.as_deref())?;
590                    for entry in entries {
591                        let mut new_row = row.clone();
592                        new_row.insert(to_var.clone(), Binding::Node(entry.other));
593                        if let Some(rv) = rel_var {
594                            new_row.insert(rv.clone(), Binding::Edge(entry.edge_id));
595                        }
596                        out.push(new_row);
597                    }
598                }
599                Ok(out)
600            }
601            LogicalPlan::VarExpand {
602                input,
603                from_var,
604                to_var,
605                rel_label,
606                direction,
607                min_hops,
608                max_hops,
609            } => {
610                let base_rows = self.eval_plan(txn, input, seed)?;
611                let mut out = Vec::new();
612                let unbounded = max_hops.is_none();
613                let effective_max = max_hops.unwrap_or(VAR_EXPAND_DEPTH_CAP);
614                for row in base_rows {
615                    let Some(Binding::Node(start_id)) = row.get(from_var).cloned() else {
616                        return Err(QueryError::UnboundVariable(from_var.clone()));
617                    };
618                    let mut visited = HashSet::new();
619                    visited.insert(start_id);
620                    if *min_hops == 0 {
621                        let mut new_row = row.clone();
622                        new_row.insert(to_var.clone(), Binding::Node(start_id));
623                        out.push(new_row);
624                    }
625                    let mut frontier = vec![start_id];
626                    let mut depth = 0u32;
627                    while depth < effective_max && !frontier.is_empty() {
628                        depth += 1;
629                        let mut next_frontier = Vec::new();
630                        for node in frontier {
631                            let entries = neighbors_for_direction(txn, node, *direction, rel_label.as_deref())?;
632                            for entry in entries {
633                                if visited.insert(entry.other) {
634                                    next_frontier.push(entry.other);
635                                    if depth >= *min_hops {
636                                        let mut new_row = row.clone();
637                                        new_row.insert(to_var.clone(), Binding::Node(entry.other));
638                                        out.push(new_row);
639                                    }
640                                }
641                            }
642                        }
643                        frontier = next_frontier;
644                        if depth == effective_max && unbounded && !frontier.is_empty() {
645                            // Unbounded (`*N..`) traversal hit the safety
646                            // cap with more still reachable — error rather
647                            // than silently truncate results, which would
648                            // be a wrong-answer failure mode for a
649                            // correctness-benchmark tool.
650                            return Err(QueryError::Parse(format!(
651                                "variable-length traversal exceeded the safety depth cap ({VAR_EXPAND_DEPTH_CAP} \
652                                 hops) — likely a cyclic graph or unexpectedly large fanout; narrow the pattern or \
653                                 add an explicit upper bound (e.g. *0..10)"
654                            )));
655                        }
656                    }
657                }
658                Ok(out)
659            }
660            LogicalPlan::Filter { input, predicate } => {
661                let rows = self.eval_plan(txn, input, seed)?;
662                let mut out = Vec::with_capacity(rows.len());
663                for row in rows {
664                    if self.eval_expr(txn, predicate, &row)? {
665                        out.push(row);
666                    }
667                }
668                Ok(out)
669            }
670        }
671    }
672
673    fn scan(&self, txn: Txn, var: &str, label: Option<&str>) -> Result<Vec<BindingRow>, QueryError> {
674        let nodes = GraphStore::all_nodes_in_txn(txn, label)?;
675        Ok(nodes
676            .into_iter()
677            .map(|n| {
678                let mut row = BindingRow::new();
679                row.insert(var.to_string(), Binding::Node(n.id));
680                row
681            })
682            .collect())
683    }
684
685    fn eval_expr(&self, txn: Txn, expr: &Expr, row: &BindingRow) -> Result<bool, QueryError> {
686        Ok(match expr {
687            Expr::And(l, r) => self.eval_expr(txn, l, row)? && self.eval_expr(txn, r, row)?,
688            Expr::Or(l, r) => self.eval_expr(txn, l, row)? || self.eval_expr(txn, r, row)?,
689            Expr::Not(e) => !self.eval_expr(txn, e, row)?,
690            Expr::Compare(pa, op, lit) => {
691                let prop_value = self.lookup_prop(txn, pa, row)?;
692                compare(&prop_value, *op, lit)
693            }
694            Expr::HasLabel(var, label) => {
695                let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
696                let Binding::Node(id) = binding else {
697                    return Err(QueryError::UnboundVariable(var.clone()));
698                };
699                let node = GraphStore::get_node_in_txn(txn, *id)?;
700                node.is_some_and(|n| n.labels.iter().any(|l| l == label))
701            }
702            Expr::VarEq(a, b) => {
703                let ba = row.get(a).ok_or_else(|| QueryError::UnboundVariable(a.clone()))?;
704                let bb = row.get(b).ok_or_else(|| QueryError::UnboundVariable(b.clone()))?;
705                match (ba, bb) {
706                    (Binding::Node(x), Binding::Node(y)) => x == y,
707                    (Binding::Edge(x), Binding::Edge(y)) => x == y,
708                    // A null-padded `Binding::Value` (from an earlier
709                    // OPTIONAL MATCH that didn't match) can't equal a
710                    // real node/edge, and comparing across binding kinds
711                    // (a node vs an edge) is never meaningful here — the
712                    // planner only ever synthesizes VarEq between two
713                    // occurrences of the same pattern variable, which are
714                    // always the same kind when both are real.
715                    _ => false,
716                }
717            }
718        })
719    }
720
721    fn lookup_prop(
722        &self,
723        txn: Txn,
724        pa: &PropAccess,
725        row: &BindingRow,
726    ) -> Result<Option<PropertyValue>, QueryError> {
727        let binding = row
728            .get(&pa.var)
729            .ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
730        match binding {
731            Binding::Node(id) => {
732                let node = GraphStore::get_node_in_txn(txn, *id)?;
733                Ok(node.and_then(|n| n.props.get(&pa.prop).cloned()))
734            }
735            Binding::Edge(id) => {
736                let edge = GraphStore::get_edge_in_txn(txn, *id)?;
737                Ok(edge.and_then(|e| e.props.get(&pa.prop).cloned()))
738            }
739            // A WITH-projected scalar (or list) has no `.prop` to access —
740            // e.g. `WITH message.id AS messageId` then `messageId.foo`
741            // isn't meaningful. Treat as absent rather than erroring,
742            // consistent with how a missing property already behaves.
743            Binding::Value(_) | Binding::List(_) => Ok(None),
744        }
745    }
746
747    fn materialize_return(
748        &self,
749        txn: Txn,
750        items: &[ReturnItem],
751        rows: &[BindingRow],
752    ) -> Result<QueryResult, QueryError> {
753        let columns = items
754            .iter()
755            .enumerate()
756            .map(|(i, item)| item.alias.clone().unwrap_or_else(|| default_column_name(&item.expr, i)))
757            .collect();
758        let out_rows = if !has_aggregate(items) {
759            let mut out_rows = Vec::with_capacity(rows.len());
760            for row in rows {
761                let mut out_row = Vec::with_capacity(items.len());
762                for item in items {
763                    out_row.push(self.eval_return_expr(txn, &item.expr, row)?);
764                }
765                out_rows.push(out_row);
766            }
767            out_rows
768        } else {
769            validate_return_items(items)?;
770            let grouped = self.resolve_grouped_rows(txn, items, rows)?;
771            grouped
772                .into_iter()
773                .map(|bindings| {
774                    bindings
775                        .iter()
776                        .map(|b| self.binding_to_value(txn, b))
777                        .collect::<Result<Vec<_>, _>>()
778                })
779                .collect::<Result<Vec<_>, _>>()?
780        };
781        Ok(QueryResult {
782            columns,
783            rows: out_rows,
784        })
785    }
786
787    fn eval_return_expr(
788        &self,
789        txn: Txn,
790        expr: &ReturnExpr,
791        row: &BindingRow,
792    ) -> Result<Value, QueryError> {
793        match expr {
794            ReturnExpr::Var(var) => {
795                let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
796                self.binding_to_value(txn, binding)
797            }
798            ReturnExpr::Prop(pa) => {
799                let value = self.lookup_prop(txn, pa, row)?;
800                Ok(match value {
801                    // Collapse "prop missing" and "prop stored as null" into
802                    // one null representation — see Value::Null docs.
803                    Some(PropertyValue::Null) | None => Value::Null,
804                    Some(pv) => Value::Property(pv),
805                })
806            }
807            ReturnExpr::Lit(lit) => Ok(match lit {
808                Literal::Null => Value::Null,
809                other => Value::Literal(other.clone()),
810            }),
811            ReturnExpr::Call { name, args, .. } => {
812                // Reaching here with an aggregate name means an aggregate
813                // call slipped past `validate_return_items` (which only
814                // allows one at a return item's top level) — grouping
815                // itself never calls `eval_return_expr` on the aggregate
816                // wrapper, only on each aggregate's own argument
817                // subexpression (see `resolve_grouped_rows`), so this is
818                // an internal-consistency error, not a normal user path.
819                if is_aggregate_name(name) {
820                    return Err(QueryError::Parse(format!(
821                        "aggregate function '{name}' can only be used as a return item's top-level expression"
822                    )));
823                }
824                let arg_values = args
825                    .iter()
826                    .map(|a| self.eval_return_expr(txn, a, row))
827                    .collect::<Result<Vec<_>, _>>()?;
828                call_builtin(name, &arg_values)
829            }
830            ReturnExpr::CountStar => Err(QueryError::Parse(
831                "count(*) can only be used as a return item's top-level expression".into(),
832            )),
833            ReturnExpr::Case { test, whens, else_ } => {
834                let test_value = match test {
835                    Some(t) => Some(self.eval_return_expr(txn, t, row)?),
836                    None => None,
837                };
838                for (when, then) in whens {
839                    let when_value = self.eval_return_expr(txn, when, row)?;
840                    // Deliberately reuses the same Null == Null -> true
841                    // convention as `compare()` below, not standard
842                    // three-valued NULL logic — IS7's `CASE r WHEN null
843                    // THEN false ELSE true END` depends on this exact
844                    // semantics to detect an OPTIONAL MATCH non-match.
845                    let matched = match &test_value {
846                        Some(tv) => value_eq(tv, &when_value),
847                        None => matches!(when_value, Value::Literal(Literal::Bool(true))),
848                    };
849                    if matched {
850                        return self.eval_return_expr(txn, then, row);
851                    }
852                }
853                match else_ {
854                    Some(e) => self.eval_return_expr(txn, e, row),
855                    None => Ok(Value::Null),
856                }
857            }
858        }
859    }
860
861    fn materialize_delete(
862        &self,
863        write_txn: &WriteTransaction,
864        vars: &[String],
865        rows: &[BindingRow],
866        detach: bool,
867    ) -> Result<QueryResult, QueryError> {
868        let mut deleted_nodes = HashSet::new();
869        let mut deleted_edges = HashSet::new();
870        for row in rows {
871            for var in vars {
872                let binding = row.get(var).ok_or_else(|| QueryError::UnboundVariable(var.clone()))?;
873                match binding {
874                    Binding::Node(id) => {
875                        if deleted_nodes.insert(*id) {
876                            GraphStore::delete_node_in_txn(write_txn, *id, detach)?;
877                        }
878                    }
879                    Binding::Edge(id) => {
880                        if deleted_edges.insert(*id) {
881                            GraphStore::delete_edge_in_txn(write_txn, *id)?;
882                        }
883                    }
884                    Binding::Value(_) | Binding::List(_) => {
885                        return Err(QueryError::UnboundVariable(format!(
886                            "'{var}' is a WITH-projected scalar, not a node/edge — DELETE needs a graph binding"
887                        )))
888                    }
889                }
890            }
891        }
892        Ok(QueryResult {
893            columns: vec![],
894            rows: vec![],
895        })
896    }
897
898    fn materialize_set(
899        &self,
900        write_txn: &WriteTransaction,
901        items: &[(PropAccess, Literal)],
902        rows: &[BindingRow],
903    ) -> Result<QueryResult, QueryError> {
904        for row in rows {
905            for (pa, lit) in items {
906                let binding = row.get(&pa.var).ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
907                let value = literal_to_value(lit);
908                match binding {
909                    Binding::Node(id) => {
910                        GraphStore::set_node_prop_in_txn(write_txn, *id, &pa.prop, value)?;
911                    }
912                    Binding::Edge(id) => {
913                        GraphStore::set_edge_prop_in_txn(write_txn, *id, &pa.prop, value)?;
914                    }
915                    Binding::Value(_) | Binding::List(_) => {
916                        return Err(QueryError::UnboundVariable(format!(
917                            "'{}' is a WITH-projected scalar, not a node/edge — SET needs a graph binding",
918                            pa.var
919                        )))
920                    }
921                }
922            }
923        }
924        Ok(QueryResult {
925            columns: vec![],
926            rows: vec![],
927        })
928    }
929}
930
931/// A statement never mutates anything iff it's a `MATCH ... RETURN` with no
932/// `DELETE`/`DETACH DELETE`/`SET` tail — `Statement::Create` and every
933/// other `Tail` variant always write. Confirmed by tracing every function
934/// reachable from pattern/WHERE/WITH evaluation: none of them ever call a
935/// table-mutating `*_in_txn` method for a `Tail::Return` statement (a
936/// label-filtered scan looks up an existing label id, it never allocates
937/// one — allocation only happens in `create_node_in_txn`/
938/// `create_edge_in_txn`). `Executor::execute` uses this to decide whether
939/// to open a `ReadTransaction` (no contention with concurrent readers or a
940/// concurrent writer) or a `WriteTransaction`.
941fn is_read_only(stmt: &Statement) -> bool {
942    matches!(stmt, Statement::Match { tail: Tail::Return(_), .. })
943}
944
945/// Recovers the real `&WriteTransaction` from a `Txn` for the two
946/// `execute_match` tail arms (`DELETE`/`SET`) that need `.insert`/
947/// `.remove`, not just `Txn`'s read-only `get`/`iter`. Panics if given
948/// `Txn::Read` — which can't happen: `Tail::Delete`/`DetachDelete`/`Set`
949/// make `is_read_only` return `false`, so `Executor::execute` always opens
950/// a `WriteTransaction` (and thus `Txn::Write`) before reaching this path.
951fn require_write_txn(txn: Txn<'_>) -> &WriteTransaction {
952    let Txn::Write(write_txn) = txn else {
953        unreachable!(
954            "materialize_delete/materialize_set only reached via the write-dispatch path in \
955             Executor::execute — is_read_only(stmt) is false for any statement with a Delete/ \
956             DetachDelete/Set tail, so execute always opens a WriteTransaction for these"
957        )
958    };
959    write_txn
960}
961
962fn default_column_name(expr: &ReturnExpr, idx: usize) -> String {
963    match expr {
964        ReturnExpr::Var(v) => v.clone(),
965        ReturnExpr::Prop(pa) => format!("{}.{}", pa.var, pa.prop),
966        ReturnExpr::Lit(_) => format!("col{idx}"),
967        ReturnExpr::Call { name, .. } => format!("{name}(...)"),
968        ReturnExpr::CountStar => "count(*)".to_string(),
969        ReturnExpr::Case { .. } => format!("case{idx}"),
970    }
971}
972
973/// The name a `WITH`/`RETURN` item is known by afterward — its alias, or
974/// a name derived from the expression (its bare var name, `col{i}`, etc).
975fn with_item_output_name((i, item): (usize, &ReturnItem)) -> String {
976    item.alias.clone().unwrap_or_else(|| default_column_name(&item.expr, i))
977}
978
979/// True iff `expr` is itself an aggregate call — `count(*)`, or a `Call`
980/// whose name is in `is_aggregate_name`'s fixed set. Does NOT look inside
981/// `expr` for a nested aggregate — see `contains_aggregate` for that.
982fn is_top_level_aggregate(expr: &ReturnExpr) -> bool {
983    match expr {
984        ReturnExpr::CountStar => true,
985        ReturnExpr::Call { name, .. } => is_aggregate_name(name),
986        _ => false,
987    }
988}
989
990/// True iff `expr` contains an aggregate call anywhere inside it, at any
991/// depth — used to reject an aggregate nested inside another aggregate's
992/// argument, or inside a non-aggregate expression's `CASE`/`Call`
993/// arguments (an aggregate must be a return item's *entire* top-level
994/// expression — see `validate_return_items`).
995fn contains_aggregate(expr: &ReturnExpr) -> bool {
996    match expr {
997        ReturnExpr::CountStar => true,
998        ReturnExpr::Call { name, args, .. } => is_aggregate_name(name) || args.iter().any(contains_aggregate),
999        ReturnExpr::Case { test, whens, else_ } => {
1000            test.as_deref().is_some_and(contains_aggregate)
1001                || whens.iter().any(|(w, t)| contains_aggregate(w) || contains_aggregate(t))
1002                || else_.as_deref().is_some_and(contains_aggregate)
1003        }
1004        ReturnExpr::Var(_) | ReturnExpr::Prop(_) | ReturnExpr::Lit(_) => false,
1005    }
1006}
1007
1008/// True iff any item's top-level expression is an aggregate call —
1009/// `materialize_with`/`materialize_return` dispatch to the grouping path
1010/// iff this is true, otherwise the existing row-at-a-time path runs
1011/// completely unchanged (zero perf/behavior impact on non-aggregating
1012/// queries).
1013fn has_aggregate(items: &[ReturnItem]) -> bool {
1014    items.iter().any(|item| is_top_level_aggregate(&item.expr))
1015}
1016
1017/// Validates a RETURN/WITH item list before any row is processed: every
1018/// aggregate call has exactly one argument (`count(*)`, the zero-argument
1019/// form, is `CountStar`, a separate variant — never reaches the `Call`
1020/// arm here), no aggregate's own argument contains a nested aggregate
1021/// call, and no non-aggregate item's expression contains an aggregate
1022/// call anywhere inside it (aggregates must be a return item's entire
1023/// top-level expression — justified by there being no arithmetic
1024/// operators anywhere in this engine yet, so `count(n) * 2`-style
1025/// composition is already impossible, and nothing in the target query set
1026/// needs an aggregate nested inside a `CASE` branch).
1027fn validate_return_items(items: &[ReturnItem]) -> Result<(), QueryError> {
1028    for item in items {
1029        match &item.expr {
1030            ReturnExpr::CountStar => {}
1031            ReturnExpr::Call { name, args, .. } if is_aggregate_name(name) => {
1032                if args.len() != 1 {
1033                    return Err(QueryError::Parse(format!(
1034                        "{name}() takes exactly one argument (use count(*) for a row count with no argument)"
1035                    )));
1036                }
1037                if contains_aggregate(&args[0]) {
1038                    return Err(QueryError::Parse(format!(
1039                        "aggregate function '{name}' can't take another aggregate as an argument"
1040                    )));
1041                }
1042            }
1043            other => {
1044                if contains_aggregate(other) {
1045                    return Err(QueryError::Parse(
1046                        "an aggregate function must be a return item's entire expression, not nested inside \
1047                         another expression"
1048                            .into(),
1049                    ));
1050                }
1051            }
1052        }
1053    }
1054    Ok(())
1055}
1056
1057/// Grouping-key hashing — deliberately at the `Binding` level (`NodeId`/
1058/// `EdgeId`/`PropertyValue`), not `Value`: cheaper (no `GraphStore` fetch
1059/// just to compute) and the correct semantics (two `Binding::Node`s are
1060/// the same group iff the same node **identity**, not equal-by-struct-
1061/// contents). `Binding::List`'s elements are `Value`s already, so those
1062/// delegate to `value_hash_key` directly.
1063fn binding_hash_key(b: &Binding) -> HashKey {
1064    match b {
1065        Binding::Node(id) => HashKey::Node(*id),
1066        Binding::Edge(id) => HashKey::Edge(*id),
1067        Binding::Value(pv) => property_value_hash_key(pv),
1068        Binding::List(items) => HashKey::List(items.iter().map(value_hash_key).collect()),
1069    }
1070}
1071
1072/// Converts a finished `AggAcc::finish()` result to the `Binding` it's
1073/// carried as through a `WITH` boundary — `collect()`'s `Value::List`
1074/// needs `Binding::List` (no list variant in `PropertyValue`, the
1075/// storage-layer type `Binding::Value` wraps), everything else collapses
1076/// to `Binding::Value` same as any other computed WITH item.
1077fn value_to_binding(v: Value) -> Binding {
1078    match v {
1079        Value::List(items) => Binding::List(items),
1080        other => Binding::Value(value_to_property_value(&other)),
1081    }
1082}
1083
1084/// `WithExpr::Compare`'s value-vs-literal comparison — reuses `compare()`
1085/// (below) by reducing a `Value` down to the `Option<PropertyValue>` shape
1086/// it expects; `Node`/`Edge`/`List` have no meaningful comparison against
1087/// a `Literal` and fall back to "absent", same as a missing property does.
1088fn compare_value(value: &Value, op: CompareOp, lit: &Literal) -> bool {
1089    let prop = match value {
1090        Value::Null => None,
1091        Value::Property(pv) => Some(pv.clone()),
1092        Value::Literal(l) => Some(literal_to_value(l)),
1093        Value::Node(_) | Value::Edge(_) | Value::List(_) => None,
1094    };
1095    compare(&prop, op, lit)
1096}
1097
1098/// Coerces a materialized `Value` down to a `PropertyValue` for storing in
1099/// `Binding::Value` — used by `item_binding` for a computed (non-bare-var)
1100/// WITH/RETURN item. `Value::Node`/`Edge` can't occur here in practice (no
1101/// non-aggregate `ReturnExpr` form produces one except `Var`, which takes
1102/// the bare-variable path instead). `Value::List` can't occur here either
1103/// — `collect()` only ever appears in an aggregating item list, which
1104/// `has_aggregate` routes to `resolve_grouped_rows`/`Binding::List`
1105/// instead of through `item_binding` at all. Both fall back to `Null`
1106/// rather than needing a fallible signature for an unreachable case.
1107fn value_to_property_value(v: &Value) -> PropertyValue {
1108    match v {
1109        Value::Null => PropertyValue::Null,
1110        Value::Property(pv) => pv.clone(),
1111        Value::Literal(lit) => literal_to_value(lit),
1112        Value::Node(_) | Value::Edge(_) | Value::List(_) => PropertyValue::Null,
1113    }
1114}
1115
1116fn literal_to_value(lit: &Literal) -> PropertyValue {
1117    match lit {
1118        Literal::Int(i) => PropertyValue::Int(*i),
1119        Literal::Float(f) => PropertyValue::Float(*f),
1120        Literal::String(s) => PropertyValue::String(s.clone()),
1121        Literal::Bool(b) => PropertyValue::Bool(*b),
1122        Literal::Null => PropertyValue::Null,
1123        Literal::Param(name) => {
1124            unreachable!("param ${name} must be substituted before execution — see params::substitute_params")
1125        }
1126    }
1127}
1128
1129fn literal_props_to_values(props: &[(String, Literal)]) -> BTreeMap<String, PropertyValue> {
1130    props.iter().map(|(k, v)| (k.clone(), literal_to_value(v))).collect()
1131}
1132
1133fn pattern_labels(labels: &[String]) -> Vec<&str> {
1134    if labels.is_empty() {
1135        vec!["Node"]
1136    } else {
1137        labels.iter().map(|s| s.as_str()).collect()
1138    }
1139}
1140
1141/// `Either` (undirected `-[r:TYPE]-`) has no single storage-level call —
1142/// query both directions and dedupe by `edge_id` (a self-loop would
1143/// otherwise appear twice, once from each direction's adjacency table).
1144fn neighbors_for_direction(
1145    txn: Txn,
1146    node: NodeId,
1147    direction: ExpandDirection,
1148    rel_label: Option<&str>,
1149) -> Result<Vec<AdjEntry>, QueryError> {
1150    Ok(match direction {
1151        ExpandDirection::Out => GraphStore::neighbors_in_txn(txn, node, Direction::Out, rel_label)?,
1152        ExpandDirection::In => GraphStore::neighbors_in_txn(txn, node, Direction::In, rel_label)?,
1153        ExpandDirection::Either => {
1154            let mut out = GraphStore::neighbors_in_txn(txn, node, Direction::Out, rel_label)?;
1155            let inbound = GraphStore::neighbors_in_txn(txn, node, Direction::In, rel_label)?;
1156            let seen: HashSet<EdgeId> = out.iter().map(|e| e.edge_id).collect();
1157            out.extend(inbound.into_iter().filter(|e| !seen.contains(&e.edge_id)));
1158            out
1159        }
1160    })
1161}
1162
1163fn compare(prop: &Option<PropertyValue>, op: CompareOp, lit: &Literal) -> bool {
1164    let Some(prop) = prop else { return false };
1165    match (prop, lit) {
1166        (PropertyValue::Int(a), Literal::Int(b)) => cmp_f64(op, *a as f64, *b as f64),
1167        (PropertyValue::Int(a), Literal::Float(b)) => cmp_f64(op, *a as f64, *b),
1168        (PropertyValue::Float(a), Literal::Float(b)) => cmp_f64(op, *a, *b),
1169        (PropertyValue::Float(a), Literal::Int(b)) => cmp_f64(op, *a, *b as f64),
1170        (PropertyValue::String(a), Literal::String(b)) => cmp_ord(op, a.as_str(), b.as_str()),
1171        (PropertyValue::Bool(a), Literal::Bool(b)) => match op {
1172            CompareOp::Eq => a == b,
1173            CompareOp::Ne => a != b,
1174            _ => false,
1175        },
1176        (PropertyValue::Null, Literal::Null) => matches!(op, CompareOp::Eq),
1177        _ => false,
1178    }
1179}
1180
1181fn cmp_f64(op: CompareOp, a: f64, b: f64) -> bool {
1182    match op {
1183        CompareOp::Eq => a == b,
1184        CompareOp::Ne => a != b,
1185        CompareOp::Lt => a < b,
1186        CompareOp::Le => a <= b,
1187        CompareOp::Gt => a > b,
1188        CompareOp::Ge => a >= b,
1189    }
1190}
1191
1192fn cmp_ord<T: PartialOrd>(op: CompareOp, a: T, b: T) -> bool {
1193    match op {
1194        CompareOp::Eq => a == b,
1195        CompareOp::Ne => a != b,
1196        CompareOp::Lt => a < b,
1197        CompareOp::Le => a <= b,
1198        CompareOp::Gt => a > b,
1199        CompareOp::Ge => a >= b,
1200    }
1201}
1202
1203/// Value equality for CASE's WHEN-comparison (and, elsewhere, DISTINCT
1204/// dedup within an aggregate). Null == Null -> true here deliberately,
1205/// matching `compare()`'s convention above, not standard three-valued NULL
1206/// logic. `Node`/`Edge` compare by id (graph identity), not full-struct
1207/// contents — cheaper, and the correct semantics regardless (two bindings
1208/// are "the same node" iff the same node, not iff their label/prop
1209/// snapshots happen to match).
1210pub(crate) fn value_eq(a: &Value, b: &Value) -> bool {
1211    match (a, b) {
1212        (Value::Null, Value::Null) => true,
1213        (Value::Null, _) | (_, Value::Null) => false,
1214        (Value::Property(pa), Value::Property(pb)) => pa == pb,
1215        (Value::Literal(la), Value::Literal(lb)) => la == lb,
1216        (Value::Property(pa), Value::Literal(lb)) => *pa == literal_to_value(lb),
1217        (Value::Literal(la), Value::Property(pb)) => literal_to_value(la) == *pb,
1218        (Value::Node(na), Value::Node(nb)) => na.id == nb.id,
1219        (Value::Edge(ea), Value::Edge(eb)) => ea.id == eb.id,
1220        (Value::List(la), Value::List(lb)) => la.len() == lb.len() && la.iter().zip(lb).all(|(x, y)| value_eq(x, y)),
1221        _ => false,
1222    }
1223}
1224
1225fn call_builtin(name: &str, args: &[Value]) -> Result<Value, QueryError> {
1226    match name.to_ascii_lowercase().as_str() {
1227        "coalesce" => Ok(args
1228            .iter()
1229            .find(|v| !matches!(v, Value::Null))
1230            .cloned()
1231            .unwrap_or(Value::Null)),
1232        "tointeger" => Ok(args.first().map(to_integer).unwrap_or(Value::Null)),
1233        other => Err(QueryError::Parse(format!("unknown function: {other}"))),
1234    }
1235}
1236
1237fn to_integer(v: &Value) -> Value {
1238    let as_str_parse = |s: &str| match s.trim().parse::<i64>() {
1239        Ok(i) => Value::Property(PropertyValue::Int(i)),
1240        Err(_) => Value::Null,
1241    };
1242    match v {
1243        Value::Property(PropertyValue::Int(i)) => Value::Property(PropertyValue::Int(*i)),
1244        Value::Property(PropertyValue::Float(f)) => Value::Property(PropertyValue::Int(*f as i64)),
1245        Value::Property(PropertyValue::String(s)) => as_str_parse(s),
1246        Value::Literal(Literal::Int(i)) => Value::Property(PropertyValue::Int(*i)),
1247        Value::Literal(Literal::Float(f)) => Value::Property(PropertyValue::Int(*f as i64)),
1248        Value::Literal(Literal::String(s)) => as_str_parse(s),
1249        _ => Value::Null,
1250    }
1251}
1252
1253/// Sorts `rows` (already-projected `RETURN`/`WITH` output, `columns`
1254/// aligned by index) by `order_by`, which evaluates against the projected
1255/// column names — never the raw pattern `BindingRow` — since every ORDER BY
1256/// key in practice is a RETURN/WITH alias, not a bare pattern variable.
1257fn apply_order_by(
1258    rows: Vec<Vec<Value>>,
1259    columns: &[String],
1260    order_by: &[(ReturnExpr, SortDir)],
1261) -> Result<Vec<Vec<Value>>, QueryError> {
1262    let mut keyed: Vec<(Vec<Value>, Vec<Value>)> = Vec::with_capacity(rows.len());
1263    for row in rows {
1264        let row_map: HashMap<String, Value> = columns.iter().cloned().zip(row.iter().cloned()).collect();
1265        let keys = order_by
1266            .iter()
1267            .map(|(expr, _)| eval_projected_expr(expr, &row_map))
1268            .collect::<Result<Vec<_>, _>>()?;
1269        keyed.push((keys, row));
1270    }
1271    keyed.sort_by(|(ka, _), (kb, _)| {
1272        for (i, (_, dir)) in order_by.iter().enumerate() {
1273            let ord = compare_with_dir(&ka[i], &kb[i], *dir);
1274            if ord != std::cmp::Ordering::Equal {
1275                return ord;
1276            }
1277        }
1278        std::cmp::Ordering::Equal
1279    });
1280    Ok(keyed.into_iter().map(|(_, row)| row).collect())
1281}
1282
1283/// Same expression shape as `eval_return_expr`, but resolves `Var`/`Prop`
1284/// against already-projected output columns instead of the graph-bound
1285/// `BindingRow` — no `WriteTransaction`/`GraphStore` access needed, since a
1286/// projected `Value::Node`/`Value::Edge` already carries its full record
1287/// (including props) from when it was first materialized.
1288fn eval_projected_expr(expr: &ReturnExpr, row: &HashMap<String, Value>) -> Result<Value, QueryError> {
1289    match expr {
1290        ReturnExpr::Var(name) => row
1291            .get(name)
1292            .cloned()
1293            .ok_or_else(|| QueryError::UnboundVariable(name.clone())),
1294        ReturnExpr::Prop(pa) => {
1295            let base = row
1296                .get(&pa.var)
1297                .ok_or_else(|| QueryError::UnboundVariable(pa.var.clone()))?;
1298            let pv = match base {
1299                Value::Node(n) => n.props.get(&pa.prop).cloned(),
1300                Value::Edge(e) => e.props.get(&pa.prop).cloned(),
1301                _ => None,
1302            };
1303            Ok(match pv {
1304                Some(PropertyValue::Null) | None => Value::Null,
1305                Some(v) => Value::Property(v),
1306            })
1307        }
1308        ReturnExpr::Lit(lit) => Ok(match lit {
1309            Literal::Null => Value::Null,
1310            other => Value::Literal(other.clone()),
1311        }),
1312        ReturnExpr::Call { name, args, .. } => {
1313            // Same internal-consistency stance as `eval_return_expr`'s
1314            // `Call` arm: by the time ORDER BY runs, aggregation has
1315            // already resolved into ordinary named output columns
1316            // (referenced here via `Var`), so a raw aggregate `Call`
1317            // reaching this point means it wasn't top-level as
1318            // `validate_return_items` requires.
1319            if is_aggregate_name(name) {
1320                return Err(QueryError::Parse(format!(
1321                    "aggregate function '{name}' can only be used as a return item's top-level expression"
1322                )));
1323            }
1324            let arg_values = args
1325                .iter()
1326                .map(|a| eval_projected_expr(a, row))
1327                .collect::<Result<Vec<_>, _>>()?;
1328            call_builtin(name, &arg_values)
1329        }
1330        ReturnExpr::CountStar => Err(QueryError::Parse(
1331            "count(*) can only be used as a return item's top-level expression".into(),
1332        )),
1333        ReturnExpr::Case { test, whens, else_ } => {
1334            let test_value = match test {
1335                Some(t) => Some(eval_projected_expr(t, row)?),
1336                None => None,
1337            };
1338            for (when, then) in whens {
1339                let when_value = eval_projected_expr(when, row)?;
1340                let matched = match &test_value {
1341                    Some(tv) => value_eq(tv, &when_value),
1342                    None => matches!(when_value, Value::Literal(Literal::Bool(true))),
1343                };
1344                if matched {
1345                    return eval_projected_expr(then, row);
1346                }
1347            }
1348            match else_ {
1349                Some(e) => eval_projected_expr(e, row),
1350                None => Ok(Value::Null),
1351            }
1352        }
1353    }
1354}
1355
1356/// NULLs sort last regardless of ASC/DESC (matches Neo4j's documented
1357/// behavior) — only non-null comparisons get reversed for DESC.
1358fn compare_with_dir(a: &Value, b: &Value, dir: SortDir) -> std::cmp::Ordering {
1359    use std::cmp::Ordering;
1360    let a_null = matches!(a, Value::Null);
1361    let b_null = matches!(b, Value::Null);
1362    match (a_null, b_null) {
1363        (true, true) => return Ordering::Equal,
1364        (true, false) => return Ordering::Greater,
1365        (false, true) => return Ordering::Less,
1366        (false, false) => {}
1367    }
1368    let ord = compare_non_null(a, b);
1369    if dir == SortDir::Desc {
1370        ord.reverse()
1371    } else {
1372        ord
1373    }
1374}
1375
1376fn compare_non_null(a: &Value, b: &Value) -> std::cmp::Ordering {
1377    use std::cmp::Ordering;
1378    let pa = value_to_comparable(a);
1379    let pb = value_to_comparable(b);
1380    match (pa, pb) {
1381        (Some(PropertyValue::Int(x)), Some(PropertyValue::Int(y))) => x.cmp(&y),
1382        (Some(PropertyValue::Int(x)), Some(PropertyValue::Float(y))) => {
1383            (x as f64).partial_cmp(&y).unwrap_or(Ordering::Equal)
1384        }
1385        (Some(PropertyValue::Float(x)), Some(PropertyValue::Int(y))) => {
1386            x.partial_cmp(&(y as f64)).unwrap_or(Ordering::Equal)
1387        }
1388        (Some(PropertyValue::Float(x)), Some(PropertyValue::Float(y))) => x.partial_cmp(&y).unwrap_or(Ordering::Equal),
1389        (Some(PropertyValue::String(x)), Some(PropertyValue::String(y))) => x.cmp(&y),
1390        (Some(PropertyValue::Bool(x)), Some(PropertyValue::Bool(y))) => x.cmp(&y),
1391        _ => Ordering::Equal,
1392    }
1393}
1394
1395fn value_to_comparable(v: &Value) -> Option<PropertyValue> {
1396    match v {
1397        Value::Property(pv) => Some(pv.clone()),
1398        Value::Literal(lit) => Some(literal_to_value(lit)),
1399        _ => None,
1400    }
1401}
1402
1403/// Ordering for `min`/`max` aggregate folding — `None` for values with no
1404/// natural order (`Node`/`Edge`/`List`, or a `Null`, which `AggAcc::fold`
1405/// never passes here anyway since null contributions are skipped before
1406/// folding). The caller turns `None` into a clear error rather than an
1407/// arbitrary "always equal" fallback — unlike ORDER BY's
1408/// `compare_non_null`, which tolerates that for presentation ordering
1409/// (see its docs), silently treating two nodes as "equal" inside an
1410/// aggregate would be a wrong-answer failure mode, not just an
1411/// unhelpful sort order.
1412pub(crate) fn comparable_ordering(a: &Value, b: &Value) -> Option<std::cmp::Ordering> {
1413    use std::cmp::Ordering;
1414    let pa = value_to_comparable(a)?;
1415    let pb = value_to_comparable(b)?;
1416    Some(match (pa, pb) {
1417        (PropertyValue::Int(x), PropertyValue::Int(y)) => x.cmp(&y),
1418        (PropertyValue::Int(x), PropertyValue::Float(y)) => (x as f64).partial_cmp(&y).unwrap_or(Ordering::Equal),
1419        (PropertyValue::Float(x), PropertyValue::Int(y)) => x.partial_cmp(&(y as f64)).unwrap_or(Ordering::Equal),
1420        (PropertyValue::Float(x), PropertyValue::Float(y)) => x.partial_cmp(&y).unwrap_or(Ordering::Equal),
1421        (PropertyValue::String(x), PropertyValue::String(y)) => x.cmp(&y),
1422        (PropertyValue::Bool(x), PropertyValue::Bool(y)) => x.cmp(&y),
1423        _ => return None,
1424    })
1425}