turso_core 0.6.1

//! The virtual database engine (VDBE).
//!
//! The VDBE is a register-based virtual machine that execute bytecode
//! instructions that represent SQL statements. When an application prepares
//! an SQL statement, the statement is compiled into a sequence of bytecode
//! instructions that perform the needed operations, such as reading or
//! writing to a b-tree, sorting, or aggregating data.
//!
//! The instruction set of the VDBE is similar to SQLite's instruction set,
//! but with the exception that bytecodes that perform I/O operations are
//! return execution back to the caller instead of blocking. This is because
//! Turso is designed for applications that need high concurrency such as
//! serverless runtimes. In addition, asynchronous I/O makes storage
//! disaggregation easier.
//!
//! You can find a full list of SQLite opcodes at:
//!
//! https://www.sqlite.org/opcode.html

use crate::translate::plan::BitSet;
use crate::types::{Extendable, Text};
use crate::{turso_assert, turso_assert_ne, turso_debug_assert, NonNan};
pub mod affinity;
pub mod array;
pub mod bloom_filter;
pub mod builder;
pub mod execute;
pub mod explain;
#[allow(dead_code)]
pub mod hash_table;
pub mod insn;
pub mod metrics;
pub mod rowset;
pub mod sorter;
pub mod vacuum;
pub mod value;
// for benchmarks
pub use crate::translate::collate::CollationSeq;
use crate::{
    error::LimboError,
    function::{AggFunc, FuncCtx},
    mvcc::{database::CommitStateMachine, MvccClock},
    numeric::Numeric,
    return_if_io,
    schema::Trigger,
    state_machine::StateMachine,
    translate::plan::TableReferences,
    types::{IOCompletions, IOResult},
    vdbe::{
        execute::{
            OpColumnState, OpDeleteState, OpDeleteSubState, OpDestroyState, OpIdxInsertState,
            OpInitCdcVersionState, OpInsertState, OpInsertSubState, OpJournalModeState,
            OpNewRowidState, OpNoConflictState, OpParseSchemaState, OpProgramState, OpRowIdState,
            OpSeekState, OpTransactionState, VacuumIntoOpContext,
        },
        hash_table::HashTable,
        metrics::StatementMetrics,
        vacuum::VacuumInPlaceOpContext,
    },
    ValueRef, WalAutoActions,
};
use smallvec::SmallVec;

#[cfg(feature = "json")]
use crate::json::JsonCacheCell;
use crate::sync::RwLock;
use crate::{
    storage::pager::Pager,
    translate::plan::ResultSetColumn,
    types::{AggContext, Cursor, ImmutableRecord, Value},
    vdbe::{builder::CursorType, insn::Insn},
};
use crate::{
    AtomicBool, CaptureDataChangesInfo, Connection, MvStore, Result, Statement, SyncMode,
    TransactionState,
};
use branches::{mark_unlikely, unlikely};
use builder::{CursorKey, QueryMode};
use execute::{
    InsnFunction, InsnFunctionStepResult, OpIdxDeleteState, OpIntegrityCheckState,
    OpOpenEphemeralState,
};
use turso_parser::ast::ResolveType;

use crate::io::TempFile;
use crate::vdbe::bloom_filter::BloomFilter;
use crate::vdbe::rowset::RowSet;
use explain::{insn_to_row_with_comment, EXPLAIN_COLUMNS, EXPLAIN_QUERY_PLAN_COLUMNS};
use std::{
    collections::HashMap,
    num::NonZero,
    ops::Deref,
    sync::{
        atomic::{AtomicI64, AtomicIsize, Ordering},
        Arc,
    },
    task::Waker,
};
use tracing::{instrument, Level};

/// State machine for committing view deltas with I/O handling
#[derive(Debug, Clone)]
pub enum ViewDeltaCommitState {
    NotStarted,
    Processing {
        views: Vec<String>, // view names (all materialized views have storage)
        current_index: usize,
    },
    Done,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
/// Represents a target for a jump instruction.
/// Stores 32-bit ints to keep the enum word-sized.
pub enum BranchOffset {
    /// A label is a named location in the program.
    /// If there are references to it, it must always be resolved to an Offset
    /// via `ProgramBuilder::preassign_label_to_next_insn` or
    /// `ProgramBuilder::link_label_to_other_label`.
    Label(u32),
    /// An offset is a direct index into the instruction list.
    Offset(InsnReference),
    /// A placeholder is a temporary value to satisfy the compiler.
    /// It must be set later.
    Placeholder,
}

impl BranchOffset {
    /// Returns true if the branch offset is an offset.
    pub fn is_offset(&self) -> bool {
        matches!(self, BranchOffset::Offset(_))
    }

    /// Returns the offset value. Panics if the branch offset is a label or placeholder.
    pub fn as_offset_int(&self) -> InsnReference {
        match self {
            BranchOffset::Label(v) => unreachable!("Unresolved label: {}", v),
            BranchOffset::Offset(v) => *v,
            BranchOffset::Placeholder => unreachable!("Unresolved placeholder"),
        }
    }

    /// Returns the branch offset as a signed integer.
    /// Used in explain output, where we don't want to panic in case we have an unresolved
    /// label or placeholder.
    pub fn as_debug_int(&self) -> i32 {
        match self {
            BranchOffset::Label(v) => *v as i32,
            BranchOffset::Offset(v) => *v as i32,
            BranchOffset::Placeholder => i32::MAX,
        }
    }
}

pub type CursorID = usize;

pub type PageIdx = i64;

// Index of insn in list of insns
type InsnReference = u32;

#[derive(Debug)]
pub enum StepResult {
    Done,
    IO,
    Row,
    Interrupt,
    Busy,
}

#[derive(Debug)]
#[allow(clippy::large_enum_variant)]
/// The commit state of the program.
/// There are two states:
/// - Ready: The program is ready to run the next instruction, or has shut down after
///   the last instruction.
/// - Committing: The program is committing a write transaction. It is waiting for the pager to finish flushing the cache to disk,
///   primarily to the WAL, but also possibly checkpointing the WAL to the database file.
enum CommitState {
    Ready,
    Committing,
    /// Committing attached database pagers after main pager commit is done.
    CommittingAttached,
    CommittingMvcc {
        state_machine: StateMachine<Box<CommitStateMachine<MvccClock>>>,
    },
    /// Committing MVCC transactions on attached databases after main MVCC commit is done.
    CommittingAttachedMvcc {
        state_machine: StateMachine<Box<CommitStateMachine<MvccClock>>>,
        db_id: usize,
        mv_store: Arc<MvStore>,
    },
}

impl CommitState {
    fn cleanup_mvcc_checkpoint_state(&mut self) {
        match self {
            CommitState::CommittingMvcc { state_machine } => {
                state_machine.inner_mut().cleanup_mvcc_checkpoint_state()
            }
            CommitState::CommittingAttachedMvcc { state_machine, .. } => {
                state_machine.inner_mut().cleanup_mvcc_checkpoint_state()
            }
            CommitState::Ready | CommitState::Committing | CommitState::CommittingAttached => {}
        }
    }

    fn cleanup_abandoned_mvcc_commit(&mut self, connection: &Connection) {
        match self {
            CommitState::CommittingAttachedMvcc {
                state_machine,
                db_id: attached_db_id,
                ..
            } if !state_machine.is_finalized() => {
                if connection
                    .database_schemas()
                    .write()
                    .remove(attached_db_id)
                    .is_some()
                {
                    connection.bump_prepare_context_generation();
                }
            }
            CommitState::CommittingMvcc { state_machine } if !state_machine.is_finalized() => {}
            _ => return, // no-op for already-finalized state machines and non-MVCC commit states
        };

        connection.rollback_attached_mvcc_txs(true);
        connection.rollback_attached_wal_txns();
    }
}

#[derive(Debug, Clone)]
pub enum Register {
    Value(Value),
    Aggregate(AggContext),
    Record(ImmutableRecord),
}

impl Register {
    #[inline]
    pub const fn is_null(&self) -> bool {
        matches!(self, Register::Value(Value::Null))
    }

    #[inline(always)]
    /// Sets the value of the register to an integer,
    /// reusing the existing Register::Value(Value::Numeric(Numeric::Integer(_))) if possible,
    /// which is faster than always creating a new one.
    pub fn set_int(&mut self, val: i64) {
        match self {
            Register::Value(Value::Numeric(Numeric::Integer(existing))) => {
                *existing = val;
            }
            Register::Value(Value::Numeric(float)) => {
                *float = Numeric::Integer(val);
            }
            Register::Value(other_value_kind) => {
                *other_value_kind = Value::from_i64(val);
            }
            _ => {
                *self = Register::Value(Value::from_i64(val));
            }
        }
    }
    /// Set the value of the register to a floating point,
    /// reusing Register::Value(Value::Numeric(Numeric::Float(_))) if possible.
    #[inline(always)]
    pub fn set_float(&mut self, val: NonNan) {
        match self {
            Register::Value(Value::Numeric(Numeric::Float(existing))) => {
                *existing = val;
            }
            Register::Value(Value::Numeric(integer)) => {
                *integer = Numeric::Float(val);
            }
            Register::Value(other_value_kind) => {
                *other_value_kind = Value::Numeric(Numeric::Float(val));
            }
            _ => {
                *self = Register::Value(Value::Numeric(Numeric::Float(val)));
            }
        }
    }

    /// Set the value of the register to a Text,
    /// reusing Register::Value(Value::Text(_)) buffer if possible.
    #[inline]
    pub fn set_text(&mut self, val: Text) {
        match self {
            Register::Value(Value::Text(existing)) => {
                existing.do_extend(&val);
            }
            Register::Value(other_value_kind) => {
                *other_value_kind = Value::Text(val);
            }
            _ => {
                *self = Register::Value(Value::Text(val));
            }
        }
    }

    /// Set the value of the register to a blob,
    /// reusing Register::Value(Value::Blob(_)) buffer if possible.
    #[inline]
    pub fn set_blob(&mut self, val: Vec<u8>) {
        match self {
            Register::Value(Value::Blob(existing)) => {
                existing.do_extend(&val);
            }
            Register::Value(other_value_kind) => {
                *other_value_kind = Value::Blob(val);
            }
            _ => {
                *self = Register::Value(Value::Blob(val));
            }
        }
    }

    // Set the value of the register to NULL,
    // reusing the existing Register::Value(Value::Null) if possible.
    pub fn set_null(&mut self) {
        match self {
            Register::Value(Value::Null) => {}
            Register::Value(other_value_kind) => {
                *other_value_kind = Value::Null;
            }
            _ => {
                *self = Register::Value(Value::Null);
            }
        }
    }

    /// Set the register to a generic Value, attempting to reuse backing allocation if compatible.
    pub fn set_value(&mut self, val: Value) {
        match self {
            Register::Value(v) => {
                *v = val;
            }
            _ => {
                *self = Register::Value(val);
            }
        }
    }
}

/// A row is a the list of registers that hold the values for a filtered row. This row is a pointer, therefore
/// after stepping again, row will be invalidated to be sure it doesn't point to somewhere unexpected.
#[derive(Debug)]
pub struct Row {
    values: *const Register,
    count: usize,
}

#[derive(Debug, Clone, Copy, PartialEq)]
pub enum TxnCleanup {
    None,
    RollbackTxn,
    /// begin_statement was called and statement is participating in an interactive transaction.
    /// If statement is abandoned and/or dropped without an apparent error, we should rollback statement
    /// to previous savepoint.
    RollbackSavepoint,
}

#[derive(Debug, Clone, Copy, PartialEq)]
pub enum ProgramExecutionState {
    /// No steps of the program was executed
    Init,
    /// Program started execution but didn't reach any terminal state
    Running,
    /// Interrupt requested for the program
    Interrupting,
    /// Terminal state: program interrupted
    Interrupted,
    /// Terminal state: program finished successfully
    Done,
    /// Terminal state: program failed with error
    Failed,
}

impl ProgramExecutionState {
    pub const fn is_running(&self) -> bool {
        matches!(
            self,
            ProgramExecutionState::Interrupting | ProgramExecutionState::Running
        )
    }
    pub const fn is_terminal(&self) -> bool {
        matches!(
            self,
            ProgramExecutionState::Interrupted
                | ProgramExecutionState::Failed
                | ProgramExecutionState::Done
        )
    }
}

/// Re-entrant state for [Insn::HashBuild].
/// Allows HashBuild to resume cleanly after async I/O without re-reading the row.
#[derive(Debug, Default)]
pub struct OpHashBuildState {
    pub key_values: Vec<Value>,
    pub key_idx: usize,
    pub payload_values: Vec<Value>,
    pub payload_idx: usize,
    pub rowid: Option<i64>,
    pub cursor_id: CursorID,
    pub hash_table_id: usize,
    pub key_start_reg: usize,
    pub num_keys: usize,
}

/// Re-entrant state for [Insn::HashProbe].
/// Allows HashProbe to resume cleanly after async probe-row buffering I/O.
#[derive(Debug, Default)]
pub struct OpHashProbeState {
    /// Cached probe key values to avoid re-reading from registers
    pub probe_keys: Vec<Value>,
    /// Hash table register being probed
    pub hash_table_id: usize,
    /// Partition index being loaded (if any)
    pub partition_idx: usize,
    /// Whether the probe row was already buffered for grace processing.
    pub probe_buffered: bool,
}

enum ActiveOpState {
    None,
    Delete(OpDeleteState),
    Destroy(OpDestroyState),
    IdxDelete(OpIdxDeleteState),
    IntegrityCheck(OpIntegrityCheckState),
    OpenEphemeral(OpOpenEphemeralState),
    Program(OpProgramState),
    NewRowid(OpNewRowidState),
    IdxInsert(OpIdxInsertState),
    Insert(OpInsertState),
    NoConflict(OpNoConflictState),
    Column(OpColumnState),
    RowId(OpRowIdState),
    Transaction(OpTransactionState),
    JournalMode(OpJournalModeState),
    ParseSchema(OpParseSchemaState),
    HashBuild(Option<OpHashBuildState>),
    HashProbe(Option<OpHashProbeState>),
    InitCdcVersion(OpInitCdcVersionState),
}

impl std::fmt::Debug for ActiveOpState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let name = match self {
            ActiveOpState::None => "None",
            ActiveOpState::Delete(_) => "Delete",
            ActiveOpState::Destroy(_) => "Destroy",
            ActiveOpState::IdxDelete(_) => "IdxDelete",
            ActiveOpState::IntegrityCheck(_) => "IntegrityCheck",
            ActiveOpState::OpenEphemeral(_) => "OpenEphemeral",
            ActiveOpState::Program(_) => "Program",
            ActiveOpState::NewRowid(_) => "NewRowid",
            ActiveOpState::IdxInsert(_) => "IdxInsert",
            ActiveOpState::Insert(_) => "Insert",
            ActiveOpState::NoConflict(_) => "NoConflict",
            ActiveOpState::Column(_) => "Column",
            ActiveOpState::RowId(_) => "RowId",
            ActiveOpState::Transaction(_) => "Transaction",
            ActiveOpState::JournalMode(_) => "JournalMode",
            ActiveOpState::ParseSchema(_) => "ParseSchema",
            ActiveOpState::HashBuild(_) => "HashBuild",
            ActiveOpState::HashProbe(_) => "HashProbe",
            ActiveOpState::InitCdcVersion(_) => "InitCdcVersion",
        };
        f.write_str(name)
    }
}

#[derive(Debug, Default)]
struct ActiveOpStateSlot {
    state: ActiveOpState,
}

macro_rules! active_state_accessor {
    ($name:ident, $variant:ident, $ty:ty, $init:expr) => {
        fn $name(&mut self) -> &mut $ty {
            if matches!(self.state, ActiveOpState::None) {
                self.state = ActiveOpState::$variant($init);
            }
            match &mut self.state {
                ActiveOpState::$variant(state) => state,
                state => unreachable!(
                    "active opcode state mismatch: expected {}, got {:?}",
                    stringify!($variant),
                    state
                ),
            }
        }
    };
}

impl Default for ActiveOpState {
    fn default() -> Self {
        Self::None
    }
}

impl ActiveOpStateSlot {
    fn clear(&mut self) {
        self.state = ActiveOpState::None;
    }

    active_state_accessor!(
        delete,
        Delete,
        OpDeleteState,
        OpDeleteState {
            sub_state: OpDeleteSubState::MaybeCaptureRecord,
            deleted_record: None,
        }
    );
    active_state_accessor!(
        destroy,
        Destroy,
        OpDestroyState,
        OpDestroyState::CreateCursor
    );
    active_state_accessor!(
        idx_delete,
        IdxDelete,
        OpIdxDeleteState,
        OpIdxDeleteState::Seeking
    );
    active_state_accessor!(
        integrity_check,
        IntegrityCheck,
        OpIntegrityCheckState,
        OpIntegrityCheckState::Start
    );
    active_state_accessor!(
        open_ephemeral,
        OpenEphemeral,
        OpOpenEphemeralState,
        OpOpenEphemeralState::Start
    );
    active_state_accessor!(program, Program, OpProgramState, OpProgramState::Start);
    active_state_accessor!(new_rowid, NewRowid, OpNewRowidState, OpNewRowidState::Start);
    active_state_accessor!(
        idx_insert,
        IdxInsert,
        OpIdxInsertState,
        OpIdxInsertState::MaybeSeek
    );
    active_state_accessor!(
        insert,
        Insert,
        OpInsertState,
        OpInsertState {
            sub_state: OpInsertSubState::MaybeCaptureRecord,
            old_record: None,
            is_noop_update: false,
        }
    );
    active_state_accessor!(
        no_conflict,
        NoConflict,
        OpNoConflictState,
        OpNoConflictState::Start
    );
    active_state_accessor!(column, Column, OpColumnState, OpColumnState::Start);
    active_state_accessor!(row_id, RowId, OpRowIdState, OpRowIdState::Start);
    active_state_accessor!(
        transaction,
        Transaction,
        OpTransactionState,
        OpTransactionState::Start
    );
    active_state_accessor!(
        journal_mode,
        JournalMode,
        OpJournalModeState,
        OpJournalModeState::default()
    );
    active_state_accessor!(parse_schema, ParseSchema, OpParseSchemaState, None);
    active_state_accessor!(hash_build, HashBuild, Option<OpHashBuildState>, None);
    active_state_accessor!(hash_probe, HashProbe, Option<OpHashProbeState>, None);
    active_state_accessor!(
        init_cdc_version,
        InitCdcVersion,
        OpInitCdcVersionState,
        None
    );

    fn program_ref(&self) -> Option<&OpProgramState> {
        match &self.state {
            ActiveOpState::Program(state) => Some(state),
            _ => None,
        }
    }

    fn program_mut(&mut self) -> Option<&mut OpProgramState> {
        match &mut self.state {
            ActiveOpState::Program(state) => Some(state),
            _ => None,
        }
    }
}

#[derive(Debug, Clone)]
pub(crate) struct DeferredSeekState {
    pub index_cursor_id: CursorID,
    pub table_cursor_id: CursorID,
}

pub(crate) enum VacuumOpState {
    None,
    IntoFile(Box<VacuumIntoOpContext>),
    InPlace(Box<VacuumInPlaceOpContext>),
}

impl Default for VacuumOpState {
    fn default() -> Self {
        Self::None
    }
}

/// The program state describes the environment in which the program executes.
pub struct ProgramState {
    pub io_completions: Option<IOCompletions>,
    pub pc: InsnReference,
    pub(crate) cursors: Vec<Option<Cursor>>,
    cursor_seqs: Vec<i64>,
    registers: Box<[Register]>,
    pub(crate) result_row: Option<Row>,
    last_compare: Option<std::cmp::Ordering>,
    deferred_seeks: Vec<Option<DeferredSeekState>>,
    /// Indicate whether a coroutine has ended for a given yield register.
    /// If an element is present, it means the coroutine with the given register number has ended.
    ended_coroutine: Vec<u32>,
    /// Indicate whether an [Insn::Once] instruction at a given program counter position has already been executed, well, once.
    once: SmallVec<[u32; 4]>,
    pub execution_state: ProgramExecutionState,
    /// Per-execution statement deadline derived from the connection query timeout.
    /// `None` means no timeout.
    pub query_deadline: Option<crate::MonotonicInstant>,
    pub parameters: Vec<Value>,
    commit_state: CommitState,
    #[cfg(feature = "json")]
    json_cache: JsonCacheCell,
    active_op_state: ActiveOpStateSlot,
    seek_state: OpSeekState,
    /// Metrics collected for the lifetime of this prepared statement.
    pub metrics: StatementMetrics,
    /// Current collation sequence set by OP_CollSeq instruction
    current_collation: Option<CollationSeq>,
    op_vacuum_state: VacuumOpState,
    /// State machine for committing view deltas with I/O handling
    view_delta_state: ViewDeltaCommitState,
    /// Marker which tells about auto transaction cleanup necessary for that connection in case of reset
    /// This is used when statement in auto-commit mode reseted after previous uncomplete execution - in which case we may need to rollback transaction started on previous attempt
    pub(crate) auto_txn_cleanup: TxnCleanup,
    pub explain_state: RwLock<ExplainState>,
    /// Scratch buffer for [Insn::HashDistinct] to avoid per-row allocations.
    distinct_key_values: Vec<Value>,
    hash_tables: HashMap<usize, HashTable>,
    /// TempFile handles for ephemeral cursors, keyed by cursor_id.
    /// Dropping removes the temp file from disk.
    ephemeral_temp_files: HashMap<usize, TempFile>,
    /// Attached pagers that have open savepoints for statement rollback.
    attached_savepoint_pagers: Vec<Arc<Pager>>,
    /// Pending error to return after FAIL mode commit completes.
    /// When a constraint error occurs with FAIL resolve type in autocommit mode,
    /// we need to commit partial changes before returning the error.
    pub(crate) pending_fail_error: Option<LimboError>,
    /// Pending CDC info to apply after the program completes successfully.
    /// Set by InitCdcVersion opcode, applied at Halt/Done so that if the
    /// transaction rolls back, the connection's CDC state remains unchanged.
    ///
    /// capture_data_changes has type Option<CaptureDataChangesInfo> (off mode is None)
    /// so, for pending_cdc_info we wrap it in one more Option<...> layer to represent if mode changed during program execution
    pub(crate) pending_cdc_info: Option<Option<CaptureDataChangesInfo>>,
    /// Cached subprogram Statements keyed by the PC of the Program instruction.
    /// Avoids re-allocating ProgramState on each trigger/FK-action fire.
    pub(crate) subprogram_stmt_cache: HashMap<usize, Box<Statement>>,
    /// RowSet objects stored by register index
    rowsets: HashMap<usize, RowSet>,
    /// Bloom filters stored by cursor ID for probabilistic set membership testing
    /// Used to avoid unnecessary seeks on ephemeral indexes and hash tables
    pub(crate) bloom_filters: HashMap<usize, BloomFilter>,
    /// Number of deferred foreign key violations when the statement started.
    /// When a statement subtransaction rolls back, the connection's deferred foreign key violations counter
    /// is reset to this value.
    fk_deferred_violations_when_stmt_started: AtomicIsize,
    /// Number of immediate foreign key violations that occurred during the active statement. If nonzero,
    /// the statement subtransactionwill roll back.
    fk_immediate_violations_during_stmt: AtomicIsize,
    uses_subjournal: bool,
    /// Whether this statement is an active write inside an explicit transaction.
    pub(crate) is_active_write: bool,
    /// Whether begin_statement was called (savepoint + FK bookkeeping active).
    has_stmt_transaction: bool,
    pub n_change: AtomicI64,
}

impl std::fmt::Debug for Program {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Program").finish()
    }
}

// See: https://github.com/tursodatabase/turso/issues/1552
// SAFETY: Rust cannot derive Send + Sync automatically mainly because of `Row` struct
// as it contains a `*const Register`.
// Program + Program State upholds Rust aliasing rules with `Row` by only giving out immutable references to
// the internal `result_row` and by invalidating the result row whenever the program is stepped.
unsafe impl Send for ProgramState {}
unsafe impl Sync for ProgramState {}
crate::assert::assert_send_sync!(ProgramState);

impl ProgramState {
    pub fn new(max_registers: usize, max_cursors: usize) -> Self {
        let cursors: Vec<Option<Cursor>> = (0..max_cursors).map(|_| None).collect();
        let cursor_seqs = vec![0i64; max_cursors];
        let registers = vec![Register::Value(Value::Null); max_registers].into_boxed_slice();
        Self {
            io_completions: None,
            pc: 0,
            cursors,
            cursor_seqs,
            registers,
            result_row: None,
            last_compare: None,
            deferred_seeks: vec![None; max_cursors],
            ended_coroutine: vec![],
            once: SmallVec::<[u32; 4]>::new(),
            execution_state: ProgramExecutionState::Init,
            query_deadline: None,
            parameters: Vec::new(),
            commit_state: CommitState::Ready,
            #[cfg(feature = "json")]
            json_cache: JsonCacheCell::new(),
            active_op_state: ActiveOpStateSlot::default(),
            seek_state: OpSeekState::Start,
            metrics: StatementMetrics::new(),
            distinct_key_values: Vec::new(),
            current_collation: None,
            op_vacuum_state: VacuumOpState::None,
            view_delta_state: ViewDeltaCommitState::NotStarted,
            auto_txn_cleanup: TxnCleanup::None,
            fk_deferred_violations_when_stmt_started: AtomicIsize::new(0),
            fk_immediate_violations_during_stmt: AtomicIsize::new(0),
            rowsets: HashMap::default(),
            bloom_filters: HashMap::default(),
            hash_tables: HashMap::default(),
            ephemeral_temp_files: HashMap::default(),
            uses_subjournal: false,
            is_active_write: false,
            has_stmt_transaction: false,
            attached_savepoint_pagers: Vec::new(),
            n_change: AtomicI64::new(0),
            explain_state: RwLock::new(ExplainState::default()),
            pending_fail_error: None,
            pending_cdc_info: None,
            subprogram_stmt_cache: HashMap::default(),
        }
    }

    pub fn set_register(&mut self, idx: usize, value: Register) {
        self.registers[idx] = value;
    }

    pub fn get_register(&self, idx: usize) -> &Register {
        &self.registers[idx]
    }

    pub fn column_count(&self) -> usize {
        self.registers.len()
    }

    pub fn column(&self, i: usize) -> Option<String> {
        Some(format!("{:?}", self.registers[i]))
    }

    pub fn interrupt(&mut self) {
        self.execution_state = ProgramExecutionState::Interrupting;
    }

    pub fn is_interrupted(&self) -> bool {
        matches!(self.execution_state, ProgramExecutionState::Interrupting)
    }

    pub fn bind_at(&mut self, index: NonZero<usize>, value: Value) {
        let i = index.get() - 1;
        if i >= self.parameters.len() {
            self.parameters.resize(i + 1, Value::Null);
        }
        let slot = &mut self.parameters[i];
        match (slot, value) {
            (Value::Null, Value::Null) => {}
            (Value::Numeric(Numeric::Integer(existing)), Value::Numeric(Numeric::Integer(new))) => {
                *existing = new
            }
            (Value::Numeric(Numeric::Float(existing)), Value::Numeric(Numeric::Float(new))) => {
                *existing = new
            }
            (Value::Text(existing), Value::Text(new)) => existing.do_extend(&new),
            (Value::Blob(existing), Value::Blob(new)) => existing.do_extend(&new),
            (slot, value) => *slot = value,
        }
    }

    pub fn clear_bindings(&mut self) {
        self.parameters.clear();
    }

    pub fn get_parameter(&self, index: NonZero<usize>) -> Value {
        let i = index.get() - 1;
        self.parameters.get(i).cloned().unwrap_or(Value::Null)
    }

    pub fn reset(&mut self, max_registers: Option<usize>, max_cursors: Option<usize>) {
        self.io_completions = None;
        self.pc = 0;

        if let Some(max_cursors) = max_cursors {
            self.cursors.resize_with(max_cursors, || None);
            self.cursor_seqs.resize(max_cursors, 0);
            self.deferred_seeks.resize(max_cursors, None);
        }
        self.result_row = None;
        if let Some(max_registers) = max_registers {
            // into_vec and into_boxed_slice do not allocate
            let mut registers = std::mem::take(&mut self.registers).into_vec();
            // As we are dropping whatever is in the result row, we can be sure that no one is referencing values from `*const Register` inside `Row`.
            registers.resize_with(max_registers, || Register::Value(Value::Null));
            self.registers = registers.into_boxed_slice();
        }
        // reset cursors as they can have cached information which will be no longer relevant on next program execution
        self.cursors.iter_mut().for_each(|c| {
            let _ = c.take();
        });
        for r in self.registers.iter_mut() {
            match r {
                Register::Value(v) => *v = Value::Null,
                _ => r.set_null(),
            }
        }
        self.last_compare = None;
        self.deferred_seeks.iter_mut().for_each(|s| *s = None);
        self.ended_coroutine.clear();
        self.once.clear();
        self.execution_state = ProgramExecutionState::Init;
        self.query_deadline = None;
        self.current_collation = None;
        #[cfg(feature = "json")]
        self.json_cache.clear();

        // A caller can reset or drop a statement after an MVCC auto-checkpoint
        // has yielded I/O. Waiting for that I/O does not step the nested
        // CheckpointStateMachine again, so release its checkpoint lock before
        // replacing commit_state with Ready.
        self.commit_state.cleanup_mvcc_checkpoint_state();
        self.active_op_state.clear();
        self.seek_state = OpSeekState::Start;
        self.current_collation = None;
        self.commit_state = CommitState::Ready;
        self.op_vacuum_state = VacuumOpState::None;
        self.view_delta_state = ViewDeltaCommitState::NotStarted;
        self.auto_txn_cleanup = TxnCleanup::None;
        self.fk_immediate_violations_during_stmt
            .store(0, Ordering::SeqCst);
        self.fk_deferred_violations_when_stmt_started
            .store(0, Ordering::SeqCst);
        self.rowsets.clear();
        self.bloom_filters.clear();
        self.hash_tables.clear();
        self.ephemeral_temp_files.clear();
        self.uses_subjournal = false;
        self.is_active_write = false;
        self.has_stmt_transaction = false;
        self.distinct_key_values.clear();
        self.attached_savepoint_pagers.clear();
        self.n_change.store(0, Ordering::SeqCst);
        *self.explain_state.write() = ExplainState::default();
        self.pending_fail_error = None;
        self.pending_cdc_info = None;
        self.subprogram_stmt_cache.clear();
    }

    /// Whether this statement owns the implicit autocommit transaction it is
    /// about to finish, including re-entry while its commit is in progress.
    #[inline]
    pub(crate) fn owns_auto_txn(&self) -> bool {
        self.auto_txn_cleanup == TxnCleanup::RollbackTxn
            || !matches!(self.commit_state, CommitState::Ready)
    }

    #[inline]
    pub fn record_rows_read(&mut self, count: u64) {
        self.metrics.rows_read = self.metrics.rows_read.saturating_add(count);
    }

    #[inline]
    pub fn record_rows_written(&mut self, count: u64) {
        self.metrics.rows_written = self.metrics.rows_written.saturating_add(count);
    }

    pub(crate) fn metrics(&self) -> StatementMetrics {
        let mut metrics = self.metrics.clone();
        if let Some(OpProgramState::Step { statement, .. }) = self.active_op_state.program_ref() {
            metrics.merge(&statement.metrics());
        }
        for statement in self.subprogram_stmt_cache.values() {
            metrics.merge(&statement.metrics());
        }
        metrics
    }

    pub(crate) fn reset_metrics(&mut self) {
        self.metrics.reset();
        if let Some(OpProgramState::Step { statement, .. }) = self.active_op_state.program_mut() {
            statement.reset_metrics();
        }
        for statement in self.subprogram_stmt_cache.values_mut() {
            statement.reset_metrics();
        }
    }

    pub(crate) fn reset_stmt_status(&mut self, counter: crate::statement::StatementStatusCounter) {
        match counter {
            crate::statement::StatementStatusCounter::FullscanStep => {
                self.metrics.fullscan_steps = 0
            }
            crate::statement::StatementStatusCounter::Sort => self.metrics.sort_operations = 0,
            crate::statement::StatementStatusCounter::VmStep => self.metrics.insn_executed = 0,
            crate::statement::StatementStatusCounter::Reprepare => self.metrics.reprepares = 0,
            crate::statement::StatementStatusCounter::RowsRead => self.metrics.rows_read = 0,
            crate::statement::StatementStatusCounter::RowsWritten => self.metrics.rows_written = 0,
        }
        if let Some(OpProgramState::Step { statement, .. }) = self.active_op_state.program_mut() {
            statement.reset_stmt_status(counter);
        }
        for statement in self.subprogram_stmt_cache.values_mut() {
            statement.reset_stmt_status(counter);
        }
    }

    pub fn get_cursor(&mut self, cursor_id: CursorID) -> &mut Cursor {
        self.cursors
            .get_mut(cursor_id)
            .unwrap_or_else(|| panic!("cursor id {cursor_id} out of bounds"))
            .as_mut()
            .unwrap_or_else(|| panic!("cursor id {cursor_id} is None"))
    }

    /// Begin a statement subtransaction.
    ///
    /// Creates a savepoint on the main DB's MvStore (or pager for WAL mode),
    /// and snapshots FK violation counters for potential statement rollback.
    /// Attached DB savepoints are opened per-DB in `op_transaction_inner`
    /// when each DB's Transaction opcode is executed.
    ///
    /// Pager/MVCC savepoints are only opened for write statements inside an
    /// explicit transaction. In autocommit mode, a statement abort is a
    /// transaction abort, so savepoints are unnecessary.
    pub fn begin_statement(
        &mut self,
        connection: &Connection,
        pager: &Arc<Pager>,
        write: bool,
    ) -> Result<IOResult<()>> {
        let in_explicit_txn = !connection.auto_commit.load(Ordering::SeqCst);
        if write && in_explicit_txn {
            // Check if MVCC is active - if so, use MVCC savepoints instead of pager savepoints
            if let Some(mv_store) = connection.mv_store().as_ref() {
                if let Some(tx_id) = connection.get_mv_tx_id() {
                    mv_store.begin_savepoint(tx_id);
                }
            } else {
                // Non-MVCC mode: use pager savepoints
                let db_size = return_if_io!(pager.with_header(|header| header.database_size.get()));
                pager.open_subjournal()?;
                pager.try_use_subjournal()?;
                let result = pager.open_savepoint(db_size);
                if result.is_err() {
                    pager.stop_use_subjournal();
                }
                result?;
                self.uses_subjournal = true;
            }
        }

        self.has_stmt_transaction = true;

        // Store the deferred foreign key violations counter at the start of the statement.
        // This is used to ensure that if an interactive transaction had deferred FK violations and a statement subtransaction rolls back,
        // the deferred FK violations are not lost.
        self.fk_deferred_violations_when_stmt_started.store(
            connection.fk_deferred_violations.load(Ordering::Acquire),
            Ordering::SeqCst,
        );
        // Reset the immediate foreign key violations counter to 0. If this is nonzero when the statement completes, the statement subtransaction will roll back.
        self.fk_immediate_violations_during_stmt
            .store(0, Ordering::Release);
        Ok(IOResult::Done(()))
    }

    /// End a statement subtransaction.
    ///
    /// Mirrors SQLite's vdbeCloseStatement (vdbeaux.c:3203-3248). Pager/MVCC
    /// savepoint management and FK violation counter restoration are independent
    /// concerns: pager savepoints may be skipped (e.g. autocommit optimization)
    /// while FK bookkeeping still needs cleanup.
    pub fn end_statement(
        &mut self,
        connection: &Connection,
        pager: &Arc<Pager>,
        end_statement: EndStatement,
    ) -> Result<()> {
        if self.is_active_write {
            connection.n_active_writes.fetch_sub(1, Ordering::SeqCst);
            self.is_active_write = false;
        }
        // If begin_statement was never called, no savepoint/FK cleanup needed.
        if !self.has_stmt_transaction {
            return Ok(());
        }
        self.has_stmt_transaction = false;

        // Drain attached pagers upfront so we can clean them up regardless of path.
        let attached_pagers: Vec<Arc<Pager>> = self.attached_savepoint_pagers.drain(..).collect();
        let result = match end_statement {
            EndStatement::ReleaseSavepoint => {
                if let Some(mv_store) = connection.mv_store().as_ref() {
                    if let Some(tx_id) = connection.get_mv_tx_id() {
                        mv_store.release_savepoint(tx_id);
                    }
                    connection.for_each_attached_mv_tx(|db_id, tx_id| {
                        if let Some(attached_mv) = connection.mv_store_for_db(db_id) {
                            attached_mv.release_savepoint(tx_id);
                        }
                    });
                    Ok(())
                } else if self.uses_subjournal || !attached_pagers.is_empty() {
                    if self.uses_subjournal {
                        pager.release_savepoint()?;
                    }
                    for p in &attached_pagers {
                        p.release_savepoint()?;
                    }
                    Ok(())
                } else {
                    Ok(())
                }
            }
            EndStatement::RollbackSavepoint => {
                // Rollback pager/MVCC savepoint if one was opened.
                let pager_err = if let Some(mv_store) = connection.mv_store().as_ref() {
                    let mut err = None;
                    if let Some(tx_id) = connection.get_mv_tx_id() {
                        if let Err(e) = mv_store.rollback_first_savepoint(tx_id) {
                            err = Some(e);
                        }
                    }
                    connection.for_each_attached_mv_tx(|db_id, tx_id| {
                        if let Some(attached_mv) = connection.mv_store_for_db(db_id) {
                            if let Err(e) = attached_mv.rollback_first_savepoint(tx_id) {
                                if err.is_none() {
                                    err = Some(e);
                                }
                            }
                        }
                    });
                    err
                } else if self.uses_subjournal {
                    match pager.rollback_to_newest_savepoint() {
                        Ok(_) => {
                            let mut err = None;
                            for p in &attached_pagers {
                                if let Err(e) = p.rollback_to_newest_savepoint() {
                                    err = Some(e);
                                    break;
                                }
                            }
                            err
                        }
                        Err(e) => Some(e),
                    }
                } else if !attached_pagers.is_empty() {
                    let mut err = None;
                    for p in &attached_pagers {
                        if let Err(e) = p.rollback_to_newest_savepoint() {
                            err = Some(e);
                        }
                    }
                    err
                } else {
                    None
                };

                // Always restore FK violation counters on statement rollback,
                // regardless of whether a pager savepoint was opened.
                // Mirrors SQLite's vdbeCloseStatement (vdbeaux.c:3243-3246).
                connection.fk_deferred_violations.store(
                    self.fk_deferred_violations_when_stmt_started
                        .load(Ordering::Acquire),
                    Ordering::SeqCst,
                );

                match pager_err {
                    Some(e) => Err(e),
                    None => Ok(()),
                }
            }
        };
        if self.uses_subjournal {
            pager.stop_use_subjournal();
            self.uses_subjournal = false;
        }
        for p in &attached_pagers {
            p.stop_use_subjournal();
        }
        result
    }

    /// Gets or creates a bloom filter for the given cursor ID.
    pub fn get_or_create_bloom_filter(&mut self, cursor_id: usize) -> &mut BloomFilter {
        self.bloom_filters.entry(cursor_id).or_default()
    }

    /// Gets or creates a bloom filter with a specific capacity for the given cursor ID.
    pub fn get_or_create_bloom_filter_with_capacity(
        &mut self,
        cursor_id: usize,
        expected_items: u32,
        false_positive_rate: f32,
    ) -> &mut BloomFilter {
        self.bloom_filters
            .entry(cursor_id)
            .or_insert_with(|| BloomFilter::with_capacity(expected_items, false_positive_rate))
    }

    /// Gets an existing bloom filter for the given cursor ID.
    pub fn get_bloom_filter(&self, cursor_id: usize) -> Option<&BloomFilter> {
        self.bloom_filters.get(&cursor_id)
    }

    /// Gets a mutable reference to an existing bloom filter for the given cursor ID.
    pub fn get_bloom_filter_mut(&mut self, cursor_id: usize) -> Option<&mut BloomFilter> {
        self.bloom_filters.get_mut(&cursor_id)
    }

    /// Removes and drops the bloom filter for the given cursor ID.
    pub fn remove_bloom_filter(&mut self, cursor_id: usize) {
        self.bloom_filters.remove(&cursor_id);
    }

    /// Checks if a bloom filter exists for the given cursor ID.
    pub fn has_bloom_filter(&self, cursor_id: usize) -> bool {
        self.bloom_filters.contains_key(&cursor_id)
    }

    pub fn get_fk_immediate_violations_during_stmt(&self) -> isize {
        self.fk_immediate_violations_during_stmt
            .load(Ordering::Acquire)
    }

    pub fn increment_fk_immediate_violations_during_stmt(&self, v: isize) {
        self.fk_immediate_violations_during_stmt
            .fetch_add(v, Ordering::AcqRel);
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Action to take at the end of a statement subtransaction.
pub enum EndStatement {
    /// Release (commit) the savepoint -- effectively removing the savepoint as it is no longer needed for undo purposes.
    ReleaseSavepoint,
    /// Rollback (abort) to the newest savepoint: read pages from the subjournal and restore them to the page cache.
    /// This is used to undo the changes made by the statement.
    RollbackSavepoint,
}

impl Register {
    pub fn get_value(&self) -> &Value {
        match self {
            Register::Value(v) => v,
            Register::Record(r) => {
                turso_assert!(!r.is_invalidated());
                r.as_blob_value()
            }
            _ => panic!("register holds unexpected value: {self:?}"),
        }
    }
}

#[macro_export]
macro_rules! must_be_btree_cursor {
    ($cursor_id:expr, $cursor_ref:expr, $state:expr, $insn_name:expr) => {{
        let (_, cursor_type) = $cursor_ref.get($cursor_id).unwrap();
        if matches!(
            cursor_type,
            CursorType::BTreeTable(_)
                | CursorType::BTreeIndex(_)
                | CursorType::MaterializedView(_, _)
        ) {
            $crate::get_cursor!($state, $cursor_id)
        } else {
            panic!("{} on unexpected cursor", $insn_name)
        }
    }};
}

/// Macro is necessary to help the borrow checker see we are only accessing state.cursor field
/// and nothing else
#[macro_export]
macro_rules! get_cursor {
    ($state:expr, $cursor_id:expr) => {
        $state
            .cursors
            .get_mut($cursor_id)
            .unwrap_or_else(|| panic!("cursor id {} out of bounds", $cursor_id))
            .as_mut()
            .unwrap_or_else(|| panic!("cursor id {} is None", $cursor_id))
    };
}

/// Tracks the state of explain mode execution, including which subprograms need to be processed.
#[derive(Default)]
pub struct ExplainState {
    /// Subprograms queued for explain output, processed after the parent program finishes.
    pending: std::collections::VecDeque<Arc<PreparedProgram>>,
    /// The subprogram currently being explained, if any.
    current: Option<Arc<PreparedProgram>>,
}

#[derive(Debug, Clone)]
pub struct PreparedProgram {
    pub max_registers: usize,
    // we store original indices because we don't want to create new vec from
    // ProgramBuilder
    pub insns: Vec<(Insn, usize)>,
    pub cursor_ref: Vec<(Option<CursorKey>, CursorType)>,
    pub comments: Vec<(InsnReference, &'static str)>,
    pub parameters: crate::parameters::Parameters,
    pub change_cnt_on: bool,
    /// Flag that detect if the sqlite statement will directly manipulate the database file.\
    /// mirrors: https://sqlite.org/c3ref/stmt_readonly.html.
    pub readonly: bool,
    pub result_columns: Vec<ResultSetColumn>,
    pub table_references: TableReferences,
    pub sql: String,
    /// Whether the statement needs to be wrapped in a statement subtransaction
    /// when run as part of an interactive (non-autocommit) transaction.
    /// See [crate::vdbe::builder::ProgramBuilder::is_multi_write] and [crate::vdbe::builder::ProgramBuilder::may_abort] for more details.
    pub needs_stmt_subtransactions: Arc<AtomicBool>,
    /// If this Program is a trigger subprogram, a ref to the trigger is stored here.
    pub trigger: Option<Arc<Trigger>>,
    /// Whether this program is a subprogram (trigger or FK action) that runs within a parent statement.
    pub is_subprogram: bool,
    /// Whether the program contains any trigger subprograms.
    pub contains_trigger_subprograms: bool,
    pub resolve_type: ResolveType,
    pub prepare_context: PrepareContext,
    /// Set of attached database indices that need write transactions.
    pub write_databases: BitSet,
    /// Set of attached database indices that need read transactions.
    pub read_databases: BitSet,
}

#[derive(Clone)]
pub struct Program {
    pub(crate) prepared: Arc<PreparedProgram>,
    pub connection: Arc<Connection>,
}

/// Captures connection settings at statement preparation time for cache invalidation.
///
/// This struct is used to detect when a cached prepared statement needs to be recompiled
/// because relevant connection settings have changed. When `matches_connection()` returns
/// false, the statement will be automatically reprepared before execution.
///
/// # Adding New Fields
///
/// If you add a new setting to `Connection` that affects statement compilation or execution,
/// When adding a new connection setting that affects query compilation, you MUST call
/// `bump_prepare_context_generation()` in its setter so that prepared statements know
/// they need to be reprepared.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PrepareContext {
    /// Identity check: the prepared statement must belong to the same database.
    database_ptr: usize,
    /// Generation counter snapshot taken at prepare time. Compared against the
    /// connection's current generation to detect setting changes (pragmas,
    /// attach/detach, extension registration, etc.) without rebuilding the full
    /// context on every step.
    generation: u64,
}

impl PrepareContext {
    pub fn from_connection(connection: &Connection) -> Self {
        Self {
            database_ptr: connection.database_ptr(),
            generation: connection.prepare_context_generation(),
        }
    }

    #[inline]
    pub fn matches_connection(&self, connection: &Connection) -> bool {
        self.database_ptr == connection.database_ptr()
            && self.generation == connection.prepare_context_generation()
    }
}

impl PreparedProgram {
    pub fn bind(self: Arc<Self>, connection: Arc<Connection>) -> Program {
        Program {
            prepared: self,
            connection,
        }
    }

    pub fn is_compatible_with(&self, connection: &Connection) -> bool {
        self.prepare_context.matches_connection(connection)
    }

    #[inline]
    pub const fn is_readonly(&self) -> bool {
        self.readonly
    }
}

impl Program {
    #[inline]
    pub fn prepared(&self) -> &Arc<PreparedProgram> {
        &self.prepared
    }

    pub fn from_prepared(prepared: Arc<PreparedProgram>, connection: Arc<Connection>) -> Self {
        Self {
            prepared,
            connection,
        }
    }

    #[inline]
    pub fn is_readonly(&self) -> bool {
        self.prepared().is_readonly()
    }
}

impl Program {
    fn get_pager_from_database_index(&self, idx: &usize) -> Result<Arc<Pager>> {
        self.connection.get_pager_from_database_index(idx)
    }

    #[inline]
    fn maybe_request_interrupt<I>(&self, state: &mut ProgramState, io: &I) -> bool
    where
        I: crate::IO + ?Sized,
    {
        let connection_interrupt = self.connection.is_interrupted();
        let hit_query_deadline = state
            .query_deadline
            .is_some_and(|deadline| io.current_time_monotonic() >= deadline);
        let progress_interrupt = self
            .connection
            .should_interrupt_for_progress(state.metrics.vm_steps);
        if connection_interrupt || hit_query_deadline || progress_interrupt {
            state.interrupt();
        }
        state.is_interrupted()
    }

    #[turso_macros::trace_stack]
    pub fn step(
        &self,
        state: &mut ProgramState,
        pager: &Arc<Pager>,
        query_mode: QueryMode,
        waker: Option<&Waker>,
    ) -> Result<StepResult> {
        state.execution_state = ProgramExecutionState::Running;
        let result = match query_mode {
            QueryMode::Normal => self.normal_step(state, pager, waker),
            QueryMode::Explain => self.explain_step(state, pager),
            QueryMode::ExplainQueryPlan => self.explain_query_plan_step(state, pager),
        };
        match &result {
            Ok(StepResult::Done) => {
                state.execution_state = ProgramExecutionState::Done;
            }
            Ok(StepResult::Interrupt) => {
                state.execution_state = ProgramExecutionState::Interrupted;
            }
            Err(_) => {
                state.execution_state = ProgramExecutionState::Failed;
            }
            _ => {}
        }
        result
    }

    fn explain_step(&self, state: &mut ProgramState, pager: &Arc<Pager>) -> Result<StepResult> {
        turso_debug_assert!(state.column_count() == EXPLAIN_COLUMNS.len());
        if self.connection.is_closed() {
            let tx_state = self.connection.get_tx_state();
            if let TransactionState::Write { .. } = tx_state {
                pager.rollback_tx(&self.connection);
            }
            return Err(LimboError::InternalError("Connection closed".to_string()));
        }

        if self.maybe_request_interrupt(state, pager.io.as_ref()) {
            return Ok(StepResult::Interrupt);
        }

        state.metrics.vm_steps = state.metrics.vm_steps.saturating_add(1);

        let mut explain_state = state.explain_state.write();

        // Advance to the next subprogram if the current one is finished
        loop {
            if let Some(ref current) = explain_state.current {
                if (state.pc as usize) < current.insns.len() {
                    break;
                }
            } else if (state.pc as usize) < self.insns.len() {
                break;
            }
            // Current program is done, pop next subprogram from queue
            if let Some(next) = explain_state.pending.pop_front() {
                explain_state.current = Some(next);
                state.pc = 0;
            } else {
                explain_state.current = None;
                return Ok(StepResult::Done);
            }
        }

        let pc = state.pc as usize;

        // Explain the current instruction from the active program.
        // We collect subprograms separately to avoid borrow conflicts with explain_state.
        let (row, subprogram) = if let Some(ref current) = explain_state.current {
            let (insn, _) = &current.insns[pc];
            let sub = if let Insn::Program {
                program: prepared, ..
            } = insn
            {
                Some(prepared.clone())
            } else {
                None
            };
            let comment = current
                .comments
                .iter()
                .find(|(offset, _)| *offset == state.pc)
                .map(|(_, c)| *c);
            (insn_to_row_with_comment(current, insn, comment), sub)
        } else {
            let (insn, _) = &self.insns[pc];
            let sub = if let Insn::Program {
                program: prepared, ..
            } = insn
            {
                Some(prepared.clone())
            } else {
                None
            };
            let comment = self
                .comments
                .iter()
                .find(|(offset, _)| *offset == state.pc)
                .map(|(_, c)| *c);
            (insn_to_row_with_comment(self, insn, comment), sub)
        };
        if let Some(sub) = subprogram {
            explain_state.pending.push_back(sub);
        }
        let (opcode, p1, p2, p3, p4, p5, comment) = row;

        state.registers[0].set_int(state.pc as i64);
        state.registers[1].set_value(Value::from_text(opcode));
        state.registers[2].set_int(p1);
        state.registers[3].set_int(p2);
        state.registers[4].set_int(p3);
        state.registers[5].set_value(p4);
        state.registers[6].set_int(p5);
        state.registers[7].set_value(Value::from_text(comment));
        state.result_row = Some(Row {
            values: &state.registers[0] as *const Register,
            count: EXPLAIN_COLUMNS.len(),
        });
        state.pc += 1;
        Ok(StepResult::Row)
    }

    fn explain_query_plan_step(
        &self,
        state: &mut ProgramState,
        pager: &Arc<Pager>,
    ) -> Result<StepResult> {
        turso_debug_assert!(state.column_count() == EXPLAIN_QUERY_PLAN_COLUMNS.len());
        loop {
            if self.connection.is_closed() {
                // Connection is closed for whatever reason, rollback the transaction.
                let state = self.connection.get_tx_state();
                if let TransactionState::Write { .. } = state {
                    pager.rollback_tx(&self.connection);
                }
                return Err(LimboError::InternalError("Connection closed".to_string()));
            }

            if self.maybe_request_interrupt(state, pager.io.as_ref()) {
                return Ok(StepResult::Interrupt);
            }

            // FIXME: do we need this?
            state.metrics.vm_steps = state.metrics.vm_steps.saturating_add(1);

            if state.pc as usize >= self.insns.len() {
                return Ok(StepResult::Done);
            }

            let Insn::Explain { p1, p2, detail } = &self.insns[state.pc as usize].0 else {
                state.pc += 1;
                continue;
            };

            state.registers[0].set_int(*p1 as i64);
            state.registers[1] =
                Register::Value(Value::from_i64(p2.as_ref().map(|p| *p).unwrap_or(0) as i64));
            state.registers[2].set_int(0);
            state.registers[3].set_value(Value::from_text(detail.clone()));
            state.result_row = Some(Row {
                values: &state.registers[0] as *const Register,
                count: EXPLAIN_QUERY_PLAN_COLUMNS.len(),
            });
            state.pc += 1;
            return Ok(StepResult::Row);
        }
    }

    #[instrument(skip_all, level = Level::DEBUG)]
    fn normal_step(
        &self,
        state: &mut ProgramState,
        pager: &Arc<Pager>,
        waker: Option<&Waker>,
    ) -> Result<StepResult> {
        let enable_tracing = tracing::enabled!(tracing::Level::TRACE);
        loop {
            if self.connection.is_closed() {
                // Connection is closed for whatever reason, rollback the transaction.
                let state = self.connection.get_tx_state();
                if let TransactionState::Write { .. } = state {
                    pager.rollback_tx(&self.connection);
                }
                return Err(LimboError::InternalError("Connection closed".to_string()));
            }
            if self.maybe_request_interrupt(state, pager.io.as_ref()) {
                self.abort(pager, None, state)?;
                return Ok(StepResult::Interrupt);
            }

            if let Some(io) = &state.io_completions {
                if !io.finished() {
                    io.set_waker(waker);
                    return Ok(StepResult::IO);
                }
                if let Some(err) = io.get_error() {
                    if pager.is_checkpointing() {
                        // Wrap IO errors that occurred during checkpointing in CheckpointFailed error,
                        // so that abort() knows not to try to rollback the transaction, because the transaction
                        // is already durable in the WAL and hence committed.
                        // This also lets the simulator know that it should shadow the results of the query because
                        // the write itself succeeded.
                        let checkpoint_err = LimboError::CheckpointFailed(err.to_string());
                        tracing::error!("Checkpoint failed: {checkpoint_err}");
                        if let Err(abort_err) = self.abort(pager, Some(&checkpoint_err), state) {
                            tracing::error!(
                                "Abort also failed during checkpoint error handling: {abort_err}"
                            );
                        }
                        return Err(checkpoint_err);
                    }
                    let err = err.into();
                    if let Err(abort_err) = self.abort(pager, Some(&err), state) {
                        tracing::error!("Abort failed during error handling: {abort_err}");
                    }
                    return Err(err);
                }
                state.io_completions = None;
            }
            // invalidate row
            let _ = state.result_row.take();
            let (insn, _) = &self.insns[state.pc as usize];
            let insn_function = insn.to_function();
            if enable_tracing {
                trace_insn(self, state.pc as InsnReference, insn);
                crate::stack::trace_remaining("program_step:opcode");
            }
            // Always increment VM steps for every loop iteration
            state.metrics.vm_steps = state.metrics.vm_steps.saturating_add(1);

            match insn_function(self, state, insn, pager) {
                Ok(InsnFunctionStepResult::Step) => {
                    // Instruction completed, moving to next
                    state.metrics.insn_executed = state.metrics.insn_executed.saturating_add(1);
                }
                Ok(InsnFunctionStepResult::Done) => {
                    // Instruction completed execution
                    state.metrics.insn_executed = state.metrics.insn_executed.saturating_add(1);
                    state.auto_txn_cleanup = TxnCleanup::None;
                    return Ok(StepResult::Done);
                }
                Ok(InsnFunctionStepResult::IO(io)) => {
                    // Instruction not complete - waiting for I/O, will resume at same PC
                    io.set_waker(waker);
                    let is_yield = io.is_explicit_yield();
                    if is_yield {
                        // Yield: return control to the cooperative scheduler so
                        // other connections can make progress (e.g. release a
                        // contended lock). Don't store in io_completions —
                        // yields aren't pending I/O, so the instruction will
                        // simply re-execute on the next step.
                        return Ok(StepResult::IO);
                    }
                    let finished = io.finished();
                    state.io_completions = Some(io);
                    if !finished {
                        return Ok(StepResult::IO);
                    }
                    // just continue the outer loop if IO is finished so db will continue execution immediately
                }
                Ok(InsnFunctionStepResult::Row) => {
                    // Instruction completed (ResultRow already incremented PC)
                    state.metrics.insn_executed = state.metrics.insn_executed.saturating_add(1);
                    return Ok(StepResult::Row);
                }
                Err(LimboError::Busy) => {
                    // Instruction blocked - will retry at same PC
                    return Ok(StepResult::Busy);
                }
                Err(LimboError::BusySnapshot)
                    if self.connection.transaction_state.get() == TransactionState::None =>
                {
                    // For interactive transactions that are already in a read transaction, retrying BusySnapshot is pointless
                    // because the snapshot will continue to be stale no matter how many times we retry.
                    // However, for auto-commits or BEGIN IMMEDIATE, failing to promote to write transaction means it was rolled
                    // back, so auto-retrying can be useful.
                    return Ok(StepResult::Busy);
                }
                Err(err) => {
                    if let Err(abort_err) = self.abort(pager, Some(&err), state) {
                        tracing::error!("Abort failed during error handling: {abort_err}");
                    }
                    return Err(err);
                }
            }
        }
    }

    #[instrument(skip_all, level = Level::DEBUG)]
    fn apply_view_deltas(
        &self,
        state: &mut ProgramState,
        rollback: bool,
        pager: &Arc<Pager>,
    ) -> Result<IOResult<()>> {
        use crate::types::IOResult;

        loop {
            match &state.view_delta_state {
                ViewDeltaCommitState::NotStarted => {
                    if self.connection.view_transaction_states.is_empty() {
                        return Ok(IOResult::Done(()));
                    }

                    if rollback {
                        // On rollback, just clear and done
                        self.connection.view_transaction_states.clear();
                        return Ok(IOResult::Done(()));
                    }

                    // Not a rollback - proceed with processing
                    let schema = self.connection.schema.read();

                    // Collect materialized views - they should all have storage
                    let mut views = Vec::new();
                    for view_name in self.connection.view_transaction_states.get_view_names() {
                        if let Some(view_mutex) = schema.get_materialized_view(&view_name) {
                            let view = view_mutex.lock();
                            let root_page = view.get_root_page();

                            // Materialized views should always have storage (root_page != 0)
                            turso_assert_ne!(
                                root_page, 0,
                                "Materialized view should have a root page",
                                { "view_name": view_name }
                            );

                            views.push(view_name);
                        }
                    }

                    state.view_delta_state = ViewDeltaCommitState::Processing {
                        views,
                        current_index: 0,
                    };
                }

                ViewDeltaCommitState::Processing {
                    views,
                    current_index,
                } => {
                    // At this point we know it's not a rollback
                    if *current_index >= views.len() {
                        // All done, clear the transaction states
                        self.connection.view_transaction_states.clear();
                        state.view_delta_state = ViewDeltaCommitState::Done;
                        return Ok(IOResult::Done(()));
                    }

                    let view_name = &views[*current_index];

                    let table_deltas = self
                        .connection
                        .view_transaction_states
                        .get(view_name)
                        .expect("view should have transaction state")
                        .get_table_deltas();

                    let schema = self.connection.schema.read();
                    if let Some(view_mutex) = schema.get_materialized_view(view_name) {
                        let mut view = view_mutex.lock();

                        // Create a DeltaSet from the per-table deltas
                        let mut delta_set = crate::incremental::compiler::DeltaSet::new();
                        for (table_name, delta) in table_deltas {
                            delta_set.insert(table_name, delta);
                        }

                        // Handle I/O from merge_delta - pass pager, circuit will create its own cursor
                        match view.merge_delta(delta_set, pager.clone())? {
                            IOResult::Done(_) => {
                                // Move to next view
                                state.view_delta_state = ViewDeltaCommitState::Processing {
                                    views: views.clone(),
                                    current_index: current_index + 1,
                                };
                            }
                            IOResult::IO(io) => {
                                // Return I/O, will resume at same index
                                return Ok(IOResult::IO(io));
                            }
                        }
                    }
                }

                ViewDeltaCommitState::Done => {
                    return Ok(IOResult::Done(()));
                }
            }
        }
    }

    pub fn commit_txn(
        &self,
        pager: Arc<Pager>,
        program_state: &mut ProgramState,
        mv_store: Option<&Arc<MvStore>>,
        rollback: bool,
    ) -> Result<IOResult<()>> {
        // Apply view deltas with I/O handling
        match self.apply_view_deltas(program_state, rollback, &pager)? {
            IOResult::IO(io) => return Ok(IOResult::IO(io)),
            IOResult::Done(_) => {}
        }

        // Reset state for next use
        program_state.view_delta_state = ViewDeltaCommitState::NotStarted;
        let tx_state = self.connection.get_tx_state();
        if tx_state == TransactionState::None
            && matches!(program_state.commit_state, CommitState::Ready)
        {
            // No main transaction and no in-progress commit — check whether
            // any attached/temp database still has an active transaction before
            // bailing out. Defer these checks to here so the common case
            // (active main transaction) doesn't pay for the lock reads.
            let has_attached_mv_tx = self.connection.next_attached_mv_tx().is_some();
            let has_attached_wal_tx =
                self.connection
                    .with_all_attached_pagers_with_index(|pagers| {
                        pagers.iter().any(|(_, pager)| pager.holds_read_lock())
                    });
            if !has_attached_mv_tx && !has_attached_wal_tx {
                return Ok(IOResult::Done(()));
            }
        }
        if self.connection.is_nested_stmt() {
            // We don't want to commit on nested statements. Let parent handle it.
            return Ok(IOResult::Done(()));
        }
        let res = if let Some(mv_store) = mv_store {
            self.commit_txn_mvcc(pager, program_state, mv_store, rollback)
        } else {
            self.commit_txn_wal(pager, program_state, rollback)
        }?;
        if !res.is_io() {
            if self.change_cnt_on {
                self.connection
                    .set_changes(program_state.n_change.load(Ordering::SeqCst));
            }
            let transaction_finished = self.connection.auto_commit.load(Ordering::SeqCst)
                && self.connection.get_tx_state() == TransactionState::None;
            if transaction_finished {
                // Finalize the in-memory TEMP schema only when the outer
                // transaction actually finishes. Updating the committed temp
                // snapshot after every statement inside an explicit
                // transaction would make a later full ROLLBACK restore
                // uncommitted temp DDL.
                if rollback {
                    self.connection.rollback_temp_schema();
                } else {
                    self.connection.commit_temp_schema();
                }
            }
        }
        Ok(res)
    }

    fn commit_txn_wal(
        &self,
        pager: Arc<Pager>,
        program_state: &mut ProgramState,
        rollback: bool,
    ) -> Result<IOResult<()>> {
        let connection = self.connection.clone();
        let auto_commit = connection.auto_commit.load(Ordering::SeqCst);
        let tx_state = connection.get_tx_state();
        tracing::debug!(
            "Halt auto_commit {}, commit_state={:?}, tx_state={:?}",
            auto_commit,
            program_state.commit_state,
            tx_state,
        );
        if matches!(program_state.commit_state, CommitState::Committing) {
            let TransactionState::Write { .. } = tx_state else {
                unreachable!("invalid state for write commit step")
            };
            self.step_end_write_txn(&pager, &connection, program_state, rollback)
        } else if matches!(program_state.commit_state, CommitState::CommittingAttached) {
            // Re-entry after IO yield from attached pager commit.
            match self.end_attached_write_txns(&connection, rollback)? {
                IOResult::Done(_) => {
                    program_state.commit_state = CommitState::Ready;
                    if pager.holds_read_lock() {
                        pager.end_read_tx();
                    }
                    self.end_attached_read_txns(&connection);
                    Ok(IOResult::Done(()))
                }
                IOResult::IO(io) => Ok(IOResult::IO(io)),
            }
        } else if auto_commit {
            match tx_state {
                TransactionState::Write { .. } => {
                    self.step_end_write_txn(&pager, &connection, program_state, rollback)
                }
                TransactionState::Read => {
                    connection.set_tx_state(TransactionState::None);
                    // Commit any attached write transactions that were opened
                    // independently of the main connection's transaction state.
                    // (e.g., UPDATE aux0.t SET ... only needs Read on main DB
                    // but holds a write lock on the attached pager.)
                    match self.end_attached_write_txns(&connection, rollback)? {
                        IOResult::Done(_) => {}
                        IOResult::IO(io) => {
                            program_state.commit_state = CommitState::CommittingAttached;
                            return Ok(IOResult::IO(io));
                        }
                    }
                    pager.end_read_tx();
                    self.end_attached_read_txns(&connection);
                    Ok(IOResult::Done(()))
                }
                TransactionState::None => {
                    match self.end_attached_write_txns(&connection, rollback)? {
                        IOResult::Done(_) => {}
                        IOResult::IO(io) => {
                            program_state.commit_state = CommitState::CommittingAttached;
                            return Ok(IOResult::IO(io));
                        }
                    }
                    self.end_attached_read_txns(&connection);
                    Ok(IOResult::Done(()))
                }
                TransactionState::PendingUpgrade { .. } => {
                    panic!("Unexpected transaction state: {tx_state:?} during auto-commit",)
                }
            }
        } else {
            Ok(IOResult::Done(()))
        }
    }

    /// Commit MVCC transactions across all databases in a multi-phase protocol:
    ///
    /// 1. **Main DB MVCC** — commit the main database's MvStore transaction.
    /// 2. **Attached MVCC** — commit each attached database's MvStore transaction.
    /// 3. **Attached WAL** — flush dirty pages on attached databases that use WAL
    ///    (e.g. :memory: attached while main is MVCC).
    ///
    /// **IMPORTANT**: This multi-phase commit is NOT atomic across databases.
    /// A crash between phases can leave the main and attached databases in
    /// inconsistent states (main committed, some attached DBs not committed).
    /// This matches SQLite's WAL mode behavior — cross-file atomicity only
    /// exists in legacy rollback journal mode, which we do not support.
    fn commit_txn_mvcc(
        &self,
        pager: Arc<Pager>,
        program_state: &mut ProgramState,
        mv_store: &Arc<MvStore>,
        rollback: bool,
    ) -> Result<IOResult<()>> {
        let conn = self.connection.clone();
        let auto_commit = conn.auto_commit.load(Ordering::SeqCst);
        if !auto_commit {
            return Ok(IOResult::Done(()));
        }

        // Phase 1: Commit main DB MVCC transaction
        if matches!(program_state.commit_state, CommitState::Ready) {
            if let Some(tx_id) = conn.get_mv_tx_id() {
                let state_machine = mv_store.commit_tx(tx_id, &conn, crate::MAIN_DB_ID)?;
                program_state.commit_state = CommitState::CommittingMvcc { state_machine };
            }
            // If no main MVCC tx, commit_state stays Ready and we fall
            // through directly to phase 2 (the CommittingMvcc and
            // CommittingAttachedMvcc checks will both miss).
        }

        if matches!(
            program_state.commit_state,
            CommitState::CommittingMvcc { .. }
        ) {
            let CommitState::CommittingMvcc { state_machine } = &mut program_state.commit_state
            else {
                unreachable!()
            };
            match self.step_end_mvcc_txn(state_machine, mv_store)? {
                IOResult::Done(_) => {
                    assert!(state_machine.is_finalized());
                    conn.set_mv_tx(None);
                    conn.set_tx_state(TransactionState::None);
                    pager.end_read_tx();
                    program_state.commit_state = CommitState::Ready;
                    // Fall through to attached phase
                }
                IOResult::IO(io) => return Ok(IOResult::IO(io)),
            }
        }

        // Phase 2: Commit MVCC transactions on attached databases
        // Resume an in-progress attached MVCC commit
        if matches!(
            program_state.commit_state,
            CommitState::CommittingAttachedMvcc { .. }
        ) {
            let (step_result, db_id) = {
                let CommitState::CommittingAttachedMvcc {
                    state_machine,
                    db_id,
                    mv_store: ref attached_mv,
                } = &mut program_state.commit_state
                else {
                    unreachable!()
                };
                (state_machine.step(attached_mv)?, *db_id)
            };
            match step_result {
                IOResult::Done(_) => {
                    let attached_pager = conn
                        .get_pager_from_database_index(&db_id)
                        .expect("attached MVCC transaction should always have a pager");
                    conn.publish_database_schema(db_id);
                    conn.set_mv_tx_for_db(db_id, None);
                    attached_pager.end_read_tx();
                    // Fall through to look for more
                }
                IOResult::IO(io) => return Ok(IOResult::IO(io)),
            }
        }

        // Start/continue committing remaining attached MVCC transactions
        loop {
            let Some((db_id, tx_id, _mode)) = conn.next_attached_mv_tx() else {
                break;
            };
            let Some(attached_mv_store) = conn.mv_store_for_db(db_id) else {
                conn.set_mv_tx_for_db(db_id, None);
                continue;
            };
            let mut state_machine = match attached_mv_store.commit_tx(tx_id, &conn, db_id) {
                Ok(sm) => sm,
                Err(e) => {
                    tracing::error!(
                        db_id,
                        "attached DB commit failed after main DB already committed; \
                         cross-database state is inconsistent: {e}"
                    );
                    // Rollback remaining uncommitted attached MVCC transactions
                    // so they don't block checkpointing until connection close.
                    conn.rollback_attached_mvcc_txs(true);
                    return Err(e);
                }
            };
            match state_machine.step(&attached_mv_store)? {
                IOResult::Done(_) => {
                    let attached_pager = conn
                        .get_pager_from_database_index(&db_id)
                        .expect("attached MVCC transaction should always have a pager");
                    conn.publish_database_schema(db_id);
                    conn.set_mv_tx_for_db(db_id, None);
                    attached_pager.end_read_tx();
                    continue;
                }
                IOResult::IO(io) => {
                    program_state.commit_state = CommitState::CommittingAttachedMvcc {
                        state_machine,
                        db_id,
                        mv_store: attached_mv_store,
                    };
                    return Ok(IOResult::IO(io));
                }
            }
        }

        // Phase 3: Commit WAL transactions on attached databases that don't use MVCC.
        // When the main DB uses MVCC, we route through commit_txn_mvcc, but attached
        // DBs may use WAL mode and need their dirty pages committed via the WAL path.
        if matches!(program_state.commit_state, CommitState::CommittingAttached) {
            // Re-entry after IO yield from attached WAL pager commit.
            match self.end_attached_write_txns(&conn, rollback)? {
                IOResult::Done(_) => {
                    program_state.commit_state = CommitState::Ready;
                    self.end_attached_read_txns(&conn);
                    return Ok(IOResult::Done(()));
                }
                IOResult::IO(io) => return Ok(IOResult::IO(io)),
            }
        }

        match self.end_attached_write_txns(&conn, rollback)? {
            IOResult::Done(_) => {}
            IOResult::IO(io) => {
                program_state.commit_state = CommitState::CommittingAttached;
                return Ok(IOResult::IO(io));
            }
        }
        self.end_attached_read_txns(&conn);

        program_state.commit_state = CommitState::Ready;
        Ok(IOResult::Done(()))
    }

    #[instrument(skip(self, pager, connection, program_state), level = Level::DEBUG)]
    fn step_end_write_txn(
        &self,
        pager: &Arc<Pager>,
        connection: &Connection,
        program_state: &mut ProgramState,
        rollback: bool,
    ) -> Result<IOResult<()>> {
        let commit_state = &mut program_state.commit_state;
        if matches!(commit_state, CommitState::CommittingAttached) {
            // Resume committing attached pagers after IO yield.
            match self.end_attached_write_txns(connection, rollback)? {
                IOResult::Done(_) => {
                    *commit_state = CommitState::Ready;
                }
                IOResult::IO(io) => {
                    return Ok(IOResult::IO(io));
                }
            }
            // Release read locks on attached pagers that only had read transactions
            // (end_attached_write_txns only handles pagers with write locks).
            self.end_attached_read_txns(connection);
            return Ok(IOResult::Done(()));
        }
        let txn_finish_result = if !rollback {
            pager.commit_tx(connection, true)
        } else {
            pager.rollback_tx(connection);
            Ok(IOResult::Done(()))
        };
        tracing::debug!("txn_finish_result: {:?}", txn_finish_result);
        match txn_finish_result? {
            IOResult::Done(_) => {
                // Main pager commit done, now commit attached database pagers
                match self.end_attached_write_txns(connection, rollback)? {
                    IOResult::Done(_) => {
                        *commit_state = CommitState::Ready;
                    }
                    IOResult::IO(io) => {
                        *commit_state = CommitState::CommittingAttached;
                        return Ok(IOResult::IO(io));
                    }
                }
            }
            IOResult::IO(io) => {
                tracing::trace!("Cacheflush IO");
                *commit_state = CommitState::Committing;
                return Ok(IOResult::IO(io));
            }
        }
        // Release read locks on attached pagers that only had read transactions
        // (end_attached_write_txns only handles pagers with write locks).
        self.end_attached_read_txns(connection);
        Ok(IOResult::Done(()))
    }

    /// End write transactions on all attached databases that hold write locks.
    /// Iterates ALL attached pagers (not just the current program's write_databases)
    /// because in explicit transactions, the COMMIT statement's program may differ
    /// from the statement that acquired the attached write lock.
    /// On IO yield, already-committed pagers are skipped on re-entry via holds_write_lock().
    fn end_attached_write_txns(
        &self,
        connection: &Connection,
        rollback: bool,
    ) -> Result<IOResult<()>> {
        connection.with_all_attached_pagers_with_index(|pagers| {
            for (db_id, attached_pager) in pagers {
                let db_id = *db_id;
                // MVCC-enabled attached DBs are committed in commit_txn_mvcc phase 2
                if connection.mv_store_for_db(db_id).is_some() {
                    continue;
                }
                if !attached_pager.holds_write_lock() {
                    continue;
                }
                if !rollback {
                    // Commit dirty pages to WAL, then end write+read transactions.
                    // We disable auto-checkpoint and avoid pager.commit_tx() since
                    // the checkpoint logic can leave read locks held.
                    match attached_pager.commit_dirty_pages(
                        WalAutoActions::empty(),
                        SyncMode::Normal,
                        false,
                    ) {
                        Ok(IOResult::Done(_)) => {}
                        Ok(IOResult::IO(io)) => {
                            // IO pending — return so the caller can yield and re-enter.
                            // commit_dirty_pages tracks its own internal state, so calling
                            // it again on re-entry will resume correctly.
                            return Ok(IOResult::IO(io));
                        }
                        Err(e) => return Err(e),
                    }
                    // WAL commit succeeded — publish the connection-local schema
                    // changes to the shared Database so other connections can see them.
                    connection.publish_database_schema(db_id);
                    attached_pager.end_write_tx();
                    attached_pager.end_read_tx();
                    attached_pager.commit_dirty_pages_end();
                } else {
                    // Discard any local schema changes on rollback
                    connection.database_schemas().write().remove(&db_id);
                    attached_pager.rollback_attached();
                }
            }
            Ok(IOResult::Done(()))
        })
    }

    /// End read transactions on all attached databases that had transactions started.
    fn end_attached_read_txns(&self, connection: &Connection) {
        connection.with_all_attached_pagers_with_index(|pagers| {
            pagers.iter().for_each(|(db_id, attached_pager)| {
                if connection.mv_store_for_db(*db_id).is_some() {
                    // MVCC-enabled attached DBs don't use WAL read transactions, so skip.
                    return;
                }
                if attached_pager.holds_write_lock() {
                    // Attached pager has a write lock, so its read transaction was ended by end_attached_write_txns: skip.
                    return;
                }
                if attached_pager.holds_read_lock() {
                    attached_pager.end_read_tx();
                }
            });
        })
    }

    #[instrument(skip(self, commit_state, mv_store), level = Level::DEBUG)]
    fn step_end_mvcc_txn(
        &self,
        commit_state: &mut StateMachine<Box<CommitStateMachine<MvccClock>>>,
        mv_store: &Arc<MvStore>,
    ) -> Result<IOResult<()>> {
        commit_state.step(mv_store)
    }

    /// Aborts the program due to various conditions (explicit error, interrupt or reset of unfinished statement) by rolling back the transaction
    /// This method is no-op if program was already finished (either aborted or executed to completion)
    /// Returns an error if cleanup operations (savepoint rollback/release) fail.
    pub fn abort(
        &self,
        pager: &Arc<Pager>,
        err: Option<&LimboError>,
        state: &mut ProgramState,
    ) -> Result<()> {
        fn capture_abort_error(
            abort_error: &mut Option<LimboError>,
            err: LimboError,
            context: &str,
        ) {
            tracing::error!("{context}: {err}");
            if abort_error.is_none() {
                *abort_error = Some(err);
            }
        }

        let mut abort_error: Option<LimboError> = None;
        // MVCC auto-checkpoint is owned by commit_state, not by normal_step().
        // If its yielded I/O fails, normal_step sees the error before
        // CommitStateMachine::Checkpoint gets another step, so the checkpoint
        // state machine cannot run its own error cleanup. abort() is the first
        // statement cleanup path that still owns that commit_state.
        state.commit_state.cleanup_mvcc_checkpoint_state();
        // If a CommitStateMachine was non-terminal when the program was
        // aborted — Statement dropped mid-IO yield, or `?` propagated a Busy
        // out of BeginCommitLogicalLog / SyncLogicalLog — release the locks
        // it acquired (`pager_commit_lock`, `exclusive_tx`) and roll back the
        // orphan tx. Without this the tx stays in `Preparing`, the next op on
        // this connection trips a `turso_assert_eq!(Active)`, and any other
        // writer parks forever on the leaked `pager_commit_lock`. The
        // following err-match's no-rollback arms (Busy / TxError / etc.)
        // would otherwise skip this cleanup.
        state
            .commit_state
            .cleanup_abandoned_mvcc_commit(&self.connection);

        // VACUUM (and VACUUM INTO) state can own internal helper statements whose drop path
        // releases nested guards. Clean it before checking whether this program
        // is itself nested; otherwise abort could skip top-level cleanup.
        if let Err(err) = execute::cleanup_vacuum_state(&self.connection, state) {
            capture_abort_error(
                &mut abort_error,
                err,
                "Failed to clean up VACUUM state during abort",
            );
        }

        // Only end trigger execution if the subprogram was actually running.
        // Cached (pooled) statements may be dropped after their trigger execution
        // was already ended by op_program; calling end again would pop the wrong
        // entry from the executing_triggers stack.
        if self.is_trigger_subprogram() && state.execution_state.is_running() {
            self.connection.end_trigger_execution();
        }
        // Errors from nested statements are handled by the parent statement.
        if !self.connection.is_nested_stmt() && !self.is_trigger_subprogram() {
            let owns_auto_txn = state.owns_auto_txn();
            if err.is_some() && !pager.is_checkpointing() {
                // For ON CONFLICT FAIL, do NOT rollback the statement savepoint —
                // changes made before the error should persist.
                // For all other resolve types (ABORT, ROLLBACK, etc.), rollback the statement.
                let is_fail_constraint = (matches!(err, Some(LimboError::Constraint(_)))
                    && self.resolve_type == ResolveType::Fail)
                    || matches!(err, Some(LimboError::Raise(ResolveType::Fail, _)));
                if !is_fail_constraint {
                    if let Err(end_stmt_err) = state.end_statement(
                        &self.connection,
                        pager,
                        EndStatement::RollbackSavepoint,
                    ) {
                        capture_abort_error(
                            &mut abort_error,
                            end_stmt_err,
                            "Failed to rollback statement savepoint during abort",
                        );
                    }
                }
            }
            match err {
                // Transaction errors, e.g. trying to start a nested transaction, do not cause a rollback.
                Some(LimboError::TxError(_)) => {}
                // Table locked errors, e.g. trying to checkpoint in an interactive transaction, do not cause a rollback.
                Some(LimboError::TableLocked) => {}
                // Busy errors do not cause a rollback.
                Some(LimboError::Busy) => {}
                // BusySnapshot errors do not cause a rollback either - user must rollback explicitly.
                // BusySnapshot is distinct from Busy in that a busy_timeout or handler should not be
                // used because it will not help - the snapshot is permanently stale and rollback is
                // the only way out for this poor transaction.
                Some(LimboError::BusySnapshot) => {}
                // Schema updated errors do not cause a rollback; the statement will be reprepared and retried,
                // and the caller is expected to handle transaction cleanup explicitly if needed.
                Some(LimboError::SchemaUpdated) => {}
                // Foreign key constraint errors: ON CONFLICT does NOT apply to FK violations.
                // FK errors always behave like ABORT: rollback statement,
                // rollback transaction in autocommit mode.
                Some(LimboError::ForeignKeyConstraint(_)) => {
                    if owns_auto_txn {
                        self.rollback_current_txn(pager);
                    }
                    self.connection.set_changes_without_total(0);
                }
                // Constraint and RAISE errors: behavior depends on the effective resolve type.
                // For normal constraints, the resolve type comes from the statement (ON CONFLICT).
                // For RAISE errors, the resolve type is embedded in the error variant itself.
                // - ROLLBACK: rollback the entire transaction regardless of autocommit mode
                // - FAIL: don't rollback anything - changes persist, transaction stays active
                // - ABORT (default): rollback statement, rollback txn if autocommit
                Some(LimboError::Constraint(_)) | Some(LimboError::Raise(_, _)) => {
                    let effective_resolve = match err {
                        Some(LimboError::Raise(rt, _)) => *rt,
                        _ => self.resolve_type,
                    };
                    match effective_resolve {
                        ResolveType::Rollback => {
                            self.rollback_current_txn(pager);
                            // All deferred FK violations are undone by the full rollback.
                            self.connection.clear_deferred_foreign_key_violations();
                        }
                        ResolveType::Fail => {
                            // FAIL: Don't rollback the transaction.
                            // Changes made before the error persist.
                            if let Err(end_stmt_err) = state.end_statement(
                                &self.connection,
                                pager,
                                EndStatement::ReleaseSavepoint,
                            ) {
                                capture_abort_error(
                                    &mut abort_error,
                                    end_stmt_err,
                                    "Failed to release statement savepoint during abort",
                                );
                            }
                            if owns_auto_txn {
                                // Autocommit FAIL: commit partial changes.
                                // This matches halt()'s FAIL+autocommit path.
                                let mv_store = self.connection.mv_store();
                                if let Err(e) = execute::vtab_commit_all(&self.connection) {
                                    capture_abort_error(
                                        &mut abort_error,
                                        e,
                                        "vtab_commit_all failed during FAIL abort",
                                    );
                                }
                                if let Err(e) = execute::index_method_pre_commit_all(state, pager) {
                                    capture_abort_error(
                                        &mut abort_error,
                                        e,
                                        "index_method_pre_commit_all failed during FAIL abort",
                                    );
                                }
                                loop {
                                    match self.commit_txn(
                                        pager.clone(),
                                        state,
                                        mv_store.as_ref(),
                                        false,
                                    ) {
                                        Ok(IOResult::Done(_)) => break,
                                        Ok(IOResult::IO(io)) => {
                                            if let Err(e) = io.wait(pager.io.as_ref()) {
                                                capture_abort_error(
                                                    &mut abort_error,
                                                    e,
                                                    "IO error during FAIL commit in abort",
                                                );
                                                break;
                                            }
                                        }
                                        Err(e) => {
                                            capture_abort_error(
                                                &mut abort_error,
                                                e,
                                                "commit_txn failed during FAIL abort",
                                            );
                                            break;
                                        }
                                    }
                                }
                            }
                        }
                        _ => {
                            if owns_auto_txn {
                                self.rollback_current_txn(pager);
                            }
                        }
                    }
                    let last_change = match effective_resolve {
                        ResolveType::Fail => state.n_change.load(Ordering::SeqCst),
                        _ => 0,
                    };
                    self.connection.set_changes_without_total(last_change);
                }
                Some(LimboError::RaiseIgnore) => {
                    tracing::error!(
                        "BUG: RaiseIgnore reached abort() - should be caught by op_program"
                    );
                    debug_assert!(
                        false,
                        "RaiseIgnore should be caught by op_program, not reach abort"
                    );
                }
                _ => match state.auto_txn_cleanup {
                    TxnCleanup::RollbackTxn => {
                        self.rollback_current_txn(pager);
                    }
                    TxnCleanup::RollbackSavepoint => {
                        if owns_auto_txn {
                            self.rollback_current_txn(pager);
                        } else if err.is_none() && !pager.is_checkpointing() {
                            if let Err(end_stmt_err) = state.end_statement(
                                &self.connection,
                                pager,
                                EndStatement::RollbackSavepoint,
                            ) {
                                capture_abort_error(
                                    &mut abort_error,
                                    end_stmt_err,
                                    "Failed to rollback statement savepoint during abort",
                                );
                            }
                        }
                    }
                    TxnCleanup::None => {
                        if owns_auto_txn || (!self.connection.get_auto_commit() && err.is_some()) {
                            self.rollback_current_txn(pager);
                        }
                    }
                },
            }
        }
        if state.uses_subjournal {
            pager.stop_use_subjournal();
            state.uses_subjournal = false;
        }
        state.auto_txn_cleanup = TxnCleanup::None;
        if let Some(err) = abort_error {
            return Err(err);
        }
        Ok(())
    }

    fn rollback_current_txn(&self, pager: &Arc<Pager>) {
        self.connection.rollback_current_txn_state(pager, true);
    }

    pub fn is_trigger_subprogram(&self) -> bool {
        self.trigger.is_some() || self.is_subprogram
    }
}

impl Deref for Program {
    type Target = PreparedProgram;

    fn deref(&self) -> &PreparedProgram {
        &self.prepared
    }
}

pub(crate) fn make_record(
    registers: &[Register],
    start_reg: &usize,
    count: &usize,
) -> ImmutableRecord {
    let regs = &registers[*start_reg..*start_reg + *count];
    ImmutableRecord::from_registers(regs, regs.len())
}

/// Split a register slice into an immutable ref and a mutable ref at two distinct indices.
pub(crate) fn split_registers(
    registers: &mut [Register],
    src: usize,
    dst: usize,
) -> (&Register, &mut Register) {
    debug_assert_ne!(src, dst, "split_registers: src and dst must differ");
    if src < dst {
        let (left, right) = registers.split_at_mut(dst);
        (&left[src], &mut right[0])
    } else {
        let (left, right) = registers.split_at_mut(src);
        (&right[0], &mut left[dst])
    }
}

pub fn registers_to_ref_values<'a>(
    registers: &'a [Register],
) -> impl ExactSizeIterator<Item = ValueRef<'a>> {
    registers.iter().map(|reg| reg.get_value().as_ref())
}

#[instrument(skip(program), level = Level::DEBUG)]
fn trace_insn(program: &Program, addr: InsnReference, insn: &Insn) {
    tracing::trace!(
        "\n{}",
        explain::insn_to_str(
            program,
            addr,
            insn,
            String::new(),
            program
                .comments
                .iter()
                .find(|(offset, _)| *offset == addr)
                .map(|(_, comment)| comment)
                .copied()
        )
    );
}

pub trait FromValueRow<'a> {
    fn from_value(value: &'a Value) -> Result<Self>
    where
        Self: Sized + 'a;
}

impl<'a> FromValueRow<'a> for i64 {
    fn from_value(value: &'a Value) -> Result<Self> {
        match value {
            Value::Numeric(Numeric::Integer(i)) => Ok(*i),
            _ => Err(LimboError::ConversionError("Expected integer value".into())),
        }
    }
}

impl<'a> FromValueRow<'a> for f64 {
    fn from_value(value: &'a Value) -> Result<Self> {
        match value {
            Value::Numeric(Numeric::Float(f)) => Ok(f64::from(*f)),
            _ => Err(LimboError::ConversionError("Expected integer value".into())),
        }
    }
}

impl<'a> FromValueRow<'a> for String {
    fn from_value(value: &'a Value) -> Result<Self> {
        match value {
            Value::Text(s) => Ok(s.as_str().to_string()),
            _ => Err(LimboError::ConversionError("Expected text value".into())),
        }
    }
}

impl<'a> FromValueRow<'a> for &'a str {
    fn from_value(value: &'a Value) -> Result<Self> {
        match value {
            Value::Text(s) => Ok(s.as_str()),
            _ => Err(LimboError::ConversionError("Expected text value".into())),
        }
    }
}

impl<'a> FromValueRow<'a> for &'a Value {
    fn from_value(value: &'a Value) -> Result<Self> {
        Ok(value)
    }
}

impl Row {
    pub fn get<'a, T: FromValueRow<'a> + 'a>(&'a self, idx: usize) -> Result<T> {
        let value = unsafe {
            self.values
                .add(idx)
                .as_ref()
                .expect("row value pointer should be valid")
        };
        let value = match value {
            Register::Value(value) => value,
            _ => unreachable!("a row should be formed of values only"),
        };
        T::from_value(value)
    }

    pub fn get_value(&self, idx: usize) -> &Value {
        let value = unsafe {
            self.values
                .add(idx)
                .as_ref()
                .expect("row value pointer should be valid")
        };
        match value {
            Register::Value(value) => value,
            _ => unreachable!("a row should be formed of values only"),
        }
    }

    pub fn get_values(&self) -> impl Iterator<Item = &Value> {
        let values = unsafe { std::slice::from_raw_parts(self.values, self.count) };
        // This should be ownedvalues
        // TODO: add check for this
        values.iter().map(|v| v.get_value())
    }

    pub fn len(&self) -> usize {
        self.count
    }

    pub fn is_empty(&self) -> bool {
        self.count == 0
    }
}

/// Extension trait for `ValueIterator` that allows writing directly to a `Register`
/// without allocating intermediate `ValueRef` values.
pub trait ValueIteratorExt {
    /// Skips `n` elements and writes the value directly to the register.
    /// Returns `Some(Ok(()))` on success, `Some(Err(...))` on parse error,
    /// or `None` if there are fewer than `n+1` elements.
    fn nth_into_register(&mut self, n: usize, dest: &mut Register) -> Option<Result<()>>;
}

impl<'a> ValueIteratorExt for crate::types::ValueIterator<'a> {
    #[inline(always)]
    fn nth_into_register(&mut self, n: usize, dest: &mut Register) -> Option<Result<()>> {
        use crate::storage::sqlite3_ondisk::read_varint;
        use crate::types::{get_serial_type_size, Extendable, Text};

        let mut header = self.header_section_ref();
        let mut data = self.data_section_ref();

        // Skip n elements
        let mut data_sum = 0;
        for _ in 0..n {
            if header.is_empty() {
                return None;
            }

            let (serial_type, bytes_read) = match read_varint(header) {
                Ok(v) => v,
                Err(e) => return Some(Err(e)),
            };
            header = &header[bytes_read..];

            data_sum += match get_serial_type_size(serial_type) {
                Ok(size) => size,
                Err(e) => return Some(Err(e)),
            };
        }

        if data_sum > data.len() {
            return Some(Err(LimboError::Corrupt(
                "Data section too small for indicated serial type size".into(),
            )));
        }
        data = &data[data_sum..];

        // Read the serial type for the target element
        if header.is_empty() {
            return None;
        }

        let (serial_type, bytes_read) = match read_varint(header) {
            Ok(v) => v,
            Err(e) => return Some(Err(e)),
        };

        // Update iterator state
        self.set_header_section(&header[bytes_read..]);

        // Decode directly into register based on serial type
        match serial_type {
            // NULL
            0 => {
                self.set_data_section(data);
                dest.set_null();
            }
            // I8
            1 => {
                if unlikely(data.is_empty()) {
                    return Some(Err(LimboError::Corrupt("Invalid 1-byte int".into())));
                }
                self.set_data_section(&data[1..]);
                dest.set_int(data[0] as i8 as i64);
            }
            // I16
            2 => {
                if unlikely(data.len() < 2) {
                    return Some(Err(LimboError::Corrupt("Invalid 2-byte int".into())));
                }
                self.set_data_section(&data[2..]);
                dest.set_int(i16::from_be_bytes([data[0], data[1]]) as i64);
            }
            // I24
            3 => {
                if unlikely(data.len() < 3) {
                    return Some(Err(LimboError::Corrupt("Invalid 3-byte int".into())));
                }
                self.set_data_section(&data[3..]);
                let sign_extension = if data[0] <= 0x7F { 0 } else { 0xFF };
                dest.set_int(
                    i32::from_be_bytes([sign_extension, data[0], data[1], data[2]]) as i64,
                );
            }
            // I32
            4 => {
                if unlikely(data.len() < 4) {
                    return Some(Err(LimboError::Corrupt("Invalid 4-byte int".into())));
                }
                self.set_data_section(&data[4..]);
                dest.set_int(i32::from_be_bytes([data[0], data[1], data[2], data[3]]) as i64);
            }
            // I48
            5 => {
                if unlikely(data.len() < 6) {
                    return Some(Err(LimboError::Corrupt("Invalid 6-byte int".into())));
                }
                self.set_data_section(&data[6..]);
                let sign_extension = if data[0] <= 0x7F { 0 } else { 0xFF };
                dest.set_int(i64::from_be_bytes([
                    sign_extension,
                    sign_extension,
                    data[0],
                    data[1],
                    data[2],
                    data[3],
                    data[4],
                    data[5],
                ]));
            }
            // I64
            6 => {
                if unlikely(data.len() < 8) {
                    return Some(Err(LimboError::Corrupt("Invalid 8-byte int".into())));
                }
                self.set_data_section(&data[8..]);
                dest.set_int(i64::from_be_bytes([
                    data[0], data[1], data[2], data[3], data[4], data[5], data[6], data[7],
                ]));
            }
            // F64
            7 => {
                if unlikely(data.len() < 8) {
                    return Some(Err(LimboError::Corrupt("Invalid 8-byte float".into())));
                }
                self.set_data_section(&data[8..]);
                let val = f64::from_be_bytes([
                    data[0], data[1], data[2], data[3], data[4], data[5], data[6], data[7],
                ]);
                if let Some(nn) = NonNan::new(val) {
                    dest.set_float(nn);
                } else {
                    dest.set_null();
                }
            }
            // CONST_INT0
            8 => {
                self.set_data_section(data);
                dest.set_int(0);
            }
            // CONST_INT1
            9 => {
                self.set_data_section(data);
                dest.set_int(1);
            }
            // Reserved
            10 | 11 => {
                mark_unlikely();
                return Some(Err(LimboError::Corrupt(format!(
                    "Reserved serial type: {serial_type}"
                ))));
            }
            // BLOB (n >= 12 && n & 1 == 0)
            n if n >= 12 && n & 1 == 0 => {
                let content_size = ((n - 12) / 2) as usize;
                if unlikely(data.len() < content_size) {
                    return Some(Err(LimboError::Corrupt("Invalid Blob value".into())));
                }
                self.set_data_section(&data[content_size..]);
                let blob_data = &data[..content_size];
                match dest {
                    Register::Value(Value::Blob(existing_blob)) => {
                        existing_blob.do_extend(&blob_data);
                    }
                    _ => {
                        dest.set_blob(blob_data.to_vec());
                    }
                }
            }
            // TEXT (n >= 13 && n & 1 == 1)
            n if n >= 13 && n & 1 == 1 => {
                let content_size = ((n - 13) / 2) as usize;
                if unlikely(data.len() < content_size) {
                    return Some(Err(LimboError::Corrupt("Invalid Text value".into())));
                }
                self.set_data_section(&data[content_size..]);
                let text_data = &data[..content_size];
                // SAFETY: TEXT serial type contains valid UTF-8
                let text_str = if cfg!(debug_assertions) {
                    match std::str::from_utf8(text_data) {
                        Ok(s) => s,
                        Err(e) => {
                            return Some(Err(LimboError::InternalError(format!(
                                "Invalid UTF-8 in TEXT serial type: {e}"
                            ))));
                        }
                    }
                } else {
                    unsafe { std::str::from_utf8_unchecked(text_data) }
                };
                match dest {
                    Register::Value(Value::Text(existing_text)) => {
                        existing_text.do_extend(&text_str);
                    }
                    _ => {
                        dest.set_text(Text::new(text_str.to_string()));
                    }
                }
            }
            _ => {
                mark_unlikely();
                return Some(Err(LimboError::Corrupt(format!(
                    "Invalid serial type: {serial_type}"
                ))));
            }
        }

        Some(Ok(()))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::panic::{catch_unwind, AssertUnwindSafe};

    #[test]
    fn active_opcode_helpers_initialize_defaults() {
        let mut state = ProgramState::new(1, 0);

        assert!(matches!(state.active_op_state.state, ActiveOpState::None));
        assert!(matches!(
            state.active_op_state.column(),
            OpColumnState::Start
        ));
        state.active_op_state.clear();
        assert!(state.active_op_state.parse_schema().is_none());
    }

    #[test]
    fn active_opcode_helpers_reject_mismatched_resumes() {
        let mut state = ProgramState::new(1, 0);
        *state.active_op_state.column() = OpColumnState::GetColumn;

        let panic = catch_unwind(AssertUnwindSafe(|| {
            let _ = state.active_op_state.parse_schema();
        }));
        assert!(panic.is_err(), "mismatched opcode resume should panic");
    }

    #[test]
    fn seek_state_is_independent_from_active_opcode_slot() {
        let mut state = ProgramState::new(1, 0);

        *state.active_op_state.insert() = OpInsertState {
            sub_state: OpInsertSubState::Seek,
            old_record: None,
            is_noop_update: false,
        };
        state.seek_state = OpSeekState::MoveLast;

        assert!(matches!(
            state.active_op_state.insert().sub_state,
            OpInsertSubState::Seek
        ));
        assert!(matches!(state.seek_state, OpSeekState::MoveLast));
    }
}

/// Shuttle tests for validating the `unsafe impl Send + Sync for ProgramState` safety claims.
///
/// The safety claims are:
/// 1. `Row` contains a `*const Register` pointing into `ProgramState.registers`
/// 2. Only immutable references (`&Row`) are given out via `result_row.as_ref()`
/// 3. `result_row` is invalidated (via `.take()`) at the start of each step iteration
///
/// These tests verify that the implementation correctly upholds these invariants
/// under concurrent access patterns.

#[cfg(all(shuttle, test))]
mod shuttle_tests {
    use super::*;
    use crate::sync::Arc;
    use crate::thread;
    use crate::types::Value;

    /// Creates a minimal ProgramState for testing.
    fn create_test_state(num_registers: usize, num_cursors: usize) -> ProgramState {
        ProgramState::new(num_registers, num_cursors)
    }

    /// Test that ProgramState can be safely sent between threads.
    /// This validates the `unsafe impl Send for ProgramState` claim.
    #[test]
    fn shuttle_program_state_send() {
        shuttle::check_random(
            || {
                let mut state = create_test_state(10, 2);

                // Write some data to registers
                state.registers[0].set_int(42);
                state.registers[1].set_text(Text::new("test".to_string()));

                // Send state to another thread
                let handle = thread::spawn(move || {
                    // Verify data is intact after send
                    assert!(matches!(
                        &state.registers[0],
                        Register::Value(Value::Numeric(Numeric::Integer(42)))
                    ));
                    if let Register::Value(Value::Text(t)) = &state.registers[1] {
                        assert_eq!(t.as_str(), "test");
                    } else {
                        panic!("Expected text value");
                    }

                    // Modify in new thread
                    state.registers[2].set_int(100);
                    state
                });

                let state = handle.join().unwrap();
                assert!(matches!(
                    &state.registers[2],
                    Register::Value(Value::Numeric(Numeric::Integer(100)))
                ));
            },
            1000,
        );
    }

    /// Test that ProgramState with a set result_row can be safely sent.
    /// The Row contains a raw pointer that must remain valid after the send.
    #[test]
    fn shuttle_program_state_send_with_row() {
        shuttle::check_random(
            || {
                let mut state = create_test_state(10, 2);

                // Set up registers with test data
                state.registers[0].set_int(1);
                state.registers[1].set_int(2);
                state.registers[2].set_int(3);

                // Create a result_row pointing to registers
                state.result_row = Some(Row {
                    values: &state.registers[0] as *const Register,
                    count: 3,
                });

                // Send to another thread - the pointer must remain valid
                // because it points to memory owned by state (the registers Vec)
                let handle = thread::spawn(move || {
                    // The row pointer should still be valid because registers moved with state
                    if let Some(row) = &state.result_row {
                        assert_eq!(row.len(), 3);
                        // Read through the pointer - this validates the pointer is still valid
                        let val = row.get::<i64>(0).unwrap();
                        assert_eq!(val, 1);
                        let val = row.get::<i64>(1).unwrap();
                        assert_eq!(val, 2);
                        let val = row.get::<i64>(2).unwrap();
                        assert_eq!(val, 3);
                    } else {
                        panic!("Expected result_row to be set");
                    }
                    state
                });

                let _ = handle.join().unwrap();
            },
            1000,
        );
    }

    /// Test concurrent reads of result_row through shared reference.
    /// This validates the `unsafe impl Sync for ProgramState` claim for read access.
    #[test]
    fn shuttle_program_state_sync_concurrent_reads() {
        shuttle::check_random(
            || {
                let mut state = create_test_state(10, 2);

                // Set up registers
                state.registers[0].set_int(42);
                state.registers[1].set_int(43);

                // Create result_row
                state.result_row = Some(Row {
                    values: &state.registers[0] as *const Register,
                    count: 2,
                });

                let state = Arc::new(state);
                let state2 = Arc::clone(&state);
                let state3 = Arc::clone(&state);

                // Multiple threads reading concurrently
                let h1 = thread::spawn(move || {
                    if let Some(row) = &state.result_row {
                        let val = row.get::<i64>(0).unwrap();
                        assert_eq!(val, 42);
                    }
                });

                let h2 = thread::spawn(move || {
                    if let Some(row) = &state2.result_row {
                        let val = row.get::<i64>(1).unwrap();
                        assert_eq!(val, 43);
                    }
                });

                let h3 = thread::spawn(move || {
                    if let Some(row) = &state3.result_row {
                        assert_eq!(row.len(), 2);
                    }
                });

                h1.join().unwrap();
                h2.join().unwrap();
                h3.join().unwrap();
            },
            1000,
        );
    }

    /// Test that Row values read through the pointer are consistent.
    /// Multiple threads reading the same row values should see the same data.
    #[test]
    fn shuttle_row_pointer_consistency() {
        shuttle::check_random(
            || {
                let mut state = create_test_state(10, 2);

                // Set up registers with distinct values
                for i in 0..5 {
                    state.registers[i].set_int(i as i64 * 10);
                }

                state.result_row = Some(Row {
                    values: &state.registers[0] as *const Register,
                    count: 5,
                });

                let state = Arc::new(state);
                let mut handles = vec![];

                for _ in 0..4 {
                    let state_clone = Arc::clone(&state);
                    let h = thread::spawn(move || {
                        if let Some(row) = &state_clone.result_row {
                            // All threads should see the same values
                            for i in 0..5 {
                                let val = row.get::<i64>(i).unwrap();
                                assert_eq!(val, i as i64 * 10);
                            }
                        }
                    });
                    handles.push(h);
                }

                for h in handles {
                    h.join().unwrap();
                }
            },
            1000,
        );
    }

    /// Test the result_row invalidation pattern.
    /// When result_row is taken (invalidated), concurrent reads should not see stale data.
    /// This simulates the pattern used in `normal_step()` where `result_row.take()` is called.
    #[test]
    fn shuttle_result_row_invalidation() {
        shuttle::check_random(
            || {
                let mut state = create_test_state(10, 2);

                state.registers[0].set_int(100);
                state.result_row = Some(Row {
                    values: &state.registers[0] as *const Register,
                    count: 1,
                });

                // Simulate the invalidation pattern from normal_step
                // In real code, this requires &mut self, so there's no concurrent access
                let taken_row = state.result_row.take();

                // After take(), result_row should be None
                assert!(state.result_row.is_none());

                // The taken row still holds valid data (until dropped)
                if let Some(row) = taken_row {
                    let val = row.get::<i64>(0).unwrap();
                    assert_eq!(val, 100);
                }
            },
            1000,
        );
    }

    /// Test register modification after row invalidation.
    /// This validates that modifying registers after take() is safe.
    #[test]
    fn shuttle_register_modification_after_invalidation() {
        shuttle::check_random(
            || {
                let mut state = create_test_state(10, 2);

                state.registers[0].set_int(1);
                state.result_row = Some(Row {
                    values: &state.registers[0] as *const Register,
                    count: 1,
                });

                // Invalidate row (simulating what normal_step does)
                let _ = state.result_row.take();

                // Now safe to modify registers
                state.registers[0].set_int(999);

                // Create new row pointing to modified registers
                state.result_row = Some(Row {
                    values: &state.registers[0] as *const Register,
                    count: 1,
                });

                // New row should see new value
                if let Some(row) = &state.result_row {
                    let val = row.get::<i64>(0).unwrap();
                    assert_eq!(val, 999);
                }
            },
            1000,
        );
    }

    /// Test sequential send-receive pattern (simulating async task scheduling).
    /// ProgramState is moved between threads in a producer-consumer pattern.
    #[test]
    fn shuttle_sequential_thread_transfer() {
        shuttle::check_random(
            || {
                let mut state = create_test_state(10, 2);
                state.registers[0].set_int(0);

                // Thread 1: increment
                let h1 = thread::spawn(move || {
                    if let Register::Value(Value::Numeric(Numeric::Integer(v))) =
                        &state.registers[0]
                    {
                        state.registers[0].set_int(v + 1);
                    }
                    state
                });

                let mut state = h1.join().unwrap();

                // Thread 2: increment
                let h2 = thread::spawn(move || {
                    if let Register::Value(Value::Numeric(Numeric::Integer(v))) =
                        &state.registers[0]
                    {
                        state.registers[0].set_int(v + 1);
                    }
                    state
                });

                let mut state = h2.join().unwrap();

                // Thread 3: increment
                let h3 = thread::spawn(move || {
                    if let Register::Value(Value::Numeric(Numeric::Integer(v))) =
                        &state.registers[0]
                    {
                        state.registers[0].set_int(v + 1);
                    }
                    state
                });

                let state = h3.join().unwrap();

                // Final value should be 3
                assert!(matches!(
                    &state.registers[0],
                    Register::Value(Value::Numeric(Numeric::Integer(3)))
                ));
            },
            1000,
        );
    }

    /// Test that ProgramState can be wrapped in Arc for shared ownership.
    /// This is the typical pattern for concurrent database operations.
    #[test]
    fn shuttle_arc_wrapped_state() {
        shuttle::check_random(
            || {
                let mut state = create_test_state(10, 2);

                // Initialize with test data
                for i in 0..5 {
                    state.registers[i].set_int(i as i64);
                }

                let state = Arc::new(state);
                let mut handles = vec![];

                // Multiple threads reading registers through Arc
                for thread_id in 0u8..4 {
                    let state_clone = Arc::clone(&state);
                    let h = thread::spawn(move || {
                        // Each thread reads all registers
                        for i in 0..5 {
                            if let Register::Value(Value::Numeric(Numeric::Integer(v))) =
                                &state_clone.registers[i]
                            {
                                assert_eq!(*v, i as i64);
                            }
                        }
                        thread_id
                    });
                    handles.push(h);
                }

                for h in handles {
                    h.join().unwrap();
                }
            },
            1000,
        );
    }

    /// Test Row::get_values iterator under concurrent access.
    #[test]
    fn shuttle_row_get_values_concurrent() {
        shuttle::check_random(
            || {
                let mut state = create_test_state(10, 2);

                state.registers[0].set_int(10);
                state.registers[1].set_int(20);
                state.registers[2].set_int(30);

                state.result_row = Some(Row {
                    values: &state.registers[0] as *const Register,
                    count: 3,
                });

                let state = Arc::new(state);
                let state2 = Arc::clone(&state);

                let h1 = thread::spawn(move || {
                    if let Some(row) = &state.result_row {
                        let values: Vec<_> = row.get_values().collect();
                        assert_eq!(values.len(), 3);
                    }
                });

                let h2 = thread::spawn(move || {
                    if let Some(row) = &state2.result_row {
                        let mut sum = 0i64;
                        for val in row.get_values() {
                            if let Value::Numeric(Numeric::Integer(i)) = val {
                                sum += i;
                            }
                        }
                        assert_eq!(sum, 60); // 10 + 20 + 30
                    }
                });

                h1.join().unwrap();
                h2.join().unwrap();
            },
            1000,
        );
    }

    /// Stress test: Many threads reading from shared ProgramState.
    #[test]
    fn shuttle_stress_concurrent_reads() {
        shuttle::check_random(
            || {
                let mut state = create_test_state(20, 2);

                // Fill registers with identifiable data
                for i in 0..20 {
                    state.registers[i].set_int(i as i64 * 100);
                }

                state.result_row = Some(Row {
                    values: &state.registers[0] as *const Register,
                    count: 20,
                });

                let state = Arc::new(state);
                let mut handles = vec![];

                for thread_id in 0..6u8 {
                    let state_clone = Arc::clone(&state);
                    let h = thread::spawn(move || {
                        // Each thread reads different parts
                        let start = (thread_id as usize * 3) % 20;
                        if let Some(row) = &state_clone.result_row {
                            for i in 0..3 {
                                let idx = (start + i) % row.len();
                                let val = row.get::<i64>(idx).unwrap();
                                assert_eq!(val, idx as i64 * 100);
                            }
                        }
                        thread_id
                    });
                    handles.push(h);
                }

                for h in handles {
                    h.join().unwrap();
                }
            },
            1000,
        );
    }
}