sqawk 0.8.2

//! SQL VM engine that executes bytecode instructions
//!
//! This module implements the virtual machine that executes bytecode instructions
//! generated by the compiler. It maintains internal state like registers, cursors,
//! and the program counter during execution.

use super::bytecode::{
    Instruction, OpCode, Program, Register, AGG_AVG, AGG_COUNT, AGG_DISTINCT, AGG_MAX, AGG_MIN,
    AGG_SUM, AGG_TYPE_MASK,
};
use crate::capacity::{
    DEFAULT_ACCUMULATOR_CAPACITY, DEFAULT_CURSOR_CAPACITY, DEFAULT_MODIFICATIONS_CAPACITY,
    DEFAULT_REGISTER_CAPACITY, DEFAULT_RESULT_CAPACITY, DEFAULT_SORTER_CAPACITY,
    DEFAULT_SORTER_ROWS_CAPACITY, DEFAULT_SORT_KEYS_CAPACITY,
};
use crate::database::Database;
use crate::error::{SqawkError, SqawkResult};
use crate::table::{ColumnDefinition, Table, Value};
use std::collections::HashMap;

/// Data backing a cursor - either borrowed table or owned ephemeral data
/// Following SQLite's model where Column/Next/Rewind work for ALL cursor types
enum CursorData<'a> {
    /// Table cursor - borrows reference to table
    Table { table: &'a Table },
    /// Ephemeral cursor - owns its data
    Ephemeral {
        rows: Vec<Vec<Value>>,
        col_count: usize,
        sort_keys: Vec<(usize, bool)>, // (col_idx, ascending)
        sorted: bool,
        sequence: i64,
        name: Option<String>, // Optional name for pending tables
    },
}

/// Unified cursor for VM execution
///
/// Works with both table references and ephemeral data.
/// This follows SQLite's model where Column/Next/Rewind work for any cursor.
struct Cursor<'a> {
    data: CursorData<'a>,
    /// Current row position (0-based)
    position: usize,
    /// Whether the cursor is at a valid row
    valid: bool,
}

impl<'a> Cursor<'a> {
    /// Create cursor from table reference
    fn from_table(table: &'a Table) -> Self {
        Self {
            data: CursorData::Table { table },
            position: 0,
            valid: false,
        }
    }

    /// Create ephemeral cursor
    fn ephemeral(col_count: usize, sort_keys: Vec<(usize, bool)>) -> Self {
        Self {
            data: CursorData::Ephemeral {
                rows: Vec::with_capacity(DEFAULT_SORTER_ROWS_CAPACITY),
                col_count,
                sort_keys,
                sorted: false,
                sequence: 0,
                name: None,
            },
            position: 0,
            valid: false,
        }
    }

    /// Create named ephemeral cursor (for pending tables)
    fn named_ephemeral(col_count: usize, table_name: String) -> Self {
        Self {
            data: CursorData::Ephemeral {
                rows: Vec::with_capacity(DEFAULT_SORTER_ROWS_CAPACITY),
                col_count,
                sort_keys: Vec::with_capacity(DEFAULT_SORT_KEYS_CAPACITY),
                sorted: false,
                sequence: 0,
                name: Some(table_name),
            },
            position: 0,
            valid: false,
        }
    }

    /// Move to the first row (Rewind opcode)
    fn rewind(&mut self) -> bool {
        self.position = 0;
        self.valid = match &self.data {
            CursorData::Table { table } => self.position < table.row_count(),
            CursorData::Ephemeral { rows, .. } => !rows.is_empty(),
        };
        self.valid
    }

    /// Move to the next row (Next opcode)
    fn next(&mut self) -> bool {
        if !self.valid {
            return false;
        }

        self.position += 1;
        self.valid = match &self.data {
            CursorData::Table { table } => self.position < table.row_count(),
            CursorData::Ephemeral { rows, .. } => self.position < rows.len(),
        };
        self.valid
    }

    /// Get the current row index (for index-based results)
    fn current_row_idx(&self) -> Option<usize> {
        if self.valid {
            Some(self.position)
        } else {
            None
        }
    }

    /// Get a column value at the current position (Column opcode)
    fn column(&self, idx: usize) -> Option<Value> {
        if !self.valid {
            return None;
        }
        match &self.data {
            CursorData::Table { table } => {
                if idx >= table.column_count() {
                    return None;
                }
                let rows = table.rows();
                if self.position < rows.len() {
                    rows[self.position].get(idx).cloned()
                } else {
                    None
                }
            }
            CursorData::Ephemeral {
                rows, col_count, ..
            } => {
                if idx >= *col_count {
                    return None;
                }
                rows.get(self.position)?.get(idx).cloned()
            }
        }
    }

    /// Get the table name (for table cursors or named ephemeral cursors)
    fn table_name(&self) -> &str {
        match &self.data {
            CursorData::Table { table } => table.name(),
            CursorData::Ephemeral { name: Some(n), .. } => n,
            CursorData::Ephemeral { name: None, .. } => "<ephemeral>",
        }
    }

    /// Get the current row index (alias for current_row_idx)
    fn current_row_index(&self) -> Option<usize> {
        self.current_row_idx()
    }

    /// Insert row into ephemeral cursor (IdxInsert opcode)
    fn insert_row(&mut self, row: Vec<Value>) -> Result<(), SqawkError> {
        match &mut self.data {
            CursorData::Ephemeral { rows, sorted, .. } => {
                rows.push(row);
                *sorted = false;
                Ok(())
            }
            CursorData::Table { .. } => Err(SqawkError::VmError(
                "Cannot insert into table cursor".into(),
            )),
        }
    }

    /// Sort ephemeral cursor (Sort opcode)
    /// Returns true if has rows (continue), false if empty (should jump)
    fn sort(&mut self) -> Result<bool, SqawkError> {
        match &mut self.data {
            CursorData::Ephemeral {
                rows,
                sort_keys,
                sorted,
                ..
            } => {
                if rows.is_empty() {
                    return Ok(false); // Empty - caller should jump
                }
                if !*sorted {
                    let keys = sort_keys.clone();
                    rows.sort_by(|a, b| {
                        for (col_idx, ascending) in &keys {
                            let cmp = compare_values(
                                a.get(*col_idx).unwrap_or(&Value::Null),
                                b.get(*col_idx).unwrap_or(&Value::Null),
                            );
                            if cmp != std::cmp::Ordering::Equal {
                                return if *ascending { cmp } else { cmp.reverse() };
                            }
                        }
                        std::cmp::Ordering::Equal
                    });
                    *sorted = true;
                }
                self.position = 0;
                self.valid = true;
                Ok(true) // Has rows - caller continues
            }
            CursorData::Table { .. } => Err(SqawkError::VmError("Cannot sort table cursor".into())),
        }
    }

    /// Generate next sequence number (Sequence opcode)
    fn next_sequence(&mut self) -> Result<i64, SqawkError> {
        match &mut self.data {
            CursorData::Ephemeral { sequence, .. } => {
                let seq = *sequence;
                *sequence += 1;
                Ok(seq)
            }
            CursorData::Table { .. } => Err(SqawkError::VmError(
                "Cannot generate sequence from table cursor".into(),
            )),
        }
    }

    /// Get row count (for debugging/validation)
    #[allow(dead_code)]
    fn row_count(&self) -> usize {
        match &self.data {
            CursorData::Table { table } => table.row_count(),
            CursorData::Ephemeral { rows, .. } => rows.len(),
        }
    }

    /// Check if this is an ephemeral cursor
    #[allow(dead_code)]
    fn is_ephemeral(&self) -> bool {
        matches!(self.data, CursorData::Ephemeral { .. })
    }

    /// Get column count
    fn column_count(&self) -> usize {
        match &self.data {
            CursorData::Table { table } => table.column_count(),
            CursorData::Ephemeral { col_count, .. } => *col_count,
        }
    }

    /// Get rows reference (for table cursors, returns table rows; for ephemeral, returns owned rows)
    fn rows(&self) -> &[Vec<Value>] {
        match &self.data {
            CursorData::Table { table } => table.rows(),
            CursorData::Ephemeral { rows, .. } => rows,
        }
    }

    /// Get a column value at a specific row offset from current position
    /// Positive offset = forward (for LEAD), negative offset = backward (for LAG)
    /// Returns None if the offset position is out of bounds
    fn column_at_offset(&self, col_idx: usize, offset: i64) -> Option<Value> {
        if !self.valid {
            return None;
        }

        let target_pos = (self.position as i64) + offset;
        if target_pos < 0 {
            return None;
        }

        let target_pos = target_pos as usize;
        match &self.data {
            CursorData::Table { table } => {
                if col_idx >= table.column_count() || target_pos >= table.row_count() {
                    return None;
                }
                let rows = table.rows();
                rows.get(target_pos)?.get(col_idx).cloned()
            }
            CursorData::Ephemeral {
                rows, col_count, ..
            } => {
                if col_idx >= *col_count || target_pos >= rows.len() {
                    return None;
                }
                rows.get(target_pos)?.get(col_idx).cloned()
            }
        }
    }
}

/// Round `num` to `digits` places after the decimal point.
///
/// Negative `digits` rounds to the left of the point, so
/// `round_to(1234.5, -2)` is `1200.0`. A magnitude large enough to push the
/// scale factor out of f64 range leaves the value alone: at that many digits
/// it is already exact.
fn round_to(num: f64, digits: i64) -> f64 {
    if digits == 0 {
        return num.round();
    }
    // Outside this range the scale factor is not representable, and the value
    // is already exact at that many digits either way.
    if !(-308..=308).contains(&digits) {
        return num;
    }
    let factor = 10f64.powi(digits as i32);
    let scaled = num * factor;
    if !scaled.is_finite() {
        return num;
    }
    scaled.round() / factor
}

/// Compare two Values for ordering
fn compare_values(a: &Value, b: &Value) -> std::cmp::Ordering {
    use std::cmp::Ordering;
    match (a, b) {
        (Value::Null, Value::Null) => Ordering::Equal,
        (Value::Null, _) => Ordering::Less,
        (_, Value::Null) => Ordering::Greater,
        (Value::Integer(a), Value::Integer(b)) => a.cmp(b),
        (Value::Float(a), Value::Float(b)) => a.partial_cmp(b).unwrap_or(Ordering::Equal),
        (Value::Integer(a), Value::Float(b)) => {
            (*a as f64).partial_cmp(b).unwrap_or(Ordering::Equal)
        }
        (Value::Float(a), Value::Integer(b)) => {
            a.partial_cmp(&(*b as f64)).unwrap_or(Ordering::Equal)
        }
        (Value::String(a), Value::String(b)) => a.cmp(b),
        (Value::Boolean(a), Value::Boolean(b)) => a.cmp(b),
        _ => format!("{:?}", a).cmp(&format!("{:?}", b)),
    }
}

/// Sorter for ORDER BY operations
struct Sorter {
    rows: Vec<Vec<Value>>,
    sort_keys: Vec<(usize, bool)>, // (column_index, ascending)
    position: usize,
    sorted: bool,
}

/// Aggregate accumulator for GROUP BY operations
#[derive(Debug, Clone)]
struct AggAccumulator {
    /// Function type: 0=COUNT, 1=SUM, 2=AVG, 3=MIN, 4=MAX
    func_type: i64,
    /// Count of values processed (for COUNT and AVG)
    count: i64,
    /// Accumulated value (sum for SUM/AVG, min/max for MIN/MAX)
    value: Option<Value>,
    /// Values already accumulated, for DISTINCT aggregates. `None` when the
    /// aggregate is not DISTINCT, so the ordinary path allocates nothing.
    seen: Option<std::collections::HashSet<String>>,
}

impl Sorter {
    fn new(_col_count: usize, sort_keys: Vec<(usize, bool)>) -> Self {
        Self {
            rows: Vec::with_capacity(DEFAULT_SORTER_ROWS_CAPACITY),
            sort_keys,
            position: 0,
            sorted: false,
        }
    }

    fn insert(&mut self, row: Vec<Value>) {
        self.rows.push(row);
        self.sorted = false;
    }

    fn sort(&mut self) {
        let keys = self.sort_keys.clone();
        self.rows.sort_by(|a, b| {
            for (col_idx, ascending) in &keys {
                if *col_idx >= a.len() || *col_idx >= b.len() {
                    continue;
                }
                let cmp = compare_values(&a[*col_idx], &b[*col_idx]);
                if cmp != std::cmp::Ordering::Equal {
                    return if *ascending { cmp } else { cmp.reverse() };
                }
            }
            std::cmp::Ordering::Equal
        });
        self.sorted = true;
        self.position = 0;
    }

    fn current_row(&self) -> Option<&Vec<Value>> {
        if self.sorted && self.position < self.rows.len() {
            Some(&self.rows[self.position])
        } else {
            None
        }
    }

    fn next(&mut self) -> bool {
        if !self.sorted {
            return false;
        }
        self.position += 1;
        self.position < self.rows.len()
    }
}

#[derive(Debug, Clone, Copy, PartialEq)]
enum TransactionState {
    /// No active transaction - the default state when no transaction has been started
    None,
    /// Transaction in progress - changes are being tracked but not yet permanent
    Active,
    /// Transaction has been committed - changes have been permanently applied
    Committed,
    /// Transaction has been rolled back - changes have been discarded
    RolledBack,
}

/// Column definition for CREATE TABLE
#[derive(Debug, Clone)]
pub struct ColumnDef {
    pub name: String,
    pub data_type: String,
}

/// Pending modification to a table
#[derive(Debug, Clone)]
pub enum TableModification {
    Insert {
        table_name: String,
        row: Vec<Value>,
    },
    Delete {
        table_name: String,
        row_index: usize,
    },
    /// Replace a row in place, preserving its position.
    ///
    /// UPDATE used to be expressed as Delete plus Insert, and the insert
    /// appends, so every updated row jumped to the end of the table -- and
    /// with --write that reordering was written to the user's file.
    Replace {
        table_name: String,
        row_index: usize,
        row: Vec<Value>,
    },
    CreateTable {
        table_name: String,
        columns: Vec<ColumnDef>,
        file_path: Option<String>,
        delimiter: Option<String>,
    },
    DropTable {
        table_name: String,
    },
    AlterTableAddColumn {
        table_name: String,
        column_name: String,
        column_type: String,
    },
    Truncate {
        table_name: String,
    },
}

/// SQL VM engine that executes bytecode
pub struct VmEngine<'a> {
    database: &'a Database,
    program: Program,
    pc: usize,
    registers: Vec<Register>,
    cursors: HashMap<usize, Cursor<'a>>,
    sorters: HashMap<usize, Sorter>,
    accumulators: HashMap<usize, AggAccumulator>,
    results: Vec<Vec<Value>>,

    /// Index-based results (Phase 4C parallel path)
    /// Each entry contains row indices from each cursor for that result row
    /// Option<usize> handles NULL rows in outer joins
    #[allow(dead_code)]
    result_row_indices: Vec<Vec<Option<usize>>>,

    /// Order of cursor IDs as sources (maps position to cursor ID)
    #[allow(dead_code)]
    cursor_source_order: Vec<usize>,

    /// Index into `results` at which each window partition begins.
    ///
    /// Recorded during the streaming window pass so a following WindowFinalize
    /// can identify each partition's extent without needing the partition key
    /// to be a projected column.
    window_partition_starts: Vec<usize>,

    /// Result of the most recent Compare, consumed by Jump.
    ///
    /// Mirrors SQLite, where OP_Compare leaves its verdict for a following
    /// OP_Jump rather than materializing it into a register.
    compare_flag: std::cmp::Ordering,

    /// Current transaction state (None, Active, Committed, or RolledBack)
    /// Tracks the lifecycle of a transaction and enforces proper operation sequencing
    transaction_state: TransactionState,

    /// Pending modifications to be applied after execution
    pending_modifications: Vec<TableModification>,

    /// Whether the engine is in verbose mode
    verbose: bool,

    /// Once-flag tracking - maps instruction address to has_executed flag
    /// Used by the Once opcode to skip repeated initialization
    once_flags: HashMap<usize, bool>,
}

impl<'a> VmEngine<'a> {
    /// Create a new VM engine
    pub fn new(database: &'a Database, verbose: bool) -> Self {
        Self {
            database,
            program: Program::new(),
            pc: 0,
            compare_flag: std::cmp::Ordering::Equal,
            window_partition_starts: Vec::new(),
            registers: Vec::with_capacity(DEFAULT_REGISTER_CAPACITY),
            cursors: HashMap::with_capacity(DEFAULT_CURSOR_CAPACITY),
            sorters: HashMap::with_capacity(DEFAULT_SORTER_CAPACITY),
            accumulators: HashMap::with_capacity(DEFAULT_ACCUMULATOR_CAPACITY),
            results: Vec::with_capacity(DEFAULT_RESULT_CAPACITY),
            result_row_indices: Vec::with_capacity(DEFAULT_RESULT_CAPACITY),
            cursor_source_order: Vec::with_capacity(DEFAULT_CURSOR_CAPACITY),
            transaction_state: TransactionState::None,
            pending_modifications: Vec::with_capacity(DEFAULT_MODIFICATIONS_CAPACITY),
            verbose,
            once_flags: HashMap::with_capacity(DEFAULT_CURSOR_CAPACITY),
        }
    }

    /// Alias for new() - both work with immutable database reference
    /// Write operations are accumulated and returned via take_modifications()
    pub fn new_mut(database: &'a Database, verbose: bool) -> Self {
        Self::new(database, verbose)
    }

    /// Initialize the VM with a program to execute
    pub fn init(&mut self, program: Program) {
        self.program = program;
        self.pc = 0;
        self.registers = Vec::with_capacity(DEFAULT_REGISTER_CAPACITY);
        self.cursors.clear();
        self.sorters.clear();
        self.accumulators.clear();
        self.results.clear();
        self.result_row_indices.clear();
        self.cursor_source_order.clear();
        self.pending_modifications.clear();
        self.once_flags.clear();
        self.window_partition_starts.clear();

        // Size the register file from the compiler's allocation count.
        //
        // This used to be inferred as max(p1, p2, p3) over the instruction
        // stream, capped at 10000. That was wrong in both directions: those
        // operands hold literals as often as register numbers, so `Integer
        // 999999 -> r` inflated the file to the cap, while a program that
        // genuinely needed more than 10000 registers got a short one -- and
        // Column errors on a short file rather than growing it.
        //
        // Hand-assembled test programs carry no count, so fall back to the old
        // scan when register_count is 0 but instructions exist.
        let reg_count = if self.program.register_count > 0 {
            self.program.register_count
        } else {
            self.program
                .instructions
                .iter()
                .map(|i| i.p1.max(i.p2).max(i.p3))
                .max()
                .unwrap_or(10)
                .min(10000)
        };
        self.registers = vec![Register::Null; (reg_count + 5) as usize];

        // Reset transaction state
        self.transaction_state = TransactionState::None;

        if self.verbose {
            println!("VM initialized with program:\n{}", self.program);
        }
    }

    /// Take pending modifications out of the engine
    pub fn take_modifications(&mut self) -> Vec<TableModification> {
        std::mem::take(&mut self.pending_modifications)
    }

    /// Execute the current program
    pub fn execute(&mut self) -> SqawkResult<()> {
        if self.program.is_empty() {
            return Err(SqawkError::VmError("No program to execute".to_string()));
        }

        // Start execution at PC 0
        self.pc = 0;

        // Main execution loop
        loop {
            // Get current instruction
            let inst_clone = self
                .program
                .get(self.pc)
                .ok_or_else(|| {
                    SqawkError::VmError(format!("Invalid program counter: {}", self.pc))
                })?
                .clone(); // Clone the instruction to avoid borrowing issues

            if self.verbose {
                println!("Executing [{}]: {}", self.pc, inst_clone);
            }

            // Execute the instruction
            match self.execute_instruction(&inst_clone)? {
                ExecuteResult::Continue => {
                    // Move to next instruction
                    self.pc += 1;
                }
                ExecuteResult::Jump(addr) => {
                    // Jump to specified address
                    self.pc = addr;
                }
                ExecuteResult::Halt => {
                    // End execution
                    break;
                }
            }
        }

        Ok(())
    }

    /// Execute a single instruction
    fn execute_instruction(&mut self, inst: &Instruction) -> SqawkResult<ExecuteResult> {
        match inst.opcode {
            OpCode::Init => {
                // Initialize VM and jump to start address in P2
                Ok(ExecuteResult::Jump(inst.p2 as usize))
            }

            OpCode::Goto => {
                // Unconditional jump to address in P2
                Ok(ExecuteResult::Jump(inst.p2 as usize))
            }

            OpCode::Halt => {
                // End execution
                Ok(ExecuteResult::Halt)
            }

            OpCode::OpenRead => {
                // Open a cursor for reading a table
                let cursor_idx = inst.p1 as usize;
                let table_name = inst.p4.as_deref().unwrap_or("");

                // Check if the table exists
                if !self.database.has_table(table_name) {
                    return Err(SqawkError::TableNotFound(table_name.to_string()));
                }

                // Get the table from the database (safe to unwrap since we checked it exists)
                // Note: We now borrow the table instead of cloning it (Phase 4B optimization)
                let table = self.database.get_table(table_name).unwrap();

                // Create a cursor for the table (borrows, no clone)
                let cursor = Cursor::from_table(table);
                self.cursors.insert(cursor_idx, cursor);

                // Track cursor source order for index-based results (Phase 4C)
                if !self.cursor_source_order.contains(&cursor_idx) {
                    self.cursor_source_order.push(cursor_idx);
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::OpenWrite => {
                // Open a cursor for writing to a table
                // Note: Actual modifications are queued and applied after execution
                let cursor_idx = inst.p1 as usize;
                let table_name = inst.p4.as_deref().unwrap_or("");

                // Check if the table exists in database or pending modifications
                if self.database.has_table(table_name) {
                    // Get the table from the database
                    let table = self.database.get_table(table_name).unwrap();
                    let cursor = Cursor::from_table(table);
                    self.cursors.insert(cursor_idx, cursor);
                } else {
                    // Check if there's a pending CreateTable for this table
                    let pending_create = self.pending_modifications.iter().find(|m| {
                        matches!(m, TableModification::CreateTable { table_name: tn, .. } if tn == table_name)
                    });

                    if let Some(TableModification::CreateTable {
                        columns,
                        table_name: pending_name,
                        ..
                    }) = pending_create.cloned()
                    {
                        // Create a named ephemeral cursor with the schema from pending CreateTable
                        let col_count = columns.len();
                        let cursor = Cursor::named_ephemeral(col_count, pending_name);
                        self.cursors.insert(cursor_idx, cursor);
                    } else {
                        return Err(SqawkError::TableNotFound(table_name.to_string()));
                    }
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Close => {
                // Close a cursor
                let cursor_idx = inst.p1 as usize;
                self.cursors.remove(&cursor_idx);

                Ok(ExecuteResult::Continue)
            }

            OpCode::Rewind => {
                // Rewind a cursor to the first row
                let cursor_idx = inst.p1 as usize;
                let jump_addr = inst.p2 as usize;

                if let Some(cursor) = self.cursors.get_mut(&cursor_idx) {
                    if cursor.rewind() {
                        // Found at least one row
                        Ok(ExecuteResult::Continue)
                    } else {
                        // Empty table, jump
                        Ok(ExecuteResult::Jump(jump_addr))
                    }
                } else {
                    Err(SqawkError::VmError(format!(
                        "Invalid cursor: {}",
                        cursor_idx
                    )))
                }
            }

            OpCode::Next => {
                // Move cursor to next row
                let cursor_idx = inst.p1 as usize;
                let jump_addr = inst.p2 as usize;

                if let Some(cursor) = self.cursors.get_mut(&cursor_idx) {
                    if cursor.next() {
                        // More rows to process
                        Ok(ExecuteResult::Jump(jump_addr))
                    } else {
                        // No more rows
                        Ok(ExecuteResult::Continue)
                    }
                } else {
                    Err(SqawkError::VmError(format!(
                        "Invalid cursor: {}",
                        cursor_idx
                    )))
                }
            }

            OpCode::Column => {
                // Read a column value into a register
                let cursor_idx = inst.p1 as usize;
                let column_idx = inst.p2 as usize;
                let register_idx = inst.p3 as usize;

                if let Some(cursor) = self.cursors.get(&cursor_idx) {
                    if let Some(value) = cursor.column(column_idx) {
                        // Store the value in the register
                        if register_idx < self.registers.len() {
                            self.registers[register_idx] = Register::from(value);
                            Ok(ExecuteResult::Continue)
                        } else {
                            Err(SqawkError::VmError(format!(
                                "Register index out of bounds: {}",
                                register_idx
                            )))
                        }
                    } else {
                        // Column not found or cursor invalid
                        Err(SqawkError::VmError(format!(
                            "Column {} not found for cursor {}",
                            column_idx, cursor_idx
                        )))
                    }
                } else {
                    Err(SqawkError::VmError(format!(
                        "Invalid cursor: {}",
                        cursor_idx
                    )))
                }
            }

            OpCode::InsertRow => {
                // Insert row from registers P2..P2+P3 into table at cursor P1
                let cursor_idx = inst.p1 as usize;
                let start_reg = inst.p2 as usize;
                let col_count = inst.p3 as usize;

                // Get table name from cursor
                let table_name = if let Some(cursor) = self.cursors.get(&cursor_idx) {
                    cursor.table_name().to_string()
                } else {
                    return Err(SqawkError::VmError(format!(
                        "Invalid cursor for InsertRow: {}",
                        cursor_idx
                    )));
                };

                // Collect values from registers
                let mut row = Vec::with_capacity(col_count);
                for i in 0..col_count {
                    let reg_value = self.get_register(start_reg + i)?;
                    row.push(Value::from(reg_value));
                }

                // Add to pending modifications
                self.pending_modifications
                    .push(TableModification::Insert { table_name, row });

                if self.verbose {
                    println!(
                        "InsertRow: queued insert of {} columns into cursor {}",
                        col_count, cursor_idx
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::UpdateRow => {
                // Replace the current row at cursor P1 with registers
                // P2..P2+P3, keeping its position in the table.
                let cursor_idx = inst.p1 as usize;
                let base_reg = inst.p2 as usize;
                let col_count = inst.p3 as usize;

                let (table_name, row_idx) = if let Some(cursor) = self.cursors.get(&cursor_idx) {
                    (cursor.table_name().to_string(), cursor.current_row_index())
                } else {
                    return Err(SqawkError::VmError(format!(
                        "Invalid cursor for UpdateRow: {}",
                        cursor_idx
                    )));
                };

                let table_name = inst
                    .p4
                    .as_deref()
                    .map(|s| s.to_string())
                    .unwrap_or(table_name);

                if let Some(idx) = row_idx {
                    let mut row = Vec::with_capacity(col_count);
                    for i in 0..col_count {
                        row.push(Value::from(self.get_register(base_reg + i)?));
                    }
                    self.pending_modifications.push(TableModification::Replace {
                        table_name,
                        row_index: idx,
                        row,
                    });
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::DeleteRow => {
                // Delete current row at cursor P1
                let cursor_idx = inst.p1 as usize;

                // Get table name and current row index from cursor
                let (table_name, row_idx) = if let Some(cursor) = self.cursors.get(&cursor_idx) {
                    (cursor.table_name().to_string(), cursor.current_row_index())
                } else {
                    return Err(SqawkError::VmError(format!(
                        "Invalid cursor for DeleteRow: {}",
                        cursor_idx
                    )));
                };

                if let Some(idx) = row_idx {
                    // Add to pending modifications
                    self.pending_modifications.push(TableModification::Delete {
                        table_name,
                        row_index: idx,
                    });

                    if self.verbose {
                        println!(
                            "DeleteRow: queued delete of row {} from cursor {}",
                            idx, cursor_idx
                        );
                    }
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::CreateTable => {
                // Create a new table from P4 specification
                // P4 format: "table_name:col1:type1,col2:type2,...|filepath|delimiter"
                let spec = inst.p4.clone().unwrap_or_default();
                // splitn, not split: the delimiter is the LAST field and may
                // itself be '|'. Splitting on every '|' turned
                // `WITH (delimiter = '|')` into an extra empty field and left
                // the table with the default comma instead.
                let parts: Vec<&str> = spec.splitn(3, '|').collect();

                if parts.is_empty() {
                    return Err(SqawkError::VmError(
                        "CreateTable: missing table specification".to_string(),
                    ));
                }

                // Parse table name and columns
                let name_and_cols: Vec<&str> = parts[0].split(':').collect();
                if name_and_cols.is_empty() {
                    return Err(SqawkError::VmError(
                        "CreateTable: missing table name".to_string(),
                    ));
                }

                let table_name = name_and_cols[0].to_string();

                // Parse columns (pairs of name:type)
                let col_count = (name_and_cols.len() - 1) / 2;
                let mut columns = Vec::with_capacity(col_count);
                let mut i = 1;
                while i + 1 < name_and_cols.len() {
                    columns.push(ColumnDef {
                        name: name_and_cols[i].to_string(),
                        data_type: name_and_cols[i + 1].to_string(),
                    });
                    i += 2;
                }

                // Parse optional file path and delimiter
                let file_path = parts
                    .get(1)
                    .filter(|s| !s.is_empty())
                    .map(|s| s.to_string());
                let delimiter = parts
                    .get(2)
                    .filter(|s| !s.is_empty())
                    .map(|s| s.to_string());

                // Add to pending modifications
                self.pending_modifications
                    .push(TableModification::CreateTable {
                        table_name: table_name.clone(),
                        columns,
                        file_path,
                        delimiter,
                    });

                if self.verbose {
                    println!("CreateTable: queued creation of table '{}'", table_name);
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::DropTable => {
                // Drop a table
                // P1 = 1 if IF EXISTS (don't error if table doesn't exist)
                // P4 = table name
                let if_exists = inst.p1 != 0;
                let table_name = inst.p4.as_deref().unwrap_or("");

                if table_name.is_empty() {
                    return Err(SqawkError::VmError(
                        "DropTable: missing table name".to_string(),
                    ));
                }

                // Check if table exists
                let table_exists = self.database.has_table(table_name);

                if !table_exists && !if_exists {
                    return Err(SqawkError::TableNotFound(table_name.to_string()));
                }

                // Only queue drop if table exists
                if table_exists {
                    self.pending_modifications
                        .push(TableModification::DropTable {
                            table_name: table_name.to_string(),
                        });

                    if self.verbose {
                        println!("DropTable: queued drop of table '{}'", table_name);
                    }
                } else if self.verbose {
                    println!(
                        "DropTable: table '{}' does not exist (IF EXISTS)",
                        table_name
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::AlterTableAdd => {
                // Add a column to a table
                // P4 = "table_name:column_name:column_type"
                let spec = inst.p4.as_deref().unwrap_or("");
                let parts: Vec<&str> = spec.split(':').collect();

                if parts.len() < 3 {
                    return Err(SqawkError::VmError(
                        "AlterTableAdd: invalid specification, expected table:column:type"
                            .to_string(),
                    ));
                }

                let table_name = parts[0].to_string();
                let column_name = parts[1].to_string();
                let column_type = parts[2].to_string();

                // Add to pending modifications
                self.pending_modifications
                    .push(TableModification::AlterTableAddColumn {
                        table_name: table_name.clone(),
                        column_name: column_name.clone(),
                        column_type: column_type.clone(),
                    });

                if self.verbose {
                    println!(
                        "AlterTableAdd: queued add column '{}' ({}) to table '{}'",
                        column_name, column_type, table_name
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Truncate => {
                // Remove all rows from a table
                // P4 = table name
                let table_name = inst.p4.as_deref().unwrap_or("");

                if table_name.is_empty() {
                    return Err(SqawkError::VmError(
                        "Truncate: missing table name".to_string(),
                    ));
                }

                // Add to pending modifications
                self.pending_modifications
                    .push(TableModification::Truncate {
                        table_name: table_name.to_string(),
                    });

                if self.verbose {
                    println!("Truncate: queued truncate of table '{}'", table_name);
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::StringFunc => {
                // Execute a string function
                // P1 = src reg, P2 = dest reg, P3 = arg regs (encoded), P4 = "FUNC_NAME:extra_args"
                let src_reg = inst.p1 as usize;
                let dest_reg = inst.p2 as usize;
                let func_spec = inst.p4.clone().unwrap_or_default();
                let parts: Vec<&str> = func_spec.split(':').collect();
                let func_name = parts.first().unwrap_or(&"");

                // Get source value
                let src_value = self.get_register(src_reg)?;
                let src_str = match src_value {
                    Register::String(s) => s,
                    Register::Integer(i) => i.to_string(),
                    Register::Float(f) => f.to_string(),
                    Register::Boolean(b) => b.to_string(),
                    Register::Null => {
                        // NULL in, NULL out
                        self.set_register(dest_reg, Register::Null)?;
                        return Ok(ExecuteResult::Continue);
                    }
                };

                let result = match *func_name {
                    "UPPER" => src_str.to_uppercase(),
                    "LOWER" => src_str.to_lowercase(),
                    "TRIM" => src_str.trim().to_string(),
                    "LTRIM" => src_str.trim_start().to_string(),
                    "RTRIM" => src_str.trim_end().to_string(),
                    "LENGTH" => {
                        self.set_register(dest_reg, Register::Integer(src_str.len() as i64))?;
                        return Ok(ExecuteResult::Continue);
                    }
                    "SUBSTR" | "SUBSTRING" => {
                        // P3 contains start register, parts[1] contains length if present
                        let start_reg = inst.p3 as usize;
                        let start = match self.get_register(start_reg)? {
                            Register::Integer(i) => i as usize,
                            _ => 1,
                        };
                        // Adjust for 1-based indexing
                        let start_idx = if start > 0 { start - 1 } else { 0 };

                        // Check if we have a length register specified in parts[1]
                        if let Some(len_str) = parts.get(1) {
                            if let Ok(len_reg) = len_str.parse::<usize>() {
                                if let Ok(Register::Integer(len)) = self.get_register(len_reg) {
                                    let len = len as usize;
                                    src_str.chars().skip(start_idx).take(len).collect()
                                } else {
                                    src_str.chars().skip(start_idx).collect()
                                }
                            } else {
                                src_str.chars().skip(start_idx).collect()
                            }
                        } else {
                            src_str.chars().skip(start_idx).collect()
                        }
                    }
                    "REPLACE" => {
                        // parts[1] = from_reg, parts[2] = to_reg
                        if parts.len() >= 3 {
                            if let (Ok(from_reg), Ok(to_reg)) =
                                (parts[1].parse::<usize>(), parts[2].parse::<usize>())
                            {
                                let from_str = match self.get_register(from_reg)? {
                                    Register::String(s) => s,
                                    Register::Integer(i) => i.to_string(),
                                    _ => String::new(),
                                };
                                let to_str = match self.get_register(to_reg)? {
                                    Register::String(s) => s,
                                    Register::Integer(i) => i.to_string(),
                                    _ => String::new(),
                                };
                                src_str.replace(&from_str, &to_str)
                            } else {
                                src_str
                            }
                        } else {
                            src_str
                        }
                    }
                    "CONCAT" => {
                        // Concatenate strings: parts[1..] contain additional register numbers
                        let mut result = src_str;
                        for part in parts.iter().skip(1) {
                            if let Ok(reg) = part.parse::<usize>() {
                                let val = match self.get_register(reg)? {
                                    Register::String(s) => s,
                                    Register::Integer(i) => i.to_string(),
                                    Register::Float(f) => f.to_string(),
                                    Register::Boolean(b) => b.to_string(),
                                    Register::Null => String::new(),
                                };
                                result.push_str(&val);
                            }
                        }
                        result
                    }
                    "LEFT" => {
                        // LEFT(str, n) - return leftmost n characters
                        let len_reg = inst.p3 as usize;
                        let len = match self.get_register(len_reg)? {
                            Register::Integer(i) => i.max(0) as usize,
                            _ => 0,
                        };
                        src_str.chars().take(len).collect()
                    }
                    "RIGHT" => {
                        // RIGHT(str, n) - return rightmost n characters
                        let len_reg = inst.p3 as usize;
                        let len = match self.get_register(len_reg)? {
                            Register::Integer(i) => i.max(0) as usize,
                            _ => 0,
                        };
                        let char_count = src_str.chars().count();
                        if len >= char_count {
                            src_str
                        } else {
                            src_str.chars().skip(char_count - len).collect()
                        }
                    }
                    _ => {
                        return Err(SqawkError::VmError(format!(
                            "Unknown string function: {}",
                            func_name
                        )));
                    }
                };

                self.set_register(dest_reg, Register::String(result))?;

                if self.verbose {
                    println!(
                        "StringFunc {}: r[{}] -> r[{}]",
                        func_name, src_reg, dest_reg
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::MathFunc => {
                // Execute a math function
                // P1 = src reg, P2 = dest reg, P4 = function name (ABS, ROUND, CEIL, FLOOR)
                let src_reg = inst.p1 as usize;
                let dest_reg = inst.p2 as usize;
                let func_name = inst.p4.as_deref().unwrap_or("");

                // Get source value as a number
                let src_value = self.get_register(src_reg)?;
                let num = match src_value {
                    Register::Integer(i) => i as f64,
                    Register::Float(f) => f,
                    Register::String(s) => s.parse::<f64>().unwrap_or(0.0),
                    Register::Null => {
                        // NULL in, NULL out
                        self.set_register(dest_reg, Register::Null)?;
                        return Ok(ExecuteResult::Continue);
                    }
                    Register::Boolean(b) => {
                        if b {
                            1.0
                        } else {
                            0.0
                        }
                    }
                };

                // P3 holds the register with ROUND's precision, or -1 when the
                // call had no second argument. Every emit site passes -1 for
                // the functions that take no precision; the name is checked
                // too so that neither convention alone has to be right.
                let precision = if func_name == "ROUND" && inst.p3 >= 0 {
                    match self.get_register(inst.p3 as usize)? {
                        Register::Integer(i) => Some(i),
                        Register::Float(f) => Some(f as i64),
                        // A precision that is not a number is a mistake in the
                        // query, not a request to round to 0 places.
                        Register::String(s) => match s.trim().parse::<i64>() {
                            Ok(i) => Some(i),
                            Err(_) => {
                                return Err(SqawkError::InvalidSqlQuery(format!(
                                    "ROUND precision must be a number, got '{}'",
                                    s
                                )));
                            }
                        },
                        Register::Boolean(b) => Some(b as i64),
                        Register::Null => {
                            // NULL precision yields NULL, as for the value.
                            self.set_register(dest_reg, Register::Null)?;
                            return Ok(ExecuteResult::Continue);
                        }
                    }
                } else {
                    None
                };

                let result = match func_name {
                    "ABS" => num.abs(),
                    "ROUND" => round_to(num, precision.unwrap_or(0)),
                    "CEIL" | "CEILING" => num.ceil(),
                    "FLOOR" => num.floor(),
                    _ => {
                        return Err(SqawkError::VmError(format!(
                            "Unknown math function: {}",
                            func_name
                        )));
                    }
                };

                // Store result - preserve integer type if possible
                let result_reg = if result.fract() == 0.0
                    && result >= i64::MIN as f64
                    && result <= i64::MAX as f64
                {
                    Register::Integer(result as i64)
                } else {
                    Register::Float(result)
                };
                self.set_register(dest_reg, result_reg)?;

                if self.verbose {
                    println!(
                        "MathFunc {}: r[{}] ({}) -> r[{}] ({})",
                        func_name, src_reg, num, dest_reg, result
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::DateFunc => {
                // Execute a date/time function
                // P1 = src reg (optional, for DATE/TIME), P2 = dest reg, P4 = function name
                let src_reg = inst.p1 as usize;
                let dest_reg = inst.p2 as usize;
                let func_name = inst.p4.as_deref().unwrap_or("");

                let result = match func_name {
                    "NOW" | "CURRENT_TIMESTAMP" => {
                        // Return current date and time in ISO 8601 format
                        use std::time::SystemTime;
                        let now = SystemTime::now()
                            .duration_since(SystemTime::UNIX_EPOCH)
                            .unwrap_or_default();
                        let secs = now.as_secs();
                        // Convert to datetime string (simplified - just date for now)
                        let days_since_epoch = secs / 86400;
                        let years = 1970 + (days_since_epoch / 365);
                        let remaining_days = days_since_epoch % 365;
                        let months = remaining_days / 30 + 1;
                        let days = remaining_days % 30 + 1;
                        let hours = (secs % 86400) / 3600;
                        let minutes = (secs % 3600) / 60;
                        let seconds = secs % 60;
                        format!(
                            "{:04}-{:02}-{:02} {:02}:{:02}:{:02}",
                            years, months, days, hours, minutes, seconds
                        )
                    }
                    "DATE" => {
                        // Extract or validate date from input
                        let src_value = self.get_register(src_reg)?;
                        match src_value {
                            Register::String(s) => {
                                // Try to extract just the date part (YYYY-MM-DD)
                                if s.len() >= 10 {
                                    s[..10].to_string()
                                } else {
                                    s
                                }
                            }
                            Register::Null => {
                                self.set_register(dest_reg, Register::Null)?;
                                return Ok(ExecuteResult::Continue);
                            }
                            _ => {
                                // Convert to string and try to extract date
                                match src_value {
                                    Register::Integer(i) => i.to_string(),
                                    Register::Float(f) => f.to_string(),
                                    _ => String::new(),
                                }
                            }
                        }
                    }
                    "TIME" => {
                        // Extract or validate time from input
                        let src_value = self.get_register(src_reg)?;
                        match src_value {
                            Register::String(s) => {
                                // Try to extract time part (HH:MM:SS)
                                if s.contains(' ') {
                                    // Datetime format, extract time after space
                                    s.split(' ').nth(1).unwrap_or(&s).to_string()
                                } else {
                                    // Already a time string or return as-is
                                    s
                                }
                            }
                            Register::Null => {
                                self.set_register(dest_reg, Register::Null)?;
                                return Ok(ExecuteResult::Continue);
                            }
                            _ => String::new(),
                        }
                    }
                    "CURRENT_DATE" => {
                        // Return current date only
                        use std::time::SystemTime;
                        let now = SystemTime::now()
                            .duration_since(SystemTime::UNIX_EPOCH)
                            .unwrap_or_default();
                        let days_since_epoch = now.as_secs() / 86400;
                        let years = 1970 + (days_since_epoch / 365);
                        let remaining_days = days_since_epoch % 365;
                        let months = remaining_days / 30 + 1;
                        let days = remaining_days % 30 + 1;
                        format!("{:04}-{:02}-{:02}", years, months, days)
                    }
                    "CURRENT_TIME" => {
                        // Return current time only
                        use std::time::SystemTime;
                        let now = SystemTime::now()
                            .duration_since(SystemTime::UNIX_EPOCH)
                            .unwrap_or_default();
                        let secs = now.as_secs();
                        let hours = (secs % 86400) / 3600;
                        let minutes = (secs % 3600) / 60;
                        let seconds = secs % 60;
                        format!("{:02}:{:02}:{:02}", hours, minutes, seconds)
                    }
                    _ => {
                        return Err(SqawkError::VmError(format!(
                            "Unknown date function: {}",
                            func_name
                        )));
                    }
                };

                self.set_register(dest_reg, Register::String(result))?;

                if self.verbose {
                    println!("DateFunc {}: -> r[{}]", func_name, dest_reg);
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Add => {
                // dest = left + right
                let left_reg = inst.p1 as usize;
                let right_reg = inst.p2 as usize;
                let dest_reg = inst.p3 as usize;

                let result = self.arithmetic_op(left_reg, right_reg, "+")?;
                self.set_register(dest_reg, result)?;

                if self.verbose {
                    println!("Add: r[{}] + r[{}] -> r[{}]", left_reg, right_reg, dest_reg);
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Subtract => {
                // dest = left - right
                let left_reg = inst.p1 as usize;
                let right_reg = inst.p2 as usize;
                let dest_reg = inst.p3 as usize;

                let result = self.arithmetic_op(left_reg, right_reg, "-")?;
                self.set_register(dest_reg, result)?;

                if self.verbose {
                    println!(
                        "Subtract: r[{}] - r[{}] -> r[{}]",
                        left_reg, right_reg, dest_reg
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Multiply => {
                // dest = left * right
                let left_reg = inst.p1 as usize;
                let right_reg = inst.p2 as usize;
                let dest_reg = inst.p3 as usize;

                let result = self.arithmetic_op(left_reg, right_reg, "*")?;
                self.set_register(dest_reg, result)?;

                if self.verbose {
                    println!(
                        "Multiply: r[{}] * r[{}] -> r[{}]",
                        left_reg, right_reg, dest_reg
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Divide => {
                // dest = left / right
                let left_reg = inst.p1 as usize;
                let right_reg = inst.p2 as usize;
                let dest_reg = inst.p3 as usize;

                let result = self.arithmetic_op(left_reg, right_reg, "/")?;
                self.set_register(dest_reg, result)?;

                if self.verbose {
                    println!(
                        "Divide: r[{}] / r[{}] -> r[{}]",
                        left_reg, right_reg, dest_reg
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Remainder => {
                // dest = left % right
                let left_reg = inst.p1 as usize;
                let right_reg = inst.p2 as usize;
                let dest_reg = inst.p3 as usize;

                let result = self.arithmetic_op(left_reg, right_reg, "%")?;
                self.set_register(dest_reg, result)?;

                if self.verbose {
                    println!(
                        "Remainder: r[{}] %% r[{}] -> r[{}]",
                        left_reg, right_reg, dest_reg
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::WindowAggStep => {
                // Window aggregate step: update window state based on current row
                // P1 = func type, P2 = value reg, P3 = cursor, P4 = window spec
                let func_type = inst.p1;
                let _value_reg = inst.p2 as usize;
                let cursor_id = inst.p3 as usize;
                let window_spec = inst.p4.clone().unwrap_or_default();

                // Parse state registers from window spec
                let state_part = window_spec.split(":state=").nth(1).unwrap_or("");
                let mut row_num_reg = 0usize;
                let mut prev_part_reg = 0usize;
                let mut rank_reg = 0usize;
                let mut prev_ord_reg = 0usize;
                let mut dense_rank_reg = 0usize;
                let mut agg_sum_reg = 0usize;
                let mut agg_count_reg = 0usize;
                let mut agg_min_reg = 0usize;
                let mut agg_max_reg = 0usize;
                let mut agg_col = 0usize;

                for part in state_part.split(',') {
                    let kv: Vec<&str> = part.split(':').collect();
                    if kv.len() == 2 {
                        let val = kv[1].parse::<usize>().unwrap_or(0);
                        match kv[0] {
                            "row_num" => row_num_reg = val,
                            "prev_part" => prev_part_reg = val,
                            "rank" => rank_reg = val,
                            "prev_ord" => prev_ord_reg = val,
                            "dense_rank" => dense_rank_reg = val,
                            "agg_sum" => agg_sum_reg = val,
                            "agg_count" => agg_count_reg = val,
                            "agg_min" => agg_min_reg = val,
                            "agg_max" => agg_max_reg = val,
                            "agg_col" => agg_col = val,
                            _ => {}
                        }
                    }
                }

                // Parse partition and order column indices
                let part_str = window_spec
                    .split(":part=")
                    .nth(1)
                    .and_then(|s| s.split(':').next())
                    .unwrap_or("");
                let ord_str = window_spec
                    .split(":ord=")
                    .nth(1)
                    .and_then(|s| s.split(':').next())
                    .unwrap_or("");

                let part_cols: Vec<usize> =
                    part_str.split(',').filter_map(|s| s.parse().ok()).collect();
                let ord_cols: Vec<usize> =
                    ord_str.split(',').filter_map(|s| s.parse().ok()).collect();

                // Get current partition key value (first partition column if any)
                let current_part_val = if !part_cols.is_empty() {
                    if let Some(cursor) = self.cursors.get(&cursor_id) {
                        cursor.column(part_cols[0]).unwrap_or(Value::Null)
                    } else {
                        Value::Null
                    }
                } else {
                    Value::Null // No partition = all rows in same partition
                };

                // Get current order key value
                let current_ord_val = if !ord_cols.is_empty() {
                    if let Some(cursor) = self.cursors.get(&cursor_id) {
                        cursor.column(ord_cols[0]).unwrap_or(Value::Null)
                    } else {
                        Value::Null
                    }
                } else {
                    Value::Null
                };

                // Check if partition changed
                // Use a special marker to detect first row: check if row_num is 0
                let is_first_row = match self.get_register(row_num_reg)? {
                    Register::Integer(n) => n == 0,
                    _ => true,
                };

                let prev_part_val = self.get_register(prev_part_reg)?;
                let partition_changed = if is_first_row {
                    true // First row is always a "new partition"
                } else {
                    match (&prev_part_val, &current_part_val) {
                        (Register::Null, Value::Null) => false, // Both null = same partition (or no partition)
                        (Register::Null, _) => true, // Was null, now has value = new partition
                        (_, Value::Null) => true,    // Had value, now null = new partition
                        (Register::Integer(p), Value::Integer(c)) => *p != *c,
                        (Register::String(p), Value::String(c)) => p != c,
                        (Register::Float(p), Value::Float(c)) => *p != *c,
                        _ => true, // Different types = partition changed
                    }
                };

                if partition_changed {
                    // The row about to be emitted starts a new partition.
                    self.window_partition_starts.push(self.results.len());
                }

                // Check if order value changed (for RANK)
                let prev_ord_val = self.get_register(prev_ord_reg)?;
                let order_changed = if is_first_row {
                    true
                } else {
                    match (&prev_ord_val, &current_ord_val) {
                        (Register::Null, Value::Null) => false,
                        (Register::Null, _) => true,
                        (_, Value::Null) => true,
                        (Register::Integer(p), Value::Integer(c)) => *p != *c,
                        (Register::String(p), Value::String(c)) => p != c,
                        (Register::Float(p), Value::Float(c)) => *p != *c,
                        _ => true,
                    }
                };

                // Update window state based on function type
                match func_type {
                    0 => {
                        // ROW_NUMBER: increment, reset on partition change
                        let current = if partition_changed {
                            1
                        } else {
                            match self.get_register(row_num_reg)? {
                                Register::Integer(n) => n + 1,
                                _ => 1,
                            }
                        };
                        self.set_register(row_num_reg, Register::Integer(current))?;
                    }
                    1 => {
                        // RANK: like row_number but same value for ties
                        let row_num = if partition_changed {
                            1
                        } else {
                            match self.get_register(row_num_reg)? {
                                Register::Integer(n) => n + 1,
                                _ => 1,
                            }
                        };
                        self.set_register(row_num_reg, Register::Integer(row_num))?;

                        if partition_changed || order_changed {
                            self.set_register(rank_reg, Register::Integer(row_num))?;
                        }
                        // else rank stays the same (tie)
                    }
                    2 => {
                        // DENSE_RANK: increment only when value changes
                        let row_num = if partition_changed {
                            1
                        } else {
                            match self.get_register(row_num_reg)? {
                                Register::Integer(n) => n + 1,
                                _ => 1,
                            }
                        };
                        self.set_register(row_num_reg, Register::Integer(row_num))?;

                        let dense = if partition_changed {
                            1
                        } else if order_changed {
                            match self.get_register(dense_rank_reg)? {
                                Register::Integer(n) => n + 1,
                                _ => 1,
                            }
                        } else {
                            match self.get_register(dense_rank_reg)? {
                                Register::Integer(n) => n,
                                _ => 1,
                            }
                        };
                        self.set_register(dense_rank_reg, Register::Integer(dense))?;
                    }
                    5..=9 => {
                        // Aggregate window functions: SUM(5), AVG(6), COUNT(7), MIN(8), MAX(9)
                        // Track row number for partition detection
                        let current = if partition_changed {
                            1
                        } else {
                            match self.get_register(row_num_reg)? {
                                Register::Integer(n) => n + 1,
                                _ => 1,
                            }
                        };
                        self.set_register(row_num_reg, Register::Integer(current))?;

                        // Get the value from the aggregate column
                        let agg_value = if let Some(cursor) = self.cursors.get(&cursor_id) {
                            cursor.column(agg_col).unwrap_or(Value::Null)
                        } else {
                            Value::Null
                        };

                        // Reset aggregates on partition change
                        if partition_changed {
                            self.set_register(agg_sum_reg, Register::Integer(0))?;
                            self.set_register(agg_count_reg, Register::Integer(0))?;
                            self.set_register(agg_min_reg, Register::Null)?;
                            self.set_register(agg_max_reg, Register::Null)?;
                        }

                        // Update aggregates based on current value
                        if !matches!(agg_value, Value::Null) {
                            // Update count
                            let new_count = match self.get_register(agg_count_reg)? {
                                Register::Integer(n) => n + 1,
                                _ => 1,
                            };
                            self.set_register(agg_count_reg, Register::Integer(new_count))?;

                            // Update sum
                            let val_num = match &agg_value {
                                Value::Integer(n) => *n as f64,
                                Value::Float(f) => *f,
                                Value::String(s) => s.parse::<f64>().unwrap_or(0.0),
                                _ => 0.0,
                            };
                            let current_sum = match self.get_register(agg_sum_reg)? {
                                Register::Integer(n) => n as f64,
                                Register::Float(f) => f,
                                _ => 0.0,
                            };
                            self.set_register(agg_sum_reg, Register::Float(current_sum + val_num))?;

                            // Update min
                            let should_update_min = match self.get_register(agg_min_reg)? {
                                Register::Null => true,
                                Register::Integer(n) => val_num < (n as f64),
                                Register::Float(f) => val_num < f,
                                _ => false,
                            };
                            if should_update_min {
                                self.set_register(agg_min_reg, Register::Float(val_num))?;
                            }

                            // Update max
                            let should_update_max = match self.get_register(agg_max_reg)? {
                                Register::Null => true,
                                Register::Integer(n) => val_num > (n as f64),
                                Register::Float(f) => val_num > f,
                                _ => false,
                            };
                            if should_update_max {
                                self.set_register(agg_max_reg, Register::Float(val_num))?;
                            }
                        }
                    }
                    _ => {
                        // Other window functions - just track row number
                        let current = if partition_changed {
                            1
                        } else {
                            match self.get_register(row_num_reg)? {
                                Register::Integer(n) => n + 1,
                                _ => 1,
                            }
                        };
                        self.set_register(row_num_reg, Register::Integer(current))?;
                    }
                }

                // Store current partition/order values for next iteration
                self.set_register(prev_part_reg, Register::from(current_part_val))?;
                self.set_register(prev_ord_reg, Register::from(current_ord_val))?;

                if self.verbose {
                    println!(
                        "WindowAggStep: func_type={}, partition_changed={}",
                        func_type, partition_changed
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::WindowFinalize => {
                // Give every row of a partition that partition's FINAL window
                // value, which for an unordered frame is the partition total.
                //
                // The streaming pass accumulates as it goes, so without this a
                // window aggregate with PARTITION BY and no ORDER BY returns a
                // RUNNING total. SQL defines the frame in that case as the
                // whole partition, so every row should see the same value.
                let col = inst.p1 as usize;
                let starts = self.window_partition_starts.clone();
                for (i, &start) in starts.iter().enumerate() {
                    let end = starts.get(i + 1).copied().unwrap_or(self.results.len());
                    if end == 0 || start >= end {
                        continue;
                    }
                    let final_value = match self.results[end - 1].get(col) {
                        Some(v) => v.clone(),
                        None => continue,
                    };
                    for row in &mut self.results[start..end] {
                        if col < row.len() {
                            row[col] = final_value.clone();
                        }
                    }
                }
                Ok(ExecuteResult::Continue)
            }

            OpCode::WindowValue => {
                // Get current window function value
                // P1 = state base reg, P2 = dest reg, P3 = func type
                // P4 = for LAG/LEAD: "LAG:cursor=N:col=N:offset=N:parts=N,N,..."
                let state_base_reg = inst.p1 as usize;
                let dest_reg = inst.p2 as usize;
                let func_type = inst.p3;

                let value = match func_type {
                    0 => {
                        // ROW_NUMBER - value is in state_base_reg (row_num_reg)
                        self.get_register(state_base_reg)?
                    }
                    1 => {
                        // RANK - value is in rank_reg (state_base_reg + 2)
                        self.get_register(state_base_reg + 2)?
                    }
                    2 => {
                        // DENSE_RANK - value is in dense_rank_reg (state_base_reg + 4)
                        self.get_register(state_base_reg + 4)?
                    }
                    3 | 4 => {
                        // LAG (3) or LEAD (4)
                        // Parse P4 to get cursor, column, offset, partition columns
                        let p4 = inst.p4.clone().unwrap_or_default();
                        let mut cursor_id = 1usize;
                        let mut col_pos = 0usize;
                        let mut offset = 1i64;
                        let mut partition_cols: Vec<usize> =
                            Vec::with_capacity(DEFAULT_SORT_KEYS_CAPACITY);

                        for part in p4.split(':') {
                            if let Some(val) = part.strip_prefix("cursor=") {
                                cursor_id = val.parse().unwrap_or(1);
                            } else if let Some(val) = part.strip_prefix("col=") {
                                col_pos = val.parse().unwrap_or(0);
                            } else if let Some(val) = part.strip_prefix("offset=") {
                                offset = val.parse().unwrap_or(1);
                            } else if let Some(parts_str) = part.strip_prefix("parts=") {
                                if !parts_str.is_empty() {
                                    partition_cols = parts_str
                                        .split(',')
                                        .filter_map(|s| s.parse::<usize>().ok())
                                        .collect();
                                }
                            }
                        }

                        // Get the current row number in partition (1-based)
                        let row_num = match self.get_register(state_base_reg)? {
                            Register::Integer(n) => n,
                            _ => 1,
                        };

                        // Calculate the row offset
                        let row_offset = if func_type == 3 { -offset } else { offset }; // LAG = negative, LEAD = positive

                        // For LAG: check if we have enough rows behind us in this partition
                        // For LEAD: check if target row exists and is in same partition
                        if let Some(cursor) = self.cursors.get(&cursor_id) {
                            let current_pos = cursor.position;

                            // For LAG: row_num is 1-based row in partition
                            // If row_num <= offset, we can't look back that far, return NULL
                            if func_type == 3 && row_num <= offset {
                                Register::Null
                            } else if func_type == 4 {
                                // For LEAD: need to check if target row is in same partition
                                let target_pos = (current_pos as i64 + row_offset) as usize;
                                let rows = cursor.rows();

                                if target_pos >= rows.len() {
                                    Register::Null
                                } else {
                                    // Check if target row has same partition keys
                                    let current_row = &rows[current_pos];
                                    let target_row = &rows[target_pos];

                                    let same_partition = if partition_cols.is_empty() {
                                        true // No partition = all in same partition
                                    } else {
                                        partition_cols
                                            .iter()
                                            .all(|&col| current_row.get(col) == target_row.get(col))
                                    };

                                    if same_partition {
                                        target_row
                                            .get(col_pos)
                                            .cloned()
                                            .map(Register::from)
                                            .unwrap_or(Register::Null)
                                    } else {
                                        Register::Null
                                    }
                                }
                            } else {
                                // LAG with valid offset
                                cursor
                                    .column_at_offset(col_pos, row_offset)
                                    .map(Register::from)
                                    .unwrap_or(Register::Null)
                            }
                        } else {
                            Register::Null
                        }
                    }
                    5 => {
                        // SUM - return running sum
                        // agg_sum is at state_base_reg + 6
                        self.get_register(state_base_reg + 6)?
                    }
                    6 => {
                        // AVG - return running average (sum / count)
                        // agg_sum is at state_base_reg + 6, agg_count is at state_base_reg + 7
                        let sum = match self.get_register(state_base_reg + 6)? {
                            Register::Float(f) => f,
                            Register::Integer(n) => n as f64,
                            _ => 0.0,
                        };
                        let count = match self.get_register(state_base_reg + 7)? {
                            Register::Integer(n) => n,
                            _ => 1,
                        };
                        if count > 0 {
                            Register::Float(sum / count as f64)
                        } else {
                            Register::Null
                        }
                    }
                    7 => {
                        // COUNT - return running count
                        // agg_count is at state_base_reg + 7
                        self.get_register(state_base_reg + 7)?
                    }
                    8 => {
                        // MIN - return running min
                        // agg_min is at state_base_reg + 8
                        self.get_register(state_base_reg + 8)?
                    }
                    9 => {
                        // MAX - return running max
                        // agg_max is at state_base_reg + 9
                        self.get_register(state_base_reg + 9)?
                    }
                    _ => {
                        // Default to row number
                        self.get_register(state_base_reg)?
                    }
                };

                self.set_register(dest_reg, value)?;

                if self.verbose {
                    println!("WindowValue: func_type={}, dest=r[{}]", func_type, dest_reg);
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Copy => {
                // Copy register P1 to register P2
                let src_reg = inst.p1 as usize;
                let dest_reg = inst.p2 as usize;

                let value = self.get_register(src_reg)?;
                self.set_register(dest_reg, value)?;

                if self.verbose {
                    println!("Copy: r[{}] -> r[{}]", src_reg, dest_reg);
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Integer => {
                // Load integer constant into register
                // Special case: p1=-1 means "use current result count as the value"
                let value = if inst.p1 == -1 {
                    self.results.len() as i64
                } else {
                    inst.p1
                };
                let register_idx = inst.p2 as usize;

                if register_idx < self.registers.len() {
                    self.registers[register_idx] = Register::Integer(value);
                    Ok(ExecuteResult::Continue)
                } else {
                    Err(SqawkError::VmError(format!(
                        "Register index out of bounds: {}",
                        register_idx
                    )))
                }
            }

            OpCode::String => {
                // Load string constant into register
                let register_idx = inst.p2 as usize;
                let string_value = inst.p4.as_deref().unwrap_or("").to_string();

                if register_idx < self.registers.len() {
                    self.registers[register_idx] = Register::String(string_value);
                    Ok(ExecuteResult::Continue)
                } else {
                    Err(SqawkError::VmError(format!(
                        "Register index out of bounds: {}",
                        register_idx
                    )))
                }
            }

            OpCode::Null => {
                // Load NULL into register
                let register_idx = inst.p2 as usize;

                if register_idx < self.registers.len() {
                    self.registers[register_idx] = Register::Null;
                    Ok(ExecuteResult::Continue)
                } else {
                    Err(SqawkError::VmError(format!(
                        "Register index out of bounds: {}",
                        register_idx
                    )))
                }
            }

            OpCode::ResultRow => {
                // Return a result row from registers
                let start_reg = inst.p1 as usize;
                let column_count = inst.p2 as usize;

                // Collect values from registers
                let mut row = Vec::with_capacity(column_count);
                for i in 0..column_count {
                    let reg_idx = start_reg + i;
                    if reg_idx < self.registers.len() {
                        // Convert register to Value
                        let value = Value::from(self.registers[reg_idx].clone());
                        row.push(value);
                    } else {
                        return Err(SqawkError::VmError(format!(
                            "Register index out of bounds: {}",
                            reg_idx
                        )));
                    }
                }

                // Add the row to results
                self.results.push(row);

                // Phase 4C: Also record row indices from each cursor (parallel path)
                // This collects the current row index from each cursor in source order
                let row_indices: Vec<Option<usize>> = self
                    .cursor_source_order
                    .iter()
                    .map(|cursor_id| {
                        self.cursors
                            .get(cursor_id)
                            .and_then(|c| c.current_row_idx())
                    })
                    .collect();
                self.result_row_indices.push(row_indices);

                Ok(ExecuteResult::Continue)
            }

            OpCode::Begin => {
                // Begin a transaction - initiates a new atomic unit of work
                // All operations performed between BEGIN and COMMIT/ROLLBACK are treated as a single
                // logical operation from the perspective of database consistency
                if self.transaction_state == TransactionState::Active {
                    return Err(SqawkError::VmError(
                        "Transaction already in progress".to_string(),
                    ));
                }

                // Initialize transaction state to track changes
                self.transaction_state = TransactionState::Active;

                if self.verbose {
                    println!("BEGIN TRANSACTION");
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Commit => {
                // Commit a transaction - makes all changes permanent
                // Finalizes the changes made during the transaction and ensures they
                // become permanent, durable parts of the database state
                if self.transaction_state != TransactionState::Active {
                    return Err(SqawkError::VmError(
                        "No transaction in progress to commit".to_string(),
                    ));
                }

                // Finalize transaction
                self.transaction_state = TransactionState::Committed;

                if self.verbose {
                    println!("COMMIT TRANSACTION");
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Rollback => {
                // Rollback a transaction - abandons all changes made during the transaction
                // Undoes all operations since BEGIN, returning the database to its prior state
                if self.transaction_state != TransactionState::Active {
                    return Err(SqawkError::VmError(
                        "No transaction in progress to rollback".to_string(),
                    ));
                }

                // Update transaction state to reflect rollback completion
                self.transaction_state = TransactionState::RolledBack;

                if self.verbose {
                    println!("ROLLBACK TRANSACTION");
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::SavePoint => {
                // Create a savepoint within the current transaction
                // A savepoint marks a point within a transaction to which you can later roll back
                // without rolling back the entire transaction
                if self.transaction_state != TransactionState::Active {
                    return Err(SqawkError::VmError(
                        "No active transaction for savepoint".to_string(),
                    ));
                }

                // In a full implementation, we would:
                // 1. Create a marker in the transaction log to identify this position
                // 2. Allow for multiple savepoints with different names
                // 3. Support rolling back to any specific savepoint
                // 4. Properly handle nested savepoints in a hierarchical manner
                let savepoint_name = inst
                    .p4
                    .as_deref()
                    .map(|s| s.to_string())
                    .unwrap_or_else(|| format!("sp_{}", inst.p1));

                if self.verbose {
                    println!("SAVEPOINT {}", savepoint_name);
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Release => {
                // Release a savepoint (commit changes up to the savepoint)
                // This permanently applies all changes made since the savepoint was created
                // while keeping the transaction active for further operations
                if self.transaction_state != TransactionState::Active {
                    return Err(SqawkError::VmError(
                        "No active transaction for savepoint release".to_string(),
                    ));
                }

                // In a complete implementation, we would:
                // 1. Find the specified savepoint in our transaction log
                // 2. Make all changes permanent up to that point
                // 3. Remove this savepoint and any subsequent nested savepoints
                // 4. Maintain the active transaction state for further operations
                let savepoint_name = inst
                    .p4
                    .as_deref()
                    .map(|s| s.to_string())
                    .unwrap_or_else(|| format!("sp_{}", inst.p1));

                if self.verbose {
                    println!("RELEASE SAVEPOINT {}", savepoint_name);
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Lt | OpCode::Le | OpCode::Gt | OpCode::Ge | OpCode::Eq | OpCode::Ne => {
                // Unified comparison operations
                self.execute_comparison(inst)
            }

            OpCode::IfZ => {
                // Jump if register P1 contains 0
                // P1: Register to test
                // P2: Jump destination (instruction index)

                // Get the register value to test
                let reg = self.get_register(inst.p1 as usize)?;

                // Check if the register contains 0 (or logical false)
                let is_zero = match reg {
                    Register::Integer(val) => val == 0,
                    Register::Float(val) => val == 0.0,
                    Register::String(ref val) => val.is_empty(),
                    Register::Null => true, // NULL is considered "zero" for this purpose
                    Register::Boolean(val) => !val, // false is considered "zero"
                };

                // If the register is zero/false, jump to P2
                if is_zero {
                    // Jump to the target instruction
                    return Ok(ExecuteResult::Jump(inst.p2 as usize));
                }

                // Otherwise, continue to the next instruction
                Ok(ExecuteResult::Continue)
            }

            OpCode::IfPos => {
                // Jump if register P1 contains a positive value (> 0)
                // P1: Register to test
                // P2: Jump destination (instruction index)

                // Get the register value to test
                let reg = self.get_register(inst.p1 as usize)?;

                // Check if the register contains a positive value
                let is_positive = match reg {
                    Register::Integer(val) => val > 0,
                    Register::Float(val) => val > 0.0,
                    // Boolean true is considered positive, false is not
                    Register::Boolean(val) => val,
                    // String and null aren't treated as positive
                    Register::String(_) | Register::Null => false,
                };

                // If the register is positive, jump to P2
                if is_positive {
                    return Ok(ExecuteResult::Jump(inst.p2 as usize));
                }

                // Otherwise, continue to the next instruction
                Ok(ExecuteResult::Continue)
            }

            OpCode::IfNeg => {
                // Jump if register P1 contains a negative value (< 0)
                // P1: Register to test
                // P2: Jump destination (instruction index)

                // Get the register value to test
                let reg = self.get_register(inst.p1 as usize)?;

                // Check if the register contains a negative value
                let is_negative = match reg {
                    Register::Integer(val) => val < 0,
                    Register::Float(val) => val < 0.0,
                    // Boolean values aren't treated as negative
                    Register::Boolean(_) => false,
                    // String and null aren't treated as negative
                    Register::String(_) | Register::Null => false,
                };

                // If the register is negative, jump to P2
                if is_negative {
                    return Ok(ExecuteResult::Jump(inst.p2 as usize));
                }

                // Otherwise, continue to the next instruction
                Ok(ExecuteResult::Continue)
            }

            OpCode::Like => {
                // LIKE pattern matching (P1 LIKE P4 pattern, result in P3)
                // P1: Register containing value to test
                // P2: Flags (bit 0 = negated, bit 1 = case_insensitive)
                // P3: Destination register for result (1 for match, 0 for no match)
                // P4: Pattern string with SQL LIKE wildcards (% = any sequence, _ = any single char)

                let value_reg = inst.p1 as usize;
                let flags = inst.p2;
                let result_reg = inst.p3 as usize;
                let pattern = inst.p4.clone().unwrap_or_default();

                let negated = (flags & 1) != 0;
                let case_insensitive = (flags & 2) != 0;

                // Get the value to test
                let value = self.get_register(value_reg)?;
                let value_str = match value {
                    Register::String(s) => s,
                    Register::Integer(i) => i.to_string(),
                    Register::Float(f) => f.to_string(),
                    Register::Boolean(b) => b.to_string(),
                    Register::Null => {
                        // NULL LIKE anything is NULL (treated as false for filtering)
                        self.set_register(result_reg, Register::Integer(0))?;
                        return Ok(ExecuteResult::Continue);
                    }
                };

                // Convert SQL LIKE pattern to regex
                let regex_pattern = self.like_pattern_to_regex(&pattern, case_insensitive);

                // Perform the match
                let matched = match regex::Regex::new(&regex_pattern) {
                    Ok(re) => re.is_match(&value_str),
                    Err(_) => false,
                };

                // Apply negation if needed
                let result = if negated { !matched } else { matched };

                self.set_register(result_reg, Register::Integer(if result { 1 } else { 0 }))?;

                Ok(ExecuteResult::Continue)
            }

            OpCode::Glob => {
                // GLOB pattern matching (Unix-style wildcards)
                // P1: Register containing value to test
                // P2: Flags (bit 0 = negated)
                // P3: Destination register for result
                // P4: Pattern string with GLOB wildcards (* = any sequence, ? = any single char)

                let value_reg = inst.p1 as usize;
                let flags = inst.p2;
                let result_reg = inst.p3 as usize;
                let pattern = inst.p4.clone().unwrap_or_default();

                let negated = (flags & 1) != 0;

                // Get the value to test
                let value = self.get_register(value_reg)?;
                let value_str = match value {
                    Register::String(s) => s,
                    Register::Integer(i) => i.to_string(),
                    Register::Float(f) => f.to_string(),
                    Register::Boolean(b) => b.to_string(),
                    Register::Null => {
                        self.set_register(result_reg, Register::Integer(0))?;
                        return Ok(ExecuteResult::Continue);
                    }
                };

                // Convert GLOB pattern to regex (case-sensitive)
                let regex_pattern = self.glob_pattern_to_regex(&pattern);

                // Perform the match
                let matched = match regex::Regex::new(&regex_pattern) {
                    Ok(re) => re.is_match(&value_str),
                    Err(_) => false,
                };

                // Apply negation if needed
                let result = if negated { !matched } else { matched };

                self.set_register(result_reg, Register::Integer(if result { 1 } else { 0 }))?;

                Ok(ExecuteResult::Continue)
            }

            OpCode::Cast => {
                // Cast value in P1 to type specified in P4, result in P2
                // P1: Source register containing value to cast
                // P2: Destination register for result
                // P4: Target type name ("INTEGER", "TEXT", "REAL", "BOOLEAN")

                let src_reg = inst.p1 as usize;
                let dest_reg = inst.p2 as usize;
                let target_type = inst.p4.as_deref().unwrap_or("TEXT");

                let src_value = self.get_register(src_reg)?;

                let result = match target_type.to_uppercase().as_str() {
                    "INTEGER" | "INT" => match src_value {
                        Register::Integer(i) => Register::Integer(i),
                        Register::Float(f) => Register::Integer(f as i64),
                        Register::String(s) => Register::Integer(s.parse::<i64>().unwrap_or(0)),
                        Register::Boolean(b) => Register::Integer(if b { 1 } else { 0 }),
                        Register::Null => Register::Null,
                    },
                    "REAL" | "FLOAT" | "DOUBLE" => match src_value {
                        Register::Integer(i) => Register::Float(i as f64),
                        Register::Float(f) => Register::Float(f),
                        Register::String(s) => Register::Float(s.parse::<f64>().unwrap_or(0.0)),
                        Register::Boolean(b) => Register::Float(if b { 1.0 } else { 0.0 }),
                        Register::Null => Register::Null,
                    },
                    "TEXT" | "VARCHAR" | "CHAR" | "STRING" => match src_value {
                        Register::Integer(i) => Register::String(i.to_string()),
                        Register::Float(f) => Register::String(f.to_string()),
                        Register::String(s) => Register::String(s),
                        Register::Boolean(b) => Register::String(b.to_string()),
                        Register::Null => Register::Null,
                    },
                    "BOOLEAN" | "BOOL" => match src_value {
                        Register::Integer(i) => Register::Boolean(i != 0),
                        Register::Float(f) => Register::Boolean(f != 0.0),
                        Register::String(s) => {
                            let lower = s.to_lowercase();
                            Register::Boolean(lower == "true" || lower == "1" || lower == "yes")
                        }
                        Register::Boolean(b) => Register::Boolean(b),
                        Register::Null => Register::Null,
                    },
                    _ => {
                        // Unknown type - default to TEXT
                        match src_value {
                            Register::Integer(i) => Register::String(i.to_string()),
                            Register::Float(f) => Register::String(f.to_string()),
                            Register::String(s) => Register::String(s),
                            Register::Boolean(b) => Register::String(b.to_string()),
                            Register::Null => Register::Null,
                        }
                    }
                };

                self.set_register(dest_reg, result)?;
                Ok(ExecuteResult::Continue)
            }

            OpCode::IsNull => {
                // Check if P1 is NULL, set P2 to 1 if NULL, 0 otherwise
                let src_reg = inst.p1 as usize;
                let dest_reg = inst.p2 as usize;

                let src_value = self.get_register(src_reg)?;
                let is_null = matches!(src_value, Register::Null);

                self.set_register(dest_reg, Register::Integer(if is_null { 1 } else { 0 }))?;
                Ok(ExecuteResult::Continue)
            }

            OpCode::Not => {
                // Logical negation with NULL propagation: NOT UNKNOWN is
                // UNKNOWN, not true.
                let src = self.get_register(inst.p1 as usize)?;
                let out = match src {
                    Register::Null => Register::Null,
                    other => Register::Integer(if Self::register_is_truthy(&other) {
                        0
                    } else {
                        1
                    }),
                };
                self.set_register(inst.p2 as usize, out)?;
                Ok(ExecuteResult::Continue)
            }

            OpCode::Compare => {
                // Pairwise-compare two register vectors of length P3, starting
                // at P1 and P2, and stash the Ordering for the next Jump.
                //
                // Uses internal ordering (compare_values), where NULL == NULL.
                // That is deliberate: this drives grouping and sorting, where
                // NULL keys must group together, whereas SQL `=` must yield
                // UNKNOWN for NULL. Keeping them separate is what lets the
                // three-valued-logic change apply to Eq without breaking
                // GROUP BY on a nullable key.
                let lhs_start = inst.p1 as usize;
                let rhs_start = inst.p2 as usize;
                let count = inst.p3 as usize;

                let mut ordering = std::cmp::Ordering::Equal;
                for i in 0..count {
                    let a: Value = self.get_register(lhs_start + i)?.into();
                    let b: Value = self.get_register(rhs_start + i)?.into();
                    ordering = compare_values(&a, &b);
                    if ordering != std::cmp::Ordering::Equal {
                        break;
                    }
                }
                self.compare_flag = ordering;
                Ok(ExecuteResult::Continue)
            }

            OpCode::Jump => {
                // Three-way branch on the last Compare.
                let target = match self.compare_flag {
                    std::cmp::Ordering::Less => inst.p1,
                    std::cmp::Ordering::Equal => inst.p2,
                    std::cmp::Ordering::Greater => inst.p3,
                };
                Ok(ExecuteResult::Jump(target as usize))
            }

            // JumpIfTrue and JumpIfFalse have been replaced by SQLite-style opcodes:
            // - IfZ (jump if zero/false)
            // - IfPos (jump if positive)
            // - IfNeg (jump if negative)
            OpCode::Noop => {
                // No operation
                Ok(ExecuteResult::Continue)
            }

            OpCode::SortResults => {
                // Sort the accumulated result rows.
                //
                // A post-processing opcode in the same family as Distinct and
                // Limit: it runs over self.results after the producing loop has
                // finished, so it is independent of how those rows were
                // produced. That is what lets ORDER BY work over GROUP BY
                // output and over joins, neither of which can feed the
                // cursor-based sorter used by a plain table scan.
                let spec = inst.p4.as_deref().unwrap_or("");
                let keys: Vec<(usize, bool)> = spec
                    .split(',')
                    .filter(|s| !s.is_empty())
                    .filter_map(|part| {
                        let (idx, dir) = part.split_once(':')?;
                        Some((
                            idx.trim().parse().ok()?,
                            !dir.trim().eq_ignore_ascii_case("desc"),
                        ))
                    })
                    .collect();

                if !keys.is_empty() {
                    // Sort results and their index tracking together so the
                    // two do not drift apart.
                    let track = self.result_row_indices.len() == self.results.len();
                    let mut order: Vec<usize> = (0..self.results.len()).collect();
                    order.sort_by(|&a, &b| {
                        for &(col, asc) in &keys {
                            let va = self.results[a].get(col).unwrap_or(&Value::Null);
                            let vb = self.results[b].get(col).unwrap_or(&Value::Null);
                            let ord = compare_values(va, vb);
                            let ord = if asc { ord } else { ord.reverse() };
                            if ord != std::cmp::Ordering::Equal {
                                return ord;
                            }
                        }
                        std::cmp::Ordering::Equal
                    });
                    self.results = order.iter().map(|&i| self.results[i].clone()).collect();
                    if track {
                        self.result_row_indices = order
                            .iter()
                            .map(|&i| self.result_row_indices[i].clone())
                            .collect();
                    }
                }
                Ok(ExecuteResult::Continue)
            }

            OpCode::Distinct => {
                // Remove duplicate rows from results
                // This is used by UNION (without ALL) to deduplicate combined results
                let mut seen = std::collections::HashSet::with_capacity(self.results.len());
                let mut unique_results = Vec::with_capacity(self.results.len());

                for row in &self.results {
                    // Create a hashable key from the row values
                    let key: Vec<String> = row.iter().map(|v| format!("{:?}", v)).collect();
                    let key_str = key.join("|");

                    if seen.insert(key_str) {
                        unique_results.push(row.clone());
                    }
                }

                self.results = unique_results;

                if self.verbose {
                    println!(
                        "DISTINCT: {} unique rows after deduplication",
                        self.results.len()
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Limit => {
                // Limit results to P1 rows, skip first P2 rows (post-processing opcode)
                // This is used after DISTINCT to apply LIMIT/OFFSET correctly
                let limit = inst.p1 as usize;
                let offset = inst.p2 as usize;

                // Index tracking is trimmed alongside the rows. It used to be
                // left untouched, so after a post-processing LIMIT the two
                // vectors disagreed and the debug validator reported a
                // mismatch on every limited query.
                let track = self.result_row_indices.len() == self.results.len();

                // First apply OFFSET - skip first N rows
                if offset > 0 && !self.results.is_empty() {
                    if offset >= self.results.len() {
                        self.results.clear();
                        if track {
                            self.result_row_indices.clear();
                        }
                    } else {
                        self.results = self.results.drain(offset..).collect();
                        if track {
                            self.result_row_indices =
                                self.result_row_indices.drain(offset..).collect();
                        }
                    }
                }

                // Then apply LIMIT - truncate to N rows
                if self.results.len() > limit {
                    self.results.truncate(limit);
                    if track && self.result_row_indices.len() > limit {
                        self.result_row_indices.truncate(limit);
                    }
                }

                if self.verbose {
                    println!(
                        "LIMIT: offset {} then truncated to {} rows",
                        offset,
                        self.results.len()
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Intersect => {
                // Keep only rows that exist in both left and right result sets
                // P1 contains the register with the count of left-side results (-1 means no marker)
                let left_count = if inst.p1 >= 0 {
                    match self.get_register(inst.p1 as usize)? {
                        Register::Integer(n) => n as usize,
                        _ => {
                            return Err(SqawkError::VmError(
                                "Invalid left count register for INTERSECT".to_string(),
                            ))
                        }
                    }
                } else {
                    // If no marker, assume half the results are from each side (fallback)
                    self.results.len() / 2
                };

                // Split results into left and right sets
                let left_rows: Vec<_> = self.results.iter().take(left_count).cloned().collect();
                let right_rows: Vec<_> = self.results.iter().skip(left_count).cloned().collect();

                // Create a set of right rows for fast lookup
                let right_set: std::collections::HashSet<_> = right_rows
                    .iter()
                    .map(|row| {
                        row.iter()
                            .map(|v| format!("{:?}", v))
                            .collect::<Vec<_>>()
                            .join("|")
                    })
                    .collect();

                // Keep only left rows that also appear in right
                let mut intersection = Vec::with_capacity(left_rows.len());
                let mut seen = std::collections::HashSet::with_capacity(left_rows.len());

                for row in &left_rows {
                    let key: Vec<String> = row.iter().map(|v| format!("{:?}", v)).collect();
                    let key_str = key.join("|");

                    if right_set.contains(&key_str) && seen.insert(key_str) {
                        intersection.push(row.clone());
                    }
                }

                self.results = intersection;

                if self.verbose {
                    println!("INTERSECT: {} rows in intersection", self.results.len());
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Except => {
                // Keep only rows from left that don't exist in right result set
                // P1 contains the register with the count of left-side results (-1 means no marker)
                let left_count = if inst.p1 >= 0 {
                    match self.get_register(inst.p1 as usize)? {
                        Register::Integer(n) => n as usize,
                        _ => {
                            return Err(SqawkError::VmError(
                                "Invalid left count register for EXCEPT".to_string(),
                            ))
                        }
                    }
                } else {
                    // If no marker, assume half the results are from each side (fallback)
                    self.results.len() / 2
                };

                // Split results into left and right sets
                let left_rows: Vec<_> = self.results.iter().take(left_count).cloned().collect();
                let right_rows: Vec<_> = self.results.iter().skip(left_count).cloned().collect();

                // Create a set of right rows for fast lookup
                let right_set: std::collections::HashSet<_> = right_rows
                    .iter()
                    .map(|row| {
                        row.iter()
                            .map(|v| format!("{:?}", v))
                            .collect::<Vec<_>>()
                            .join("|")
                    })
                    .collect();

                // Keep only left rows that don't appear in right
                let mut difference = Vec::with_capacity(left_rows.len());
                let mut seen = std::collections::HashSet::with_capacity(left_rows.len());

                for row in &left_rows {
                    let key: Vec<String> = row.iter().map(|v| format!("{:?}", v)).collect();
                    let key_str = key.join("|");

                    if !right_set.contains(&key_str) && seen.insert(key_str) {
                        difference.push(row.clone());
                    }
                }

                self.results = difference;

                if self.verbose {
                    println!("EXCEPT: {} rows in difference", self.results.len());
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::NullRow => {
                // Load NULL values into registers P1 through P1+P2-1
                // P1: Start register
                // P2: Number of registers to set to NULL
                let start_reg = inst.p1 as usize;
                let count = inst.p2 as usize;

                for i in 0..count {
                    self.set_register(start_reg + i, Register::Null)?;
                }

                if self.verbose {
                    println!(
                        "NullRow: Set registers {} to {} to NULL",
                        start_reg,
                        start_reg + count - 1
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::RewindInner => {
                // Rewind cursor P1 for inner loop of nested join
                // P1: Cursor index
                // P2: Jump address if table is empty
                let cursor_idx = inst.p1 as usize;
                let jump_addr = inst.p2 as usize;

                if let Some(cursor) = self.cursors.get_mut(&cursor_idx) {
                    if cursor.rewind() {
                        // Cursor has rows, continue
                        Ok(ExecuteResult::Continue)
                    } else {
                        // Table is empty, jump to P2
                        Ok(ExecuteResult::Jump(jump_addr))
                    }
                } else {
                    Err(SqawkError::VmError(format!(
                        "Cursor {} not found for RewindInner",
                        cursor_idx
                    )))
                }
            }

            OpCode::MarkMatch => {
                // Set register P1 to 1 to indicate a match was found
                let reg = inst.p1 as usize;
                self.set_register(reg, Register::Integer(1))?;

                if self.verbose {
                    println!("MarkMatch: Set r[{}] = 1", reg);
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::CheckMatch => {
                // If register P1 is 0 (no match), jump to P2
                // Then reset register P1 to 0 for next iteration
                let reg = inst.p1 as usize;
                let jump_addr = inst.p2 as usize;

                let should_jump = match self.get_register(reg)? {
                    Register::Integer(0) => true,
                    Register::Integer(_) => false,
                    Register::Null => true, // Treat NULL as no match
                    _ => true,              // Default to no match for unexpected types
                };

                // Reset the match flag for the next iteration
                self.set_register(reg, Register::Integer(0))?;

                if should_jump {
                    if self.verbose {
                        println!("CheckMatch: No match found, jumping to {}", jump_addr);
                    }
                    Ok(ExecuteResult::Jump(jump_addr))
                } else {
                    if self.verbose {
                        println!("CheckMatch: Match found, continuing");
                    }
                    Ok(ExecuteResult::Continue)
                }
            }

            OpCode::DecrJumpZero => {
                // Decrement register P1. If result is zero, jump to P2.
                // Used for LIMIT - when counter hits zero, stop outputting rows.
                // Note: LIMIT 0 is handled at compile time with a Goto.
                let reg = inst.p1 as usize;
                let jump_addr = inst.p2 as usize;

                let current = match self.get_register(reg)? {
                    Register::Integer(n) => n,
                    _ => 0,
                };

                let new_val = current - 1;
                self.set_register(reg, Register::Integer(new_val))?;

                if self.verbose {
                    println!(
                        "DecrJumpZero: r[{}] = {} -> {}, jump if zero to {}",
                        reg, current, new_val, jump_addr
                    );
                }

                if new_val == 0 {
                    Ok(ExecuteResult::Jump(jump_addr))
                } else {
                    Ok(ExecuteResult::Continue)
                }
            }

            OpCode::SorterOpen => {
                // Open a sorter with P1 = sorter ID, P2 = column count, P4 = sort spec
                let sorter_id = inst.p1 as usize;
                let col_count = inst.p2 as usize;
                let sort_spec = inst.p4.clone().unwrap_or_default();

                // Parse sort specification: "col_idx:asc,col_idx:desc,..."
                let sort_keys: Vec<(usize, bool)> = sort_spec
                    .split(',')
                    .filter_map(|spec| {
                        let parts: Vec<&str> = spec.split(':').collect();
                        if parts.len() == 2 {
                            let col_idx = parts[0].parse::<usize>().ok()?;
                            let ascending = parts[1] != "desc";
                            Some((col_idx, ascending))
                        } else {
                            None
                        }
                    })
                    .collect();

                let sorter = Sorter::new(col_count, sort_keys);
                self.sorters.insert(sorter_id, sorter);

                if self.verbose {
                    println!(
                        "SorterOpen: Created sorter {} with {} columns, spec: {}",
                        sorter_id, col_count, sort_spec
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::SorterInsert => {
                // Insert row into sorter P1. P2 = start register, P3 = column count
                let sorter_id = inst.p1 as usize;
                let start_reg = inst.p2 as usize;
                let col_count = inst.p3 as usize;

                // Collect values from registers
                let mut row = Vec::with_capacity(col_count);
                for i in 0..col_count {
                    let reg_idx = start_reg + i;
                    let value = Value::from(self.get_register(reg_idx)?);
                    row.push(value);
                }

                if let Some(sorter) = self.sorters.get_mut(&sorter_id) {
                    sorter.insert(row);
                    if self.verbose {
                        println!("SorterInsert: Added row to sorter {}", sorter_id);
                    }
                } else {
                    return Err(SqawkError::VmError(format!(
                        "Sorter {} not found",
                        sorter_id
                    )));
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::SorterSort => {
                // Sort the sorter P1
                let sorter_id = inst.p1 as usize;

                if let Some(sorter) = self.sorters.get_mut(&sorter_id) {
                    sorter.sort();
                    if self.verbose {
                        println!(
                            "SorterSort: Sorted {} rows in sorter {}",
                            sorter.rows.len(),
                            sorter_id
                        );
                    }
                } else {
                    return Err(SqawkError::VmError(format!(
                        "Sorter {} not found",
                        sorter_id
                    )));
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::SorterData => {
                // Copy current sorter P1 row to registers starting at P2. P3 = column count
                let sorter_id = inst.p1 as usize;
                let start_reg = inst.p2 as usize;
                let col_count = inst.p3 as usize;

                let (row_data, position) = {
                    let sorter = self.sorters.get(&sorter_id).ok_or_else(|| {
                        SqawkError::VmError(format!("Sorter {} not found", sorter_id))
                    })?;
                    let row = sorter.current_row().ok_or_else(|| {
                        SqawkError::VmError(format!("No current row in sorter {}", sorter_id))
                    })?;
                    (row.clone(), sorter.position)
                };

                for (i, value) in row_data.iter().take(col_count).enumerate() {
                    self.set_register(start_reg + i, Register::from(value.clone()))?;
                }

                if self.verbose {
                    println!(
                        "SorterData: Copied row {} from sorter {} to r[{}..{}]",
                        position,
                        sorter_id,
                        start_reg,
                        start_reg + col_count - 1
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::SorterNext => {
                // Advance sorter P1. Jump to P2 if more rows, else continue.
                let sorter_id = inst.p1 as usize;
                let jump_addr = inst.p2 as usize;

                if let Some(sorter) = self.sorters.get_mut(&sorter_id) {
                    if sorter.next() {
                        if self.verbose {
                            println!(
                                "SorterNext: Advanced to row {} in sorter {}, jumping to {}",
                                sorter.position, sorter_id, jump_addr
                            );
                        }
                        Ok(ExecuteResult::Jump(jump_addr))
                    } else {
                        if self.verbose {
                            println!(
                                "SorterNext: No more rows in sorter {}, continuing",
                                sorter_id
                            );
                        }
                        Ok(ExecuteResult::Continue)
                    }
                } else {
                    Err(SqawkError::VmError(format!(
                        "Sorter {} not found",
                        sorter_id
                    )))
                }
            }

            OpCode::OpenEphemeral => {
                // Open ephemeral cursor P1 with P2 columns. P4 = sort key spec.
                // Uses unified cursor model - Column/Next/Rewind work for ephemeral cursors.
                let cursor_id = inst.p1 as usize;
                let col_count = inst.p2 as usize;
                let sort_spec = inst.p4.clone().unwrap_or_default();

                // Parse sort specification: "col_idx:asc,col_idx:desc,..."
                let sort_keys: Vec<(usize, bool)> = sort_spec
                    .split(',')
                    .filter_map(|spec| {
                        let parts: Vec<&str> = spec.split(':').collect();
                        if parts.len() == 2 {
                            let col_idx = parts[0].parse::<usize>().ok()?;
                            let ascending = parts[1] != "desc";
                            Some((col_idx, ascending))
                        } else {
                            None
                        }
                    })
                    .collect();

                // Create ephemeral cursor in the unified cursors map
                let cursor = Cursor::ephemeral(col_count, sort_keys);
                self.cursors.insert(cursor_id, cursor);

                if self.verbose {
                    println!(
                        "OpenEphemeral: Created ephemeral cursor {} with {} columns, spec: {}",
                        cursor_id, col_count, sort_spec
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::IdxInsert => {
                // Insert into ephemeral cursor P1. P2 = start register, P3 = column count.
                let cursor_id = inst.p1 as usize;
                let start_reg = inst.p2 as usize;
                let col_count = inst.p3 as usize;

                // Collect values from registers
                let mut row = Vec::with_capacity(col_count);
                for i in 0..col_count {
                    let reg_idx = start_reg + i;
                    let value = Value::from(self.get_register(reg_idx)?);
                    row.push(value);
                }

                if let Some(cursor) = self.cursors.get_mut(&cursor_id) {
                    cursor.insert_row(row)?;
                    if self.verbose {
                        println!("IdxInsert: Added row to cursor {}", cursor_id);
                    }
                } else {
                    return Err(SqawkError::VmError(format!(
                        "Cursor {} not found",
                        cursor_id
                    )));
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::Sort => {
                // Sort ephemeral cursor P1. Jump to P2 if empty, else continue.
                let cursor_id = inst.p1 as usize;
                let jump_if_empty = inst.p2 as usize;

                if let Some(cursor) = self.cursors.get_mut(&cursor_id) {
                    let has_rows = cursor.sort()?;
                    if has_rows {
                        if self.verbose {
                            println!(
                                "Sort: Sorted {} rows in cursor {}",
                                cursor.row_count(),
                                cursor_id
                            );
                        }
                        Ok(ExecuteResult::Continue)
                    } else {
                        if self.verbose {
                            println!(
                                "Sort: Cursor {} is empty, jumping to {}",
                                cursor_id, jump_if_empty
                            );
                        }
                        Ok(ExecuteResult::Jump(jump_if_empty))
                    }
                } else {
                    Err(SqawkError::VmError(format!(
                        "Cursor {} not found",
                        cursor_id
                    )))
                }
            }

            OpCode::Sequence => {
                // P1 = ephemeral cursor, P2 = dest register. Generate next sequence number.
                let cursor_id = inst.p1 as usize;
                let dest_reg = inst.p2 as usize;

                if let Some(cursor) = self.cursors.get_mut(&cursor_id) {
                    let seq = cursor.next_sequence()?;
                    self.set_register(dest_reg, Register::Integer(seq))?;
                    if self.verbose {
                        println!(
                            "Sequence: Generated sequence {} from cursor {} into r[{}]",
                            seq, cursor_id, dest_reg
                        );
                    }
                } else {
                    return Err(SqawkError::VmError(format!(
                        "Cursor {} not found",
                        cursor_id
                    )));
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::AggStep => {
                // Step an aggregate function
                // P1 = function type (0=COUNT, 1=SUM, 2=AVG, 3=MIN, 4=MAX)
                // P2 = value register
                // P3 = accumulator register (used as key for accumulator storage)
                let func_type = inst.p1 & AGG_TYPE_MASK;
                let is_distinct = inst.p1 & AGG_DISTINCT != 0;
                let value_reg = inst.p2 as usize;
                let acc_reg = inst.p3 as usize;

                // Get the value to accumulate (skip for COUNT(*) where p2=-1)
                let value = if inst.p2 >= 0 {
                    Some(Value::from(self.get_register(value_reg)?))
                } else {
                    None
                };

                // Check if the accumulator register is Null - if so, reset the accumulator
                // This matches SQLite's behavior where Null initialization signals a fresh start
                let acc_reg_value = self.get_register(acc_reg)?;
                if matches!(acc_reg_value, Register::Null) {
                    // Reset the accumulator for this register
                    self.accumulators.insert(
                        acc_reg,
                        AggAccumulator {
                            func_type,
                            count: 0,
                            value: None,
                            seen: if is_distinct {
                                Some(std::collections::HashSet::new())
                            } else {
                                None
                            },
                        },
                    );
                    // Mark the register as non-Null to indicate accumulator is active
                    self.registers[acc_reg] = Register::Integer(0);
                }

                // Get or create the accumulator
                let acc = self.accumulators.entry(acc_reg).or_insert(AggAccumulator {
                    func_type,
                    count: 0,
                    value: None,
                    seen: if is_distinct {
                        Some(std::collections::HashSet::new())
                    } else {
                        None
                    },
                });

                // DISTINCT: ignore a value already accumulated for this group.
                // NULLs are not tracked -- every aggregate already skips them.
                if is_distinct {
                    if let Some(val) = value.as_ref() {
                        if !matches!(val, Value::Null) {
                            let key = format!("{:?}", val);
                            let seen = acc.seen.get_or_insert_with(Default::default);
                            if !seen.insert(key) {
                                return Ok(ExecuteResult::Continue);
                            }
                        }
                    }
                }

                match func_type {
                    AGG_COUNT => {
                        // COUNT - count non-NULL values (or all rows if value is None)
                        if value.is_none() || !matches!(value.as_ref(), Some(Value::Null)) {
                            acc.count += 1;
                        }
                    }
                    AGG_SUM => {
                        // SUM
                        if let Some(val) = value {
                            if !matches!(val, Value::Null) {
                                acc.count += 1;
                                acc.value = Some(match (&acc.value, &val) {
                                    (None, v) => v.clone(),
                                    (Some(Value::Integer(a)), Value::Integer(b)) => {
                                        Value::Integer(a + b)
                                    }
                                    (Some(Value::Float(a)), Value::Float(b)) => Value::Float(a + b),
                                    (Some(Value::Integer(a)), Value::Float(b)) => {
                                        Value::Float(*a as f64 + b)
                                    }
                                    (Some(Value::Float(a)), Value::Integer(b)) => {
                                        Value::Float(a + *b as f64)
                                    }
                                    (Some(existing), _) => existing.clone(),
                                });
                            }
                        }
                    }
                    AGG_AVG => {
                        // AVG (accumulate sum, divide in AggFinal)
                        if let Some(val) = value {
                            if !matches!(val, Value::Null) {
                                acc.count += 1;
                                acc.value = Some(match (&acc.value, &val) {
                                    (None, v) => v.clone(),
                                    (Some(Value::Integer(a)), Value::Integer(b)) => {
                                        Value::Integer(a + b)
                                    }
                                    (Some(Value::Float(a)), Value::Float(b)) => Value::Float(a + b),
                                    (Some(Value::Integer(a)), Value::Float(b)) => {
                                        Value::Float(*a as f64 + b)
                                    }
                                    (Some(Value::Float(a)), Value::Integer(b)) => {
                                        Value::Float(a + *b as f64)
                                    }
                                    (Some(existing), _) => existing.clone(),
                                });
                            }
                        }
                    }
                    AGG_MIN => {
                        // MIN
                        if let Some(val) = value {
                            if !matches!(val, Value::Null) {
                                acc.count += 1;
                                acc.value = Some(match &acc.value {
                                    None => val,
                                    Some(existing) => {
                                        if compare_values(&val, existing)
                                            == std::cmp::Ordering::Less
                                        {
                                            val
                                        } else {
                                            existing.clone()
                                        }
                                    }
                                });
                            }
                        }
                    }
                    AGG_MAX => {
                        // MAX
                        if let Some(val) = value {
                            if !matches!(val, Value::Null) {
                                acc.count += 1;
                                acc.value = Some(match &acc.value {
                                    None => val,
                                    Some(existing) => {
                                        if compare_values(&val, existing)
                                            == std::cmp::Ordering::Greater
                                        {
                                            val
                                        } else {
                                            existing.clone()
                                        }
                                    }
                                });
                            }
                        }
                    }
                    _ => {
                        return Err(SqawkError::VmError(format!(
                            "Unknown aggregate function type: {}",
                            func_type
                        )));
                    }
                }

                if self.verbose {
                    println!(
                        "AggStep: func={} acc_reg={} count={}",
                        func_type, acc_reg, acc.count
                    );
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::AggFinal => {
                // Finalize an aggregate and store result
                // P1 = accumulator register
                // P2 = result register
                // P3 = function type (0=COUNT, 1=SUM, 2=AVG, 3=MIN, 4=MAX) - used when no accumulator exists
                let acc_reg = inst.p1 as usize;
                let result_reg = inst.p2 as usize;
                let func_type_hint = inst.p3; // Fallback function type

                // First check if the accumulator register is Null - this means no AggStep was called
                // (the register was initialized to Null and never updated by AggStep)
                let acc_reg_value = self.get_register(acc_reg)?;
                let result = if matches!(acc_reg_value, Register::Null) {
                    // No AggStep was called this iteration - return default based on function type
                    if func_type_hint == AGG_COUNT {
                        Register::Integer(0) // COUNT of no rows is 0
                    } else {
                        Register::Null // SUM/AVG/MIN/MAX of no rows is NULL
                    }
                } else if let Some(acc) = self.accumulators.get(&acc_reg) {
                    match acc.func_type {
                        AGG_COUNT => {
                            // COUNT - return the count
                            Register::Integer(acc.count)
                        }
                        AGG_SUM => {
                            // SUM - return the accumulated sum
                            match &acc.value {
                                Some(Value::Integer(i)) => Register::Integer(*i),
                                Some(Value::Float(f)) => Register::Float(*f),
                                _ => Register::Null,
                            }
                        }
                        AGG_AVG => {
                            // AVG - divide sum by count
                            if acc.count == 0 {
                                Register::Null
                            } else {
                                match &acc.value {
                                    Some(Value::Integer(i)) => {
                                        Register::Float(*i as f64 / acc.count as f64)
                                    }
                                    Some(Value::Float(f)) => Register::Float(f / acc.count as f64),
                                    _ => Register::Null,
                                }
                            }
                        }
                        AGG_MIN | AGG_MAX => {
                            // MIN/MAX - return the accumulated value
                            match &acc.value {
                                Some(Value::Integer(i)) => Register::Integer(*i),
                                Some(Value::Float(f)) => Register::Float(*f),
                                Some(Value::String(s)) => Register::String(s.clone().into_owned()),
                                Some(Value::Boolean(b)) => Register::Boolean(*b),
                                _ => Register::Null,
                            }
                        }
                        _ => Register::Null,
                    }
                } else {
                    // No accumulator found - return appropriate default based on function type
                    // COUNT with no rows returns 0, others return NULL
                    if func_type_hint == AGG_COUNT {
                        Register::Integer(0) // COUNT of no rows is 0
                    } else {
                        Register::Null
                    }
                };

                self.set_register(result_reg, result)?;

                if self.verbose {
                    println!("AggFinal: acc_reg={} result_reg={}", acc_reg, result_reg);
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::AggReset => {
                // Reset accumulator at register P1
                let acc_reg = inst.p1 as usize;

                // Remove the accumulator so it will be re-initialized on next AggStep
                self.accumulators.remove(&acc_reg);

                if self.verbose {
                    println!("AggReset: cleared accumulator at reg {}", acc_reg);
                }

                Ok(ExecuteResult::Continue)
            }

            OpCode::InitCoroutine => {
                // InitCoroutine P1, P2, P3
                // P1 = register to store coroutine return/entry address
                // P2 = address to jump to (skip over coroutine body)
                // P3 = coroutine entry point (first instruction of coroutine)
                let return_reg = inst.p1 as usize;
                let skip_addr = inst.p2 as usize;
                let entry_point = inst.p3 as usize;

                // Store the entry point in the register - this will be swapped with PC on Yield
                self.registers[return_reg] = Register::Integer(entry_point as i64);

                // Jump over the coroutine body to continue main execution
                Ok(ExecuteResult::Jump(skip_addr))
            }

            OpCode::Yield => {
                // Yield P1
                // P1 = register containing coroutine PC (swap with current PC)
                // Cooperative context switch between main code and coroutine
                let coro_reg = inst.p1 as usize;

                // Get saved coroutine PC from register
                let saved_pc = match &self.registers[coro_reg] {
                    Register::Integer(pc) => *pc as usize,
                    _ => {
                        return Err(SqawkError::VmError(
                            "Yield: register must contain integer PC".into(),
                        ))
                    }
                };

                // Save current PC + 1 (next instruction) into the register
                // This is where we'll return after EndCoroutine
                self.registers[coro_reg] = Register::Integer((self.pc + 1) as i64);

                // Jump to the saved coroutine location
                Ok(ExecuteResult::Jump(saved_pc))
            }

            OpCode::EndCoroutine => {
                // EndCoroutine P1
                // P1 = register containing return address (saved by last Yield)
                // Jump back to caller, prepare register for next Yield
                let coro_reg = inst.p1 as usize;

                // Get return address from register
                let return_addr = match &self.registers[coro_reg] {
                    Register::Integer(pc) => *pc as usize,
                    _ => {
                        return Err(SqawkError::VmError(
                            "EndCoroutine: register must contain integer return address".into(),
                        ))
                    }
                };

                // Store next instruction (PC + 1) for the next Yield call
                // This will be the coroutine's resume point
                self.registers[coro_reg] = Register::Integer((self.pc + 1) as i64);

                // Jump back to caller
                Ok(ExecuteResult::Jump(return_addr))
            }

            OpCode::Once => {
                // Once P1, P2
                // First time: continue to next instruction
                // Subsequently: jump to P2
                // Uses instruction address as unique key for tracking
                let skip_addr = inst.p2 as usize;

                if self.once_flags.get(&self.pc).copied().unwrap_or(false) {
                    // Already executed, skip
                    Ok(ExecuteResult::Jump(skip_addr))
                } else {
                    // First time, mark as executed and continue
                    self.once_flags.insert(self.pc, true);
                    Ok(ExecuteResult::Continue)
                }
            }

            OpCode::Exists => {
                // Exists P1, P2
                // P1 = result register (set to 1 if cursor P2 has rows, else 0)
                // P2 = cursor ID to check
                let result_reg = inst.p1 as usize;
                let cursor_idx = inst.p2 as usize;

                let has_rows = if let Some(cursor) = self.cursors.get(&cursor_idx) {
                    cursor.row_count() > 0
                } else {
                    false
                };

                self.registers[result_reg] = Register::Integer(if has_rows { 1 } else { 0 });
                Ok(ExecuteResult::Continue)
            }

            OpCode::NotExists => {
                // NotExists P1, P2
                // P1 = result register (set to 1 if cursor P2 has no rows, else 0)
                // P2 = cursor ID to check
                let result_reg = inst.p1 as usize;
                let cursor_idx = inst.p2 as usize;

                let has_rows = if let Some(cursor) = self.cursors.get(&cursor_idx) {
                    cursor.row_count() > 0
                } else {
                    false
                };

                self.registers[result_reg] = Register::Integer(if has_rows { 0 } else { 1 });
                Ok(ExecuteResult::Continue)
            }

            #[allow(unreachable_patterns)]
            _ => {
                // This code is unreachable since all opcodes are handled above,
                // but we keep it for future-proofing and to silence compiler warnings
                Err(SqawkError::UnsupportedSqlFeature(format!(
                    "Unsupported VM opcode: {:?}",
                    inst.opcode
                )))
            }
        }
    }

    // Removed unused methods get_results, has_results, get_column_names,
    // get_affected_rows, and get_modified_tables

    /// Create a table from the results using the result schema
    pub fn create_result_table(&self) -> SqawkResult<Option<Table>> {
        let schema = &self.program.result_schema;

        // If we have no results but DO have a schema, create an empty table with headers
        // This handles SELECT from empty tables
        if self.results.is_empty() {
            if !schema.is_empty() {
                // Build column definitions from schema
                let col_defs: Vec<ColumnDefinition> = schema
                    .columns
                    .iter()
                    .map(|col| ColumnDefinition {
                        name: col.name.clone(),
                        data_type: col.data_type,
                    })
                    .collect();
                let table = Table::new_with_schema("result", col_defs, None, None);
                return Ok(Some(table));
            }
            return Ok(None);
        }

        // Build column definitions from schema, filling in generic names if schema is incomplete
        let col_count = self.results[0].len();
        let col_defs: Vec<ColumnDefinition> = (0..col_count)
            .map(|i| {
                if i < schema.columns.len() {
                    ColumnDefinition {
                        name: schema.columns[i].name.clone(),
                        data_type: schema.columns[i].data_type,
                    }
                } else {
                    ColumnDefinition {
                        name: format!("col{}", i),
                        data_type: crate::table::DataType::Text,
                    }
                }
            })
            .collect();

        let mut table = Table::new_with_schema("result", col_defs, None, None);

        // Add rows
        for row in &self.results {
            table.add_row(row.clone())?;
        }

        // Phase 4D: Debug validation - verify index tracking is correct
        // Only validate in debug builds to avoid performance impact
        #[cfg(debug_assertions)]
        self.validate_index_tracking();

        Ok(Some(table))
    }

    /// Validate that the index-based result tracking matches the materialized results
    ///
    /// This method is used in debug builds to ensure the index tracking in Phase 4C
    /// is producing correct results before we can fully switch to index-only mode.
    #[cfg(debug_assertions)]
    fn validate_index_tracking(&self) {
        // Skip validation if we have no results or no index tracking
        if self.results.is_empty() || self.result_row_indices.is_empty() {
            return;
        }

        // Verify counts match
        if self.results.len() != self.result_row_indices.len() {
            eprintln!(
                "Index tracking mismatch: {} results vs {} index entries",
                self.results.len(),
                self.result_row_indices.len()
            );
            return;
        }

        // For single-table queries (one cursor), validate that indices point to correct rows
        if self.cursor_source_order.len() == 1 {
            let cursor_id = self.cursor_source_order[0];
            if let Some(cursor) = self.cursors.get(&cursor_id) {
                let rows = cursor.rows();

                for row_indices in self.result_row_indices.iter() {
                    if let Some(Some(row_idx)) = row_indices.first() {
                        // Verify the row exists
                        if *row_idx < rows.len() {
                            // The index tracking is recording correctly
                            // (We can't fully validate column values here since ResultRow
                            // may include computed expressions, not just column references)
                        }
                    }
                }
            }
        }

        // For multi-table queries (JOINs), verify the index structure is consistent
        if self.cursor_source_order.len() > 1 {
            // Verify all row_indices entries have the expected number of cursor entries
            for row_indices in &self.result_row_indices {
                if row_indices.len() != self.cursor_source_order.len() {
                    eprintln!(
                        "JOIN index tracking mismatch: expected {} cursor indices, got {}",
                        self.cursor_source_order.len(),
                        row_indices.len()
                    );
                }
            }
        }
    }

    /// Materialize a JOIN row from tracked indices (Phase 4E)
    ///
    /// This method reconstructs a result row by reading values directly from
    /// source tables using the tracked row indices. This is more memory-efficient
    /// than storing cloned row data for each result.
    ///
    /// # Arguments
    /// * `row_idx` - Index into result_row_indices
    ///
    /// # Returns
    /// The materialized row, or None if indices are invalid
    #[allow(dead_code)]
    fn materialize_join_row_from_indices(
        &self,
        row_idx: usize,
    ) -> Option<Vec<crate::table::Value>> {
        let indices = self.result_row_indices.get(row_idx)?;
        // Estimate total column count from all cursors
        let total_cols: usize = self
            .cursor_source_order
            .iter()
            .filter_map(|id| self.cursors.get(id))
            .map(|c| c.column_count())
            .sum();
        let mut row = Vec::with_capacity(total_cols);

        // For each cursor in source order, get the row and append its columns
        for (cursor_pos, cursor_id) in self.cursor_source_order.iter().enumerate() {
            if let Some(cursor) = self.cursors.get(cursor_id) {
                let source_row_idx = indices.get(cursor_pos).copied().flatten();

                if let Some(row_idx) = source_row_idx {
                    // Get row from source cursor
                    let cursor_rows = cursor.rows();
                    if let Some(source_row) = cursor_rows.get(row_idx) {
                        row.extend(source_row.iter().cloned());
                    } else {
                        // Row index out of bounds - fill with NULLs
                        for _ in 0..cursor.column_count() {
                            row.push(crate::table::Value::Null);
                        }
                    }
                } else {
                    // NULL row (outer join) - fill with NULLs
                    for _ in 0..cursor.column_count() {
                        row.push(crate::table::Value::Null);
                    }
                }
            }
        }

        Some(row)
    }

    /// Get the number of cursors (tables) involved in the current query
    #[allow(dead_code)]
    fn cursor_count(&self) -> usize {
        self.cursor_source_order.len()
    }

    /// Check if this is a JOIN query (multiple cursors)
    #[allow(dead_code)]
    fn is_join_query(&self) -> bool {
        self.cursor_source_order.len() > 1
    }

    /// Get a register value by index
    ///
    /// # Arguments
    /// * `idx` - The register index
    ///
    /// # Returns
    /// The register value or an error if the index is out of bounds
    /// Whether a register counts as true.
    ///
    /// Matches IfZ's notion of "zero" exactly, so `Not` and the conditional
    /// jumps cannot disagree about what a value means.
    fn register_is_truthy(reg: &Register) -> bool {
        match reg {
            Register::Integer(v) => *v != 0,
            Register::Float(v) => *v != 0.0,
            Register::String(v) => !v.is_empty(),
            Register::Boolean(v) => *v,
            Register::Null => false,
        }
    }

    fn get_register(&self, idx: usize) -> SqawkResult<Register> {
        if idx < self.registers.len() {
            Ok(self.registers[idx].clone())
        } else {
            Err(SqawkError::VmError(format!(
                "Register index out of bounds: {}",
                idx
            )))
        }
    }

    /// Set a register value by index
    ///
    /// # Arguments
    /// * `idx` - The register index
    /// * `value` - The value to set
    ///
    /// # Returns
    /// Ok(()) if successful, or an error if the index is out of bounds
    fn set_register(&mut self, idx: usize, value: Register) -> SqawkResult<()> {
        // Ensure the register exists
        while idx >= self.registers.len() {
            self.registers.push(Register::Null);
        }

        // Set the value
        self.registers[idx] = value;

        Ok(())
    }

    /// Execute a comparison operation between two registers
    ///
    /// # Arguments
    /// * `inst` - The instruction containing:
    ///   - p1: left operand register index
    ///   - p2: right operand register index
    ///   - p3: destination register for result (1 for true, 0 for false)
    ///   - opcode: the comparison type (Lt, Le, Gt, Ge, Eq, Ne)
    fn execute_comparison(&mut self, inst: &Instruction) -> SqawkResult<ExecuteResult> {
        use std::cmp::Ordering;

        let reg1 = self.get_register(inst.p1 as usize)?;
        let reg2 = self.get_register(inst.p2 as usize)?;

        // SQL three-valued logic: a comparison involving NULL is UNKNOWN, not
        // an error and not false. The result register holds NULL for UNKNOWN.
        //
        // IfZ already treats a NULL register as zero, so every WHERE, HAVING
        // and ON site rejects an UNKNOWN row without any control-flow change.
        let result = match Self::sql_compare(&reg1, &reg2) {
            None => Register::Null,
            Some(ord) => {
                let truth = match inst.opcode {
                    OpCode::Lt => ord == Ordering::Less,
                    OpCode::Le => ord != Ordering::Greater,
                    OpCode::Gt => ord == Ordering::Greater,
                    OpCode::Ge => ord != Ordering::Less,
                    OpCode::Eq => ord == Ordering::Equal,
                    OpCode::Ne => ord != Ordering::Equal,
                    _ => {
                        return Err(SqawkError::VmError(format!(
                            "Invalid comparison opcode: {:?}",
                            inst.opcode
                        )))
                    }
                };
                Register::Integer(if truth { 1 } else { 0 })
            }
        };

        self.set_register(inst.p3 as usize, result)?;
        Ok(ExecuteResult::Continue)
    }

    /// SQL comparison of two registers.
    ///
    /// `None` means UNKNOWN, which is the result whenever either operand is
    /// NULL -- including NULL against NULL. (Grouping and sorting need the
    /// opposite, NULL equal to NULL, which is why the Compare opcode has its
    /// own internal ordering rather than sharing this.)
    ///
    /// Mixed string/number operands are coerced when the string parses as a
    /// number, and compared as text otherwise. This is the awk-shaped choice
    /// rather than the SQLite one, for two reasons: a CSV column is untyped
    /// text that sqawk guesses a type for PER CELL, so a single column can
    /// hold both; and arithmetic already coerces this way, so `x + '1'` and
    /// `x > '1'` agree instead of disagreeing.
    fn sql_compare(a: &Register, b: &Register) -> Option<std::cmp::Ordering> {
        // Any NULL operand yields UNKNOWN.
        if matches!(a, Register::Null) || matches!(b, Register::Null) {
            return None;
        }

        fn as_number(reg: &Register) -> Option<f64> {
            match reg {
                Register::Integer(i) => Some(*i as f64),
                Register::Float(f) => Some(*f),
                Register::Boolean(v) => Some(if *v { 1.0 } else { 0.0 }),
                Register::String(s) => s.trim().parse::<f64>().ok(),
                Register::Null => None,
            }
        }

        match (a, b) {
            // Integer against integer compares as i64, not through f64, so
            // values above 2^53 stay exact.
            (Register::Integer(x), Register::Integer(y)) => Some(x.cmp(y)),
            (Register::String(x), Register::String(y)) => Some(x.cmp(y)),
            (Register::Boolean(x), Register::Boolean(y)) => Some(x.cmp(y)),
            _ => match (as_number(a), as_number(b)) {
                (Some(x), Some(y)) => x.partial_cmp(&y),
                // A string that does not parse as a number falls back to a
                // textual comparison against the other side's rendering.
                _ => {
                    let x: Value = a.clone().into();
                    let y: Value = b.clone().into();
                    Some(x.to_string().cmp(&y.to_string()))
                }
            },
        }
    }

    /// Perform an arithmetic operation on two register values
    ///
    /// # Arguments
    /// * `left_reg` - Left operand register index
    /// * `right_reg` - Right operand register index
    /// * `op` - The operator: "+", "-", "*", "/", "%"
    ///
    /// # Returns
    /// The result as a Register value
    fn arithmetic_op(&self, left_reg: usize, right_reg: usize, op: &str) -> SqawkResult<Register> {
        let left = self.get_register(left_reg)?;
        let right = self.get_register(right_reg)?;

        // Handle NULL propagation
        if matches!(left, Register::Null) || matches!(right, Register::Null) {
            return Ok(Register::Null);
        }

        // Convert to numbers
        let left_num = match left {
            Register::Integer(i) => i as f64,
            Register::Float(f) => f,
            Register::String(s) => s.parse::<f64>().unwrap_or(0.0),
            Register::Boolean(b) => {
                if b {
                    1.0
                } else {
                    0.0
                }
            }
            Register::Null => unreachable!(),
        };

        let right_num = match right {
            Register::Integer(i) => i as f64,
            Register::Float(f) => f,
            Register::String(s) => s.parse::<f64>().unwrap_or(0.0),
            Register::Boolean(b) => {
                if b {
                    1.0
                } else {
                    0.0
                }
            }
            Register::Null => unreachable!(),
        };

        let result = match op {
            "+" => left_num + right_num,
            "-" => left_num - right_num,
            "*" => left_num * right_num,
            "/" => {
                if right_num == 0.0 {
                    return Ok(Register::Null); // Division by zero returns NULL
                }
                left_num / right_num
            }
            "%" => {
                if right_num == 0.0 {
                    return Ok(Register::Null); // Modulo by zero returns NULL
                }
                left_num % right_num
            }
            _ => {
                return Err(SqawkError::VmError(format!(
                    "Unknown arithmetic operator: {}",
                    op
                )));
            }
        };

        // Preserve integer type if possible
        if result.fract() == 0.0 && result >= i64::MIN as f64 && result <= i64::MAX as f64 {
            Ok(Register::Integer(result as i64))
        } else {
            Ok(Register::Float(result))
        }
    }

    /// Convert a SQL LIKE pattern to a regex pattern
    ///
    /// SQL LIKE uses:
    /// - % to match any sequence of zero or more characters
    /// - _ to match any single character
    ///
    /// # Arguments
    /// * `pattern` - The SQL LIKE pattern
    /// * `case_insensitive` - Whether to perform case-insensitive matching (for ILIKE)
    ///
    /// # Returns
    /// A regex pattern string
    fn like_pattern_to_regex(&self, pattern: &str, case_insensitive: bool) -> String {
        let mut regex = String::new();

        // Add case-insensitive flag if needed
        if case_insensitive {
            regex.push_str("(?i)");
        }

        // Anchor at start
        regex.push('^');

        // Convert each character
        let chars: Vec<char> = pattern.chars().collect();
        let mut i = 0;
        while i < chars.len() {
            match chars[i] {
                '%' => regex.push_str(".*"),
                '_' => regex.push('.'),
                // Escape regex special characters
                '.' | '*' | '+' | '?' | '^' | '$' | '(' | ')' | '[' | ']' | '{' | '}' | '|'
                | '\\' => {
                    regex.push('\\');
                    regex.push(chars[i]);
                }
                c => regex.push(c),
            }
            i += 1;
        }

        // Anchor at end
        regex.push('$');

        regex
    }

    /// Convert a GLOB pattern to a regex pattern
    ///
    /// GLOB uses Unix-style wildcards:
    /// - * to match any sequence of zero or more characters
    /// - ? to match any single character
    /// - [...] for character classes
    ///
    /// # Arguments
    /// * `pattern` - The GLOB pattern
    ///
    /// # Returns
    /// A regex pattern string
    fn glob_pattern_to_regex(&self, pattern: &str) -> String {
        let mut regex = String::new();

        // Anchor at start
        regex.push('^');

        // Convert each character
        let chars: Vec<char> = pattern.chars().collect();
        let mut i = 0;
        while i < chars.len() {
            match chars[i] {
                '*' => regex.push_str(".*"),
                '?' => regex.push('.'),
                '[' => {
                    // Pass through character class as-is (with some escaping)
                    regex.push('[');
                    i += 1;
                    while i < chars.len() && chars[i] != ']' {
                        if chars[i] == '\\' && i + 1 < chars.len() {
                            regex.push('\\');
                            regex.push(chars[i + 1]);
                            i += 2;
                        } else {
                            regex.push(chars[i]);
                            i += 1;
                        }
                    }
                    if i < chars.len() {
                        regex.push(']');
                    }
                }
                // Escape regex special characters (except those handled above)
                '.' | '+' | '^' | '$' | '(' | ')' | '{' | '}' | '|' | '\\' => {
                    regex.push('\\');
                    regex.push(chars[i]);
                }
                c => regex.push(c),
            }
            i += 1;
        }

        // Anchor at end
        regex.push('$');

        regex
    }
}

/// Result of executing an instruction
enum ExecuteResult {
    /// Continue to next instruction
    Continue,
    /// Jump to specified address
    Jump(usize),
    /// Halt execution
    Halt,
}