sqlrite/sql/

executor.rs

//! Query executors — evaluate parsed SQL statements against the in-memory
//! storage and produce formatted output.

use std::cmp::Ordering;

use prettytable::{Cell as PrintCell, Row as PrintRow, Table as PrintTable};
use sqlparser::ast::{
    AlterTable, AlterTableOperation, AssignmentTarget, BinaryOperator, CreateIndex, Delete, Expr,
    FromTable, FunctionArg, FunctionArgExpr, FunctionArguments, IndexType, ObjectName,
    ObjectNamePart, RenameTableNameKind, Statement, TableFactor, TableWithJoins, UnaryOperator,
    Update, Value as AstValue,
};

use crate::error::{Result, SQLRiteError};
use crate::sql::db::database::Database;
use crate::sql::db::secondary_index::{IndexOrigin, SecondaryIndex};
use crate::sql::db::table::{
    DataType, FtsIndexEntry, HnswIndexEntry, Table, Value, parse_vector_literal,
};
use crate::sql::fts::{Bm25Params, PostingList};
use crate::sql::hnsw::{DistanceMetric, HnswIndex};
use crate::sql::parser::select::{OrderByClause, Projection, SelectQuery};

/// Executes a parsed `SelectQuery` against the database and returns a
/// human-readable rendering of the result set (prettytable). Also returns
/// the number of rows produced, for the top-level status message.
/// Structured result of a SELECT: column names in projection order,
/// and each matching row as a `Vec<Value>` aligned with the columns.
/// Phase 5a introduced this so the public `Connection` / `Statement`
/// API has typed rows to yield; the existing `execute_select` that
/// returns pre-rendered text is now a thin wrapper on top.
pub struct SelectResult {
    pub columns: Vec<String>,
    pub rows: Vec<Vec<Value>>,
}

/// Executes a SELECT and returns structured rows. The typed rows are
/// what the new public API streams to callers; the REPL / Tauri app
/// pre-render into a prettytable via `execute_select`.
pub fn execute_select_rows(query: SelectQuery, db: &Database) -> Result<SelectResult> {
    let table = db
        .get_table(query.table_name.clone())
        .map_err(|_| SQLRiteError::Internal(format!("Table '{}' not found", query.table_name)))?;

    // Resolve projection to a concrete ordered column list.
    let projected_cols: Vec<String> = match &query.projection {
        Projection::All => table.column_names(),
        Projection::Columns(cols) => {
            for c in cols {
                if !table.contains_column(c.to_string()) {
                    return Err(SQLRiteError::Internal(format!(
                        "Column '{c}' does not exist on table '{}'",
                        query.table_name
                    )));
                }
            }
            cols.clone()
        }
    };

    // Collect matching rowids. If the WHERE is the shape `col = literal`
    // and `col` has a secondary index, probe the index for an O(log N)
    // seek; otherwise fall back to the full table scan.
    let matching = match select_rowids(table, query.selection.as_ref())? {
        RowidSource::IndexProbe(rowids) => rowids,
        RowidSource::FullScan => {
            let mut out = Vec::new();
            for rowid in table.rowids() {
                if let Some(expr) = &query.selection {
                    if !eval_predicate(expr, table, rowid)? {
                        continue;
                    }
                }
                out.push(rowid);
            }
            out
        }
    };
    let mut matching = matching;

    // Phase 7c — bounded-heap top-k optimization.
    //
    // The naive "ORDER BY <expr>" path (Phase 7b) sorts every matching
    // rowid: O(N log N) sort_by + a truncate. For KNN queries
    //
    //     SELECT id FROM docs
    //     ORDER BY vec_distance_l2(embedding, [...])
    //     LIMIT 10;
    //
    // N is the table row count and k is the LIMIT. With a bounded
    // max-heap of size k we can find the top-k in O(N log k) — same
    // sort_by-per-row cost on the heap operations, but k is typically
    // 10-100 while N can be millions.
    //
    // Phase 7d.2 — HNSW ANN probe.
    //
    // Even better than the bounded heap: if the ORDER BY expression is
    // exactly `vec_distance_l2(<col>, <bracket-array literal>)` AND
    // `<col>` has an HNSW index attached, skip the linear scan
    // entirely and probe the graph in O(log N). Approximate but
    // typically ≥ 0.95 recall (verified by the recall tests in
    // src/sql/hnsw.rs).
    //
    // We branch in cases:
    //   1. ORDER BY + LIMIT k matches the HNSW probe pattern  → graph probe.
    //   2. ORDER BY + LIMIT k matches the FTS probe pattern   → posting probe.
    //   3. ORDER BY + LIMIT k where k < |matching|            → bounded heap (7c).
    //   4. ORDER BY without LIMIT, or LIMIT >= |matching|     → full sort.
    //   5. LIMIT without ORDER BY                              → just truncate.
    match (&query.order_by, query.limit) {
        (Some(order), Some(k)) if try_hnsw_probe(table, &order.expr, k).is_some() => {
            matching = try_hnsw_probe(table, &order.expr, k).unwrap();
        }
        (Some(order), Some(k))
            if try_fts_probe(table, &order.expr, order.ascending, k).is_some() =>
        {
            matching = try_fts_probe(table, &order.expr, order.ascending, k).unwrap();
        }
        (Some(order), Some(k)) if k < matching.len() => {
            matching = select_topk(&matching, table, order, k)?;
        }
        (Some(order), _) => {
            sort_rowids(&mut matching, table, order)?;
            if let Some(k) = query.limit {
                matching.truncate(k);
            }
        }
        (None, Some(k)) => {
            matching.truncate(k);
        }
        (None, None) => {}
    }

    // Build typed rows. Missing cells surface as `Value::Null` — that
    // maps a column-not-present-for-this-rowid case onto the public
    // `Row::get` → `Option<T>` surface cleanly.
    let mut rows: Vec<Vec<Value>> = Vec::with_capacity(matching.len());
    for rowid in &matching {
        let row: Vec<Value> = projected_cols
            .iter()
            .map(|col| table.get_value(col, *rowid).unwrap_or(Value::Null))
            .collect();
        rows.push(row);
    }

    Ok(SelectResult {
        columns: projected_cols,
        rows,
    })
}

/// Executes a SELECT and returns `(rendered_table, row_count)`. The
/// REPL and Tauri app use this to keep the table-printing behaviour
/// the engine has always shipped. Structured callers use
/// `execute_select_rows` instead.
pub fn execute_select(query: SelectQuery, db: &Database) -> Result<(String, usize)> {
    let result = execute_select_rows(query, db)?;
    let row_count = result.rows.len();

    let mut print_table = PrintTable::new();
    let header_cells: Vec<PrintCell> = result.columns.iter().map(|c| PrintCell::new(c)).collect();
    print_table.add_row(PrintRow::new(header_cells));

    for row in &result.rows {
        let cells: Vec<PrintCell> = row
            .iter()
            .map(|v| PrintCell::new(&v.to_display_string()))
            .collect();
        print_table.add_row(PrintRow::new(cells));
    }

    Ok((print_table.to_string(), row_count))
}

/// Executes a DELETE statement. Returns the number of rows removed.
pub fn execute_delete(stmt: &Statement, db: &mut Database) -> Result<usize> {
    let Statement::Delete(Delete {
        from, selection, ..
    }) = stmt
    else {
        return Err(SQLRiteError::Internal(
            "execute_delete called on a non-DELETE statement".to_string(),
        ));
    };

    let tables = match from {
        FromTable::WithFromKeyword(t) | FromTable::WithoutKeyword(t) => t,
    };
    let table_name = extract_single_table_name(tables)?;

    // Compute matching rowids with an immutable borrow, then mutate.
    let matching: Vec<i64> = {
        let table = db
            .get_table(table_name.clone())
            .map_err(|_| SQLRiteError::Internal(format!("Table '{table_name}' not found")))?;
        match select_rowids(table, selection.as_ref())? {
            RowidSource::IndexProbe(rowids) => rowids,
            RowidSource::FullScan => {
                let mut out = Vec::new();
                for rowid in table.rowids() {
                    if let Some(expr) = selection {
                        if !eval_predicate(expr, table, rowid)? {
                            continue;
                        }
                    }
                    out.push(rowid);
                }
                out
            }
        }
    };

    let table = db.get_table_mut(table_name)?;
    for rowid in &matching {
        table.delete_row(*rowid);
    }
    // Phase 7d.3 — any DELETE invalidates every HNSW index on this
    // table (the deleted node could still appear in other nodes'
    // neighbor lists, breaking subsequent searches). Mark dirty so
    // the next save rebuilds from current rows before serializing.
    //
    // Phase 8b — same posture for FTS indexes (Q7 — rebuild-on-save
    // mirrors HNSW). The deleted rowid still appears in posting
    // lists; leaving it would surface zombie hits in future queries.
    if !matching.is_empty() {
        for entry in &mut table.hnsw_indexes {
            entry.needs_rebuild = true;
        }
        for entry in &mut table.fts_indexes {
            entry.needs_rebuild = true;
        }
    }
    Ok(matching.len())
}

/// Executes an UPDATE statement. Returns the number of rows updated.
pub fn execute_update(stmt: &Statement, db: &mut Database) -> Result<usize> {
    let Statement::Update(Update {
        table,
        assignments,
        from,
        selection,
        ..
    }) = stmt
    else {
        return Err(SQLRiteError::Internal(
            "execute_update called on a non-UPDATE statement".to_string(),
        ));
    };

    if from.is_some() {
        return Err(SQLRiteError::NotImplemented(
            "UPDATE ... FROM is not supported yet".to_string(),
        ));
    }

    let table_name = extract_table_name(table)?;

    // Resolve assignment targets to plain column names and verify they exist.
    let mut parsed_assignments: Vec<(String, Expr)> = Vec::with_capacity(assignments.len());
    {
        let tbl = db
            .get_table(table_name.clone())
            .map_err(|_| SQLRiteError::Internal(format!("Table '{table_name}' not found")))?;
        for a in assignments {
            let col = match &a.target {
                AssignmentTarget::ColumnName(name) => name
                    .0
                    .last()
                    .map(|p| p.to_string())
                    .ok_or_else(|| SQLRiteError::Internal("empty column name".to_string()))?,
                AssignmentTarget::Tuple(_) => {
                    return Err(SQLRiteError::NotImplemented(
                        "tuple assignment targets are not supported".to_string(),
                    ));
                }
            };
            if !tbl.contains_column(col.clone()) {
                return Err(SQLRiteError::Internal(format!(
                    "UPDATE references unknown column '{col}'"
                )));
            }
            parsed_assignments.push((col, a.value.clone()));
        }
    }

    // Gather matching rowids + the new values to write for each assignment, under
    // an immutable borrow. Uses the index-probe fast path when the WHERE is
    // `col = literal` on an indexed column.
    let work: Vec<(i64, Vec<(String, Value)>)> = {
        let tbl = db.get_table(table_name.clone())?;
        let matched_rowids: Vec<i64> = match select_rowids(tbl, selection.as_ref())? {
            RowidSource::IndexProbe(rowids) => rowids,
            RowidSource::FullScan => {
                let mut out = Vec::new();
                for rowid in tbl.rowids() {
                    if let Some(expr) = selection {
                        if !eval_predicate(expr, tbl, rowid)? {
                            continue;
                        }
                    }
                    out.push(rowid);
                }
                out
            }
        };
        let mut rows_to_update = Vec::new();
        for rowid in matched_rowids {
            let mut values = Vec::with_capacity(parsed_assignments.len());
            for (col, expr) in &parsed_assignments {
                // UPDATE's RHS is evaluated in the context of the row being updated,
                // so column references on the right resolve to the current row's values.
                let v = eval_expr(expr, tbl, rowid)?;
                values.push((col.clone(), v));
            }
            rows_to_update.push((rowid, values));
        }
        rows_to_update
    };

    let tbl = db.get_table_mut(table_name)?;
    for (rowid, values) in &work {
        for (col, v) in values {
            tbl.set_value(col, *rowid, v.clone())?;
        }
    }

    // Phase 7d.3 — UPDATE may have changed a vector column that an
    // HNSW index covers. Mark every covering index dirty so save
    // rebuilds from current rows. (Updates that only touched
    // non-vector columns also mark dirty, which is over-conservative
    // but harmless — the rebuild walks rows anyway, and the cost is
    // only paid on save.)
    //
    // Phase 8b — same shape for FTS indexes covering updated TEXT cols.
    if !work.is_empty() {
        let updated_columns: std::collections::HashSet<&str> = work
            .iter()
            .flat_map(|(_, values)| values.iter().map(|(c, _)| c.as_str()))
            .collect();
        for entry in &mut tbl.hnsw_indexes {
            if updated_columns.contains(entry.column_name.as_str()) {
                entry.needs_rebuild = true;
            }
        }
        for entry in &mut tbl.fts_indexes {
            if updated_columns.contains(entry.column_name.as_str()) {
                entry.needs_rebuild = true;
            }
        }
    }
    Ok(work.len())
}

/// Handles `CREATE INDEX [UNIQUE] <name> ON <table> [USING <method>] (<column>)`.
/// Single-column indexes only.
///
/// Two flavours, branching on the optional `USING <method>` clause:
///   - **No USING, or `USING btree`**: regular B-Tree secondary index
///     (Phase 3e). Indexable types: Integer, Text.
///   - **`USING hnsw`**: HNSW ANN index (Phase 7d.2). Indexable types:
///     Vector(N) only. Distance metric is L2 by default; cosine and
///     dot variants are deferred to Phase 7d.x.
///
/// Returns the (possibly synthesized) index name for the status message.
pub fn execute_create_index(stmt: &Statement, db: &mut Database) -> Result<String> {
    let Statement::CreateIndex(CreateIndex {
        name,
        table_name,
        columns,
        using,
        unique,
        if_not_exists,
        predicate,
        ..
    }) = stmt
    else {
        return Err(SQLRiteError::Internal(
            "execute_create_index called on a non-CREATE-INDEX statement".to_string(),
        ));
    };

    if predicate.is_some() {
        return Err(SQLRiteError::NotImplemented(
            "partial indexes (CREATE INDEX ... WHERE) are not supported yet".to_string(),
        ));
    }

    if columns.len() != 1 {
        return Err(SQLRiteError::NotImplemented(format!(
            "multi-column indexes are not supported yet ({} columns given)",
            columns.len()
        )));
    }

    let index_name = name.as_ref().map(|n| n.to_string()).ok_or_else(|| {
        SQLRiteError::NotImplemented(
            "anonymous CREATE INDEX (no name) is not supported — give it a name".to_string(),
        )
    })?;

    // Detect USING <method>. The `using` field on CreateIndex covers the
    // pre-column form `CREATE INDEX … USING hnsw (col)`. (sqlparser also
    // accepts a post-column form `… (col) USING hnsw` and parks that in
    // `index_options`; we don't bother with it — the canonical form is
    // pre-column and matches PG/pgvector convention.)
    let method = match using {
        Some(IndexType::Custom(ident)) if ident.value.eq_ignore_ascii_case("hnsw") => {
            IndexMethod::Hnsw
        }
        Some(IndexType::Custom(ident)) if ident.value.eq_ignore_ascii_case("fts") => {
            IndexMethod::Fts
        }
        Some(IndexType::Custom(ident)) if ident.value.eq_ignore_ascii_case("btree") => {
            IndexMethod::Btree
        }
        Some(other) => {
            return Err(SQLRiteError::NotImplemented(format!(
                "CREATE INDEX … USING {other:?} is not supported \
                 (try `hnsw`, `fts`, or no USING clause)"
            )));
        }
        None => IndexMethod::Btree,
    };

    let table_name_str = table_name.to_string();
    let column_name = match &columns[0].column.expr {
        Expr::Identifier(ident) => ident.value.clone(),
        Expr::CompoundIdentifier(parts) => parts
            .last()
            .map(|p| p.value.clone())
            .ok_or_else(|| SQLRiteError::Internal("empty compound identifier".to_string()))?,
        other => {
            return Err(SQLRiteError::NotImplemented(format!(
                "CREATE INDEX only supports simple column references, got {other:?}"
            )));
        }
    };

    // Validate: table exists, column exists, type matches the index method,
    // name is unique across both index kinds. Snapshot (rowid, value) pairs
    // up front under the immutable borrow so the mutable attach later
    // doesn't fight over `self`.
    let (datatype, existing_rowids_and_values): (DataType, Vec<(i64, Value)>) = {
        let table = db.get_table(table_name_str.clone()).map_err(|_| {
            SQLRiteError::General(format!(
                "CREATE INDEX references unknown table '{table_name_str}'"
            ))
        })?;
        if !table.contains_column(column_name.clone()) {
            return Err(SQLRiteError::General(format!(
                "CREATE INDEX references unknown column '{column_name}' on table '{table_name_str}'"
            )));
        }
        let col = table
            .columns
            .iter()
            .find(|c| c.column_name == column_name)
            .expect("we just verified the column exists");

        // Name uniqueness check spans ALL index kinds — btree, hnsw, and
        // fts share one namespace per table.
        if table.index_by_name(&index_name).is_some()
            || table.hnsw_indexes.iter().any(|i| i.name == index_name)
            || table.fts_indexes.iter().any(|i| i.name == index_name)
        {
            if *if_not_exists {
                return Ok(index_name);
            }
            return Err(SQLRiteError::General(format!(
                "index '{index_name}' already exists"
            )));
        }
        let datatype = clone_datatype(&col.datatype);

        let mut pairs = Vec::new();
        for rowid in table.rowids() {
            if let Some(v) = table.get_value(&column_name, rowid) {
                pairs.push((rowid, v));
            }
        }
        (datatype, pairs)
    };

    match method {
        IndexMethod::Btree => create_btree_index(
            db,
            &table_name_str,
            &index_name,
            &column_name,
            &datatype,
            *unique,
            &existing_rowids_and_values,
        ),
        IndexMethod::Hnsw => create_hnsw_index(
            db,
            &table_name_str,
            &index_name,
            &column_name,
            &datatype,
            *unique,
            &existing_rowids_and_values,
        ),
        IndexMethod::Fts => create_fts_index(
            db,
            &table_name_str,
            &index_name,
            &column_name,
            &datatype,
            *unique,
            &existing_rowids_and_values,
        ),
    }
}

/// Executes `DROP TABLE [IF EXISTS] <name>;`. Mirrors SQLite's single-target
/// shape: sqlparser parses `DROP TABLE a, b` as one statement with
/// `names: vec![a, b]`, but we reject the multi-target form to keep error
/// semantics simple (no partial-failure rollback).
///
/// On success the table — and every index attached to it — disappears from
/// the in-memory `Database`. The next auto-save rebuilds `sqlrite_master`
/// from scratch and simply doesn't write a row for the dropped table or
/// its indexes; pages previously occupied by them become orphans on disk
/// (no free-list yet — file size doesn't shrink until a future VACUUM).
pub fn execute_drop_table(
    names: &[ObjectName],
    if_exists: bool,
    db: &mut Database,
) -> Result<usize> {
    if names.len() != 1 {
        return Err(SQLRiteError::NotImplemented(
            "DROP TABLE supports a single table per statement".to_string(),
        ));
    }
    let name = names[0].to_string();

    if name == crate::sql::pager::MASTER_TABLE_NAME {
        return Err(SQLRiteError::General(format!(
            "'{}' is a reserved name used by the internal schema catalog",
            crate::sql::pager::MASTER_TABLE_NAME
        )));
    }

    if !db.contains_table(name.clone()) {
        return if if_exists {
            Ok(0)
        } else {
            Err(SQLRiteError::General(format!(
                "Table '{name}' does not exist"
            )))
        };
    }

    db.tables.remove(&name);
    Ok(1)
}

/// Executes `DROP INDEX [IF EXISTS] <name>;`. The statement does not name a
/// table, so we walk every table looking for the index across all three
/// index families (B-Tree secondary, HNSW, FTS).
///
/// Refuses to drop auto-indexes (`origin == IndexOrigin::Auto`) — those are
/// invariants of the table's PRIMARY KEY / UNIQUE constraints and should
/// only disappear when the column or table they depend on is dropped.
/// SQLite has the same rule for its `sqlite_autoindex_*` indexes.
pub fn execute_drop_index(
    names: &[ObjectName],
    if_exists: bool,
    db: &mut Database,
) -> Result<usize> {
    if names.len() != 1 {
        return Err(SQLRiteError::NotImplemented(
            "DROP INDEX supports a single index per statement".to_string(),
        ));
    }
    let name = names[0].to_string();

    for table in db.tables.values_mut() {
        if let Some(secondary) = table.secondary_indexes.iter().find(|i| i.name == name) {
            if secondary.origin == IndexOrigin::Auto {
                return Err(SQLRiteError::General(format!(
                    "cannot drop auto-created index '{name}' (drop the column or table instead)"
                )));
            }
            table.secondary_indexes.retain(|i| i.name != name);
            return Ok(1);
        }
        if table.hnsw_indexes.iter().any(|i| i.name == name) {
            table.hnsw_indexes.retain(|i| i.name != name);
            return Ok(1);
        }
        if table.fts_indexes.iter().any(|i| i.name == name) {
            table.fts_indexes.retain(|i| i.name != name);
            return Ok(1);
        }
    }

    if if_exists {
        Ok(0)
    } else {
        Err(SQLRiteError::General(format!(
            "Index '{name}' does not exist"
        )))
    }
}

/// Executes `ALTER TABLE [IF EXISTS] <name> <op>;` for one operation per
/// statement. Supports four sub-operations matching SQLite:
///
///   - `RENAME TO <new>`
///   - `RENAME COLUMN <old> TO <new>`
///   - `ADD COLUMN <coldef>` (NOT NULL requires DEFAULT on a non-empty table;
///     PK / UNIQUE constraints rejected — would need backfill + uniqueness)
///   - `DROP COLUMN <name>` (refuses PK column and only-column)
///
/// Multi-operation ALTER (`ALTER TABLE foo RENAME TO bar, ADD COLUMN x ...`)
/// is rejected; SQLite forbids it too.
pub fn execute_alter_table(alter: AlterTable, db: &mut Database) -> Result<String> {
    let table_name = alter.name.to_string();

    if table_name == crate::sql::pager::MASTER_TABLE_NAME {
        return Err(SQLRiteError::General(format!(
            "'{}' is a reserved name used by the internal schema catalog",
            crate::sql::pager::MASTER_TABLE_NAME
        )));
    }

    if !db.contains_table(table_name.clone()) {
        return if alter.if_exists {
            Ok("ALTER TABLE: no-op (table does not exist)".to_string())
        } else {
            Err(SQLRiteError::General(format!(
                "Table '{table_name}' does not exist"
            )))
        };
    }

    if alter.operations.len() != 1 {
        return Err(SQLRiteError::NotImplemented(
            "ALTER TABLE supports one operation per statement".to_string(),
        ));
    }

    match &alter.operations[0] {
        AlterTableOperation::RenameTable { table_name: kind } => {
            let new_name = match kind {
                RenameTableNameKind::To(name) => name.to_string(),
                RenameTableNameKind::As(_) => {
                    return Err(SQLRiteError::NotImplemented(
                        "ALTER TABLE ... RENAME AS (MySQL-only) is not supported; use RENAME TO"
                            .to_string(),
                    ));
                }
            };
            alter_rename_table(db, &table_name, &new_name)?;
            Ok(format!(
                "ALTER TABLE '{table_name}' RENAME TO '{new_name}' executed."
            ))
        }
        AlterTableOperation::RenameColumn {
            old_column_name,
            new_column_name,
        } => {
            let old = old_column_name.value.clone();
            let new = new_column_name.value.clone();
            db.get_table_mut(table_name.clone())?
                .rename_column(&old, &new)?;
            Ok(format!(
                "ALTER TABLE '{table_name}' RENAME COLUMN '{old}' TO '{new}' executed."
            ))
        }
        AlterTableOperation::AddColumn {
            column_def,
            if_not_exists,
            ..
        } => {
            let parsed = crate::sql::parser::create::parse_one_column(column_def)?;
            let table = db.get_table_mut(table_name.clone())?;
            if *if_not_exists && table.contains_column(parsed.name.clone()) {
                return Ok(format!(
                    "ALTER TABLE '{table_name}' ADD COLUMN: no-op (column '{}' already exists)",
                    parsed.name
                ));
            }
            let col_name = parsed.name.clone();
            table.add_column(parsed)?;
            Ok(format!(
                "ALTER TABLE '{table_name}' ADD COLUMN '{col_name}' executed."
            ))
        }
        AlterTableOperation::DropColumn {
            column_names,
            if_exists,
            ..
        } => {
            if column_names.len() != 1 {
                return Err(SQLRiteError::NotImplemented(
                    "ALTER TABLE DROP COLUMN supports a single column per statement".to_string(),
                ));
            }
            let col_name = column_names[0].value.clone();
            let table = db.get_table_mut(table_name.clone())?;
            if *if_exists && !table.contains_column(col_name.clone()) {
                return Ok(format!(
                    "ALTER TABLE '{table_name}' DROP COLUMN: no-op (column '{col_name}' does not exist)"
                ));
            }
            table.drop_column(&col_name)?;
            Ok(format!(
                "ALTER TABLE '{table_name}' DROP COLUMN '{col_name}' executed."
            ))
        }
        other => Err(SQLRiteError::NotImplemented(format!(
            "ALTER TABLE operation {other:?} is not supported"
        ))),
    }
}

/// Executes `VACUUM;` (SQLR-6). Compacts the database file: rewrites
/// every live table, index, and the catalog contiguously from page 1,
/// drops the freelist, and truncates the tail at the next checkpoint.
///
/// Refuses to run inside a transaction (would publish in-flight writes
/// out of band); refuses on read-only databases (handled upstream by
/// the read-only mutation gate); and is a no-op on in-memory databases
/// (no file to compact). Bare `VACUUM;` only — non-default options
/// (`FULL`, `REINDEX`, table targets, etc.) are rejected.
pub fn execute_vacuum(db: &mut Database) -> Result<String> {
    if db.in_transaction() {
        return Err(SQLRiteError::General(
            "VACUUM cannot run inside a transaction".to_string(),
        ));
    }
    let path = match db.source_path.clone() {
        Some(p) => p,
        None => {
            return Ok("VACUUM is a no-op for in-memory databases".to_string());
        }
    };
    // Checkpoint before AND after VACUUM so the main-file size we report
    // reflects only what VACUUM actually reclaimed — without the leading
    // checkpoint, `size_before` would be the stale main-file snapshot
    // (typically 2 pages) while WAL holds the live bytes, making the
    // bytes-reclaimed delta meaningless.
    if let Some(pager) = db.pager.as_mut() {
        let _ = pager.checkpoint();
    }
    let size_before = std::fs::metadata(&path).ok().map(|m| m.len()).unwrap_or(0);
    let pages_before = db
        .pager
        .as_ref()
        .map(|p| p.header().page_count)
        .unwrap_or(0);
    crate::sql::pager::vacuum_database(db, &path)?;
    // Second checkpoint so the main file shrinks now — VACUUM's whole
    // purpose is to reclaim bytes, so paying the I/O up front is fair.
    if let Some(pager) = db.pager.as_mut() {
        let _ = pager.checkpoint();
    }
    let size_after = std::fs::metadata(&path).ok().map(|m| m.len()).unwrap_or(0);
    let pages_after = db
        .pager
        .as_ref()
        .map(|p| p.header().page_count)
        .unwrap_or(0);
    let pages_reclaimed = pages_before.saturating_sub(pages_after);
    let bytes_reclaimed = size_before.saturating_sub(size_after);
    Ok(format!(
        "VACUUM completed. {pages_reclaimed} pages reclaimed ({bytes_reclaimed} bytes)."
    ))
}

/// Renames a table in `db.tables`. Updates `tb_name`, every secondary
/// index's `table_name` field, and any auto-index whose name embedded
/// the old table name. HNSW / FTS index entries don't carry a
/// `table_name` field — they're addressed implicitly via the `Table`
/// they live inside, so they move with the rename for free.
fn alter_rename_table(db: &mut Database, old: &str, new: &str) -> Result<()> {
    if new == crate::sql::pager::MASTER_TABLE_NAME {
        return Err(SQLRiteError::General(format!(
            "'{}' is a reserved name used by the internal schema catalog",
            crate::sql::pager::MASTER_TABLE_NAME
        )));
    }
    if old == new {
        return Ok(());
    }
    if db.contains_table(new.to_string()) {
        return Err(SQLRiteError::General(format!(
            "target table '{new}' already exists"
        )));
    }

    let mut table = db
        .tables
        .remove(old)
        .ok_or_else(|| SQLRiteError::General(format!("Table '{old}' does not exist")))?;
    table.tb_name = new.to_string();
    for idx in table.secondary_indexes.iter_mut() {
        idx.table_name = new.to_string();
        if idx.origin == IndexOrigin::Auto
            && idx.name == SecondaryIndex::auto_name(old, &idx.column_name)
        {
            idx.name = SecondaryIndex::auto_name(new, &idx.column_name);
        }
    }
    db.tables.insert(new.to_string(), table);
    Ok(())
}

/// `USING <method>` choices recognized by `execute_create_index`. A
/// missing USING clause defaults to `Btree` so existing CREATE INDEX
/// statements (Phase 3e) keep working unchanged.
#[derive(Debug, Clone, Copy)]
enum IndexMethod {
    Btree,
    Hnsw,
    /// Phase 8b — full-text inverted index over a TEXT column.
    Fts,
}

/// Builds a Phase 3e B-Tree secondary index and attaches it to the table.
fn create_btree_index(
    db: &mut Database,
    table_name: &str,
    index_name: &str,
    column_name: &str,
    datatype: &DataType,
    unique: bool,
    existing: &[(i64, Value)],
) -> Result<String> {
    let mut idx = SecondaryIndex::new(
        index_name.to_string(),
        table_name.to_string(),
        column_name.to_string(),
        datatype,
        unique,
        IndexOrigin::Explicit,
    )?;

    // Populate from existing rows. UNIQUE violations here mean the
    // existing data already breaks the new index's constraint — a
    // common source of user confusion, so be explicit.
    for (rowid, v) in existing {
        if unique && idx.would_violate_unique(v) {
            return Err(SQLRiteError::General(format!(
                "cannot create UNIQUE index '{index_name}': column '{column_name}' \
                 already contains the duplicate value {}",
                v.to_display_string()
            )));
        }
        idx.insert(v, *rowid)?;
    }

    let table_mut = db.get_table_mut(table_name.to_string())?;
    table_mut.secondary_indexes.push(idx);
    Ok(index_name.to_string())
}

/// Builds a Phase 7d.2 HNSW index and attaches it to the table.
fn create_hnsw_index(
    db: &mut Database,
    table_name: &str,
    index_name: &str,
    column_name: &str,
    datatype: &DataType,
    unique: bool,
    existing: &[(i64, Value)],
) -> Result<String> {
    // HNSW only makes sense on VECTOR columns. Reject anything else
    // with a clear message — this is the most likely user error.
    let dim = match datatype {
        DataType::Vector(d) => *d,
        other => {
            return Err(SQLRiteError::General(format!(
                "USING hnsw requires a VECTOR column; '{column_name}' is {other}"
            )));
        }
    };

    if unique {
        return Err(SQLRiteError::General(
            "UNIQUE has no meaning for HNSW indexes".to_string(),
        ));
    }

    // Build the in-memory graph. Distance metric is L2 by default
    // (Phase 7d.2 doesn't yet expose a knob for picking cosine/dot —
    // see `docs/phase-7-plan.md` for the deferral).
    //
    // Seed: hash the index name so different indexes get different
    // graph topologies, but the same index always gets the same one
    // — useful when debugging recall / index size.
    let seed = hash_str_to_seed(index_name);
    let mut idx = HnswIndex::new(DistanceMetric::L2, seed);

    // Snapshot the (rowid, vector) pairs into a side map so the
    // get_vec closure below can serve them by id without re-borrowing
    // the table (we're already holding `existing` — flatten it).
    let mut vec_map: std::collections::HashMap<i64, Vec<f32>> =
        std::collections::HashMap::with_capacity(existing.len());
    for (rowid, v) in existing {
        match v {
            Value::Vector(vec) => {
                if vec.len() != dim {
                    return Err(SQLRiteError::Internal(format!(
                        "row {rowid} stores a {}-dim vector in column '{column_name}' \
                         declared as VECTOR({dim}) — schema invariant violated",
                        vec.len()
                    )));
                }
                vec_map.insert(*rowid, vec.clone());
            }
            // Non-vector values (theoretical NULL, type coercion bug)
            // get skipped — they wouldn't have a sensible graph
            // position anyway.
            _ => continue,
        }
    }

    for (rowid, _) in existing {
        if let Some(v) = vec_map.get(rowid) {
            let v_clone = v.clone();
            idx.insert(*rowid, &v_clone, |id| {
                vec_map.get(&id).cloned().unwrap_or_default()
            });
        }
    }

    let table_mut = db.get_table_mut(table_name.to_string())?;
    table_mut.hnsw_indexes.push(HnswIndexEntry {
        name: index_name.to_string(),
        column_name: column_name.to_string(),
        index: idx,
        // Freshly built — no DELETE/UPDATE has invalidated it yet.
        needs_rebuild: false,
    });
    Ok(index_name.to_string())
}

/// Builds a Phase 8b FTS inverted index and attaches it to the table.
/// Mirrors [`create_hnsw_index`] in shape: validate column type,
/// tokenize each existing row's text into the in-memory posting list,
/// push an `FtsIndexEntry`.
fn create_fts_index(
    db: &mut Database,
    table_name: &str,
    index_name: &str,
    column_name: &str,
    datatype: &DataType,
    unique: bool,
    existing: &[(i64, Value)],
) -> Result<String> {
    // FTS is a TEXT-only feature for the MVP. JSON columns share the
    // Row::Text storage but their content is structured — full-text
    // indexing JSON keys + values would need a different design (and
    // is out of scope per the Phase 8 plan's "Out of scope" section).
    match datatype {
        DataType::Text => {}
        other => {
            return Err(SQLRiteError::General(format!(
                "USING fts requires a TEXT column; '{column_name}' is {other}"
            )));
        }
    }

    if unique {
        return Err(SQLRiteError::General(
            "UNIQUE has no meaning for FTS indexes".to_string(),
        ));
    }

    let mut idx = PostingList::new();
    for (rowid, v) in existing {
        if let Value::Text(text) = v {
            idx.insert(*rowid, text);
        }
        // Non-text values (Null, type coercion bugs) get skipped — same
        // posture as create_hnsw_index for non-vector values.
    }

    let table_mut = db.get_table_mut(table_name.to_string())?;
    table_mut.fts_indexes.push(FtsIndexEntry {
        name: index_name.to_string(),
        column_name: column_name.to_string(),
        index: idx,
        needs_rebuild: false,
    });
    Ok(index_name.to_string())
}

/// Stable, deterministic hash of a string into a u64 RNG seed. FNV-1a;
/// avoids pulling in `std::hash::DefaultHasher` (which is randomized
/// per process).
fn hash_str_to_seed(s: &str) -> u64 {
    let mut h: u64 = 0xCBF29CE484222325;
    for b in s.as_bytes() {
        h ^= *b as u64;
        h = h.wrapping_mul(0x100000001B3);
    }
    h
}

/// Cheap clone helper — `DataType` intentionally doesn't derive `Clone`
/// because the enum has no ergonomic reason to be cloneable elsewhere.
fn clone_datatype(dt: &DataType) -> DataType {
    match dt {
        DataType::Integer => DataType::Integer,
        DataType::Text => DataType::Text,
        DataType::Real => DataType::Real,
        DataType::Bool => DataType::Bool,
        DataType::Vector(dim) => DataType::Vector(*dim),
        DataType::Json => DataType::Json,
        DataType::None => DataType::None,
        DataType::Invalid => DataType::Invalid,
    }
}

fn extract_single_table_name(tables: &[TableWithJoins]) -> Result<String> {
    if tables.len() != 1 {
        return Err(SQLRiteError::NotImplemented(
            "multi-table DELETE is not supported yet".to_string(),
        ));
    }
    extract_table_name(&tables[0])
}

fn extract_table_name(twj: &TableWithJoins) -> Result<String> {
    if !twj.joins.is_empty() {
        return Err(SQLRiteError::NotImplemented(
            "JOIN is not supported yet".to_string(),
        ));
    }
    match &twj.relation {
        TableFactor::Table { name, .. } => Ok(name.to_string()),
        _ => Err(SQLRiteError::NotImplemented(
            "only plain table references are supported".to_string(),
        )),
    }
}

/// Tells the executor how to produce its candidate rowid list.
enum RowidSource {
    /// The WHERE was simple enough to probe a secondary index directly.
    /// The `Vec` already contains exactly the rows the index matched;
    /// no further WHERE evaluation is needed (the probe is precise).
    IndexProbe(Vec<i64>),
    /// No applicable index; caller falls back to walking `table.rowids()`
    /// and evaluating the WHERE on each row.
    FullScan,
}

/// Try to satisfy `WHERE` with an index probe. Currently supports the
/// simplest shape: a single `col = literal` (or `literal = col`) where
/// `col` is on a secondary index. AND/OR/range predicates fall back to
/// full scan — those can be layered on later without changing the caller.
fn select_rowids(table: &Table, selection: Option<&Expr>) -> Result<RowidSource> {
    let Some(expr) = selection else {
        return Ok(RowidSource::FullScan);
    };
    let Some((col, literal)) = try_extract_equality(expr) else {
        return Ok(RowidSource::FullScan);
    };
    let Some(idx) = table.index_for_column(&col) else {
        return Ok(RowidSource::FullScan);
    };

    // Convert the literal into a runtime Value. If the literal type doesn't
    // match the column's index we still need correct semantics — evaluate
    // the WHERE against every row. Fall back to full scan.
    let literal_value = match convert_literal(&literal) {
        Ok(v) => v,
        Err(_) => return Ok(RowidSource::FullScan),
    };

    // Index lookup returns the full list of rowids matching this equality
    // predicate. For unique indexes that's at most one; for non-unique it
    // can be many.
    let mut rowids = idx.lookup(&literal_value);
    rowids.sort_unstable();
    Ok(RowidSource::IndexProbe(rowids))
}

/// Recognizes `expr` as a simple equality on a column reference against a
/// literal. Returns `(column_name, literal_value)` if the shape matches;
/// `None` otherwise. Accepts both `col = literal` and `literal = col`.
fn try_extract_equality(expr: &Expr) -> Option<(String, sqlparser::ast::Value)> {
    // Peel off Nested parens so `WHERE (x = 1)` is recognized too.
    let peeled = match expr {
        Expr::Nested(inner) => inner.as_ref(),
        other => other,
    };
    let Expr::BinaryOp { left, op, right } = peeled else {
        return None;
    };
    if !matches!(op, BinaryOperator::Eq) {
        return None;
    }
    let col_from = |e: &Expr| -> Option<String> {
        match e {
            Expr::Identifier(ident) => Some(ident.value.clone()),
            Expr::CompoundIdentifier(parts) => parts.last().map(|p| p.value.clone()),
            _ => None,
        }
    };
    let literal_from = |e: &Expr| -> Option<sqlparser::ast::Value> {
        if let Expr::Value(v) = e {
            Some(v.value.clone())
        } else {
            None
        }
    };
    if let (Some(c), Some(l)) = (col_from(left), literal_from(right)) {
        return Some((c, l));
    }
    if let (Some(l), Some(c)) = (literal_from(left), col_from(right)) {
        return Some((c, l));
    }
    None
}

/// Recognizes the HNSW-probable query pattern and probes the graph
/// if a matching index exists.
///
/// Looks for ORDER BY `vec_distance_l2(<col>, <bracket-array literal>)`
/// where the table has an HNSW index attached to `<col>`. On a match,
/// returns the top-k rowids straight from the graph (O(log N)). On
/// any miss — different function name, no matching index, query
/// dimension wrong, etc. — returns `None` and the caller falls through
/// to the bounded-heap brute-force path (7c) or the full sort (7b),
/// preserving correct results regardless of whether the HNSW pathway
/// kicked in.
///
/// Phase 7d.2 caveats:
/// - Only `vec_distance_l2` is recognized. Cosine and dot fall through
///   to brute-force because we don't yet expose a per-index distance
///   knob (deferred to Phase 7d.x — see `docs/phase-7-plan.md`).
/// - Only ASCENDING order makes sense for "k nearest" — DESC ORDER BY
///   `vec_distance_l2(...) LIMIT k` would mean "k farthest", which
///   isn't what the index is built for. We don't bother to detect
///   `ascending == false` here; the optimizer just skips and the
///   fallback path handles it correctly (slower).
fn try_hnsw_probe(table: &Table, order_expr: &Expr, k: usize) -> Option<Vec<i64>> {
    if k == 0 {
        return None;
    }

    // Pattern-match: order expr must be a function call vec_distance_l2(a, b).
    let func = match order_expr {
        Expr::Function(f) => f,
        _ => return None,
    };
    let fname = match func.name.0.as_slice() {
        [ObjectNamePart::Identifier(ident)] => ident.value.to_lowercase(),
        _ => return None,
    };
    if fname != "vec_distance_l2" {
        return None;
    }

    // Extract the two args as raw Exprs.
    let arg_list = match &func.args {
        FunctionArguments::List(l) => &l.args,
        _ => return None,
    };
    if arg_list.len() != 2 {
        return None;
    }
    let exprs: Vec<&Expr> = arg_list
        .iter()
        .filter_map(|a| match a {
            FunctionArg::Unnamed(FunctionArgExpr::Expr(e)) => Some(e),
            _ => None,
        })
        .collect();
    if exprs.len() != 2 {
        return None;
    }

    // One arg must be a column reference (the indexed col); the other
    // must be a bracket-array literal (the query vector). Try both
    // orderings — pgvector's idiom puts the column on the left, but
    // SQL is commutative for distance.
    let (col_name, query_vec) = match identify_indexed_arg_and_literal(exprs[0], exprs[1]) {
        Some(v) => v,
        None => match identify_indexed_arg_and_literal(exprs[1], exprs[0]) {
            Some(v) => v,
            None => return None,
        },
    };

    // Find the HNSW index on this column.
    let entry = table
        .hnsw_indexes
        .iter()
        .find(|e| e.column_name == col_name)?;

    // Dimension sanity check — the query vector must match the
    // indexed column's declared dimension. If it doesn't, the brute-
    // force fallback would also error at the vec_distance_l2 dim-check;
    // returning None here lets that path produce the user-visible
    // error message.
    let declared_dim = match table.columns.iter().find(|c| c.column_name == col_name) {
        Some(c) => match &c.datatype {
            DataType::Vector(d) => *d,
            _ => return None,
        },
        None => return None,
    };
    if query_vec.len() != declared_dim {
        return None;
    }

    // Probe the graph. Vectors are looked up from the table's row
    // storage — a closure rather than a `&Table` so the algorithm
    // module stays decoupled from the SQL types.
    let column_for_closure = col_name.clone();
    let table_ref = table;
    let result = entry.index.search(&query_vec, k, |id| {
        match table_ref.get_value(&column_for_closure, id) {
            Some(Value::Vector(v)) => v,
            _ => Vec::new(),
        }
    });
    Some(result)
}

/// Phase 8b — FTS optimizer hook.
///
/// Recognizes `ORDER BY bm25_score(<col>, '<query>') DESC LIMIT <k>`
/// and serves it from the FTS index instead of full-scanning. Returns
/// `Some(rowids)` already sorted by descending BM25 (with rowid
/// ascending as tie-break), or `None` to fall through to scalar eval.
///
/// **Known limitation (mirrors `try_hnsw_probe`).** This shortcut
/// ignores any `WHERE` clause. The canonical FTS query has a
/// `WHERE fts_match(<col>, '<q>')` predicate, which is implicitly
/// satisfied by the probe results — so dropping it is harmless.
/// Anything *else* in the WHERE (`AND status = 'published'`) gets
/// silently skipped on the optimizer path. Per Phase 8 plan Q6 we
/// match HNSW's posture here; a correctness-preserving multi-index
/// composer is deferred.
fn try_fts_probe(table: &Table, order_expr: &Expr, ascending: bool, k: usize) -> Option<Vec<i64>> {
    if k == 0 || ascending {
        // BM25 is "higher = better"; ASC ranking is almost certainly a
        // user mistake. Fall through so the caller gets either an
        // explicit error from scalar eval or the slow correct path.
        return None;
    }

    let func = match order_expr {
        Expr::Function(f) => f,
        _ => return None,
    };
    let fname = match func.name.0.as_slice() {
        [ObjectNamePart::Identifier(ident)] => ident.value.to_lowercase(),
        _ => return None,
    };
    if fname != "bm25_score" {
        return None;
    }

    let arg_list = match &func.args {
        FunctionArguments::List(l) => &l.args,
        _ => return None,
    };
    if arg_list.len() != 2 {
        return None;
    }
    let exprs: Vec<&Expr> = arg_list
        .iter()
        .filter_map(|a| match a {
            FunctionArg::Unnamed(FunctionArgExpr::Expr(e)) => Some(e),
            _ => None,
        })
        .collect();
    if exprs.len() != 2 {
        return None;
    }

    // Arg 0 must be a bare column identifier.
    let col_name = match exprs[0] {
        Expr::Identifier(ident) if ident.quote_style.is_none() => ident.value.clone(),
        _ => return None,
    };

    // Arg 1 must be a single-quoted string literal. Anything else
    // (column reference, function call) requires per-row evaluation —
    // we'd lose the whole point of the probe.
    let query = match exprs[1] {
        Expr::Value(v) => match &v.value {
            AstValue::SingleQuotedString(s) => s.clone(),
            _ => return None,
        },
        _ => return None,
    };

    let entry = table
        .fts_indexes
        .iter()
        .find(|e| e.column_name == col_name)?;

    let scored = entry.index.query(&query, &Bm25Params::default());
    let mut out: Vec<i64> = scored.into_iter().map(|(id, _)| id).collect();
    if out.len() > k {
        out.truncate(k);
    }
    Some(out)
}

/// Helper for `try_hnsw_probe`: given two function args, identify which
/// one is a bare column identifier (the indexed column) and which is a
/// bracket-array literal (the query vector). Returns
/// `Some((column_name, query_vec))` on a match, `None` otherwise.
fn identify_indexed_arg_and_literal(a: &Expr, b: &Expr) -> Option<(String, Vec<f32>)> {
    let col_name = match a {
        Expr::Identifier(ident) if ident.quote_style.is_none() => ident.value.clone(),
        _ => return None,
    };
    let lit_str = match b {
        Expr::Identifier(ident) if ident.quote_style == Some('[') => {
            format!("[{}]", ident.value)
        }
        _ => return None,
    };
    let v = parse_vector_literal(&lit_str).ok()?;
    Some((col_name, v))
}

/// One entry in the bounded-heap top-k path. Holds a pre-evaluated
/// sort key + the rowid it came from. The `asc` flag inverts `Ord`
/// so a single `BinaryHeap<HeapEntry>` works for both ASC and DESC
/// without wrapping in `std::cmp::Reverse` at the call site:
///
///   - ASC LIMIT k = "k smallest": natural Ord. Max-heap top is the
///     largest currently kept; new items smaller than top displace.
///   - DESC LIMIT k = "k largest": Ord reversed. Max-heap top is now
///     the smallest currently kept (under reversed Ord, smallest
///     looks largest); new items larger than top displace.
///
/// In both cases the displacement test reduces to "new entry < heap top".
struct HeapEntry {
    key: Value,
    rowid: i64,
    asc: bool,
}

impl PartialEq for HeapEntry {
    fn eq(&self, other: &Self) -> bool {
        self.cmp(other) == Ordering::Equal
    }
}

impl Eq for HeapEntry {}

impl PartialOrd for HeapEntry {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for HeapEntry {
    fn cmp(&self, other: &Self) -> Ordering {
        let raw = compare_values(Some(&self.key), Some(&other.key));
        if self.asc { raw } else { raw.reverse() }
    }
}

/// Bounded-heap top-k selection. Returns at most `k` rowids in the
/// caller's desired order (ascending key for `order.ascending`,
/// descending otherwise).
///
/// O(N log k) where N = `matching.len()`. Caller must check
/// `k < matching.len()` for this to be a win — for k ≥ N the
/// `sort_rowids` full-sort path is the same asymptotic cost without
/// the heap overhead.
fn select_topk(
    matching: &[i64],
    table: &Table,
    order: &OrderByClause,
    k: usize,
) -> Result<Vec<i64>> {
    use std::collections::BinaryHeap;

    if k == 0 || matching.is_empty() {
        return Ok(Vec::new());
    }

    let mut heap: BinaryHeap<HeapEntry> = BinaryHeap::with_capacity(k + 1);

    for &rowid in matching {
        let key = eval_expr(&order.expr, table, rowid)?;
        let entry = HeapEntry {
            key,
            rowid,
            asc: order.ascending,
        };

        if heap.len() < k {
            heap.push(entry);
        } else {
            // peek() returns the largest under our direction-aware Ord
            // — the worst entry currently kept. Displace it iff the
            // new entry is "better" (i.e. compares Less).
            if entry < *heap.peek().unwrap() {
                heap.pop();
                heap.push(entry);
            }
        }
    }

    // `into_sorted_vec` returns ascending under our direction-aware Ord:
    //   ASC: ascending by raw key (what we want)
    //   DESC: ascending under reversed Ord = descending by raw key (what
    //         we want for an ORDER BY DESC LIMIT k result)
    Ok(heap
        .into_sorted_vec()
        .into_iter()
        .map(|e| e.rowid)
        .collect())
}

fn sort_rowids(rowids: &mut [i64], table: &Table, order: &OrderByClause) -> Result<()> {
    // Phase 7b: ORDER BY now accepts any expression (column ref,
    // arithmetic, function call, …). Pre-compute the sort key for
    // every rowid up front so the comparator is called O(N log N)
    // times against pre-evaluated Values rather than re-evaluating
    // the expression O(N log N) times. Not strictly necessary today,
    // but vital once 7d's HNSW index lands and this same code path
    // could be running tens of millions of distance computations.
    let mut keys: Vec<(i64, Result<Value>)> = rowids
        .iter()
        .map(|r| (*r, eval_expr(&order.expr, table, *r)))
        .collect();

    // Surface the FIRST evaluation error if any. We could be lazy
    // and let sort_by encounter it, but `Ord::cmp` can't return a
    // Result and we'd have to swallow errors silently.
    for (_, k) in &keys {
        if let Err(e) = k {
            return Err(SQLRiteError::General(format!(
                "ORDER BY expression failed: {e}"
            )));
        }
    }

    keys.sort_by(|(_, ka), (_, kb)| {
        // Both unwrap()s are safe — we just verified above that
        // every key Result is Ok.
        let va = ka.as_ref().unwrap();
        let vb = kb.as_ref().unwrap();
        let ord = compare_values(Some(va), Some(vb));
        if order.ascending { ord } else { ord.reverse() }
    });

    // Write the sorted rowids back into the caller's slice.
    for (i, (rowid, _)) in keys.into_iter().enumerate() {
        rowids[i] = rowid;
    }
    Ok(())
}

fn compare_values(a: Option<&Value>, b: Option<&Value>) -> Ordering {
    match (a, b) {
        (None, None) => Ordering::Equal,
        (None, _) => Ordering::Less,
        (_, None) => Ordering::Greater,
        (Some(a), Some(b)) => match (a, b) {
            (Value::Null, Value::Null) => Ordering::Equal,
            (Value::Null, _) => Ordering::Less,
            (_, Value::Null) => Ordering::Greater,
            (Value::Integer(x), Value::Integer(y)) => x.cmp(y),
            (Value::Real(x), Value::Real(y)) => x.partial_cmp(y).unwrap_or(Ordering::Equal),
            (Value::Integer(x), Value::Real(y)) => {
                (*x as f64).partial_cmp(y).unwrap_or(Ordering::Equal)
            }
            (Value::Real(x), Value::Integer(y)) => {
                x.partial_cmp(&(*y as f64)).unwrap_or(Ordering::Equal)
            }
            (Value::Text(x), Value::Text(y)) => x.cmp(y),
            (Value::Bool(x), Value::Bool(y)) => x.cmp(y),
            // Cross-type fallback: stringify and compare; keeps ORDER BY total.
            (x, y) => x.to_display_string().cmp(&y.to_display_string()),
        },
    }
}

/// Returns `true` if the row at `rowid` matches the predicate expression.
pub fn eval_predicate(expr: &Expr, table: &Table, rowid: i64) -> Result<bool> {
    let v = eval_expr(expr, table, rowid)?;
    match v {
        Value::Bool(b) => Ok(b),
        Value::Null => Ok(false), // SQL NULL in a WHERE is treated as false
        Value::Integer(i) => Ok(i != 0),
        other => Err(SQLRiteError::Internal(format!(
            "WHERE clause must evaluate to boolean, got {}",
            other.to_display_string()
        ))),
    }
}

fn eval_expr(expr: &Expr, table: &Table, rowid: i64) -> Result<Value> {
    match expr {
        Expr::Nested(inner) => eval_expr(inner, table, rowid),

        Expr::Identifier(ident) => {
            // Phase 7b — sqlparser parses bracket-array literals like
            // `[0.1, 0.2, 0.3]` as bracket-quoted identifiers (it inherits
            // MSSQL `[name]` syntax). When we see `quote_style == Some('[')`
            // in expression-evaluation position (SELECT projection, WHERE,
            // ORDER BY, function args), parse the bracketed content as a
            // vector literal so the rest of the executor can compare /
            // distance-compute against it. Same trick the INSERT parser
            // uses; the executor needed its own copy because expression
            // eval runs on a different code path.
            if ident.quote_style == Some('[') {
                let raw = format!("[{}]", ident.value);
                let v = parse_vector_literal(&raw)?;
                return Ok(Value::Vector(v));
            }
            Ok(table.get_value(&ident.value, rowid).unwrap_or(Value::Null))
        }

        Expr::CompoundIdentifier(parts) => {
            // Accept `table.col` — we only have one table in scope, so ignore the qualifier.
            let col = parts
                .last()
                .map(|i| i.value.as_str())
                .ok_or_else(|| SQLRiteError::Internal("empty compound identifier".to_string()))?;
            Ok(table.get_value(col, rowid).unwrap_or(Value::Null))
        }

        Expr::Value(v) => convert_literal(&v.value),

        Expr::UnaryOp { op, expr } => {
            let inner = eval_expr(expr, table, rowid)?;
            match op {
                UnaryOperator::Not => match inner {
                    Value::Bool(b) => Ok(Value::Bool(!b)),
                    Value::Null => Ok(Value::Null),
                    other => Err(SQLRiteError::Internal(format!(
                        "NOT applied to non-boolean value: {}",
                        other.to_display_string()
                    ))),
                },
                UnaryOperator::Minus => match inner {
                    Value::Integer(i) => Ok(Value::Integer(-i)),
                    Value::Real(f) => Ok(Value::Real(-f)),
                    Value::Null => Ok(Value::Null),
                    other => Err(SQLRiteError::Internal(format!(
                        "unary minus on non-numeric value: {}",
                        other.to_display_string()
                    ))),
                },
                UnaryOperator::Plus => Ok(inner),
                other => Err(SQLRiteError::NotImplemented(format!(
                    "unary operator {other:?} is not supported"
                ))),
            }
        }

        Expr::BinaryOp { left, op, right } => match op {
            BinaryOperator::And => {
                let l = eval_expr(left, table, rowid)?;
                let r = eval_expr(right, table, rowid)?;
                Ok(Value::Bool(as_bool(&l)? && as_bool(&r)?))
            }
            BinaryOperator::Or => {
                let l = eval_expr(left, table, rowid)?;
                let r = eval_expr(right, table, rowid)?;
                Ok(Value::Bool(as_bool(&l)? || as_bool(&r)?))
            }
            cmp @ (BinaryOperator::Eq
            | BinaryOperator::NotEq
            | BinaryOperator::Lt
            | BinaryOperator::LtEq
            | BinaryOperator::Gt
            | BinaryOperator::GtEq) => {
                let l = eval_expr(left, table, rowid)?;
                let r = eval_expr(right, table, rowid)?;
                // Any comparison involving NULL is unknown → false in a WHERE.
                if matches!(l, Value::Null) || matches!(r, Value::Null) {
                    return Ok(Value::Bool(false));
                }
                let ord = compare_values(Some(&l), Some(&r));
                let result = match cmp {
                    BinaryOperator::Eq => ord == Ordering::Equal,
                    BinaryOperator::NotEq => ord != Ordering::Equal,
                    BinaryOperator::Lt => ord == Ordering::Less,
                    BinaryOperator::LtEq => ord != Ordering::Greater,
                    BinaryOperator::Gt => ord == Ordering::Greater,
                    BinaryOperator::GtEq => ord != Ordering::Less,
                    _ => unreachable!(),
                };
                Ok(Value::Bool(result))
            }
            arith @ (BinaryOperator::Plus
            | BinaryOperator::Minus
            | BinaryOperator::Multiply
            | BinaryOperator::Divide
            | BinaryOperator::Modulo) => {
                let l = eval_expr(left, table, rowid)?;
                let r = eval_expr(right, table, rowid)?;
                eval_arith(arith, &l, &r)
            }
            BinaryOperator::StringConcat => {
                let l = eval_expr(left, table, rowid)?;
                let r = eval_expr(right, table, rowid)?;
                if matches!(l, Value::Null) || matches!(r, Value::Null) {
                    return Ok(Value::Null);
                }
                Ok(Value::Text(format!(
                    "{}{}",
                    l.to_display_string(),
                    r.to_display_string()
                )))
            }
            other => Err(SQLRiteError::NotImplemented(format!(
                "binary operator {other:?} is not supported yet"
            ))),
        },

        // Phase 7b — function-call dispatch. Currently only the three
        // vector-distance functions; this match arm becomes the single
        // place to register more SQL functions later (e.g. abs(),
        // length(), …) without re-touching the rest of the executor.
        //
        // Operator forms (`<->` `<=>` `<#>`) are NOT plumbed here: two
        // of three don't parse natively in sqlparser (we'd need a
        // string-preprocessing pass or a sqlparser fork). Deferred to
        // a follow-up sub-phase; see docs/phase-7-plan.md's "Scope
        // corrections" note.
        Expr::Function(func) => eval_function(func, table, rowid),

        other => Err(SQLRiteError::NotImplemented(format!(
            "unsupported expression in WHERE/projection: {other:?}"
        ))),
    }
}

/// Dispatches an `Expr::Function` to its built-in implementation.
/// Currently only the three vec_distance_* functions; other functions
/// surface as `NotImplemented` errors with the function name in the
/// message so users see what they tried.
fn eval_function(func: &sqlparser::ast::Function, table: &Table, rowid: i64) -> Result<Value> {
    // Function name lives in `name.0[0]` for unqualified calls. Anything
    // qualified (e.g. `pkg.fn(...)`) falls through to NotImplemented.
    let name = match func.name.0.as_slice() {
        [ObjectNamePart::Identifier(ident)] => ident.value.to_lowercase(),
        _ => {
            return Err(SQLRiteError::NotImplemented(format!(
                "qualified function names not supported: {:?}",
                func.name
            )));
        }
    };

    match name.as_str() {
        "vec_distance_l2" | "vec_distance_cosine" | "vec_distance_dot" => {
            let (a, b) = extract_two_vector_args(&name, &func.args, table, rowid)?;
            let dist = match name.as_str() {
                "vec_distance_l2" => vec_distance_l2(&a, &b),
                "vec_distance_cosine" => vec_distance_cosine(&a, &b)?,
                "vec_distance_dot" => vec_distance_dot(&a, &b),
                _ => unreachable!(),
            };
            // Widen f32 → f64 for the runtime Value. Vectors are stored
            // as f32 (consistent with industry convention for embeddings),
            // but the executor's numeric type is f64 so distances slot
            // into Value::Real cleanly and can be compared / ordered with
            // other reals via the existing arithmetic + comparison paths.
            Ok(Value::Real(dist as f64))
        }
        // Phase 7e — JSON functions. All four parse the JSON text on
        // demand (we don't cache parsed values), then resolve a path
        // (default `$` = root). The path resolver handles `.key` for
        // object access and `[N]` for array index. SQLite-style.
        "json_extract" => json_fn_extract(&name, &func.args, table, rowid),
        "json_type" => json_fn_type(&name, &func.args, table, rowid),
        "json_array_length" => json_fn_array_length(&name, &func.args, table, rowid),
        "json_object_keys" => json_fn_object_keys(&name, &func.args, table, rowid),
        // Phase 8b — FTS scalars. Both consult an FTS index attached to
        // the named column; both error if no index exists (the index is
        // a hard prerequisite, mirroring SQLite FTS5's MATCH).
        "fts_match" => {
            let (entry, query) = resolve_fts_args(&name, &func.args, table, rowid)?;
            Ok(Value::Bool(entry.index.matches(rowid, &query)))
        }
        "bm25_score" => {
            let (entry, query) = resolve_fts_args(&name, &func.args, table, rowid)?;
            let s = entry.index.score(rowid, &query, &Bm25Params::default());
            Ok(Value::Real(s))
        }
        other => Err(SQLRiteError::NotImplemented(format!(
            "unknown function: {other}(...)"
        ))),
    }
}

/// Helper for `fts_match` / `bm25_score`: pull the column reference out
/// of arg 0 (a bare identifier — we need the *name*, not the per-row
/// value), evaluate arg 1 as a Text query string, and look up the FTS
/// index attached to that column. Errors if any step fails.
fn resolve_fts_args<'t>(
    fn_name: &str,
    args: &FunctionArguments,
    table: &'t Table,
    rowid: i64,
) -> Result<(&'t FtsIndexEntry, String)> {
    let arg_list = match args {
        FunctionArguments::List(l) => &l.args,
        _ => {
            return Err(SQLRiteError::General(format!(
                "{fn_name}() expects exactly two arguments: (column, query_text)"
            )));
        }
    };
    if arg_list.len() != 2 {
        return Err(SQLRiteError::General(format!(
            "{fn_name}() expects exactly 2 arguments, got {}",
            arg_list.len()
        )));
    }

    // Arg 0: bare column identifier. Must resolve syntactically to a
    // column name (we can't accept arbitrary expressions because we
    // need the column to look up the index, not the column's value).
    let col_expr = match &arg_list[0] {
        FunctionArg::Unnamed(FunctionArgExpr::Expr(e)) => e,
        other => {
            return Err(SQLRiteError::NotImplemented(format!(
                "{fn_name}() argument 0 must be a column name, got {other:?}"
            )));
        }
    };
    let col_name = match col_expr {
        Expr::Identifier(ident) => ident.value.clone(),
        Expr::CompoundIdentifier(parts) => parts
            .last()
            .map(|p| p.value.clone())
            .ok_or_else(|| SQLRiteError::Internal("empty compound identifier".to_string()))?,
        other => {
            return Err(SQLRiteError::General(format!(
                "{fn_name}() argument 0 must be a column reference, got {other:?}"
            )));
        }
    };

    // Arg 1: query string. Evaluated through the normal expression
    // pipeline so callers can pass a literal `'rust db'` or an
    // expression that yields TEXT.
    let q_expr = match &arg_list[1] {
        FunctionArg::Unnamed(FunctionArgExpr::Expr(e)) => e,
        other => {
            return Err(SQLRiteError::NotImplemented(format!(
                "{fn_name}() argument 1 must be a text expression, got {other:?}"
            )));
        }
    };
    let query = match eval_expr(q_expr, table, rowid)? {
        Value::Text(s) => s,
        other => {
            return Err(SQLRiteError::General(format!(
                "{fn_name}() argument 1 must be TEXT, got {}",
                other.to_display_string()
            )));
        }
    };

    let entry = table
        .fts_indexes
        .iter()
        .find(|e| e.column_name == col_name)
        .ok_or_else(|| {
            SQLRiteError::General(format!(
                "{fn_name}({col_name}, ...): no FTS index on column '{col_name}' \
                 (run CREATE INDEX <name> ON <table> USING fts({col_name}) first)"
            ))
        })?;
    Ok((entry, query))
}

// -----------------------------------------------------------------
// Phase 7e — JSON path-extraction functions
// -----------------------------------------------------------------

/// Extracts the JSON-typed text + optional path string out of a
/// function call's args. Used by all four json_* functions.
///
/// Arity rules (matching SQLite JSON1):
///   - 1 arg  → JSON value, path defaults to `$` (root)
///   - 2 args → (JSON value, path text)
///
/// Returns `(json_text, path)` so caller can serde_json::from_str
/// + walk_json_path on it.
fn extract_json_and_path(
    fn_name: &str,
    args: &FunctionArguments,
    table: &Table,
    rowid: i64,
) -> Result<(String, String)> {
    let arg_list = match args {
        FunctionArguments::List(l) => &l.args,
        _ => {
            return Err(SQLRiteError::General(format!(
                "{fn_name}() expects 1 or 2 arguments"
            )));
        }
    };
    if !(arg_list.len() == 1 || arg_list.len() == 2) {
        return Err(SQLRiteError::General(format!(
            "{fn_name}() expects 1 or 2 arguments, got {}",
            arg_list.len()
        )));
    }
    // Evaluate first arg → must produce text.
    let first_expr = match &arg_list[0] {
        FunctionArg::Unnamed(FunctionArgExpr::Expr(e)) => e,
        other => {
            return Err(SQLRiteError::NotImplemented(format!(
                "{fn_name}() argument 0 has unsupported shape: {other:?}"
            )));
        }
    };
    let json_text = match eval_expr(first_expr, table, rowid)? {
        Value::Text(s) => s,
        Value::Null => {
            return Err(SQLRiteError::General(format!(
                "{fn_name}() called on NULL — JSON column has no value for this row"
            )));
        }
        other => {
            return Err(SQLRiteError::General(format!(
                "{fn_name}() argument 0 is not JSON-typed: got {}",
                other.to_display_string()
            )));
        }
    };

    // Path defaults to root `$` when omitted.
    let path = if arg_list.len() == 2 {
        let path_expr = match &arg_list[1] {
            FunctionArg::Unnamed(FunctionArgExpr::Expr(e)) => e,
            other => {
                return Err(SQLRiteError::NotImplemented(format!(
                    "{fn_name}() argument 1 has unsupported shape: {other:?}"
                )));
            }
        };
        match eval_expr(path_expr, table, rowid)? {
            Value::Text(s) => s,
            other => {
                return Err(SQLRiteError::General(format!(
                    "{fn_name}() path argument must be a string literal, got {}",
                    other.to_display_string()
                )));
            }
        }
    } else {
        "$".to_string()
    };

    Ok((json_text, path))
}

/// Walks a `serde_json::Value` along a JSONPath subset:
///   - `$` is the root
///   - `.key` for object access (key may not contain `.` or `[`)
///   - `[N]` for array index (N a non-negative integer)
///   - chains arbitrarily: `$.foo.bar[0].baz`
///
/// Returns `Ok(None)` for "path didn't match anything" (NULL in SQL),
/// `Err` for malformed paths. Matches SQLite JSON1's semantic
/// distinction: missing-key = NULL, malformed-path = error.
fn walk_json_path<'a>(
    value: &'a serde_json::Value,
    path: &str,
) -> Result<Option<&'a serde_json::Value>> {
    let mut chars = path.chars().peekable();
    if chars.next() != Some('$') {
        return Err(SQLRiteError::General(format!(
            "JSON path must start with '$', got `{path}`"
        )));
    }
    let mut current = value;
    while let Some(&c) = chars.peek() {
        match c {
            '.' => {
                chars.next();
                let mut key = String::new();
                while let Some(&c) = chars.peek() {
                    if c == '.' || c == '[' {
                        break;
                    }
                    key.push(c);
                    chars.next();
                }
                if key.is_empty() {
                    return Err(SQLRiteError::General(format!(
                        "JSON path has empty key after '.' in `{path}`"
                    )));
                }
                match current.get(&key) {
                    Some(v) => current = v,
                    None => return Ok(None),
                }
            }
            '[' => {
                chars.next();
                let mut idx_str = String::new();
                while let Some(&c) = chars.peek() {
                    if c == ']' {
                        break;
                    }
                    idx_str.push(c);
                    chars.next();
                }
                if chars.next() != Some(']') {
                    return Err(SQLRiteError::General(format!(
                        "JSON path has unclosed `[` in `{path}`"
                    )));
                }
                let idx: usize = idx_str.trim().parse().map_err(|_| {
                    SQLRiteError::General(format!(
                        "JSON path has non-integer index `[{idx_str}]` in `{path}`"
                    ))
                })?;
                match current.get(idx) {
                    Some(v) => current = v,
                    None => return Ok(None),
                }
            }
            other => {
                return Err(SQLRiteError::General(format!(
                    "JSON path has unexpected character `{other}` in `{path}` \
                     (expected `.`, `[`, or end-of-path)"
                )));
            }
        }
    }
    Ok(Some(current))
}

/// Converts a serde_json scalar to a SQLRite Value. For composite
/// types (object, array) returns the JSON-encoded text — callers
/// pattern-match on shape from the calling json_* function.
fn json_value_to_sql(v: &serde_json::Value) -> Value {
    match v {
        serde_json::Value::Null => Value::Null,
        serde_json::Value::Bool(b) => Value::Bool(*b),
        serde_json::Value::Number(n) => {
            // Match SQLite: integer if it fits an i64, else f64.
            if let Some(i) = n.as_i64() {
                Value::Integer(i)
            } else if let Some(f) = n.as_f64() {
                Value::Real(f)
            } else {
                Value::Null
            }
        }
        serde_json::Value::String(s) => Value::Text(s.clone()),
        // Objects + arrays come out as JSON-encoded text. Same as
        // SQLite's json_extract: composite results round-trip through
        // text rather than being modeled as a richer Value type.
        composite => Value::Text(composite.to_string()),
    }
}

fn json_fn_extract(
    name: &str,
    args: &FunctionArguments,
    table: &Table,
    rowid: i64,
) -> Result<Value> {
    let (json_text, path) = extract_json_and_path(name, args, table, rowid)?;
    let parsed: serde_json::Value = serde_json::from_str(&json_text).map_err(|e| {
        SQLRiteError::General(format!("{name}() got invalid JSON `{json_text}`: {e}"))
    })?;
    match walk_json_path(&parsed, &path)? {
        Some(v) => Ok(json_value_to_sql(v)),
        None => Ok(Value::Null),
    }
}

fn json_fn_type(name: &str, args: &FunctionArguments, table: &Table, rowid: i64) -> Result<Value> {
    let (json_text, path) = extract_json_and_path(name, args, table, rowid)?;
    let parsed: serde_json::Value = serde_json::from_str(&json_text).map_err(|e| {
        SQLRiteError::General(format!("{name}() got invalid JSON `{json_text}`: {e}"))
    })?;
    let resolved = match walk_json_path(&parsed, &path)? {
        Some(v) => v,
        None => return Ok(Value::Null),
    };
    let ty = match resolved {
        serde_json::Value::Null => "null",
        serde_json::Value::Bool(true) => "true",
        serde_json::Value::Bool(false) => "false",
        serde_json::Value::Number(n) => {
            if n.is_i64() || n.is_u64() {
                "integer"
            } else {
                "real"
            }
        }
        serde_json::Value::String(_) => "text",
        serde_json::Value::Array(_) => "array",
        serde_json::Value::Object(_) => "object",
    };
    Ok(Value::Text(ty.to_string()))
}

fn json_fn_array_length(
    name: &str,
    args: &FunctionArguments,
    table: &Table,
    rowid: i64,
) -> Result<Value> {
    let (json_text, path) = extract_json_and_path(name, args, table, rowid)?;
    let parsed: serde_json::Value = serde_json::from_str(&json_text).map_err(|e| {
        SQLRiteError::General(format!("{name}() got invalid JSON `{json_text}`: {e}"))
    })?;
    let resolved = match walk_json_path(&parsed, &path)? {
        Some(v) => v,
        None => return Ok(Value::Null),
    };
    match resolved.as_array() {
        Some(arr) => Ok(Value::Integer(arr.len() as i64)),
        None => Err(SQLRiteError::General(format!(
            "{name}() resolved to a non-array value at path `{path}`"
        ))),
    }
}

fn json_fn_object_keys(
    name: &str,
    args: &FunctionArguments,
    table: &Table,
    rowid: i64,
) -> Result<Value> {
    let (json_text, path) = extract_json_and_path(name, args, table, rowid)?;
    let parsed: serde_json::Value = serde_json::from_str(&json_text).map_err(|e| {
        SQLRiteError::General(format!("{name}() got invalid JSON `{json_text}`: {e}"))
    })?;
    let resolved = match walk_json_path(&parsed, &path)? {
        Some(v) => v,
        None => return Ok(Value::Null),
    };
    let obj = resolved.as_object().ok_or_else(|| {
        SQLRiteError::General(format!(
            "{name}() resolved to a non-object value at path `{path}`"
        ))
    })?;
    // SQLite's json_object_keys is a table-valued function (one row
    // per key). Without set-returning function support we can't
    // reproduce that shape; instead return the keys as a JSON array
    // text. Caller can iterate via json_array_length + json_extract,
    // or just treat it as a serialized list. Document this divergence
    // in supported-sql.md.
    let keys: Vec<serde_json::Value> = obj
        .keys()
        .map(|k| serde_json::Value::String(k.clone()))
        .collect();
    Ok(Value::Text(serde_json::Value::Array(keys).to_string()))
}

/// Extracts exactly two `Vec<f32>` arguments from a function call,
/// validating arity and that both sides are Vector-typed with matching
/// dimensions. Used by all three vec_distance_* functions.
fn extract_two_vector_args(
    fn_name: &str,
    args: &FunctionArguments,
    table: &Table,
    rowid: i64,
) -> Result<(Vec<f32>, Vec<f32>)> {
    let arg_list = match args {
        FunctionArguments::List(l) => &l.args,
        _ => {
            return Err(SQLRiteError::General(format!(
                "{fn_name}() expects exactly two vector arguments"
            )));
        }
    };
    if arg_list.len() != 2 {
        return Err(SQLRiteError::General(format!(
            "{fn_name}() expects exactly 2 arguments, got {}",
            arg_list.len()
        )));
    }
    let mut out: Vec<Vec<f32>> = Vec::with_capacity(2);
    for (i, arg) in arg_list.iter().enumerate() {
        let expr = match arg {
            FunctionArg::Unnamed(FunctionArgExpr::Expr(e)) => e,
            other => {
                return Err(SQLRiteError::NotImplemented(format!(
                    "{fn_name}() argument {i} has unsupported shape: {other:?}"
                )));
            }
        };
        let val = eval_expr(expr, table, rowid)?;
        match val {
            Value::Vector(v) => out.push(v),
            other => {
                return Err(SQLRiteError::General(format!(
                    "{fn_name}() argument {i} is not a vector: got {}",
                    other.to_display_string()
                )));
            }
        }
    }
    let b = out.pop().unwrap();
    let a = out.pop().unwrap();
    if a.len() != b.len() {
        return Err(SQLRiteError::General(format!(
            "{fn_name}(): vector dimensions don't match (lhs={}, rhs={})",
            a.len(),
            b.len()
        )));
    }
    Ok((a, b))
}

/// Euclidean (L2) distance: √Σ(aᵢ − bᵢ)².
/// Smaller-is-closer; identical vectors return 0.0.
pub(crate) fn vec_distance_l2(a: &[f32], b: &[f32]) -> f32 {
    debug_assert_eq!(a.len(), b.len());
    let mut sum = 0.0f32;
    for i in 0..a.len() {
        let d = a[i] - b[i];
        sum += d * d;
    }
    sum.sqrt()
}

/// Cosine distance: 1 − (a·b) / (‖a‖·‖b‖).
/// Smaller-is-closer; identical (non-zero) vectors return 0.0,
/// orthogonal vectors return 1.0, opposite-direction vectors return 2.0.
///
/// Errors if either vector has zero magnitude — cosine similarity is
/// undefined for the zero vector and silently returning NaN would
/// poison `ORDER BY` ranking. Callers who want the silent-NaN
/// behavior can compute `vec_distance_dot(a, b) / (norm(a) * norm(b))`
/// themselves.
pub(crate) fn vec_distance_cosine(a: &[f32], b: &[f32]) -> Result<f32> {
    debug_assert_eq!(a.len(), b.len());
    let mut dot = 0.0f32;
    let mut norm_a_sq = 0.0f32;
    let mut norm_b_sq = 0.0f32;
    for i in 0..a.len() {
        dot += a[i] * b[i];
        norm_a_sq += a[i] * a[i];
        norm_b_sq += b[i] * b[i];
    }
    let denom = (norm_a_sq * norm_b_sq).sqrt();
    if denom == 0.0 {
        return Err(SQLRiteError::General(
            "vec_distance_cosine() is undefined for zero-magnitude vectors".to_string(),
        ));
    }
    Ok(1.0 - dot / denom)
}

/// Negated dot product: −(a·b).
/// pgvector convention — negated so smaller-is-closer like L2 / cosine.
/// For unit-norm vectors `vec_distance_dot(a, b) == vec_distance_cosine(a, b) - 1`.
pub(crate) fn vec_distance_dot(a: &[f32], b: &[f32]) -> f32 {
    debug_assert_eq!(a.len(), b.len());
    let mut dot = 0.0f32;
    for i in 0..a.len() {
        dot += a[i] * b[i];
    }
    -dot
}

/// Evaluates an integer/real arithmetic op. NULL on either side propagates.
/// Mixed Integer/Real promotes to Real. Divide/Modulo by zero → error.
fn eval_arith(op: &BinaryOperator, l: &Value, r: &Value) -> Result<Value> {
    if matches!(l, Value::Null) || matches!(r, Value::Null) {
        return Ok(Value::Null);
    }
    match (l, r) {
        (Value::Integer(a), Value::Integer(b)) => match op {
            BinaryOperator::Plus => Ok(Value::Integer(a.wrapping_add(*b))),
            BinaryOperator::Minus => Ok(Value::Integer(a.wrapping_sub(*b))),
            BinaryOperator::Multiply => Ok(Value::Integer(a.wrapping_mul(*b))),
            BinaryOperator::Divide => {
                if *b == 0 {
                    Err(SQLRiteError::General("division by zero".to_string()))
                } else {
                    Ok(Value::Integer(a / b))
                }
            }
            BinaryOperator::Modulo => {
                if *b == 0 {
                    Err(SQLRiteError::General("modulo by zero".to_string()))
                } else {
                    Ok(Value::Integer(a % b))
                }
            }
            _ => unreachable!(),
        },
        // Anything involving a Real promotes both sides to f64.
        (a, b) => {
            let af = as_number(a)?;
            let bf = as_number(b)?;
            match op {
                BinaryOperator::Plus => Ok(Value::Real(af + bf)),
                BinaryOperator::Minus => Ok(Value::Real(af - bf)),
                BinaryOperator::Multiply => Ok(Value::Real(af * bf)),
                BinaryOperator::Divide => {
                    if bf == 0.0 {
                        Err(SQLRiteError::General("division by zero".to_string()))
                    } else {
                        Ok(Value::Real(af / bf))
                    }
                }
                BinaryOperator::Modulo => {
                    if bf == 0.0 {
                        Err(SQLRiteError::General("modulo by zero".to_string()))
                    } else {
                        Ok(Value::Real(af % bf))
                    }
                }
                _ => unreachable!(),
            }
        }
    }
}

fn as_number(v: &Value) -> Result<f64> {
    match v {
        Value::Integer(i) => Ok(*i as f64),
        Value::Real(f) => Ok(*f),
        Value::Bool(b) => Ok(if *b { 1.0 } else { 0.0 }),
        other => Err(SQLRiteError::General(format!(
            "arithmetic on non-numeric value '{}'",
            other.to_display_string()
        ))),
    }
}

fn as_bool(v: &Value) -> Result<bool> {
    match v {
        Value::Bool(b) => Ok(*b),
        Value::Null => Ok(false),
        Value::Integer(i) => Ok(*i != 0),
        other => Err(SQLRiteError::Internal(format!(
            "expected boolean, got {}",
            other.to_display_string()
        ))),
    }
}

fn convert_literal(v: &sqlparser::ast::Value) -> Result<Value> {
    use sqlparser::ast::Value as AstValue;
    match v {
        AstValue::Number(n, _) => {
            if let Ok(i) = n.parse::<i64>() {
                Ok(Value::Integer(i))
            } else if let Ok(f) = n.parse::<f64>() {
                Ok(Value::Real(f))
            } else {
                Err(SQLRiteError::Internal(format!(
                    "could not parse numeric literal '{n}'"
                )))
            }
        }
        AstValue::SingleQuotedString(s) => Ok(Value::Text(s.clone())),
        AstValue::Boolean(b) => Ok(Value::Bool(*b)),
        AstValue::Null => Ok(Value::Null),
        other => Err(SQLRiteError::NotImplemented(format!(
            "unsupported literal value: {other:?}"
        ))),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    // -----------------------------------------------------------------
    // Phase 7b — Vector distance function math
    // -----------------------------------------------------------------

    /// Float comparison helper — distance results need a small epsilon
    /// because we accumulate sums across many f32 multiplies.
    fn approx_eq(a: f32, b: f32, eps: f32) -> bool {
        (a - b).abs() < eps
    }

    #[test]
    fn vec_distance_l2_identical_is_zero() {
        let v = vec![0.1, 0.2, 0.3];
        assert_eq!(vec_distance_l2(&v, &v), 0.0);
    }

    #[test]
    fn vec_distance_l2_unit_basis_is_sqrt2() {
        // [1, 0] vs [0, 1]: distance = √((1-0)² + (0-1)²) = √2 ≈ 1.414
        let a = vec![1.0, 0.0];
        let b = vec![0.0, 1.0];
        assert!(approx_eq(vec_distance_l2(&a, &b), 2.0_f32.sqrt(), 1e-6));
    }

    #[test]
    fn vec_distance_l2_known_value() {
        // [0, 0, 0] vs [3, 4, 0]: √(9 + 16 + 0) = 5 (the classic 3-4-5 triangle).
        let a = vec![0.0, 0.0, 0.0];
        let b = vec![3.0, 4.0, 0.0];
        assert!(approx_eq(vec_distance_l2(&a, &b), 5.0, 1e-6));
    }

    #[test]
    fn vec_distance_cosine_identical_is_zero() {
        let v = vec![0.1, 0.2, 0.3];
        let d = vec_distance_cosine(&v, &v).unwrap();
        assert!(approx_eq(d, 0.0, 1e-6), "cos(v,v) = {d}, expected ≈ 0");
    }

    #[test]
    fn vec_distance_cosine_orthogonal_is_one() {
        // Two orthogonal unit vectors should have cosine distance = 1.0
        // (cosine similarity = 0 → distance = 1 - 0 = 1).
        let a = vec![1.0, 0.0];
        let b = vec![0.0, 1.0];
        assert!(approx_eq(vec_distance_cosine(&a, &b).unwrap(), 1.0, 1e-6));
    }

    #[test]
    fn vec_distance_cosine_opposite_is_two() {
        // a and -a have cosine similarity = -1 → distance = 1 - (-1) = 2.
        let a = vec![1.0, 0.0, 0.0];
        let b = vec![-1.0, 0.0, 0.0];
        assert!(approx_eq(vec_distance_cosine(&a, &b).unwrap(), 2.0, 1e-6));
    }

    #[test]
    fn vec_distance_cosine_zero_magnitude_errors() {
        // Cosine is undefined for the zero vector — error rather than NaN.
        let a = vec![0.0, 0.0];
        let b = vec![1.0, 0.0];
        let err = vec_distance_cosine(&a, &b).unwrap_err();
        assert!(format!("{err}").contains("zero-magnitude"));
    }

    #[test]
    fn vec_distance_dot_negates() {
        // a·b = 1*4 + 2*5 + 3*6 = 32. Negated → -32.
        let a = vec![1.0, 2.0, 3.0];
        let b = vec![4.0, 5.0, 6.0];
        assert!(approx_eq(vec_distance_dot(&a, &b), -32.0, 1e-6));
    }

    #[test]
    fn vec_distance_dot_orthogonal_is_zero() {
        // Orthogonal vectors have dot product 0 → negated is also 0.
        let a = vec![1.0, 0.0];
        let b = vec![0.0, 1.0];
        assert_eq!(vec_distance_dot(&a, &b), 0.0);
    }

    #[test]
    fn vec_distance_dot_unit_norm_matches_cosine_minus_one() {
        // For unit-norm vectors: dot(a,b) = cos(a,b)
        // → -dot(a,b) = -cos(a,b) = (1 - cos(a,b)) - 1 = vec_distance_cosine(a,b) - 1.
        // Useful sanity check that the two functions agree on unit vectors.
        let a = vec![0.6f32, 0.8]; // unit norm: √(0.36+0.64) = 1
        let b = vec![0.8f32, 0.6]; // unit norm too
        let dot = vec_distance_dot(&a, &b);
        let cos = vec_distance_cosine(&a, &b).unwrap();
        assert!(approx_eq(dot, cos - 1.0, 1e-5));
    }

    // -----------------------------------------------------------------
    // Phase 7c — bounded-heap top-k correctness + benchmark
    // -----------------------------------------------------------------

    use crate::sql::db::database::Database;
    use crate::sql::parser::select::SelectQuery;
    use sqlparser::dialect::SQLiteDialect;
    use sqlparser::parser::Parser;

    /// Builds a `docs(id INTEGER PK, score REAL)` table with N rows of
    /// distinct positive scores so top-k tests aren't sensitive to
    /// tie-breaking (heap is unstable; full-sort is stable; we want
    /// both to agree without arguing about equal-score row order).
    ///
    /// **Why positive scores:** the INSERT parser doesn't currently
    /// handle `Expr::UnaryOp(Minus, …)` for negative number literals
    /// (it would parse `-3.14` as a unary expression and the value
    /// extractor would skip it). That's a pre-existing bug, out of
    /// scope for 7c. Using the Knuth multiplicative hash gives us
    /// distinct positive scrambled values without dancing around the
    /// negative-literal limitation.
    fn seed_score_table(n: usize) -> Database {
        let mut db = Database::new("tempdb".to_string());
        crate::sql::process_command(
            "CREATE TABLE docs (id INTEGER PRIMARY KEY, score REAL);",
            &mut db,
        )
        .expect("create");
        for i in 0..n {
            // Knuth multiplicative hash mod 1_000_000 — distinct,
            // dense in [0, 999_999], no collisions for n up to ~tens
            // of thousands.
            let score = ((i as u64).wrapping_mul(2_654_435_761) % 1_000_000) as f64;
            let sql = format!("INSERT INTO docs (score) VALUES ({score});");
            crate::sql::process_command(&sql, &mut db).expect("insert");
        }
        db
    }

    /// Helper: parses an SQL SELECT into a SelectQuery so we can drive
    /// `select_topk` / `sort_rowids` directly without the rest of the
    /// process_command pipeline.
    fn parse_select(sql: &str) -> SelectQuery {
        let dialect = SQLiteDialect {};
        let mut ast = Parser::parse_sql(&dialect, sql).expect("parse");
        let stmt = ast.pop().expect("one statement");
        SelectQuery::new(&stmt).expect("select-query")
    }

    #[test]
    fn topk_matches_full_sort_asc() {
        // Build N=200, top-k=10. Bounded heap output must equal
        // full-sort-then-truncate output (both produce ASC order).
        let db = seed_score_table(200);
        let table = db.get_table("docs".to_string()).unwrap();
        let q = parse_select("SELECT * FROM docs ORDER BY score ASC LIMIT 10;");
        let order = q.order_by.as_ref().unwrap();
        let all_rowids = table.rowids();

        // Full-sort path
        let mut full = all_rowids.clone();
        sort_rowids(&mut full, table, order).unwrap();
        full.truncate(10);

        // Bounded-heap path
        let topk = select_topk(&all_rowids, table, order, 10).unwrap();

        assert_eq!(topk, full, "top-k via heap should match full-sort+truncate");
    }

    #[test]
    fn topk_matches_full_sort_desc() {
        // Same with DESC — verifies the direction-aware Ord wrapper.
        let db = seed_score_table(200);
        let table = db.get_table("docs".to_string()).unwrap();
        let q = parse_select("SELECT * FROM docs ORDER BY score DESC LIMIT 10;");
        let order = q.order_by.as_ref().unwrap();
        let all_rowids = table.rowids();

        let mut full = all_rowids.clone();
        sort_rowids(&mut full, table, order).unwrap();
        full.truncate(10);

        let topk = select_topk(&all_rowids, table, order, 10).unwrap();

        assert_eq!(
            topk, full,
            "top-k DESC via heap should match full-sort+truncate"
        );
    }

    #[test]
    fn topk_k_larger_than_n_returns_everything_sorted() {
        // The executor branches off to the full-sort path when k >= N,
        // but if a caller invokes select_topk directly with k > N, it
        // should still produce all-sorted output (no truncation
        // because we don't have N items to truncate to k).
        let db = seed_score_table(50);
        let table = db.get_table("docs".to_string()).unwrap();
        let q = parse_select("SELECT * FROM docs ORDER BY score ASC LIMIT 1000;");
        let order = q.order_by.as_ref().unwrap();
        let topk = select_topk(&table.rowids(), table, order, 1000).unwrap();
        assert_eq!(topk.len(), 50);
        // All scores in ascending order.
        let scores: Vec<f64> = topk
            .iter()
            .filter_map(|r| match table.get_value("score", *r) {
                Some(Value::Real(f)) => Some(f),
                _ => None,
            })
            .collect();
        assert!(scores.windows(2).all(|w| w[0] <= w[1]));
    }

    #[test]
    fn topk_k_zero_returns_empty() {
        let db = seed_score_table(10);
        let table = db.get_table("docs".to_string()).unwrap();
        let q = parse_select("SELECT * FROM docs ORDER BY score ASC LIMIT 1;");
        let order = q.order_by.as_ref().unwrap();
        let topk = select_topk(&table.rowids(), table, order, 0).unwrap();
        assert!(topk.is_empty());
    }

    #[test]
    fn topk_empty_input_returns_empty() {
        let db = seed_score_table(0);
        let table = db.get_table("docs".to_string()).unwrap();
        let q = parse_select("SELECT * FROM docs ORDER BY score ASC LIMIT 5;");
        let order = q.order_by.as_ref().unwrap();
        let topk = select_topk(&[], table, order, 5).unwrap();
        assert!(topk.is_empty());
    }

    #[test]
    fn topk_works_through_select_executor_with_distance_function() {
        // Integration check that the executor actually picks the
        // bounded-heap path on a KNN-shaped query and produces the
        // correct top-k.
        let mut db = Database::new("tempdb".to_string());
        crate::sql::process_command(
            "CREATE TABLE docs (id INTEGER PRIMARY KEY, e VECTOR(2));",
            &mut db,
        )
        .unwrap();
        // Five rows with distinct distances from probe [1.0, 0.0]:
        //   id=1 [1.0, 0.0]   distance=0
        //   id=2 [2.0, 0.0]   distance=1
        //   id=3 [0.0, 3.0]   distance=√(1+9) = √10 ≈ 3.16
        //   id=4 [1.0, 4.0]   distance=4
        //   id=5 [10.0, 10.0] distance=√(81+100) ≈ 13.45
        for v in &[
            "[1.0, 0.0]",
            "[2.0, 0.0]",
            "[0.0, 3.0]",
            "[1.0, 4.0]",
            "[10.0, 10.0]",
        ] {
            crate::sql::process_command(&format!("INSERT INTO docs (e) VALUES ({v});"), &mut db)
                .unwrap();
        }
        let resp = crate::sql::process_command(
            "SELECT id FROM docs ORDER BY vec_distance_l2(e, [1.0, 0.0]) ASC LIMIT 3;",
            &mut db,
        )
        .unwrap();
        // Top-3 closest to [1.0, 0.0] are id=1, id=2, id=3 (in that order).
        // The status message tells us how many rows came back.
        assert!(resp.contains("3 rows returned"), "got: {resp}");
    }

    /// Manual benchmark — not run by default. Recommended invocation:
    ///
    ///     cargo test -p sqlrite-engine --lib topk_benchmark --release \
    ///         -- --ignored --nocapture
    ///
    /// (`--release` matters: Rust's optimized sort gets very fast under
    /// optimization, so the heap's relative advantage is best observed
    /// against a sort that's also been optimized.)
    ///
    /// Measured numbers on an Apple Silicon laptop with N=10_000 + k=10:
    ///   - bounded heap:    ~820µs
    ///   - full sort+trunc: ~1.5ms
    ///   - ratio:           ~1.8×
    ///
    /// The advantage is real but moderate at this size because the sort
    /// key here is a single REAL column read (cheap) and Rust's sort_by
    /// has a very low constant factor. The asymptotic O(N log k) vs
    /// O(N log N) advantage scales with N and with per-row work — KNN
    /// queries where the sort key is `vec_distance_l2(col, [...])` are
    /// where this path really pays off, because each key evaluation is
    /// itself O(dim) and the heap path skips the per-row evaluation
    /// in the comparator (see `sort_rowids` for the contrast).
    #[test]
    #[ignore]
    fn topk_benchmark() {
        use std::time::Instant;
        const N: usize = 10_000;
        const K: usize = 10;

        let db = seed_score_table(N);
        let table = db.get_table("docs".to_string()).unwrap();
        let q = parse_select("SELECT * FROM docs ORDER BY score ASC LIMIT 10;");
        let order = q.order_by.as_ref().unwrap();
        let all_rowids = table.rowids();

        // Time bounded heap.
        let t0 = Instant::now();
        let _topk = select_topk(&all_rowids, table, order, K).unwrap();
        let heap_dur = t0.elapsed();

        // Time full sort + truncate.
        let t1 = Instant::now();
        let mut full = all_rowids.clone();
        sort_rowids(&mut full, table, order).unwrap();
        full.truncate(K);
        let sort_dur = t1.elapsed();

        let ratio = sort_dur.as_secs_f64() / heap_dur.as_secs_f64().max(1e-9);
        println!("\n--- topk_benchmark (N={N}, k={K}) ---");
        println!("  bounded heap:   {heap_dur:?}");
        println!("  full sort+trunc: {sort_dur:?}");
        println!("  speedup ratio:  {ratio:.2}×");

        // Soft assertion. Floor is 1.4× because the cheap-key
        // benchmark hovers around 1.8× empirically; setting this too
        // close to the measured value risks flaky CI on slower
        // runners. Floor of 1.4× still catches an actual regression
        // (e.g., if select_topk became O(N²) or stopped using the
        // heap entirely).
        assert!(
            ratio > 1.4,
            "bounded heap should be substantially faster than full sort, but ratio = {ratio:.2}"
        );
    }
}