powdb_query/executor/

mod.rs

//! PowDB query executor.

// Submodules that don't use macros defined in this file.
mod compiled;
mod eval;
pub mod mem_budget;

use crate::ast::*;
use crate::canonicalize::canonicalize_with_form;
use crate::plan::*;
use crate::plan_cache::PlanCache;
use crate::planner;
use crate::result::{QueryError, QueryResult};
use powdb_storage::catalog::Catalog;
use powdb_storage::row::{decode_row, RowLayout, ROW_MAGIC, ROW_PREFIX_SIZE};
use powdb_storage::types::*;
use powdb_storage::view::ViewRegistry;
pub use powdb_storage::wal::{WalDurabilityTicket, WalSyncMode};

use std::io;
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::time::Instant;
use tracing::{error, info, warn, Level};

use self::compiled::*;
use self::eval::*;

/// Legacy sentinel string constant — kept for backward compatibility with
/// any external code matching on the string representation. New code should
/// match on `QueryError::ReadonlyNeedsWrite` directly.
pub const READONLY_NEEDS_WRITE: &str = "__POWDB_READONLY_NEEDS_WRITE__";

/// Return the byte offset where the row body starts.
///
/// v0.5 rows begin with the `PROW` magic/version prefix. Legacy rows start
/// directly with the row body. Raw executor fast paths must add this base
/// before reading body-relative bitmap/data offsets.
#[inline]
pub(crate) fn row_body_base(row: &[u8]) -> usize {
    if row.len() >= ROW_PREFIX_SIZE && &row[0..4] == ROW_MAGIC {
        ROW_PREFIX_SIZE
    } else {
        0
    }
}

/// Query frontend dialect. PowQL remains the default/native dialect; SQL is
/// an explicit frontend that lowers to the same AST before planning.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QueryDialect {
    PowQL,
    Sql,
}

/// Plan cache capacity. Bench workloads fill ~15 slots; real apps will sit
/// comfortably in 256. Lookup is O(1), collisions clear the cache (see
/// `plan_cache::PlanCache::insert`).
const PLAN_CACHE_CAPACITY: usize = 256;
const SQL_RAW_CACHE_SALT: u64 = 0x7261_772d_7371_6c01;

#[inline]
fn sql_raw_cache_hash(hash: u64) -> u64 {
    hash ^ SQL_RAW_CACHE_SALT
}
/// Re-exported rather than re-declared: `Catalog::install_auto_wal_archive`
/// takes this exact type, and a second hand-written copy of it here is a
/// parity that has to be maintained by hand.
use powdb_storage::catalog::WalArchiveHook;

/// Maximum number of rows a join may produce before the executor aborts.
/// Prevents Cartesian-product blowups (e.g. `T cross join T` on 10K rows
/// would produce 100M rows in memory without this cap).
pub(super) const MAX_JOIN_ROWS: usize = 1_000_000;

/// Maximum candidate pairs allowed for a fallback nested-loop join. This is
/// grounded in the release benchmark's conservative 250 ms evaluation budget.
pub(super) const MAX_NESTED_LOOP_PAIRS: usize = 6_400_000;

/// Maximum number of rows that may be materialized for sorting.
/// Queries that exceed this should add a LIMIT clause to narrow the input
/// before sorting.
pub(super) const MAX_SORT_ROWS: usize = 10_000_000;

#[inline]
pub(super) fn check_join_limit(row_count: usize) -> Result<(), QueryError> {
    if row_count > MAX_JOIN_ROWS {
        return Err(QueryError::JoinLimitExceeded);
    }
    Ok(())
}

// ─── Mission D11 Phase 1: scalar hot-loop helpers ─────────────────────────
//
// These macros expand into the scan body of `agg_single_col_fast` and sit
// inside the `for_each_row_raw` closure. They exist to:
//
//   1. Split the loop on presence of a predicate *outside* the hot body,
//      so the no-predicate path (agg_sum/agg_min/agg_max bench workloads)
//      never pays the `Option<CompiledPredicate>` branch per row.
//   2. Drop two bounds checks per row by reading the null bitmap byte
//      and the 8-byte value via raw pointer casts.
//
// SAFETY (shared across every call site below):
//
//   - `$bmp_byte` is `col_idx / 8` where `col_idx < n_cols`, and the row body
//     encoding stores `bitmap_size = n_cols.div_ceil(8)` bytes of bitmap
//     starting at body offset 2. So `bmp_off = row_body_base(row) + 2 +
//     $bmp_byte < row_len`, and `get_unchecked(bmp_off)` is inside the
//     row slice.
//   - `$off = 2 + bitmap_size + fixed_offsets[col_idx]` is body-relative for a fixed-size
//     column. Every fixed-size column contributes `fixed_size(type_id)`
//     bytes to the fixed region, so the row always has
//     `[data_off .. data_off + 8]` available for any i64/f64 column, where
//     `data_off = row_body_base(row) + $off` — enforced by the row encoder
//     (`storage/src/row.rs`) and the schema invariant that a row with a
//     given schema has enough body bytes for `2 + bitmap_size + fixed_region_size`.
//   - Both macros are only invoked from `agg_single_col_fast`, which
//     early-returns if the column isn't Int/Float (8-byte fixed) and
//     early-returns if `fast.fixed_offsets[col_idx]` is `None`.
macro_rules! agg_int_loop {
    (
        $self:expr, $table:expr, $pred:expr,
        $bmp_byte:expr, $bmp_bit:expr, $off:expr,
        |$v:ident : i64| $body:block
    ) => {{
        let bmp_byte = $bmp_byte;
        let bmp_bit = $bmp_bit;
        let off = $off;
        if let Some(pred) = &$pred {
            for_each_row_raw_cancellable(&$self.catalog, $table, |_rid, data| {
                if !pred(data) {
                    return;
                }
                let base = row_body_base(data);
                let bmp_off = base + 2 + bmp_byte;
                let data_off = base + off;
                // Bounds guard: skip corrupt/truncated rows that are too
                // short to contain the bitmap byte or the 8-byte value.
                if bmp_off >= data.len() || data_off + 8 > data.len() {
                    return;
                }
                // SAFETY: `bmp_off < data.len()` is checked above.
                // The bitmap byte lives at body offset 2..2+bitmap_size in
                // the row encoding, and bmp_byte = col_idx / 8 < bitmap_size.
                // Corrupt rows are rejected by the bounds guard.
                let bmp = unsafe { *data.get_unchecked(bmp_off) };
                if (bmp >> bmp_bit) & 1 == 1 {
                    return;
                }
                // SAFETY: `data_off + 8 <= data.len()` is checked above.
                // `data_off = base + 2 + bitmap_size + fixed_offsets[col_idx]`
                // points to an 8-byte i64 in the fixed-size region of the row.
                // The pointer cast is valid because we read exactly 8
                // bytes via from_le_bytes. Corrupt rows are rejected by
                // the bounds guard.
                let $v: i64 =
                    unsafe { i64::from_le_bytes(*(data.as_ptr().add(data_off) as *const [u8; 8])) };
                $body
            })?;
        } else {
            for_each_row_raw_cancellable(&$self.catalog, $table, |_rid, data| {
                let base = row_body_base(data);
                let bmp_off = base + 2 + bmp_byte;
                let data_off = base + off;
                // Bounds guard: skip corrupt/truncated rows.
                if bmp_off >= data.len() || data_off + 8 > data.len() {
                    return;
                }
                // SAFETY: `bmp_off < data.len()` is checked above.
                // See the predicate branch for the full invariant.
                let bmp = unsafe { *data.get_unchecked(bmp_off) };
                if (bmp >> bmp_bit) & 1 == 1 {
                    return;
                }
                // SAFETY: `data_off + 8 <= data.len()` is checked above.
                // See the predicate branch for the full invariant.
                let $v: i64 =
                    unsafe { i64::from_le_bytes(*(data.as_ptr().add(data_off) as *const [u8; 8])) };
                $body
            })?;
        }
    }};
}

macro_rules! agg_float_loop {
    (
        $self:expr, $table:expr, $pred:expr,
        $bmp_byte:expr, $bmp_bit:expr, $off:expr,
        |$v:ident : f64| $body:block
    ) => {{
        let bmp_byte = $bmp_byte;
        let bmp_bit = $bmp_bit;
        let off = $off;
        if let Some(pred) = &$pred {
            for_each_row_raw_cancellable(&$self.catalog, $table, |_rid, data| {
                if !pred(data) {
                    return;
                }
                let base = row_body_base(data);
                let bmp_off = base + 2 + bmp_byte;
                let data_off = base + off;
                // Bounds guard: skip corrupt/truncated rows that are too
                // short to contain the bitmap byte or the 8-byte value.
                if bmp_off >= data.len() || data_off + 8 > data.len() {
                    return;
                }
                // SAFETY: `bmp_off < data.len()` is checked above.
                // The bitmap byte lives at body offset 2..2+bitmap_size in
                // the row encoding, and bmp_byte = col_idx / 8 < bitmap_size.
                // Corrupt rows are rejected by the bounds guard.
                let bmp = unsafe { *data.get_unchecked(bmp_off) };
                if (bmp >> bmp_bit) & 1 == 1 {
                    return;
                }
                // SAFETY: `data_off + 8 <= data.len()` is checked above.
                // `data_off = base + 2 + bitmap_size + fixed_offsets[col_idx]`
                // points to an 8-byte f64 in the fixed-size region of the row.
                // The pointer cast is valid because we read exactly 8
                // bytes via from_le_bytes. Corrupt rows are rejected by
                // the bounds guard.
                let $v: f64 =
                    unsafe { f64::from_le_bytes(*(data.as_ptr().add(data_off) as *const [u8; 8])) };
                $body
            })?;
        } else {
            for_each_row_raw_cancellable(&$self.catalog, $table, |_rid, data| {
                let base = row_body_base(data);
                let bmp_off = base + 2 + bmp_byte;
                let data_off = base + off;
                // Bounds guard: skip corrupt/truncated rows.
                if bmp_off >= data.len() || data_off + 8 > data.len() {
                    return;
                }
                // SAFETY: `bmp_off < data.len()` is checked above.
                // See the predicate branch for the full invariant.
                let bmp = unsafe { *data.get_unchecked(bmp_off) };
                if (bmp >> bmp_bit) & 1 == 1 {
                    return;
                }
                // SAFETY: `data_off + 8 <= data.len()` is checked above.
                // See the predicate branch for the full invariant.
                let $v: f64 =
                    unsafe { f64::from_le_bytes(*(data.as_ptr().add(data_off) as *const [u8; 8])) };
                $body
            })?;
        }
    }};
}

// Submodules that use the macros above — must be declared after macro_rules!.
mod plan_exec;
mod prepared;

#[cfg(test)]
mod tests;

// Re-exports for the public API
pub use self::prepared::PreparedQuery;

use self::plan_exec::{
    aggregate_rows, aggregate_rows_with_provenance, compare_order_values,
    cooperative_stable_sort_by, counts_every_row, exec_group_by, exec_group_by_with_provenance,
    execute_materialized_join, execute_window, for_each_row_raw_cancellable, format_plan_tree,
    literal_limit, predicate_column_indices_json, range_matches, synthesize_range_predicate,
    union_rows, validate_plan, LoweredPlan,
};

/// Mission infra-1: classify a parsed statement as read-only vs. mutating.
/// Used by [`Engine::execute_powql_readonly`] and by the server handler
/// to decide between the RwLock reader and writer sides. `Union` recurses
/// because each side can independently be read/write (though in practice
/// both sides are reads — the parser only builds Union from query shapes).
pub fn is_read_only_statement(stmt: &Statement) -> bool {
    match stmt {
        Statement::Query(_) => true,
        Statement::ListTypes | Statement::Describe(_) | Statement::ListLinks => true,
        Statement::Union(u) => is_read_only_statement(&u.left) && is_read_only_statement(&u.right),
        Statement::Insert(_)
        | Statement::Upsert(_)
        | Statement::UpdateQuery(_)
        | Statement::DeleteQuery(_)
        | Statement::CreateType(_)
        | Statement::CreateLink(_)
        | Statement::DropLink(_)
        | Statement::AlterTable(_)
        | Statement::DropTable(_)
        | Statement::CreateView(_)
        | Statement::RefreshView(_)
        | Statement::DropView(_) => false,
        Statement::Begin | Statement::Commit | Statement::Rollback => false,
        Statement::Explain(inner) => is_read_only_statement(inner),
    }
}

/// Map a read-only executor result into the read-only-engine surface: the
/// internal "this statement writes" sentinel ([`QueryError::ReadonlyNeedsWrite`])
/// becomes the terminal, operator-facing [`QueryError::ReadonlyMode`]. There is
/// no writer to escalate to in read-only mode, so the sentinel never leaves the
/// engine.
fn to_readonly_terminal(
    result: Result<QueryResult, QueryError>,
) -> Result<QueryResult, QueryError> {
    match result {
        Err(QueryError::ReadonlyNeedsWrite) => Err(QueryError::ReadonlyMode),
        other => other,
    }
}

/// Return whether executing this read plan would have to refresh a dirty
/// materialized view. This is intentionally a whole-plan preflight: the server
/// may retry only this typed condition under exclusive admission, so it must be
/// raised before any input branch performs row work.
fn plan_reads_dirty_view(plan: &PlanNode, views: &ViewRegistry) -> bool {
    match plan {
        PlanNode::SeqScan { table }
        | PlanNode::AliasScan { table, .. }
        | PlanNode::IndexScan { table, .. }
        | PlanNode::RangeScan { table, .. }
        | PlanNode::ExprIndexScan { table, .. }
        | PlanNode::ExprRangeScan { table, .. }
        | PlanNode::OrderedExprIndexScan { table, .. } => views.is_dirty(table),

        PlanNode::Filter { input, .. }
        | PlanNode::Project { input, .. }
        | PlanNode::Sort { input, .. }
        | PlanNode::Limit { input, .. }
        | PlanNode::Offset { input, .. }
        | PlanNode::Aggregate { input, .. }
        | PlanNode::Distinct { input }
        | PlanNode::GroupBy { input, .. }
        | PlanNode::Window { input, .. } => plan_reads_dirty_view(input, views),

        PlanNode::NestedLoopJoin { left, right, .. } | PlanNode::Union { left, right, .. } => {
            plan_reads_dirty_view(left, views) || plan_reads_dirty_view(right, views)
        }

        // Nested sub-query fields scan their child tables directly, so a
        // dirty child view needs the same escalation as the parent input.
        PlanNode::NestedProject { input, fields } => {
            plan_reads_dirty_view(input, views)
                || fields.iter().any(|field| match field {
                    crate::plan::NestedProjectField::Nested(nested) => {
                        let mut dirty = false;
                        nested.visit_tables(&mut |table| dirty |= views.is_dirty(table));
                        // A block link traversal's child table is a placeholder
                        // until execution-time catalog resolution, so
                        // visit_tables cannot see it; escalate conservatively
                        // whenever any view is dirty.
                        dirty
                            || (plan_exec::nested_fields_have_via_link(std::slice::from_ref(field))
                                && any_view_dirty(views))
                    }
                    crate::plan::NestedProjectField::Plain(_) => false,
                    // Scalar link hop tables are likewise unknown here.
                    crate::plan::NestedProjectField::Link(link) => {
                        link.resolved.is_none() && any_view_dirty(views)
                    }
                })
        }

        // EXPLAIN formats its input without executing it, so inspecting a plan
        // that names a dirty view never requires a refresh.
        PlanNode::Explain { .. }
        | PlanNode::AlterTable { .. }
        | PlanNode::DropTable { .. }
        | PlanNode::Insert { .. }
        | PlanNode::Upsert { .. }
        | PlanNode::Update { .. }
        | PlanNode::Delete { .. }
        | PlanNode::CreateTable { .. }
        | PlanNode::CreateLink { .. }
        | PlanNode::DropLink { .. }
        | PlanNode::ListTypes
        | PlanNode::Describe { .. }
        | PlanNode::ListLinks
        | PlanNode::CreateView { .. }
        | PlanNode::RefreshView { .. }
        | PlanNode::DropView { .. }
        | PlanNode::Begin
        | PlanNode::Commit
        | PlanNode::Rollback => false,
    }
}

/// Open the materialized-view registry over `data_dir`, marking every view
/// whose stored source no longer parses as dirty.
///
/// Dirty is how a read is made to go through `Engine::refresh_view`, which is
/// where an unreadable source is reported (see
/// `plan_exec::dispatch::parse_stored_view_source`). Without this, such a view
/// is CLEAN forever: its dependency list was empty for the same reason its
/// source will not parse, so no mutation ever dirties it, no read ever
/// refreshes it, and every read serves stale rows with no error at all. Marking
/// it dirty converts that into the typed error, and touches only memory, so a
/// database that is merely being inspected is not modified.
/// Whether `data_dir` already holds a catalog, i.e. whether reopening it is
/// meant to find an existing database.
///
/// The file name is the on-disk identity of a PowDB directory, shared by the
/// backup, sync and server crates; it is a format constant, not a storage
/// implementation detail, and changing it would be a format break carrying its
/// own migration.
fn catalog_file_present(data_dir: &Path) -> bool {
    data_dir.join("catalog.bin").exists()
}

/// Open the view registry, refusing the whole database when the file is there
/// but unreadable.
///
/// [`ViewRegistry::open`] already answers "no views" for an absent file, so an
/// error here means `views.bin` exists and did not decode. Treating that as an
/// empty registry lost every `materialize` statement ever run: reads of a view
/// name reported an unknown table, the backing tables stayed on disk with
/// nothing to refresh them, and the next `materialize` overwrote the file that
/// still held the definitions. A damaged heap and a missing catalog both refuse
/// the open (see `new_inner`), and so does this.
fn open_view_registry(data_dir: &Path) -> io::Result<ViewRegistry> {
    let mut registry = ViewRegistry::open(data_dir).map_err(|e| {
        io::Error::new(
            e.kind(),
            format!(
                "{} could not be read ({e}); it defines this database's materialized views,                  so it must be restored or removed before the database can be opened",
                data_dir
                    .join(powdb_storage::data_dir::VIEW_REGISTRY_FILE)
                    .display()
            ),
        )
    })?;
    let unreadable: Vec<String> = registry
        .list_views()
        .iter()
        .filter(|name| {
            registry
                .get(name)
                .is_some_and(|def| crate::parser::parse(&def.query).is_err())
        })
        .map(|name| (*name).to_string())
        .collect();
    for name in unreadable {
        warn!(
            view = %name,
            "materialized view has an unparseable stored source; reads will report an error \
             rather than serve rows that cannot be recomputed"
        );
        registry.mark_dirty_in_memory(&name);
    }
    Ok(registry)
}

/// Names of the dirty materialized views this plan scans, taken from table
/// names the plan itself carries.
///
/// Deliberately narrower than [`plan_reads_dirty_view`]: that one escalates
/// conservatively for link traversals whose child table is only knowable after
/// execution-time catalog resolution, which is the right answer for a preflight
/// that can only say yes or no, but it cannot name a table to refresh. The
/// write path refreshes those where it resolves them.
fn collect_dirty_scanned_views(plan: &PlanNode, views: &ViewRegistry, out: &mut Vec<String>) {
    match plan {
        PlanNode::SeqScan { table }
        | PlanNode::AliasScan { table, .. }
        | PlanNode::IndexScan { table, .. }
        | PlanNode::RangeScan { table, .. }
        | PlanNode::ExprIndexScan { table, .. }
        | PlanNode::ExprRangeScan { table, .. }
        | PlanNode::OrderedExprIndexScan { table, .. } => {
            if views.is_dirty(table) && !out.iter().any(|name| name == table) {
                out.push(table.clone());
            }
        }

        PlanNode::Filter { input, .. }
        | PlanNode::Project { input, .. }
        | PlanNode::Sort { input, .. }
        | PlanNode::Limit { input, .. }
        | PlanNode::Offset { input, .. }
        | PlanNode::Aggregate { input, .. }
        | PlanNode::Distinct { input }
        | PlanNode::GroupBy { input, .. }
        | PlanNode::Window { input, .. }
        | PlanNode::NestedProject { input, .. } => collect_dirty_scanned_views(input, views, out),

        PlanNode::NestedLoopJoin { left, right, .. } | PlanNode::Union { left, right, .. } => {
            collect_dirty_scanned_views(left, views, out);
            collect_dirty_scanned_views(right, views, out);
        }

        // EXPLAIN formats its input without executing it, and the write
        // statements own their own view bookkeeping.
        _ => {}
    }
}

/// True when any registered materialized view is currently dirty. Conservative
/// escalation test for plans whose scanned tables are not knowable before
/// execution-time catalog resolution (link traversals resolve their child
/// tables from the persistent catalog at query time).
fn any_view_dirty(views: &ViewRegistry) -> bool {
    views.list_views().iter().any(|v| views.is_dirty(v))
}

pub struct Engine {
    catalog: Catalog,
    /// Exclusive PID-based lock on the data directory, held for the engine's
    /// lifetime so two separate processes can't open the same dir and corrupt
    /// the heap/WAL. Released on clean drop; a `mem::forget` crash leaves a
    /// stale lock the next open takes over. Leading `_`: it does its work
    /// through `Drop`, never read directly.
    _dir_lock: powdb_storage::dir_lock::DirLock,
    /// Mission D9 — cached parsed+planned query trees keyed by canonical
    /// hash. Saves the ~3μs parse+plan cost on repeat queries that differ
    /// only in literal values.
    ///
    /// Mission infra-1: wrapped in `Mutex` so the read path can be driven
    /// by `&self`. The critical section is extremely short — a single
    /// hashmap lookup + plan clone on a hit, or a single insert on a miss.
    /// A full `RwLock` would be over-engineered here; the contention window
    /// is smaller than the read-path scan work it gates.
    plan_cache: Mutex<PlanCache>,
    /// Mission C Phase 13: reusable `Vec<Value>` scratch buffer for the
    /// prepared-insert fast path. `execute_prepared` used to allocate a
    /// fresh `vec![Value::Empty; n_cols]` on every insert; recycling this
    /// buffer shaves one heap alloc per row on `insert_batch_1k`.
    insert_values_scratch: Vec<Value>,
    /// Materialized view registry: tracks view definitions, dependencies,
    /// and dirty state. Views are backed by regular catalog tables; this
    /// registry adds the lifecycle metadata.
    view_registry: ViewRegistry,
    in_transaction: bool,
    /// WS2 — per-query memory budget ceiling (bytes). The running total lives
    /// in a thread-local (see [`mem_budget`]) and is reset at every top-level
    /// query entry, so sort/join/GROUP BY/IN-list materialization can be capped
    /// without OOM-killing the process. This field holds only the *limit* (a
    /// plain `usize`, so `Engine` stays `Sync` for the concurrent read path).
    /// Default [`mem_budget::DEFAULT_QUERY_MEMORY_LIMIT`] (256 MB); overridable
    /// via `Engine::with_memory_limit` (server reads `POWDB_QUERY_MEMORY_LIMIT`).
    query_memory_limit: usize,
    /// Maximum candidate pairs a fallback nested-loop join may evaluate before
    /// it is rejected. Default [`MAX_NESTED_LOOP_PAIRS`], overridable via
    /// [`Engine::set_nested_loop_pair_limit`] (server reads
    /// `POWDB_MAX_NESTED_LOOP_PAIRS`). A plain `usize` so `Engine` stays `Sync`.
    nested_loop_pair_limit: usize,
    wal_archive_hook: Option<WalArchiveHook>,
    /// True when opened via [`Engine::open_read_only`] for snapshot serving. In
    /// this mode the catalog and its files are read-only, the `DirLock` is a
    /// shared reader lock, and every mutating execute path returns the terminal
    /// [`QueryError::ReadonlyMode`] instead of ever touching disk.
    read_only: bool,
    /// Test-only: when true, every executor fast path and the compiled
    /// predicate entry decline to fire, so the query runs through the generic
    /// evaluator. See [`Engine::set_force_generic_path`]. The field only
    /// exists under the `testing` feature, and [`Engine::generic_path_forced`]
    /// is a compile-time `false` without it, so no shipped build pays a branch.
    #[cfg(feature = "testing")]
    force_generic_path: bool,
    /// Test-only: the name of every fast-path check site that has declined
    /// because `force_generic_path` was set. A bare count would only tell a
    /// test that *something* declined; a query usually passes several check
    /// sites, so a count stays non-zero even when the one site under test lost
    /// its guard. Recording names is what makes "this shape really was
    /// diverted" provable, and a missed site is a hole in the whole runner.
    /// Behind a `Mutex` because the read path drives the engine by `&self` and
    /// `Engine` must stay `Sync`; it is only ever locked while the switch is on.
    #[cfg(feature = "testing")]
    forced_generic_sites: std::sync::Mutex<Vec<&'static str>>,
}

impl Engine {
    /// Open or create a PowDB engine rooted at `data_dir`.
    ///
    /// If the directory already contains a catalog, it is reopened.
    /// Otherwise a fresh empty database is created.
    ///
    /// # Examples
    ///
    /// ```
    /// use powdb_query::executor::Engine;
    ///
    /// let dir = tempfile::tempdir().unwrap();
    /// let engine = Engine::new(dir.path()).unwrap();
    /// // Engine is ready — the directory now contains a catalog.
    /// ```
    pub fn new(data_dir: &Path) -> io::Result<Self> {
        Self::new_inner(data_dir, None)
    }

    /// Open or create an engine that archives WAL records before any recovery,
    /// rollback, or drop checkpoint truncates them. This keeps the query crate
    /// independent of replication metadata while giving sync-aware callers one
    /// lifecycle boundary for retained-history preservation.
    pub fn new_with_wal_archive<F>(data_dir: &Path, archive: F) -> io::Result<Self>
    where
        F: Fn(&Path, &[powdb_storage::wal::WalRecord]) -> io::Result<()> + Send + Sync + 'static,
    {
        Self::new_inner(data_dir, Some(Arc::new(archive)))
    }

    fn new_inner(data_dir: &Path, wal_archive_hook: Option<WalArchiveHook>) -> io::Result<Self> {
        powdb_storage::create_data_dir_secure(data_dir)?;
        // Refuse to open a directory another live process already holds, before
        // touching any on-disk state (concurrent writers corrupt the heap/WAL).
        let dir_lock = powdb_storage::dir_lock::DirLock::acquire(data_dir)?;
        // Try to reopen an existing database first; only create a fresh
        // catalog when there isn't one already on disk.
        let catalog_result = match &wal_archive_hook {
            Some(hook) => {
                let hook = Arc::clone(hook);
                Catalog::open_with_wal_archive(data_dir, move |dir, records| hook(dir, records))
            }
            None => Catalog::open(data_dir),
        };
        let mut catalog = match catalog_result {
            Ok(c) => {
                info!(data_dir = %data_dir.display(), "engine reopened existing database");
                c
            }
            // Exactly one NotFound means "there is no database here": the one
            // `Catalog::open` raises because the catalog file itself is absent.
            // Every other file the open path touches raises the same kind, a
            // table's `.heap` most obviously, and `Catalog::create` renames a
            // fresh empty catalog over the existing one and truncates the WAL.
            // Treating any NotFound as a fresh directory therefore turned a
            // single missing heap into silent, total loss of every table,
            // including the ones still intact on disk. Damage must stay loud so
            // the missing file can be restored.
            Err(e) if e.kind() == io::ErrorKind::NotFound && !catalog_file_present(data_dir) => {
                info!(data_dir = %data_dir.display(), "engine initialized fresh database");
                Catalog::create(data_dir)?
            }
            Err(e) => return Err(e),
        };
        // `Catalog::open_with_wal_archive` borrows its hook for the open and
        // cannot keep it, so a catalog that had one used to stop checkpointing
        // altogether rather than truncate records it could not publish. Hand it
        // one it can keep: the automatic checkpoint archives, then truncates.
        if let Some(hook) = &wal_archive_hook {
            catalog.install_auto_wal_archive(Arc::clone(hook));
        }
        let view_registry = open_view_registry(data_dir)?;
        Ok(Engine {
            catalog,
            _dir_lock: dir_lock,
            plan_cache: Mutex::new(PlanCache::new(PLAN_CACHE_CAPACITY)),
            insert_values_scratch: Vec::new(),
            view_registry,
            in_transaction: false,
            query_memory_limit: mem_budget::DEFAULT_QUERY_MEMORY_LIMIT,
            nested_loop_pair_limit: MAX_NESTED_LOOP_PAIRS,
            wal_archive_hook,
            read_only: false,
            #[cfg(feature = "testing")]
            force_generic_path: false,
            #[cfg(feature = "testing")]
            forced_generic_sites: std::sync::Mutex::new(Vec::new()),
        })
    }

    /// Open an engine **read-only** over a quiescent data directory for snapshot
    /// serving (tier 1 of the replica story). This is the supported way to serve
    /// a restored backup or a checkpointed replica with no write gate at all:
    ///
    /// - the catalog and every heap/index/WAL file are opened read-only
    ///   ([`Catalog::open_read_only`]), so nothing on disk is ever mutated;
    /// - a **shared reader** [`DirLock`](powdb_storage::dir_lock::DirLock) is
    ///   taken, so N read-only processes may serve the same directory
    ///   concurrently, while a read-write open refuses to start against live
    ///   readers (and readers refuse a live writer);
    /// - a non-empty WAL is refused with an actionable error rather than replayed
    ///   (the directory must be recovered by a read-write engine first);
    /// - every mutating statement returns the terminal [`QueryError::ReadonlyMode`].
    ///
    /// Materialized-view refresh must happen before snapshotting: a query over a
    /// stale (dirty) view is refused in this mode rather than silently escalating.
    pub fn open_read_only(data_dir: &Path) -> io::Result<Self> {
        // Validate readability without ever chmod-ing the directory (a read-only
        // open must leave it byte-identical).
        powdb_storage::validate_data_dir_read_only(data_dir)?;
        // A shared reader lock: coexists with other readers, refuses a live writer.
        let dir_lock = powdb_storage::dir_lock::DirLock::acquire_reader(data_dir)?;
        let catalog = Catalog::open_read_only(data_dir)?;
        info!(data_dir = %data_dir.display(), "engine opened read-only for snapshot serving");
        let view_registry = open_view_registry(data_dir)?;
        Ok(Engine {
            catalog,
            _dir_lock: dir_lock,
            plan_cache: Mutex::new(PlanCache::new(PLAN_CACHE_CAPACITY)),
            insert_values_scratch: Vec::new(),
            view_registry,
            in_transaction: false,
            query_memory_limit: mem_budget::DEFAULT_QUERY_MEMORY_LIMIT,
            nested_loop_pair_limit: MAX_NESTED_LOOP_PAIRS,
            // No WAL-archive hook: a read-only engine never writes, so its Drop
            // must never checkpoint (the hook is what would drive that).
            wal_archive_hook: None,
            read_only: true,
            #[cfg(feature = "testing")]
            force_generic_path: false,
            #[cfg(feature = "testing")]
            forced_generic_sites: std::sync::Mutex::new(Vec::new()),
        })
    }

    /// Read-only open with an explicit per-query memory budget (bytes).
    pub fn open_read_only_with_memory_limit(
        data_dir: &Path,
        limit_bytes: usize,
    ) -> io::Result<Self> {
        let mut engine = Engine::open_read_only(data_dir)?;
        engine.set_query_memory_limit(limit_bytes);
        Ok(engine)
    }

    /// Whether this engine was opened read-only for snapshot serving.
    pub fn is_read_only(&self) -> bool {
        self.read_only
    }

    /// Plan `stmt` and lower it. **The only place in the executor that calls
    /// the planner.**
    ///
    /// Planner output is not executable as it stands: the planner is pure, so
    /// it emits `IndexScan` / `RangeScan` probes speculatively and leaves every
    /// literal exactly as written, and lowering is what decides whether those
    /// probes exist and what key bytes they address. Running raw planner output
    /// is therefore not a missed optimization, it is a different answer.
    ///
    /// Eight subquery materialization sites used to plan a statement and hand
    /// the result straight to `execute_plan` / `execute_plan_readonly`. The
    /// consequence was that a predicate answered correctly at the top level and
    /// incorrectly one level of nesting down:
    /// `count(H filter .price < 3)` gave 2 while
    /// `count(H filter .n in (H filter .price < 3 { .n }))` gave 0 over the
    /// same rows and the same index.
    ///
    /// Funnelling every plan through here is what makes that unrepeatable: a
    /// ninth site has to call this to get a plan at all, and what it gets back
    /// is a [`LoweredPlan`], which is the only thing the execution entry points
    /// accept. `no_execution_entry_point_can_receive_an_unlowered_plan` in
    /// `tests/cross_type_index_parity.rs` fails the build if a second call to
    /// the planner appears anywhere under `src/executor/`.
    fn plan_and_lower(&self, stmt: Statement) -> Result<LoweredPlan, QueryError> {
        Ok(self.plan_and_lower_cacheable(stmt)?.1)
    }

    /// [`Engine::plan_and_lower`] for the callers that also need the raw plan,
    /// which is only ever the plan cache: the cache stores the pre-lowering
    /// tree, because lowering is a function of catalog state and the cache
    /// outlives DDL. Lowering therefore runs on every cache hit as well.
    fn plan_and_lower_cacheable(
        &self,
        stmt: Statement,
    ) -> Result<(PlanNode, LoweredPlan), QueryError> {
        let plan =
            crate::planner::plan_statement(stmt).map_err(|e| QueryError::Parse(e.to_string()))?;
        let lowered = self.lower(&plan)?;
        Ok((plan, lowered))
    }

    /// [`Engine::plan_and_lower_cacheable`] from PowQL text. The only call to
    /// `planner::plan` under `src/executor/`, for the same reason.
    fn plan_text_and_lower(&self, input: &str) -> Result<(PlanNode, LoweredPlan), QueryError> {
        let plan = planner::plan(input).map_err(|e| QueryError::Parse(e.to_string()))?;
        let lowered = self.lower(&plan)?;
        Ok((plan, lowered))
    }

    /// Lower a plan that came from somewhere other than the planner: the plan
    /// cache, or a fallback the executor built from an already-lowered tree.
    /// Lowering is idempotent, so calling it on a plan that has been through it
    /// already is a no-op.
    fn lower(&self, plan: &PlanNode) -> Result<LoweredPlan, QueryError> {
        LoweredPlan::of(&self.catalog, plan)
    }

    /// Test-only: the EXPLAIN text of `query`'s plan after `passes` rounds of
    /// lowering.
    ///
    /// Routing every plan through [`Engine::plan_and_lower`] is only sound if
    /// lowering is idempotent: a plan that reaches an entry point already
    /// lowered (a re-lowered cache hit, a prepared template, a fallback the
    /// executor rebuilt from a lowered tree) must come out unchanged rather
    /// than rewritten a second time. This exposes the pass so a test can hold
    /// that directly instead of inferring it from answers.
    ///
    /// Only available with the `testing` feature.
    #[cfg(feature = "testing")]
    pub fn lowered_plan_text(&self, query: &str, passes: usize) -> Result<String, QueryError> {
        let (_, mut plan) = self.plan_text_and_lower(query)?;
        for _ in 1..passes.max(1) {
            plan = self.lower(plan.node())?;
        }
        Ok(format_plan_tree(&self.catalog, plan.node(), 0))
    }

    /// Run a lowered plan on the mutable path. `Engine::execute_plan` stays
    /// public and takes a bare `&PlanNode` for embedders that build plans
    /// themselves, and lowers it before calling this; every path inside the
    /// executor goes through here so that what it runs is a plan the type
    /// system says was lowered.
    fn execute_lowered(&mut self, plan: &LoweredPlan) -> Result<QueryResult, QueryError> {
        self.refresh_dirty_views_read_by(plan.node())?;
        let result = self.dispatch_mut(plan.node());
        Self::raise_arith_fault(result)
    }

    /// Turn a fault the expression evaluator recorded during `result` into the
    /// error it had no channel to report itself.
    ///
    /// An overflowing `+` or a per-row zero divisor evaluates to `Value::Empty`,
    /// which is indistinguishable from a missing column: the row came back with
    /// a NULL in a projection, matched nothing in a predicate, and an `update`
    /// wrote that NULL into a required column. An existing error wins, because
    /// it is the more specific one; either way the slot is cleared.
    fn raise_arith_fault(
        result: Result<QueryResult, QueryError>,
    ) -> Result<QueryResult, QueryError> {
        match (result, take_arith_fault()) {
            (Ok(_), Some(message)) => Err(QueryError::Execution(message)),
            (result, _) => result,
        }
    }

    /// Refresh every stale materialized view this plan is about to read, before
    /// any of it runs.
    ///
    /// The dispatch arms each carry their own dirty check, and the fast paths
    /// that short-circuit them do not: `V filter .id > 0 { .id }` is served by
    /// `project_filter_limit_fast`, which takes the table name and scans the
    /// backing heap directly, so a stale view answered with its pre-mutation
    /// rows while the bare `V` next to it answered correctly. Checking at every
    /// fast path is how that happened; checking once, here, at the boundary
    /// every statement and subquery crosses, is what makes it not recur.
    ///
    /// The walk only runs when a view is dirty at all, and only refreshes views
    /// the plan actually names, so refreshing one cannot re-enter itself
    /// through its own source query.
    fn refresh_dirty_views_read_by(&mut self, plan: &PlanNode) -> Result<(), QueryError> {
        if !any_view_dirty(&self.view_registry) {
            return Ok(());
        }
        let mut stale: Vec<String> = Vec::new();
        collect_dirty_scanned_views(plan, &self.view_registry, &mut stale);
        for name in stale {
            self.refresh_view(&name)?;
        }
        Ok(())
    }

    /// Test-only: force every query onto the generic evaluator.
    ///
    /// The planner is pure, so one query text runs through different physical
    /// code depending on catalog state and plan shape: a compiled byte-level
    /// predicate, a fused scan, a bounded top-N heap, an index probe, or the
    /// generic decode-and-evaluate loop. Those paths are supposed to be
    /// indistinguishable from outside; twice now they were not, and the
    /// disagreement shipped because no test ran the same query both ways.
    ///
    /// With this set, predicate compilation declines and every fast-path match
    /// site falls through, so a test can execute a query optimized and
    /// unoptimized and diff the two results.
    /// It changes which code runs, never what the answer should be: any
    /// difference it exposes is a bug in one of the two paths.
    ///
    /// Each decline records the name of the site that declined; read them back
    /// with [`Engine::forced_generic_sites`]. A comparison whose shape never
    /// declined anywhere is comparing one code path with itself, so the names
    /// are what keep the comparison honest.
    ///
    /// Only available with the `testing` feature; there is no way to reach it
    /// from a shipped build.
    #[cfg(feature = "testing")]
    pub fn set_force_generic_path(&mut self, force: bool) {
        self.force_generic_path = force;
    }

    /// Whether fast paths are currently suppressed, recording `site` when they
    /// are. `site` names the specific fast path being declined, so a test can
    /// assert that the shape it ran was diverted *there* rather than merely
    /// somewhere. Compiles to a constant `false` without the `testing` feature,
    /// so every guarded branch folds away in a shipped build.
    #[cfg(feature = "testing")]
    #[inline]
    pub(in crate::executor) fn generic_path_forced(&self, site: &'static str) -> bool {
        if self.force_generic_path {
            if let Ok(mut sites) = self.forced_generic_sites.lock() {
                sites.push(site);
            }
            return true;
        }
        false
    }

    /// Test-only: the fast-path check sites that have declined so far because
    /// [`Engine::set_force_generic_path`] is on, in the order they declined. A
    /// shape that is supposed to reach a named fast path but never appears here
    /// has a missing check site, not a passing test.
    #[cfg(feature = "testing")]
    pub fn forced_generic_sites(&self) -> Vec<&'static str> {
        match self.forced_generic_sites.lock() {
            Ok(sites) => sites.clone(),
            Err(poisoned) => poisoned.into_inner().clone(),
        }
    }

    /// Test-only: forget the recorded declines so one query's declines can be
    /// observed in isolation.
    #[cfg(feature = "testing")]
    pub fn reset_forced_generic_sites(&self) {
        match self.forced_generic_sites.lock() {
            Ok(mut sites) => sites.clear(),
            Err(poisoned) => poisoned.into_inner().clear(),
        }
    }

    #[cfg(not(feature = "testing"))]
    #[inline]
    pub(in crate::executor) fn generic_path_forced(&self, _site: &'static str) -> bool {
        false
    }

    /// Compile `predicate` into a byte-level closure, unless fast paths are
    /// suppressed. This is the single entry point every executor site uses, so
    /// [`Engine::set_force_generic_path`] cannot be bypassed by a caller that
    /// forgets the check.
    #[inline]
    pub(in crate::executor) fn compile_predicate_unless_forced(
        &self,
        site: &'static str,
        predicate: &Expr,
        columns: &[String],
        layout: &FastLayout,
        schema: &Schema,
    ) -> Option<CompiledPredicate> {
        if self.generic_path_forced(site) {
            return None;
        }
        compile_predicate(predicate, columns, layout, schema)
    }

    /// Open or create an engine with an explicit per-query memory limit
    /// (bytes). Used by the server to apply `POWDB_QUERY_MEMORY_LIMIT`, and by
    /// tests that need a tiny limit to exercise the budget guard.
    pub fn with_memory_limit(data_dir: &Path, limit_bytes: usize) -> io::Result<Self> {
        let mut engine = Engine::new(data_dir)?;
        engine.set_query_memory_limit(limit_bytes);
        Ok(engine)
    }

    /// Open or create an archive-aware engine with an explicit per-query memory
    /// limit.
    pub fn with_memory_limit_and_wal_archive<F>(
        data_dir: &Path,
        limit_bytes: usize,
        archive: F,
    ) -> io::Result<Self>
    where
        F: Fn(&Path, &[powdb_storage::wal::WalRecord]) -> io::Result<()> + Send + Sync + 'static,
    {
        let mut engine = Engine::new_with_wal_archive(data_dir, archive)?;
        engine.set_query_memory_limit(limit_bytes);
        Ok(engine)
    }

    /// Current per-query memory limit in bytes.
    pub fn query_memory_limit(&self) -> usize {
        self.query_memory_limit
    }

    /// Override the per-query memory limit in bytes (builder-style).
    pub fn set_query_memory_limit(&mut self, limit_bytes: usize) {
        self.query_memory_limit = limit_bytes;
    }

    /// Current fallback nested-loop join candidate-pair cap.
    pub fn nested_loop_pair_limit(&self) -> usize {
        self.nested_loop_pair_limit
    }

    /// Override the fallback nested-loop join candidate-pair cap. Used by the
    /// server to apply `POWDB_MAX_NESTED_LOOP_PAIRS`, and by tests that need a
    /// tiny cap to exercise the guard on a small join. A zero limit is clamped
    /// to 1 so a valid single-pair join is never rejected outright.
    pub fn set_nested_loop_pair_limit(&mut self, limit: usize) {
        self.nested_loop_pair_limit = limit.max(1);
    }

    /// Set the WAL durability mode (see [`WalSyncMode`]). `Full` (the default)
    /// fsyncs every commit; `Normal` moves the fsync to a background flusher
    /// with a bounded crash-loss window; `Off` is bench-only (no durability).
    /// Wired from the server's `POWDB_SYNC_MODE` / `--sync-mode` config.
    pub fn set_wal_sync_mode(&mut self, mode: WalSyncMode) {
        self.catalog.set_wal_sync_mode(mode);
    }

    /// Run `f` with commit durability deferred — the WAL group-commit entry
    /// point for callers that serialize writers behind an exclusive lock.
    ///
    /// Inside `f`, Full-mode commit points register the WAL generation they
    /// need durable instead of fsyncing inline. The returned ticket (if any)
    /// must be waited on before the statement's result is acknowledged; the
    /// caller should release its exclusive engine lock first, so other
    /// committers can append while the fsync runs. That overlap is what lets
    /// one fsync cover many commits. A lone committer's wait performs the
    /// fsync immediately — group commit never introduces a delay.
    ///
    /// `Normal`/`Off` sync modes return no ticket; their durability
    /// contracts are unchanged. If `f` panics the engine must not be reused
    /// (the deferral flag may still be set); lock poisoning enforces this
    /// for callers that share the engine behind a lock.
    pub fn run_with_deferred_durability<T>(
        &mut self,
        f: impl FnOnce(&mut Engine) -> T,
    ) -> (T, Option<WalDurabilityTicket>) {
        self.catalog.set_wal_sync_deferred(true);
        let out = f(self);
        self.catalog.set_wal_sync_deferred(false);
        let ticket = self.catalog.take_wal_durability_ticket();
        (out, ticket)
    }

    /// Number of fsyncs issued against the WAL (test/metrics hook).
    pub fn wal_fsync_count(&self) -> u64 {
        self.catalog.wal_fsync_count()
    }

    /// Roll back the active explicit transaction while archiving any committed
    /// pre-transaction WAL records that recovery must replay and truncate.
    /// This is the sync-aware counterpart to the ordinary `rollback` statement;
    /// callers provide the archive hook so the query crate stays independent of
    /// replication metadata.
    pub fn rollback_transaction_with_wal_archive<F>(
        &mut self,
        archive: F,
    ) -> Result<QueryResult, QueryError>
    where
        F: FnMut(&Path, &[powdb_storage::wal::WalRecord]) -> io::Result<()>,
    {
        if !self.in_transaction {
            return Err(QueryError::Execution(
                "no active transaction to roll back".into(),
            ));
        }
        self.catalog
            .rollback_to_last_sync_with_wal_archive(archive)
            .map_err(QueryError::from_storage_io)?;
        self.finish_rollback_after_catalog_restore()
    }

    pub fn rollback_transaction_preserving_wal_archive(
        &mut self,
    ) -> Result<QueryResult, QueryError> {
        let Some(hook) = self.wal_archive_hook.clone() else {
            if !self.in_transaction {
                return Err(QueryError::Execution(
                    "no active transaction to roll back".into(),
                ));
            }
            self.catalog
                .rollback_to_last_sync()
                .map_err(QueryError::from_storage_io)?;
            return self.finish_rollback_after_catalog_restore();
        };
        self.rollback_transaction_with_wal_archive(move |dir, records| hook(dir, records))
    }

    fn finish_rollback_after_catalog_restore(&mut self) -> Result<QueryResult, QueryError> {
        self.in_transaction = false;
        if let Ok(mut cache) = self.plan_cache.lock() {
            cache.clear();
        }
        self.view_registry =
            open_view_registry(self.catalog.data_dir()).map_err(QueryError::from_storage_io)?;
        Ok(QueryResult::Executed {
            message: "transaction rolled back".to_string(),
        })
    }

    /// Enter a budgeted-statement frame for the current query. The returned
    /// guard must be held for the duration of the statement; on its drop the
    /// reentrancy depth is decremented. Only the *outermost* statement entry
    /// zeroes this thread's running total, so a nested `execute_powql` (the
    /// source query of a `create_view`/`refresh_view`) does NOT discard the
    /// outer frame's accounting. The accumulator is thread-local, so this never
    /// touches another concurrent query's total.
    #[must_use = "the budget guard must outlive the statement body"]
    pub(super) fn enter_memory_budget(&self) -> mem_budget::EnterGuard {
        mem_budget::enter()
    }

    /// Charge the estimated footprint of a freshly materialized batch of rows
    /// against the current per-query budget. Returns
    /// [`QueryError::MemoryLimitExceeded`] cleanly if the batch would push the
    /// query over its limit. Used at every full-materialization point (sort
    /// buffer, join build side, GROUP BY hash table, IN-list).
    pub(super) fn charge_rows(&self, rows: &[Vec<Value>]) -> Result<(), QueryError> {
        let mut total = 0usize;
        let mut cancel = crate::cancel::CancelCheck::new();
        for row in rows {
            cancel.tick()?;
            total = total.saturating_add(mem_budget::estimate_row_size(row));
        }
        mem_budget::charge(total, self.query_memory_limit)
    }

    /// Charge a materialized IN-list (the literal expressions pulled out of an
    /// uncorrelated `IN (subquery)`) against the current per-query budget.
    /// Each item is conservatively sized at the `Expr` slot plus, for string
    /// literals, the owned heap bytes.
    pub(super) fn charge_in_list(&self, list: &[crate::ast::Expr]) -> Result<(), QueryError> {
        let base = std::mem::size_of::<crate::ast::Expr>();
        let mut total = std::mem::size_of::<Vec<crate::ast::Expr>>();
        let mut cancel = crate::cancel::CancelCheck::new();
        for item in list {
            cancel.tick()?;
            total = total.saturating_add(base);
            if let crate::ast::Expr::Literal(crate::ast::Literal::String(s)) = item {
                total = total.saturating_add(s.capacity());
            }
        }
        mem_budget::charge(total, self.query_memory_limit)
    }

    /// Dispatch to the requested query frontend.
    pub fn execute_with_dialect(
        &mut self,
        dialect: QueryDialect,
        input: &str,
    ) -> Result<QueryResult, QueryError> {
        match dialect {
            QueryDialect::PowQL => self.execute_powql(input),
            QueryDialect::Sql => self.execute_sql(input),
        }
    }

    /// Read-only variant of [`Engine::execute_with_dialect`].
    pub fn execute_readonly_with_dialect(
        &self,
        dialect: QueryDialect,
        input: &str,
    ) -> Result<QueryResult, QueryError> {
        match dialect {
            QueryDialect::PowQL => self.execute_powql_readonly(input),
            QueryDialect::Sql => self.execute_sql_readonly(input),
        }
    }

    /// The statement boundary. An autocommit statement is durable before it
    /// returns; inside an explicit transaction the boundary belongs to
    /// `commit`, so nothing is flushed here. Flushing per statement inside a
    /// transaction buys no durability (an unfinished transaction is rolled
    /// back on replay) and costs one fsync per row.
    fn commit_statement(&mut self) -> Result<(), QueryError> {
        if self.in_transaction {
            return Ok(());
        }
        self.catalog
            .commit_autocommit()
            .map_err(QueryError::from_storage_io)
    }

    /// Parse + plan + execute a PowQL query.
    ///
    /// # Examples
    ///
    /// ```
    /// use powdb_query::executor::Engine;
    /// use powdb_query::result::QueryResult;
    ///
    /// let dir = tempfile::tempdir().unwrap();
    /// let mut engine = Engine::new(dir.path()).unwrap();
    ///
    /// // Create a table and insert a row.
    /// engine.execute_powql("type User { required name: str, age: int }").unwrap();
    /// engine.execute_powql(r#"insert User { name := "Alice", age := 30 }"#).unwrap();
    ///
    /// // Query rows back.
    /// let result = engine.execute_powql("User").unwrap();
    /// assert_eq!(result.row_count(), 1);
    /// ```
    ///
    /// Mission D6 — tracing collapse: the previous implementation ran 4
    /// `Instant::now()` + 3 `elapsed().as_micros()` calls + formatted an
    /// `info!` span on every query, even when tracing was disabled. On a
    /// sub-microsecond `point_lookup_indexed` call that overhead was
    /// 100-200ns — 20%+ of the whole query. We now measure time only when
    /// INFO is actually enabled via `tracing::enabled!`, and we moved the
    /// noisy `debug!(?plan)` line behind the same gate so the Debug
    /// formatter can't run unconditionally either.
    ///
    /// Mission D9 — plan cache: on the hot path we canonicalise the query
    /// text (lex + FNV-1a hash with literal values stripped), check the
    /// cache, and on a hit substitute the new literals into a clone of the
    /// cached plan. This skips re-lexing, re-parsing, and re-planning —
    /// around 3μs per call on bench workloads. On a miss we plan as before
    /// and insert the plan under its canonical hash.
    pub fn execute_powql(&mut self, input: &str) -> Result<QueryResult, QueryError> {
        if self.read_only {
            // Snapshot-serving mode: run reads through the read-only executor and
            // turn the "this statement writes" sentinel into the terminal
            // ReadonlyMode error. No mutation ever reaches disk.
            return to_readonly_terminal(self.execute_powql_readonly(input));
        }
        // WS2: each *outermost* statement starts with the full memory
        // allowance. The guard holds the reentrancy depth so a nested
        // `execute_powql` (e.g. a view's source query) does not reset the
        // outer frame's accounting mid-statement.
        let _budget = self.enter_memory_budget();
        // A token may be cancelled before execution starts (for example, EOF
        // detected while this job was waiting for the engine lock). Check once
        // at the statement boundary so even point operations with no long loop
        // honor cancellation before they can mutate state.
        crate::cancel::check()?;
        // Hot path: tracing disabled. Zero syscalls, zero formatting.
        if !tracing::enabled!(Level::INFO) {
            // D9: try the plan cache first. Canonicalisation lexes the
            // query once; on a hit we skip the parser and planner entirely.
            if let Ok((hash, canonical, literals)) = canonicalize_with_form(input) {
                let cached = self
                    .plan_cache
                    .lock()
                    .map_err(|e| QueryError::Execution(format!("plan cache lock poisoned: {e}")))?
                    .get_with_substitution(hash, &canonical, &literals);
                if let Some(plan) = cached {
                    let plan = self.lower(&plan)?;
                    let result = self.execute_lowered(&plan);
                    // Mission B (post-review): statement-boundary WAL
                    // group commit. Catalog::wal_log now only appends;
                    // the fsync happens here exactly once per statement.
                    // `sync_wal` is a no-op when nothing was buffered
                    // (pure reads pay zero fsync).
                    self.commit_statement()?;
                    return result;
                }
                // Miss — plan, insert, execute.
                let (raw, plan) = self.plan_text_and_lower(input)?;
                self.plan_cache
                    .lock()
                    .map_err(|e| QueryError::Execution(format!("plan cache lock poisoned: {e}")))?
                    .insert(hash, canonical, raw, literals.len());
                let result = self.execute_lowered(&plan);
                self.commit_statement()?;
                return result;
            }
            // Lex error — fall through to the planner so the caller gets a
            // consistent error shape.
            let (_, plan) = self.plan_text_and_lower(input)?;
            let result = self.execute_lowered(&plan);
            self.commit_statement()?;
            return result;
        }

        // Instrumented path — only taken under explicit tracing subscribers.
        let total_start = Instant::now();
        let plan_start = Instant::now();
        // `plan_us` now covers planning AND lowering. They are one step from
        // the caller's point of view (lowering is not optional, an unlowered
        // plan answers differently), and keeping them one call is what keeps
        // the planner reachable from a single place.
        let (_, plan) = self.plan_text_and_lower(input).inspect_err(|err| {
            error!(query = %input, error = %err, "query plan failed");
        })?;
        let plan_us = plan_start.elapsed().as_micros();

        let exec_start = Instant::now();
        let result = self.execute_lowered(&plan);
        self.commit_statement()?;
        let exec_us = exec_start.elapsed().as_micros();

        let total_us = total_start.elapsed().as_micros();
        match &result {
            Ok(r) => {
                info!(
                    query = %input,
                    plan_us = plan_us,
                    exec_us = exec_us,
                    total_us = total_us,
                    rows = r.row_count(),
                    "query ok"
                );
            }
            Err(e) => {
                error!(
                    query = %input,
                    plan_us = plan_us,
                    exec_us = exec_us,
                    error = %e,
                    "query failed"
                );
            }
        }
        result
    }

    /// Parse + plan + execute a SQL query through the SQL frontend.
    ///
    /// SQL is lowered to the existing PowDB AST and to canonical PowQL text.
    /// The canonical PowQL text is used as the plan-cache key, so equivalent
    /// SQL and PowQL spellings share cached plans.
    pub fn execute_sql(&mut self, input: &str) -> Result<QueryResult, QueryError> {
        if self.read_only {
            return to_readonly_terminal(self.execute_sql_readonly(input));
        }
        let _budget = self.enter_memory_budget();
        crate::cancel::check()?;
        let parsed = crate::sql::parse_sql_with_canonical(input)
            .map_err(|e| QueryError::Parse(e.to_string()))?;

        if !tracing::enabled!(Level::INFO) {
            if let Ok((hash, canonical, literals)) = canonicalize_with_form(&parsed.canonical_powql)
            {
                let hash = if crate::sql::statement_has_aggregate(&parsed.statement) {
                    sql_raw_cache_hash(hash)
                } else {
                    hash
                };
                let cached = self
                    .plan_cache
                    .lock()
                    .map_err(|e| QueryError::Execution(format!("plan cache lock poisoned: {e}")))?
                    .get_with_substitution(hash, &canonical, &literals);
                if let Some(plan) = cached {
                    let plan = self.lower(&plan)?;
                    let result = self.execute_lowered(&plan);
                    self.commit_statement()?;
                    return result;
                }

                let (raw, plan) = self.plan_and_lower_cacheable(parsed.statement)?;
                self.plan_cache
                    .lock()
                    .map_err(|e| QueryError::Execution(format!("plan cache lock poisoned: {e}")))?
                    .insert(hash, canonical, raw, literals.len());
                let result = self.execute_lowered(&plan);
                self.commit_statement()?;
                return result;
            }
        }

        let plan = self.plan_and_lower(parsed.statement)?;
        let result = self.execute_lowered(&plan);
        self.commit_statement()?;
        result
    }

    /// Read-only variant of [`Engine::execute_sql`].
    pub fn execute_sql_readonly(&self, input: &str) -> Result<QueryResult, QueryError> {
        let _budget = self.enter_memory_budget();
        crate::cancel::check()?;
        let parsed = crate::sql::parse_sql_with_canonical(input)
            .map_err(|e| QueryError::Parse(e.to_string()))?;
        if !is_read_only_statement(&parsed.statement) {
            return Err(QueryError::ReadonlyNeedsWrite);
        }

        if let Ok((hash, canonical, literals)) = canonicalize_with_form(&parsed.canonical_powql) {
            let hash = if crate::sql::statement_has_aggregate(&parsed.statement) {
                sql_raw_cache_hash(hash)
            } else {
                hash
            };
            let cached = self
                .plan_cache
                .lock()
                .map_err(|e| QueryError::Execution(format!("plan cache lock poisoned: {e}")))?
                .get_with_substitution(hash, &canonical, &literals);
            if let Some(plan) = cached {
                let plan = self.lower(&plan)?;
                return self.execute_plan_readonly(&plan);
            }
            let (raw, plan) = self.plan_and_lower_cacheable(parsed.statement)?;
            self.plan_cache
                .lock()
                .map_err(|e| QueryError::Execution(format!("plan cache lock poisoned: {e}")))?
                .insert(hash, canonical, raw, literals.len());
            return self.execute_plan_readonly(&plan);
        }

        let plan = self.plan_and_lower(parsed.statement)?;
        self.execute_plan_readonly(&plan)
    }

    /// Execute PowQL with `$N` placeholders bound to positional `params`.
    ///
    /// Task 4: parameters are substituted as literal *tokens* before
    /// parsing (see [`crate::parser::parse_with_params`]), so untrusted
    /// input can never change the query's shape. This path deliberately
    /// **bypasses the plan cache** — template caching is a follow-up — and
    /// otherwise mirrors the non-cached tail of [`Engine::execute_powql`].
    pub fn execute_powql_with_params(
        &mut self,
        input: &str,
        params: &[crate::ast::ParamValue],
    ) -> Result<QueryResult, QueryError> {
        if self.read_only {
            return to_readonly_terminal(self.execute_powql_readonly_with_params(input, params));
        }
        let _budget = self.enter_memory_budget();
        crate::cancel::check()?;
        let stmt = crate::parser::parse_with_params(input, params)
            .map_err(|e| QueryError::Parse(e.to_string()))?;
        let plan = self.plan_and_lower(stmt)?;
        let result = self.execute_lowered(&plan);
        self.commit_statement()?;
        result
    }

    /// Read-only variant of [`Engine::execute_powql_with_params`].
    ///
    /// Mirrors [`Engine::execute_powql_readonly`]: parses with bound
    /// params, rejects any write statement with
    /// [`QueryError::ReadonlyNeedsWrite`] so the caller can escalate to the
    /// write lock, then executes under a shared borrow. No plan-cache
    /// interaction.
    pub fn execute_powql_readonly_with_params(
        &self,
        input: &str,
        params: &[crate::ast::ParamValue],
    ) -> Result<QueryResult, QueryError> {
        let _budget = self.enter_memory_budget();
        crate::cancel::check()?;
        let stmt = crate::parser::parse_with_params(input, params)
            .map_err(|e| QueryError::Parse(e.to_string()))?;
        if !is_read_only_statement(&stmt) {
            return Err(QueryError::ReadonlyNeedsWrite);
        }
        let plan = self.plan_and_lower(stmt)?;
        self.execute_plan_readonly(&plan)
    }

    /// Cancellation-aware variant of [`Engine::execute_powql`]. Installs
    /// `cancel` as the current thread's cancellation token for the duration of
    /// the statement, so cancellable read and mutation-target discovery loops
    /// poll it. Mutation application checks once before its first write, then
    /// finishes without polling because the engine has no statement savepoint
    /// with which to undo a written prefix. The base methods also honor an
    /// already installed token; a caller with no token (embedded/direct use)
    /// never cancels.
    pub fn execute_powql_with_cancel(
        &mut self,
        input: &str,
        cancel: Arc<crate::cancel::ExecCancel>,
    ) -> Result<QueryResult, QueryError> {
        let _cancel_guard = crate::cancel::install(cancel);
        self.execute_powql(input)
    }

    /// Cancellation-aware variant of [`Engine::execute_sql`].
    pub fn execute_sql_with_cancel(
        &mut self,
        input: &str,
        cancel: Arc<crate::cancel::ExecCancel>,
    ) -> Result<QueryResult, QueryError> {
        let _cancel_guard = crate::cancel::install(cancel);
        self.execute_sql(input)
    }

    /// Cancellation-aware variant of [`Engine::execute_powql_readonly`].
    pub fn execute_powql_readonly_with_cancel(
        &self,
        input: &str,
        cancel: Arc<crate::cancel::ExecCancel>,
    ) -> Result<QueryResult, QueryError> {
        let _cancel_guard = crate::cancel::install(cancel);
        self.execute_powql_readonly(input)
    }

    /// Cancellation-aware variant of [`Engine::execute_sql_readonly`].
    pub fn execute_sql_readonly_with_cancel(
        &self,
        input: &str,
        cancel: Arc<crate::cancel::ExecCancel>,
    ) -> Result<QueryResult, QueryError> {
        let _cancel_guard = crate::cancel::install(cancel);
        self.execute_sql_readonly(input)
    }

    /// Cancellation-aware variant of [`Engine::execute_powql_with_params`].
    pub fn execute_powql_with_params_and_cancel(
        &mut self,
        input: &str,
        params: &[crate::ast::ParamValue],
        cancel: Arc<crate::cancel::ExecCancel>,
    ) -> Result<QueryResult, QueryError> {
        let _cancel_guard = crate::cancel::install(cancel);
        self.execute_powql_with_params(input, params)
    }

    /// Cancellation-aware variant of [`Engine::execute_powql_readonly_with_params`].
    pub fn execute_powql_readonly_with_params_and_cancel(
        &self,
        input: &str,
        params: &[crate::ast::ParamValue],
        cancel: Arc<crate::cancel::ExecCancel>,
    ) -> Result<QueryResult, QueryError> {
        let _cancel_guard = crate::cancel::install(cancel);
        self.execute_powql_readonly_with_params(input, params)
    }

    /// Plan cache stats — useful for benches and debugging.
    pub fn plan_cache_stats(&self) -> (u64, u64, usize) {
        let cache = self.plan_cache.lock().unwrap_or_else(|e| e.into_inner());
        (cache.hits, cache.misses, cache.len())
    }

    /// Mission infra-1: read-only entry point.
    ///
    /// Parses + plans + executes a PowQL query using only a shared borrow
    /// on the engine. Rejects any statement that would mutate state
    /// (Insert/Update/Delete/CreateTable/AlterTable/DropTable/CreateView/
    /// RefreshView/DropView) by returning [`READONLY_NEEDS_WRITE`] so the
    /// caller can escalate to the write lock.
    ///
    /// Also returns [`READONLY_NEEDS_WRITE`] if a materialized view in the
    /// query is dirty — refreshing one requires `&mut self`, so the caller
    /// must retake the write lock for the first refresh.
    ///
    /// This method is the concurrent-read fast path behind
    /// `Arc<RwLock<Engine>>`: multiple threads can call it simultaneously
    /// under a shared `.read()` lock and each will scan independently.
    pub fn execute_powql_readonly(&self, input: &str) -> Result<QueryResult, QueryError> {
        // WS2: each *outermost* statement starts with the full memory
        // allowance. The guard holds the reentrancy depth so a nested
        // `execute_powql*` does not reset the outer frame's accounting.
        let _budget = self.enter_memory_budget();
        crate::cancel::check()?;
        // Parse the statement first so we can classify read vs. write
        // without touching the catalog. This is the same lex+parse cost
        // the hot path would pay anyway.
        let stmt = crate::parser::parse(input).map_err(|e| QueryError::Parse(e.to_string()))?;
        if !is_read_only_statement(&stmt) {
            return Err(QueryError::ReadonlyNeedsWrite);
        }

        // Try the plan cache first — identical hash scheme to
        // `execute_powql` so both paths share cache state. The mutex
        // section is just a hashmap lookup + plan clone.
        if let Ok((hash, canonical, literals)) = canonicalize_with_form(input) {
            let cached = self
                .plan_cache
                .lock()
                .map_err(|e| QueryError::Execution(format!("plan cache lock poisoned: {e}")))?
                .get_with_substitution(hash, &canonical, &literals);
            if let Some(plan) = cached {
                let plan = self.lower(&plan)?;
                return self.execute_plan_readonly(&plan);
            }
            // Miss: plan + insert + execute. The planner is pure, so this
            // is safe from `&self`.
            let (raw, plan) = self.plan_and_lower_cacheable(stmt)?;
            self.plan_cache
                .lock()
                .map_err(|e| QueryError::Execution(format!("plan cache lock poisoned: {e}")))?
                .insert(hash, canonical, raw, literals.len());
            return self.execute_plan_readonly(&plan);
        }
        // Lex error — fall through to the planner for a consistent error
        // shape (though `parse` above would usually have caught it).
        let plan = self.plan_and_lower(stmt)?;
        self.execute_plan_readonly(&plan)
    }

    /// Read-only version of [`Engine::execute_plan`]. Dispatches the
    /// read-path plan variants by calling `&self` helpers and errors with
    /// [`READONLY_NEEDS_WRITE`] on any write variant. This is the
    /// recursion target for composite read plans under the RwLock reader.
    ///
    /// The dispatch mirrors `execute_plan` for the read branches but does
    /// not carry any of the fast-paths that need `&mut self` (e.g. plan-
    /// cache mutation on inner subqueries is handled via the shared mutex
    /// in [`Engine::execute_powql_readonly`]; in-flight subquery
    /// materialisation uses [`Engine::materialize_subqueries_readonly`]).
    fn execute_plan_readonly(&self, plan: &LoweredPlan) -> Result<QueryResult, QueryError> {
        Self::raise_arith_fault(self.dispatch_readonly(plan.node()))
    }

    /// The read-path dispatch itself. Takes a bare `&PlanNode` because it is
    /// the recursion target: every child of a lowered plan is lowered, so a
    /// subtree needs no second wrapper. Reaching it from outside an already
    /// lowered tree is what [`Engine::execute_plan_readonly`] exists to
    /// prevent, which is why this one is private and unlowered plans cannot
    /// name it.
    fn dispatch_readonly(&self, plan: &PlanNode) -> Result<QueryResult, QueryError> {
        // Detect every dirty materialized-view source before executing any
        // branch of the plan. Without this preflight, a join could fully scan
        // its clean left input before discovering a dirty right input, then
        // repeat that work after the server upgrades to writer admission.
        // Alias scans also need this centralized check: they do not pass
        // through the SeqScan arm below.
        if plan_reads_dirty_view(plan, &self.view_registry) {
            return Err(QueryError::ReadonlyNeedsWrite);
        }
        // Mirror the mutable path: reject a stray aggregate FunctionCall, an
        // unknown table or column, a mistyped comparison or JSON path base and
        // a negative slice count before evaluating any row (see `dispatch_mut`
        // for the rationale).
        validate_plan(&self.catalog, plan)?;
        match plan {
            PlanNode::ExprIndexScan { .. }
            | PlanNode::ExprRangeScan { .. }
            | PlanNode::OrderedExprIndexScan { .. } => {
                if let Some(result) = self.execute_expression_index_plan(plan, None)? {
                    return Ok(result);
                }
                let fallback = self.lower(plan)?;
                self.execute_plan_readonly(&fallback)
            }
            PlanNode::SeqScan { table } => {
                // Dirty view means we'd need to refresh it — can't do that
                // under `&self`. Escalate to the write path.
                if self.view_registry.is_dirty(table) {
                    return Err(QueryError::ReadonlyNeedsWrite);
                }
                let schema = self
                    .catalog
                    .schema(table)
                    .ok_or_else(|| QueryError::TableNotFound(table.clone()))?
                    .clone();
                let columns: Vec<String> = schema.columns.iter().map(|c| c.name.clone()).collect();
                // Cooperative cancellation: a full-table scan of a huge table
                // must stay stoppable.
                let mut cancel = crate::cancel::CancelCheck::new();
                let mut rows: Vec<Vec<Value>> = Vec::new();
                for item in self.catalog.scan(table).map_err(|e| e.to_string())? {
                    let (_, row) = item.map_err(crate::result::QueryError::from_storage_io)?;
                    cancel.tick()?;
                    rows.push(row);
                }
                Ok(QueryResult::Rows { columns, rows })
            }

            PlanNode::AliasScan { table, alias } => {
                let schema = self
                    .catalog
                    .schema(table)
                    .ok_or_else(|| QueryError::TableNotFound(table.clone()))?
                    .clone();
                let columns: Vec<String> = schema
                    .columns
                    .iter()
                    .map(|c| format!("{alias}.{}", c.name))
                    .collect();
                let mut cancel = crate::cancel::CancelCheck::new();
                let mut rows: Vec<Vec<Value>> = Vec::new();
                for item in self.catalog.scan(table).map_err(|e| e.to_string())? {
                    let (_, row) = item.map_err(crate::result::QueryError::from_storage_io)?;
                    cancel.tick()?;
                    rows.push(row);
                }
                Ok(QueryResult::Rows { columns, rows })
            }

            PlanNode::NestedProject { input, fields } => {
                // Dirty child views were escalated by the preflight above;
                // the assembly itself only reads.
                // Resolve link traversals against the persistent catalog before
                // assembly (like the mutable dispatch), so child tables and
                // scalar hop chains are concrete.
                let resolved;
                let fields: &[crate::plan::NestedProjectField] =
                    if plan_exec::nested_fields_have_via_link(fields) {
                        let outer = plan_exec::scan_source_table(input).ok_or_else(|| {
                            QueryError::Execution(
                                "link traversal requires a plain aliased table scan as its parent"
                                    .into(),
                            )
                        })?;
                        resolved = self.resolve_nested_via_links(fields, outer)?;
                        &resolved
                    } else {
                        fields
                    };
                let parent = self.dispatch_readonly(input)?;
                self.execute_nested_project(parent, fields)
            }

            PlanNode::IndexScan { table, column, key } => {
                let schema = self
                    .catalog
                    .schema(table)
                    .ok_or_else(|| QueryError::TableNotFound(table.clone()))?
                    .clone();
                let columns: Vec<String> = schema.columns.iter().map(|c| c.name.clone()).collect();
                let key_value = literal_to_value(key)?;
                let tbl = self
                    .catalog
                    .get_table(table)
                    .ok_or_else(|| QueryError::TableNotFound(table.clone()))?;

                if tbl.has_index(column) {
                    // Use index_lookup_all to handle both unique and
                    // non-unique indexes — returns all matching RowIds.
                    let rids = tbl.index_lookup_all(column, &key_value);
                    let mut rows: Vec<Vec<Value>> = Vec::with_capacity(rids.len());
                    let mut cancel = crate::cancel::CancelCheck::new();
                    for rid in rids {
                        cancel.tick()?;
                        // Overflow safety (P0-3/P0-4): `tbl.get` reassembles
                        // spilled columns (the old `heap.get` + `decode_row`
                        // returned Empty / wrapped a >= 64KB value).
                        if let Some(row) = tbl.get(rid).map_err(QueryError::from_storage_io)? {
                            rows.push(row);
                        }
                    }
                    return Ok(QueryResult::Rows { columns, rows });
                }

                // No index: synthetic eq predicate + compiled scan.
                // Overflow safety (P0-4/P1): v2-capable tables use the decoded
                // last-resort scan below (raw scan drops/mis-reads spilled cols).
                let fast = FastLayout::new(&schema);
                let synth_pred = Expr::BinaryOp(
                    Box::new(Expr::Field(column.clone())),
                    BinOp::Eq,
                    Box::new(key.clone()),
                );
                if !tbl.has_overflow_rows() {
                    if let Some(compiled) = self.compile_predicate_unless_forced(
                        "readonly:index-scan-scan-fallback:predicate",
                        &synth_pred,
                        &columns,
                        &fast,
                        &schema,
                    ) {
                        let mut rows: Vec<Vec<Value>> = Vec::with_capacity(64);
                        for_each_row_raw_cancellable(&self.catalog, table, |_rid, data| {
                            if compiled(data) {
                                rows.push(decode_row(&schema, data));
                            }
                        })?;
                        return Ok(QueryResult::Rows { columns, rows });
                    }
                }

                // Last resort: slow eq-check.
                let col_idx =
                    schema
                        .column_index(column)
                        .ok_or_else(|| QueryError::ColumnNotFound {
                            table: String::new(),
                            column: column.clone(),
                        })?;
                let mut cancel = crate::cancel::CancelCheck::new();
                let mut rows: Vec<Vec<Value>> = Vec::new();
                for item in tbl.scan() {
                    let (_, row) = item.map_err(crate::result::QueryError::from_storage_io)?;
                    cancel.tick()?;
                    if row[col_idx] == key_value {
                        rows.push(row);
                    }
                }
                Ok(QueryResult::Rows { columns, rows })
            }

            PlanNode::RangeScan {
                table,
                column,
                start,
                end,
            } => {
                let tbl = self
                    .catalog
                    .get_table(table)
                    .ok_or_else(|| QueryError::TableNotFound(table.clone()))?;
                let columns: Vec<String> = tbl
                    .schema()
                    .columns
                    .iter()
                    .map(|c| c.name.clone())
                    .collect();
                let schema = tbl.schema().clone();

                let start_val = match start {
                    Some((expr, _)) => Some(literal_to_value(expr)?),
                    None => None,
                };
                let end_val = match end {
                    Some((expr, _)) => Some(literal_to_value(expr)?),
                    None => None,
                };
                let start_inclusive = start.as_ref().map(|(_, inc)| *inc).unwrap_or(true);
                let end_inclusive = end.as_ref().map(|(_, inc)| *inc).unwrap_or(true);

                // Range scans only use the btree fast path for unique indexes.
                // Non-unique indexes store composite keys that don't compare
                // directly against raw column values.
                if tbl.is_index_unique(column) == Some(true) {
                    if let Some(btree) = tbl.index(column) {
                        let hits: Vec<(Value, RowId)> = match (&start_val, &end_val) {
                            (Some(s), Some(e)) => btree.range(s, e).collect(),
                            (Some(s), None) => btree.range_from(s),
                            (None, Some(e)) => btree.range_to(e),
                            (None, None) => {
                                // Unbounded both sides — equivalent to seq scan.
                                let mut cancel = crate::cancel::CancelCheck::new();
                                let mut rows: Vec<Vec<Value>> = Vec::new();
                                for item in tbl.scan() {
                                    let (_, row) =
                                        item.map_err(crate::result::QueryError::from_storage_io)?;
                                    cancel.tick()?;
                                    rows.push(row);
                                }
                                return Ok(QueryResult::Rows { columns, rows });
                            }
                        };
                        let mut rows: Vec<Vec<Value>> = Vec::with_capacity(hits.len());
                        let mut cancel = crate::cancel::CancelCheck::new();
                        for (key, rid) in hits {
                            cancel.tick()?;
                            // Filter for exclusive bounds.
                            if !start_inclusive {
                                if let Some(ref s) = start_val {
                                    if &key == s {
                                        continue;
                                    }
                                }
                            }
                            if !end_inclusive {
                                if let Some(ref e) = end_val {
                                    if &key == e {
                                        continue;
                                    }
                                }
                            }
                            // Overflow safety (P0-3): reassemble spilled cols.
                            if let Some(row) = tbl.get(rid).map_err(QueryError::from_storage_io)? {
                                rows.push(row);
                            }
                        }
                        return Ok(QueryResult::Rows { columns, rows });
                    }
                }

                // Fallback: no index — synthesize the range predicate and scan.
                // Overflow safety (P0-4): v2-capable tables use the decoded
                // last-resort scan below.
                let fast = FastLayout::new(&schema);
                let synth = synthesize_range_predicate(column, start, end);
                if !tbl.has_overflow_rows() {
                    if let Some(compiled) = self.compile_predicate_unless_forced(
                        "readonly:range-scan-scan-fallback:predicate",
                        &synth,
                        &columns,
                        &fast,
                        &schema,
                    ) {
                        let mut rows: Vec<Vec<Value>> = Vec::with_capacity(64);
                        for_each_row_raw_cancellable(&self.catalog, table, |_rid, data| {
                            if compiled(data) {
                                rows.push(decode_row(&schema, data));
                            }
                        })?;
                        return Ok(QueryResult::Rows { columns, rows });
                    }
                }

                // Last resort: decoded row eval.
                let col_idx =
                    schema
                        .column_index(column)
                        .ok_or_else(|| QueryError::ColumnNotFound {
                            table: String::new(),
                            column: column.clone(),
                        })?;
                let mut cancel = crate::cancel::CancelCheck::new();
                let mut rows: Vec<Vec<Value>> = Vec::new();
                for item in tbl.scan() {
                    let (_, row) = item.map_err(crate::result::QueryError::from_storage_io)?;
                    cancel.tick()?;
                    if range_matches(
                        &row[col_idx],
                        &start_val,
                        start_inclusive,
                        &end_val,
                        end_inclusive,
                    ) {
                        rows.push(row);
                    }
                }
                Ok(QueryResult::Rows { columns, rows })
            }

            PlanNode::Filter { input, predicate } => {
                // Materialise subqueries using the `&self` variant.
                // Uncorrelated subqueries are replaced with InList/Bool;
                // correlated ones are left as InSubquery/ExistsSubquery
                // for per-row materialisation below.
                let materialized;
                let predicate = if contains_subquery(predicate) {
                    materialized = self.materialize_subqueries_readonly(predicate)?;
                    &materialized
                } else {
                    predicate
                };

                // Correlated subquery path: per-row materialisation.
                if contains_subquery(predicate) {
                    let result = self.dispatch_readonly(input)?;
                    return match result {
                        QueryResult::Rows { columns, rows } => {
                            let mut filtered = Vec::new();
                            // Cooperative cancellation: this runs a subquery per
                            // outer row, so a large outer scan must stay stoppable.
                            let mut cancel = crate::cancel::CancelCheck::new();
                            for row in rows {
                                cancel.tick()?;
                                let row_pred = self.materialize_correlated_for_row_readonly(
                                    predicate, &row, &columns,
                                )?;
                                if eval_predicate(&row_pred, &row, &columns) {
                                    filtered.push(row);
                                }
                            }
                            Ok(QueryResult::Rows {
                                columns,
                                rows: filtered,
                            })
                        }
                        _ => Err("filter requires row input".into()),
                    };
                }

                // Lane A fast path: Filter over an equality-driven index scan
                // (mirrors the mutable path). Pure `&self`, so it is shared.
                if matches!(
                    input.as_ref(),
                    PlanNode::IndexScan { .. } | PlanNode::ExprIndexScan { .. }
                ) {
                    if let Some(result) = self.try_filter_index_residual_fast(input, predicate)? {
                        return Ok(result);
                    }
                }

                // Fused Filter+SeqScan fast path.
                // Overflow safety (P0-4/P1): v2-capable tables fall through to
                // the decoded general path below.
                if let PlanNode::SeqScan { table } = input.as_ref() {
                    if !self.catalog.table_has_overflow(table)
                        && !self.generic_path_forced("readonly:filter-seqscan-raw")
                    {
                        if self.view_registry.is_dirty(table) {
                            return Err(QueryError::ReadonlyNeedsWrite);
                        }
                        let schema = self
                            .catalog
                            .schema(table)
                            .ok_or_else(|| QueryError::TableNotFound(table.clone()))?
                            .clone();
                        let columns: Vec<String> =
                            schema.columns.iter().map(|c| c.name.clone()).collect();
                        let fast = FastLayout::new(&schema);
                        let row_layout = RowLayout::new(&schema);
                        let mut rows: Vec<Vec<Value>> = Vec::with_capacity(64);

                        // Cooperative cancellation: full-table compiled/selective
                        // predicate scan must stay stoppable (see the write-path
                        // Filter fast path for the same pattern).
                        let mut cancel = crate::cancel::CancelCheck::new();
                        let mut cancel_err: Option<QueryError> = None;
                        if let Some(compiled) = self.compile_predicate_unless_forced(
                            "readonly:filter-seqscan:predicate",
                            predicate,
                            &columns,
                            &fast,
                            &schema,
                        ) {
                            self.catalog
                                .try_for_each_row_raw(table, |_rid, data| {
                                    if let Err(e) = cancel.tick() {
                                        cancel_err = Some(e);
                                        return std::ops::ControlFlow::Break(());
                                    }
                                    if compiled(data) {
                                        rows.push(decode_row(&schema, data));
                                    }
                                    std::ops::ControlFlow::Continue(())
                                })
                                .map_err(QueryError::from_storage_io)?;
                        } else {
                            let pred_cols = predicate_column_indices_json(predicate, &columns);
                            self.catalog
                                .try_for_each_row_raw(table, |_rid, data| {
                                    if let Err(e) = cancel.tick() {
                                        cancel_err = Some(e);
                                        return std::ops::ControlFlow::Break(());
                                    }
                                    let pred_row =
                                        decode_selective(&schema, &row_layout, data, &pred_cols);
                                    if eval_predicate(predicate, &pred_row, &columns) {
                                        rows.push(decode_row(&schema, data));
                                    }
                                    std::ops::ControlFlow::Continue(())
                                })
                                .map_err(QueryError::from_storage_io)?;
                        }
                        if let Some(e) = cancel_err {
                            return Err(e);
                        }

                        return Ok(QueryResult::Rows { columns, rows });
                    }
                }

                // General path.
                let result = self.dispatch_readonly(input)?;
                match result {
                    QueryResult::Rows { columns, rows } => {
                        let mut cancel = crate::cancel::CancelCheck::new();
                        let mut filtered: Vec<Vec<Value>> = Vec::new();
                        for row in rows {
                            cancel.tick()?;
                            if eval_predicate(predicate, &row, &columns) {
                                filtered.push(row);
                            }
                        }
                        Ok(QueryResult::Rows {
                            columns,
                            rows: filtered,
                        })
                    }
                    _ => Err("filter requires row input".into()),
                }
            }

            PlanNode::Project { input, fields } => {
                if matches!(
                    input.as_ref(),
                    PlanNode::ExprIndexScan { .. }
                        | PlanNode::ExprRangeScan { .. }
                        | PlanNode::OrderedExprIndexScan { .. }
                ) {
                    if let Some(result) = self.execute_expression_index_plan(input, Some(fields))? {
                        return Ok(result);
                    }
                }
                // Fast path: Project over IndexScan. Avoids full-row decode
                // by calling decode_column only for projected fields.
                if let PlanNode::IndexScan { table, column, key } = input.as_ref() {
                    let key_value = literal_to_value(key)?;
                    let tbl = self
                        .catalog
                        .get_table(table)
                        .ok_or_else(|| QueryError::TableNotFound(table.clone()))?;
                    let schema = tbl.schema();

                    let proj_columns: Vec<String> = fields
                        .iter()
                        .map(|f| {
                            f.alias
                                .clone()
                                .unwrap_or_else(|| projection_output_name(&f.expr))
                        })
                        .collect();

                    let proj_indices: Vec<usize> = fields
                        .iter()
                        .filter_map(|f| {
                            if let Expr::Field(name) = &f.expr {
                                schema.column_index(name)
                            } else {
                                None
                            }
                        })
                        .collect();

                    // Plain-field projections only; a computed projection
                    // (e.g. `length(.v)`) falls through to the generic
                    // expression-evaluating path (its column is otherwise
                    // dropped — proj_indices only collects Fields).
                    let all_plain_fields = fields.iter().all(|f| matches!(f.expr, Expr::Field(_)));
                    if tbl.has_index(column)
                        && all_plain_fields
                        && !self.generic_path_forced("readonly:project-over-index-scan")
                    {
                        let rids = tbl.index_lookup_all(column, &key_value);
                        let mut rows: Vec<Vec<Value>> = Vec::with_capacity(rids.len());
                        let mut cancel = crate::cancel::CancelCheck::new();
                        for rid in rids {
                            cancel.tick()?;
                            // Overflow safety (P0-3/P0-4): reassemble via
                            // `tbl.get` so spilled projected columns return
                            // their value, not Empty / a wrapped >= 64KB blob.
                            if let Some(full) = tbl.get(rid).map_err(QueryError::from_storage_io)? {
                                let row: Vec<Value> =
                                    proj_indices.iter().map(|&ci| full[ci].clone()).collect();
                                rows.push(row);
                            }
                        }
                        return Ok(QueryResult::Rows {
                            columns: proj_columns,
                            rows,
                        });
                    }
                }

                // Fast paths over Limit(Sort(...)) / Limit(Filter(...)) / Limit(SeqScan).
                if let PlanNode::Limit {
                    input: inner,
                    count: limit_expr,
                } = input.as_ref()
                {
                    if let PlanNode::Sort {
                        input: sort_input,
                        keys,
                    } = inner.as_ref()
                    {
                        if keys.len() == 1 {
                            if let (Expr::Field(sort_field), Some(limit)) =
                                (&keys[0].expr, literal_limit(limit_expr))
                            {
                                let descending = keys[0].descending;
                                let (table_opt, pred_opt): (Option<&str>, Option<&Expr>) =
                                    match sort_input.as_ref() {
                                        PlanNode::SeqScan { table } => (Some(table.as_str()), None),
                                        PlanNode::Filter {
                                            input: fi,
                                            predicate,
                                        } => {
                                            if let PlanNode::SeqScan { table } = fi.as_ref() {
                                                (Some(table.as_str()), Some(predicate))
                                            } else {
                                                (None, None)
                                            }
                                        }
                                        _ => (None, None),
                                    };
                                if let Some(table) = table_opt {
                                    if let Some(result) = self.project_filter_sort_limit_fast(
                                        table, fields, sort_field, descending, limit, pred_opt,
                                    )? {
                                        return Ok(result);
                                    }
                                }
                            }
                        }
                    }
                    if let PlanNode::Filter {
                        input: fi,
                        predicate,
                    } = inner.as_ref()
                    {
                        if let (PlanNode::SeqScan { table }, Some(limit)) =
                            (fi.as_ref(), literal_limit(limit_expr))
                        {
                            if let Some(result) = self.project_filter_limit_fast(
                                table,
                                fields,
                                limit,
                                Some(predicate),
                            )? {
                                return Ok(result);
                            }
                        }
                    }
                    if let (PlanNode::SeqScan { table }, Some(limit)) =
                        (inner.as_ref(), literal_limit(limit_expr))
                    {
                        if let Some(result) =
                            self.project_filter_limit_fast(table, fields, limit, None)?
                        {
                            return Ok(result);
                        }
                    }
                }

                // Project(Filter(SeqScan)) without Limit.
                if let PlanNode::Filter {
                    input: fi,
                    predicate,
                } = input.as_ref()
                {
                    if let PlanNode::SeqScan { table } = fi.as_ref() {
                        if let Some(result) = self.project_filter_limit_fast(
                            table,
                            fields,
                            usize::MAX,
                            Some(predicate),
                        )? {
                            return Ok(result);
                        }
                    }
                }

                // Project(SeqScan) without Filter or Limit.
                if let PlanNode::SeqScan { table } = input.as_ref() {
                    if let Some(result) =
                        self.project_filter_limit_fast(table, fields, usize::MAX, None)?
                    {
                        return Ok(result);
                    }
                }

                // Generic path.
                let result = self.dispatch_readonly(input)?;
                match result {
                    QueryResult::Rows { columns, rows } => {
                        let proj_columns: Vec<String> = fields
                            .iter()
                            .map(|f| {
                                f.alias
                                    .clone()
                                    .unwrap_or_else(|| projection_output_name(&f.expr))
                            })
                            .collect();
                        let mut cancel = crate::cancel::CancelCheck::new();
                        let mut proj_rows: Vec<Vec<Value>> = Vec::with_capacity(rows.len());
                        for row in &rows {
                            cancel.tick()?;
                            proj_rows.push(
                                fields
                                    .iter()
                                    .map(|f| eval_expr(&f.expr, row, &columns))
                                    .collect(),
                            );
                        }
                        Ok(QueryResult::Rows {
                            columns: proj_columns,
                            rows: proj_rows,
                        })
                    }
                    _ => Err("project requires row input".into()),
                }
            }

            PlanNode::Sort { input, keys } => {
                let result = self.dispatch_readonly(input)?;
                match result {
                    QueryResult::Rows { columns, mut rows } => {
                        if rows.len() > MAX_SORT_ROWS {
                            return Err(QueryError::SortLimitExceeded);
                        }
                        // WS2: byte-budget guard on the sort buffer.
                        self.charge_rows(&rows)?;
                        let key_specs: Vec<(Option<usize>, &Expr, bool)> = keys
                            .iter()
                            .map(|k| {
                                let stored_name = match &k.expr {
                                    Expr::Field(name) => Some(name.clone()),
                                    Expr::QualifiedField { qualifier, field } => {
                                        Some(format!("{qualifier}.{field}"))
                                    }
                                    _ => None,
                                };
                                // Same resolver the read-write sort path and
                                // every other clause use, so the readonly path
                                // does not disagree with them about whether
                                // `order .amount` inside a join names a column.
                                let index = stored_name
                                    .as_ref()
                                    .and_then(|name| resolve_column_index(name, &columns));
                                if let Some(name) = stored_name {
                                    if index.is_none() {
                                        return Err(QueryError::ColumnNotFound {
                                            table: String::new(),
                                            column: name,
                                        });
                                    }
                                }
                                Ok((index, &k.expr, k.descending))
                            })
                            .collect::<Result<_, QueryError>>()?;
                        cooperative_stable_sort_by(&mut rows, self.query_memory_limit, |a, b| {
                            for &(col_idx, expr, descending) in &key_specs {
                                let (left_value, right_value) = match col_idx {
                                    Some(index) => (&a[index], &b[index]),
                                    None => {
                                        let left = eval_expr(expr, a, &columns);
                                        let right = eval_expr(expr, b, &columns);
                                        let cmp = compare_order_values(&left, &right, descending);
                                        if cmp != std::cmp::Ordering::Equal {
                                            return cmp;
                                        }
                                        continue;
                                    }
                                };
                                let cmp = compare_order_values(left_value, right_value, descending);
                                if cmp != std::cmp::Ordering::Equal {
                                    return cmp;
                                }
                            }
                            std::cmp::Ordering::Equal
                        })?;
                        Ok(QueryResult::Rows { columns, rows })
                    }
                    _ => Err("sort requires row input".into()),
                }
            }

            PlanNode::Limit { input, count } => {
                let result = self.dispatch_readonly(input)?;
                let n = match count {
                    Expr::Literal(Literal::Int(v)) => *v as usize,
                    _ => return Err("limit must be integer literal".into()),
                };
                match result {
                    QueryResult::Rows { columns, rows } => {
                        let mut cancel = crate::cancel::CancelCheck::new();
                        let mut limited = Vec::with_capacity(n.min(rows.len()));
                        for row in rows.into_iter().take(n) {
                            cancel.tick()?;
                            limited.push(row);
                        }
                        Ok(QueryResult::Rows {
                            columns,
                            rows: limited,
                        })
                    }
                    _ => Err("limit requires row input".into()),
                }
            }

            PlanNode::Offset { input, count } => {
                let result = self.dispatch_readonly(input)?;
                let n = match count {
                    Expr::Literal(Literal::Int(v)) => *v as usize,
                    _ => return Err("offset must be integer literal".into()),
                };
                match result {
                    QueryResult::Rows { columns, rows } => {
                        let mut cancel = crate::cancel::CancelCheck::new();
                        let mut offset = Vec::with_capacity(rows.len().saturating_sub(n));
                        for (index, row) in rows.into_iter().enumerate() {
                            cancel.tick()?;
                            if index >= n {
                                offset.push(row);
                            }
                        }
                        Ok(QueryResult::Rows {
                            columns,
                            rows: offset,
                        })
                    }
                    _ => Err("offset requires row input".into()),
                }
            }

            PlanNode::Aggregate {
                input,
                function,
                argument,
                mode: _,
                provenance_alias,
            } => {
                if let Some(provenance_alias) = provenance_alias {
                    let input = self.materialize_rows_with_provenance(input)?;
                    self.charge_rows(&input.rows)?;
                    return aggregate_rows_with_provenance(
                        *function,
                        argument.as_ref(),
                        &input,
                        provenance_alias,
                        self.query_memory_limit,
                    );
                }
                // Fast path: count() over SeqScan.
                // Overflow safety (P0-4): v2-capable tables use the decoded
                // generic path (raw count drops >= 64KB rows). A count with a
                // target column (`count(T { .v })`) counts non-null values, so
                // it must not take this row-counting path.
                if *function == AggFunc::Count
                    && counts_every_row(argument.as_ref())
                    && !self.generic_path_forced("readonly:count-fast-block")
                {
                    if let PlanNode::SeqScan { table } = input.as_ref() {
                        if !self.catalog.table_has_overflow(table) {
                            // A dirty materialized view must be refreshed before
                            // it can be counted, which needs `&mut self`. Escalate
                            // to the write path (F3: count(View) returned stale).
                            if self.view_registry.is_dirty(table) {
                                return Err(QueryError::ReadonlyNeedsWrite);
                            }
                            let mut count: i64 = 0;
                            for_each_row_raw_cancellable(&self.catalog, table, |_rid, _data| {
                                count += 1;
                            })?;
                            return Ok(QueryResult::Scalar(Value::Int(count)));
                        }
                    }
                    if let PlanNode::Filter {
                        input: inner,
                        predicate,
                    } = input.as_ref()
                    {
                        // Only take the fast path for a plain Filter(SeqScan)
                        // with no subquery in the predicate. A subquery
                        // predicate (`count(T filter .x in (...))`) must be
                        // resolved first; the fast path evaluates the raw
                        // predicate with no subquery materialisation, which
                        // silently yields 0 (F1). Falling through routes it to
                        // the generic path that runs the subquery correctly.
                        if let PlanNode::SeqScan { table } = inner.as_ref() {
                            if self.view_registry.is_dirty(table) {
                                // F3: count(View filter ...) over a dirty view.
                                return Err(QueryError::ReadonlyNeedsWrite);
                            }
                        }
                        if let (PlanNode::SeqScan { table }, false) =
                            (inner.as_ref(), contains_subquery(predicate))
                        {
                            if !self.catalog.table_has_overflow(table) {
                                let schema = self
                                    .catalog
                                    .schema(table)
                                    .ok_or_else(|| QueryError::TableNotFound(table.clone()))?
                                    .clone();
                                let columns: Vec<String> =
                                    schema.columns.iter().map(|c| c.name.clone()).collect();
                                let fast = FastLayout::new(&schema);
                                let row_layout = RowLayout::new(&schema);

                                if let Some(compiled) = self.compile_predicate_unless_forced(
                                    "readonly:count-filter:predicate",
                                    predicate,
                                    &columns,
                                    &fast,
                                    &schema,
                                ) {
                                    let mut count: i64 = 0;
                                    for_each_row_raw_cancellable(
                                        &self.catalog,
                                        table,
                                        |_rid, data| {
                                            if compiled(data) {
                                                count += 1;
                                            }
                                        },
                                    )?;
                                    return Ok(QueryResult::Scalar(Value::Int(count)));
                                }

                                let pred_cols = predicate_column_indices_json(predicate, &columns);
                                let mut count: i64 = 0;
                                for_each_row_raw_cancellable(
                                    &self.catalog,
                                    table,
                                    |_rid, data| {
                                        let pred_row = decode_selective(
                                            &schema,
                                            &row_layout,
                                            data,
                                            &pred_cols,
                                        );
                                        if eval_predicate(predicate, &pred_row, &columns) {
                                            count += 1;
                                        }
                                    },
                                )?;
                                return Ok(QueryResult::Scalar(Value::Int(count)));
                            }
                        }
                    }
                }

                // Fast path: sum/avg/min/max over single fixed-size numeric.
                if matches!(
                    function,
                    AggFunc::Sum
                        | AggFunc::Avg
                        | AggFunc::Min
                        | AggFunc::Max
                        | AggFunc::CountDistinct
                ) {
                    if let Some(Expr::Field(col)) = argument.as_ref() {
                        let (table_opt, pred_opt): (Option<&str>, Option<&Expr>) =
                            match input.as_ref() {
                                PlanNode::SeqScan { table } => (Some(table.as_str()), None),
                                PlanNode::Filter {
                                    input: inner,
                                    predicate,
                                } => {
                                    if let PlanNode::SeqScan { table } = inner.as_ref() {
                                        (Some(table.as_str()), Some(predicate))
                                    } else {
                                        (None, None)
                                    }
                                }
                                _ => (None, None),
                            };
                        if let Some(table) = table_opt {
                            if let Some(result) =
                                self.agg_single_col_fast(table, col, *function, pred_opt)?
                            {
                                return Ok(result);
                            }
                        }
                    }
                }

                // Generic path.
                let result = self.dispatch_readonly(input)?;
                match result {
                    QueryResult::Rows { columns, rows } => {
                        aggregate_rows(*function, argument.as_ref(), &columns, &rows)
                    }
                    _ => Err("aggregate requires row input".into()),
                }
            }

            PlanNode::Distinct { input } => {
                let result = self.dispatch_readonly(input)?;
                match result {
                    QueryResult::Rows { columns, rows } => {
                        let mut seen = std::collections::HashSet::new();
                        let mut unique_rows = Vec::new();
                        let mut cancel = crate::cancel::CancelCheck::new();
                        for row in rows {
                            cancel.tick()?;
                            if seen.insert(row.clone()) {
                                unique_rows.push(row);
                            }
                        }
                        Ok(QueryResult::Rows {
                            columns,
                            rows: unique_rows,
                        })
                    }
                    other => Ok(other),
                }
            }

            PlanNode::GroupBy {
                input,
                keys,
                aggregates,
                having,
            } => {
                if aggregates
                    .iter()
                    .any(|aggregate| aggregate.provenance_alias.is_some())
                {
                    let input = self.materialize_rows_with_provenance(input)?;
                    self.charge_rows(&input.rows)?;
                    return exec_group_by_with_provenance(
                        input,
                        keys,
                        aggregates,
                        having,
                        self.query_memory_limit,
                    );
                }
                let result = self.dispatch_readonly(input)?;
                match result {
                    QueryResult::Rows { columns, rows } => {
                        // WS2: byte-budget guard on the GROUP BY input buffer
                        // (the hash table is bounded by the input it groups).
                        self.charge_rows(&rows)?;
                        exec_group_by(columns, rows, keys, aggregates, having)
                    }
                    _ => Err("group by requires row input".into()),
                }
            }

            PlanNode::NestedLoopJoin {
                left,
                right,
                on,
                kind,
            } => {
                let left_result = self.dispatch_readonly(left)?;
                let right_result = self.dispatch_readonly(right)?;
                let (left_columns, left_rows) = match left_result {
                    QueryResult::Rows { columns, rows } => (columns, rows),
                    _ => return Err("join left side must produce rows".into()),
                };
                let (right_columns, right_rows) = match right_result {
                    QueryResult::Rows { columns, rows } => (columns, rows),
                    _ => return Err("join right side must produce rows".into()),
                };

                // WS2: byte-budget guard on the join build side.
                self.charge_rows(&left_rows)?;
                self.charge_rows(&right_rows)?;

                execute_materialized_join(
                    left_columns,
                    left_rows,
                    right_columns,
                    right_rows,
                    on.as_ref(),
                    *kind,
                    self.nested_loop_pair_limit,
                )
            }

            PlanNode::Window { input, windows } => {
                let result = self.dispatch_readonly(input)?;
                execute_window(result, windows, self.query_memory_limit)
            }

            PlanNode::Union { left, right, all } => {
                let left_result = self.dispatch_readonly(left)?;
                let right_result = self.dispatch_readonly(right)?;
                let (left_cols, left_rows) = match left_result {
                    QueryResult::Rows { columns, rows } => (columns, rows),
                    _ => return Err("UNION requires query results on left side".into()),
                };
                let (_, right_rows) = match right_result {
                    QueryResult::Rows { columns, rows } => (columns, rows),
                    _ => return Err("UNION requires query results on right side".into()),
                };
                Ok(QueryResult::Rows {
                    columns: left_cols,
                    rows: union_rows(left_rows, right_rows, *all)?,
                })
            }

            PlanNode::Explain { input } => {
                // Every execute entry point runs lower_unindexed_scans before
                // dispatch and lowering recurses into Explain, so `input` is
                // already the plan that will actually run.
                let text = format_plan_tree(&self.catalog, input, 0);
                Ok(QueryResult::Rows {
                    columns: vec!["plan".to_string()],
                    rows: text
                        .lines()
                        .map(|line| vec![Value::Str(line.to_string())])
                        .collect(),
                })
            }

            PlanNode::ListTypes => self.introspect_list_types(),

            PlanNode::Describe { table } => self.introspect_describe(table),

            PlanNode::ListLinks => self.introspect_list_links(),

            // All write variants — caller must escalate to the write lock.
            PlanNode::Insert { .. }
            | PlanNode::Update { .. }
            | PlanNode::Delete { .. }
            | PlanNode::Upsert { .. }
            | PlanNode::CreateTable { .. }
            | PlanNode::CreateLink { .. }
            | PlanNode::DropLink { .. }
            | PlanNode::AlterTable { .. }
            | PlanNode::DropTable { .. }
            | PlanNode::CreateView { .. }
            | PlanNode::RefreshView { .. }
            | PlanNode::DropView { .. }
            | PlanNode::Begin
            | PlanNode::Commit
            | PlanNode::Rollback => Err(QueryError::ReadonlyNeedsWrite),
        }
    }

    /// `&self` variant of [`Engine::materialize_subqueries`]. Used by the
    /// read path so `Filter` predicates with `InSubquery`/`ExistsSubquery`
    /// children can evaluate their inner queries without taking the write
    /// lock. Inner queries that would themselves need a write (e.g. dirty
    /// view) escalate via [`READONLY_NEEDS_WRITE`] just like the top-level
    /// read path does.
    fn materialize_subqueries_readonly(&self, expr: &Expr) -> Result<Expr, QueryError> {
        match expr {
            Expr::InSubquery {
                expr: inner,
                subquery,
                negated,
            } => {
                if is_correlated_subquery(subquery, &self.catalog) {
                    // Pass through — will be materialized per-row in the
                    // Filter handler's correlated subquery path.
                    let inner = self.materialize_subqueries_readonly(inner)?;
                    return Ok(Expr::InSubquery {
                        expr: Box::new(inner),
                        subquery: subquery.clone(),
                        negated: *negated,
                    });
                }
                let inner = self.materialize_subqueries_readonly(inner)?;
                let sub_plan = self.plan_and_lower(Statement::Query(*subquery.clone()))?;
                let result = self.execute_plan_readonly(&sub_plan)?;
                let values = match result {
                    QueryResult::Rows { rows, .. } => {
                        let mut values = Vec::with_capacity(rows.len());
                        let mut cancel = crate::cancel::CancelCheck::new();
                        for mut row in rows {
                            cancel.tick()?;
                            if !row.is_empty() {
                                values.push(value_to_expr(row.swap_remove(0)));
                            }
                        }
                        values
                    }
                    _ => Vec::new(),
                };
                // WS2: byte-budget guard on the materialized IN-list.
                self.charge_in_list(&values)?;
                Ok(Expr::InList {
                    expr: Box::new(inner),
                    list: values,
                    negated: *negated,
                })
            }
            Expr::ExistsSubquery { subquery, negated } => {
                if is_correlated_subquery(subquery, &self.catalog) {
                    return Ok(expr.clone());
                }
                let sub_plan = self.plan_and_lower(Statement::Query(*subquery.clone()))?;
                let result = self.execute_plan_readonly(&sub_plan)?;
                let has_rows = match result {
                    QueryResult::Rows { rows, .. } => !rows.is_empty(),
                    _ => false,
                };
                let truth = if *negated { !has_rows } else { has_rows };
                Ok(Expr::Literal(Literal::Bool(truth)))
            }
            Expr::BinaryOp(l, op, r) => {
                let l = self.materialize_subqueries_readonly(l)?;
                let r = self.materialize_subqueries_readonly(r)?;
                Ok(Expr::BinaryOp(Box::new(l), *op, Box::new(r)))
            }
            Expr::UnaryOp(op, inner) => {
                let inner = self.materialize_subqueries_readonly(inner)?;
                Ok(Expr::UnaryOp(*op, Box::new(inner)))
            }
            Expr::Case { whens, else_expr } => {
                let whens = whens
                    .iter()
                    .map(|(c, r)| {
                        let c = self.materialize_subqueries_readonly(c)?;
                        let r = self.materialize_subqueries_readonly(r)?;
                        Ok((Box::new(c), Box::new(r)))
                    })
                    .collect::<Result<Vec<_>, QueryError>>()?;
                let else_expr = match else_expr {
                    Some(e) => Some(Box::new(self.materialize_subqueries_readonly(e)?)),
                    None => None,
                };
                Ok(Expr::Case { whens, else_expr })
            }
            other => Ok(other.clone()),
        }
    }

    /// Per-row materialisation of correlated subqueries. For each row in the
    /// outer query, substitute outer column references in the subquery's
    /// filter with the current row's literal values, execute the modified
    /// subquery, and return the result as an InList or Bool literal.
    fn materialize_correlated_for_row_readonly(
        &self,
        expr: &Expr,
        outer_row: &[Value],
        outer_columns: &[String],
    ) -> Result<Expr, QueryError> {
        match expr {
            Expr::InSubquery {
                expr: inner,
                subquery,
                negated,
            } => {
                let inner =
                    self.materialize_correlated_for_row_readonly(inner, outer_row, outer_columns)?;
                let mut sub = *subquery.clone();
                if let Some(ref filter) = sub.filter {
                    sub.filter = Some(substitute_outer_refs(
                        filter,
                        &sub.source,
                        &self.catalog,
                        outer_row,
                        outer_columns,
                    ));
                }
                let sub_plan = self.plan_and_lower(Statement::Query(sub))?;
                let result = self.execute_plan_readonly(&sub_plan)?;
                let values = match result {
                    QueryResult::Rows { rows, .. } => {
                        let mut values = Vec::with_capacity(rows.len());
                        let mut cancel = crate::cancel::CancelCheck::new();
                        for mut row in rows {
                            cancel.tick()?;
                            if !row.is_empty() {
                                values.push(value_to_expr(row.swap_remove(0)));
                            }
                        }
                        values
                    }
                    _ => Vec::new(),
                };
                // WS2: byte-budget guard on the per-row materialized IN-list.
                self.charge_in_list(&values)?;
                Ok(Expr::InList {
                    expr: Box::new(inner),
                    list: values,
                    negated: *negated,
                })
            }
            Expr::ExistsSubquery { subquery, negated } => {
                let mut sub = *subquery.clone();
                if let Some(ref filter) = sub.filter {
                    sub.filter = Some(substitute_outer_refs(
                        filter,
                        &sub.source,
                        &self.catalog,
                        outer_row,
                        outer_columns,
                    ));
                }
                let sub_plan = self.plan_and_lower(Statement::Query(sub))?;
                let result = self.execute_plan_readonly(&sub_plan)?;
                let has_rows = match result {
                    QueryResult::Rows { rows, .. } => !rows.is_empty(),
                    _ => false,
                };
                let truth = if *negated { !has_rows } else { has_rows };
                Ok(Expr::Literal(Literal::Bool(truth)))
            }
            Expr::BinaryOp(l, op, r) => {
                let l =
                    self.materialize_correlated_for_row_readonly(l, outer_row, outer_columns)?;
                let r =
                    self.materialize_correlated_for_row_readonly(r, outer_row, outer_columns)?;
                Ok(Expr::BinaryOp(Box::new(l), *op, Box::new(r)))
            }
            Expr::UnaryOp(op, inner) => {
                let inner =
                    self.materialize_correlated_for_row_readonly(inner, outer_row, outer_columns)?;
                Ok(Expr::UnaryOp(*op, Box::new(inner)))
            }
            other => Ok(other.clone()),
        }
    }

    pub fn catalog(&self) -> &Catalog {
        &self.catalog
    }

    pub fn catalog_mut(&mut self) -> &mut Catalog {
        &mut self.catalog
    }
}

impl Drop for Engine {
    fn drop(&mut self) {
        let Some(hook) = self.wal_archive_hook.clone() else {
            return;
        };
        if let Err(err) = self
            .catalog
            .checkpoint_with_wal_archive(move |dir, records| hook(dir, records))
        {
            error!(error = %err, "sync-aware engine checkpoint on drop failed");
        }
    }
}