core_query/cypher/

plan.rs

//! Cypher logical planner: `Query` → `Vec<PlanOp>`.
//!
//! Pure (no `GraphView`). Never panics. Bound-destination handling lives on
//! `Expand` (see `PlanOp::Expand`); `JoinBound` is only emitted for the
//! *start* node of a MATCH whose variable is already bound.

use super::ast::{
    ret_val_label, AggArg, AggFunc, Expr, LimitSkip, NodePat, Operand, OptionalClause, OrderItem,
    OrderTarget, Pattern, Query, RelDir, RelPat, RetItem, RetVal, UnwindExpr, WithStage,
};
use crate::filter::CmpOp;
use std::collections::BTreeSet;

/// One operator in the logical plan. Patterns compile left-to-right into
/// scan / join / expand ops; WHERE is a single `Filter`; then `Project`,
/// rewritten `OrderBy`, `Skip`, `Limit` in that order.
#[derive(Debug, Clone, PartialEq)]
pub enum PlanOp {
    /// Seed rows from all nodes, or those with `label`.
    ScanLabel {
        var: String,
        label: Option<String>,
    },
    /// Point lookup by IdMap key. Emitted when a MATCH node property map is
    /// exactly one equality on field `id` (mixed maps stay ScanLabel+LookupProps).
    ScanKey {
        var: String,
        key: Operand,
        label: Option<String>,
    },
    /// Indexed equality lookup: seed rows from nodes of `label` whose scalar
    /// `field` equals `value`. Emitted when a MATCH node property map is exactly
    /// one equality on a non-`id` field. The executor uses the property index
    /// when `(label, field)` is declared, else falls back to a scan+filter, so
    /// this op is always correct regardless of whether the index exists.
    IndexScan {
        var: String,
        label: Option<String>,
        field: String,
        value: Operand,
    },
    /// Compound indexed equality lookup: seed rows from nodes whose scalar
    /// properties all satisfy two or more equalities. Emitted when a MATCH node
    /// has ≥2 non-`id` scalar equalities (inline map, WHERE fold, or a mix).
    ///
    /// The executor resolves each `(field, value)` via `nodes_with_prop`:
    /// indexed fields contribute a candidate list; unindexed fields become
    /// per-node post-filters (`node_matches`). When ALL fields are unindexed it
    /// falls back to a full label scan + filter — identical semantics to
    /// `ScanLabel + LookupProps` but expressed as a single op for later passes.
    /// `INDEX_INTERSECT_FIRES` advances only when at least one field is indexed.
    IndexIntersect {
        var: String,
        label: Option<String>,
        equalities: Vec<(String, Operand)>,
    },
    /// Retain rows whose `var` node matches the pattern-map props.
    LookupProps {
        var: String,
        props: Vec<(String, Operand)>,
    },
    /// Expand from `from` along `etype`/`dir`, binding `rel_var` and `to`.
    ///
    /// Destination label/prop checks ride on this op (`to_label` / `to_props`).
    /// If `to` is already bound in the row, the executor keeps only edges
    /// that land on that bound id (JoinBound semantics *inside* Expand).
    /// `rel_var` is always `Some` after planning: user name or `_rN`.
    Expand {
        from: String,
        rel_var: Option<String>,
        etypes: Vec<String>,
        dir: RelDir,
        to: String,
        to_label: Option<String>,
        to_props: Vec<(String, Operand)>,
    },
    /// Pattern-*start* node whose var is already bound: label/prop re-check only.
    JoinBound {
        var: String,
        label: Option<String>,
        props: Vec<(String, Operand)>,
    },
    Filter {
        expr: Expr,
    },
    Project {
        items: Vec<RetItem>,
    },
    /// Deduplicate projected rows (`RETURN DISTINCT`). Hashed with the same
    /// numeric unify as `GroupAggregate`. Caps distinct rows at the
    /// intermediate-row budget.
    Distinct,
    /// After `plan`, every item's `target` is `OrderTarget::Alias(column)`
    /// where `column` is a projected column name. The executor resolves
    /// ORDER BY against the post-Project table only.
    OrderBy {
        items: Vec<OrderItem>,
    },
    Skip(LimitSkip),
    Limit(LimitSkip),
    /// Single aggregate over all matched rows (no grouping).
    ///
    /// Execution routes to a streaming accumulator path (O(1) memory).
    /// The 1 M intermediate-row budget does **not** apply: the accumulator
    /// holds a single running value regardless of how many source rows exist.
    ///
    /// Null/non-numeric values in `arg` are silently skipped for SUM/AVG/MIN/MAX.
    Aggregate {
        func: AggFunc,
        arg: AggArg,
        /// Projected column name — alias if provided, else the canonical
        /// function call string (`COUNT(*)`, `SUM(n.age)`, etc.).
        column: String,
    },
    /// Grouped aggregation: one or more group-key items and one or more aggregate
    /// functions, computed per distinct group.
    ///
    /// The executor streams through all matching rows, computing one
    /// `Option<ValueKey>` per group-key item; `None` represents a null value,
    /// and null keys group together (openCypher semantics).
    ///
    /// Group count is capped at 1,000,000; exceeding the cap is an error.
    ///
    /// `ORDER BY` / `SKIP` / `LIMIT` ops that follow in the plan apply to the
    /// finished group table (sort the groups, then slice).  `row_bound()` always
    /// returns `None` for plans containing this op so that LIMIT is never pushed
    /// into producers.
    GroupAggregate {
        /// Non-aggregate RETURN items: `(projected_column_name, ret_item)`.
        keys: Vec<(String, RetItem)>,
        /// Aggregate RETURN items: `(func, arg, projected_column_name)`.
        aggs: Vec<(AggFunc, AggArg, String)>,
    },
    /// Variable-length path expansion: BFS from `from`, emitting one row per
    /// (start, end, depth) path found with `min ≤ depth ≤ max`.
    ///
    /// Per-path edge-uniqueness (Cypher relationship isomorphism): a single
    /// path may not reuse the same edge (`EdgeRef`) twice; node revisits ARE
    /// allowed.  `rel_var`, when present, is bound to a virtual path cell
    /// whose sole accessible property is `length` (hop count as `Int`).
    ///
    /// Always executes via the **staged path** regardless of LIMIT.
    /// The 1 M intermediate-row budget applies to the output row count.
    VarExpand {
        from: String,
        rel_var: Option<String>,
        etypes: Vec<String>,
        dir: RelDir,
        to: String,
        min: u8,
        max: u8,
    },
    /// Shortest path between two already-bound nodes via BFS.
    ///
    /// Both `from` and `to` must be bound in the current row before this op
    /// executes.  BFS terminates at the first depth where `to` is reached.
    /// If `to` is unreachable within `max_hops`, zero rows are emitted.
    /// Exactly one row is emitted when a path exists.
    ///
    /// `rel_var`, when present, binds to a virtual path cell; `r.length`
    /// yields the hop count as `Int`.
    ///
    /// Always executes via the **staged path** regardless of LIMIT.
    ShortestPath {
        from: String,
        rel_var: Option<String>,
        etypes: Vec<String>,
        dir: RelDir,
        to: String,
        max_hops: u8,
    },
    /// Non-aggregate WITH: apply filter / order / skip / limit to the current
    /// row set without projecting. Node bindings in the row survive as-is so
    /// that subsequent MATCH clauses can join against them.
    ///
    /// Always executes via the **staged path** (row_bound returns None for any
    /// plan containing this op).
    With {
        items: Vec<RetItem>,
        where_expr: Option<Expr>,
        order_by: Vec<OrderItem>,
        skip: Option<LimitSkip>,
        limit: Option<LimitSkip>,
    },
    /// UNWIND: expand each input row into N rows by iterating a list value.
    ///
    /// - `list: UnwindExpr::Lit(v)` → use a literal list.
    /// - `list: UnwindExpr::Prop { var, field }` → resolve the list from a node property.
    /// - `list: UnwindExpr::Var(name)` → look up a scalar binding from a prior WITH.
    ///
    /// null / empty list → 0 output rows (openCypher).
    /// Non-list → named error at execution time.
    ///
    /// Always executes via the **staged path** (row_bound returns None).
    Unwind {
        expr: UnwindExpr,
        alias: String,
    },
    /// OPTIONAL MATCH: left-outer-join semantics.
    ///
    /// For each input row the `inner` plan is executed in isolation.  If the
    /// inner plan produces at least one output row, those rows replace the
    /// input row (inner join semantics for the rows that match).  If the inner
    /// plan produces **zero** rows, the input row survives with every variable
    /// listed in `optional_vars` set to null (left-outer fallback).
    ///
    /// `optional_vars` lists the variables that are introduced inside the
    /// optional pattern (i.e., the variables that must be nulled when the
    /// pattern fails).  Variables that were already bound before the optional
    /// clause are not listed here — they continue to hold their original values
    /// in the null row.
    ///
    /// Always executes via the **staged path** (row_bound returns None).
    LeftOuterApply {
        inner: Vec<PlanOp>,
        optional_vars: Vec<String>,
    },
}

/// Compute the effective row bound for LIMIT push-down.
///
/// Returns `Some(SKIP + LIMIT)` when the plan can terminate producers early —
/// that is, when the plan contains a `Limit` op **and no `OrderBy`** op.
/// An `OrderBy` requires full materialisation before slicing, so the bound
/// cannot be pushed past it.
///
/// Returns `None` when:
/// - No `Limit` op is present, or
/// - An `OrderBy` op is present (sorting needs every row).
///
/// # Decision table
///
/// | Plan shape (tail before Project)           | push-down? | note                                         |
/// |--------------------------------------------|------------|----------------------------------------------|
/// | Scan / Expand → Project → Limit            | YES        | pull-based; all stages stop at bound         |
/// | Scan / Expand → Filter → Project → Limit   | YES        | pull-based; Filter + earlier stages all stop |
/// | … → OrderBy → … Limit                      | NO         | sort requires all rows first                 |
///
/// When `row_bound` returns `Some`, the executor uses a demand-driven
/// (pull-based) strategy: **all** producer stages (Scan, Expand, Filter, …)
/// terminate as soon as `bound` final rows have been collected.  No
/// intermediate table is ever fully materialised for the bounded path.
///
/// `SKIP + LIMIT` is used instead of plain `LIMIT` so that there are enough
/// rows to apply the `SKIP` offset and still yield `LIMIT` final rows.
/// Saturating addition is used to guard against pathological large values.
pub fn row_bound(ops: &[PlanOp]) -> Option<usize> {
    // ORDER BY and DISTINCT require full materialisation — bound cannot be pushed.
    if ops
        .iter()
        .any(|op| matches!(op, PlanOp::OrderBy { .. } | PlanOp::Distinct))
    {
        return None;
    }
    // Aggregate plans use the streaming accumulator path, not the pull path.
    if ops.iter().any(|op| matches!(op, PlanOp::Aggregate { .. })) {
        return None;
    }
    // GroupAggregate plans are a sink over the full row stream; ORDER BY and LIMIT
    // apply to the finished group table, never to producers.
    if ops
        .iter()
        .any(|op| matches!(op, PlanOp::GroupAggregate { .. }))
    {
        return None;
    }
    // VarExpand / ShortestPath always use the staged path so that the 1M row
    // budget applies and BFS state is cleanly managed stage-by-stage.
    if ops
        .iter()
        .any(|op| matches!(op, PlanOp::VarExpand { .. } | PlanOp::ShortestPath { .. }))
    {
        return None;
    }
    // Pipeline plans (WITH / UNWIND / LeftOuterApply) always use the staged
    // path so that intermediate rows are correctly bounded and sequenced.
    if ops.iter().any(|op| {
        matches!(
            op,
            PlanOp::With { .. } | PlanOp::Unwind { .. } | PlanOp::LeftOuterApply { .. }
        )
    }) {
        return None;
    }
    let limit_n = ops.iter().rev().find_map(|op| match op {
        PlanOp::Limit(LimitSkip::Exact(n)) => Some(*n),
        PlanOp::Limit(LimitSkip::Param(_)) => None, // param-limit: can't determine bound statically
        _ => None,
    })?;
    // A param Skip means the skip count is unknown at plan time — force staged.
    if ops
        .iter()
        .any(|op| matches!(op, PlanOp::Skip(LimitSkip::Param(_))))
    {
        return None;
    }
    let skip_n = ops
        .iter()
        .rev()
        .find_map(|op| match op {
            PlanOp::Skip(LimitSkip::Exact(n)) => Some(*n),
            _ => None,
        })
        .unwrap_or(0);
    Some((skip_n as usize).saturating_add(limit_n as usize))
}

/// Returns `true` when the plan shape is supported by `subscribe_query`.
///
/// Allowlisted shapes (documented subset — not full Cypher):
///   • `MATCH (n:Label) WHERE … RETURN …`
///     → ScanLabel or ScanKey, optional LookupProps, optional Filter, Project,
///       optional Limit.
///   • `MATCH (a)-[r:TYPE]->(b) RETURN …`
///     → ScanLabel or ScanKey, exactly one Expand, optional Filter, Project,
///       optional Limit.
///
/// Everything else is rejected: SKIP (creates unstable offset windows), multi-hop
/// Expand chains, ORDER BY, DISTINCT, aggregates, variable-length paths, OPTIONAL
/// MATCH, WITH, UNWIND, JoinBound (multi-MATCH). Use LIMIT to bound re-execution
/// cost (`subscribe_query` does a full re-run per commit).
pub fn is_subscribable(ops: &[PlanOp]) -> bool {
    // All ops must be from the allowlisted set. Skip is excluded: SKIP N shifts
    // the result window on every commit, causing spurious Added/Removed churn for
    // rows whose data never changed.
    ops.iter().all(|op| {
        matches!(
            op,
            PlanOp::ScanLabel { .. }
                | PlanOp::ScanKey { .. }
                | PlanOp::IndexScan { .. }
                | PlanOp::IndexIntersect { .. }
                | PlanOp::LookupProps { .. }
                | PlanOp::Expand { .. }
                | PlanOp::Filter { .. }
                | PlanOp::Project { .. }
                | PlanOp::Limit(_)
        )
    })
    // At least one scan.
    && ops.iter().any(|op| {
        matches!(
            op,
            PlanOp::ScanLabel { .. }
                | PlanOp::ScanKey { .. }
                | PlanOp::IndexScan { .. }
                | PlanOp::IndexIntersect { .. }
        )
    })
    // Exactly one Project (ensures it is a RETURN query).
    && ops.iter().any(|op| matches!(op, PlanOp::Project { .. }))
    // At most one Expand: multi-hop chains are outside the documented subset.
    && ops
        .iter()
        .filter(|op| matches!(op, PlanOp::Expand { .. }))
        .count()
        <= 1
}

/// Compile `q` into a logical plan. Errors are contextual `String`s; never panics.
pub fn plan(q: &Query) -> Result<Vec<PlanOp>, String> {
    let mut bound = BTreeSet::new();
    let mut rel_bound = BTreeSet::new();
    let mut ops = Vec::new();
    let mut node_anon = 0u32;
    let mut rel_anon = 0u32;

    for pat in &q.matches {
        compile_pattern(
            pat,
            &mut ops,
            &mut bound,
            &mut rel_bound,
            &mut node_anon,
            &mut rel_anon,
        )?;
    }

    // OPTIONAL MATCH clauses (after required MATCHes).
    for oc in &q.optional_clauses {
        compile_optional_clause(
            oc,
            &mut ops,
            &mut bound,
            &mut rel_bound,
            &mut node_anon,
            &mut rel_anon,
        )?;
    }

    // Top-level UNWIND clauses.
    for uw in &q.unwinds {
        check_unwind_bound(&uw.list, &bound)?;
        bound.insert(uw.alias.clone());
        ops.push(PlanOp::Unwind {
            expr: uw.list.clone(),
            alias: uw.alias.clone(),
        });
    }

    if let Some(expr) = &q.where_expr {
        check_expr_bound(expr, &bound)?;
        ops.push(PlanOp::Filter { expr: expr.clone() });
    }
    // Fold WHERE single-equality predicates into IndexScan ops before any
    // aggregate or project ops are appended.
    ops = fold_where_equalities(ops);

    // Post-UNWIND WHERE: filter expanded rows using UNWIND alias bindings.
    if let Some(expr) = &q.post_unwind_where {
        check_expr_bound(expr, &bound)?;
        ops.push(PlanOp::Filter { expr: expr.clone() });
    }

    // WITH pipeline stages.
    for stage in &q.stages {
        compile_with_stage(
            stage,
            &mut ops,
            &mut bound,
            &mut rel_bound,
            &mut node_anon,
            &mut rel_anon,
        )?;
    }

    check_return_bound(&q.returns, &bound, &rel_bound)?;
    check_duplicate_aliases(&q.returns)?;
    check_duplicate_columns(&q.returns)?;
    if q.distinct
        && q.returns
            .iter()
            .any(|r| matches!(&r.value, RetVal::Agg { .. }))
    {
        return Err(
            "RETURN DISTINCT is not supported with aggregate functions; use grouping".to_string(),
        );
    }

    // Detect aggregate vs non-aggregate items in RETURN.
    // For pipeline plans (with stages or top-level UNWIND), single-aggregate
    // path is not used — route to GroupAggregate or Project.
    let is_pipeline = !q.stages.is_empty()
        || !q.unwinds.is_empty()
        || q.post_unwind_where.is_some()
        || !q.optional_clauses.is_empty();
    let agg_count = q
        .returns
        .iter()
        .filter(|r| matches!(&r.value, RetVal::Agg { .. }))
        .count();

    if agg_count == 1 && q.returns.len() == 1 && !is_pipeline {
        // Single-aggregate fast path: streaming O(1) accumulator, no grouping.
        let item = &q.returns[0];
        let (func, arg) = match &item.value {
            RetVal::Agg { func, arg } => (func.clone(), arg.clone()),
            _ => unreachable!(),
        };
        // Validate: SUM/AVG/MIN/MAX require a Prop arg, not Star.
        if let (AggFunc::Sum | AggFunc::Avg | AggFunc::Min | AggFunc::Max, AggArg::Star) =
            (&func, &arg)
        {
            return Err(format!(
                "{name} does not accept '*'; use a property expression like `{name}(n.prop)`",
                name = func_name(&func),
            ));
        }
        let column = item
            .alias
            .clone()
            .unwrap_or_else(|| agg_column_name(&func, &arg));
        ops.push(PlanOp::Aggregate { func, arg, column });
        // ORDER BY and LIMIT/SKIP are ignored for single-aggregate queries
        // (always returns exactly one row).
        return Ok(ops);
    }

    if agg_count > 0 {
        // GroupAggregate: handles grouped (mix of key items and aggregates) as
        // well as multi-aggregate-no-keys (all RETURN items are aggregates).
        let mut keys: Vec<(String, RetItem)> = Vec::new();
        let mut aggs: Vec<(AggFunc, AggArg, String)> = Vec::new();
        for item in &q.returns {
            match &item.value {
                RetVal::Agg { func, arg } => {
                    // Validate: SUM/AVG/MIN/MAX require a Prop arg, not Star.
                    if let (
                        AggFunc::Sum | AggFunc::Avg | AggFunc::Min | AggFunc::Max,
                        AggArg::Star,
                    ) = (func, arg)
                    {
                        return Err(format!(
                            "{name} does not accept '*'; use a property expression like `{name}(n.prop)`",
                            name = func_name(func),
                        ));
                    }
                    let column = item
                        .alias
                        .clone()
                        .unwrap_or_else(|| agg_column_name(func, arg));
                    aggs.push((func.clone(), arg.clone(), column));
                }
                _ => {
                    keys.push((column_name(item), item.clone()));
                }
            }
        }
        ops.push(PlanOp::GroupAggregate { keys, aggs });
        // ORDER BY + SKIP + LIMIT apply to the finished group result table.
        if !q.order_by.is_empty() {
            let mut items = Vec::with_capacity(q.order_by.len());
            for item in &q.order_by {
                items.push(rewrite_order_item(item, &q.returns, &bound, &rel_bound)?);
            }
            ops.push(PlanOp::OrderBy { items });
        }
        if let Some(ls) = &q.skip {
            ops.push(PlanOp::Skip(ls.clone()));
        }
        if let Some(ls) = &q.limit {
            ops.push(PlanOp::Limit(ls.clone()));
        }
        return Ok(ops);
    }

    ops.push(PlanOp::Project {
        items: q.returns.clone(),
    });
    if q.distinct {
        ops.push(PlanOp::Distinct);
    }

    if !q.order_by.is_empty() {
        let mut items = Vec::with_capacity(q.order_by.len());
        for item in &q.order_by {
            items.push(rewrite_order_item(item, &q.returns, &bound, &rel_bound)?);
        }
        ops.push(PlanOp::OrderBy { items });
    }

    if let Some(ls) = &q.skip {
        ops.push(PlanOp::Skip(ls.clone()));
    }
    if let Some(ls) = &q.limit {
        ops.push(PlanOp::Limit(ls.clone()));
    }

    Ok(ops)
}

/// Compile one WITH pipeline stage.
fn compile_with_stage(
    stage: &WithStage,
    ops: &mut Vec<PlanOp>,
    bound: &mut BTreeSet<String>,
    rel_bound: &mut BTreeSet<String>,
    node_anon: &mut u32,
    rel_anon: &mut u32,
) -> Result<(), String> {
    let agg_count = stage
        .items
        .iter()
        .filter(|r| matches!(&r.value, RetVal::Agg { .. }))
        .count();

    if agg_count > 0 {
        // Aggregate WITH → compile GroupAggregate + optional Filter/OrderBy/Skip/Limit.
        let mut keys: Vec<(String, RetItem)> = Vec::new();
        let mut aggs: Vec<(AggFunc, AggArg, String)> = Vec::new();
        for item in &stage.items {
            match &item.value {
                RetVal::Agg { func, arg } => {
                    if let (
                        AggFunc::Sum | AggFunc::Avg | AggFunc::Min | AggFunc::Max,
                        AggArg::Star,
                    ) = (func, arg)
                    {
                        return Err(format!(
                            "{name} does not accept '*'; use a property expression like `{name}(n.prop)`",
                            name = func_name(func),
                        ));
                    }
                    let col = item
                        .alias
                        .clone()
                        .unwrap_or_else(|| agg_column_name(func, arg));
                    aggs.push((func.clone(), arg.clone(), col));
                }
                _ => {
                    keys.push((column_name(item), item.clone()));
                }
            }
        }
        ops.push(PlanOp::GroupAggregate {
            keys: keys.clone(),
            aggs: aggs.clone(),
        });

        // Update bound to reflect only what GroupAggregate outputs.
        bound.clear();
        rel_bound.clear();
        for (col, _) in &keys {
            bound.insert(col.clone());
        }
        for (_, _, col) in &aggs {
            bound.insert(col.clone());
        }

        // Optional HAVING filter (WHERE after WITH with aggregates).
        if let Some(expr) = &stage.where_expr {
            check_expr_bound(expr, bound)?;
            ops.push(PlanOp::Filter { expr: expr.clone() });
        }
        // ORDER BY on the group result rows — validate targets against GroupAggregate output.
        // `bound` was updated above (lines 435–442) to hold only group output columns.
        if !stage.order_by.is_empty() {
            for item in &stage.order_by {
                match &item.target {
                    OrderTarget::Prop { var, .. } | OrderTarget::Var(var) => {
                        require_bound(var, bound, "ORDER BY in aggregate WITH")?;
                    }
                    OrderTarget::Alias(name) => {
                        require_bound(name, bound, "ORDER BY in aggregate WITH")?;
                    }
                }
            }
            ops.push(PlanOp::OrderBy {
                items: stage.order_by.clone(),
            });
        }
        if let Some(ls) = &stage.skip {
            ops.push(PlanOp::Skip(ls.clone()));
        }
        if let Some(ls) = &stage.limit {
            ops.push(PlanOp::Limit(ls.clone()));
        }
    } else {
        // Non-aggregate WITH → validate items and emit PlanOp::With.
        check_return_bound(&stage.items, bound, rel_bound)?;

        // `WITH … WHERE …` filters *after* the projection — the executor
        // projects each row through the WITH items and only then runs the
        // filter, exactly so that `WITH t, t.age AS age WHERE age > 30`
        // works. The pre-flight check has to scope the same way: against the
        // variables that survive the WITH as well as those already in scope.
        // Checking the pre-WITH set alone rejected every alias a
        // non-aggregate WITH introduced (`unbound variable `age` in WHERE`)
        // even though the plan it would have produced ran correctly, while
        // the aggregate branch above has always scoped its HAVING clause to
        // the group's output columns.
        let with_col_names: BTreeSet<String> = stage.items.iter().map(column_name).collect();
        let with_scope: BTreeSet<String> = bound.union(&with_col_names).cloned().collect();
        if let Some(expr) = &stage.where_expr {
            check_expr_bound(expr, &with_scope)?;
        }
        // ORDER BY items reference either var names or prop paths — no rewrite needed
        // here; exec_order_by_rows handles raw row ordering.
        // ORDER BY may reference the WITH output columns (aliases) in addition to
        // variables already in scope before the WITH.
        for item in &stage.order_by {
            match &item.target {
                OrderTarget::Prop { var, .. } | OrderTarget::Var(var) => {
                    if !bound.contains(var.as_str()) && !with_col_names.contains(var.as_str()) {
                        return Err(format!("unbound variable `{var}` in ORDER BY in WITH"));
                    }
                }
                OrderTarget::Alias(name) => {
                    if !bound.contains(name.as_str()) && !with_col_names.contains(name.as_str()) {
                        return Err(format!("unbound variable `{name}` in ORDER BY in WITH"));
                    }
                }
            }
        }
        ops.push(PlanOp::With {
            items: stage.items.clone(),
            where_expr: stage.where_expr.clone(),
            order_by: stage.order_by.clone(),
            skip: stage.skip.clone(),
            limit: stage.limit.clone(),
        });

        // Update bound: after non-aggregate WITH, only the WITH items survive.
        let mut new_bound: BTreeSet<String> = BTreeSet::new();
        let mut new_rel_bound: BTreeSet<String> = BTreeSet::new();
        for item in &stage.items {
            let col = column_name(item);
            new_bound.insert(col.clone());
            // Preserve rel-bound status for relationship variables carried through.
            match &item.value {
                RetVal::Var(v) if rel_bound.contains(v.as_str()) => {
                    new_rel_bound.insert(col);
                }
                _ => {}
            }
        }
        *bound = new_bound;
        *rel_bound = new_rel_bound;
    }

    // MATCH clauses that follow this WITH.
    for pat in &stage.matches {
        compile_pattern(pat, ops, bound, rel_bound, node_anon, rel_anon)?;
    }
    // OPTIONAL MATCH clauses that follow those MATCHes.
    for oc in &stage.optional_clauses {
        compile_optional_clause(oc, ops, bound, rel_bound, node_anon, rel_anon)?;
    }
    // UNWIND clauses that follow this WITH.
    for uw in &stage.unwinds {
        check_unwind_bound(&uw.list, bound)?;
        bound.insert(uw.alias.clone());
        ops.push(PlanOp::Unwind {
            expr: uw.list.clone(),
            alias: uw.alias.clone(),
        });
    }
    // WHERE that follows those MATCHes.
    if let Some(expr) = &stage.post_where {
        check_expr_bound(expr, bound)?;
        ops.push(PlanOp::Filter { expr: expr.clone() });
    }

    Ok(())
}

fn id_lookup(props: &[(String, Operand)]) -> Option<&Operand> {
    if props.len() == 1 && props[0].0 == "id" {
        Some(&props[0].1)
    } else {
        None
    }
}

/// A single equality on a non-`id` field with a literal or `$param` value —
/// the shape eligible for an `IndexScan`. Returns `(field, value)`.
fn index_lookup(props: &[(String, Operand)]) -> Option<(&str, &Operand)> {
    if props.len() == 1
        && props[0].0 != "id"
        && matches!(props[0].1, Operand::Lit(_) | Operand::Param(_))
    {
        Some((props[0].0.as_str(), &props[0].1))
    } else {
        None
    }
}

/// Two or more non-`id` equalities all with literal or `$param` values —
/// the shape eligible for `IndexIntersect`. Returns the full equality list
/// when ALL props qualify (no `id` field, all Lit|Param operands, len ≥ 2).
fn multi_index_lookup(props: &[(String, Operand)]) -> Option<Vec<(String, Operand)>> {
    if props.len() < 2 {
        return None;
    }
    if props
        .iter()
        .any(|(f, v)| f == "id" || !matches!(v, Operand::Lit(_) | Operand::Param(_)))
    {
        return None;
    }
    Some(props.to_vec())
}

/// Split an `Expr::And` chain into a flat list of sub-expressions. Used by the
/// WHERE-equality fold pass (T1) and compound-equality folding (T2).
pub(super) fn split_and(expr: Expr) -> Vec<Expr> {
    match expr {
        Expr::And(l, r) => {
            let mut v = split_and(*l);
            v.extend(split_and(*r));
            v
        }
        other => vec![other],
    }
}

/// Reassemble a flat list of expressions into an `Expr::And` chain.
/// Returns `None` when the list is empty (the caller must drop the Filter op).
pub(super) fn join_and(mut exprs: Vec<Expr>) -> Option<Expr> {
    if exprs.is_empty() {
        return None;
    }
    let mut result = exprs.remove(0);
    for e in exprs {
        result = Expr::And(Box::new(result), Box::new(e));
    }
    Some(result)
}

/// Post-pass: fold WHERE-clause single-equality predicates into `IndexScan` ops.
///
/// Eligibility (conservative):
/// - The anchoring scan must be `ScanLabel` — `ScanKey` is already O(1);
///   `IndexScan` already holds one equality (T2 handles compound).
/// - No `Expand` op may appear between the `ScanLabel` and the `Filter` being
///   folded: the conservative rule avoids cross-expand pushdown for now.
/// - The Filter must contain at least one `Cmp{Prop{var==scan_var,field}, Eq,
///   Lit|Param}` term. Only the first such term is taken; the rest stay as a
///   residual `Filter` for T2 / the executor to handle.
///
/// Folding is always correct regardless of whether the field is indexed:
/// the `IndexScan` executor arm (exec.rs:1379-1386) falls back to a full scan
/// + single-equality retain when `nodes_with_prop` returns `None`.
///
/// **Semantic note:** folding changes the set of *candidate* nodes that reach
/// residual predicates — only nodes matching the equality are visited, not every
/// node in the label. For valid data this produces identical result rows.
/// However, if a node's residual property would cause a type error (e.g. calling
/// a string function on an Int) that node must also match the folded equality to
/// trigger the error; nodes eliminated by the `IndexScan` will not surface it.
/// This is consistent with predicate pushdown in all standard query engines.
pub(super) fn fold_where_equalities(mut ops: Vec<PlanOp>) -> Vec<PlanOp> {
    // Locate the anchoring scan op: ScanLabel (most common), or IndexScan produced
    // by the inline-prop path (merge with WHERE equalities → IndexIntersect).
    // IndexIntersect anchors are not re-folded here; T2's exec handles them.
    let Some(scan_pos) = ops
        .iter()
        .position(|op| matches!(op, PlanOp::ScanLabel { .. } | PlanOp::IndexScan { .. }))
    else {
        return ops;
    };

    // Extract the scan variable, label, and any equality already committed by an
    // inline-prop IndexScan (used when merging inline+WHERE into IndexIntersect).
    let (scan_var, scan_label, existing_eq) = match &ops[scan_pos] {
        PlanOp::ScanLabel { var, label } => (var.clone(), label.clone(), None),
        PlanOp::IndexScan {
            var,
            label,
            field,
            value,
        } => (
            var.clone(),
            label.clone(),
            Some((field.clone(), value.clone())),
        ),
        _ => unreachable!(),
    };

    // Find the first Filter after the anchoring scan op.
    let Some(rel_pos) = ops[scan_pos + 1..]
        .iter()
        .position(|op| matches!(op, PlanOp::Filter { .. }))
    else {
        return ops;
    };
    let filter_pos = scan_pos + 1 + rel_pos;

    // Conservative: do not fold if any Expand lies between the anchor and the Filter.
    if ops[scan_pos + 1..filter_pos]
        .iter()
        .any(|op| matches!(op, PlanOp::Expand { .. }))
    {
        return ops;
    }

    let filter_expr = match &ops[filter_pos] {
        PlanOp::Filter { expr } => expr.clone(),
        _ => unreachable!(),
    };

    // Flatten the AND chain. Identity equality (`n.key` / `n.id` / `key(n)` /
    // `id(n)` vs lit/param) is a tagged IndexScan (`IdentityEq`): ScanKey plus
    // stored-wins. Other equalities stay residual so they are not mixed into
    // IndexIntersect (which would lose the point lookup).
    let mut terms = split_and(filter_expr);
    if let Some((idx, form, value)) = find_identity_eq(&terms, &scan_var) {
        terms.remove(idx);
        if let Some((f, v)) = existing_eq {
            terms.push(Expr::Cmp {
                lhs: Operand::Prop {
                    var: scan_var.clone(),
                    field: f,
                },
                op: CmpOp::Eq,
                rhs: v,
            });
        }
        ops[scan_pos] = match form {
            // Stored-wins: ScanKey the value, keep it only if stored identity
            // agrees, then union nodes whose stored `field` is the value.
            IdentityEqForm::Prop(field) => PlanOp::IndexScan {
                var: scan_var,
                label: scan_label,
                field,
                value,
            },
            // `key(n)` / `id(n)` is the id-map key and nothing else, which is
            // exactly `ScanKey` — it already filters label and visibility.
            IdentityEqForm::Func => PlanOp::ScanKey {
                var: scan_var,
                key: value,
                label: scan_label,
            },
        };
        match join_and(terms) {
            Some(residual) => ops[filter_pos] = PlanOp::Filter { expr: residual },
            None => {
                ops.remove(filter_pos);
            }
        }
        return ops;
    }

    // Flatten the AND chain and collect ALL eligible equalities on the scan var.
    let mut extracted: Vec<(String, Operand)> = Vec::new();
    let mut i = 0;
    while i < terms.len() {
        if matches!(
            &terms[i],
            Expr::Cmp {
                lhs: Operand::Prop { var, .. },
                op: CmpOp::Eq,
                rhs: Operand::Lit(_) | Operand::Param(_),
            } if var == &scan_var
        ) {
            let term = terms.remove(i);
            match term {
                Expr::Cmp {
                    lhs: Operand::Prop { field, .. },
                    rhs,
                    ..
                } => extracted.push((field, rhs)),
                _ => unreachable!(),
            }
        } else {
            i += 1;
        }
    }

    if extracted.is_empty() {
        return ops;
    }

    // Merge any pre-existing inline equality (from IndexScan) with the WHERE equalities.
    let mut all_equalities: Vec<(String, Operand)> = Vec::new();
    if let Some(eq) = existing_eq {
        all_equalities.push(eq);
    }
    all_equalities.extend(extracted);

    // Promote the anchor op.
    ops[scan_pos] = if all_equalities.len() == 1 {
        let (field, value) = all_equalities.remove(0);
        PlanOp::IndexScan {
            var: scan_var,
            label: scan_label,
            field,
            value,
        }
    } else {
        PlanOp::IndexIntersect {
            var: scan_var,
            label: scan_label,
            equalities: all_equalities,
        }
    };

    // Drop or narrow the Filter.
    match join_and(terms) {
        Some(residual) => ops[filter_pos] = PlanOp::Filter { expr: residual },
        None => {
            ops.remove(filter_pos);
        }
    }

    ops
}

/// `n.key` / `n.id` (and `key(n)` / `id(n)`) equality against a literal or
/// `$param`. `n.label` / `labels(n)` are not identity-eq (not a key).
fn is_identity_eq_field(field: &str) -> bool {
    field == "key" || field == "id"
}

fn identity_func_field(name: &str) -> Option<&'static str> {
    match name.to_ascii_lowercase().as_str() {
        "key" => Some("key"),
        "id" => Some("id"),
        _ => None,
    }
}

/// Which spelling of identity equality a term used.
///
/// The two are **not** interchangeable and must not share a plan op:
/// `n.key` / `n.id` are stored-wins (a stored property of that name beats the
/// id-map key), while `key(n)` / `id(n)` are always the id-map key. Folding the
/// function form into the stored-wins `IndexScan` tag makes a node whose stored
/// `key` property is `K` match `key(n) = 'K'`, and hides a node whose stored
/// `id` differs from its key from `id(n) = <its key>`.
enum IdentityEqForm {
    /// `n.key` / `n.id` — tagged `IndexScan`: ScanKey plus stored-wins union.
    Prop(String),
    /// `key(n)` / `id(n)` — a plain `ScanKey` point lookup, nothing else.
    Func,
}

fn identity_eq_term(term: &Expr, scan_var: &str) -> Option<(IdentityEqForm, Operand)> {
    let Expr::Cmp {
        lhs,
        op: CmpOp::Eq,
        rhs,
    } = term
    else {
        return None;
    };
    if !matches!(rhs, Operand::Lit(_) | Operand::Param(_)) {
        return None;
    }
    match lhs {
        Operand::Prop { var, field } if var == scan_var && is_identity_eq_field(field) => {
            Some((IdentityEqForm::Prop(field.clone()), rhs.clone()))
        }
        Operand::FuncCall { name, args } if args.len() == 1 => {
            let Operand::Var(v) = &args[0] else {
                return None;
            };
            if v != scan_var {
                return None;
            }
            identity_func_field(name)?;
            Some((IdentityEqForm::Func, rhs.clone()))
        }
        _ => None,
    }
}

fn find_identity_eq(terms: &[Expr], scan_var: &str) -> Option<(usize, IdentityEqForm, Operand)> {
    for (i, term) in terms.iter().enumerate() {
        if let Some((form, value)) = identity_eq_term(term, scan_var) {
            return Some((i, form, value));
        }
    }
    None
}

fn invert_dir(d: RelDir) -> RelDir {
    match d {
        RelDir::Right => RelDir::Left,
        RelDir::Left => RelDir::Right,
        RelDir::Undirected => RelDir::Undirected,
    }
}

fn compile_pattern(
    pat: &Pattern,
    ops: &mut Vec<PlanOp>,
    bound: &mut BTreeSet<String>,
    rel_bound: &mut BTreeSet<String>,
    node_anon: &mut u32,
    rel_anon: &mut u32,
) -> Result<(), String> {
    let start = name_node(&pat.start, node_anon, bound);
    if pat.shortest {
        // shortestPath requires both endpoints already bound.
        if !bound.contains(&start) {
            return Err(format!(
                "shortestPath: source node `{start}` is not bound; \
                 bind both endpoints before shortestPath"
            ));
        }
        ops.push(PlanOp::JoinBound {
            var: start.clone(),
            label: pat.start.label.clone(),
            props: pat.start.props.clone(),
        });
    } else if bound.contains(&start) {
        ops.push(PlanOp::JoinBound {
            var: start.clone(),
            label: pat.start.label.clone(),
            props: pat.start.props.clone(),
        });
    } else if pat.chain.len() == 1
        && pat.chain[0].0.hops.is_none()
        && pat.chain[0]
            .1
            .var
            .as_ref()
            .is_some_and(|v| bound.contains(v))
    {
        // Expand-from-bound: leftmost unbound, rightmost dest already bound,
        // single-rel *fixed-hop* pattern. Start from dest, invert dir, expand
        // toward start. Variable-length (`*min..max`) is not reversed: VarExpand
        // has no dest label/prop filter, so reversing would drop start checks.
        let (rel, dest) = &pat.chain[0];
        let dest_name = name_node(dest, node_anon, bound);
        let rel_name = name_rel(rel, rel_anon, bound);
        bound.insert(rel_name.clone());
        rel_bound.insert(rel_name.clone());
        if dest.label.is_some() || !dest.props.is_empty() {
            ops.push(PlanOp::JoinBound {
                var: dest_name.clone(),
                label: dest.label.clone(),
                props: dest.props.clone(),
            });
        }
        ops.push(PlanOp::Expand {
            from: dest_name,
            rel_var: Some(rel_name),
            etypes: rel.etypes.clone(),
            dir: invert_dir(rel.dir),
            to: start.clone(),
            to_label: pat.start.label.clone(),
            to_props: pat.start.props.clone(),
        });
        bound.insert(start);
        return Ok(());
    } else if let Some(key) = id_lookup(&pat.start.props) {
        ops.push(PlanOp::ScanKey {
            var: start.clone(),
            key: key.clone(),
            label: pat.start.label.clone(),
        });
        bound.insert(start.clone());
    } else if let Some((field, value)) = index_lookup(&pat.start.props) {
        ops.push(PlanOp::IndexScan {
            var: start.clone(),
            label: pat.start.label.clone(),
            field: field.to_string(),
            value: value.clone(),
        });
        bound.insert(start.clone());
    } else if let Some(equalities) = multi_index_lookup(&pat.start.props) {
        ops.push(PlanOp::IndexIntersect {
            var: start.clone(),
            label: pat.start.label.clone(),
            equalities,
        });
        bound.insert(start.clone());
    } else {
        ops.push(PlanOp::ScanLabel {
            var: start.clone(),
            label: pat.start.label.clone(),
        });
        if !pat.start.props.is_empty() {
            ops.push(PlanOp::LookupProps {
                var: start.clone(),
                props: pat.start.props.clone(),
            });
        }
        bound.insert(start.clone());
    }

    let mut from = start;
    for (rel, dest) in &pat.chain {
        let rel_name = name_rel(rel, rel_anon, bound);
        bound.insert(rel_name.clone());
        rel_bound.insert(rel_name.clone());
        let to = name_node(dest, node_anon, bound);

        if let Some(hops) = rel.hops {
            if pat.shortest {
                // shortestPath: destination must also already be bound.
                if !bound.contains(&to) {
                    return Err(format!(
                        "shortestPath: destination node `{to}` is not bound; \
                         bind both endpoints before shortestPath"
                    ));
                }
                // A minimum hop count > 1 is not supported for shortestPath —
                // the BFS always returns the shortest (lowest-hop) path, so a
                // min constraint would silently be ignored.  Reject explicitly.
                if hops.min > 1 {
                    return Err(format!(
                        "shortestPath does not support a minimum hop count \
                         (got min={}); use a plain variable-length pattern \
                         if you need a minimum",
                        hops.min
                    ));
                }
                ops.push(PlanOp::ShortestPath {
                    from: from.clone(),
                    rel_var: Some(rel_name),
                    etypes: rel.etypes.clone(),
                    dir: rel.dir,
                    to: to.clone(),
                    max_hops: hops.max,
                });
            } else {
                ops.push(PlanOp::VarExpand {
                    from: from.clone(),
                    rel_var: Some(rel_name),
                    etypes: rel.etypes.clone(),
                    dir: rel.dir,
                    to: to.clone(),
                    min: hops.min,
                    max: hops.max,
                });
                bound.insert(to.clone());
            }
        } else {
            ops.push(PlanOp::Expand {
                from: from.clone(),
                rel_var: Some(rel_name),
                etypes: rel.etypes.clone(),
                dir: rel.dir,
                to: to.clone(),
                to_label: dest.label.clone(),
                to_props: dest.props.clone(),
            });
            bound.insert(to.clone());
        }
        from = to;
    }
    Ok(())
}

/// Compile one `OPTIONAL MATCH` clause into a `LeftOuterApply` op.
///
/// The inner plan is compiled from the pattern(s) and optional WHERE, starting
/// from a copy of the outer bound set.  Variables introduced inside the optional
/// scope are collected as `optional_vars` — they will be nulled in the fallback
/// row when the inner plan produces no results.
fn compile_optional_clause(
    oc: &OptionalClause,
    ops: &mut Vec<PlanOp>,
    bound: &mut BTreeSet<String>,
    rel_bound: &mut BTreeSet<String>,
    node_anon: &mut u32,
    rel_anon: &mut u32,
) -> Result<(), String> {
    // Clone the outer bound state; the inner plan compiles against it.
    let mut inner_bound = bound.clone();
    let mut inner_rel_bound = rel_bound.clone();
    let mut inner_ops: Vec<PlanOp> = Vec::new();

    for pat in &oc.patterns {
        compile_pattern(
            pat,
            &mut inner_ops,
            &mut inner_bound,
            &mut inner_rel_bound,
            node_anon,
            rel_anon,
        )?;
    }
    if let Some(expr) = &oc.where_expr {
        check_expr_bound(expr, &inner_bound)?;
        inner_ops.push(PlanOp::Filter { expr: expr.clone() });
    }

    // Variables newly introduced by the optional clause.
    let optional_vars: Vec<String> = inner_bound
        .difference(bound)
        .chain(inner_rel_bound.difference(rel_bound))
        .cloned()
        .collect();

    // Merge inner-introduced vars into the outer bound set so subsequent
    // clauses can reference them (they may be null, but they are "bound").
    for v in &optional_vars {
        bound.insert(v.clone());
    }
    for v in inner_rel_bound
        .difference(&*rel_bound)
        .cloned()
        .collect::<Vec<_>>()
    {
        rel_bound.insert(v);
    }

    ops.push(PlanOp::LeftOuterApply {
        inner: inner_ops,
        optional_vars,
    });
    Ok(())
}

fn name_node(node: &NodePat, counter: &mut u32, bound: &BTreeSet<String>) -> String {
    match &node.var {
        Some(v) => v.clone(),
        None => fresh("_n", counter, bound),
    }
}

fn name_rel(rel: &RelPat, counter: &mut u32, bound: &BTreeSet<String>) -> String {
    match &rel.var {
        Some(v) => v.clone(),
        None => fresh("_r", counter, bound),
    }
}

/// Stable `_nN` / `_rN` in encounter order. Skips names already bound so a
/// user var `_n0` does not collide with the next anonymous node.
fn fresh(prefix: &str, counter: &mut u32, bound: &BTreeSet<String>) -> String {
    for _ in 0..=u32::MAX {
        let name = format!("{prefix}{counter}");
        *counter = counter.wrapping_add(1);
        if !bound.contains(&name) {
            return name;
        }
    }
    format!("{prefix}x")
}

fn check_expr_bound(expr: &Expr, bound: &BTreeSet<String>) -> Result<(), String> {
    match expr {
        Expr::And(lhs, rhs) | Expr::Or(lhs, rhs) => {
            check_expr_bound(lhs, bound)?;
            check_expr_bound(rhs, bound)
        }
        Expr::Not(inner) => check_expr_bound(inner, bound),
        Expr::Cmp { lhs, rhs, .. } => {
            check_operand_bound(lhs, bound, "WHERE")?;
            check_operand_bound(rhs, bound, "WHERE")
        }
        Expr::Truthy(op) => check_operand_bound(op, bound, "WHERE"),
        Expr::IsNull(op) | Expr::IsNotNull(op) => check_operand_bound(op, bound, "WHERE"),
        Expr::In { expr, list } => {
            check_operand_bound(expr, bound, "WHERE")?;
            for item in list {
                check_operand_bound(item, bound, "WHERE")?;
            }
            Ok(())
        }
    }
}

fn check_operand_bound(
    operand: &Operand,
    bound: &BTreeSet<String>,
    clause: &str,
) -> Result<(), String> {
    match operand {
        Operand::Prop { var, .. } => require_bound(var, bound, clause),
        Operand::Lit(_) | Operand::Param(_) => Ok(()),
        Operand::Var(name) => require_bound(name, bound, clause),
        Operand::BinArith { left, right, .. } => {
            check_operand_bound(left, bound, clause)?;
            check_operand_bound(right, bound, clause)
        }
        Operand::Index { base, index } => {
            check_operand_bound(base, bound, clause)?;
            check_operand_bound(index, bound, clause)
        }
        Operand::FuncCall { args, .. } => {
            for arg in args {
                check_operand_bound(arg, bound, clause)?;
            }
            Ok(())
        }
        Operand::Case { branches, default } => {
            for (cond, value) in branches {
                check_expr_bound(cond, bound)?;
                check_operand_bound(value, bound, clause)?;
            }
            if let Some(d) = default {
                check_operand_bound(d, bound, clause)?;
            }
            Ok(())
        }
    }
}

/// Validate that any variable referenced in an UNWIND expression is already bound.
fn check_unwind_bound(expr: &UnwindExpr, bound: &BTreeSet<String>) -> Result<(), String> {
    match expr {
        UnwindExpr::Lit(_) => Ok(()),
        UnwindExpr::Prop { var, .. } => require_bound(var, bound, "UNWIND"),
        UnwindExpr::Var(name) => require_bound(name, bound, "UNWIND"),
    }
}

fn require_bound(var: &str, bound: &BTreeSet<String>, clause: &str) -> Result<(), String> {
    if bound.contains(var) {
        Ok(())
    } else {
        Err(format!("unbound variable `{var}` in {clause}"))
    }
}

fn reject_bare_rel(var: &str, rel_bound: &BTreeSet<String>) -> Result<(), String> {
    if rel_bound.contains(var) {
        Err(format!(
            "cannot return relationship variable '{var}' bare; return its properties ({var}.field) instead"
        ))
    } else {
        Ok(())
    }
}

fn check_return_bound(
    items: &[RetItem],
    bound: &BTreeSet<String>,
    rel_bound: &BTreeSet<String>,
) -> Result<(), String> {
    for item in items {
        match &item.value {
            RetVal::Var(v) => {
                require_bound(v, bound, "RETURN")?;
                reject_bare_rel(v, rel_bound)?;
            }
            RetVal::Prop { var, .. } => {
                require_bound(var, bound, "RETURN")?;
            }
            RetVal::Agg { arg, .. } => check_agg_arg_bound(arg, bound)?,
            RetVal::FuncCall { args, .. } => {
                for arg in args {
                    check_operand_bound(arg, bound, "RETURN")?;
                }
            }
            RetVal::ScalarExpr(op) => {
                check_operand_bound(op, bound, "RETURN")?;
            }
        }
    }
    Ok(())
}

fn check_duplicate_aliases(items: &[RetItem]) -> Result<(), String> {
    let mut seen = BTreeSet::new();
    for item in items {
        if let Some(alias) = &item.alias {
            if !seen.insert(alias.clone()) {
                return Err(format!("duplicate RETURN alias `{alias}`"));
            }
        }
    }
    Ok(())
}

fn check_duplicate_columns(items: &[RetItem]) -> Result<(), String> {
    let mut seen = BTreeSet::new();
    for item in items {
        let col = column_name(item);
        if !seen.insert(col.clone()) {
            return Err(format!("duplicate RETURN column `{col}`"));
        }
    }
    Ok(())
}

/// Projected column name: alias if given, else the bare var, else `var.field`,
/// else the canonical aggregate call string, else `funcname(...)`, else `<expr>`.
fn column_name(item: &RetItem) -> String {
    if let Some(alias) = &item.alias {
        return alias.clone();
    }
    ret_val_label(&item.value).unwrap_or_else(|| match &item.value {
        RetVal::Agg { func, arg } => agg_column_name(func, arg),
        _ => unreachable!("ret_val_label names every non-aggregate item"),
    })
}

/// Every variable an aggregate argument reads must be bound, through any
/// `DISTINCT` wrapper.
fn check_agg_arg_bound(arg: &AggArg, bound: &BTreeSet<String>) -> Result<(), String> {
    match arg {
        AggArg::Star => Ok(()),
        AggArg::Var(v) => require_bound(v, bound, "RETURN"),
        AggArg::Prop { var, .. } => require_bound(var, bound, "RETURN"),
        AggArg::Distinct(inner) => check_agg_arg_bound(inner, bound),
    }
}

/// Canonical string for an aggregate without an alias, e.g. `COUNT(*)`,
/// `SUM(n.age)`, `COUNT(DISTINCT t)`.
fn agg_column_name(func: &AggFunc, arg: &AggArg) -> String {
    let f = func_name(func);
    format!("{f}({})", agg_arg_name(arg))
}

fn agg_arg_name(arg: &AggArg) -> String {
    match arg {
        AggArg::Star => "*".to_string(),
        AggArg::Var(v) => v.clone(),
        AggArg::Prop { var, field } => format!("{var}.{field}"),
        AggArg::Distinct(inner) => format!("DISTINCT {}", agg_arg_name(inner)),
    }
}

fn func_name(func: &AggFunc) -> &'static str {
    match func {
        AggFunc::Count => "COUNT",
        AggFunc::Sum => "SUM",
        AggFunc::Avg => "AVG",
        AggFunc::Min => "MIN",
        AggFunc::Max => "MAX",
        AggFunc::Collect => "COLLECT",
    }
}

fn rewrite_order_item(
    item: &OrderItem,
    returns: &[RetItem],
    bound: &BTreeSet<String>,
    rel_bound: &BTreeSet<String>,
) -> Result<OrderItem, String> {
    let column = match &item.target {
        OrderTarget::Alias(name) => {
            if returns
                .iter()
                .any(|r| r.alias.as_deref() == Some(name.as_str()))
            {
                name.clone()
            } else {
                return Err(format!("ORDER BY target `{name}` is not present in RETURN"));
            }
        }
        OrderTarget::Var(v) => {
            require_bound(v, bound, "ORDER BY")?;
            reject_bare_rel(v, rel_bound)?;
            match returns
                .iter()
                .find(|r| matches!(&r.value, RetVal::Var(x) if x == v))
            {
                Some(r) => column_name(r),
                None => {
                    return Err(format!("ORDER BY target `{v}` is not present in RETURN"));
                }
            }
        }
        OrderTarget::Prop { var, field } => {
            require_bound(var, bound, "ORDER BY")?;
            match returns.iter().find(
                |r| matches!(&r.value, RetVal::Prop { var: v, field: f } if v == var && f == field),
            ) {
                Some(r) => column_name(r),
                None => {
                    return Err(format!(
                        "ORDER BY target `{var}.{field}` is not present in RETURN"
                    ));
                }
            }
        }
    };
    Ok(OrderItem {
        target: OrderTarget::Alias(column),
        descending: item.descending,
    })
}

#[cfg(test)]
mod tests {
    use super::{plan, PlanOp};
    use crate::cypher::ast::{Expr, LimitSkip, Operand, OrderItem, OrderTarget, RetItem, RetVal};
    use crate::cypher::{lex, parse, RelDir};
    use crate::filter::CmpOp;
    use core_storage::Value;

    fn plan_src(src: &str) -> Result<Vec<PlanOp>, String> {
        plan(&parse(&lex(src)?)?)
    }

    fn assert_plan_err(src: &str, needle: &str) -> String {
        let result = std::panic::catch_unwind(|| plan_src(src));
        assert!(result.is_ok(), "plan({src:?}) panicked");
        let err = result
            .unwrap()
            .expect_err(&format!("plan({src:?}) must be Err"));
        assert!(
            err.contains(needle),
            "error must mention {needle:?}, got: {err}"
        );
        err
    }

    /// Dogfood query from T6. Shape:
    /// - MATCH 1: `t` unbound with `{id: $tid}` → `ScanKey`.
    /// - MATCH 2: `c` unbound, dest `t` already bound, single-rel → reverse:
    ///   Expand from `t` dir Left (inbound) to `c` (Company label on `to`).
    /// - MATCH 3: start `c` already bound → `JoinBound`; expand to bound `t`.
    #[test]
    fn dogfood_query_exact_plan() {
        let src = "\
MATCH (t:Talent {id: $tid}) \
MATCH (c:Company)-[i:INDUSTRY_ALIGNMENT]->(t) \
MATCH (c)-[s:SPECIALTY_MATCH]->(t) \
WHERE i.score >= 0.5 AND s.score >= 0.5 \
RETURN c, i.score AS industry, s.score AS specialty \
ORDER BY industry DESC, specialty DESC \
LIMIT 10";
        let got = plan_src(src).expect("dogfood query must plan");
        let expected = vec![
            PlanOp::ScanKey {
                var: "t".into(),
                key: Operand::Param("tid".into()),
                label: Some("Talent".into()),
            },
            PlanOp::Expand {
                from: "t".into(),
                rel_var: Some("i".into()),
                etypes: vec!["INDUSTRY_ALIGNMENT".into()],
                dir: RelDir::Left,
                to: "c".into(),
                to_label: Some("Company".into()),
                to_props: vec![],
            },
            PlanOp::JoinBound {
                var: "c".into(),
                label: None,
                props: vec![],
            },
            PlanOp::Expand {
                from: "c".into(),
                rel_var: Some("s".into()),
                etypes: vec!["SPECIALTY_MATCH".into()],
                dir: RelDir::Right,
                to: "t".into(),
                to_label: None,
                to_props: vec![],
            },
            PlanOp::Filter {
                expr: Expr::And(
                    Box::new(Expr::Cmp {
                        lhs: Operand::Prop {
                            var: "i".into(),
                            field: "score".into(),
                        },
                        op: CmpOp::Ge,
                        rhs: Operand::Lit(Value::Float(0.5)),
                    }),
                    Box::new(Expr::Cmp {
                        lhs: Operand::Prop {
                            var: "s".into(),
                            field: "score".into(),
                        },
                        op: CmpOp::Ge,
                        rhs: Operand::Lit(Value::Float(0.5)),
                    }),
                ),
            },
            PlanOp::Project {
                items: vec![
                    RetItem {
                        value: RetVal::Var("c".into()),
                        alias: None,
                    },
                    RetItem {
                        value: RetVal::Prop {
                            var: "i".into(),
                            field: "score".into(),
                        },
                        alias: Some("industry".into()),
                    },
                    RetItem {
                        value: RetVal::Prop {
                            var: "s".into(),
                            field: "score".into(),
                        },
                        alias: Some("specialty".into()),
                    },
                ],
            },
            PlanOp::OrderBy {
                items: vec![
                    OrderItem {
                        target: OrderTarget::Alias("industry".into()),
                        descending: true,
                    },
                    OrderItem {
                        target: OrderTarget::Alias("specialty".into()),
                        descending: true,
                    },
                ],
            },
            PlanOp::Limit(LimitSkip::Exact(10)),
        ];
        assert_eq!(got, expected);
    }

    /// Anonymous names increment in encounter order across the whole query.
    /// MATCH 1: start `_n0`, rel `_r0`, dest `a`.
    /// MATCH 2: start `_n1` unbound, dest already-bound `a` → reverse Expand
    /// from `a` dir Left to `_n1`.
    #[test]
    fn anonymous_node_and_rel_names_are_stable() {
        let got = plan_src("MATCH ()-[]->(a) MATCH ()-[]->(a) RETURN a").unwrap();
        assert_eq!(
            got,
            vec![
                PlanOp::ScanLabel {
                    var: "_n0".into(),
                    label: None,
                },
                PlanOp::Expand {
                    from: "_n0".into(),
                    rel_var: Some("_r0".into()),
                    etypes: vec![],
                    dir: RelDir::Right,
                    to: "a".into(),
                    to_label: None,
                    to_props: vec![],
                },
                PlanOp::Expand {
                    from: "a".into(),
                    rel_var: Some("_r1".into()),
                    etypes: vec![],
                    dir: RelDir::Left,
                    to: "_n1".into(),
                    to_label: None,
                    to_props: vec![],
                },
                PlanOp::Project {
                    items: vec![RetItem {
                        value: RetVal::Var("a".into()),
                        alias: None,
                    }],
                },
            ]
        );
    }

    #[test]
    fn props_on_scan_node_emit_scan_then_lookup() {
        let got = plan_src("MATCH (t:Talent {id: $tid}) RETURN t").unwrap();
        assert_eq!(
            got,
            vec![
                PlanOp::ScanKey {
                    var: "t".into(),
                    key: Operand::Param("tid".into()),
                    label: Some("Talent".into()),
                },
                PlanOp::Project {
                    items: vec![RetItem {
                        value: RetVal::Var("t".into()),
                        alias: None,
                    }],
                },
            ]
        );
    }

    #[test]
    fn mixed_id_map_stays_scan_label_then_lookup() {
        let got = plan_src("MATCH (t:Talent {id: $k, name: 'x'}) RETURN t").unwrap();
        assert_eq!(
            got,
            vec![
                PlanOp::ScanLabel {
                    var: "t".into(),
                    label: Some("Talent".into()),
                },
                PlanOp::LookupProps {
                    var: "t".into(),
                    props: vec![
                        ("id".into(), Operand::Param("k".into())),
                        ("name".into(), Operand::Lit(Value::Str("x".into()))),
                    ],
                },
                PlanOp::Project {
                    items: vec![RetItem {
                        value: RetVal::Var("t".into()),
                        alias: None,
                    }],
                },
            ]
        );
    }

    #[test]
    fn plan_id_map_is_scan_key() {
        let toks = crate::cypher::lex("MATCH (n:Person {id: $k}) RETURN n").unwrap();
        let q = crate::cypher::parse(&toks).unwrap();
        let ops = plan(&q).unwrap();
        assert!(matches!(ops[0], PlanOp::ScanKey { .. }), "{ops:?}");
    }

    #[test]
    fn plan_expands_from_bound_key() {
        let cy =
            "MATCH (t:Talent {id: $tid}) MATCH (c:Company)-[i:INDUSTRY_ALIGNMENT]->(t) RETURN c";
        let ops = plan(&crate::cypher::parse(&crate::cypher::lex(cy).unwrap()).unwrap()).unwrap();
        // first: ScanKey t; then Expand from t, dir Left (inbound)
        assert!(matches!(&ops[0], PlanOp::ScanKey { var, .. } if var == "t"));
        match &ops[1] {
            PlanOp::Expand { from, dir, to, .. } => {
                assert_eq!(from, "t");
                assert_eq!(to, "c");
                assert_eq!(*dir, RelDir::Left);
            }
            other => panic!("{other:?}"),
        }
    }

    /// VarExpand has no dest label/prop filter. Reversing
    /// `MATCH (c:Company)-[*1..2]->(t)` would drop `:Company` and bind
    /// non-Company `c`. Keep LTR: ScanLabel Company then VarExpand.
    #[test]
    fn plan_does_not_reverse_variable_length_from_bound() {
        let cy = "MATCH (t {id: $tid}) MATCH (c:Company)-[*1..2]->(t) RETURN c";
        let ops = plan_src(cy).unwrap();
        assert!(
            matches!(&ops[0], PlanOp::ScanKey { var, .. } if var == "t"),
            "{ops:?}"
        );
        assert!(
            matches!(&ops[1], PlanOp::ScanLabel { var, label } if var == "c" && label.as_deref() == Some("Company")),
            "{ops:?}"
        );
        match &ops[2] {
            PlanOp::VarExpand {
                from,
                dir,
                to,
                min,
                max,
                ..
            } => {
                assert_eq!(from, "c");
                assert_eq!(to, "t");
                assert_eq!(*dir, RelDir::Right);
                assert_eq!(*min, 1);
                assert_eq!(*max, 2);
            }
            other => panic!("{other:?}"),
        }
    }

    #[test]
    fn unbound_var_in_where_is_err() {
        let err = assert_plan_err("MATCH (a) WHERE b.x = 1 RETURN a", "b");
        assert!(
            err.to_ascii_lowercase().contains("unbound")
                && err.to_ascii_lowercase().contains("where"),
            "expected unbound-in-WHERE context, got: {err}"
        );
    }

    #[test]
    fn unbound_var_in_return_is_err() {
        let err = assert_plan_err("MATCH (a) RETURN b", "b");
        assert!(
            err.to_ascii_lowercase().contains("unbound")
                && err.to_ascii_lowercase().contains("return"),
            "expected unbound-in-RETURN context, got: {err}"
        );
    }

    #[test]
    fn unbound_var_in_order_by_is_err() {
        let err = assert_plan_err("MATCH (a) RETURN a ORDER BY b", "b");
        assert!(
            err.to_ascii_lowercase().contains("unbound")
                && (err.to_ascii_lowercase().contains("order")),
            "expected unbound-in-ORDER context, got: {err}"
        );
    }

    #[test]
    fn duplicate_alias_is_err() {
        let err = assert_plan_err("MATCH (a) RETURN a AS x, a.id AS x", "x");
        assert!(
            err.to_ascii_lowercase().contains("duplicate")
                && err.to_ascii_lowercase().contains("alias"),
            "expected duplicate-alias context, got: {err}"
        );
    }

    #[test]
    fn duplicate_column_name_is_err() {
        let err = assert_plan_err("MATCH (a) RETURN a, a", "a");
        assert!(
            err.to_ascii_lowercase().contains("duplicate")
                && err.to_ascii_lowercase().contains("column"),
            "expected duplicate-column context, got: {err}"
        );
    }

    #[test]
    fn order_by_target_absent_from_return_is_err() {
        // `a` is bound, but `a.x` is not a RETURN item.
        let err = assert_plan_err("MATCH (a) RETURN a ORDER BY a.x", "a.x");
        assert!(
            err.to_ascii_lowercase().contains("return"),
            "expected ORDER BY target-not-in-RETURN context, got: {err}"
        );
    }

    /// Alias → that alias's column; bare var → its RETURN column (alias if
    /// given, else the var name); un-aliased prop → `var.field`.
    #[test]
    fn order_by_targets_rewrite_to_projected_column_names() {
        let got = plan_src(
            "MATCH (a)-[r]->(b) \
             RETURN a, a.name AS nm, b.age \
             ORDER BY nm DESC, a ASC, b.age",
        )
        .unwrap();
        let order = got
            .iter()
            .find_map(|op| match op {
                PlanOp::OrderBy { items } => Some(items),
                _ => None,
            })
            .expect("plan must contain OrderBy");
        assert_eq!(
            order,
            &vec![
                OrderItem {
                    target: OrderTarget::Alias("nm".into()),
                    descending: true,
                },
                OrderItem {
                    target: OrderTarget::Alias("a".into()),
                    descending: false,
                },
                OrderItem {
                    target: OrderTarget::Alias("b.age".into()),
                    descending: false,
                },
            ]
        );

        let aliased_var = plan_src("MATCH (a) RETURN a AS person ORDER BY a").unwrap();
        let order = aliased_var
            .iter()
            .find_map(|op| match op {
                PlanOp::OrderBy { items } => Some(items),
                _ => None,
            })
            .expect("plan must contain OrderBy");
        assert_eq!(
            order,
            &vec![OrderItem {
                target: OrderTarget::Alias("person".into()),
                descending: false,
            }]
        );
    }

    #[test]
    fn bound_pattern_start_is_join_bound_then_expand() {
        let got = plan_src("MATCH (a:L) MATCH (a)-[r:T]->(b) RETURN a, b").unwrap();
        assert_eq!(
            got,
            vec![
                PlanOp::ScanLabel {
                    var: "a".into(),
                    label: Some("L".into()),
                },
                PlanOp::JoinBound {
                    var: "a".into(),
                    label: None,
                    props: vec![],
                },
                PlanOp::Expand {
                    from: "a".into(),
                    rel_var: Some("r".into()),
                    etypes: vec!["T".into()],
                    dir: RelDir::Right,
                    to: "b".into(),
                    to_label: None,
                    to_props: vec![],
                },
                PlanOp::Project {
                    items: vec![
                        RetItem {
                            value: RetVal::Var("a".into()),
                            alias: None,
                        },
                        RetItem {
                            value: RetVal::Var("b".into()),
                            alias: None,
                        },
                    ],
                },
            ]
        );
    }

    #[test]
    fn bound_dest_extra_checks_ride_on_expand() {
        let got = plan_src("MATCH (t:Talent) MATCH (c)-[r]->(t:Talent {id: 1}) RETURN t").unwrap();
        assert_eq!(
            got,
            vec![
                PlanOp::ScanLabel {
                    var: "t".into(),
                    label: Some("Talent".into()),
                },
                PlanOp::JoinBound {
                    var: "t".into(),
                    label: Some("Talent".into()),
                    props: vec![("id".into(), Operand::Lit(Value::Int(1)))],
                },
                PlanOp::Expand {
                    from: "t".into(),
                    rel_var: Some("r".into()),
                    etypes: vec![],
                    dir: RelDir::Left,
                    to: "c".into(),
                    to_label: None,
                    to_props: vec![],
                },
                PlanOp::Project {
                    items: vec![RetItem {
                        value: RetVal::Var("t".into()),
                        alias: None,
                    }],
                },
            ]
        );
    }

    #[test]
    fn return_distinct_emits_distinct_after_project() {
        let ops = plan_src("MATCH (n) RETURN DISTINCT n").expect("DISTINCT must plan");
        let proj = ops
            .iter()
            .position(|op| matches!(op, PlanOp::Project { .. }))
            .expect("Project");
        assert!(
            matches!(ops.get(proj + 1), Some(PlanOp::Distinct)),
            "DISTINCT must follow Project, got: {ops:?}"
        );
        let bounded = plan_src("MATCH (n) RETURN DISTINCT n LIMIT 1").unwrap();
        assert!(
            super::row_bound(&bounded).is_none(),
            "DISTINCT + LIMIT must not push LIMIT into producers"
        );
    }

    #[test]
    fn skip_then_limit_follow_project() {
        let got = plan_src("MATCH (a) RETURN a SKIP 2 LIMIT 3").unwrap();
        assert_eq!(
            got,
            vec![
                PlanOp::ScanLabel {
                    var: "a".into(),
                    label: None,
                },
                PlanOp::Project {
                    items: vec![RetItem {
                        value: RetVal::Var("a".into()),
                        alias: None,
                    }],
                },
                PlanOp::Skip(LimitSkip::Exact(2)),
                PlanOp::Limit(LimitSkip::Exact(3)),
            ]
        );
    }

    #[test]
    fn aliased_prop_order_by_rewrites_to_alias_column() {
        let got = plan_src("MATCH (a) RETURN a.name AS nm ORDER BY a.name").unwrap();
        let order = got
            .iter()
            .find_map(|op| match op {
                PlanOp::OrderBy { items } => Some(items),
                _ => None,
            })
            .unwrap();
        assert_eq!(
            order,
            &vec![OrderItem {
                target: OrderTarget::Alias("nm".into()),
                descending: false,
            }]
        );
    }

    #[test]
    fn plan_never_panics_on_hand_built_query() {
        use crate::cypher::ast::{NodePat, Pattern, Query};
        let q = Query {
            matches: vec![],
            optional_clauses: vec![],
            where_expr: None,
            unwinds: vec![],
            post_unwind_where: None,
            stages: vec![],
            returns: vec![],
            order_by: vec![],
            distinct: false,
            skip: None,
            limit: None,
        };
        let result = std::panic::catch_unwind(|| plan(&q));
        assert!(result.is_ok(), "plan panicked on empty Query");
        let _ = result.unwrap();

        let q = Query {
            matches: vec![Pattern {
                start: NodePat {
                    var: None,
                    label: None,
                    props: vec![],
                },
                chain: vec![],
                shortest: false,
            }],
            optional_clauses: vec![],
            where_expr: Some(Expr::Not(Box::new(Expr::Cmp {
                lhs: Operand::Param("p".into()),
                op: CmpOp::Eq,
                rhs: Operand::Lit(Value::Int(1)),
            }))),
            unwinds: vec![],
            post_unwind_where: None,
            stages: vec![],
            returns: vec![],
            distinct: false,
            order_by: vec![OrderItem {
                target: OrderTarget::Alias("missing".into()),
                descending: true,
            }],
            skip: Some(LimitSkip::Exact(0)),
            limit: Some(LimitSkip::Exact(0)),
        };
        let result = std::panic::catch_unwind(|| plan(&q));
        assert!(result.is_ok(), "plan panicked on hand-built Query");
        let _ = result.unwrap();
    }

    #[test]
    fn bare_relationship_var_in_return_is_err() {
        let err = assert_plan_err("MATCH (a)-[r:T]->(b) RETURN r", "r");
        assert!(
            err.to_ascii_lowercase().contains("relationship"),
            "expected bare-rel RETURN guidance, got: {err}"
        );
    }

    #[test]
    fn relationship_prop_in_return_is_ok() {
        plan_src("MATCH (a)-[r:T]->(b) RETURN r.w").expect("rel prop RETURN must plan");
    }

    #[test]
    fn bare_relationship_var_in_order_by_is_err() {
        // RETURN r.w is legal; ORDER BY r is a bare rel var (defense, not just
        // "not in RETURN").
        let err = assert_plan_err("MATCH (a)-[r:T]->(b) RETURN r.w ORDER BY r", "r");
        assert!(
            err.to_ascii_lowercase().contains("relationship"),
            "expected bare-rel ORDER BY guidance, got: {err}"
        );
    }

    #[test]
    fn relationship_prop_in_order_by_is_ok() {
        plan_src("MATCH (a)-[r:T]->(b) RETURN r.w ORDER BY r.w")
            .expect("rel prop ORDER BY must plan");
    }

    // ── Variable-length path and shortestPath planner tests ───────────────────

    #[test]
    fn var_expand_op_emitted_for_star_rel() {
        use super::row_bound;
        let ops = plan_src("MATCH (a)-[r:T*2..4]->(b) RETURN b").unwrap();
        let has_var = ops
            .iter()
            .any(|op| matches!(op, PlanOp::VarExpand { min: 2, max: 4, .. }));
        assert!(has_var, "expected VarExpand(2..4) in plan, got: {ops:?}");
        // row_bound must be None even when no ORDER BY (VarExpand overrides pull routing)
        assert_eq!(
            row_bound(&ops),
            None,
            "VarExpand plan must not use pull path"
        );
    }

    #[test]
    fn var_expand_with_limit_still_takes_staged_path() {
        use super::row_bound;
        let ops = plan_src("MATCH (a)-[r:T*1..3]->(b) RETURN b LIMIT 5").unwrap();
        // Staged path: row_bound returns None for VarExpand.
        assert_eq!(
            row_bound(&ops),
            None,
            "VarExpand + LIMIT must still use staged path"
        );
        let has_var = ops.iter().any(|op| matches!(op, PlanOp::VarExpand { .. }));
        assert!(has_var, "plan must contain VarExpand");
        let has_limit = ops
            .iter()
            .any(|op| matches!(op, PlanOp::Limit(LimitSkip::Exact(5))));
        assert!(has_limit, "plan must still emit Limit op");
    }

    #[test]
    fn shortest_path_op_emitted_for_shortest_path_clause() {
        let ops =
            plan_src("MATCH (a:N) MATCH (b:N) MATCH shortestPath((a)-[r:T*..3]->(b)) RETURN a")
                .unwrap();
        let has_sp = ops
            .iter()
            .any(|op| matches!(op, PlanOp::ShortestPath { max_hops: 3, .. }));
        assert!(
            has_sp,
            "expected ShortestPath op with max_hops=3, got: {ops:?}"
        );
    }

    #[test]
    fn shortest_path_unbound_endpoint_is_err() {
        let err = assert_plan_err(
            "MATCH shortestPath((a)-[r:T*..3]->(b)) RETURN a",
            "shortestPath",
        );
        assert!(
            err.contains("not bound") || err.contains("bound"),
            "error must mention binding, got: {err}"
        );
    }

    #[test]
    fn var_expand_rel_var_is_in_rel_bound() {
        // r.length should be allowed in RETURN (prop access)
        plan_src("MATCH (a)-[r:T*1..3]->(b) RETURN r.length").expect("r.length must plan");
        // bare r must be rejected
        assert_plan_err("MATCH (a)-[r:T*1..3]->(b) RETURN r", "r");
    }

    #[test]
    fn shortest_path_min_gt_1_is_plan_err() {
        // shortestPath with *2..5 must be rejected: min>1 is not supported
        let err = assert_plan_err(
            "MATCH (a:N) MATCH (b:N) MATCH shortestPath((a)-[r:T*2..5]->(b)) RETURN r.length",
            "shortestPath",
        );
        assert!(
            err.contains("minimum"),
            "error must mention minimum hop count, got: {err}"
        );
    }

    // ── is_subscribable tests ──────────────────────────────────────────────────

    fn subscribable(src: &str) -> bool {
        let ops = plan_src(src).expect("must plan");
        super::is_subscribable(&ops)
    }

    #[test]
    fn is_subscribable_passes_simple_label_scan() {
        assert!(subscribable("MATCH (n:Person) RETURN n"));
        assert!(subscribable("MATCH (n:Person) WHERE n.age > 18 RETURN n"));
        assert!(subscribable("MATCH (n:Person) RETURN n LIMIT 100"));
    }

    #[test]
    fn is_subscribable_passes_single_hop_expand() {
        assert!(subscribable(
            "MATCH (a:Person)-[r:KNOWS]->(b:Person) RETURN a"
        ));
        assert!(subscribable(
            "MATCH (a:Person)-[r:KNOWS]->(b:Person) RETURN a LIMIT 50"
        ));
    }

    #[test]
    fn is_subscribable_rejects_multi_hop_expand() {
        // Two Expand ops: outside the documented single-hop subset.
        assert!(
            !subscribable("MATCH (a:Person)-[r1:KNOWS]->(b:Person)-[r2:LIKES]->(c:Thing) RETURN a"),
            "two-hop chain must be rejected"
        );
    }

    #[test]
    fn is_subscribable_rejects_skip() {
        // SKIP creates unstable offset windows — explicitly excluded.
        assert!(
            !subscribable("MATCH (n:Person) RETURN n SKIP 10 LIMIT 50"),
            "SKIP must be rejected"
        );
        assert!(
            !subscribable("MATCH (n:Person) RETURN n SKIP 10"),
            "bare SKIP must be rejected"
        );
    }

    #[test]
    fn is_subscribable_rejects_order_by() {
        assert!(!subscribable("MATCH (n:Person) RETURN n ORDER BY n"));
    }

    #[test]
    fn is_subscribable_rejects_aggregates() {
        assert!(!subscribable("MATCH (n:Person) RETURN COUNT(*)"));
    }

    #[test]
    fn is_subscribable_rejects_var_expand() {
        assert!(!subscribable(
            "MATCH (a:Person)-[r:KNOWS*1..3]->(b) RETURN b"
        ));
    }

    // --- WHERE equality fold tests (T1) ---

    #[test]
    fn where_equality_folds_to_index_scan() {
        let ops = plan_src("MATCH (n:Person) WHERE n.city = 'austin' RETURN n.key").unwrap();
        assert!(
            matches!(&ops[0], PlanOp::IndexScan { field, .. } if field == "city"),
            "WHERE single equality must fold to IndexScan, got {:?}",
            ops[0]
        );
        assert!(
            !ops.iter().any(|op| matches!(op, PlanOp::Filter { .. })),
            "consumed predicate must not remain as Filter"
        );
    }

    #[test]
    fn where_equality_param_folds_to_index_scan() {
        let ops = plan_src("MATCH (n:Person) WHERE n.city = $c RETURN n.key").unwrap();
        assert!(
            matches!(&ops[0], PlanOp::IndexScan { .. }),
            "param WHERE equality must fold to IndexScan, got {:?}",
            ops[0]
        );
    }

    #[test]
    fn where_and_keeps_residual_filter() {
        let ops = plan_src("MATCH (n:Person) WHERE n.city = 'austin' AND n.age > 30 RETURN n.key")
            .unwrap();
        assert!(
            matches!(&ops[0], PlanOp::IndexScan { field, .. } if field == "city"),
            "equality must fold to IndexScan, got {:?}",
            ops[0]
        );
        assert!(
            ops.iter().any(|op| matches!(op, PlanOp::Filter { .. })),
            "n.age > 30 must remain as residual Filter"
        );
    }

    #[test]
    fn where_on_expanded_var_does_not_fold() {
        let ops =
            plan_src("MATCH (a:Person)-[:KNOWS]->(b:Person) WHERE b.city = 'austin' RETURN a.key")
                .unwrap();
        assert!(
            matches!(&ops[0], PlanOp::ScanLabel { .. } | PlanOp::IndexScan { .. }),
            "first op must be a scan, got {:?}",
            ops[0]
        );
        assert!(
            ops.iter().any(|op| matches!(op, PlanOp::Filter { .. })),
            "b.city filter must remain"
        );
    }

    #[test]
    fn where_inline_prop_and_where_equality_both_usable() {
        // T2: inline prop + WHERE equality on same var → IndexIntersect with both.
        let ops = plan_src("MATCH (n:Person {team: 'core'}) WHERE n.city = 'austin' RETURN n.key")
            .unwrap();
        assert!(
            matches!(&ops[0], PlanOp::IndexIntersect { equalities, .. } if equalities.len() == 2),
            "inline+WHERE equalities must merge to IndexIntersect(2), got {:?}",
            ops[0]
        );
        assert!(
            !ops.iter().any(|op| matches!(op, PlanOp::Filter { .. })),
            "both equalities fully folded; no residual Filter expected"
        );
    }

    // --- IndexIntersect tests (T2) ---

    #[test]
    fn single_equality_inline_stays_index_scan() {
        // Regression: single inline prop must stay IndexScan, not IndexIntersect.
        let ops = plan_src("MATCH (n:Person {city: 'austin'}) RETURN n").unwrap();
        assert!(
            matches!(&ops[0], PlanOp::IndexScan { field, .. } if field == "city"),
            "single-equality inline prop must emit IndexScan, got {:?}",
            ops[0]
        );
    }

    #[test]
    fn compound_inline_props_emit_index_intersect() {
        let ops = plan_src("MATCH (n:Doc {namespace: 'a', status: 'live'}) RETURN n.key").unwrap();
        assert!(
            matches!(&ops[0], PlanOp::IndexIntersect { equalities, .. } if equalities.len() == 2),
            "two inline props must emit IndexIntersect(2), got {:?}",
            ops[0]
        );
    }

    #[test]
    fn where_two_equalities_emit_index_intersect() {
        let ops = plan_src("MATCH (n:Doc) WHERE n.namespace = 'a' AND n.status = $s RETURN n.key")
            .unwrap();
        assert!(
            matches!(&ops[0], PlanOp::IndexIntersect { equalities, .. } if equalities.len() == 2),
            "two WHERE equalities must emit IndexIntersect(2), got {:?}",
            ops[0]
        );
        assert!(
            !ops.iter().any(|op| matches!(op, PlanOp::Filter { .. })),
            "both equalities fully consumed; no residual Filter expected"
        );
    }

    #[test]
    fn mixed_inline_and_where_equalities_merge() {
        let ops = plan_src("MATCH (n:Doc {namespace: 'a'}) WHERE n.status = 'live' RETURN n.key")
            .unwrap();
        assert!(
            matches!(&ops[0], PlanOp::IndexIntersect { equalities, .. } if equalities.len() == 2),
            "inline+WHERE equalities must merge to IndexIntersect(2), got {:?}",
            ops[0]
        );
    }

    #[test]
    fn where_three_equalities_emit_index_intersect() {
        let ops = plan_src(
            "MATCH (n:Doc) WHERE n.namespace = 'a' AND n.status = 'live' AND n.kind = $k RETURN n",
        )
        .unwrap();
        assert!(
            matches!(&ops[0], PlanOp::IndexIntersect { equalities, .. } if equalities.len() == 3),
            "three WHERE equalities must emit IndexIntersect(3), got {:?}",
            ops[0]
        );
        assert!(
            !ops.iter().any(|op| matches!(op, PlanOp::Filter { .. })),
            "all three equalities fully consumed; no residual Filter expected"
        );
    }
}