lora_executor/executor/

helpers.rs

//! Cross-cutting executor helpers used by both the read-only and the
//! mutable executor (and, via `pub(crate)` re-exports through
//! `super::mod`, by the streaming pull pipeline in `crate::pull`).
//!
//! Roughly four groups:
//!
//! 1. Row-set primitives: `dedup_rows` / `dedup_rows_by_vars` for
//!    UNION / DISTINCT, [`compute_aggregate_expr`] for the buffered
//!    aggregation path, [`compare_sort_item`] for the buffered Sort
//!    operator. The streaming pipeline in `crate::pull` calls
//!    [`compute_aggregate_expr`] when the streamable-fold fast-path
//!    classifier rejects a projection.
//! 2. Label / property scans: [`scan_node_ids_for_label_groups`],
//!    [`indexed_node_property_candidates`],
//!    [`node_matches_label_groups`], [`node_matches_property_filter`],
//!    [`label_group_candidates_prefiltered`]. Both NodeByLabelScan and
//!    NodeByPropertyScan share these helpers across the buffered and
//!    streaming pipelines.
//! 3. Path construction: [`build_path_value`] for `PathBuild`,
//!    [`variable_length_expand`] (the buffered BFS path used as the
//!    fallback under non-streaming variable-length expansions) and
//!    [`filter_shortest_paths`] for SHORTEST PATH.
//! 4. Value classification: [`value_matches_property_value`] (used
//!    by every property prefilter), [`hydrate_node_record`] /
//!    [`hydrate_relationship_record`] (single-record hydration), and
//!    the [`GroupValueKey`] dedup / group key.
//!
//! Also hosts the small `eval_properties_expr`, `eval_aggregate_arg_values`,
//! `eval_first_or_null`, `dedup_values`, `as_f64_lossy`,
//! `compare_values_for_sort`, `compare_values_total`,
//! `single_label_hint`, `property_lookup_values`, `type_rank`,
//! `flatten_label_groups` private helpers, and the
//! `MAX_VAR_LEN_HOPS` cap on unbounded variable-length expansion.

use std::cmp::Ordering;
use std::collections::{BTreeMap, BTreeSet};

use web_time::Instant;

use lora_analyzer::symbols::VarId;
use lora_analyzer::{AggregateFunction, FunctionId, ResolvedExpr, ResolvedMapSelector};
use lora_ast::{Direction, RangeLiteral, SortDirection};
use lora_compiler::physical::{
    ExpandExec, HashAggregationExec, LimitExec, NodeByLabelScanExec, NodeByPropertyScanExec,
    NodeScanExec, PhysicalNodeId, PhysicalOp, PhysicalPlan, ProjectionExec, UnwindExec,
};
use lora_store::{GraphStorage, NodeId, Properties, PropertyValue, RelationshipId};

use crate::errors::{value_kind, ExecResult, ExecutorError};
use crate::eval::{eval_expr, eval_expr_result, eval_truthy_result, EvalContext};
use crate::value::{lora_value_to_property, LoraPath, LoraValue, Row};

/// Deadline guard. Returns `QueryTimeout` once the deadline has
/// elapsed; both executors call this every operator-level recursion
/// step and from inside per-row inner loops.
#[inline]
/// Pre-size for a scan that emits `rows x candidates` rows. Capped: a
/// cartesian product (`MATCH (a:A), (b:B)`) would otherwise reserve its
/// full size up front, hundreds of gigabytes at 20k x 20k, before the
/// first deadline check or the first row. Past the cap the Vec grows on
/// demand.
fn scan_output_capacity(rows: usize, candidates: usize) -> usize {
    const MAX_PRESIZE_ROWS: usize = 1 << 16;
    rows.saturating_mul(candidates).min(MAX_PRESIZE_ROWS)
}

/// `check_optional_deadline`, discarding the rows built so far on timeout.
macro_rules! check_deadline_or_discard {
    ($deadline:expr, $out:ident) => {
        if let Err(err) = check_optional_deadline($deadline) {
            discard_rows(std::mem::take(&mut $out));
            return Err(err);
        }
    };
}

pub(super) fn check_deadline_at(deadline: Instant) -> ExecResult<()> {
    // Also true when the query was cancelled; see `crate::cancel`.
    if crate::cancel::eval_tripped() || crate::cancel::deadline_reached(deadline) {
        Err(ExecutorError::QueryTimeout)
    } else {
        Ok(())
    }
}

/// Project `item` from `source` into `projected`. A bare variable takes the
/// source's slot as is, so a large value it holds is shared, not copied.
#[inline]
pub(crate) fn project_item<S: GraphStorage>(
    projected: &mut Row,
    source: &Row,
    item: &lora_analyzer::ResolvedProjection,
    eval_ctx: &EvalContext<'_, S>,
) -> ExecResult<()> {
    if let ResolvedExpr::Variable(var) = &item.expr {
        if projected.insert_named_from(item.output, item.name.clone(), source, *var) {
            return Ok(());
        }
    }
    let value = eval_expr_result(&item.expr, source, eval_ctx).map_err(ExecutorError::from_eval)?;
    projected.insert_named(item.output, item.name.clone(), value);
    Ok(())
}

/// [`project_item`] for a projection that keeps the row's existing bindings.
#[inline]
pub(crate) fn project_item_in_place<S: GraphStorage>(
    row: &mut Row,
    item: &lora_analyzer::ResolvedProjection,
    eval_ctx: &EvalContext<'_, S>,
) -> ExecResult<()> {
    if let ResolvedExpr::Variable(var) = &item.expr {
        if row.insert_named_from_self(item.output, item.name.clone(), *var) {
            return Ok(());
        }
    }
    let value = eval_expr_result(&item.expr, row, eval_ctx).map_err(ExecutorError::from_eval)?;
    row.insert_named(item.output, item.name.clone(), value);
    Ok(())
}

/// The deadline check for buffered per-row loops that are not handed the
/// deadline explicitly (filter, projection, unwind): they read the one
/// the executor made active for this thread. Each row evaluates at least
/// one expression, so one clock read per row is noise.
#[inline]
fn check_active_deadline() -> ExecResult<()> {
    check_optional_deadline(crate::cancel::active_deadline())
}

pub(super) fn filter_rows_checked<S: GraphStorage>(
    input_rows: Vec<Row>,
    predicate: &ResolvedExpr,
    eval_ctx: &EvalContext<'_, S>,
) -> ExecResult<Vec<Row>> {
    let mut out = Vec::with_capacity(input_rows.len());
    for row in input_rows {
        check_active_deadline()?;
        if eval_truthy_result(predicate, &row, eval_ctx).map_err(ExecutorError::from_eval)? {
            out.push(row);
        }
    }
    Ok(out)
}

pub(super) fn project_rows_checked<S: GraphStorage>(
    input_rows: Vec<Row>,
    op: &ProjectionExec,
    eval_ctx: &EvalContext<'_, S>,
) -> ExecResult<Vec<Row>> {
    let mut out = Vec::with_capacity(input_rows.len());

    for row in input_rows {
        check_active_deadline()?;
        if op.include_existing {
            let mut projected = row;
            for item in &op.items {
                project_item_in_place(&mut projected, item, eval_ctx)?;
            }
            out.push(projected);
        } else {
            let mut projected = Row::new();
            for item in &op.items {
                project_item(&mut projected, &row, item, eval_ctx)?;
            }
            out.push(projected);
        }
    }

    Ok(if op.distinct {
        dedup_rows_by_vars(out)
    } else {
        out
    })
}

pub(super) fn unwind_rows<S: GraphStorage>(
    input_rows: Vec<Row>,
    op: &UnwindExec,
    eval_ctx: &EvalContext<'_, S>,
) -> ExecResult<Vec<Row>> {
    let mut out = Vec::with_capacity(input_rows.len());

    for row in input_rows {
        check_active_deadline()?;
        // `eval_expr_result` so a failing list expression (a bad function
        // argument, a missing index) is an error, not an empty UNWIND.
        match eval_expr_result(&op.expr, &row, eval_ctx).map_err(ExecutorError::from_eval)? {
            LoraValue::List(values) => {
                let mut values = values.into_iter();
                let last = values.next_back();
                for value in values {
                    let mut new_row = row.clone();
                    new_row.insert(op.alias, value);
                    out.push(new_row);
                }
                // The input row moves into the last element's output.
                if let Some(value) = last {
                    let mut new_row = row;
                    new_row.insert(op.alias, value);
                    out.push(new_row);
                }
            }
            LoraValue::Null => {}
            other => {
                let mut new_row = row;
                new_row.insert(op.alias, other);
                out.push(new_row);
            }
        }
    }

    Ok(out)
}

/// A `SKIP` or `LIMIT` count, evaluated against an empty row. `null`, a
/// negative number or a non-integer is an error, as in Neo4j: reading
/// `LIMIT null` as "no limit" would turn an omitted parameter into an
/// unbounded result.
pub(crate) fn eval_row_count<S: GraphStorage>(
    clause: &str,
    expr: &ResolvedExpr,
    eval_ctx: &EvalContext<'_, S>,
) -> ExecResult<usize> {
    let value = eval_expr_result(expr, &Row::new(), eval_ctx).map_err(ExecutorError::from_eval)?;
    let n = match value {
        LoraValue::Int(n) => Some(n),
        LoraValue::Float(f) if f.fract() == 0.0 && f.is_finite() => Some(f as i64),
        _ => None,
    };
    match n {
        Some(n) if n >= 0 => Ok(n as usize),
        _ => Err(ExecutorError::RuntimeError(format!(
            "{clause} expects a non-negative integer, got {}",
            describe_count(&value)
        ))),
    }
}

fn describe_count(value: &LoraValue) -> String {
    match value {
        LoraValue::Null => "null".to_string(),
        LoraValue::Int(n) => n.to_string(),
        LoraValue::Float(f) => f.to_string(),
        other => value_kind(other),
    }
}

pub(super) fn limit_rows<S: GraphStorage>(
    mut rows: Vec<Row>,
    op: &LimitExec,
    eval_ctx: &EvalContext<'_, S>,
) -> ExecResult<Vec<Row>> {
    let limit = match op.limit.as_ref() {
        Some(e) => eval_row_count("LIMIT", e, eval_ctx)?,
        None => rows.len(),
    };
    let skip = match op.skip.as_ref() {
        Some(e) => eval_row_count("SKIP", e, eval_ctx)?,
        None => 0,
    };

    if skip >= rows.len() {
        return Ok(Vec::new());
    }

    rows.drain(0..skip);
    rows.truncate(limit);
    Ok(rows)
}

#[inline]
pub(crate) fn bound_node_id_for_expand(row: &Row, var: VarId) -> ExecResult<Option<NodeId>> {
    match row.get(var) {
        Some(LoraValue::Node(id)) => Ok(Some(*id)),
        Some(other) => Err(ExecutorError::ExpectedNodeForExpand {
            var: format!("{var:?}"),
            found: value_kind(other),
        }),
        None => Ok(None),
    }
}

#[inline]
pub(crate) fn bound_relationship_id_for_expand(
    row: &Row,
    var: VarId,
) -> ExecResult<Option<RelationshipId>> {
    match row.get(var) {
        Some(LoraValue::Relationship(id)) => Ok(Some(*id)),
        Some(other) => Err(ExecutorError::ExpectedRelationshipForExpand {
            var: format!("{var:?}"),
            found: value_kind(other),
        }),
        None => Ok(None),
    }
}

pub(super) fn node_scan_rows<S: GraphStorage>(
    storage: &S,
    base_rows: Vec<Row>,
    op: &NodeScanExec,
    deadline: Option<Instant>,
) -> ExecResult<Vec<Row>> {
    let node_ids = storage.all_node_ids();
    let mut out = Vec::with_capacity(scan_output_capacity(base_rows.len(), node_ids.len()));

    if deadline.is_none() {
        for row in base_rows {
            if let Some(existing_id) = bound_node_id_for_expand(&row, op.var)? {
                if storage.has_node(existing_id) {
                    out.push(row);
                }
                continue;
            }

            for &id in &node_ids {
                let mut new_row = row.clone();
                new_row.insert(op.var, LoraValue::Node(id));
                out.push(new_row);
            }
        }
        return Ok(out);
    }

    for row in base_rows {
        check_deadline_or_discard!(deadline, out);
        if let Some(existing_id) = bound_node_id_for_expand(&row, op.var)? {
            if storage.has_node(existing_id) {
                out.push(row);
            }
            continue;
        }

        for &id in &node_ids {
            check_deadline_or_discard!(deadline, out);
            let mut new_row = row.clone();
            new_row.insert(op.var, LoraValue::Node(id));
            out.push(new_row);
        }
    }

    Ok(out)
}

pub(super) fn node_by_label_scan_rows<S: GraphStorage>(
    storage: &S,
    base_rows: Vec<Row>,
    op: &NodeByLabelScanExec,
    deadline: Option<Instant>,
) -> ExecResult<Vec<Row>> {
    let candidate_ids = scan_node_ids_for_label_groups(storage, &op.labels);
    let candidates_prefiltered = label_group_candidates_prefiltered(&op.labels);
    let mut out = Vec::with_capacity(scan_output_capacity(base_rows.len(), candidate_ids.len()));

    if deadline.is_none() {
        for row in base_rows {
            if let Some(existing_id) = bound_node_id_for_expand(&row, op.var)? {
                let labels_ok = storage
                    .with_node(existing_id, |n| {
                        node_matches_label_groups(&n.labels, &op.labels)
                    })
                    .unwrap_or(false);
                if labels_ok {
                    out.push(row);
                }
                continue;
            }

            for &id in &candidate_ids {
                if !candidates_prefiltered {
                    let labels_ok = storage
                        .with_node(id, |n| node_matches_label_groups(&n.labels, &op.labels))
                        .unwrap_or(false);
                    if !labels_ok {
                        continue;
                    }
                }
                let mut new_row = row.clone();
                new_row.insert(op.var, LoraValue::Node(id));
                out.push(new_row);
            }
        }
        return Ok(out);
    }

    for row in base_rows {
        check_deadline_or_discard!(deadline, out);
        if let Some(existing_id) = bound_node_id_for_expand(&row, op.var)? {
            let labels_ok = storage
                .with_node(existing_id, |n| {
                    node_matches_label_groups(&n.labels, &op.labels)
                })
                .unwrap_or(false);
            if labels_ok {
                out.push(row);
            }
            continue;
        }

        for &id in &candidate_ids {
            check_deadline_or_discard!(deadline, out);
            if !candidates_prefiltered {
                let labels_ok = storage
                    .with_node(id, |n| node_matches_label_groups(&n.labels, &op.labels))
                    .unwrap_or(false);
                if !labels_ok {
                    continue;
                }
            }
            let mut new_row = row.clone();
            new_row.insert(op.var, LoraValue::Node(id));
            out.push(new_row);
        }
    }

    Ok(out)
}

pub(super) fn node_by_property_scan_rows<S: GraphStorage>(
    storage: &S,
    params: &BTreeMap<String, LoraValue>,
    base_rows: Vec<Row>,
    op: &NodeByPropertyScanExec,
    deadline: Option<Instant>,
) -> ExecResult<Vec<Row>> {
    let eval_ctx = EvalContext { storage, params };
    let mut out = Vec::new();

    for row in base_rows {
        check_deadline_or_discard!(deadline, out);
        let expected = eval_expr(&op.value, &row, &eval_ctx);

        if let Some(existing_id) = bound_node_id_for_expand(&row, op.var)? {
            if property_scan_matches(
                storage,
                existing_id,
                &op.labels,
                &op.key,
                &expected,
                op.in_list,
            ) {
                out.push(row);
            }
            continue;
        }

        let candidates =
            property_scan_candidates(storage, &op.labels, &op.key, &expected, op.in_list);
        for id in candidates.ids {
            check_deadline_or_discard!(deadline, out);
            if !candidates.prefiltered
                && !property_scan_matches(storage, id, &op.labels, &op.key, &expected, op.in_list)
            {
                continue;
            }
            let mut new_row = row.clone();
            new_row.insert(op.var, LoraValue::Node(id));
            out.push(new_row);
        }
    }

    Ok(out)
}

/// Release rows a timed-out operator had already built. A scan that ran
/// for most of its deadline may hold tens of millions of rows, and freeing
/// them on the query's thread used to take as long again as building them
/// (a 1 s deadline returned after 2-3 s). Large batches are dropped on the
/// rayon pool instead, so the timeout reaches the caller at once.
pub(crate) fn discard_rows(rows: Vec<Row>) {
    #[cfg(all(feature = "parallel", not(target_arch = "wasm32")))]
    if rows.len() >= 4096 {
        rayon::spawn(move || drop(rows));
        return;
    }
    drop(rows);
}

/// Flatten per-input chunks a parallel scan produced, or return the first
/// error, discarding the chunks already built (see [`discard_rows`]).
#[cfg(all(feature = "parallel", not(target_arch = "wasm32")))]
pub(crate) fn flatten_row_chunks(mut chunks: Vec<ExecResult<Vec<Row>>>) -> ExecResult<Vec<Row>> {
    // Look for a failure before copying anything: the chunks built before
    // a timeout can hold millions of rows.
    if let Some(at) = chunks.iter().position(Result::is_err) {
        let err = match chunks.swap_remove(at) {
            Err(err) => err,
            Ok(_) => unreachable!("position() found an error"),
        };
        // Free them off this thread without copying them first.
        rayon::spawn(move || drop(chunks));
        return Err(err);
    }
    let total = chunks.iter().map(|c| c.as_ref().map_or(0, Vec::len)).sum();
    let mut rows = Vec::with_capacity(total);
    for chunk in chunks.into_iter().flatten() {
        rows.extend(chunk);
    }
    Ok(rows)
}

#[inline]
fn check_optional_deadline(deadline: Option<Instant>) -> ExecResult<()> {
    match deadline {
        Some(deadline) => check_deadline_at(deadline),
        None => Ok(()),
    }
}

pub(crate) fn plan_may_need_hydration(plan: &PhysicalPlan) -> bool {
    op_may_need_hydration(plan, plan.root)
}

pub(crate) fn count_all_scan_aggregation_rows<S: GraphStorage>(
    storage: &S,
    plan: &PhysicalPlan,
    op: &HashAggregationExec,
) -> Option<Vec<Row>> {
    if !op.group_by.is_empty() {
        return None;
    }
    let specs = crate::pull::classify_streamable_aggregates(&op.aggregates)?;
    let scan_var = scan_subtree_var(plan, op.input)?;
    // `count(*)` counts rows; `count(n)` of the scanned node counts the
    // same rows, since a scan never binds `n` to null.
    let counts_rows = specs.iter().all(|spec| match spec.kind {
        crate::pull::StreamableAggKind::CountAll => true,
        crate::pull::StreamableAggKind::CountField => {
            matches!(&spec.arg, Some(ResolvedExpr::Variable(v)) if *v == scan_var)
        }
        _ => false,
    });
    if !counts_rows {
        return None;
    }

    let count = count_rows_for_scan_subtree(storage, plan, op.input)? as i64;
    let value = LoraValue::Int(count);
    let mut row = Row::new();
    for proj in &op.aggregates {
        row.insert_named(proj.output, proj.name.clone(), value.clone());
    }
    Some(vec![row])
}

fn scan_subtree_var(plan: &PhysicalPlan, node_id: PhysicalNodeId) -> Option<VarId> {
    match &plan.nodes[node_id] {
        PhysicalOp::NodeScan(op) => Some(op.var),
        PhysicalOp::NodeByLabelScan(op) => Some(op.var),
        _ => None,
    }
}

fn count_rows_for_scan_subtree<S: GraphStorage>(
    storage: &S,
    plan: &PhysicalPlan,
    node_id: PhysicalNodeId,
) -> Option<usize> {
    match &plan.nodes[node_id] {
        PhysicalOp::NodeScan(op) if scan_input_is_argument(plan, op.input) => {
            Some(storage.node_count())
        }
        PhysicalOp::NodeByLabelScan(op) if scan_input_is_argument(plan, op.input) => {
            // One label: the label index already holds the answer.
            if let [group] = op.labels.as_slice() {
                if let [label] = group.as_slice() {
                    return Some(storage.node_count_by_label(label));
                }
            }
            let ids = scan_node_ids_for_label_groups(storage, &op.labels);
            if label_group_candidates_prefiltered(&op.labels) {
                return Some(ids.len());
            }
            Some(
                ids.into_iter()
                    .filter(|&id| {
                        storage
                            .with_node(id, |node| {
                                node_matches_label_groups(&node.labels, &op.labels)
                            })
                            .unwrap_or(false)
                    })
                    .count(),
            )
        }
        _ => None,
    }
}

fn scan_input_is_argument(plan: &PhysicalPlan, input: Option<PhysicalNodeId>) -> bool {
    match input {
        None => true,
        Some(id) => matches!(plan.nodes.get(id), Some(PhysicalOp::Argument(_))),
    }
}

fn op_may_need_hydration(plan: &PhysicalPlan, node_id: PhysicalNodeId) -> bool {
    match &plan.nodes[node_id] {
        PhysicalOp::Projection(op) if !op.include_existing => op
            .items
            .iter()
            .any(|item| expr_may_produce_hydratable_value(&item.expr)),
        PhysicalOp::HashAggregation(op) => {
            op.group_by
                .iter()
                .any(|item| expr_may_produce_hydratable_value(&item.expr))
                || op
                    .aggregates
                    .iter()
                    .any(|item| expr_may_produce_hydratable_value(&item.expr))
        }
        PhysicalOp::Sort(op) => op_may_need_hydration(plan, op.input),
        PhysicalOp::Limit(op) => op_may_need_hydration(plan, op.input),
        PhysicalOp::Filter(op) => op_may_need_hydration(plan, op.input),
        PhysicalOp::Unwind(op) => op_may_need_hydration(plan, op.input),
        PhysicalOp::PathBuild(op) => op_may_need_hydration(plan, op.input),
        _ => true,
    }
}

fn expr_may_produce_hydratable_value(expr: &ResolvedExpr) -> bool {
    match expr {
        ResolvedExpr::Variable(_) | ResolvedExpr::Parameter(_) => true,
        ResolvedExpr::Literal(_)
        | ResolvedExpr::Property { .. }
        | ResolvedExpr::ExistsSubquery { .. }
        | ResolvedExpr::Binary { .. }
        | ResolvedExpr::Unary { .. }
        | ResolvedExpr::ListPredicate { .. } => false,
        ResolvedExpr::Function { function, args, .. } => {
            function_may_produce_hydratable_value(*function, args)
        }
        ResolvedExpr::List(items) => items.iter().any(expr_may_produce_hydratable_value),
        ResolvedExpr::Map(items) => items
            .iter()
            .any(|(_, value)| expr_may_produce_hydratable_value(value)),
        ResolvedExpr::Case {
            alternatives,
            else_expr,
            ..
        } => {
            alternatives
                .iter()
                .any(|(_, value)| expr_may_produce_hydratable_value(value))
                || else_expr
                    .as_deref()
                    .is_some_and(expr_may_produce_hydratable_value)
        }
        ResolvedExpr::ListComprehension { map_expr, .. } => map_expr
            .as_deref()
            .map(expr_may_produce_hydratable_value)
            .unwrap_or(true),
        ResolvedExpr::Reduce { expr, .. } => expr_may_produce_hydratable_value(expr),
        ResolvedExpr::MapProjection { selectors, .. } => selectors.iter().any(|selector| {
            matches!(selector, ResolvedMapSelector::Literal(_, expr) if expr_may_produce_hydratable_value(expr))
        }),
        ResolvedExpr::Index { expr, .. } | ResolvedExpr::Slice { expr, .. } => {
            expr_may_produce_hydratable_value(expr)
        }
        ResolvedExpr::PatternComprehension { map_expr, .. } => {
            expr_may_produce_hydratable_value(map_expr)
        }
    }
}

fn function_may_produce_hydratable_value(function: FunctionId, args: &[ResolvedExpr]) -> bool {
    match function.name() {
        "path.nodes" | "path.edges" | "path.first" | "path.last" | "list.first" | "list.last" => {
            true
        }
        "value.coalesce" | "value.first_non_null" | "collect" => {
            args.iter().any(expr_may_produce_hydratable_value)
        }
        "list.rest" | "value.reverse" | "list.reverse" => {
            args.first().is_some_and(expr_may_produce_hydratable_value)
        }
        _ => false,
    }
}

pub(super) fn expand_rows<S: GraphStorage>(
    storage: &S,
    params: &BTreeMap<String, LoraValue>,
    input_rows: Vec<Row>,
    op: &ExpandExec,
) -> ExecResult<Vec<Row>> {
    let eval_ctx = EvalContext { storage, params };
    // Most expansions emit at least one row per input; sizing up front
    // avoids re-copying every row as a large output buffer grows.
    let mut out = Vec::with_capacity(input_rows.len());
    // Matching (rel, dst) pairs for the current input row. Collected
    // first so the input row can be *moved* into its last output
    // instead of cloned: a degree-k node costs k-1 row clones, and a
    // degree-1 hop (chains, trees walked upward) costs none.
    let mut matches: Vec<(u64, u64)> = Vec::new();

    for row in input_rows {
        let Some(src_node_id) = bound_node_id_for_expand(&row, op.src)? else {
            continue;
        };

        let mut rel_property_filter = None;
        matches.clear();

        storage.try_for_each_expand_id(
            src_node_id,
            op.direction,
            &op.types,
            |rel_id, dst_id| {
                if let Some(expr) = op.rel_properties.as_ref() {
                    if rel_property_filter.is_none() {
                        let expected = eval_expr(expr, &row, &eval_ctx);
                        let LoraValue::Map(map) = expected else {
                            return Err(ExecutorError::ExpectedPropertyMap {
                                found: value_kind(&expected),
                            });
                        };
                        rel_property_filter = Some(map);
                    }

                    let Some(map) = rel_property_filter.as_ref() else {
                        return Ok(());
                    };
                    let matches = storage
                        .with_relationship(rel_id, |rel| {
                            map.iter().all(|(key, expected)| {
                                rel.properties
                                    .get(key.as_str())
                                    .map(|actual| value_matches_property_value(expected, actual))
                                    .unwrap_or(false)
                            })
                        })
                        .unwrap_or(false);
                    if !matches {
                        return Ok(());
                    }
                }

                if let Some(existing_id) = bound_node_id_for_expand(&row, op.dst)? {
                    if existing_id != dst_id {
                        return Ok(());
                    }
                }

                if let Some(rel_var) = op.rel {
                    if let Some(existing_id) = bound_relationship_id_for_expand(&row, rel_var)? {
                        if existing_id != rel_id {
                            return Ok(());
                        }
                    }
                }

                matches.push((rel_id, dst_id));
                Ok(())
            },
        )?;

        let Some((&last, rest)) = matches.split_last() else {
            continue;
        };
        for &(rel_id, dst_id) in rest {
            let mut new_row = row.clone();
            bind_expand_target(&mut new_row, op, rel_id, dst_id);
            out.push(new_row);
        }
        let mut new_row = row;
        bind_expand_target(&mut new_row, op, last.0, last.1);
        out.push(new_row);
    }

    Ok(out)
}

#[inline]
fn bind_expand_target(row: &mut Row, op: &ExpandExec, rel_id: u64, dst_id: u64) {
    if !row.contains_key(op.dst) {
        row.insert(op.dst, LoraValue::Node(dst_id));
    }
    if let Some(rel_var) = op.rel {
        if !row.contains_key(rel_var) {
            row.insert(rel_var, LoraValue::Relationship(rel_id));
        }
    }
}

pub(super) fn expand_var_len_rows<S: GraphStorage>(
    storage: &S,
    input_rows: Vec<Row>,
    op: &ExpandExec,
    range: &RangeLiteral,
) -> ExecResult<Vec<Row>> {
    let (min_hops, max_hops) = resolve_range(range);
    let bind_relationships = op.rel.is_some();
    let mut out = Vec::new();

    for row in input_rows {
        let Some(src_node_id) = bound_node_id_for_expand(&row, op.src)? else {
            continue;
        };
        // A destination the row already binds is checked, never rebound.
        let bound_dst = bound_node_id_for_expand(&row, op.dst)?;

        let expansions = variable_length_expand(
            storage,
            src_node_id,
            op.direction,
            &op.types,
            min_hops,
            max_hops,
            bind_relationships,
        );

        for result in expansions {
            if bound_dst.is_some_and(|dst| dst != result.dst_node_id) {
                continue;
            }
            let mut new_row = row.clone();
            new_row.insert(op.dst, LoraValue::Node(result.dst_node_id));

            if let Some(rel_var) = op.rel {
                let rel_list = LoraValue::List(
                    result
                        .rel_ids
                        .into_iter()
                        .map(LoraValue::Relationship)
                        .collect(),
                );
                new_row.insert(rel_var, rel_list);
            }

            out.push(new_row);
        }
    }

    Ok(out)
}

pub(super) fn properties_to_value_map(props: &Properties) -> LoraValue {
    let mut map = BTreeMap::new();
    for (k, v) in props.iter() {
        map.insert(k.to_string(), LoraValue::from(v));
    }
    LoraValue::Map(map)
}

/// Dedup rows that share the same schema (same VarId set). Compares rows by
/// a Vec<GroupValueKey> keyed on VarId iteration order — avoids the per-row
/// column-name String clones of `dedup_rows`. Used by DISTINCT projection.
pub(crate) fn dedup_rows_by_vars(rows: Vec<Row>) -> Vec<Row> {
    let mut seen: BTreeSet<Vec<GroupValueKey>> = BTreeSet::new();
    let mut out = Vec::new();

    for row in rows {
        let key: Vec<GroupValueKey> = row
            .iter()
            .map(|(_, val)| GroupValueKey::from_value(val))
            .collect();
        if seen.insert(key) {
            out.push(row);
        }
    }

    out
}

/// Dedup rows using named entries so rows with different VarIds but the same
/// column name + value are collapsed. Needed for UNION where each branch has
/// its own VarIds.
pub(crate) fn dedup_rows(rows: Vec<Row>) -> Vec<Row> {
    let mut seen: BTreeSet<Vec<(String, GroupValueKey)>> = BTreeSet::new();
    let mut out = Vec::new();

    for row in rows {
        let key: Vec<(String, GroupValueKey)> = row
            .iter_named()
            .map(|(_, name, val)| (name.into_owned(), GroupValueKey::from_value(val)))
            .collect();
        if seen.insert(key) {
            out.push(row);
        }
    }

    out
}

pub(super) fn eval_properties_expr<S: GraphStorage>(
    expr: &ResolvedExpr,
    row: &Row,
    storage: &S,
    params: &BTreeMap<String, LoraValue>,
) -> ExecResult<Properties> {
    let eval_ctx = EvalContext { storage, params };

    // A `null` value in a property map means "no property": it is never
    // stored, so `keys(n)` and existence checks do not see it.
    if let ResolvedExpr::Map(items) = expr {
        let mut out = Properties::new();
        for (k, v) in items {
            let value = eval_expr(v, row, &eval_ctx);
            if matches!(value, LoraValue::Null) {
                continue;
            }
            let prop = lora_value_to_property(value)
                .map_err(|e| ExecutorError::RuntimeError(e.to_string()))?;
            out.insert(lora_store::intern(k), prop);
        }
        return Ok(out);
    }

    match eval_expr(expr, row, &eval_ctx) {
        LoraValue::Map(map) => {
            let mut out = Properties::new();
            for (k, v) in map {
                if matches!(v, LoraValue::Null) {
                    continue;
                }
                let prop = lora_value_to_property(v)
                    .map_err(|e| ExecutorError::RuntimeError(e.to_string()))?;
                // Route every CREATE/SET property key through the
                // process-wide intern table so all nodes sharing a
                // column name share one heap allocation. This is the
                // hot path for `UNWIND $rows AS r CREATE (...)` and
                // is where the bulk CSV-import memory win actually
                // lands.
                out.insert(lora_store::intern_owned(k), prop);
            }
            Ok(out)
        }
        other => Err(ExecutorError::ExpectedPropertyMap {
            found: value_kind(&other),
        }),
    }
}

pub(crate) fn compute_aggregate_expr<S: GraphStorage>(
    expr: &ResolvedExpr,
    rows: &[Row],
    eval_ctx: &EvalContext<'_, S>,
) -> ExecResult<LoraValue> {
    match expr {
        ResolvedExpr::Function {
            function,
            distinct,
            args,
        } => {
            let func = function.as_aggregate();
            match func {
                Some(AggregateFunction::Count) => {
                    if args.is_empty() {
                        return Ok(LoraValue::Int(rows.len() as i64));
                    }

                    let mut values = eval_aggregate_arg_values(&args[0], rows, eval_ctx)?;
                    values.retain(|v| !matches!(v, LoraValue::Null));

                    if *distinct {
                        values = dedup_values(values);
                    }

                    Ok(LoraValue::Int(values.len() as i64))
                }

                Some(AggregateFunction::Collect) => {
                    if args.is_empty() {
                        return Ok(LoraValue::List(Vec::new()));
                    }

                    let mut values = eval_aggregate_arg_values(&args[0], rows, eval_ctx)?;
                    // openCypher: collect() ignores null values. Only a
                    // top-level null is dropped; a list or map that
                    // merely contains nulls is kept as-is.
                    values.retain(|v| !matches!(v, LoraValue::Null));

                    if *distinct {
                        values = dedup_values(values);
                    }

                    Ok(LoraValue::List(values))
                }

                Some(kind @ (AggregateFunction::Sum | AggregateFunction::Avg)) => {
                    if args.is_empty() {
                        return Ok(LoraValue::Null);
                    }

                    let mut values = eval_aggregate_arg_values(&args[0], rows, eval_ctx)?;

                    if *distinct {
                        values = dedup_values(values);
                    }

                    // One implementation with the streaming fold.
                    let stream_kind = if kind == AggregateFunction::Sum {
                        crate::pull::StreamableAggKind::Sum
                    } else {
                        crate::pull::StreamableAggKind::Avg
                    };
                    let mut state = crate::pull::AggState::seed(stream_kind);
                    for value in values {
                        state.fold(stream_kind, value);
                    }
                    state.finalize(stream_kind)
                }

                Some(AggregateFunction::Min) => {
                    if args.is_empty() {
                        return Ok(LoraValue::Null);
                    }

                    let mut values = eval_aggregate_arg_values(&args[0], rows, eval_ctx)?;
                    values.retain(|v| !matches!(v, LoraValue::Null));

                    if *distinct {
                        values = dedup_values(values);
                    }

                    Ok(values
                        .into_iter()
                        .min_by(compare_values_total)
                        .unwrap_or(LoraValue::Null))
                }

                Some(AggregateFunction::Max) => {
                    if args.is_empty() {
                        return Ok(LoraValue::Null);
                    }

                    let mut values = eval_aggregate_arg_values(&args[0], rows, eval_ctx)?;
                    values.retain(|v| !matches!(v, LoraValue::Null));

                    if *distinct {
                        values = dedup_values(values);
                    }

                    Ok(values
                        .into_iter()
                        .max_by(compare_values_total)
                        .unwrap_or(LoraValue::Null))
                }

                Some(AggregateFunction::Stdev | AggregateFunction::Stdevp) => {
                    if args.is_empty() {
                        return Ok(LoraValue::Null);
                    }

                    let nums: Vec<f64> = aggregate_numbers(&args[0], *distinct, rows, eval_ctx)?
                        .into_iter()
                        .map(|(n, _)| n)
                        .collect();

                    let is_population = matches!(func, Some(AggregateFunction::Stdevp));

                    if nums.is_empty() || (!is_population && nums.len() < 2) {
                        return Ok(LoraValue::Float(0.0));
                    }

                    let mean = nums.iter().sum::<f64>() / nums.len() as f64;
                    let variance_sum: f64 = nums.iter().map(|x| (x - mean).powi(2)).sum();
                    let denom = if is_population {
                        nums.len() as f64
                    } else {
                        (nums.len() - 1) as f64
                    };
                    Ok(LoraValue::Float((variance_sum / denom).sqrt()))
                }

                Some(AggregateFunction::PercentileCont) => {
                    if args.len() < 2 {
                        return Ok(LoraValue::Null);
                    }

                    let Some(first) = rows.first() else {
                        return Ok(LoraValue::Null);
                    };

                    let percentile = eval_expr_result(&args[1], first, eval_ctx)
                        .map_err(ExecutorError::from_eval)?
                        .as_f64()
                        .map(normalize_percentile)
                        .unwrap_or(0.5);
                    let mut nums: Vec<f64> =
                        aggregate_numbers(&args[0], *distinct, rows, eval_ctx)?
                            .into_iter()
                            .map(|(n, _)| n)
                            .collect();

                    if nums.is_empty() {
                        return Ok(LoraValue::Null);
                    }

                    nums.sort_by(|a, b| a.partial_cmp(b).unwrap_or(Ordering::Equal));

                    let index = percentile * (nums.len() - 1) as f64;
                    let lower = index.floor() as usize;
                    let upper = index.ceil() as usize;
                    let fraction = index - lower as f64;

                    if lower == upper || upper >= nums.len() {
                        Ok(LoraValue::Float(nums[lower]))
                    } else {
                        Ok(LoraValue::Float(
                            nums[lower] * (1.0 - fraction) + nums[upper] * fraction,
                        ))
                    }
                }

                Some(AggregateFunction::PercentileDisc) => {
                    if args.len() < 2 {
                        return Ok(LoraValue::Null);
                    }

                    let Some(first) = rows.first() else {
                        return Ok(LoraValue::Null);
                    };

                    let percentile = eval_expr_result(&args[1], first, eval_ctx)
                        .map_err(ExecutorError::from_eval)?
                        .as_f64()
                        .map(normalize_percentile)
                        .unwrap_or(0.5);
                    let mut nums = aggregate_numbers(&args[0], *distinct, rows, eval_ctx)?;

                    if nums.is_empty() {
                        return Ok(LoraValue::Null);
                    }

                    nums.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(Ordering::Equal));

                    // The value itself, as it was: an integer stays an integer.
                    let index = (percentile * (nums.len() - 1) as f64).round() as usize;
                    let index = index.min(nums.len() - 1);
                    Ok(nums.swap_remove(index).1)
                }

                _ => eval_first_or_null(expr, rows, eval_ctx),
            }
        }

        _ => eval_first_or_null(expr, rows, eval_ctx),
    }
}

/// The numeric values of an aggregate's argument over `rows`, each with
/// its numeric reading, deduplicated for DISTINCT; others are ignored.
fn aggregate_numbers<S: GraphStorage>(
    expr: &ResolvedExpr,
    distinct: bool,
    rows: &[Row],
    eval_ctx: &EvalContext<'_, S>,
) -> ExecResult<Vec<(f64, LoraValue)>> {
    let mut values = eval_aggregate_arg_values(expr, rows, eval_ctx)?;
    if distinct {
        values = dedup_values(values);
    }
    Ok(values
        .into_iter()
        .filter_map(|v| match v {
            LoraValue::Int(i) => Some((i as f64, v)),
            LoraValue::Float(f) => Some((f, v)),
            _ => None,
        })
        .collect())
}

fn eval_aggregate_arg_values<S: GraphStorage>(
    expr: &ResolvedExpr,
    rows: &[Row],
    eval_ctx: &EvalContext<'_, S>,
) -> ExecResult<Vec<LoraValue>> {
    rows.iter()
        .map(|row| eval_expr_result(expr, row, eval_ctx).map_err(ExecutorError::from_eval))
        .collect()
}

fn normalize_percentile(value: f64) -> f64 {
    if value.is_finite() {
        value.clamp(0.0, 1.0)
    } else {
        0.5
    }
}

fn eval_first_or_null<S: GraphStorage>(
    expr: &ResolvedExpr,
    rows: &[Row],
    eval_ctx: &EvalContext<'_, S>,
) -> ExecResult<LoraValue> {
    match rows.first() {
        Some(row) => eval_expr_result(expr, row, eval_ctx).map_err(ExecutorError::from_eval),
        None => Ok(LoraValue::Null),
    }
}

fn dedup_values(values: Vec<LoraValue>) -> Vec<LoraValue> {
    let mut seen: BTreeSet<GroupValueKey> = BTreeSet::new();
    let mut out = Vec::new();

    for value in values {
        let key = GroupValueKey::from_value(&value);
        if seen.insert(key) {
            out.push(value);
        }
    }

    out
}

pub(crate) fn compare_values_total(a: &LoraValue, b: &LoraValue) -> Ordering {
    use LoraValue::*;

    match (a, b) {
        (Bool(x), Bool(y)) => x.cmp(y),
        (Int(x), Int(y)) => x.cmp(y),
        (Float(x), Float(y)) => x.partial_cmp(y).unwrap_or(Ordering::Equal),
        (Int(x), Float(y)) => (*x as f64).partial_cmp(y).unwrap_or(Ordering::Equal),
        (Float(x), Int(y)) => x.partial_cmp(&(*y as f64)).unwrap_or(Ordering::Equal),
        (String(x), String(y)) => x.cmp(y),
        (Binary(x), Binary(y)) => x.segments().cmp(y.segments()),
        (Node(x), Node(y)) => x.cmp(y),
        (Relationship(x), Relationship(y)) => x.cmp(y),
        (Date(x), Date(y)) => x.cmp(y),
        (DateTime(x), DateTime(y)) => x.cmp(y),
        (LocalDateTime(x), LocalDateTime(y)) => x.cmp(y),
        (Time(x), Time(y)) => x.cmp(y),
        (LocalTime(x), LocalTime(y)) => x.cmp(y),
        (Duration(x), Duration(y)) => x.cmp(y),
        (Vector(x), Vector(y)) => x.to_key_string().cmp(&y.to_key_string()),
        _ => type_rank(a)
            .cmp(&type_rank(b))
            .then_with(|| format!("{a:?}").cmp(&format!("{b:?}"))),
    }
}

pub fn value_matches_property_value(expected: &LoraValue, actual: &PropertyValue) -> bool {
    match (expected, actual) {
        (LoraValue::Null, PropertyValue::Null) => true,
        (LoraValue::Bool(a), PropertyValue::Bool(b)) => a == b,
        (LoraValue::Int(a), PropertyValue::Int(b)) => a == b,
        (LoraValue::Float(a), PropertyValue::Float(b)) => a == b,
        (LoraValue::Int(a), PropertyValue::Float(b)) => (*a as f64) == *b,
        (LoraValue::Float(a), PropertyValue::Int(b)) => *a == (*b as f64),
        (LoraValue::String(a), PropertyValue::String(b)) => a == b,
        (LoraValue::Binary(a), PropertyValue::Binary(b)) => a == b,

        (LoraValue::List(xs), PropertyValue::List(ys)) => {
            xs.len() == ys.len()
                && xs
                    .iter()
                    .zip(ys.iter())
                    .all(|(x, y)| value_matches_property_value(x, y))
        }

        (LoraValue::Map(xm), PropertyValue::Map(ym)) => xm.iter().all(|(k, xv)| {
            ym.get(k)
                .map(|yv| value_matches_property_value(xv, yv))
                .unwrap_or(false)
        }),

        (LoraValue::Date(a), PropertyValue::Date(b)) => a == b,
        (LoraValue::DateTime(a), PropertyValue::DateTime(b)) => a == b,
        (LoraValue::LocalDateTime(a), PropertyValue::LocalDateTime(b)) => a == b,
        (LoraValue::Time(a), PropertyValue::Time(b)) => a == b,
        (LoraValue::LocalTime(a), PropertyValue::LocalTime(b)) => a == b,
        (LoraValue::Duration(a), PropertyValue::Duration(b)) => a == b,
        (LoraValue::Point(a), PropertyValue::Point(b)) => a == b,
        (LoraValue::Vector(a), PropertyValue::Vector(b)) => a == b,

        _ => false,
    }
}

pub(crate) fn node_matches_property_filter<S: GraphStorage>(
    storage: &S,
    node_id: NodeId,
    labels: &[Vec<String>],
    key: &str,
    expected: &LoraValue,
) -> bool {
    storage
        .with_node(node_id, |node| {
            node_matches_label_groups(&node.labels, labels)
                && node
                    .properties
                    .get(key)
                    .map(|actual| value_matches_property_value(expected, actual))
                    .unwrap_or(false)
        })
        .unwrap_or(false)
}

fn single_label_hint(labels: &[Vec<String>]) -> Option<&str> {
    if labels.len() == 1 && labels[0].len() == 1 {
        Some(labels[0][0].as_str())
    } else {
        None
    }
}

fn property_lookup_values(expected: &LoraValue) -> Option<Vec<PropertyValue>> {
    let property = lora_value_to_property(expected.clone()).ok()?;
    let mut values = vec![property.clone()];

    match property {
        PropertyValue::Int(i) => {
            values.push(PropertyValue::Float(i as f64));
        }
        PropertyValue::Float(f)
            if f.is_finite()
                && f.fract() == 0.0
                && f >= i64::MIN as f64
                && f <= i64::MAX as f64 =>
        {
            values.push(PropertyValue::Int(f as i64));
        }
        _ => {}
    }

    Some(values)
}

pub(crate) struct NodePropertyCandidates {
    pub(crate) ids: Vec<NodeId>,
    pub(crate) prefiltered: bool,
}

pub(crate) fn node_by_property_range_scan_rows<S: GraphStorage>(
    storage: &S,
    params: &BTreeMap<String, LoraValue>,
    base_rows: Vec<Row>,
    op: &lora_compiler::NodeByPropertyRangeScanExec,
    deadline: Option<Instant>,
) -> ExecResult<Vec<Row>> {
    let eval_ctx = EvalContext { storage, params };
    let mut out = Vec::new();
    let mut other_kinds = OtherKindScan::default();

    for row in base_rows {
        check_optional_deadline(deadline)?;
        let lo_value = op.lo.as_ref().map(|expr| eval_expr(expr, &row, &eval_ctx));
        // A null upper bound reads as none: the Filter kept above every
        // index scan still judges each row (a STARTS WITH prefix with no
        // successor, `string.prefix_end` null, scans to the end).
        let hi_value = op
            .hi
            .as_ref()
            .map(|expr| eval_expr(expr, &row, &eval_ctx))
            .filter(|v| !matches!(v, LoraValue::Null));
        let lo_prop = lo_value
            .clone()
            .and_then(|v| lora_value_to_property(v).ok());
        let hi_prop = hi_value
            .clone()
            .and_then(|v| lora_value_to_property(v).ok());
        let filter = NodeRangeFilter {
            labels: &op.labels,
            key: &op.key,
            lo: lo_value.as_ref(),
            lo_inclusive: op.lo_inclusive,
            hi: hi_value.as_ref(),
            hi_inclusive: op.hi_inclusive,
        };
        let bounds = [lo_value.as_ref(), hi_value.as_ref()];
        let bound_id = bound_node_id_for_expand(&row, op.var)?;

        if let Some(existing_id) = bound_id {
            if node_matches_range_filter(storage, existing_id, &filter)
                || bound_node_has_other_kind(storage, existing_id, op, bounds)
            {
                out.push(row);
            }
            continue;
        }

        // Temporals of another kind first, for the Filter above to judge.
        for id in other_kind_node_ids(storage, op, bounds, &mut other_kinds) {
            let mut new_row = row.clone();
            new_row.insert(op.var, LoraValue::Node(id));
            out.push(new_row);
        }

        if op.order.is_some() {
            let mut cursor =
                OrderedRangeCursor::new(storage, op, lo_value.clone(), hi_value.clone());
            while let Some(id) = cursor.next_id(storage) {
                check_optional_deadline(deadline)?;
                let mut new_row = row.clone();
                new_row.insert(op.var, LoraValue::Node(id));
                out.push(new_row);
            }
            continue;
        }

        let candidate_ids = match single_label_hint(&op.labels) {
            Some(label) => storage
                .node_range_candidates(label, &op.key, lo_prop.as_ref(), hi_prop.as_ref())
                .unwrap_or_else(|| scan_node_ids_for_label_groups(storage, &op.labels)),
            None => scan_node_ids_for_label_groups(storage, &op.labels),
        };

        for id in candidate_ids {
            check_optional_deadline(deadline)?;
            if node_matches_range_filter(storage, id, &filter) {
                let mut new_row = row.clone();
                new_row.insert(op.var, LoraValue::Node(id));
                out.push(new_row);
            }
        }
    }

    Ok(out)
}

pub(crate) fn node_by_text_scan_rows<S: GraphStorage>(
    storage: &S,
    params: &BTreeMap<String, LoraValue>,
    base_rows: Vec<Row>,
    op: &lora_compiler::NodeByTextScanExec,
    deadline: Option<Instant>,
) -> ExecResult<Vec<Row>> {
    let eval_ctx = EvalContext { storage, params };
    let mut out = Vec::new();

    for row in base_rows {
        check_optional_deadline(deadline)?;
        let query = eval_expr(&op.query, &row, &eval_ctx);
        let LoraValue::String(query_str) = &query else {
            // Non-string query → predicate cannot match anything;
            // skip the row entirely (matches scan + filter behaviour).
            continue;
        };

        if let Some(existing_id) = bound_node_id_for_expand(&row, op.var)? {
            if node_matches_text_filter(
                storage,
                existing_id,
                &op.labels,
                &op.key,
                op.predicate,
                query_str,
            ) {
                out.push(row);
            }
            continue;
        }

        let candidate_ids = match single_label_hint(&op.labels) {
            Some(label) => storage
                .node_text_candidates(label, &op.key, query_str)
                .unwrap_or_else(|| scan_node_ids_for_label_groups(storage, &op.labels)),
            None => scan_node_ids_for_label_groups(storage, &op.labels),
        };

        for id in candidate_ids {
            check_optional_deadline(deadline)?;
            if node_matches_text_filter(storage, id, &op.labels, &op.key, op.predicate, query_str) {
                let mut new_row = row.clone();
                new_row.insert(op.var, LoraValue::Node(id));
                out.push(new_row);
            }
        }
    }

    Ok(out)
}

/// The temporal kinds of a range scan's bounds, at most two.
fn temporal_bound_kinds(bounds: [Option<&LoraValue>; 2]) -> Vec<(&'static str, &LoraValue)> {
    let mut kinds: Vec<(&'static str, &LoraValue)> = Vec::with_capacity(2);
    for bound in bounds.into_iter().flatten() {
        if let Some(kind) = temporal_kind_name(bound) {
            if !kinds.iter().any(|(k, _)| *k == kind) {
                kinds.push((kind, bound));
            }
        }
    }
    kinds
}

/// Whether `value` is a temporal of a kind none of `kinds` is.
fn is_other_temporal_kind(value: &PropertyValue, kinds: &[(&'static str, &LoraValue)]) -> bool {
    temporal_kind_name(&LoraValue::from(value))
        .is_some_and(|kind| kinds.iter().any(|(k, _)| *k != kind))
}

/// Per-call memo of a label or type scan, for when no index can list the
/// other-kind ids. Held for one scan call only: in the pull path writes
/// can land between rows, so it is never kept across them.
#[derive(Default)]
pub(crate) struct OtherKindScan<Id> {
    temporals: Option<Vec<(Id, &'static str)>>,
}

impl<Id: Copy> OtherKindScan<Id> {
    fn ids(
        &mut self,
        kinds: &[(&'static str, &LoraValue)],
        scan: impl FnOnce() -> Vec<(Id, &'static str)>,
    ) -> Vec<Id> {
        self.temporals
            .get_or_insert_with(scan)
            .iter()
            .filter(|(_, kind)| kinds.iter().any(|(k, _)| k != kind))
            .map(|(id, _)| *id)
            .collect()
    }
}

/// Nodes a range scan emits besides its matches: those whose `op.key`
/// holds a temporal of another kind than a temporal bound. The index keeps
/// each kind apart and would skip them without a word; emitted, they reach
/// the `Filter` above the scan, which compares them as an unindexed scan
/// does. So an unguarded `x >= date(…)` fails on a DATETIME `x`, while a
/// guard such as `type.of(x) = 'DATE' AND x >= date(…)` drops the row
/// before the comparison (`AND` short-circuits). Reads only the other
/// kinds' runs of the index (two probes plus the ids found); without an
/// index, the label once per call.
pub(crate) fn other_kind_node_ids<S: GraphStorage>(
    storage: &S,
    op: &lora_compiler::NodeByPropertyRangeScanExec,
    bounds: [Option<&LoraValue>; 2],
    fallback: &mut OtherKindScan<NodeId>,
) -> Vec<NodeId> {
    let kinds = temporal_bound_kinds(bounds);
    if kinds.is_empty() {
        return Vec::new();
    }
    // A label every matching node has: its index lists the candidates.
    let index_label = op
        .labels
        .iter()
        .find(|group| group.len() == 1)
        .map(|group| group[0].as_str());
    let mut ids = BTreeSet::new();
    let mut need_scan = index_label.is_none();
    if let Some(label) = index_label {
        for (_, bound) in &kinds {
            let listed = lora_value_to_property((*bound).clone())
                .ok()
                .and_then(|like| storage.node_range_other_temporal_kind_ids(label, &op.key, &like));
            match listed {
                Some(listed) => ids.extend(listed),
                None => need_scan = true,
            }
        }
        // Several bound kinds: a value of one bound's kind is "other" for
        // the other bound, which each probe already lists.
    }
    if need_scan {
        ids.extend(fallback.ids(&kinds, || {
            scan_node_ids_for_label_groups(storage, &op.labels)
                .into_iter()
                .filter_map(|id| {
                    storage
                        .with_node(id, |n| {
                            n.properties
                                .get(op.key.as_str())
                                .and_then(|v| temporal_kind_name(&LoraValue::from(v)))
                        })
                        .flatten()
                        .map(|kind| (id, kind))
                })
                .collect()
        }));
        return ids.into_iter().collect();
    }
    let single = op.labels.len() == 1 && op.labels[0].len() == 1;
    ids.into_iter()
        .filter(|id| {
            single
                || storage
                    .with_node(*id, |n| node_matches_label_groups(&n.labels, &op.labels))
                    .unwrap_or(false)
        })
        .collect()
}

/// Whether the bound node `id` matches the scan's labels and holds a
/// temporal of another kind than a temporal bound; see
/// [`other_kind_node_ids`].
pub(crate) fn bound_node_has_other_kind<S: GraphStorage>(
    storage: &S,
    id: NodeId,
    op: &lora_compiler::NodeByPropertyRangeScanExec,
    bounds: [Option<&LoraValue>; 2],
) -> bool {
    let kinds = temporal_bound_kinds(bounds);
    !kinds.is_empty()
        && storage
            .with_node(id, |n| {
                node_matches_label_groups(&n.labels, &op.labels)
                    && n.properties
                        .get(op.key.as_str())
                        .is_some_and(|v| is_other_temporal_kind(v, &kinds))
            })
            .unwrap_or(false)
}

/// Relationship counterpart of [`other_kind_node_ids`].
fn other_kind_rel_ids<S: GraphStorage>(
    storage: &S,
    op: &lora_compiler::RelByPropertyRangeScanExec,
    bounds: [Option<&LoraValue>; 2],
    fallback: &mut OtherKindScan<RelationshipId>,
) -> Vec<RelationshipId> {
    let kinds = temporal_bound_kinds(bounds);
    if kinds.is_empty() {
        return Vec::new();
    }
    let mut ids = BTreeSet::new();
    let mut need_scan = op.types.is_empty();
    for ty in &op.types {
        for (_, bound) in &kinds {
            let listed = lora_value_to_property((*bound).clone())
                .ok()
                .and_then(|like| {
                    storage.relationship_range_other_temporal_kind_ids(ty, &op.key, &like)
                });
            match listed {
                Some(listed) => ids.extend(listed),
                None => need_scan = true,
            }
        }
    }
    if need_scan {
        ids.extend(fallback.ids(&kinds, || {
            rel_candidate_ids(storage, &op.types, |_| None)
                .into_iter()
                .filter_map(|id| {
                    storage
                        .with_relationship(id, |r| {
                            r.properties
                                .get(op.key.as_str())
                                .and_then(|v| temporal_kind_name(&LoraValue::from(v)))
                        })
                        .flatten()
                        .map(|kind| (id, kind))
                })
                .collect()
        }));
    }
    ids.into_iter().collect()
}

fn temporal_kind_name(value: &LoraValue) -> Option<&'static str> {
    Some(match value {
        LoraValue::Date(_) => "DATE",
        LoraValue::DateTime(_) => "DATETIME",
        LoraValue::LocalDateTime(_) => "LOCAL_DATETIME",
        LoraValue::Time(_) => "TIME",
        LoraValue::LocalTime(_) => "LOCAL_TIME",
        _ => return None,
    })
}

pub(crate) struct NodeRangeFilter<'a> {
    labels: &'a [Vec<String>],
    key: &'a str,
    lo: Option<&'a LoraValue>,
    lo_inclusive: bool,
    hi: Option<&'a LoraValue>,
    hi_inclusive: bool,
}

fn node_matches_range_filter<S: GraphStorage>(
    storage: &S,
    id: NodeId,
    filter: &NodeRangeFilter<'_>,
) -> bool {
    storage
        .with_node(id, |n| {
            if !node_matches_label_groups(&n.labels, filter.labels) {
                return false;
            }
            let Some(actual) = n.properties.get(filter.key) else {
                return false;
            };
            let actual_lv = lora_store_property_to_value(actual);
            range_predicate_holds(
                &actual_lv,
                filter.lo,
                filter.lo_inclusive,
                filter.hi,
                filter.hi_inclusive,
            )
        })
        .unwrap_or(false)
}

fn node_matches_text_filter<S: GraphStorage>(
    storage: &S,
    id: NodeId,
    labels: &[Vec<String>],
    key: &str,
    predicate: lora_compiler::TextPredicate,
    query: &str,
) -> bool {
    storage
        .with_node(id, |n| {
            if !node_matches_label_groups(&n.labels, labels) {
                return false;
            }
            let Some(PropertyValue::String(actual)) = n.properties.get(key) else {
                return false;
            };
            text_predicate_holds(actual, predicate, query)
        })
        .unwrap_or(false)
}

fn text_predicate_holds(
    actual: &str,
    predicate: lora_compiler::TextPredicate,
    query: &str,
) -> bool {
    match predicate {
        lora_compiler::TextPredicate::StartsWith => actual.starts_with(query),
        lora_compiler::TextPredicate::EndsWith => actual.ends_with(query),
        lora_compiler::TextPredicate::Contains => actual.contains(query),
    }
}

fn range_predicate_holds(
    actual: &LoraValue,
    lo: Option<&LoraValue>,
    lo_inclusive: bool,
    hi: Option<&LoraValue>,
    hi_inclusive: bool,
) -> bool {
    if let Some(lo) = lo {
        match range_comparison(actual, lo) {
            None => return false,
            Some(Ordering::Less) => return false,
            Some(Ordering::Equal) if !lo_inclusive => return false,
            _ => {}
        }
    }
    if let Some(hi) = hi {
        match range_comparison(actual, hi) {
            None => return false,
            Some(Ordering::Greater) => return false,
            Some(Ordering::Equal) if !hi_inclusive => return false,
            _ => {}
        }
    }
    true
}

fn range_comparison(actual: &LoraValue, bound: &LoraValue) -> Option<Ordering> {
    match (actual, bound) {
        (LoraValue::Null, _) | (_, LoraValue::Null) => None,
        (LoraValue::String(a), LoraValue::String(b)) => Some(a.cmp(b)),
        (
            LoraValue::Date(_)
            | LoraValue::DateTime(_)
            | LoraValue::LocalDateTime(_)
            | LoraValue::Time(_)
            | LoraValue::LocalTime(_),
            _,
        ) => actual.temporal_cmp(bound),
        (LoraValue::Duration(a), LoraValue::Duration(b)) => a
            .total_seconds_approx()
            .partial_cmp(&b.total_seconds_approx()),
        _ => actual.as_f64()?.partial_cmp(&bound.as_f64()?),
    }
}

fn lora_store_property_to_value(value: &PropertyValue) -> LoraValue {
    LoraValue::from(value)
}

pub(crate) fn node_by_point_scan_rows<S: GraphStorage>(
    storage: &S,
    params: &BTreeMap<String, LoraValue>,
    base_rows: Vec<Row>,
    op: &lora_compiler::NodeByPointScanExec,
    deadline: Option<Instant>,
) -> ExecResult<Vec<Row>> {
    let eval_ctx = EvalContext { storage, params };
    let mut out = Vec::new();

    for row in base_rows {
        check_optional_deadline(deadline)?;

        // Resolve the predicate's literal/scalar inputs against the
        // current row. These end up as captured `LoraValue`s used both
        // to probe the spatial index and to refilter every candidate.
        let probe = match &op.predicate {
            lora_compiler::PointPredicate::WithinBBox {
                lower_left,
                upper_right,
            } => {
                let ll = eval_expr(lower_left, &row, &eval_ctx);
                let ur = eval_expr(upper_right, &row, &eval_ctx);
                match (ll, ur) {
                    (LoraValue::Point(a), LoraValue::Point(b)) => {
                        Probe::WithinBBox { ll: a, ur: b }
                    }
                    _ => continue,
                }
            }
            lora_compiler::PointPredicate::WithinDistance {
                center,
                max_distance,
                inclusive,
            } => {
                let c = eval_expr(center, &row, &eval_ctx);
                let d = eval_expr(max_distance, &row, &eval_ctx);
                match (c, d) {
                    (LoraValue::Point(c), LoraValue::Float(d)) => Probe::WithinDistance {
                        center: c,
                        max: d,
                        inclusive: *inclusive,
                    },
                    (LoraValue::Point(c), LoraValue::Int(d)) => Probe::WithinDistance {
                        center: c,
                        max: d as f64,
                        inclusive: *inclusive,
                    },
                    _ => continue,
                }
            }
        };

        if let Some(existing_id) = bound_node_id_for_expand(&row, op.var)? {
            if node_matches_point_filter(storage, existing_id, &op.labels, &op.key, &probe) {
                out.push(row);
            }
            continue;
        }

        let candidate_ids = match single_label_hint(&op.labels) {
            Some(label) => match &probe {
                Probe::WithinBBox { ll, ur } => {
                    // Two seeks when the box crosses the antimeridian.
                    let (ranges, n) = lora_store::bbox_x_ranges(ll, ur);
                    let (lo_y, hi_y) = (ll.y.min(ur.y), ll.y.max(ur.y));
                    let mut ids = Vec::new();
                    let mut indexed = true;
                    for (lo_x, hi_x) in &ranges[..n] {
                        match storage.node_point_within_bbox(
                            label,
                            &op.key,
                            (*lo_x, lo_y),
                            (*hi_x, hi_y),
                        ) {
                            Some(found) => ids.extend(found),
                            None => indexed = false,
                        }
                    }
                    if !indexed {
                        scan_node_ids_for_label_groups(storage, &op.labels)
                    } else {
                        if n > 1 {
                            ids.sort_unstable();
                            ids.dedup();
                        }
                        ids
                    }
                }
                Probe::WithinDistance { center, max, .. } => storage
                    .node_point_within_distance(label, &op.key, (center.x, center.y), *max)
                    .unwrap_or_else(|| scan_node_ids_for_label_groups(storage, &op.labels)),
            },
            None => scan_node_ids_for_label_groups(storage, &op.labels),
        };

        for id in candidate_ids {
            check_optional_deadline(deadline)?;
            if node_matches_point_filter(storage, id, &op.labels, &op.key, &probe) {
                let mut new_row = row.clone();
                new_row.insert(op.var, LoraValue::Node(id));
                out.push(new_row);
            }
        }
    }

    Ok(out)
}

fn node_matches_point_filter<S: GraphStorage>(
    storage: &S,
    id: NodeId,
    labels: &[Vec<String>],
    key: &str,
    probe: &Probe,
) -> bool {
    storage
        .with_node(id, |n| {
            if !node_matches_label_groups(&n.labels, labels) {
                return false;
            }
            let Some(PropertyValue::Point(point)) = n.properties.get(key) else {
                return false;
            };
            point_predicate_holds(point, probe)
        })
        .unwrap_or(false)
}

enum Probe {
    WithinBBox {
        ll: lora_store::LoraPoint,
        ur: lora_store::LoraPoint,
    },
    WithinDistance {
        center: lora_store::LoraPoint,
        max: f64,
        inclusive: bool,
    },
}

fn point_predicate_holds(actual: &lora_store::LoraPoint, probe: &Probe) -> bool {
    match probe {
        Probe::WithinBBox { ll, ur } => lora_store::bbox_contains(actual, ll, ur).unwrap_or(false),
        Probe::WithinDistance {
            center,
            max,
            inclusive,
        } => {
            let Some(d) = lora_store::point_distance(actual, center) else {
                return false;
            };
            if *inclusive {
                d <= *max
            } else {
                d < *max
            }
        }
    }
}

pub(crate) fn indexed_node_property_candidates<S: GraphStorage>(
    storage: &S,
    labels: &[Vec<String>],
    key: &str,
    expected: &LoraValue,
) -> NodePropertyCandidates {
    let Some(values) = property_lookup_values(expected) else {
        return NodePropertyCandidates {
            ids: scan_node_ids_for_label_groups(storage, labels),
            prefiltered: false,
        };
    };

    let label_hint = single_label_hint(labels);
    let mut seen = BTreeSet::new();
    let mut out = Vec::new();
    for value in values {
        for id in storage.find_node_ids_by_property(label_hint, key, &value) {
            if seen.insert(id) {
                out.push(id);
            }
        }
    }
    NodePropertyCandidates {
        ids: out,
        prefiltered: labels.is_empty() || label_hint.is_some(),
    }
}

/// Candidate ids for a `NodeByPropertyScan`. With `in_list` the scan
/// seeks `key IN expected`: one lookup per distinct list element, the
/// union deduplicated. A `null` list matches nothing. Any other non-list
/// value yields every labelled node unfiltered, so the `Filter` kept
/// above the scan evaluates (and reports) the predicate itself.
pub(crate) fn property_scan_candidates<S: GraphStorage>(
    storage: &S,
    labels: &[Vec<String>],
    key: &str,
    expected: &LoraValue,
    in_list: bool,
) -> NodePropertyCandidates {
    if !in_list {
        return indexed_node_property_candidates(storage, labels, key, expected);
    }
    match expected {
        LoraValue::Null => NodePropertyCandidates {
            ids: Vec::new(),
            prefiltered: true,
        },
        LoraValue::List(items) => {
            let mut seen = BTreeSet::new();
            let mut ids = Vec::new();
            for item in items {
                if matches!(item, LoraValue::Null) {
                    continue;
                }
                let candidates = indexed_node_property_candidates(storage, labels, key, item);
                for id in candidates.ids {
                    if seen.contains(&id) {
                        continue;
                    }
                    if !candidates.prefiltered
                        && !node_matches_property_filter(storage, id, labels, key, item)
                    {
                        continue;
                    }
                    seen.insert(id);
                    ids.push(id);
                }
            }
            NodePropertyCandidates {
                ids,
                prefiltered: true,
            }
        }
        _ => NodePropertyCandidates {
            ids: scan_node_ids_for_label_groups(storage, labels)
                .into_iter()
                .filter(|&id| {
                    storage
                        .with_node(id, |n| node_matches_label_groups(&n.labels, labels))
                        .unwrap_or(false)
                })
                .collect(),
            prefiltered: true,
        },
    }
}

/// Whether `node_id` passes a `NodeByPropertyScan` (labels plus
/// `key = expected`, or `key IN expected` when `in_list`). Mirrors
/// [`property_scan_candidates`], including its non-list fallback.
pub(crate) fn property_scan_matches<S: GraphStorage>(
    storage: &S,
    node_id: NodeId,
    labels: &[Vec<String>],
    key: &str,
    expected: &LoraValue,
    in_list: bool,
) -> bool {
    if !in_list {
        return node_matches_property_filter(storage, node_id, labels, key, expected);
    }
    match expected {
        LoraValue::Null => false,
        LoraValue::List(items) => items.iter().any(|item| {
            !matches!(item, LoraValue::Null)
                && node_matches_property_filter(storage, node_id, labels, key, item)
        }),
        _ => storage
            .with_node(node_id, |n| node_matches_label_groups(&n.labels, labels))
            .unwrap_or(false),
    }
}

/// Build a LoraPath from the node and relationship variables currently in a row.
///
/// For variable-length relationships (stored as a List of Relationship values),
/// intermediate nodes are reconstructed from the storage by walking the
/// relationship chain.
pub(crate) fn build_path_value<S: GraphStorage>(
    row: &Row,
    node_vars: &[VarId],
    rel_vars: &[VarId],
    storage: &S,
) -> LoraValue {
    let (raw_nodes, rels, has_var_len) = path_bindings(row, node_vars, rel_vars);

    let nodes = if has_var_len && !rels.is_empty() && raw_nodes.len() == 2 {
        reconstruct_var_len_nodes(raw_nodes[0], &rels, storage)
    } else {
        raw_nodes
    };

    LoraValue::Path(LoraPath { nodes, rels })
}

#[inline]
fn path_bindings(
    row: &Row,
    node_vars: &[VarId],
    rel_vars: &[VarId],
) -> (Vec<NodeId>, Vec<RelationshipId>, bool) {
    let mut raw_nodes = Vec::new();
    let mut rels = Vec::new();
    let mut has_var_len = false;

    for &nv in node_vars {
        match row.get(nv) {
            Some(LoraValue::Node(id)) => raw_nodes.push(*id),
            Some(LoraValue::List(items)) => {
                for item in items {
                    if let LoraValue::Node(id) = item {
                        raw_nodes.push(*id);
                    }
                }
            }
            _ => {}
        }
    }

    for &rv in rel_vars {
        match row.get(rv) {
            Some(LoraValue::Relationship(id)) => rels.push(*id),
            Some(LoraValue::List(items)) => {
                has_var_len = true;
                for item in items {
                    if let LoraValue::Relationship(id) = item {
                        rels.push(*id);
                    }
                }
            }
            _ => {}
        }
    }

    (raw_nodes, rels, has_var_len)
}

#[inline]
fn reconstruct_var_len_nodes<S: GraphStorage>(
    start: NodeId,
    rels: &[RelationshipId],
    storage: &S,
) -> Vec<NodeId> {
    let mut ordered = Vec::with_capacity(rels.len() + 1);
    ordered.push(start);
    let mut current = start;
    for &rel_id in rels {
        if let Some((src, dst)) = storage.relationship_endpoints(rel_id) {
            let next = if src == current { dst } else { src };
            ordered.push(next);
            current = next;
        }
    }
    ordered
}

fn type_rank(v: &LoraValue) -> u8 {
    match v {
        LoraValue::Null => 0,
        LoraValue::Bool(_) => 1,
        LoraValue::Int(_) | LoraValue::Float(_) => 2,
        LoraValue::String(_) => 3,
        LoraValue::Binary(_) => 4,
        LoraValue::Date(_) => 5,
        LoraValue::DateTime(_) => 6,
        LoraValue::LocalDateTime(_) => 7,
        LoraValue::Time(_) => 8,
        LoraValue::LocalTime(_) => 9,
        LoraValue::Duration(_) => 10,
        LoraValue::Point(_) => 11,
        LoraValue::Vector(_) => 12,
        LoraValue::List(_) => 13,
        LoraValue::Map(_) => 14,
        LoraValue::Node(_) => 15,
        LoraValue::Relationship(_) => 16,
        LoraValue::Path(_) => 17,
    }
}

/// Check whether a node's labels satisfy all label groups.
/// Each group is a disjunction (OR): the node must have at least one label
/// from the group.  Groups are conjunctive (AND): all groups must be satisfied.
pub(crate) fn node_matches_label_groups(node_labels: &[String], groups: &[Vec<String>]) -> bool {
    groups
        .iter()
        .all(|group| group.iter().any(|l| node_labels.iter().any(|nl| nl == l)))
}

/// Scan the graph for candidate node IDs matching the label groups. Uses the
/// label index for the pick-first-label phase and avoids cloning NodeRecords.
pub(crate) fn scan_node_ids_for_label_groups<S: GraphStorage>(
    storage: &S,
    groups: &[Vec<String>],
) -> Vec<NodeId> {
    if groups.is_empty() {
        return storage.all_node_ids();
    }
    if groups.len() == 1 {
        return label_group_candidate_ids(storage, &groups[0]);
    }

    let mut best: Option<Vec<NodeId>> = None;
    for group in groups {
        let ids = label_group_candidate_ids(storage, group);
        if ids.is_empty() {
            return Vec::new();
        }
        if best
            .as_ref()
            .map(|current| ids.len() < current.len())
            .unwrap_or(true)
        {
            best = Some(ids);
        }
    }

    best.unwrap_or_default()
}

pub(crate) fn label_group_candidates_prefiltered(groups: &[Vec<String>]) -> bool {
    groups.len() <= 1
}

fn label_group_candidate_ids<S: GraphStorage>(storage: &S, group: &[String]) -> Vec<NodeId> {
    match group {
        [] => Vec::new(),
        [label] => storage.node_ids_by_label(label),
        labels => {
            let mut seen = BTreeSet::new();
            let mut out = Vec::new();
            for label in labels {
                for id in storage.node_ids_by_label(label) {
                    if seen.insert(id) {
                        out.push(id);
                    }
                }
            }
            out
        }
    }
}

pub(crate) fn hydrate_node_record(node: &lora_store::NodeRecord) -> LoraValue {
    let mut map = BTreeMap::new();
    map.insert("kind".to_string(), LoraValue::String("node".to_string()));
    map.insert("id".to_string(), LoraValue::Int(node.id as i64));
    map.insert(
        "labels".to_string(),
        LoraValue::List(
            node.labels
                .iter()
                .map(|s| LoraValue::String(s.clone()))
                .collect(),
        ),
    );
    map.insert(
        "properties".to_string(),
        properties_to_value_map(&node.properties),
    );
    LoraValue::Map(map)
}

pub(crate) fn hydrate_relationship_record(rel: &lora_store::RelationshipRecord) -> LoraValue {
    let mut map = BTreeMap::new();
    map.insert(
        "kind".to_string(),
        LoraValue::String("relationship".to_string()),
    );
    map.insert("id".to_string(), LoraValue::Int(rel.id as i64));
    map.insert("startId".to_string(), LoraValue::Int(rel.src as i64));
    map.insert("endId".to_string(), LoraValue::Int(rel.dst as i64));
    map.insert("type".to_string(), LoraValue::String(rel.rel_type.clone()));
    map.insert(
        "properties".to_string(),
        properties_to_value_map(&rel.properties),
    );
    LoraValue::Map(map)
}

/// Flatten label groups into a simple Vec<String> (for CREATE/MERGE where
/// disjunction doesn't apply — all labels are created).
pub(super) fn flatten_label_groups(groups: &[Vec<String>]) -> Vec<String> {
    groups.iter().flat_map(|g| g.iter().cloned()).collect()
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub(crate) enum GroupValueKey {
    Null,
    Bool(bool),
    Int(i64),
    Float(String),
    String(String),
    Binary(Vec<Vec<u8>>),
    List(Vec<GroupValueKey>),
    Map(Vec<(String, GroupValueKey)>),
    Node(u64),
    Relationship(u64),
}

impl GroupValueKey {
    pub(crate) fn from_value(v: &LoraValue) -> Self {
        match v {
            LoraValue::Null => Self::Null,
            LoraValue::Bool(x) => Self::Bool(*x),
            LoraValue::Int(x) => Self::Int(*x),
            LoraValue::Float(x) => Self::Float(x.to_string()),
            LoraValue::String(x) => Self::String(x.clone()),
            LoraValue::Binary(x) => Self::Binary(x.segments().to_vec()),
            LoraValue::List(xs) => Self::List(xs.iter().map(Self::from_value).collect()),
            LoraValue::Map(m) => Self::Map(
                m.iter()
                    .map(|(k, v)| (k.clone(), Self::from_value(v)))
                    .collect(),
            ),
            LoraValue::Node(id) => Self::Node(*id),
            LoraValue::Relationship(id) => Self::Relationship(*id),
            LoraValue::Path(_) => Self::Null,
            // Temporal types: use their string representation as group key
            LoraValue::Date(d) => Self::String(d.to_string()),
            LoraValue::DateTime(dt) => Self::String(dt.to_string()),
            LoraValue::LocalDateTime(dt) => Self::String(dt.to_string()),
            LoraValue::Time(t) => Self::String(t.to_string()),
            LoraValue::LocalTime(t) => Self::String(t.to_string()),
            LoraValue::Duration(dur) => Self::String(dur.to_string()),
            LoraValue::Point(p) => Self::String(p.to_string()),
            LoraValue::Vector(v) => Self::String(format!("vector:{}", v.to_key_string())),
        }
    }
}

/// Compute effective (min_hops, max_hops) from a `RangeLiteral`.
///
/// Lora semantics:
/// - `*`       → 1..∞   (start=None, end=None)
/// - `*2..5`   → 2..5   (start=Some(2), end=Some(5))
/// - `*..3`    → 1..3   (start=None, end=Some(3))
/// - `*2..`    → 2..∞   (start=Some(2), end=None)
/// - `*3`      → 3..3   (start=Some(3), end=None, no dots → exactly 3)
/// - `*0..1`   → 0..1
///
/// For unbounded upper, we cap at `MAX_VAR_LEN_HOPS` to prevent runaway.
const MAX_VAR_LEN_HOPS: u64 = 100;

pub(crate) fn resolve_range(range: &RangeLiteral) -> (u64, u64) {
    let min_hops = range.start.unwrap_or(1);
    let max_hops = range.end.unwrap_or(MAX_VAR_LEN_HOPS);
    (min_hops, max_hops)
}

/// An entry produced during BFS variable-length expansion.
pub(crate) struct VarLenResult {
    /// The destination node at the end of this path.
    pub(crate) dst_node_id: NodeId,
    /// The relationship IDs traversed (in order).
    pub(crate) rel_ids: Vec<u64>,
}

/// Perform variable-length expansion from `start_node_id` following
/// relationships of the given `types` and `direction`, collecting all
/// reachable nodes at hop distances in `[min_hops, max_hops]`.
///
/// Uses BFS with relationship-uniqueness per path (each path does not
/// reuse the same relationship, but may revisit nodes).
pub(crate) fn variable_length_expand<S: GraphStorage>(
    storage: &S,
    start_node_id: NodeId,
    direction: Direction,
    types: &[String],
    min_hops: u64,
    max_hops: u64,
    bind_relationships: bool,
) -> Vec<VarLenResult> {
    let mut results = Vec::new();

    // Each frontier entry: (current_node_id, relationships_used_so_far)
    let mut frontier: Vec<(NodeId, Vec<u64>)> = vec![(start_node_id, Vec::new())];

    for depth in 1..=max_hops {
        // On the final hop we don't need to build next_frontier at all; every
        // path gets recorded and then the loop terminates. Avoids one full
        // pass of Vec clones on deep traversals.
        let is_last_hop = depth == max_hops;
        let mut next_frontier: Vec<(NodeId, Vec<u64>)> = Vec::new();

        for (current_node, rels_used) in &frontier {
            // ID-only expand avoids cloning full records/properties for every
            // neighbour on every hop.
            for (rel_id, neighbor_id) in storage.expand_ids(*current_node, direction, types) {
                // Relationship-uniqueness: skip if this relationship was already
                // traversed on this particular path.
                if rels_used.contains(&rel_id) {
                    continue;
                }

                if is_last_hop {
                    // Terminal hop: just record the result. Allocate rel_ids
                    // once (no duplicate clone) by extending a fresh copy.
                    if depth >= min_hops {
                        let mut rel_ids = Vec::with_capacity(rels_used.len() + 1);
                        rel_ids.extend_from_slice(rels_used);
                        rel_ids.push(rel_id);
                        results.push(VarLenResult {
                            dst_node_id: neighbor_id,
                            rel_ids: if bind_relationships {
                                rel_ids
                            } else {
                                Vec::new()
                            },
                        });
                    }
                    continue;
                }

                let mut new_rels = Vec::with_capacity(rels_used.len() + 1);
                new_rels.extend_from_slice(rels_used);
                new_rels.push(rel_id);

                if depth >= min_hops {
                    results.push(VarLenResult {
                        dst_node_id: neighbor_id,
                        rel_ids: if bind_relationships {
                            new_rels.clone()
                        } else {
                            Vec::new()
                        },
                    });
                }

                next_frontier.push((neighbor_id, new_rels));
            }
        }

        if is_last_hop || next_frontier.is_empty() {
            break;
        }

        frontier = next_frontier;
    }

    // Handle min_hops == 0: include the start node itself at depth 0.
    if min_hops == 0 {
        results.insert(
            0,
            VarLenResult {
                dst_node_id: start_node_id,
                rel_ids: Vec::new(),
            },
        );
    }

    results
}

/// Filter rows to keep only shortest paths.
/// `all` = false → keep one shortest path; `all` = true → keep all shortest.
pub(crate) fn filter_shortest_paths(rows: Vec<Row>, path_var: VarId, all: bool) -> Vec<Row> {
    if rows.is_empty() {
        return rows;
    }

    // Compute path length for each row
    let lengths: Vec<usize> = rows
        .iter()
        .map(|row| match row.get(path_var) {
            Some(LoraValue::Path(p)) => p.rels.len(),
            _ => usize::MAX,
        })
        .collect();

    let min_len = lengths.iter().copied().min().unwrap_or(usize::MAX);

    let mut result: Vec<Row> = rows
        .into_iter()
        .zip(lengths.iter())
        .filter(|(_, len)| **len == min_len)
        .map(|(row, _)| row)
        .collect();

    if !all && result.len() > 1 {
        result.truncate(1);
    }

    result
}

// ---------- Rel scans ----------
//
// Mirror of the `node_by_*_scan_rows` helpers above for the
// relationship-targeted index operators. Each helper resolves the
// candidate set via the corresponding `relationship_*_candidates`
// trait method (falling back to `rel_ids_by_type` when no scope is
// active), refilters with the precise predicate, and emits one row
// per matching relationship per input row.
//
// Direction handling: the optimizer only emits a Rel*Scan when both
// endpoints of the pattern have no upstream constraint. With a
// directed pattern (Direction::Right), each rel is bound as
// (src=stored_src, rel, dst=stored_dst). With Direction::Left those
// are swapped. With Direction::Undirected we emit both orientations
// to preserve parity with the Expand-based plan it replaces.

fn emit_rel_rows(
    direction: Direction,
    src_var: VarId,
    rel_var: VarId,
    dst_var: VarId,
    rel: &lora_store::RelationshipRecord,
    base: &Row,
    out: &mut Vec<Row>,
) -> ExecResult<()> {
    match direction {
        Direction::Right => {
            emit_one_rel_row(
                src_var, rel_var, dst_var, rel.src, rel.id, rel.dst, base, out,
            )?;
        }
        Direction::Left => {
            emit_one_rel_row(
                src_var, rel_var, dst_var, rel.dst, rel.id, rel.src, base, out,
            )?;
        }
        Direction::Undirected => {
            emit_one_rel_row(
                src_var, rel_var, dst_var, rel.src, rel.id, rel.dst, base, out,
            )?;
            // Self-loops produce a single row even under undirected
            // semantics — emitting two copies of (a=a, b=a) for
            // a=stored_src=stored_dst would double-count.
            if rel.src != rel.dst {
                emit_one_rel_row(
                    src_var, rel_var, dst_var, rel.dst, rel.id, rel.src, base, out,
                )?;
            }
        }
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
fn emit_one_rel_row(
    src_var: VarId,
    rel_var: VarId,
    dst_var: VarId,
    src_id: NodeId,
    rel_id: RelationshipId,
    dst_id: NodeId,
    base: &Row,
    out: &mut Vec<Row>,
) -> ExecResult<()> {
    let mut row = base.clone();
    if bind_node_value(&mut row, src_var, src_id)?
        && bind_relationship_value(&mut row, rel_var, rel_id)?
        && bind_node_value(&mut row, dst_var, dst_id)?
    {
        out.push(row);
    }
    Ok(())
}

fn bind_node_value(row: &mut Row, var: VarId, id: NodeId) -> ExecResult<bool> {
    match row.get(var) {
        Some(LoraValue::Node(existing)) => Ok(*existing == id),
        Some(other) => Err(ExecutorError::ExpectedNodeForExpand {
            var: format!("{var:?}"),
            found: value_kind(other),
        }),
        None => {
            row.insert(var, LoraValue::Node(id));
            Ok(true)
        }
    }
}

fn bind_relationship_value(row: &mut Row, var: VarId, id: RelationshipId) -> ExecResult<bool> {
    match row.get(var) {
        Some(LoraValue::Relationship(existing)) => Ok(*existing == id),
        Some(other) => Err(ExecutorError::ExpectedRelationshipForExpand {
            var: format!("{var:?}"),
            found: value_kind(other),
        }),
        None => {
            row.insert(var, LoraValue::Relationship(id));
            Ok(true)
        }
    }
}

fn rel_candidate_ids<S, F>(storage: &S, types: &[String], indexed: F) -> Vec<RelationshipId>
where
    S: GraphStorage,
    F: Fn(&str) -> Option<Vec<RelationshipId>>,
{
    if types.is_empty() {
        // No type constraint → no rel-typed scope to probe; fall back
        // to scanning every relationship.
        return storage.all_rel_ids();
    }
    let mut all = Vec::new();
    let mut seen = BTreeSet::new();
    for ty in types {
        let ids = indexed(ty).unwrap_or_else(|| storage.rel_ids_by_type(ty));
        for id in ids {
            if seen.insert(id) {
                all.push(id);
            }
        }
    }
    all
}

pub(crate) fn rel_by_property_range_scan_rows<S: GraphStorage>(
    storage: &S,
    params: &BTreeMap<String, LoraValue>,
    base_rows: Vec<Row>,
    op: &lora_compiler::RelByPropertyRangeScanExec,
    deadline: Option<Instant>,
) -> ExecResult<Vec<Row>> {
    let eval_ctx = EvalContext { storage, params };
    let mut out = Vec::new();
    let mut other_kinds = OtherKindScan::default();

    for row in base_rows {
        check_optional_deadline(deadline)?;
        let lo_value = op.lo.as_ref().map(|expr| eval_expr(expr, &row, &eval_ctx));
        // A null upper bound reads as none: the Filter kept above every
        // index scan still judges each row (a STARTS WITH prefix with no
        // successor, `string.prefix_end` null, scans to the end).
        let hi_value = op
            .hi
            .as_ref()
            .map(|expr| eval_expr(expr, &row, &eval_ctx))
            .filter(|v| !matches!(v, LoraValue::Null));
        let lo_prop = lo_value
            .clone()
            .and_then(|v| lora_value_to_property(v).ok());
        let hi_prop = hi_value
            .clone()
            .and_then(|v| lora_value_to_property(v).ok());

        // Temporals of another kind first, for the Filter above to judge
        // (see `other_kind_node_ids`).
        let bounds = [lo_value.as_ref(), hi_value.as_ref()];
        for rel_id in other_kind_rel_ids(storage, op, bounds, &mut other_kinds) {
            if let Some(result) = storage.with_relationship(rel_id, |rel| {
                if !op.types.is_empty() && !op.types.iter().any(|t| t == &rel.rel_type) {
                    return Ok(());
                }
                emit_rel_rows(op.direction, op.src, op.rel, op.dst, rel, &row, &mut out)
            }) {
                result?;
            }
        }
        let candidate_ids = rel_candidate_ids(storage, &op.types, |ty| {
            storage.relationship_range_candidates(ty, &op.key, lo_prop.as_ref(), hi_prop.as_ref())
        });

        for rel_id in candidate_ids {
            check_optional_deadline(deadline)?;
            if let Some(result) = storage.with_relationship(rel_id, |rel| {
                if !op.types.is_empty() && !op.types.iter().any(|t| t == &rel.rel_type) {
                    return Ok(());
                }
                let Some(actual) = rel.properties.get(op.key.as_str()) else {
                    return Ok(());
                };
                let actual_lv = LoraValue::from(actual);
                if !range_predicate_holds(
                    &actual_lv,
                    lo_value.as_ref(),
                    op.lo_inclusive,
                    hi_value.as_ref(),
                    op.hi_inclusive,
                ) {
                    return Ok(());
                }
                emit_rel_rows(op.direction, op.src, op.rel, op.dst, rel, &row, &mut out)
            }) {
                result?;
            }
        }
    }

    Ok(out)
}

pub(crate) fn rel_by_text_scan_rows<S: GraphStorage>(
    storage: &S,
    params: &BTreeMap<String, LoraValue>,
    base_rows: Vec<Row>,
    op: &lora_compiler::RelByTextScanExec,
    deadline: Option<Instant>,
) -> ExecResult<Vec<Row>> {
    let eval_ctx = EvalContext { storage, params };
    let mut out = Vec::new();

    for row in base_rows {
        check_optional_deadline(deadline)?;
        let query = eval_expr(&op.query, &row, &eval_ctx);
        let LoraValue::String(query_str) = &query else {
            continue;
        };

        let candidate_ids = rel_candidate_ids(storage, &op.types, |ty| {
            storage.relationship_text_candidates(ty, &op.key, query_str)
        });

        for rel_id in candidate_ids {
            check_optional_deadline(deadline)?;
            if let Some(result) = storage.with_relationship(rel_id, |rel| {
                if !op.types.is_empty() && !op.types.iter().any(|t| t == &rel.rel_type) {
                    return Ok(());
                }
                let Some(PropertyValue::String(actual)) = rel.properties.get(op.key.as_str())
                else {
                    return Ok(());
                };
                if !text_predicate_holds(actual, op.predicate, query_str) {
                    return Ok(());
                }
                emit_rel_rows(op.direction, op.src, op.rel, op.dst, rel, &row, &mut out)
            }) {
                result?;
            }
        }
    }

    Ok(out)
}

pub(crate) fn rel_by_point_scan_rows<S: GraphStorage>(
    storage: &S,
    params: &BTreeMap<String, LoraValue>,
    base_rows: Vec<Row>,
    op: &lora_compiler::RelByPointScanExec,
    deadline: Option<Instant>,
) -> ExecResult<Vec<Row>> {
    let eval_ctx = EvalContext { storage, params };
    let mut out = Vec::new();

    for row in base_rows {
        check_optional_deadline(deadline)?;

        let probe = match &op.predicate {
            lora_compiler::PointPredicate::WithinBBox {
                lower_left,
                upper_right,
            } => {
                let ll = eval_expr(lower_left, &row, &eval_ctx);
                let ur = eval_expr(upper_right, &row, &eval_ctx);
                match (ll, ur) {
                    (LoraValue::Point(a), LoraValue::Point(b)) => {
                        Probe::WithinBBox { ll: a, ur: b }
                    }
                    _ => continue,
                }
            }
            lora_compiler::PointPredicate::WithinDistance {
                center,
                max_distance,
                inclusive,
            } => {
                let c = eval_expr(center, &row, &eval_ctx);
                let d = eval_expr(max_distance, &row, &eval_ctx);
                match (c, d) {
                    (LoraValue::Point(c), LoraValue::Float(d)) => Probe::WithinDistance {
                        center: c,
                        max: d,
                        inclusive: *inclusive,
                    },
                    (LoraValue::Point(c), LoraValue::Int(d)) => Probe::WithinDistance {
                        center: c,
                        max: d as f64,
                        inclusive: *inclusive,
                    },
                    _ => continue,
                }
            }
        };

        let candidate_ids = rel_candidate_ids(storage, &op.types, |ty| match &probe {
            Probe::WithinBBox { ll, ur } => {
                // Two seeks when the box crosses the antimeridian.
                let (ranges, n) = lora_store::bbox_x_ranges(ll, ur);
                let (lo_y, hi_y) = (ll.y.min(ur.y), ll.y.max(ur.y));
                let mut ids = Vec::new();
                for (lo_x, hi_x) in &ranges[..n] {
                    ids.extend(storage.relationship_point_within_bbox(
                        ty,
                        &op.key,
                        (*lo_x, lo_y),
                        (*hi_x, hi_y),
                    )?);
                }
                if n > 1 {
                    ids.sort_unstable();
                    ids.dedup();
                }
                Some(ids)
            }
            Probe::WithinDistance { center, max, .. } => {
                storage.relationship_point_within_distance(ty, &op.key, (center.x, center.y), *max)
            }
        });

        for rel_id in candidate_ids {
            check_optional_deadline(deadline)?;
            if let Some(result) = storage.with_relationship(rel_id, |rel| {
                if !op.types.is_empty() && !op.types.iter().any(|t| t == &rel.rel_type) {
                    return Ok(());
                }
                let Some(PropertyValue::Point(actual)) = rel.properties.get(op.key.as_str()) else {
                    return Ok(());
                };
                if !point_predicate_holds(actual, &probe) {
                    return Ok(());
                }
                emit_rel_rows(op.direction, op.src, op.rel, op.dst, rel, &row, &mut out)
            }) {
                result?;
            }
        }
    }

    Ok(out)
}

/// Node ids of an ordered range scan (`op.order` set), in `ORDER BY op.key`
/// order, each already checked against the exact bounds and labels.
///
/// The index serves the order when every bound is a string: a string
/// comparison only lets strings through, and among strings the index's
/// order is Cypher's order. Ids are then pulled from the index in chunks
/// that grow as rows are consumed, so a `LIMIT` above stops the scan
/// after a few small index reads. For any other bound the index order is
/// not Cypher's (it orders every integer before every float), so the
/// cursor collects the candidates and sorts them by value itself.
pub(crate) struct OrderedRangeCursor {
    lo: Option<LoraValue>,
    hi: Option<LoraValue>,
    mode: OrderedMode,
    pending: std::vec::IntoIter<NodeId>,
    labels: Vec<Vec<String>>,
    key: String,
    lo_inclusive: bool,
    hi_inclusive: bool,
}

enum OrderedMode {
    Index {
        label: String,
        descending: bool,
        after: Option<(PropertyValue, NodeId)>,
        chunk: usize,
        exhausted: bool,
    },
    Buffered,
}

impl OrderedRangeCursor {
    pub(crate) fn new<S: GraphStorage>(
        storage: &S,
        op: &lora_compiler::NodeByPropertyRangeScanExec,
        lo: Option<LoraValue>,
        hi: Option<LoraValue>,
    ) -> Self {
        let descending = matches!(op.order, Some(SortDirection::Desc));
        // Strings and temporals of one kind have a single contiguous,
        // correctly ordered run in the index, so the walk can stream from
        // it. Numbers cannot: integers and floats are keyed separately.
        let mut bounds = lo.iter().chain(hi.iter());
        let string_bounds = match bounds.next() {
            Some(LoraValue::String(_)) => bounds.all(|v| matches!(v, LoraValue::String(_))),
            Some(
                first @ (LoraValue::Date(_)
                | LoraValue::DateTime(_)
                | LoraValue::LocalDateTime(_)
                | LoraValue::Time(_)
                | LoraValue::LocalTime(_)),
            ) => bounds.all(|v| std::mem::discriminant(v) == std::mem::discriminant(first)),
            _ => false,
        };
        let index_label = single_label_hint(&op.labels).filter(|label| {
            string_bounds
                && storage
                    .node_range_ordered_chunk(label, &op.key, None, None, descending, None, 0)
                    .is_some()
        });

        let mut cursor = Self {
            lo,
            hi,
            mode: OrderedMode::Buffered,
            pending: Vec::new().into_iter(),
            labels: op.labels.clone(),
            key: op.key.clone(),
            lo_inclusive: op.lo_inclusive,
            hi_inclusive: op.hi_inclusive,
        };
        match index_label {
            Some(label) => {
                cursor.mode = OrderedMode::Index {
                    label: label.to_string(),
                    descending,
                    after: None,
                    chunk: 64,
                    exhausted: false,
                }
            }
            None => {
                cursor.pending = cursor
                    .sorted_candidates(storage, op, descending)
                    .into_iter()
            }
        }
        cursor
    }

    pub(crate) fn filter(&self) -> NodeRangeFilter<'_> {
        NodeRangeFilter {
            labels: &self.labels,
            key: &self.key,
            lo: self.lo.as_ref(),
            lo_inclusive: self.lo_inclusive,
            hi: self.hi.as_ref(),
            hi_inclusive: self.hi_inclusive,
        }
    }

    /// Fallback: every matching id, sorted by the property value in Cypher
    /// order (ties by id).
    fn sorted_candidates<S: GraphStorage>(
        &self,
        storage: &S,
        op: &lora_compiler::NodeByPropertyRangeScanExec,
        descending: bool,
    ) -> Vec<NodeId> {
        let lo_prop = self.lo.clone().and_then(|v| lora_value_to_property(v).ok());
        let hi_prop = self.hi.clone().and_then(|v| lora_value_to_property(v).ok());
        let candidates = match single_label_hint(&op.labels) {
            Some(label) => storage
                .node_range_candidates(label, &op.key, lo_prop.as_ref(), hi_prop.as_ref())
                .unwrap_or_else(|| scan_node_ids_for_label_groups(storage, &op.labels)),
            None => scan_node_ids_for_label_groups(storage, &op.labels),
        };
        let filter = self.filter();
        let mut keyed: Vec<(LoraValue, NodeId)> = candidates
            .into_iter()
            .filter(|&id| node_matches_range_filter(storage, id, &filter))
            .map(|id| {
                let value = storage
                    .with_node(id, |n| n.properties.get(op.key.as_str()).cloned())
                    .flatten()
                    .map(LoraValue::from)
                    .unwrap_or(LoraValue::Null);
                (value, id)
            })
            .collect();
        keyed.sort_by(|(a, ai), (b, bi)| {
            let ord = compare_values_total(a, b).then(ai.cmp(bi));
            if descending {
                ord.reverse()
            } else {
                ord
            }
        });
        keyed.into_iter().map(|(_, id)| id).collect()
    }

    pub(crate) fn next_id<S: GraphStorage>(&mut self, storage: &S) -> Option<NodeId> {
        loop {
            if let Some(id) = self.pending.next() {
                if matches!(self.mode, OrderedMode::Buffered)
                    || node_matches_range_filter(storage, id, &self.filter())
                {
                    return Some(id);
                }
                continue;
            }
            let lo_prop = self.lo.clone().and_then(|v| lora_value_to_property(v).ok());
            let hi_prop = self.hi.clone().and_then(|v| lora_value_to_property(v).ok());
            let OrderedMode::Index {
                label,
                descending,
                after,
                chunk,
                exhausted,
            } = &mut self.mode
            else {
                return None;
            };
            if *exhausted {
                return None;
            }
            let ids = storage.node_range_ordered_chunk(
                label,
                &self.key,
                lo_prop.as_ref(),
                hi_prop.as_ref(),
                *descending,
                after.as_ref().map(|(v, id)| (v, *id)),
                *chunk,
            )?;
            if ids.len() < *chunk {
                *exhausted = true;
            }
            let last = *ids.last()?;
            let last_value = storage
                .with_node(last, |n| n.properties.get(self.key.as_str()).cloned())
                .flatten()?;
            *after = Some((last_value, last));
            *chunk = (*chunk * 2).min(4096);
            self.pending = ids.into_iter();
        }
    }
}