spg-engine 7.37.23

//! Index-access planning — B-tree equality seeks, PK fast paths, GIN /
//! trigram / NSW-vector lookups, and the `col = literal` helpers behind
//! them. Split out of `lib.rs` (v7.32 engine modularisation). Pure
//! planners over (WHERE/ORDER expr, schema, catalog, table): they return
//! candidate row indices/locators or `None` to fall back to a scan.

use alloc::borrow::Cow;
use alloc::boxed::Box;
use alloc::vec::Vec;
use core::ops::Bound;

use spg_sql::ast::{BinOp, Expr, Literal, SelectStatement};
use spg_storage::{Catalog, ColumnSchema, IndexKey, Row, Table, Value};

use crate::eval::{self, EvalContext};
use crate::{
    CancelToken, Engine, EngineError, QueryResult, apply_offset_and_limit, build_projection,
    memoize,
};

/// Try to plan a WHERE clause as an equality lookup against an existing
/// index. Returns the candidate row indices on success; `None` means the
/// caller should fall back to a full scan.
///
/// v0.8 recognises a single top-level `col = literal` (in either operand
/// order). AND chains and range scans land in later milestones.
/// Look for `ORDER BY col <dist-op> literal LIMIT k` against an
/// NSW-indexed vector column. Recognised distance ops: `<->` (L2),
/// `<#>` (inner product), `<=>` (cosine). When a WHERE clause is
/// present, the planner does an "over-fetch and filter" pass — it
/// asks the graph for `k * over_fetch` candidates, evaluates WHERE
/// against each, and trims back to `k`. Returns the row indices in
/// ascending-distance order when the plan applies.
pub(crate) fn try_nsw_knn(
    stmt: &SelectStatement,
    table: &Table,
    schema_cols: &[ColumnSchema],
    table_alias: &str,
    snapshot: &spg_storage::snapshot::Snapshot,
) -> Option<Vec<usize>> {
    if stmt.distinct {
        return None;
    }
    let limit = usize::try_from(stmt.limit_literal()?).ok()?;
    if limit == 0 {
        return None;
    }
    // v6.4.0 — NSW kNN dispatch needs a single ORDER BY key on the
    // distance metric. Multi-key ORDER BY falls through to the
    // generic sort path.
    if stmt.order_by.len() != 1 {
        return None;
    }
    let order = &stmt.order_by[0];
    // NSW kNN returns rows ascending by distance — DESC inverts the
    // natural order, so the planner can't handle it without a sort
    // pass. Fall back to the generic ORDER BY path.
    if order.desc {
        return None;
    }
    let Expr::Binary { lhs, op, rhs } = &order.expr else {
        return None;
    };
    let metric = match op {
        BinOp::L2Distance => spg_storage::NswMetric::L2,
        BinOp::InnerProduct => spg_storage::NswMetric::InnerProduct,
        BinOp::CosineDistance => spg_storage::NswMetric::Cosine,
        _ => return None,
    };
    // Accept both `col <op> literal` and `literal <op> col`.
    let ((Expr::Column(col), literal) | (literal, Expr::Column(col))) =
        (lhs.as_ref(), rhs.as_ref())
    else {
        return None;
    };
    if let Some(q) = &col.qualifier
        && q != table_alias
    {
        return None;
    }
    let col_pos = schema_cols.iter().position(|s| s.name == col.name)?;
    let query = literal_to_vector(literal)?;
    let idx = spg_storage::nsw_index_on(table, col_pos)?;
    if let Some(where_expr) = &stmt.where_ {
        // Over-fetch and filter. The factor (10×) is a heuristic that
        // covers typical selectivity for the corpus tests; v2.x will
        // make it configurable.
        let over_fetch = limit.saturating_mul(10).max(NSW_OVER_FETCH_FLOOR);
        let candidates = spg_storage::nsw_query(table, &idx.name, &query, over_fetch, metric);
        let ctx = EvalContext::new(schema_cols, Some(table_alias));
        let mut kept: Vec<usize> = Vec::with_capacity(limit);
        for i in candidates {
            // Phase C.3 step 2c — MVCC read gate. Skip hot rows this
            // snapshot cannot see so an in-place writer's dead/old
            // version never surfaces on the NSW kNN fast path, and an
            // invisible row never counts toward LIMIT. No-op today.
            if !table.is_row_visible(i, snapshot) {
                continue;
            }
            let row = &table.rows()[i];
            let cond = eval::eval_expr(where_expr, row, &ctx).ok()?;
            if crate::eval::predicate_is_true(&cond, "WHERE", ctx.mysql_dialect).ok()? {
                kept.push(i);
                if kept.len() >= limit {
                    break;
                }
            }
        }
        Some(kept)
    } else {
        // Phase C.3 step 2c — MVCC read gate on the WHERE-less kNN
        // path too: drop hot rows this snapshot cannot see so a
        // writer's dead/old version never surfaces. No-op today (every
        // hot header is committed-alive so `is_row_visible` is `true`).
        Some(
            spg_storage::nsw_query(table, &idx.name, &query, limit, metric)
                .into_iter()
                .filter(|&i| table.is_row_visible(i, snapshot))
                .collect(),
        )
    }
}

/// Lower bound on the over-fetch pool when WHERE is present — even
/// for tiny `LIMIT 1` queries we keep enough candidates to absorb a
/// few WHERE rejections.
const NSW_OVER_FETCH_FLOOR: usize = 32;

/// v7.34.5 — drive the row scan via a BTree index walk in the
/// requested ORDER BY direction, emitting matched row indices lazily
/// and stopping after `LIMIT + OFFSET` survive WHERE. Avoids the
/// full-table materialise + partial-sort tail that the mailrs
/// `content_worker` baseline pinned at 80 ms across 250 k rows.
/// Returns `Some(row_indices)` in the order the caller's ORDER BY
/// asked for so `materialise_in_order` can skip its own sort pass;
/// `None` falls through to the existing scan + sort path.
///
/// Eligibility (any failure → `None`):
///   * no `DISTINCT`, no `LIMIT WITH TIES` (both need a full sort).
///   * `LIMIT N` literal present, `N > 0`.
///   * `OFFSET` literal absent or small enough that scanning past it
///     stays cheap (the walker collects `N + OFFSET` rows then trims).
///   * `ORDER BY` is exactly one entry, the expression is a bare
///     column on this table, and the column has a BTree index.
///   * No GROUP BY / HAVING (those are aggregate-bound paths handled
///     elsewhere; this helper sits on `exec_bare_select`'s primary).
///   * Cold-tier locators short-circuit to `None` so the walker
///     never crosses a tier boundary mid-scan; the legacy path
///     handles cold rows.
pub(crate) fn try_pk_walk_top_n<'a>(
    stmt: &SelectStatement,
    catalog: &'a spg_storage::Catalog,
    table: &'a Table,
    schema_cols: &[ColumnSchema],
    table_alias: &str,
    engine: &Engine,
    cancel: CancelToken<'_>,
) -> Option<Vec<Cow<'a, Row<'static>>>> {
    if stmt.distinct || stmt.limit_with_ties {
        return None;
    }
    if stmt.group_by.is_some() || stmt.having.is_some() {
        return None;
    }
    let limit = usize::try_from(stmt.limit_literal()?).ok()?;
    if limit == 0 {
        return None;
    }
    let offset = stmt
        .offset_literal()
        .and_then(|n| usize::try_from(n).ok())
        .unwrap_or(0);
    // Cap absolute walk cost: if OFFSET is huge, the walker would emit
    // `offset + limit` candidates before returning anything, which can
    // exceed the partial-sort fast path's complexity. The legacy path
    // already handles large-OFFSET sweeps with select_nth_unstable.
    const WALKER_OFFSET_CAP: usize = 65_536;
    if offset > WALKER_OFFSET_CAP {
        return None;
    }
    let want = offset.checked_add(limit)?;
    if stmt.order_by.len() != 1 {
        return None;
    }
    let order = &stmt.order_by[0];
    let Expr::Column(col) = &order.expr else {
        return None;
    };
    if let Some(q) = &col.qualifier
        && q != table_alias
    {
        return None;
    }
    let col_pos = schema_cols
        .iter()
        .position(|s| s.name.eq_ignore_ascii_case(&col.name))?;
    let index = table.index_on(col_pos)?;
    if !matches!(index.kind, spg_storage::IndexKind::BTree(_)) {
        return None;
    }
    // r1020 — a NULL key is not in the btree, so walking the btree cannot
    // see the rows that carry one. That is a silent wrong answer, in both
    // directions, and it shipped:
    //
    //   ORDER BY k DESC LIMIT 3   PG: NULL NULL 30    us: 30 20 10
    //   ORDER BY k ASC  LIMIT 5   PG: 10 20 30 NULL NULL
    //                             us: 10 20 30          (two rows dropped)
    //
    // DESC returns the wrong rows because PG orders NULLS FIRST there;
    // ASC returns too few because the walk runs out of indexed rows and
    // has nothing to fall back on. The unbounded ORDER BY is unaffected —
    // it sorts, and the sort sees every row.
    //
    // A NOT NULL column cannot carry one, so the fast path is exact there
    // and keeps its win. A nullable column falls back to the sort. That is
    // conservative: most such queries have no NULLs at all, and reclaiming
    // them means the walker learning to emit NULL-keyed rows at the right
    // end, which is a change to what it walks rather than a guard on when.
    // DESC is unrecoverable here: PG orders NULLS FIRST, so a NULL-keyed
    // row belongs at the very front and the walk would have to know about
    // it before emitting anything. ASC is recoverable, because NULLs belong
    // last — see the short-walk check after the loop.
    let key_nullable = schema_cols
        .get(col_pos)
        .is_none_or(|c: &ColumnSchema| c.nullable);
    if key_nullable && order.desc {
        return None;
    }
    let where_expr = stmt.where_.as_ref();
    let ctx = EvalContext::new(schema_cols, Some(table_alias));
    let table_name = table.schema().name.as_str();
    let mut kept: Vec<Cow<'a, Row>> = Vec::with_capacity(want);
    // v7.37.x (mailrs prod content_worker NOT EXISTS) — share one
    // MemoizeCache across the walk so a materialised InList (from
    // `subquery_replacement` of an uncorrelated InSubquery / the
    // NOT EXISTS pullup) hits the in-set fast path instead of an
    // O(N×M) linear scan per row. Without the memo, 25 k outer ×
    // 25 k InList on the prod messages × attachment_content shape
    // cost ~6 s.
    let mut memo = memoize::MemoizeCache::new();
    // v7.37.5-A2b (profile-guided) — content_worker prod profile
    // showed 24% self-time in `resolve_column` (linear scan of
    // ctx.columns per cell access) because the per-row WHERE eval
    // path went through `eval_with_in_sets` → `eval_expr(lhs, ...)`
    // → `Expr::Column` → `resolve_column`. The same WHERE expression
    // is fully-compilable (an `m.id NOT IN (literal_list)` after
    // `subquery_replacement`) — pre-compile it once and run
    // `eval_compiled` per row, pre-resolving the column position.
    // Falls back to the eval path for non-compilable WHEREs.
    let compiled_where: Option<eval::CompiledExpr> = where_expr
        .filter(|w| eval::fully_compilable(w))
        .map(|w| eval::compile_expr(w, &ctx));
    let mut eval_stack: Vec<spg_storage::Value<'static>> = Vec::new();
    // The walker yields `(IndexKey, &Vec<RowLocator>)` ordered by key
    // in the requested direction; per-key locator order within the
    // map preserves insertion order, which matches the legacy stable-
    // sort tie-break for equal keys.
    let walker: Box<dyn Iterator<Item = (&spg_storage::IndexKey, &Vec<spg_storage::RowLocator>)>> =
        if order.desc {
            Box::new(index.iter_desc())
        } else {
            Box::new(index.iter_asc())
        };
    // Phase C.3 step 2b — MVCC read gate. Compute the reader's
    // snapshot once; hot rows this snapshot cannot see are skipped so
    // an in-place writer's dead/old version never surfaces on the
    // BTree-walk fast path. Cold-tier locators are frozen segments =
    // always visible, so they stay ungated. No-op today: every hot
    // header is frozen/committed-alive, so `is_row_visible` is `true`.
    let scan_snapshot = engine.current_snapshot();
    for (key, locators) in walker {
        for loc in locators {
            // v7.34.7 (mailrs prod #6 follow-up) — single-table walker
            // gains the same hot/cold dispatch the JOIN walker got in
            // 7.34.6. `mailrs_prod_plain_limit` (`SELECT ... FROM
            // messages WHERE ... ORDER BY id DESC LIMIT N`) was the
            // remaining shape that bailed on the first cold locator
            // and fell back to the legacy materialise + partial-sort
            // path on the 803 MB prod catalog.
            let row_cow: Cow<'a, Row> = match *loc {
                spg_storage::RowLocator::Hot(row_idx) => {
                    if !table.is_row_visible(row_idx, &scan_snapshot) {
                        continue;
                    }
                    match table.rows().get(row_idx) {
                        Some(r) => Cow::Borrowed(r),
                        None => continue,
                    }
                }
                spg_storage::RowLocator::Cold { segment_id, .. } => {
                    match catalog.resolve_cold_locator(table_name, segment_id, key) {
                        Some(r) => Cow::Owned(r),
                        None => continue,
                    }
                }
            };
            if let Some(cw) = &compiled_where {
                // v7.37.5-A2b — compiled path: column positions resolved
                // at compile time; `eval_compiled` is allocator-free per
                // row (stack reused).
                let cond = eval::eval_compiled(cw, row_cow.as_ref(), &ctx, &mut eval_stack).ok()?;
                if !crate::eval::predicate_is_true(&cond, "WHERE", ctx.mysql_dialect).ok()? {
                    continue;
                }
            } else if let Some(w) = where_expr {
                let cond = engine
                    .eval_expr_with_correlated(w, row_cow.as_ref(), &ctx, cancel, Some(&mut memo))
                    .ok()?;
                if !crate::eval::predicate_is_true(&cond, "WHERE", ctx.mysql_dialect).ok()? {
                    continue;
                }
            }
            kept.push(row_cow);
            if kept.len() >= want {
                return Some(kept);
            }
        }
    }
    // r1020 — the walk ran out of INDEXED rows before filling the request.
    // On a nullable key that is exactly the case where NULL-keyed rows —
    // which no btree holds — would have completed it, and PG places them
    // last under ASC. Returning what we have would be a short answer, so
    // hand the query back to the sort, which sees every row.
    //
    // A NOT NULL key cannot be short for that reason, so it keeps its
    // result. On a nullable key this costs the walk when the limit was not
    // satisfied anyway, and keeps the fast path for every request the
    // indexed rows do satisfy — which is the shape that wins.
    if key_nullable && kept.len() < want {
        return None;
    }
    Some(kept)
}

/// Pull a `Vec<f32>` out of a literal-or-cast expression. Returns
/// `None` for anything we can't fold at plan time.
pub(crate) fn literal_to_vector(e: &Expr) -> Option<Vec<f32>> {
    match e {
        Expr::Literal(Literal::Vector(v)) => Some(v.clone()),
        Expr::Cast { expr, .. } => literal_to_vector(expr),
        _ => None,
    }
}

/// Materialise rows in a planner-supplied order (used by the NSW path)
/// without re-running ORDER BY. The projection + LIMIT slot mirror the
/// equivalent block in `exec_bare_select`.
pub(crate) fn materialise_in_order(
    stmt: &SelectStatement,
    schema_cols: &[ColumnSchema],
    table_alias: &str,
    ordered_rows: &[Cow<'_, Row<'static>>],
    mysql: bool,
) -> Result<QueryResult, EngineError> {
    let ctx = EvalContext::new(schema_cols, Some(table_alias));
    let projection = build_projection(&stmt.items, schema_cols, table_alias, mysql)?;
    let mut output_rows: Vec<Row<'static>> = Vec::with_capacity(ordered_rows.len());
    for row_cow in ordered_rows {
        let row = row_cow.as_ref();
        let mut values = Vec::with_capacity(projection.len());
        for p in &projection {
            values.push(eval::eval_expr(&p.expr, row, &ctx)?);
        }
        output_rows.push(Row::new(values));
    }
    apply_offset_and_limit(
        &mut output_rows,
        stmt.offset_literal(),
        stmt.limit_literal(),
    );
    let columns: Vec<ColumnSchema> = projection
        .into_iter()
        .map(|p| ColumnSchema::new(p.output_name, p.ty, p.nullable))
        .collect();
    Ok(QueryResult::Rows {
        columns,
        rows: output_rows,
    })
}

/// v7.20 P4 — hot-row POSITION seek for the mutation paths
/// (UPDATE / DELETE index their planned writes by position in
/// `table.rows()`, so the Cow-row shape `try_index_seek`
/// returns doesn't fit). Same top-level-AND recursion and
/// col=literal resolution; the caller re-applies the full WHERE
/// to every returned row so the index only narrows candidates.
///
/// Returns `None` (→ caller full-scans) when no equality leaf
/// hits an index OR any matching locator lives in the cold tier
/// — the mutation paths operate on hot rows, and the PK
/// promote-then-walk upstream already handles the
/// cold-single-row case.
pub(crate) fn try_index_seek_positions(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    table: &Table,
    table_alias: &str,
    snapshot: &spg_storage::snapshot::Snapshot,
) -> Option<Vec<usize>> {
    // v7.37.16 — two-sided range (BETWEEN) position seek, mirroring the
    // SELECT path's try_range_seek (same parse, same rows/4 selectivity
    // cap so a wide range still seq-scans). Must run BEFORE the AND
    // recursion: a BETWEEN desugars to `col>=a AND col<=b`, whose halves
    // alone are one-sided and would fall through to a full scan. The
    // caller re-applies the full WHERE per candidate and sorts, exactly
    // as for the equality seek. A cold locator falls through to the
    // legacy paths (the mutation walk operates on hot rows only).
    // v7.39 (round 461 / round 490) — the cap's budget is counted in rows a
    // caller will actually look at.
    //
    // The cap exists so an index walk never costs more than the scan it
    // replaces: return more than a quarter of the table and the seek is
    // refused. But the index holds one locator per row VERSION, and a
    // churned table's index still carries the dead ones — so the count being
    // compared was inflated by exactly the rows the caller then discards.
    //
    // Measured (round 460): delete-and-reinsert 1000 rows on a 50k table
    // with autovacuum off, and by cycle 20 a 1000-row range returns ~22000
    // candidates against a cap of 17750, so the seek is refused and every
    // DELETE becomes a 71000-row scan. Round 461 bought that back by adding
    // `dead_rows()` to the budget, which stopped the refusal but still
    // carried every dead version through a `Vec`, a sort, and a per-
    // candidate visibility test — 61000 of them by cycle 60 (round 490).
    //
    // The dead versions are now dropped INSIDE the walk, by the same
    // `is_row_visible` test the caller applies, so the budget is back to its
    // original meaning and round 461's compensation is unnecessary.
    let seek_cap = table.rows().len() / 4;
    if let Some((col_pos, lo, hi)) = parse_range_bounds(where_expr, schema_cols, table_alias)
        && let Some(idx) = table.index_on(col_pos)
        && let Some(locators) =
            idx.lookup_range_capped_by(bound_as_ref(&lo), bound_as_ref(&hi), seek_cap, |l| {
                match l {
                    spg_storage::RowLocator::Hot(i) => table.is_row_visible(i, snapshot),
                    // A cold locator makes the whole seek bail below; keep it
                    // so that decision is reached rather than silently
                    // dropping it here.
                    spg_storage::RowLocator::Cold { .. } => true,
                }
            })
    {
        let mut out = Vec::with_capacity(locators.len());
        let mut all_hot = true;
        for loc in &locators {
            match *loc {
                spg_storage::RowLocator::Hot(i) => out.push(i),
                spg_storage::RowLocator::Cold { .. } => {
                    all_hot = false;
                    break;
                }
            }
        }
        if all_hot {
            // v7.39 (pg_stat knife B) — one index scan.
            table.note_index_scan(out.len() as u64);
            return Some(out);
        }
    }
    if let Expr::Binary {
        lhs,
        op: BinOp::And,
        rhs,
    } = where_expr
    {
        if let Some(p) = try_index_seek_positions(lhs, schema_cols, table, table_alias, snapshot) {
            return Some(p);
        }
        return try_index_seek_positions(rhs, schema_cols, table, table_alias, snapshot);
    }
    let Expr::Binary {
        lhs,
        op: BinOp::Eq,
        rhs,
    } = where_expr
    else {
        return None;
    };
    let (col_pos, value) = resolve_col_literal_pair(lhs, rhs, schema_cols, table_alias)
        .or_else(|| resolve_col_literal_pair(rhs, lhs, schema_cols, table_alias))?;
    let idx = table.index_on(col_pos)?;
    let key = IndexKey::from_value(&value)?;
    let locators = idx.lookup_eq(&key);
    let mut out = Vec::with_capacity(locators.len());
    for loc in locators {
        match *loc {
            // v7.39 (round 490) — dead versions are dropped here for the
            // same reason as in the range walk above: the caller tests
            // exactly this and skips.
            spg_storage::RowLocator::Hot(i) => {
                if table.is_row_visible(i, snapshot) {
                    out.push(i);
                }
            }
            spg_storage::RowLocator::Cold { .. } => return None,
        }
    }
    // v7.39 (pg_stat knife B) — one index scan.
    table.note_index_scan(out.len() as u64);
    Some(out)
}

/// v7.38 (perf, index range scan) — flip a comparison operator for the
/// `literal <op> column` orientation (`5 < v` ≡ `v > 5`).
fn flip_comparison(op: BinOp) -> Option<BinOp> {
    match op {
        BinOp::Gt => Some(BinOp::Lt),
        BinOp::GtEq => Some(BinOp::LtEq),
        BinOp::Lt => Some(BinOp::Gt),
        BinOp::LtEq => Some(BinOp::GtEq),
        _ => None,
    }
}

/// Parse a single `col <op> lit` / `lit <op> col` range comparison into
/// `(col_pos, lo, hi)` where exactly one of `lo` / `hi` is bounded. Returns
/// None for anything that isn't a `> >= < <=` comparison of an indexable
/// column against a literal.
fn parse_one_sided_range(
    e: &Expr,
    schema_cols: &[ColumnSchema],
    table_alias: &str,
) -> Option<(usize, Bound<IndexKey>, Bound<IndexKey>)> {
    let Expr::Binary { lhs, op, rhs } = e else {
        return None;
    };
    // `col <op> lit` keeps `op`; `lit <op> col` flips it.
    let (col_pos, value, op) =
        if let Some((p, v)) = resolve_col_literal_pair(lhs, rhs, schema_cols, table_alias) {
            (p, v, *op)
        } else if let Some((p, v)) = resolve_col_literal_pair(rhs, lhs, schema_cols, table_alias) {
            (p, v, flip_comparison(*op)?)
        } else {
            return None;
        };
    let key = IndexKey::from_value(&value)?;
    let bounds = match op {
        BinOp::Gt => (Bound::Excluded(key), Bound::Unbounded),
        BinOp::GtEq => (Bound::Included(key), Bound::Unbounded),
        BinOp::Lt => (Bound::Unbounded, Bound::Excluded(key)),
        BinOp::LtEq => (Bound::Unbounded, Bound::Included(key)),
        _ => return None,
    };
    Some((col_pos, bounds.0, bounds.1))
}

/// Parse a two-sided range predicate — `(col >=/> a) AND (col <=/< b)` on the
/// SAME column (the BETWEEN shape) — into `(col_pos, lo, hi)` with both ends
/// bounded. Deliberately TWO-sided only: a one-sided `col > x` is usually
/// non-selective (matches a large fraction), where an index range scan +
/// per-candidate re-eval loses to a tight seq scan; those stay on the seq
/// path. A two-sided BETWEEN is typically selective enough to win, and the
/// cap in `try_range_seek` catches the rare wide one.
fn parse_range_bounds(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    table_alias: &str,
) -> Option<(usize, Bound<IndexKey>, Bound<IndexKey>)> {
    let bounds = parse_range_bounds_inner(where_expr, schema_cols, table_alias)?;
    // v7.39 (enum order knife) — the index orders enum labels
    // lexicographically but PG's enum order is the catalog member order,
    // so a range walk would under-select (and the caller's WHERE re-eval
    // cannot restore missing rows). Bail to a seq scan, whose Binary
    // comparisons are member-order aware. Eq / IN-list seeks stay: label
    // equality is exact.
    if schema_cols
        .get(bounds.0)
        .is_some_and(|c| c.user_enum_type.is_some())
    {
        return None;
    }
    Some(bounds)
}

fn parse_range_bounds_inner(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    table_alias: &str,
) -> Option<(usize, Bound<IndexKey>, Bound<IndexKey>)> {
    let Expr::Binary {
        lhs,
        op: BinOp::And,
        rhs,
    } = where_expr
    else {
        return None;
    };
    let (c1, lo1, hi1) = parse_one_sided_range(lhs, schema_cols, table_alias)?;
    let (c2, lo2, hi2) = parse_one_sided_range(rhs, schema_cols, table_alias)?;
    if c1 != c2 {
        return None;
    }
    // One side must supply the lower bound, the other the upper.
    let lo = match (lo1, lo2) {
        (Bound::Unbounded, b) | (b, Bound::Unbounded) => b,
        _ => return None, // two lower bounds — not a clean [lo, hi]
    };
    let hi = match (hi1, hi2) {
        (Bound::Unbounded, b) | (b, Bound::Unbounded) => b,
        _ => return None,
    };
    // Both ends must be bounded (a real BETWEEN), else fall back to seq scan.
    if matches!(lo, Bound::Unbounded) || matches!(hi, Bound::Unbounded) {
        return None;
    }
    Some((c1, lo, hi))
}

fn bound_as_ref(b: &Bound<IndexKey>) -> Bound<&IndexKey> {
    match b {
        Bound::Included(k) => Bound::Included(k),
        Bound::Excluded(k) => Bound::Excluded(k),
        Bound::Unbounded => Bound::Unbounded,
    }
}

/// v7.38 (perf, index range scan) — plan `col <op> lit` / `col BETWEEN a AND b`
/// as an index range walk. Returns the candidate rows in the key range, or None
/// (→ caller seq-scans) when there's no usable index, the shape isn't a range,
/// the range isn't selective (> half the table — the cap), or the range touches
/// a cold-tier locator (whose key we'd need to resolve the segment row and the
/// flattened walk doesn't carry). The caller re-applies the full WHERE per row,
/// so returning a superset is correct.
/// v7.39 (round 560) — rebuild the column's value from its index key.
///
/// `IndexKey::Int` holds an i64 whatever the column was declared as, so
/// the DECLARED type decides what comes back — `SELECT k` on an INT
/// column must answer int4, not int8. A type whose key is lossy in the
/// other direction (a date and a timestamp both key as Int) is not
/// reconstructible and falls back to reading the row.
fn value_from_key(
    key: &spg_storage::IndexKey,
    declared: spg_storage::DataType,
) -> Option<spg_storage::Value<'static>> {
    use spg_storage::{IndexKey as K, Value};
    Some(match (key, declared) {
        (K::Int(n), spg_storage::DataType::SmallInt) => Value::SmallInt(i16::try_from(*n).ok()?),
        (K::Int(n), spg_storage::DataType::Int) => Value::Int(i32::try_from(*n).ok()?),
        (K::Int(n), spg_storage::DataType::BigInt) => Value::BigInt(*n),
        (K::Text(t), spg_storage::DataType::Text) => Value::text(t.clone()),
        (K::Bool(b), spg_storage::DataType::Bool) => Value::Bool(*b),
        (K::Uuid(u), spg_storage::DataType::Uuid) => Value::Uuid(*u),
        _ => return None,
    })
}

/// v7.39 (round 560) — an index-only range scan.
///
/// When the projection needs nothing but the indexed column, the value
/// is already in the index KEY and the row never has to be read.
/// `try_range_seek` below throws the key away, keeps the locator and
/// fetches the row for a value the walk had in hand. Measured over
/// pgwire on a 500k table, projecting `k` for a 100k-row range:
///
/// ```text
///     PG18  Index Only Scan   3.6 ms      SPG  30 ms
/// ```
///
/// 8x, widening with the row count (2x at 1k rows).
///
/// PG needs its visibility map for this — a heap tuple carries its own
/// visibility, so an index entry alone cannot say whether the row is
/// live, and PG falls back to the heap for any page the map does not
/// mark all-visible. SPG keeps a header array beside the rows, so the
/// locator answers it directly and there is no map to be stale.
///
/// No selectivity cap: the ceiling on `try_range_seek` exists because
/// its caller materialises every candidate, which this does not do.
pub(crate) fn try_index_only_range(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    table: &Table,
    table_alias: &str,
    snapshot: &spg_storage::snapshot::Snapshot,
    projected: usize,
) -> Option<Vec<spg_storage::Value<'static>>> {
    let mut out: Vec<spg_storage::Value<'static>> = Vec::new();
    match index_only_range_each(
        where_expr,
        schema_cols,
        table,
        table_alias,
        snapshot,
        projected,
        &mut |v| {
            out.push(v);
            Ok(())
        },
    ) {
        Some(Ok(_)) => Some(out),
        // The walk only errors on an index that disagrees with its own
        // schema; this caller has emitted nothing, so it can still fall
        // back to the ordinary path rather than surface it.
        Some(Err(_)) | None => None,
    }
}

/// v7.39 (round 564) — the walk itself, handing each value to a sink.
///
/// `try_index_only_range` collects into a `Vec`; the streaming caller
/// wants to emit as it goes and must not have to collect first. Both go
/// through here so the shape rules and the visibility rules have one
/// copy between them.
///
/// `None` means the shape does not apply and NOTHING has been handed to
/// the sink — the caller may still fall back. `Some(Err(_))` means rows
/// may already have gone out.
pub(crate) fn index_only_range_each(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    table: &Table,
    table_alias: &str,
    snapshot: &spg_storage::snapshot::Snapshot,
    projected: usize,
    sink: &mut dyn FnMut(spg_storage::Value<'static>) -> Result<(), EngineError>,
) -> Option<Result<usize, EngineError>> {
    let (ty, lo, hi, idx) =
        index_only_precheck(where_expr, schema_cols, table, table_alias, projected)?;
    let entries = idx.range_keyed(bound_as_ref(&lo), bound_as_ref(&hi))?;
    let mut headers = table.header_runs();
    let mut n = 0usize;
    for (key, loc) in entries {
        let spg_storage::RowLocator::Hot(i) = loc else {
            return Some(Err(EngineError::Unsupported(
                "index-only scan met a locator outside the hot tier".into(),
            )));
        };
        if !headers.visible(i, snapshot) {
            continue;
        }
        // Unreachable given `key_restores_type` above and keys built
        // from the column's own values — an index that disagrees with
        // its schema is worth saying so about, not walking past.
        let Some(v) = value_from_key(key, ty) else {
            return Some(Err(EngineError::Unsupported(
                "index-only scan: index key does not restore the column type".into(),
            )));
        };
        if let Err(e) = sink(v) {
            return Some(Err(e));
        }
        n += 1;
    }
    Some(Ok(n))
}

/// v7.39 (round 565) — everything decidable about this scan before it
/// walks: the shape of the predicate, the tier, the type, the index.
///
/// Split out because EXPLAIN has to answer the same question and must
/// not answer it from its own copy of the rules. Round 564 named the
/// path in the executor and left EXPLAIN calling it `Index Scan`, so a
/// reader comparing two plans that run 2x apart saw one plan.
pub(crate) fn index_only_precheck<'t>(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    table: &'t Table,
    table_alias: &str,
    projected: usize,
) -> Option<(
    spg_storage::DataType,
    Bound<IndexKey>,
    Bound<IndexKey>,
    &'t spg_storage::Index,
)> {
    let (col_pos, lo, hi) = parse_index_only_bounds(where_expr, schema_cols, table_alias)?;
    if col_pos != projected {
        return None;
    }
    // A cold-tier locator carries no position to test visibility with,
    // and its value lives off-heap; anything cold falls back.
    if table.has_cold_rows_fast() {
        return None;
    }
    let ty = schema_cols[col_pos].ty;
    // Decided BEFORE the walk, not per row: a streaming caller cannot
    // take a row back once it has gone out, so "this type does not come
    // back from its key" has to be a shape rejection, not a discovery
    // made halfway through.
    if !key_restores_type(ty) {
        return None;
    }
    let idx = table.index_on(col_pos)?;
    Some((ty, lo, hi, idx))
}

/// v7.39 (round 566) — the predicates an index-only scan can serve:
/// a two-sided range, or an equality, which is the degenerate range
/// `[k, k]`.
///
/// Equality was left out when round 560 built this, and it is the
/// commoner query. On 500k rows with 50 distinct values, `WHERE g = 7`
/// returns 10k of them:
///
/// ```text
///     SPG  9.5 ms      PG18  4.5 ms      2.1x
/// ```
///
/// — the same loss the range shape carried before round 564, in the
/// shape people write more often.
///
/// The enum guard `parse_range_bounds` applies does not reach here, and
/// deliberately: it exists because the index orders labels
/// lexicographically while PG orders them by catalog position, so a
/// RANGE walk would under-select. Equality does not depend on the
/// order — its own comment in `try_index_seek_positions` says so.
fn parse_index_only_bounds(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    table_alias: &str,
) -> Option<(usize, Bound<IndexKey>, Bound<IndexKey>)> {
    if let Some(r) = parse_range_bounds(where_expr, schema_cols, table_alias) {
        return Some(r);
    }
    let Expr::Binary {
        lhs,
        op: BinOp::Eq,
        rhs,
    } = where_expr
    else {
        return None;
    };
    let (col_pos, value) = resolve_col_literal_pair(lhs, rhs, schema_cols, table_alias)
        .or_else(|| resolve_col_literal_pair(rhs, lhs, schema_cols, table_alias))?;
    // `col = NULL` is never true, and an index that stores NULL keys
    // would happily hand rows back for it.
    if value.is_null() {
        return None;
    }
    let key = IndexKey::from_value(&value)?;
    Some((col_pos, Bound::Included(key.clone()), Bound::Included(key)))
}

/// Which declared types an index key comes back as unambiguously.
///
/// The pairs `value_from_key` accepts, as a question about the TYPE
/// alone. A date and a timestamp both key as `IndexKey::Int`, so
/// neither is here.
fn key_restores_type(ty: spg_storage::DataType) -> bool {
    use spg_storage::DataType as T;
    matches!(
        ty,
        T::SmallInt | T::Int | T::BigInt | T::Text | T::Bool | T::Uuid
    )
}

fn try_range_seek<'a>(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    table: &'a Table,
    table_alias: &str,
    snapshot: &spg_storage::snapshot::Snapshot,
) -> Option<Vec<Cow<'a, Row<'static>>>> {
    let (col_pos, lo, hi) = parse_range_bounds(where_expr, schema_cols, table_alias)?;
    let idx = table.index_on(col_pos)?;
    // Selectivity cap: the caller still materialises + re-evals every candidate
    // this returns, so the index range scan only pays off when the range is a
    // small fraction of the table. Empirically ~50%-selective ranges regress
    // (index-walk + per-candidate materialise > a tight seq scan); a quarter is
    // a safe margin that keeps the clear wins and falls back (→ None → seq scan)
    // otherwise, so no endpoint regresses.
    let cap = table.rows().len() / 4;
    // v7.39 (round 490) — drop the dead versions inside the walk. This loop
    // already tested `is_row_visible` and skipped; doing it one level down
    // means the cap stops counting them too (see `lookup_range_capped_by`).
    let locators =
        idx.lookup_range_capped_by(bound_as_ref(&lo), bound_as_ref(&hi), cap, |l| match l {
            spg_storage::RowLocator::Hot(i) => table.is_row_visible(i, snapshot),
            spg_storage::RowLocator::Cold { .. } => true,
        })?;
    let mut out: Vec<Cow<'a, Row>> = Vec::with_capacity(locators.len());
    for loc in &locators {
        match *loc {
            spg_storage::RowLocator::Hot(i) => {
                if let Some(row) = table.rows().get(i) {
                    out.push(Cow::Borrowed(row));
                }
            }
            // A range walk flattens locators without their per-key handle, so a
            // cold-tier row can't be resolved the way the Eq seek does. Bail to
            // a seq scan (correct — the caller re-applies the WHERE to all rows).
            spg_storage::RowLocator::Cold { .. } => return None,
        }
    }
    // v7.39 (pg_stat knife B) — one index scan.
    table.note_index_scan(out.len() as u64);
    Some(out)
}

/// v7.38 (perf, exact-range count) — `count(*)` over a WHERE that is EXACTLY a
/// two-sided indexed range: the visible in-range locators ARE the matching
/// rows (`parse_range_bounds` only matches a pure two-sided range on the
/// indexed column, so there's no residual predicate), so we tally them without
/// materialising a single row or re-evaluating the WHERE. Returns None (→ the
/// caller runs the general aggregate path) when the shape doesn't fit or a
/// cold-tier locator is in range (its visibility needs the resolved row). No
/// selectivity cap — counting is cheap and always ≤ a full scan.
pub(crate) fn try_range_count(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    table: &Table,
    table_alias: &str,
    snapshot: &spg_storage::snapshot::Snapshot,
) -> Option<i64> {
    let (col_pos, lo, hi) = parse_range_bounds(where_expr, schema_cols, table_alias)?;
    let idx = table.index_on(col_pos)?;
    let locators = idx.lookup_range_capped(bound_as_ref(&lo), bound_as_ref(&hi), usize::MAX)?;
    let mut count: i64 = 0;
    for loc in &locators {
        match *loc {
            spg_storage::RowLocator::Hot(i) => {
                if table.is_row_visible(i, snapshot) {
                    count += 1;
                }
            }
            spg_storage::RowLocator::Cold { .. } => return None,
        }
    }
    Some(count)
}

pub(crate) fn try_index_seek<'a>(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    catalog: &'a Catalog,
    table: &'a Table,
    table_alias: &str,
    snapshot: &spg_storage::snapshot::Snapshot,
) -> Option<Vec<Cow<'a, Row<'static>>>> {
    // v7.38 (perf) — a range predicate (`col BETWEEN a AND b`, `col > x`) walks
    // the index range instead of full-scanning. Tried before the `AND` recurse
    // so a two-sided BETWEEN is caught as one range; a mixed predicate like
    // `id = 1 AND created > $1` isn't a pure range, so this returns None and the
    // Eq path below still seeks on `id`.
    if let Some(rows) = try_range_seek(where_expr, schema_cols, table, table_alias, snapshot) {
        return Some(rows);
    }
    // v7.11.3 — recurse through top-level `AND` so a PG-style
    // composite predicate like `WHERE id = 1 AND created_at > $1`
    // still hits the index on `id`. The caller re-applies the
    // full WHERE expression to each returned row, so dropping the
    // residual conjuncts here is correct — the index just narrows
    // the candidate set.
    if let Expr::Binary {
        lhs,
        op: BinOp::And,
        rhs,
    } = where_expr
    {
        // Try LHS first (typical convention: leading equality on
        // the indexed column comes first in user-written SQL).
        if let Some(rows) = try_index_seek(lhs, schema_cols, catalog, table, table_alias, snapshot)
        {
            return Some(rows);
        }
        return try_index_seek(rhs, schema_cols, catalog, table, table_alias, snapshot);
    }
    // v7.33 (mailrs 7.33.0) — `indexed_col IN (lit, …)` seeks each literal
    // and unions the rows (PG's bitmap index scan) instead of a full scan
    // + per-row membership test. The single-table path otherwise tested a
    // 60-element list against every row (24k × 60 string compares ~66 ms).
    if let Some(rows) = try_inlist_seek(
        where_expr,
        schema_cols,
        catalog,
        table,
        table_alias,
        snapshot,
    ) {
        return Some(rows);
    }
    let Expr::Binary {
        lhs,
        op: BinOp::Eq,
        rhs,
    } = where_expr
    else {
        return None;
    };
    let (col_pos, value) = resolve_col_literal_pair(lhs, rhs, schema_cols, table_alias)
        .or_else(|| resolve_col_literal_pair(rhs, lhs, schema_cols, table_alias))?;
    let idx = table.index_on(col_pos)?;
    let key = IndexKey::from_value(&value)?;
    let locators = idx.lookup_eq(&key);
    let table_name = table.schema().name.as_str();
    // v5.1: each locator dispatches to either the hot tier (zero-
    // copy borrow of `table.rows()[i]`) or a cold-tier segment
    // (one page read + dense row decode, ~µs scale). Cold rows are
    // returned as `Cow::Owned` so the caller's `&Row<'static>` iteration
    // doesn't see a tier distinction; pre-freezer (no cold
    // segments loaded) every locator is `Hot` and every entry is
    // `Cow::Borrowed` — identical cost to the pre-v5.1 path.
    let mut out: Vec<Cow<'a, Row>> = Vec::with_capacity(locators.len());
    for loc in locators {
        match *loc {
            spg_storage::RowLocator::Hot(i) => {
                // Phase C.3 step 2c — MVCC read gate: skip hot rows this
                // snapshot cannot see. No-op today.
                if !table.is_row_visible(i, snapshot) {
                    continue;
                }
                if let Some(row) = table.rows().get(i) {
                    out.push(Cow::Borrowed(row));
                }
            }
            spg_storage::RowLocator::Cold { segment_id, .. } => {
                if let Some(row) = catalog.resolve_cold_locator(table_name, segment_id, &key) {
                    out.push(Cow::Owned(row));
                }
            }
        }
    }
    // v7.39 (pg_stat knife B) — one index scan.
    table.note_index_scan(out.len() as u64);
    Some(out)
}

/// v7.33 (mailrs 7.33.0) — `indexed_col IN (lit, …)` candidate seek.
/// Returns the union of per-literal index lookups when `where_expr` is a
/// non-negated IN-list whose LHS is an indexed column (qualified to this
/// table or bare) and whose elements are all literals; None otherwise, so
/// the caller falls through to its Eq seek / full scan. The caller
/// re-applies the full WHERE per row, so the exact per-literal seek set is
/// correct (duplicate keys just revisit a row, which the re-eval dedups by
/// truth, not identity — harmless for a candidate set).
fn try_inlist_seek<'a>(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    catalog: &'a Catalog,
    table: &'a Table,
    table_alias: &str,
    snapshot: &spg_storage::snapshot::Snapshot,
) -> Option<Vec<Cow<'a, Row<'static>>>> {
    let Expr::InList {
        expr,
        list,
        negated: false,
    } = where_expr
    else {
        return None;
    };
    let Expr::Column(c) = expr.as_ref() else {
        return None;
    };
    if !c
        .qualifier
        .as_deref()
        .is_none_or(|q| q.eq_ignore_ascii_case(table_alias))
    {
        return None;
    }
    let col_pos = schema_cols.iter().position(|s| s.name == c.name)?;
    let idx = table.index_on(col_pos)?;
    // Every element must be a literal; bail (full scan) otherwise.
    let mut keys: Vec<IndexKey> = Vec::with_capacity(list.len());
    for e in list {
        let Expr::Literal(l) = e else {
            return None;
        };
        keys.push(IndexKey::from_value(&eval::literal_to_value(l))?);
    }
    let table_name = table.schema().name.as_str();
    let mut out: Vec<Cow<'a, Row>> = Vec::new();
    for key in &keys {
        for loc in idx.lookup_eq(key) {
            match *loc {
                spg_storage::RowLocator::Hot(i) => {
                    // Phase C.3 step 2c — MVCC read gate. No-op today.
                    if !table.is_row_visible(i, snapshot) {
                        continue;
                    }
                    if let Some(row) = table.rows().get(i) {
                        out.push(Cow::Borrowed(row));
                    }
                }
                spg_storage::RowLocator::Cold { segment_id, .. } => {
                    if let Some(row) = catalog.resolve_cold_locator(table_name, segment_id, key) {
                        out.push(Cow::Owned(row));
                    }
                }
            }
        }
    }
    // v7.39 (pg_stat knife B) — one index scan.
    table.note_index_scan(out.len() as u64);
    Some(out)
}

/// v7.12.3 — GIN-accelerated candidate seek for `WHERE col @@ <ts_query>`.
///
/// Recurses through top-level `AND` like [`try_index_seek`] so a
/// composite predicate `WHERE search_vector @@ q AND id > $1` still
/// hits the GIN index on `search_vector` — the caller re-applies the
/// full WHERE expression to each returned candidate, so dropping the
/// `id > $1` residual here stays semantically correct.
///
/// Returns `None` when:
///   - no leaf is a `col @@ <rhs>` shape on a GIN-indexed column;
///   - the RHS can't be const-evaluated to a `Value::TsQuery`
///     (typically because it references row columns);
///   - the resolved `TsQuery` uses query shapes the MVP doesn't
///     accelerate (`Not`, `Phrase` — those fall through to full scan).
///
/// On `Some(rows)` the caller iterates only `rows` and re-evaluates
/// the full `@@` predicate per row, so an over-approximate candidate
/// set is safe.
pub(crate) fn try_gin_seek<'a>(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    catalog: &'a Catalog,
    table: &'a Table,
    table_alias: &str,
    ctx: &eval::EvalContext<'_>,
    snapshot: &spg_storage::snapshot::Snapshot,
) -> Option<Vec<Cow<'a, Row<'static>>>> {
    if let Expr::Binary {
        lhs,
        op: BinOp::And,
        rhs,
    } = where_expr
    {
        if let Some(rows) =
            try_gin_seek(lhs, schema_cols, catalog, table, table_alias, ctx, snapshot)
        {
            return Some(rows);
        }
        return try_gin_seek(rhs, schema_cols, catalog, table, table_alias, ctx, snapshot);
    }
    // v7.17.0 Phase 3.P0-44 — MySQL `MATCH(col1, col2) AGAINST (...)`
    // desugars into `(to_tsvector(col1) @@ q) OR (to_tsvector(col2) @@ q)`
    // in the parser. To accelerate the multi-column case, walk OR the same
    // way we walk AND: only emit a candidate set if BOTH sides can seek
    // (otherwise the OR result is unbounded and we must fall through to
    // the full scan). Candidates are union'd; the caller's WHERE re-eval
    // verifies the full predicate per row, so duplicates / supersets stay
    // semantically safe.
    if let Expr::Binary {
        lhs,
        op: BinOp::Or,
        rhs,
    } = where_expr
    {
        let left = try_gin_seek(lhs, schema_cols, catalog, table, table_alias, ctx, snapshot)?;
        let right = try_gin_seek(rhs, schema_cols, catalog, table, table_alias, ctx, snapshot)?;
        let mut out: Vec<Cow<'a, Row>> = Vec::with_capacity(left.len() + right.len());
        out.extend(left);
        out.extend(right);
        return Some(out);
    }
    let Expr::Binary {
        lhs,
        op: BinOp::TsMatch,
        rhs,
    } = where_expr
    else {
        return None;
    };
    // Either side can be the column; pgvector idiom (`vec @@ q`)
    // hits the first arm, FROM-clause-derived (`plainto_tsquery($1)
    // q ... WHERE search_vector @@ q`) the same. CROSS JOIN derived
    // tables resolve `q` to a Column too.
    let (col_pos, query) = resolve_gin_col_query(lhs, rhs, schema_cols, table_alias, ctx)
        .or_else(|| resolve_gin_col_query(rhs, lhs, schema_cols, table_alias, ctx))?;
    // v7.17.0 Phase 3.P0-44 — MySQL `FULLTEXT KEY` builds a
    // `IndexKind::GinFulltext` posting list (Phase 2.2). It shares
    // the same `gin_lookup_word` shape as the tsvector-typed GIN,
    // so the MATCH-AGAINST `@@` predicate (desugared by the parser
    // into `to_tsvector(col) @@ plainto_tsquery('term')`) routes
    // through the same candidate-set seek.
    let idx = table
        .indices()
        .iter()
        .find(|i| i.column_position == col_pos && (i.is_gin() || i.is_gin_fulltext()))?;
    let candidates = gin_query_candidates(idx, &query)?;
    let _ = catalog; // cold-tier row resolution unused in MVP; see below.
    let mut out: Vec<Cow<'a, Row>> = Vec::with_capacity(candidates.len());
    for loc in candidates {
        match loc {
            spg_storage::RowLocator::Hot(i) => {
                // Phase C.3 step 2c — MVCC read gate. No-op today.
                if !table.is_row_visible(i, snapshot) {
                    continue;
                }
                if let Some(row) = table.rows().get(i) {
                    out.push(Cow::Borrowed(row));
                }
            }
            // GIN cold-tier rows in the MVP: skipped, matching the
            // full-scan `@@` path which itself only iterates
            // `table.rows()` (hot tier). When v7.13+ adds cold-tier
            // scan-time materialisation for `@@`, the parallel
            // resolution lands here; until then both paths see the
            // same hot-only candidate set so correctness is preserved.
            spg_storage::RowLocator::Cold { .. } => {}
        }
    }
    Some(out)
}

/// v7.37.8(sentori Epic 5 P2)— JSONB-GIN candidate seek for
/// `WHERE col @> <jsonb_literal>`. Mirrors `try_gin_seek`'s
/// AND walker(individual GIN seeks union safely under AND)but
/// drops OR — a non-overlapping containment query on one branch
/// vs another would broaden the candidate set unsafely without
/// the OR's full-result containment requirement that's already
/// checked per row downstream.
pub(crate) fn try_gin_jsonb_seek<'a>(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    table: &'a Table,
    table_alias: &str,
    snapshot: &spg_storage::snapshot::Snapshot,
) -> Option<Vec<Cow<'a, Row<'static>>>> {
    if let Expr::Binary {
        lhs,
        op: BinOp::And,
        rhs,
    } = where_expr
    {
        if let Some(rows) = try_gin_jsonb_seek(lhs, schema_cols, table, table_alias, snapshot) {
            return Some(rows);
        }
        return try_gin_jsonb_seek(rhs, schema_cols, table, table_alias, snapshot);
    }
    let Expr::Binary {
        lhs,
        op: BinOp::JsonContains,
        rhs,
    } = where_expr
    else {
        return None;
    };
    // Column on the left, jsonb literal on the right — sentori's
    // shape. Resolve column ↔ literal generally so `<lit> @> <col>`
    // (PG also accepts) lands here too(swap drops the seek because
    // `lit @> col` requires the literal to contain a column-defined
    // value, which can't be answered by a constant token lookup).
    let col_pos = resolve_jsonb_column(lhs, schema_cols, table_alias)?;
    let literal = resolve_jsonb_literal(rhs)?;
    let idx = table
        .indices()
        .iter()
        .find(|i| i.column_position == col_pos && i.is_gin_jsonb())?;
    let tokens = spg_storage::jsonb_gin::extract_tokens(&literal);
    if tokens.is_empty() {
        // An empty token list — `'{}' @> '{}'` — is trivially true
        // for every row. Best to let the full scan handle it so the
        // existing eval path renders the empty-containment answer.
        return None;
    }
    // Intersect posting lists.
    let mut candidates: Vec<spg_storage::RowLocator> = idx.gin_jsonb_lookup(&tokens[0]).to_vec();
    candidates.sort_by_key(locator_sort_key);
    candidates.dedup_by_key(|l| locator_sort_key(l));
    for tok in &tokens[1..] {
        let mut next: Vec<spg_storage::RowLocator> = idx.gin_jsonb_lookup(tok).to_vec();
        next.sort_by_key(locator_sort_key);
        next.dedup_by_key(|l| locator_sort_key(l));
        // Sorted-merge intersection.
        let mut out: Vec<spg_storage::RowLocator> = Vec::new();
        let (mut i, mut j) = (0usize, 0usize);
        while i < candidates.len() && j < next.len() {
            let lk = locator_sort_key(&candidates[i]);
            let rk = locator_sort_key(&next[j]);
            match lk.cmp(&rk) {
                core::cmp::Ordering::Less => i += 1,
                core::cmp::Ordering::Greater => j += 1,
                core::cmp::Ordering::Equal => {
                    out.push(candidates[i]);
                    i += 1;
                    j += 1;
                }
            }
        }
        candidates = out;
        if candidates.is_empty() {
            break;
        }
    }
    let mut out: Vec<Cow<'a, Row>> = Vec::with_capacity(candidates.len());
    for loc in candidates {
        // Phase C.3 step 2c — MVCC read gate on the Hot arm. No-op today.
        if let spg_storage::RowLocator::Hot(i) = loc
            && table.is_row_visible(i, snapshot)
            && let Some(row) = table.rows().get(i)
        {
            out.push(Cow::Borrowed(row));
        }
    }
    Some(out)
}

fn resolve_jsonb_column(
    e: &Expr,
    schema_cols: &[ColumnSchema],
    table_alias: &str,
) -> Option<usize> {
    if let Expr::Column(c) = e {
        if let Some(q) = &c.qualifier
            && !q.eq_ignore_ascii_case(table_alias)
        {
            return None;
        }
        let pos = schema_cols
            .iter()
            .position(|s| s.name.eq_ignore_ascii_case(&c.name))?;
        if matches!(
            schema_cols[pos].ty,
            spg_storage::DataType::Json | spg_storage::DataType::Jsonb
        ) {
            return Some(pos);
        }
    }
    None
}

fn resolve_jsonb_literal(e: &Expr) -> Option<alloc::string::String> {
    use spg_sql::ast::Literal;
    match e {
        // `'{"team":"ios"}'::jsonb` is the canonical sentori shape.
        // The parser surfaces it as `Cast { expr: Literal(String),
        // target: Jsonb }`. We accept any cast target since the
        // engine's full `@>` re-eval guards correctness; here we
        // only need the string body so the JSONB tokenizer can run.
        Expr::Cast { expr, .. } => match expr.as_ref() {
            Expr::Literal(Literal::String(s)) => Some(s.clone()),
            _ => None,
        },
        // Bare `Literal::String` works too — `WHERE col @> '{}'`
        // without a cast still names a JSONB-shape literal because
        // the operator's left side is JSONB-typed.
        Expr::Literal(Literal::String(s)) => Some(s.clone()),
        _ => None,
    }
}

/// v7.15.0 — trigram-GIN-accelerated candidate seek for
/// `WHERE col LIKE '<pat>'` and `WHERE col ILIKE '<pat>'` when
/// the column has a `gin_trgm_ops` GIN index.
///
/// Walks top-level `AND` so multi-predicate WHEREs (`col LIKE
/// 'foo%' AND id > 1`) still hit the trigram index; the caller
/// re-evaluates the full WHERE per candidate row, so dropping
/// non-LIKE conjuncts here stays semantically correct.
///
/// Returns `None` when:
///   - no leaf is `col LIKE/ILIKE <literal>` on a trigram-GIN-
///     indexed column;
///   - the pattern's literal runs are too short to constrain
///     (pattern decomposes into `< 3`-char runs, e.g. `%ab%`);
///   - the pattern doesn't const-evaluate to a TEXT.
pub(crate) fn try_trgm_seek<'a>(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    table: &'a Table,
    table_alias: &str,
    snapshot: &spg_storage::snapshot::Snapshot,
) -> Option<Vec<Cow<'a, Row<'static>>>> {
    if let Expr::Binary {
        lhs,
        op: BinOp::And,
        rhs,
    } = where_expr
    {
        if let Some(rows) = try_trgm_seek(lhs, schema_cols, table, table_alias, snapshot) {
            return Some(rows);
        }
        return try_trgm_seek(rhs, schema_cols, table, table_alias, snapshot);
    }
    // LIKE node is what carries the column reference + pattern.
    // ILIKE is the same AST node — PG's LIKE/ILIKE both lower
    // through `Expr::Like { expr, pattern, negated }`. The trigram
    // index posting-list keys are already lower-cased and
    // case-folded, so we only need the pattern's literal runs.
    let Expr::Like { expr, pattern, .. } = where_expr else {
        return None;
    };
    // Column side.
    let Expr::Column(c) = expr.as_ref() else {
        return None;
    };
    if let Some(q) = &c.qualifier
        && q != table_alias
    {
        return None;
    }
    let col_pos = schema_cols
        .iter()
        .position(|s| s.name.eq_ignore_ascii_case(&c.name))?;
    // Index must exist on that column AND be a trigram-GIN.
    let idx = table
        .indices()
        .iter()
        .find(|i| i.column_position == col_pos && i.is_gin_trgm())?;
    // Pattern side must be a literal TEXT — anything else (column
    // ref, function call, parameter that hasn't been bound yet)
    // falls through to full scan.
    let Expr::Literal(spg_sql::ast::Literal::String(pat)) = pattern.as_ref() else {
        return None;
    };
    let trigrams = spg_storage::trgm::trigrams_from_like_pattern(pat)?;
    // Intersect every trigram's posting list. Empty intersection
    // → empty candidate set (caller short-circuits its row loop).
    let mut iter = trigrams.iter();
    let first = iter.next()?;
    let mut acc: Vec<spg_storage::RowLocator> = {
        let mut v = idx
            .gin_trgm_lookup(spg_storage::trgm::trigram_str(first))
            .to_vec();
        v.sort_by_key(locator_sort_key);
        v.dedup_by_key(|l| locator_sort_key(l));
        v
    };
    for tri in iter {
        let mut next: Vec<spg_storage::RowLocator> = idx
            .gin_trgm_lookup(spg_storage::trgm::trigram_str(tri))
            .to_vec();
        next.sort_by_key(locator_sort_key);
        next.dedup_by_key(|l| locator_sort_key(l));
        // Sorted-merge intersection.
        let mut merged: Vec<spg_storage::RowLocator> =
            Vec::with_capacity(acc.len().min(next.len()));
        let (mut i, mut j) = (0usize, 0usize);
        while i < acc.len() && j < next.len() {
            let lk = locator_sort_key(&acc[i]);
            let rk = locator_sort_key(&next[j]);
            match lk.cmp(&rk) {
                core::cmp::Ordering::Less => i += 1,
                core::cmp::Ordering::Greater => j += 1,
                core::cmp::Ordering::Equal => {
                    merged.push(acc[i]);
                    i += 1;
                    j += 1;
                }
            }
        }
        acc = merged;
        if acc.is_empty() {
            break;
        }
    }
    let mut out: Vec<Cow<'a, Row>> = Vec::with_capacity(acc.len());
    for loc in acc {
        // Phase C.3 step 2c — MVCC read gate on the Hot arm. No-op today.
        if let spg_storage::RowLocator::Hot(i) = loc
            && table.is_row_visible(i, snapshot)
            && let Some(row) = table.rows().get(i)
        {
            out.push(Cow::Borrowed(row));
        }
        // Cold-tier rows: skipped in MVP (same as try_gin_seek).
    }
    Some(out)
}

/// v7.12.3 — extract `(column_position, TsQueryAst)` when one side of
/// the binary is a column reference to a GIN-indexed tsvector column
/// and the other side const-evaluates to a `Value::TsQuery`. Returns
/// `None` if the column reference is for the wrong table alias, or if
/// the RHS expression depends on row data.
pub(crate) fn resolve_gin_col_query(
    col_side: &Expr,
    query_side: &Expr,
    schema_cols: &[ColumnSchema],
    table_alias: &str,
    ctx: &eval::EvalContext<'_>,
) -> Option<(usize, spg_storage::TsQueryAst)> {
    // v7.17.0 Phase 3.P0-44 — the MATCH AGAINST desugar wraps the
    // column in `to_tsvector('simple', col)`, so we peel that wrapper
    // before the column lookup. Direct `col @@ tsquery` paths (the
    // tsvector-typed v7.12 surface) skip the wrapper entirely.
    let column = match col_side {
        Expr::Column(c) => c,
        Expr::FunctionCall { name, args }
            if name.eq_ignore_ascii_case("to_tsvector") && !args.is_empty() =>
        {
            // PG `to_tsvector` accepts either `to_tsvector(col)` or
            // `to_tsvector(config, col)`. In both shapes the column
            // we care about is the final argument.
            if let Expr::Column(c) = args.last().unwrap() {
                c
            } else {
                return None;
            }
        }
        _ => return None,
    };
    let c = column;
    if let Some(q) = &c.qualifier
        && q != table_alias
    {
        return None;
    }
    let pos = schema_cols.iter().position(|s| s.name == c.name)?;
    // Const-evaluate the query side with an empty row — fails fast
    // (with a `ColumnNotFound` / similar) if the expression actually
    // depends on row data, which is exactly the bail signal we want.
    let empty_row = Row::new(Vec::new());
    let v = eval::eval_expr(query_side, &empty_row, ctx).ok()?;
    let Value::TsQuery(q) = v else { return None };
    Some((pos, q))
}

/// v7.12.3 — walk a `TsQueryAst` against an [`IndexKind::Gin`] index
/// to produce a candidate row-locator set. Returns `None` for query
/// shapes the MVP doesn't accelerate (`Not` / `Phrase` — both bail to
/// full scan since their semantics need either complementation across
/// the whole row set or positional verification beyond what the
/// posting list carries).
///
/// Candidate sets are over-approximate — the caller re-applies the
/// full `@@` predicate per row, so reporting "row was in some
/// posting list" without verifying positions / weights stays correct.
pub(crate) fn gin_query_candidates(
    idx: &spg_storage::Index,
    query: &spg_storage::TsQueryAst,
) -> Option<Vec<spg_storage::RowLocator>> {
    use spg_storage::TsQueryAst;
    match query {
        TsQueryAst::Term { word, .. } => {
            let mut v: Vec<spg_storage::RowLocator> = idx.gin_lookup_word(word).to_vec();
            v.sort_by_key(locator_sort_key);
            v.dedup_by_key(|l| locator_sort_key(l));
            Some(v)
        }
        TsQueryAst::And(l, r) => {
            let mut left = gin_query_candidates(idx, l)?;
            let mut right = gin_query_candidates(idx, r)?;
            left.sort_by_key(locator_sort_key);
            right.sort_by_key(locator_sort_key);
            // Sorted-merge intersection.
            let mut out: Vec<spg_storage::RowLocator> = Vec::new();
            let (mut i, mut j) = (0usize, 0usize);
            while i < left.len() && j < right.len() {
                let lk = locator_sort_key(&left[i]);
                let rk = locator_sort_key(&right[j]);
                match lk.cmp(&rk) {
                    core::cmp::Ordering::Less => i += 1,
                    core::cmp::Ordering::Greater => j += 1,
                    core::cmp::Ordering::Equal => {
                        out.push(left[i]);
                        i += 1;
                        j += 1;
                    }
                }
            }
            Some(out)
        }
        TsQueryAst::Or(l, r) => {
            let mut out = gin_query_candidates(idx, l)?;
            out.extend(gin_query_candidates(idx, r)?);
            out.sort_by_key(locator_sort_key);
            out.dedup_by_key(|l| locator_sort_key(l));
            Some(out)
        }
        // Not / Phrase bail to full scan in the MVP. Not needs
        // complementation against the whole row set (not represented
        // in the posting-list view); Phrase needs positional
        // verification beyond what `word → rows` carries.
        TsQueryAst::Not(_) | TsQueryAst::Phrase { .. } => None,
    }
}

/// v7.12.3 — total ordering on `RowLocator` for sort/dedup purposes
/// inside the GIN intersection / union loops. Hot rows order by their
/// row index; Cold rows order after all Hot rows, then by
/// `(segment_id, the cold sub-key)`.
pub(crate) fn locator_sort_key(l: &spg_storage::RowLocator) -> (u8, u64, u64) {
    match *l {
        spg_storage::RowLocator::Hot(i) => (0, i as u64, 0),
        spg_storage::RowLocator::Cold {
            segment_id,
            page_offset,
        } => (1, u64::from(segment_id), u64::from(page_offset)),
    }
}

/// v5.2.3: extract `(column_position, IndexKey)` when `where_expr`
/// is a simple `col = literal` predicate suitable for a `BTree` index
/// seek. Used by `exec_update_cancel` / `exec_delete_cancel` to
/// decide whether a write touches a cold-tier row (which requires
/// promote-on-write / shadow-on-delete) before falling through to
/// the hot-tier row walk.
///
/// Returns `None` for any predicate shape the planner can't push
/// down to an index seek — complex WHERE clauses always take the
/// hot-only path (cold rows are immutable to non-indexed writes
/// until a future scan-fanout sub-version).
pub(crate) fn try_pk_predicate(
    where_expr: &Expr,
    schema_cols: &[ColumnSchema],
    table_alias: &str,
) -> Option<(usize, IndexKey)> {
    let Expr::Binary {
        lhs,
        op: BinOp::Eq,
        rhs,
    } = where_expr
    else {
        return None;
    };
    let (col_pos, value) = resolve_col_literal_pair(lhs, rhs, schema_cols, table_alias)
        .or_else(|| resolve_col_literal_pair(rhs, lhs, schema_cols, table_alias))?;
    let key = IndexKey::from_value(&value)?;
    Some((col_pos, key))
}

pub(crate) fn resolve_col_literal_pair(
    col_side: &Expr,
    lit_side: &Expr,
    schema_cols: &[ColumnSchema],
    table_alias: &str,
) -> Option<(usize, Value<'static>)> {
    let Expr::Column(c) = col_side else {
        return None;
    };
    if let Some(q) = &c.qualifier
        && q != table_alias
    {
        return None;
    }
    let pos = schema_cols.iter().position(|s| s.name == c.name)?;
    let Expr::Literal(l) = lit_side else {
        return None;
    };
    let v = match l {
        Literal::Integer(n) => {
            if let Ok(small) = i32::try_from(*n) {
                Value::Int(small)
            } else {
                Value::BigInt(*n)
            }
        }
        Literal::Float(x) => Value::Float(*x),
        Literal::Numeric { unscaled, scale } => Value::Numeric {
            scaled: *unscaled,
            scale: *scale,
            kind: spg_storage::NumericKind::Finite,
        },
        Literal::NumericBig(s) => crate::conversions::big_literal_to_value(s),
        Literal::String(s) => Value::text(s.clone()),
        Literal::Bool(b) => Value::Bool(*b),
        Literal::Null => Value::Null,
        // Vector, array and Interval literals can't be used as B-tree
        // index keys. Tell the planner to fall back to full-scan.
        Literal::Vector(_)
        | Literal::Interval { .. }
        | Literal::TextArray(_)
        | Literal::IntArray(_)
        | Literal::BigIntArray(_) => return None,
    };
    // v7.39 (round 564) — a string literal means whatever the COLUMN
    // says it means, and until here it always meant text.
    //
    // `WHERE d = '2026-01-02'` on a `DATE` column built an
    // `IndexKey::Text`, while the rows under that index are keyed
    // `IndexKey::Int` — days. The seek looked in a key space nothing
    // lives in and found nothing, so:
    //
    //     no index     d = '2026-01-02'   ->  1 row   (PG: 1)
    //     with index   d = '2026-01-02'   ->  0 rows  (PG: 1)
    //
    // Creating an index changed the answer, which is the one thing an
    // index may never do. Two-sided ranges went the same way; a
    // one-sided `d > '…'` survived only because it is not seeked at all.
    // Every string-literal comparison against a date, timestamp, time,
    // uuid or bool column was affected.
    //
    // Only string literals, and only against a non-text column: parsing
    // a string to its column type is exact or it raises, so a
    // coercion that succeeds cannot seek to the wrong key. Numeric
    // coercions are deliberately NOT done here — `WHERE i = 1.5` on an
    // integer column must not round its way to the rows holding 2.
    // A coercion that fails means the ordinary path re-raises it.
    let ty = schema_cols[pos].ty;
    if matches!(l, Literal::String(_))
        && !matches!(
            ty,
            spg_storage::DataType::Text
                | spg_storage::DataType::Varchar(_)
                | spg_storage::DataType::Char(_)
        )
    {
        return Some((
            pos,
            crate::conversions::coerce_value(v, ty, &c.name, pos).ok()?,
        ));
    }
    Some((pos, v))
}