radixdb-executor 1.1.0

// Copyright 2026 RadixDB Contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// CompiledEvaluator Bridge
//
// Provides an Evaluator-compatible API using the Expression VM internally.
// This allows gradual migration from AST-based evaluation to bytecode execution.
//
// Design:
// - Matches Evaluator's public API (new, init_columns, set_row_array, evaluate)
// - Uses per-evaluator local cache for compiled programs
// - Uses ExprVM for execution
//
// Performance Optimization:
// - For closure-based filtering, use `RowFilter` instead of creating evaluators per-row
// - `RowFilter` pre-compiles the expression once and shares `CompactArc<Program>` across threads
// - The VM is lightweight and can be created per-thread without performance penalty

use std::hash::{Hash, Hasher};
use std::num::NonZeroUsize;
use std::sync::Arc;

use lru::LruCache;
use parking_lot::Mutex;
use radixdb_core::ParamVec;
use radixdb_core::{CompactArc, StringMap};
use rustc_hash::{FxHashMap, FxHasher};

use super::compiler::{CompileContext, ExprCompiler};
use super::execution_context::ExecuteContext;
use super::program::Program;
use super::vm::ExprVM;
use radixdb_core::{Error, Result, Row, Value};
use radixdb_functions::{global_registry, FunctionRegistry};
use radixdb_sql::ast::Expression;

use crate::context::{ExecutionContext, StoredFunctionInvoker};

// ============================================================================
// PROGRAM CACHE - Global cache for compiled expression programs
// ============================================================================

/// Maximum number of cached programs (LRU eviction)
const PROGRAM_CACHE_SIZE: usize = 256;

/// Global cache for compiled programs using O(1) LRU eviction.
/// Uses parking_lot::Mutex for efficient locking.
#[derive(Clone)]
struct ProgramCacheEntry {
    expression: Expression,
    columns: Vec<String>,
    registry_generation: u64,
    program: SharedProgram,
}

#[derive(Clone)]
struct LocalProgramCacheEntry {
    expression: Expression,
    registry_generation: u64,
    program: SharedProgram,
}

static PROGRAM_CACHE: Mutex<Option<LruCache<u64, ProgramCacheEntry>>> = Mutex::new(None);

/// Clear the program cache. Call on database drop to release memory.
pub fn clear_program_cache() {
    let mut guard = PROGRAM_CACHE.lock();
    *guard = None;
}

fn checked_alias_map(aliases: &[(String, usize)]) -> Result<StringMap<u16>> {
    aliases
        .iter()
        .map(|(name, index)| {
            let index = u16::try_from(*index).map_err(|_| {
                Error::invalid_argument(format!(
                    "expression alias '{}' index {} exceeds the u16 bytecode limit",
                    name, index
                ))
            })?;
            Ok((name.to_lowercase(), index))
        })
        .collect()
}

/// Compute cache key from expression and columns using efficient recursive hashing.
/// This avoids the overhead of Debug formatting by directly hashing expression structure.
/// Uses FxHasher which is 2-5x faster than SipHash for small keys.
fn compute_cache_key(expr: &Expression, columns: &[String], registry_generation: u64) -> u64 {
    let mut hasher = FxHasher::default();
    // Use efficient recursive hashing (same as CompiledEvaluator::hash_expression)
    hash_expression(expr, &mut hasher);
    // Hash column names
    columns.hash(&mut hasher);
    registry_generation.hash(&mut hasher);
    hasher.finish()
}

/// Compute a u64 hash of an expression without string allocation.
/// This is O(expression_size) and avoids Debug formatting overhead.
/// Use this for cache keys instead of format!("{:?}", expr).
/// Uses FxHasher which is 2-5x faster than SipHash for small keys.
#[inline]
pub fn compute_expression_hash(expr: &Expression) -> u64 {
    let mut hasher = FxHasher::default();
    hash_expression(expr, &mut hasher);
    hasher.finish()
}

/// Recursively hash an expression without string allocation.
/// This is O(expression_size) and avoids Debug formatting overhead.
fn hash_expression(expr: &Expression, hasher: &mut FxHasher) {
    // First hash the discriminant to distinguish variants
    std::mem::discriminant(expr).hash(hasher);

    match expr {
        Expression::Identifier(id) => {
            id.value_lower.hash(hasher);
        }
        Expression::QualifiedIdentifier(qid) => {
            qid.qualifier.value_lower.hash(hasher);
            qid.name.value_lower.hash(hasher);
        }
        Expression::IntegerLiteral(lit) => {
            lit.value.hash(hasher);
        }
        Expression::FloatLiteral(lit) => {
            lit.value.to_bits().hash(hasher);
        }
        Expression::StringLiteral(lit) => {
            lit.value.hash(hasher);
            lit.type_hint.hash(hasher);
        }
        Expression::BooleanLiteral(lit) => {
            lit.value.hash(hasher);
        }
        Expression::NullLiteral(_) => {
            // Just discriminant is enough
        }
        Expression::BoundValue(value) => {
            value.hash(hasher);
        }
        Expression::IntervalLiteral(lit) => {
            lit.value.hash(hasher);
            lit.unit.hash(hasher);
        }
        Expression::Parameter(param) => {
            param.index.hash(hasher);
            param.name.hash(hasher);
        }
        Expression::Prefix(prefix) => {
            std::mem::discriminant(&prefix.op_type).hash(hasher);
            hash_expression(&prefix.right, hasher);
        }
        Expression::Infix(infix) => {
            std::mem::discriminant(&infix.op_type).hash(hasher);
            hash_expression(&infix.left, hasher);
            hash_expression(&infix.right, hasher);
        }
        Expression::List(list) => {
            list.elements.len().hash(hasher);
            for val in &list.elements {
                hash_expression(val, hasher);
            }
        }
        Expression::Distinct(dist) => {
            hash_expression(&dist.expr, hasher);
        }
        Expression::Exists(exists) => {
            // Use pointer identity for hashing - avoids expensive Debug format allocation
            // The subquery AST is stable during query execution
            (exists.subquery.as_ref() as *const _ as usize).hash(hasher);
        }
        Expression::AllAny(aa) => {
            aa.operator.hash(hasher);
            std::mem::discriminant(&aa.all_any_type).hash(hasher);
            hash_expression(&aa.left, hasher);
            // Use pointer identity for hashing - avoids expensive Debug format allocation
            (aa.subquery.as_ref() as *const _ as usize).hash(hasher);
        }
        Expression::In(in_expr) => {
            in_expr.not.hash(hasher);
            hash_expression(&in_expr.left, hasher);
            hash_expression(&in_expr.right, hasher);
        }
        Expression::InHashSet(in_hash) => {
            in_hash.not.hash(hasher);
            hash_expression(&in_hash.column, hasher);
            let mut values: Vec<&Value> = in_hash.values.iter().collect();
            values.sort_unstable();
            values.hash(hasher);
        }
        Expression::Between(between) => {
            between.not.hash(hasher);
            hash_expression(&between.expr, hasher);
            hash_expression(&between.lower, hasher);
            hash_expression(&between.upper, hasher);
        }
        Expression::Like(like) => {
            like.operator.hash(hasher);
            hash_expression(&like.left, hasher);
            hash_expression(&like.pattern, hasher);
            if let Some(ref escape) = like.escape {
                true.hash(hasher);
                hash_expression(escape, hasher);
            } else {
                false.hash(hasher);
            }
        }
        Expression::ScalarSubquery(sq) => {
            // Use pointer identity for hashing - avoids expensive Debug format allocation
            (sq.subquery.as_ref() as *const _ as usize).hash(hasher);
        }
        Expression::ExpressionList(list) => {
            list.expressions.len().hash(hasher);
            for e in &list.expressions {
                hash_expression(e, hasher);
            }
        }
        Expression::Case(case) => {
            if let Some(ref val) = case.value {
                true.hash(hasher);
                hash_expression(val, hasher);
            } else {
                false.hash(hasher);
            }
            case.when_clauses.len().hash(hasher);
            for when_clause in &case.when_clauses {
                hash_expression(&when_clause.condition, hasher);
                hash_expression(&when_clause.then_result, hasher);
            }
            if let Some(ref else_val) = case.else_value {
                true.hash(hasher);
                hash_expression(else_val, hasher);
            } else {
                false.hash(hasher);
            }
        }
        Expression::Cast(cast) => {
            hash_expression(&cast.expr, hasher);
            cast.type_name.hash(hasher);
        }
        Expression::FunctionCall(func) => {
            func.function.hash(hasher);
            func.is_distinct.hash(hasher);
            func.arguments.len().hash(hasher);
            for arg in &func.arguments {
                hash_expression(arg, hasher);
            }
            if let Some(ref filter) = func.filter {
                true.hash(hasher);
                hash_expression(filter, hasher);
            } else {
                false.hash(hasher);
            }
        }
        Expression::Aliased(aliased) => {
            aliased.alias.value_lower.hash(hasher);
            hash_expression(&aliased.expression, hasher);
        }
        Expression::Window(window) => {
            window.function.function.hash(hasher);
            window.function.is_distinct.hash(hasher);
            window.function.arguments.len().hash(hasher);
            for arg in &window.function.arguments {
                hash_expression(arg, hasher);
            }
            window.partition_by.len().hash(hasher);
            for e in &window.partition_by {
                hash_expression(e, hasher);
            }
            window.order_by.len().hash(hasher);
            for order in &window.order_by {
                hash_expression(&order.expression, hasher);
                order.ascending.hash(hasher);
                order.nulls_first.hash(hasher);
            }
        }
        Expression::TableSource(ts) => {
            ts.name.value_lower.hash(hasher);
            if let Some(ref alias) = ts.alias {
                true.hash(hasher);
                alias.value_lower.hash(hasher);
            } else {
                false.hash(hasher);
            }
        }
        Expression::JoinSource(js) => {
            // Use pointer identity for hashing - avoids expensive Debug format allocation
            (js.as_ref() as *const _ as usize).hash(hasher);
        }
        Expression::SubquerySource(sq) => {
            if let Some(ref alias) = sq.alias {
                true.hash(hasher);
                alias.value_lower.hash(hasher);
            } else {
                false.hash(hasher);
            }
            // Use pointer identity for hashing - avoids expensive Debug format allocation
            (sq.subquery.as_ref() as *const _ as usize).hash(hasher);
        }
        Expression::ValuesSource(vs) => {
            if let Some(ref alias) = vs.alias {
                true.hash(hasher);
                alias.value_lower.hash(hasher);
            } else {
                false.hash(hasher);
            }
            vs.rows.len().hash(hasher);
        }
        Expression::CteReference(cte) => {
            cte.name.value_lower.hash(hasher);
        }
        Expression::FunctionTableSource(fts) => {
            fts.function.value_lower.hash(hasher);
            for arg in &fts.arguments {
                hash_expression(arg, hasher);
            }
        }
        Expression::Star(_) => {
            // Just discriminant
        }
        Expression::QualifiedStar(qs) => {
            qs.qualifier.hash(hasher);
        }
        Expression::Default(_) => {
            // Just discriminant
        }
    }
}

/// Try to get a cached program, or compile and cache it.
/// Uses O(1) LRU cache with parking_lot::Mutex for efficient concurrent access.
fn compile_expression_cached(expr: &Expression, columns: &[String]) -> Result<SharedProgram> {
    let registry = global_registry();
    let registry_generation = registry.generation();
    let cache_key = compute_cache_key(expr, columns, registry_generation);

    // Try cache first (O(1) lookup and LRU update)
    {
        let mut guard = PROGRAM_CACHE.lock();
        let cache = guard.get_or_insert_with(|| {
            // SAFETY: PROGRAM_CACHE_SIZE is always > 0
            LruCache::new(NonZeroUsize::new(PROGRAM_CACHE_SIZE).unwrap())
        });
        if let Some(entry) = cache.get(&cache_key) {
            if entry.registry_generation == registry_generation
                && entry.expression == *expr
                && entry.columns == columns
            {
                return Ok(entry.program.clone());
            }
        }
    }

    // Cache miss - compile the expression (outside lock to avoid blocking)
    let ctx = CompileContext::new(columns, registry);
    let compiler = ExprCompiler::new(&ctx);
    let program: SharedProgram = compiler
        .compile(expr)
        .map(CompactArc::new)
        .map_err(|e| Error::internal(format!("Compile error: {}", e)))?;

    // Store in cache (O(1) insertion with automatic LRU eviction)
    {
        let mut guard = PROGRAM_CACHE.lock();
        let cache = guard
            .get_or_insert_with(|| LruCache::new(NonZeroUsize::new(PROGRAM_CACHE_SIZE).unwrap()));
        cache.put(
            cache_key,
            ProgramCacheEntry {
                expression: expr.clone(),
                columns: columns.to_vec(),
                registry_generation,
                program: program.clone(),
            },
        );
    }

    Ok(program)
}

// ============================================================================
// STANDALONE COMPILATION FUNCTIONS
// ============================================================================

/// Compile an expression to a program for a given column schema.
///
/// This is the recommended way to compile expressions for use in closures
/// or parallel execution. The returned `CompactArc<Program>` is `Send + Sync` and
/// can be shared across threads efficiently.
///
/// **Note:** Results are cached globally for performance. Repeated calls
/// with the same expression and columns will return the cached program.
///
/// # Arguments
/// * `expr` - The expression to compile
/// * `columns` - Column names for the schema
///
/// # Returns
/// * `CompactArc<Program>` that can be executed with `RowFilter` or `ExprVM`
pub fn compile_expression(expr: &Expression, columns: &[String]) -> Result<SharedProgram> {
    compile_expression_cached(expr, columns)
}

/// Evaluate a column-free AST expression to a concrete Value at query time.
///
/// Returns `Some(value)` if the expression is entirely self-contained (no column
/// references) and can be evaluated. Returns `None` if the expression references
/// columns, contains context-dependent functions, or evaluation fails.
///
/// This is used by pushdown rules to resolve compound constant expressions like
/// `NOW() - INTERVAL '24 hours'` into concrete Values for index/storage filtering.
///
/// Non-deterministic functions like NOW() and RANDOM() ARE allowed here — they
/// produce valid values with a blank context (they read system clock / RNG).
/// Only context-dependent functions (CURRENT_TRANSACTION_ID) are rejected because
/// they require ExecuteContext fields that are unavailable here.
///
/// Note: this is distinct from compile-time constant folding (which rejects ALL
/// non-deterministic functions to avoid caching stale values in the program LRU).
pub fn try_eval_constant_expr(expr: &Expression) -> Option<Value> {
    use std::cell::RefCell;

    // Reject expressions that require execution context (e.g. transaction_id).
    // CURRENT_TRANSACTION_ID is the only such function; it emits Op::LoadTransactionId
    // which returns NULL with a blank context.
    if contains_context_dependent_function(expr) {
        return None;
    }

    thread_local! {
        static EVAL_VM: RefCell<ExprVM> = RefCell::new(ExprVM::new());
        static EVAL_ROW: Row = Row::new();
    }

    let empty_cols: &[String] = &[];
    let ctx = CompileContext::with_global_registry(empty_cols);
    let compiler = ExprCompiler::new(&ctx);
    let program = compiler.compile(expr).ok()?;

    EVAL_ROW.with(|empty_row| {
        let exec_ctx = ExecuteContext::new(empty_row);
        EVAL_VM.with(|vm_cell| {
            let mut vm = vm_cell.borrow_mut();
            vm.execute(&program, &exec_ctx).ok()
        })
    })
}

/// Check if an expression contains functions that depend on ExecuteContext
/// (transaction state, session variables, etc.) and cannot be evaluated
/// with a blank context. Currently only CURRENT_TRANSACTION_ID.
fn contains_context_dependent_function(expr: &Expression) -> bool {
    match expr {
        Expression::FunctionCall(func) => {
            func.function.eq_ignore_ascii_case("CURRENT_TRANSACTION_ID")
                || func
                    .arguments
                    .iter()
                    .any(contains_context_dependent_function)
        }
        Expression::Infix(infix) => {
            contains_context_dependent_function(&infix.left)
                || contains_context_dependent_function(&infix.right)
        }
        Expression::Prefix(prefix) => contains_context_dependent_function(&prefix.right),
        Expression::Cast(cast) => contains_context_dependent_function(&cast.expr),
        Expression::Case(case) => {
            case.value
                .as_ref()
                .is_some_and(|v| contains_context_dependent_function(v))
                || case.when_clauses.iter().any(|w| {
                    contains_context_dependent_function(&w.condition)
                        || contains_context_dependent_function(&w.then_result)
                })
                || case
                    .else_value
                    .as_ref()
                    .is_some_and(|v| contains_context_dependent_function(v))
        }
        Expression::Between(between) => {
            contains_context_dependent_function(&between.expr)
                || contains_context_dependent_function(&between.lower)
                || contains_context_dependent_function(&between.upper)
        }
        Expression::In(in_expr) => {
            contains_context_dependent_function(&in_expr.left)
                || contains_context_dependent_function(&in_expr.right)
        }
        Expression::Like(like) => {
            contains_context_dependent_function(&like.left)
                || contains_context_dependent_function(&like.pattern)
                || like
                    .escape
                    .as_ref()
                    .is_some_and(|e| contains_context_dependent_function(e))
        }
        Expression::List(list) => list
            .elements
            .iter()
            .any(contains_context_dependent_function),
        Expression::ExpressionList(list) => list
            .expressions
            .iter()
            .any(contains_context_dependent_function),
        Expression::Aliased(aliased) => contains_context_dependent_function(&aliased.expression),
        Expression::Distinct(distinct) => contains_context_dependent_function(&distinct.expr),
        Expression::AllAny(all_any) => contains_context_dependent_function(&all_any.left),
        Expression::InHashSet(in_hash) => contains_context_dependent_function(&in_hash.column),
        _ => false,
    }
}

/// Compile an expression with full context (parameters, outer columns, etc.)
///
/// Use this when you need parameters or correlated subquery support.
pub fn compile_expression_with_context(
    expr: &Expression,
    columns: &[String],
    outer_columns: Option<&[String]>,
    function_registry: &FunctionRegistry,
) -> Result<SharedProgram> {
    let mut ctx = CompileContext::new(columns, function_registry);
    if let Some(outer_cols) = outer_columns {
        ctx = ctx.with_outer_columns(outer_cols);
    }
    let compiler = ExprCompiler::new(&ctx);
    compiler
        .compile(expr)
        .map(CompactArc::new)
        .map_err(|e| Error::internal(format!("Compile error: {}", e)))
}

// ============================================================================
// ROW FILTER - Lightweight, Send+Sync filter for closures
// ============================================================================

/// A lightweight, thread-safe row filter for closure-based filtering.
///
/// `RowFilter` pre-compiles the expression once and can be cloned cheaply
/// (it uses `CompactArc<Program>` internally). Each thread should create its own
/// `ExprVM` for execution.
///
/// # Example
/// ```ignore
/// // Create filter once
/// let filter = RowFilter::new(&where_expr, &columns)?;
///
/// // Use in closure (filter is cloned into closure)
/// let predicate = move |row: &Row| filter.matches(row);
///
/// // Or use with parallel iteration
/// rows.par_iter().filter(|row| filter.matches(row)).collect()
/// ```
#[derive(Clone)]
pub struct RowFilter {
    /// Pre-compiled program (shared across clones)
    program: SharedProgram,
    /// Query parameters (shared) - uses `CompactArc<Vec<Value>>` to match `ExecutionContext`
    params: CompactArc<ParamVec>,
    /// Named parameters (shared)
    named_params: Arc<FxHashMap<String, Value>>,
    /// Transaction ID for CURRENT_TRANSACTION_ID()
    transaction_id: Option<u64>,
    stored_function_invoker: Option<Arc<dyn StoredFunctionInvoker>>,
    /// Bound outer row used by correlated predicates.
    outer_row: Option<Arc<FxHashMap<CompactArc<str>, Value>>>,
}

impl RowFilter {
    /// Create a new row filter by compiling the given expression.
    ///
    /// # Arguments
    /// * `expr` - The boolean expression to use as filter
    /// * `columns` - Column names matching the row schema
    pub fn new(expr: &Expression, columns: &[String]) -> Result<Self> {
        let program = compile_expression(expr, columns)?;
        Ok(Self {
            program,
            params: CompactArc::new(ParamVec::new()),
            named_params: Arc::new(FxHashMap::default()),
            transaction_id: None,
            stored_function_invoker: None,
            outer_row: None,
        })
    }

    /// Create a row filter with expression aliases for HAVING clause evaluation.
    ///
    /// Expression aliases map expression strings (like "SUM(amount)") to column
    /// indices in the result row. This is used for HAVING clauses where aggregate
    /// expressions need to reference pre-computed aggregate results.
    ///
    /// # Arguments
    /// * `expr` - The boolean expression to use as filter
    /// * `columns` - Column names matching the row schema
    /// * `aliases` - Slice of (expression_name, column_index) pairs
    ///
    /// # Example
    /// ```ignore
    /// // For HAVING SUM(amount) > 100, where SUM(amount) is at column 2
    /// let aliases = vec![("sum(amount)".to_string(), 2)];
    /// let filter = RowFilter::with_aliases(&having_expr, &columns, &aliases)?;
    ///
    /// // Filter rows
    /// for row in rows {
    ///     if filter.matches(&row) {
    ///         // row passes HAVING clause
    ///     }
    /// }
    /// ```
    pub fn with_aliases(
        expr: &Expression,
        columns: &[String],
        aliases: &[(String, usize)],
    ) -> Result<Self> {
        let alias_map = checked_alias_map(aliases)?;

        let ctx = CompileContext::with_global_registry(columns).with_expression_aliases(alias_map);
        let compiler = ExprCompiler::new(&ctx);
        let program = compiler
            .compile(expr)
            .map(CompactArc::new)
            .map_err(|e| Error::internal(format!("Compile error: {}", e)))?;

        Ok(Self {
            program,
            params: CompactArc::new(ParamVec::new()),
            named_params: Arc::new(FxHashMap::default()),
            transaction_id: None,
            stored_function_invoker: None,
            outer_row: None,
        })
    }

    /// Compile a HAVING/filter expression with aliases and lexical outer scope
    /// bound at the same time. Compiling first and attaching only parameters
    /// later cannot emit `LoadOuterColumn` instructions.
    pub fn with_aliases_and_context(
        expr: &Expression,
        columns: &[String],
        aliases: &[(String, usize)],
        execution: &ExecutionContext,
    ) -> Result<Self> {
        let alias_map = checked_alias_map(aliases)?;
        let mut context =
            CompileContext::with_global_registry(columns).with_expression_aliases(alias_map);
        if let Some(outer) = execution.outer_row() {
            let outer_columns: Vec<String> = outer.keys().map(ToString::to_string).collect();
            context = context.with_outer_columns(&outer_columns);
        }
        let compiler = ExprCompiler::new(&context);
        let program = compiler
            .compile(expr)
            .map(CompactArc::new)
            .map_err(|error| Error::internal(format!("Compile error: {error}")))?;

        Ok(Self {
            program,
            params: CompactArc::clone(execution.params_arc()),
            named_params: Arc::clone(execution.named_params_arc()),
            transaction_id: execution.transaction_id(),
            stored_function_invoker: execution.stored_function_invoker().cloned(),
            outer_row: execution.outer_row().cloned().map(Arc::new),
        })
    }

    /// Create a filter with query parameters.
    pub fn with_params(mut self, params: ParamVec) -> Self {
        self.params = CompactArc::new(params);
        self
    }

    /// Create a filter with named parameters.
    pub fn with_named_params(mut self, named_params: FxHashMap<String, Value>) -> Self {
        self.named_params = Arc::new(named_params);
        self
    }

    /// Create a filter from execution context.
    ///
    /// PERF: Both `params` and `named_params` share the Arc - zero cloning.
    pub fn with_context(mut self, ctx: &ExecutionContext) -> Self {
        // Share params Arc - no cloning needed (both use CompactArc<Vec<Value>>)
        self.params = CompactArc::clone(ctx.params_arc());
        // Share named_params Arc - no cloning needed
        self.named_params = Arc::clone(ctx.named_params_arc());
        self.transaction_id = ctx.transaction_id();
        self.stored_function_invoker = ctx.stored_function_invoker().cloned();
        self.outer_row = ctx.outer_row().cloned().map(Arc::new);
        self
    }

    /// Create a filter from a pre-compiled program.
    pub fn from_program(program: SharedProgram) -> Self {
        Self {
            program,
            params: CompactArc::new(ParamVec::new()),
            named_params: Arc::new(FxHashMap::default()),
            transaction_id: None,
            stored_function_invoker: None,
            outer_row: None,
        }
    }

    /// Check if a row matches the filter condition.
    ///
    /// This method is thread-safe and can be called from multiple threads.
    /// Each call uses a thread-local VM for execution.
    #[inline]
    pub fn matches(&self, row: &Row) -> Result<bool> {
        // Use thread-local VM for zero allocation in hot path
        thread_local! {
            static VM: std::cell::RefCell<ExprVM> = std::cell::RefCell::new(ExprVM::new());
        }

        VM.with(|vm| {
            let mut ctx = ExecuteContext::new(row);

            if !self.params.is_empty() {
                ctx = ctx.with_params(&self.params);
            }
            if !self.named_params.is_empty() {
                ctx = ctx.with_named_params(&self.named_params);
            }
            if let Some(outer_row) = self.outer_row.as_deref() {
                ctx = ctx.with_outer_row(outer_row);
            }
            ctx = ctx
                .with_transaction_id(self.transaction_id)
                .with_stored_function_invoker(self.stored_function_invoker.as_ref());

            // Use try_borrow_mut to avoid panic on recursive calls (e.g., nested subqueries).
            // If the VM is already borrowed, create a temporary one for this call.
            if let Ok(mut borrowed_vm) = vm.try_borrow_mut() {
                borrowed_vm.execute_bool(&self.program, &ctx)
            } else {
                // Fallback: create a fresh VM for recursive calls
                let mut temp_vm = ExprVM::new();
                temp_vm.execute_bool(&self.program, &ctx)
            }
        })
    }

    /// Explicit checked alias retained for callers that spell out fallibility.
    /// `matches()` and this method both propagate runtime and result-type errors.
    #[inline]
    pub fn matches_checked(&self, row: &Row) -> Result<bool> {
        thread_local! {
            static VM: std::cell::RefCell<ExprVM> = std::cell::RefCell::new(ExprVM::new());
        }

        VM.with(|vm| {
            let mut ctx = ExecuteContext::new(row);

            if !self.params.is_empty() {
                ctx = ctx.with_params(&self.params);
            }
            if !self.named_params.is_empty() {
                ctx = ctx.with_named_params(&self.named_params);
            }
            if let Some(outer_row) = self.outer_row.as_deref() {
                ctx = ctx.with_outer_row(outer_row);
            }
            ctx = ctx
                .with_transaction_id(self.transaction_id)
                .with_stored_function_invoker(self.stored_function_invoker.as_ref());

            if let Ok(mut borrowed_vm) = vm.try_borrow_mut() {
                borrowed_vm.execute_bool_checked(&self.program, &ctx)
            } else {
                let mut temp_vm = ExprVM::new();
                temp_vm.execute_bool_checked(&self.program, &ctx)
            }
        })
    }

    /// Evaluate a predicate over a compact JOIN row without materializing it.
    #[inline]
    pub fn matches_deferred_checked(&self, row: &radixdb_storage::DeferredRow) -> Result<bool> {
        thread_local! {
            static VM: std::cell::RefCell<ExprVM> = std::cell::RefCell::new(ExprVM::new());
        }

        VM.with(|vm| {
            let mut ctx = ExecuteContext::for_deferred(row);

            if !self.params.is_empty() {
                ctx = ctx.with_params(&self.params);
            }
            if !self.named_params.is_empty() {
                ctx = ctx.with_named_params(&self.named_params);
            }
            if let Some(outer_row) = self.outer_row.as_deref() {
                ctx = ctx.with_outer_row(outer_row);
            }
            ctx = ctx
                .with_transaction_id(self.transaction_id)
                .with_stored_function_invoker(self.stored_function_invoker.as_ref());

            if let Ok(mut borrowed_vm) = vm.try_borrow_mut() {
                borrowed_vm.execute_bool_checked(&self.program, &ctx)
            } else {
                let mut temp_vm = ExprVM::new();
                temp_vm.execute_bool_checked(&self.program, &ctx)
            }
        })
    }

    /// Evaluate a predicate over a virtual executor row without materializing it.
    #[inline]
    pub fn matches_row_ref_checked(&self, row: &crate::operator::RowRef) -> Result<bool> {
        thread_local! {
            static VM: std::cell::RefCell<ExprVM> = std::cell::RefCell::new(ExprVM::new());
        }

        VM.with(|vm| {
            let mut ctx = ExecuteContext::for_row_ref(row);

            if !self.params.is_empty() {
                ctx = ctx.with_params(&self.params);
            }
            if !self.named_params.is_empty() {
                ctx = ctx.with_named_params(&self.named_params);
            }
            if let Some(outer_row) = self.outer_row.as_deref() {
                ctx = ctx.with_outer_row(outer_row);
            }
            ctx = ctx
                .with_transaction_id(self.transaction_id)
                .with_stored_function_invoker(self.stored_function_invoker.as_ref());

            if let Ok(mut borrowed_vm) = vm.try_borrow_mut() {
                borrowed_vm.execute_bool_checked(&self.program, &ctx)
            } else {
                ExprVM::new().execute_bool_checked(&self.program, &ctx)
            }
        })
    }

    /// Filter a RowVec in-place, removing rows that don't match.
    /// Returns Err if the filter expression produces a runtime error
    /// (e.g. invalid REGEXP pattern supplied via a parameter).
    pub fn retain_checked(&self, rows: &mut radixdb_core::RowVec) -> Result<()> {
        let mut error: Option<radixdb_core::Error> = None;
        rows.retain(|(_, row)| {
            if error.is_some() {
                return false;
            }
            match self.matches_checked(row) {
                Ok(b) => b,
                Err(e) => {
                    error = Some(e);
                    false
                }
            }
        });
        match error {
            Some(e) => Err(e),
            None => Ok(()),
        }
    }

    /// Evaluate the filter expression and return the value.
    #[inline]
    pub fn evaluate(&self, row: &Row) -> Result<Value> {
        thread_local! {
            static VM: std::cell::RefCell<ExprVM> = std::cell::RefCell::new(ExprVM::new());
        }

        VM.with(|vm| {
            let mut ctx = ExecuteContext::new(row);

            if !self.params.is_empty() {
                ctx = ctx.with_params(&self.params);
            }
            if !self.named_params.is_empty() {
                ctx = ctx.with_named_params(&self.named_params);
            }
            ctx = ctx
                .with_transaction_id(self.transaction_id)
                .with_stored_function_invoker(self.stored_function_invoker.as_ref());

            // Use try_borrow_mut to avoid panic on recursive calls (e.g., nested subqueries).
            // If the VM is already borrowed, create a temporary one for this call.
            if let Ok(mut borrowed_vm) = vm.try_borrow_mut() {
                borrowed_vm.execute_cow(&self.program, &ctx)
            } else {
                // Fallback: create a fresh VM for recursive calls
                let mut temp_vm = ExprVM::new();
                temp_vm.execute_cow(&self.program, &ctx)
            }
        })
    }

    /// Get the underlying program (for advanced use cases).
    pub fn program(&self) -> &SharedProgram {
        &self.program
    }
}

// Static assertions to verify RowFilter implements Send + Sync.
// This is safer than unsafe impl because it will fail at compile time
// if any field doesn't implement Send/Sync, rather than causing UB at runtime.
// All fields are Send + Sync:
// - CompactArc<Program> is Send + Sync (Program is immutable)
// - CompactArc<Value> is Send + Sync
// - Arc<FxHashMap<String, Value>> is Send + Sync
const _: () = {
    const fn assert_send_sync<T: Send + Sync>() {}
    let _ = assert_send_sync::<RowFilter>;
};

// ============================================================================
// JOIN FILTER - For join condition evaluation
// ============================================================================

/// A filter for join condition evaluation between two rows.
#[derive(Clone)]
pub struct JoinFilter {
    /// Pre-compiled program
    program: SharedProgram,
    /// Query parameters (shared Arc to avoid cloning)
    params: CompactArc<ParamVec>,
    /// Named parameters (shared Arc to avoid cloning)
    named_params: Arc<FxHashMap<String, Value>>,
    /// Transaction ID for context-dependent functions in residual predicates.
    transaction_id: Option<u64>,
    stored_function_invoker: Option<Arc<dyn StoredFunctionInvoker>>,
}

impl JoinFilter {
    /// Create a join filter by compiling the condition.
    ///
    /// # Arguments
    /// * `expr` - The join condition expression
    /// * `left_columns` - Column names for the left table
    /// * `right_columns` - Column names for the right table
    pub fn new(
        expr: &Expression,
        left_columns: &[String],
        right_columns: &[String],
        function_registry: &FunctionRegistry,
    ) -> Result<Self> {
        let ctx =
            CompileContext::new(left_columns, function_registry).with_second_row(right_columns);
        let compiler = ExprCompiler::new(&ctx);
        let program = compiler
            .compile(expr)
            .map_err(|e| Error::internal(format!("Compile error: {}", e)))?;
        Ok(Self {
            program: CompactArc::new(program),
            params: CompactArc::new(ParamVec::new()),
            named_params: Arc::new(FxHashMap::default()),
            transaction_id: None,
            stored_function_invoker: None,
        })
    }

    /// Set parameters from execution context.
    /// This is required when the join condition contains parameter placeholders ($1, $2, etc.).
    #[inline]
    pub fn with_context(mut self, ctx: &ExecutionContext) -> Self {
        self.params = CompactArc::clone(ctx.params_arc());
        self.named_params = Arc::clone(ctx.named_params_arc());
        self.transaction_id = ctx.transaction_id();
        self.stored_function_invoker = ctx.stored_function_invoker().cloned();
        self
    }

    /// Check if a pair of rows satisfies the join condition.
    #[inline]
    pub fn matches(&self, left_row: &Row, right_row: &Row) -> Result<bool> {
        thread_local! {
            static VM: std::cell::RefCell<ExprVM> = std::cell::RefCell::new(ExprVM::new());
        }

        VM.with(|vm| {
            let mut ctx = ExecuteContext::for_join(left_row, right_row)
                .with_transaction_id(self.transaction_id)
                .with_stored_function_invoker(self.stored_function_invoker.as_ref());

            // Apply params if present (required for parameter placeholders like $1, $2)
            if !self.params.is_empty() {
                ctx = ctx.with_params(&self.params);
            }
            if !self.named_params.is_empty() {
                ctx = ctx.with_named_params(&self.named_params);
            }

            // Use try_borrow_mut to avoid panic on recursive calls (e.g., nested subqueries).
            // If the VM is already borrowed, create a temporary one for this call.
            if let Ok(mut borrowed_vm) = vm.try_borrow_mut() {
                borrowed_vm.execute_bool(&self.program, &ctx)
            } else {
                // Fallback: create a fresh VM for recursive calls
                let mut temp_vm = ExprVM::new();
                temp_vm.execute_bool(&self.program, &ctx)
            }
        })
    }

    /// Checked join predicate evaluation. Runtime expression errors must be
    /// propagated by physical operators rather than converted into non-matches.
    #[inline]
    pub fn matches_checked(&self, left_row: &Row, right_row: &Row) -> Result<bool> {
        thread_local! {
            static VM: std::cell::RefCell<ExprVM> = std::cell::RefCell::new(ExprVM::new());
        }

        VM.with(|vm| {
            let mut ctx = ExecuteContext::for_join(left_row, right_row)
                .with_transaction_id(self.transaction_id)
                .with_stored_function_invoker(self.stored_function_invoker.as_ref());
            if !self.params.is_empty() {
                ctx = ctx.with_params(&self.params);
            }
            if !self.named_params.is_empty() {
                ctx = ctx.with_named_params(&self.named_params);
            }

            if let Ok(mut borrowed_vm) = vm.try_borrow_mut() {
                borrowed_vm.execute_bool_checked(&self.program, &ctx)
            } else {
                ExprVM::new().execute_bool_checked(&self.program, &ctx)
            }
        })
    }

    /// Checked predicate evaluation over a deferred JOIN outer row.
    ///
    /// This preserves the same compiled program and SQL semantics as
    /// `matches_checked`, but lets a following JOIN edge read projected slots
    /// without forcing the preceding edge to materialize an owned `Row`.
    #[inline]
    pub fn matches_row_ref_checked(
        &self,
        left_row: &crate::operator::RowRef,
        right_row: &Row,
    ) -> Result<bool> {
        thread_local! {
            static VM: std::cell::RefCell<ExprVM> = std::cell::RefCell::new(ExprVM::new());
        }

        VM.with(|vm| {
            let mut ctx = ExecuteContext::for_join_ref(left_row, right_row)
                .with_transaction_id(self.transaction_id)
                .with_stored_function_invoker(self.stored_function_invoker.as_ref());
            if !self.params.is_empty() {
                ctx = ctx.with_params(&self.params);
            }
            if !self.named_params.is_empty() {
                ctx = ctx.with_named_params(&self.named_params);
            }

            if let Ok(mut borrowed_vm) = vm.try_borrow_mut() {
                borrowed_vm.execute_bool_checked(&self.program, &ctx)
            } else {
                ExprVM::new().execute_bool_checked(&self.program, &ctx)
            }
        })
    }

    /// Checked predicate evaluation over two virtual JOIN inputs.
    ///
    /// This is the hash-join residual path: equality candidates remain row
    /// references until the complete `ON` predicate accepts the pair.
    #[inline]
    pub fn matches_row_refs_checked(
        &self,
        left_row: &crate::operator::RowRef,
        right_row: &crate::operator::RowRef,
    ) -> Result<bool> {
        thread_local! {
            static VM: std::cell::RefCell<ExprVM> = std::cell::RefCell::new(ExprVM::new());
        }

        VM.with(|vm| {
            let mut ctx = ExecuteContext::for_join_refs(left_row, right_row)
                .with_transaction_id(self.transaction_id)
                .with_stored_function_invoker(self.stored_function_invoker.as_ref());
            if !self.params.is_empty() {
                ctx = ctx.with_params(&self.params);
            }
            if !self.named_params.is_empty() {
                ctx = ctx.with_named_params(&self.named_params);
            }

            if let Ok(mut borrowed_vm) = vm.try_borrow_mut() {
                borrowed_vm.execute_bool_checked(&self.program, &ctx)
            } else {
                ExprVM::new().execute_bool_checked(&self.program, &ctx)
            }
        })
    }

    /// Get the underlying program.
    pub fn program(&self) -> &SharedProgram {
        &self.program
    }
}

// Static assertions to verify JoinFilter implements Send + Sync.
// This is safer than unsafe impl because it will fail at compile time
// if any field doesn't implement Send/Sync, rather than causing UB at runtime.
const _: () = {
    const fn assert_send_sync<T: Send + Sync>() {}
    let _ = assert_send_sync::<JoinFilter>;
};

// ============================================================================
// EXPRESSION EVAL - Direct VM usage for maximum performance
// ============================================================================

/// Lightweight expression evaluator using direct VM execution.
///
/// `ExpressionEval` provides the simplest possible API for expression evaluation:
/// 1. Compile the expression once with `new()`
/// 2. Evaluate rows with `eval()` or `eval_bool()`
///
/// This is the recommended replacement for `CompiledEvaluator` when you have
/// a single expression to evaluate repeatedly.
///
/// # Example
/// ```ignore
/// // Compile once
/// let eval = ExpressionEval::compile(&expr, &columns)?;
///
/// // Evaluate many rows
/// for row in rows {
///     let value = eval.eval(&row)?;
///     // or for boolean: let matches = eval.eval_bool(&row);
/// }
/// ```
pub struct ExpressionEval {
    /// Pre-compiled program
    program: SharedProgram,
    /// VM instance (reusable, maintains stack)
    vm: ExprVM,
    /// Query parameters (shared) - uses `CompactArc<Vec<Value>>` to match `ExecutionContext`
    params: CompactArc<ParamVec>,
    /// Named parameters (shared) - uses Arc to match ExecutionContext
    named_params: Arc<FxHashMap<String, Value>>,
    /// Outer row context for correlated subqueries
    outer_row: Option<FxHashMap<CompactArc<str>, Value>>,
    /// Transaction ID
    transaction_id: Option<u64>,
    stored_function_invoker: Option<Arc<dyn StoredFunctionInvoker>>,
}

impl ExpressionEval {
    /// Compile an expression for evaluation.
    pub fn compile(expr: &Expression, columns: &[String]) -> Result<Self> {
        let program = compile_expression(expr, columns)?;
        Ok(Self {
            program,
            vm: ExprVM::new(),
            params: CompactArc::new(ParamVec::new()),
            named_params: Arc::new(FxHashMap::default()),
            outer_row: None,
            transaction_id: None,
            stored_function_invoker: None,
        })
    }

    /// Compile with expression aliases for HAVING clause evaluation.
    ///
    /// Expression aliases map expression strings (like "SUM(amount)") to column
    /// indices in the result row. This is used for HAVING clauses where aggregate
    /// expressions need to reference pre-computed aggregate results.
    ///
    /// # Arguments
    /// * `expr` - The expression to compile
    /// * `columns` - Column names for the result row
    /// * `aliases` - Slice of (expression_name, column_index) pairs
    ///
    /// # Example
    /// ```ignore
    /// // For HAVING SUM(amount) > 100, where SUM(amount) is at column 2
    /// let aliases = vec![("sum(amount)".to_string(), 2)];
    /// let eval = ExpressionEval::compile_with_aliases(&having_expr, &columns, &aliases)?;
    /// ```
    pub fn compile_with_aliases(
        expr: &Expression,
        columns: &[String],
        aliases: &[(String, usize)],
    ) -> Result<Self> {
        let alias_map = checked_alias_map(aliases)?;

        Self::compile_with_options(
            expr,
            columns,
            None,
            None,
            Some(alias_map),
            global_registry(),
        )
    }

    /// Compile with full context options.
    pub fn compile_with_options(
        expr: &Expression,
        columns: &[String],
        columns2: Option<&[String]>,
        outer_columns: Option<&[String]>,
        expression_aliases: Option<StringMap<u16>>,
        function_registry: &FunctionRegistry,
    ) -> Result<Self> {
        let mut ctx = CompileContext::new(columns, function_registry);
        if let Some(cols2) = columns2 {
            ctx = ctx.with_second_row(cols2);
        }
        if let Some(outer) = outer_columns {
            ctx = ctx.with_outer_columns(outer);
        }
        if let Some(aliases) = expression_aliases {
            ctx = ctx.with_expression_aliases(aliases);
        }
        let compiler = ExprCompiler::new(&ctx);
        let program = compiler
            .compile(expr)
            .map(CompactArc::new)
            .map_err(|e| Error::internal(format!("Compile error: {}", e)))?;
        Ok(Self {
            program,
            vm: ExprVM::new(),
            params: CompactArc::new(ParamVec::new()),
            named_params: Arc::new(FxHashMap::default()),
            outer_row: None,
            transaction_id: None,
            stored_function_invoker: None,
        })
    }

    /// Create from a pre-compiled program.
    pub fn from_program(program: SharedProgram) -> Self {
        Self {
            program,
            vm: ExprVM::new(),
            params: CompactArc::new(ParamVec::new()),
            named_params: Arc::new(FxHashMap::default()),
            outer_row: None,
            transaction_id: None,
            stored_function_invoker: None,
        }
    }

    /// Set query parameters.
    pub fn with_params(mut self, params: ParamVec) -> Self {
        self.params = CompactArc::new(params);
        self
    }

    /// Set named parameters.
    pub fn with_named_params(mut self, named_params: FxHashMap<String, Value>) -> Self {
        self.named_params = Arc::new(named_params);
        self
    }

    /// Set context from ExecutionContext.
    ///
    /// PERF: Both `params` and `named_params` share the Arc - zero cloning.
    pub fn with_context(mut self, ctx: &ExecutionContext) -> Self {
        // Share params Arc - no cloning needed
        self.params = CompactArc::clone(ctx.params_arc());
        // Share named_params Arc - no cloning needed
        self.named_params = Arc::clone(ctx.named_params_arc());
        if let Some(outer) = ctx.outer_row() {
            // Clone the map directly (CompactArc<str> clones are cheap)
            let arc_map = outer.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
            self.outer_row = Some(arc_map);
        }
        self.transaction_id = ctx.transaction_id();
        self.stored_function_invoker = ctx.stored_function_invoker().cloned();
        self
    }

    /// Set transaction ID.
    pub fn with_transaction_id(mut self, txn_id: Option<u64>) -> Self {
        self.transaction_id = txn_id;
        self
    }

    /// Set outer row for correlated subqueries.
    /// Accepts `CompactArc<str>` keys directly to avoid conversion overhead.
    pub fn set_outer_row(&mut self, outer: &FxHashMap<CompactArc<str>, Value>) {
        // Clone the map (CompactArc clones are cheap, Value clones may be expensive but needed)
        let map = outer.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
        self.outer_row = Some(map);
    }

    /// Clear outer row.
    pub fn clear_outer_row(&mut self) {
        self.outer_row = None;
    }

    /// Evaluate the expression for a row.
    #[inline]
    pub fn eval(&mut self, row: &Row) -> Result<Value> {
        let mut ctx = ExecuteContext::new(row);

        if !self.params.is_empty() {
            ctx = ctx.with_params(&self.params);
        }
        if !self.named_params.is_empty() {
            ctx = ctx.with_named_params(&self.named_params);
        }
        if let Some(ref outer) = self.outer_row {
            ctx = ctx.with_outer_row(outer);
        }
        ctx = ctx
            .with_transaction_id(self.transaction_id)
            .with_stored_function_invoker(self.stored_function_invoker.as_ref());

        self.vm.execute_cow(&self.program, &ctx)
    }

    /// Evaluate as boolean (for WHERE/HAVING).
    #[inline]
    pub fn eval_bool(&mut self, row: &Row) -> Result<bool> {
        let mut ctx = ExecuteContext::new(row);

        if !self.params.is_empty() {
            ctx = ctx.with_params(&self.params);
        }
        if !self.named_params.is_empty() {
            ctx = ctx.with_named_params(&self.named_params);
        }
        if let Some(ref outer) = self.outer_row {
            ctx = ctx.with_outer_row(outer);
        }
        ctx = ctx
            .with_transaction_id(self.transaction_id)
            .with_stored_function_invoker(self.stored_function_invoker.as_ref());

        self.vm.execute_bool(&self.program, &ctx)
    }

    /// Like eval_bool but returns errors instead of swallowing them.
    #[inline]
    pub fn eval_bool_checked(&mut self, row: &Row) -> Result<bool> {
        let mut ctx = ExecuteContext::new(row);

        if !self.params.is_empty() {
            ctx = ctx.with_params(&self.params);
        }
        if !self.named_params.is_empty() {
            ctx = ctx.with_named_params(&self.named_params);
        }
        if let Some(ref outer) = self.outer_row {
            ctx = ctx.with_outer_row(outer);
        }
        ctx = ctx
            .with_transaction_id(self.transaction_id)
            .with_stored_function_invoker(self.stored_function_invoker.as_ref());

        self.vm.execute_bool_checked(&self.program, &ctx)
    }

    /// Evaluate with two rows (for joins).
    #[inline]
    pub fn eval_join(&mut self, left: &Row, right: &Row) -> Result<Value> {
        let ctx = ExecuteContext::for_join(left, right)
            .with_transaction_id(self.transaction_id)
            .with_stored_function_invoker(self.stored_function_invoker.as_ref());
        self.vm.execute_cow(&self.program, &ctx)
    }

    /// Evaluate join as boolean.
    #[inline]
    pub fn eval_join_bool(&mut self, left: &Row, right: &Row) -> Result<bool> {
        let ctx = ExecuteContext::for_join(left, right)
            .with_transaction_id(self.transaction_id)
            .with_stored_function_invoker(self.stored_function_invoker.as_ref());
        self.vm.execute_bool(&self.program, &ctx)
    }

    /// Evaluate with a row reference.
    #[inline]
    pub fn eval_slice(&mut self, row: &Row) -> Result<Value> {
        let mut ctx = ExecuteContext::new(row);

        if !self.params.is_empty() {
            ctx = ctx.with_params(&self.params);
        }
        if !self.named_params.is_empty() {
            ctx = ctx.with_named_params(&self.named_params);
        }
        if let Some(ref outer) = self.outer_row {
            ctx = ctx.with_outer_row(outer);
        }
        ctx = ctx
            .with_transaction_id(self.transaction_id)
            .with_stored_function_invoker(self.stored_function_invoker.as_ref());

        self.vm.execute_cow(&self.program, &ctx)
    }

    /// Evaluate as boolean.
    #[inline]
    pub fn eval_slice_bool(&mut self, row: &Row) -> Result<bool> {
        let mut ctx = ExecuteContext::new(row);

        if !self.params.is_empty() {
            ctx = ctx.with_params(&self.params);
        }
        if !self.named_params.is_empty() {
            ctx = ctx.with_named_params(&self.named_params);
        }
        if let Some(ref outer) = self.outer_row {
            ctx = ctx.with_outer_row(outer);
        }
        ctx = ctx
            .with_transaction_id(self.transaction_id)
            .with_stored_function_invoker(self.stored_function_invoker.as_ref());

        self.vm.execute_bool(&self.program, &ctx)
    }

    /// Get the underlying program.
    pub fn program(&self) -> &SharedProgram {
        &self.program
    }
}

// ============================================================================
// MULTI-EXPRESSION EVALUATOR - For SELECT projections
// ============================================================================

/// Evaluates multiple expressions efficiently (for SELECT projections).
///
/// Pre-compiles all expressions once, then evaluates them together for each row.
pub struct MultiExpressionEval {
    /// Pre-compiled programs for each expression
    programs: Vec<SharedProgram>,
    /// Single VM instance (reused for all expressions)
    vm: ExprVM,
    /// Query parameters (shared) - uses `CompactArc<Vec<Value>>` to match `ExecutionContext`
    params: CompactArc<ParamVec>,
    /// Named parameters (shared) - uses Arc to match ExecutionContext
    named_params: Arc<FxHashMap<String, Value>>,
    /// Transaction ID
    transaction_id: Option<u64>,
    stored_function_invoker: Option<Arc<dyn StoredFunctionInvoker>>,
}

impl MultiExpressionEval {
    /// Compile multiple expressions.
    pub fn compile(exprs: &[Expression], columns: &[String]) -> Result<Self> {
        let ctx = CompileContext::with_global_registry(columns);
        let compiler = ExprCompiler::new(&ctx);

        let programs = exprs
            .iter()
            .map(|expr| {
                compiler
                    .compile(expr)
                    .map(CompactArc::new)
                    .map_err(|e| Error::internal(format!("Compile error: {}", e)))
            })
            .collect::<Result<Vec<_>>>()?;

        Ok(Self {
            programs,
            vm: ExprVM::new(),
            params: CompactArc::new(ParamVec::new()),
            named_params: Arc::new(FxHashMap::default()),
            transaction_id: None,
            stored_function_invoker: None,
        })
    }

    /// Compile multiple expressions with expression aliases.
    ///
    /// Expression aliases map expression strings (like "SUM(amount)") to column
    /// indices in the result row. This is used for window function ORDER BY
    /// clauses where aggregate expressions need to reference pre-computed results.
    ///
    /// # Arguments
    /// * `exprs` - The expressions to compile
    /// * `columns` - Column names for the result row
    /// * `aliases` - Slice of (expression_name, column_index) pairs
    pub fn compile_with_aliases(
        exprs: &[Expression],
        columns: &[String],
        aliases: &[(String, usize)],
    ) -> Result<Self> {
        let alias_map = checked_alias_map(aliases)?;

        let ctx = CompileContext::with_global_registry(columns).with_expression_aliases(alias_map);
        let compiler = ExprCompiler::new(&ctx);

        let programs = exprs
            .iter()
            .map(|expr| {
                compiler
                    .compile(expr)
                    .map(CompactArc::new)
                    .map_err(|e| Error::internal(format!("Compile error: {}", e)))
            })
            .collect::<Result<Vec<_>>>()?;

        Ok(Self {
            programs,
            vm: ExprVM::new(),
            params: CompactArc::new(ParamVec::new()),
            named_params: Arc::new(FxHashMap::default()),
            transaction_id: None,
            stored_function_invoker: None,
        })
    }

    /// Set query parameters.
    pub fn with_params(mut self, params: ParamVec) -> Self {
        self.params = CompactArc::new(params);
        self
    }

    /// Set from execution context.
    ///
    /// PERF: Both `params` and `named_params` share the Arc - zero cloning.
    pub fn with_context(mut self, ctx: &ExecutionContext) -> Self {
        // Share params Arc - no cloning needed
        self.params = CompactArc::clone(ctx.params_arc());
        // Share named_params Arc - no cloning needed
        self.named_params = Arc::clone(ctx.named_params_arc());
        self.transaction_id = ctx.transaction_id();
        self.stored_function_invoker = ctx.stored_function_invoker().cloned();
        self
    }

    /// Evaluate all expressions for a row, returning values in order.
    #[inline]
    pub fn eval_all(&mut self, row: &Row) -> Result<Vec<Value>> {
        let mut ctx = ExecuteContext::new(row);

        if !self.params.is_empty() {
            ctx = ctx.with_params(&self.params);
        }
        if !self.named_params.is_empty() {
            ctx = ctx.with_named_params(&self.named_params);
        }
        ctx = ctx
            .with_transaction_id(self.transaction_id)
            .with_stored_function_invoker(self.stored_function_invoker.as_ref());

        self.programs
            .iter()
            .map(|prog| self.vm.execute_cow(prog, &ctx))
            .collect()
    }

    /// Evaluate all expressions, writing results into provided buffer.
    #[inline]
    pub fn eval_into(&mut self, row: &Row, output: &mut Vec<Value>) -> Result<()> {
        let mut ctx = ExecuteContext::new(row);

        if !self.params.is_empty() {
            ctx = ctx.with_params(&self.params);
        }
        if !self.named_params.is_empty() {
            ctx = ctx.with_named_params(&self.named_params);
        }
        ctx = ctx
            .with_transaction_id(self.transaction_id)
            .with_stored_function_invoker(self.stored_function_invoker.as_ref());

        output.clear();
        for prog in &self.programs {
            output.push(self.vm.execute_cow(prog, &ctx)?);
        }
        Ok(())
    }

    /// Number of expressions.
    pub fn len(&self) -> usize {
        self.programs.len()
    }

    /// Check if empty.
    pub fn is_empty(&self) -> bool {
        self.programs.is_empty()
    }
}

/// Shared program reference for zero-copy caching
pub type SharedProgram = CompactArc<Program>;

// ============================================================================
// COMPILED EVALUATOR - DEPRECATED, use ExpressionEval instead
// ============================================================================

/// Compiled expression evaluator using the Expression VM.
///
/// # Deprecated
///
/// **This type is deprecated.** Use the new, more efficient alternatives:
///
/// - [`ExpressionEval`] - For single expression evaluation (most common case)
/// - [`RowFilter`] - For closure-based filtering (Send+Sync safe)
/// - [`JoinFilter`] - For join condition evaluation
/// - [`MultiExpressionEval`] - For SELECT projections (multiple expressions)
///
/// The new APIs pre-compile expressions eagerly rather than lazily, avoiding
/// cache invalidation issues and providing better performance.
///
/// ## Migration Guide
///
/// **Before (CompiledEvaluator):**
/// ```ignore
/// let mut eval = CompiledEvaluator::new(&registry);
/// eval.init_columns(&columns);
/// for row in rows {
///     eval.set_row_array(&row);
///     let value = eval.evaluate(&expr)?;
/// }
/// ```
///
/// **After (ExpressionEval):**
/// ```ignore
/// let mut eval = ExpressionEval::compile(&expr, &columns)?;
/// for row in rows {
///     let value = eval.eval(&row)?;
/// }
/// ```
///
/// # When to use CompiledEvaluator vs new APIs
///
/// **Use the new APIs (recommended for most cases):**
/// - [`ExpressionEval`] - Single expression with static schema
/// - [`RowFilter`] - WHERE clause filtering (thread-safe)
/// - [`MultiExpressionEval`] - SELECT projections (multiple expressions)
///
/// **Use CompiledEvaluator when:**
/// - Expressions change per-row (e.g., after `process_correlated_expression`)
/// - You need dynamic/lazy expression compilation
/// - Complex scenarios with correlated subqueries
pub struct CompiledEvaluator<'a> {
    /// Function registry for compilation
    function_registry: &'a FunctionRegistry,

    /// Column names for compilation context (Arc for zero-copy sharing)
    columns: CompactArc<Vec<String>>,

    /// Second row columns (for joins)
    columns2: Option<Vec<String>>,

    /// Outer query columns (for correlated subqueries)
    outer_columns: Option<Vec<String>>,

    /// Query parameters (positional) - uses `CompactArc<Vec<Value>>` to match `ExecutionContext`
    params: CompactArc<ParamVec>,

    /// Query parameters (named) - uses Arc to match ExecutionContext
    named_params: Arc<FxHashMap<String, Value>>,

    /// Outer row context for correlated subqueries
    outer_row: Option<FxHashMap<CompactArc<str>, Value>>,

    /// Current transaction ID
    transaction_id: Option<u64>,
    stored_function_invoker: Option<Arc<dyn StoredFunctionInvoker>>,

    /// Expression aliases for HAVING clause
    expression_aliases: StringMap<u16>,

    /// Column aliases
    column_aliases: StringMap<String>,

    /// VM instance (reusable)
    vm: ExprVM,

    /// Local cache with collision-checked expression identity.
    local_cache: FxHashMap<u64, LocalProgramCacheEntry>,

    /// Deferred schema/alias validation failures for infallible setters.
    context_errors: Vec<String>,

    /// Current row values for execution (owned copy for safety)
    current_row: Option<Row>,

    /// Second row for joins (owned copy for safety)
    current_row2: Option<Row>,
}

// CompiledEvaluator is Send + Sync because all fields are Send + Sync:
// - function_registry: &FunctionRegistry is Send + Sync (shared reference to thread-safe registry)
// - All other fields are owned types that are Send + Sync

impl<'a> CompiledEvaluator<'a> {
    /// Create a new compiled evaluator with a function registry reference
    pub fn new(function_registry: &'a FunctionRegistry) -> Self {
        Self {
            function_registry,
            columns: CompactArc::new(Vec::new()),
            columns2: None,
            outer_columns: None,
            params: CompactArc::new(ParamVec::new()),
            named_params: Arc::new(FxHashMap::default()),
            outer_row: None,
            transaction_id: None,
            stored_function_invoker: None,
            expression_aliases: StringMap::new(),
            column_aliases: StringMap::new(),
            vm: ExprVM::new(),
            local_cache: FxHashMap::default(),
            context_errors: Vec::new(),
            current_row: None,
            current_row2: None,
        }
    }

    /// Create an evaluator using the global function registry.
    pub fn with_defaults() -> CompiledEvaluator<'static> {
        CompiledEvaluator::new(global_registry())
    }

    fn column_limit_error(label: &str, len: usize) -> Option<String> {
        (len > (u16::MAX as usize + 1)).then(|| {
            format!(
                "{label} has {len} columns; expression bytecode supports at most {}",
                u16::MAX as usize + 1
            )
        })
    }

    /// Clear all state for reuse.
    pub fn clear(&mut self) {
        self.columns = CompactArc::new(Vec::new());
        self.columns2 = None;
        self.outer_columns = None;
        self.params = CompactArc::new(ParamVec::new());
        self.named_params = Arc::new(FxHashMap::default());
        self.outer_row = None;
        self.transaction_id = None;
        self.stored_function_invoker = None;
        self.expression_aliases.clear();
        self.column_aliases.clear();
        self.local_cache.clear();
        self.context_errors.clear();
        self.current_row = None;
        self.current_row2 = None;
    }

    fn replace_context_error(&mut self, label: &str, error: Option<String>) {
        self.context_errors
            .retain(|existing| !existing.starts_with(label));
        if let Some(error) = error {
            self.context_errors.push(error);
        }
    }

    fn context_error(&self) -> Option<String> {
        (!self.context_errors.is_empty()).then(|| self.context_errors.join("; "))
    }

    /// Set the current transaction ID
    pub fn set_transaction_id(&mut self, txn_id: u64) {
        self.transaction_id = Some(txn_id);
    }

    /// Set query parameters (positional) - fluent API
    pub fn with_params(mut self, params: ParamVec) -> Self {
        self.params = CompactArc::new(params);
        self
    }

    /// Set named query parameters - fluent API
    pub fn with_named_params(mut self, named_params: FxHashMap<String, Value>) -> Self {
        self.named_params = Arc::new(named_params);
        self
    }

    /// Set parameters from execution context - fluent API
    ///
    /// PERF: Both `params` and `named_params` share the Arc - zero cloning.
    pub fn with_context(mut self, ctx: &ExecutionContext) -> Self {
        // Share params Arc - no cloning needed
        self.params = CompactArc::clone(ctx.params_arc());
        // Share named_params Arc - no cloning needed
        self.named_params = Arc::clone(ctx.named_params_arc());

        // Set outer row context for correlated subqueries
        if let Some(outer) = ctx.outer_row() {
            // Clone the map directly (CompactArc<str> clones are cheap)
            let arc_map = outer.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
            // Convert CompactArc<str> keys to String for outer_columns (needed for compilation)
            let outer_cols: Vec<String> = outer.keys().map(|k| k.to_string()).collect();
            self.outer_row = Some(arc_map);
            // Also set up outer_columns for compilation
            if !outer_cols.is_empty() {
                self.outer_columns = Some(outer_cols);
                let error = Self::column_limit_error(
                    "outer row",
                    self.outer_columns.as_ref().map_or(0, Vec::len),
                );
                self.replace_context_error("outer row", error);
                // Invalidate local cache since compilation context changed
                self.local_cache.clear();
            }
        }

        self.transaction_id = ctx.transaction_id();
        self.stored_function_invoker = ctx.stored_function_invoker().cloned();
        self
    }

    /// Bind an owned row and its column names for fluent evaluation.
    pub fn with_row(mut self, row: Row, columns: &[String]) -> Self {
        self.init_columns(columns);
        self.current_row = Some(row);
        self.current_row2 = None;
        self
    }

    /// Initialize the column index mapping (call once before set_row_array)
    ///
    /// Repeated semantic schemas avoid cloning; pointer identity is never used
    /// because allocator reuse is not a schema identity.
    pub fn init_columns(&mut self, columns: &[String]) {
        if self.columns.as_ref() == columns {
            return;
        }

        self.columns = CompactArc::new(columns.to_vec());
        let error = Self::column_limit_error("primary row", columns.len());
        self.replace_context_error("primary row", error);
        // Clear local cache since compilation context changed
        self.local_cache.clear();
    }

    /// Initialize columns from an Arc (zero-copy when schema already has Arc)
    ///
    /// This is the preferred method when the caller already has a `CompactArc<Vec<String>>`,
    /// such as from `Schema::column_names_arc()`. It avoids all string cloning.
    #[inline]
    pub fn init_columns_arc(&mut self, columns: CompactArc<Vec<String>>) {
        if self.columns.as_ref() == columns.as_ref() {
            return;
        }

        let error = Self::column_limit_error("primary row", columns.len());
        self.replace_context_error("primary row", error);
        self.columns = columns;
        // Clear local cache since compilation context changed
        self.local_cache.clear();
    }

    /// Add aggregate expression aliases for HAVING clause evaluation
    pub fn add_aggregate_aliases(&mut self, aliases: &[(String, usize)]) {
        for (expr_name, idx) in aliases {
            let lower = expr_name.to_lowercase();
            match u16::try_from(*idx) {
                Ok(index) => {
                    self.expression_aliases.insert(lower, index);
                }
                Err(_) => {
                    self.context_errors.push(format!(
                        "expression alias '{}' index {} exceeds the u16 bytecode limit",
                        expr_name, idx
                    ));
                }
            }
        }
        // Invalidate local cache since compilation context changed
        self.local_cache.clear();
    }

    /// Add expression aliases for HAVING clause with GROUP BY expressions
    pub fn add_expression_aliases(&mut self, aliases: &[(String, usize)]) {
        for (expr_str, idx) in aliases {
            let lower = expr_str.to_lowercase();
            match u16::try_from(*idx) {
                Ok(index) => {
                    self.expression_aliases.insert(lower, index);
                }
                Err(_) => {
                    self.context_errors.push(format!(
                        "expression alias '{}' index {} exceeds the u16 bytecode limit",
                        expr_str, idx
                    ));
                }
            }
        }
        // Invalidate local cache since compilation context changed
        self.local_cache.clear();
    }

    /// Set the row using array-based access (optimized - no map rebuilding)
    /// Call init_columns() once before using this method.
    #[inline]
    pub fn set_row_array(&mut self, row: &Row) {
        self.current_row = Some(row.clone());
        // Clear join mode
        self.current_row2 = None;
    }

    /// Set the outer row context for correlated subqueries
    /// Accepts `CompactArc<str>` keys directly to avoid conversion overhead.
    #[inline]
    pub fn set_outer_row(&mut self, outer_row: Option<&FxHashMap<CompactArc<str>, Value>>) {
        if let Some(outer) = outer_row {
            // Clone the map (CompactArc clones are cheap)
            let map = outer.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
            self.outer_row = Some(map);
        } else {
            self.outer_row = None;
        }
    }

    /// Set the outer row context by taking ownership
    /// Accepts `CompactArc<str>` keys directly to avoid conversion overhead.
    #[inline]
    pub fn set_outer_row_owned(&mut self, outer_row: FxHashMap<CompactArc<str>, Value>) {
        // Collect outer column names for compilation (convert CompactArc<str> to String for outer_columns)
        let outer_cols: Vec<String> = outer_row.keys().map(|k| k.to_string()).collect();
        self.outer_row = Some(outer_row);
        // Also set up outer_columns for compilation so LoadOuterColumn can be emitted
        if !outer_cols.is_empty() {
            // Sort for deterministic order
            let mut sorted_cols = outer_cols;
            sorted_cols.sort();
            self.outer_columns = Some(sorted_cols);
            let error = Self::column_limit_error(
                "outer row",
                self.outer_columns.as_ref().map_or(0, Vec::len),
            );
            self.replace_context_error("outer row", error);
            // Invalidate local cache since compilation context changed
            self.local_cache.clear();
        }
    }

    /// Take ownership of the outer row back (for reuse)
    /// Returns `CompactArc<str>` keys directly to avoid conversion overhead.
    #[inline]
    pub fn take_outer_row(&mut self) -> FxHashMap<CompactArc<str>, Value> {
        self.outer_row.take().unwrap_or_default()
    }

    /// Clear the outer row context
    #[inline]
    pub fn clear_outer_row(&mut self) {
        self.outer_row = None;
    }

    /// Compute hash of expression content for local cache key.
    /// Fast recursive hash that avoids string allocation.
    /// Uses FxHasher which is 2-5x faster than SipHash for small keys.
    #[inline]
    fn expr_hash(&self, expr: &Expression) -> u64 {
        let mut hasher = FxHasher::default();
        Self::hash_expression(expr, &mut hasher);
        hasher.finish()
    }

    /// Recursively hash an expression without string allocation
    fn hash_expression(expr: &Expression, hasher: &mut FxHasher) {
        // First hash the discriminant to distinguish variants
        std::mem::discriminant(expr).hash(hasher);

        match expr {
            Expression::Identifier(id) => {
                id.value_lower.hash(hasher);
            }
            Expression::QualifiedIdentifier(qid) => {
                qid.qualifier.value_lower.hash(hasher);
                qid.name.value_lower.hash(hasher);
            }
            Expression::IntegerLiteral(lit) => {
                lit.value.hash(hasher);
            }
            Expression::FloatLiteral(lit) => {
                lit.value.to_bits().hash(hasher);
            }
            Expression::StringLiteral(lit) => {
                lit.value.hash(hasher);
                lit.type_hint.hash(hasher);
            }
            Expression::BooleanLiteral(lit) => {
                lit.value.hash(hasher);
            }
            Expression::NullLiteral(_) => {
                // Just discriminant is enough
            }
            Expression::BoundValue(value) => {
                value.hash(hasher);
            }
            Expression::IntervalLiteral(lit) => {
                lit.value.hash(hasher);
                lit.unit.hash(hasher);
            }
            Expression::Parameter(param) => {
                param.index.hash(hasher);
                param.name.hash(hasher);
            }
            Expression::Prefix(prefix) => {
                std::mem::discriminant(&prefix.op_type).hash(hasher);
                Self::hash_expression(&prefix.right, hasher);
            }
            Expression::Infix(infix) => {
                std::mem::discriminant(&infix.op_type).hash(hasher);
                Self::hash_expression(&infix.left, hasher);
                Self::hash_expression(&infix.right, hasher);
            }
            Expression::List(list) => {
                list.elements.len().hash(hasher);
                for val in &list.elements {
                    Self::hash_expression(val, hasher);
                }
            }
            Expression::Distinct(dist) => {
                Self::hash_expression(&dist.expr, hasher);
            }
            Expression::Exists(exists) => {
                // Use pointer identity for hashing - avoids expensive Debug format allocation
                (exists.subquery.as_ref() as *const _ as usize).hash(hasher);
            }
            Expression::AllAny(aa) => {
                aa.operator.hash(hasher);
                std::mem::discriminant(&aa.all_any_type).hash(hasher);
                Self::hash_expression(&aa.left, hasher);
                // Use pointer identity for hashing - avoids expensive Debug format allocation
                (aa.subquery.as_ref() as *const _ as usize).hash(hasher);
            }
            Expression::In(in_expr) => {
                in_expr.not.hash(hasher);
                Self::hash_expression(&in_expr.left, hasher);
                Self::hash_expression(&in_expr.right, hasher);
            }
            Expression::InHashSet(in_hash) => {
                in_hash.not.hash(hasher);
                Self::hash_expression(&in_hash.column, hasher);
                let mut values: Vec<&Value> = in_hash.values.iter().collect();
                values.sort_unstable();
                values.hash(hasher);
            }
            Expression::Between(between) => {
                between.not.hash(hasher);
                Self::hash_expression(&between.expr, hasher);
                Self::hash_expression(&between.lower, hasher);
                Self::hash_expression(&between.upper, hasher);
            }
            Expression::Like(like) => {
                like.operator.hash(hasher);
                Self::hash_expression(&like.left, hasher);
                Self::hash_expression(&like.pattern, hasher);
                if let Some(ref escape) = like.escape {
                    true.hash(hasher);
                    Self::hash_expression(escape, hasher);
                } else {
                    false.hash(hasher);
                }
            }
            Expression::ScalarSubquery(sq) => {
                // Use pointer identity for hashing - avoids expensive Debug format allocation
                (sq.subquery.as_ref() as *const _ as usize).hash(hasher);
            }
            Expression::ExpressionList(list) => {
                list.expressions.len().hash(hasher);
                for expr in &list.expressions {
                    Self::hash_expression(expr, hasher);
                }
            }
            Expression::Case(case) => {
                if let Some(ref val) = case.value {
                    true.hash(hasher);
                    Self::hash_expression(val, hasher);
                } else {
                    false.hash(hasher);
                }
                case.when_clauses.len().hash(hasher);
                for when_clause in &case.when_clauses {
                    Self::hash_expression(&when_clause.condition, hasher);
                    Self::hash_expression(&when_clause.then_result, hasher);
                }
                if let Some(ref else_val) = case.else_value {
                    true.hash(hasher);
                    Self::hash_expression(else_val, hasher);
                } else {
                    false.hash(hasher);
                }
            }
            Expression::Cast(cast) => {
                Self::hash_expression(&cast.expr, hasher);
                cast.type_name.hash(hasher);
            }
            Expression::FunctionCall(func) => {
                func.function.hash(hasher);
                func.is_distinct.hash(hasher);
                func.arguments.len().hash(hasher);
                for arg in &func.arguments {
                    Self::hash_expression(arg, hasher);
                }
                if let Some(ref filter) = func.filter {
                    true.hash(hasher);
                    Self::hash_expression(filter, hasher);
                } else {
                    false.hash(hasher);
                }
            }
            Expression::Aliased(aliased) => {
                aliased.alias.value_lower.hash(hasher);
                Self::hash_expression(&aliased.expression, hasher);
            }
            Expression::Window(window) => {
                // Hash the FunctionCall directly (not as Expression)
                window.function.function.hash(hasher);
                window.function.is_distinct.hash(hasher);
                window.function.arguments.len().hash(hasher);
                for arg in &window.function.arguments {
                    Self::hash_expression(arg, hasher);
                }
                window.partition_by.len().hash(hasher);
                for expr in &window.partition_by {
                    Self::hash_expression(expr, hasher);
                }
                window.order_by.len().hash(hasher);
                for order in &window.order_by {
                    Self::hash_expression(&order.expression, hasher);
                    order.ascending.hash(hasher);
                    order.nulls_first.hash(hasher);
                }
            }
            Expression::TableSource(ts) => {
                ts.name.value_lower.hash(hasher);
                if let Some(ref alias) = ts.alias {
                    true.hash(hasher);
                    alias.value_lower.hash(hasher);
                } else {
                    false.hash(hasher);
                }
            }
            Expression::JoinSource(js) => {
                // Use pointer identity for hashing - avoids expensive Debug format allocation
                (js.as_ref() as *const _ as usize).hash(hasher);
            }
            Expression::SubquerySource(sq) => {
                if let Some(ref alias) = sq.alias {
                    true.hash(hasher);
                    alias.value_lower.hash(hasher);
                } else {
                    false.hash(hasher);
                }
                // Use pointer identity for hashing - avoids expensive Debug format allocation
                (sq.subquery.as_ref() as *const _ as usize).hash(hasher);
            }
            Expression::ValuesSource(vs) => {
                if let Some(ref alias) = vs.alias {
                    true.hash(hasher);
                    alias.value_lower.hash(hasher);
                } else {
                    false.hash(hasher);
                }
                vs.rows.len().hash(hasher);
            }
            Expression::CteReference(cte) => {
                cte.name.value_lower.hash(hasher);
            }
            Expression::FunctionTableSource(fts) => {
                fts.function.value_lower.hash(hasher);
                for arg in &fts.arguments {
                    Self::hash_expression(arg, hasher);
                }
            }
            Expression::Star(_) => {
                // Just discriminant
            }
            Expression::QualifiedStar(qs) => {
                qs.qualifier.hash(hasher);
            }
            Expression::Default(_) => {
                // Just discriminant
            }
        }
    }

    /// Get or compile a program for the expression.
    /// Uses local cache for fast lookup within single query evaluation.
    fn get_or_compile(&mut self, expr: &Expression) -> Result<SharedProgram> {
        if let Some(error) = self.context_error() {
            return Err(Error::invalid_argument(error));
        }
        let registry_generation = self.function_registry.generation();
        let mut expr_key = self.expr_hash(expr);
        expr_key ^= registry_generation.rotate_left(17);

        // Check local cache (fast path, no synchronization)
        if let Some(entry) = self.local_cache.get(&expr_key) {
            if entry.registry_generation == registry_generation && entry.expression == *expr {
                return Ok(CompactArc::clone(&entry.program));
            }
        }

        // Cache miss: compile the expression
        let program = CompactArc::new(self.compile_expression(expr)?);
        self.local_cache.insert(
            expr_key,
            LocalProgramCacheEntry {
                expression: expr.clone(),
                registry_generation,
                program: CompactArc::clone(&program),
            },
        );

        Ok(program)
    }

    /// Compile an expression to a Program
    fn compile_expression(&self, expr: &Expression) -> Result<Program> {
        if let Some(error) = self.context_error() {
            return Err(Error::invalid_argument(error));
        }
        let mut ctx = CompileContext::new(&self.columns, self.function_registry);

        // Add second row columns if available
        if let Some(ref cols2) = self.columns2 {
            ctx = ctx.with_second_row(cols2);
        }

        // Add outer columns if available
        if let Some(ref outer_cols) = self.outer_columns {
            ctx = ctx.with_outer_columns(outer_cols);
        }

        // Add expression aliases
        if !self.expression_aliases.is_empty() {
            ctx = ctx.with_expression_aliases(self.expression_aliases.clone());
        }

        // Add column aliases
        if !self.column_aliases.is_empty() {
            ctx = ctx.with_column_aliases(self.column_aliases.clone());
        }

        let compiler = ExprCompiler::new(&ctx);
        compiler
            .compile(expr)
            .map_err(|e| Error::internal(format!("Compile error: {}", e)))
    }

    /// Evaluate an expression to a Value
    pub fn evaluate(&mut self, expr: &Expression) -> Result<Value> {
        // Compile the expression first
        let program = self.get_or_compile(expr)?;

        // Static empty row for fallback
        static EMPTY_ROW: std::sync::LazyLock<Row> = std::sync::LazyLock::new(Row::new);

        // Get row data from owned copy
        let row = self.current_row.as_ref().unwrap_or(&EMPTY_ROW);

        // Get second row if in join mode
        let row2 = self.current_row2.as_ref();

        // Build execution context
        let mut ctx = if let Some(r2) = row2 {
            ExecuteContext::for_join(row, r2)
        } else {
            ExecuteContext::new(row)
        };

        // Add parameters
        if !self.params.is_empty() {
            ctx = ctx.with_params(&self.params);
        }

        // Add named parameters
        if !self.named_params.is_empty() {
            ctx = ctx.with_named_params(&self.named_params);
        }

        // Add outer row
        if let Some(ref outer) = self.outer_row {
            ctx = ctx.with_outer_row(outer);
        }

        // Add transaction ID
        ctx = ctx
            .with_transaction_id(self.transaction_id)
            .with_stored_function_invoker(self.stored_function_invoker.as_ref());

        // Execute
        self.vm.execute_cow(&program, &ctx)
    }

    /// Evaluate an expression as a boolean (for WHERE/HAVING clauses)
    ///
    /// Returns false for NULL results (SQL three-valued logic).
    pub fn evaluate_bool(&mut self, expr: &Expression) -> Result<bool> {
        // Compile the expression first
        let program = self.get_or_compile(expr)?;

        // Static empty row for fallback
        static EMPTY_ROW: std::sync::LazyLock<Row> = std::sync::LazyLock::new(Row::new);

        // Get row data from owned copy
        let row = self.current_row.as_ref().unwrap_or(&EMPTY_ROW);

        // Get second row if in join mode
        let row2 = self.current_row2.as_ref();

        // Build execution context
        let mut ctx = if let Some(r2) = row2 {
            ExecuteContext::for_join(row, r2)
        } else {
            ExecuteContext::new(row)
        };

        // Add parameters
        if !self.params.is_empty() {
            ctx = ctx.with_params(&self.params);
        }

        // Add named parameters
        if !self.named_params.is_empty() {
            ctx = ctx.with_named_params(&self.named_params);
        }

        // Add outer row
        if let Some(ref outer) = self.outer_row {
            ctx = ctx.with_outer_row(outer);
        }

        // Add transaction ID
        ctx = ctx
            .with_transaction_id(self.transaction_id)
            .with_stored_function_invoker(self.stored_function_invoker.as_ref());

        // Execute and convert to bool. This must use the checked VM path:
        // callers rely on `Result<bool>` to distinguish SQL false/NULL from
        // runtime expression errors.
        self.vm.execute_bool_checked(&program, &ctx)
    }
}

impl Default for CompiledEvaluator<'static> {
    fn default() -> Self {
        Self::with_defaults()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use radixdb_sql::ast::{
        Expression, FunctionCall, Identifier, InfixExpression, InfixOperator, IntegerLiteral,
    };
    use radixdb_sql::token::{Position, Token, TokenType};

    #[derive(Default)]
    struct MutableScalarOne;

    #[derive(Default)]
    struct MutableScalarTwo;

    macro_rules! mutable_scalar {
        ($type:ty, $value:expr) => {
            impl radixdb_functions::ScalarFunction for $type {
                fn name(&self) -> &str {
                    "MUTABLE_TEST"
                }

                fn info(&self) -> radixdb_functions::FunctionInfo {
                    radixdb_functions::FunctionInfo::new(
                        "MUTABLE_TEST",
                        radixdb_functions::FunctionType::Scalar,
                        "cache generation test",
                        radixdb_functions::FunctionSignature::new(
                            radixdb_functions::FunctionDataType::Integer,
                            vec![],
                            0,
                            0,
                        ),
                    )
                }

                fn evaluate(&self, _args: &[Value]) -> Result<Value> {
                    Ok(Value::Integer($value))
                }
            }
        };
    }

    mutable_scalar!(MutableScalarOne, 1);
    mutable_scalar!(MutableScalarTwo, 2);

    fn dummy_token() -> Token {
        Token::new(TokenType::Eof, "", Position::default())
    }

    fn make_identifier(name: &str) -> Expression {
        Expression::Identifier(Identifier {
            token: dummy_token(),
            value: name.into(),
            value_lower: name.to_lowercase().into(),
        })
    }

    fn make_int_literal(value: i64) -> Expression {
        Expression::IntegerLiteral(IntegerLiteral {
            token: dummy_token(),
            value,
        })
    }

    fn make_infix(left: Expression, op: InfixOperator, right: Expression) -> Expression {
        let op_str = match op {
            InfixOperator::GreaterThan => ">",
            InfixOperator::LessThan => "<",
            InfixOperator::Equal => "=",
            InfixOperator::Add => "+",
            InfixOperator::Multiply => "*",
            _ => "?",
        };
        Expression::Infix(InfixExpression {
            token: dummy_token(),
            left: Box::new(left),
            operator: op_str.into(),
            op_type: op,
            right: Box::new(right),
        })
    }

    fn make_function(name: &str) -> Expression {
        Expression::FunctionCall(Box::new(FunctionCall {
            token: dummy_token(),
            function: name.into(),
            arguments: Vec::new(),
            is_distinct: false,
            order_by: Vec::new(),
            filter: None,
        }))
    }

    // =========================================================================
    // compute_expression_hash tests
    // =========================================================================

    #[test]
    fn test_compute_expression_hash_same_expr() {
        let expr1 = make_int_literal(42);
        let expr2 = make_int_literal(42);
        assert_eq!(
            compute_expression_hash(&expr1),
            compute_expression_hash(&expr2)
        );
    }

    #[test]
    fn test_compute_expression_hash_different_expr() {
        let expr1 = make_int_literal(42);
        let expr2 = make_int_literal(43);
        assert_ne!(
            compute_expression_hash(&expr1),
            compute_expression_hash(&expr2)
        );
    }

    #[test]
    fn test_compute_expression_hash_complex() {
        // col > 5
        let expr1 = make_infix(
            make_identifier("col"),
            InfixOperator::GreaterThan,
            make_int_literal(5),
        );
        // col > 5 (same)
        let expr2 = make_infix(
            make_identifier("col"),
            InfixOperator::GreaterThan,
            make_int_literal(5),
        );
        assert_eq!(
            compute_expression_hash(&expr1),
            compute_expression_hash(&expr2)
        );
    }

    // =========================================================================
    // compile_expression tests
    // =========================================================================

    #[test]
    fn test_compile_expression_basic() {
        // col > 5
        let expr = make_infix(
            make_identifier("col"),
            InfixOperator::GreaterThan,
            make_int_literal(5),
        );
        let columns = vec!["col".to_string()];
        let program = compile_expression(&expr, &columns);
        assert!(program.is_ok());
    }

    #[test]
    fn test_compile_expression_unknown_column() {
        // unknown_col > 5
        let expr = make_infix(
            make_identifier("unknown_col"),
            InfixOperator::GreaterThan,
            make_int_literal(5),
        );
        let columns = vec!["col".to_string()];
        // Unknown columns cause compilation errors
        let program = compile_expression(&expr, &columns);
        assert!(program.is_err());
    }

    // =========================================================================
    // RowFilter tests
    // =========================================================================

    #[test]
    fn test_row_filter_new() {
        // col > 5
        let expr = make_infix(
            make_identifier("col"),
            InfixOperator::GreaterThan,
            make_int_literal(5),
        );
        let columns = vec!["col".to_string()];
        let filter = RowFilter::new(&expr, &columns);
        assert!(filter.is_ok());
    }

    #[test]
    fn test_row_filter_matches_true() {
        // col > 5
        let expr = make_infix(
            make_identifier("col"),
            InfixOperator::GreaterThan,
            make_int_literal(5),
        );
        let columns = vec!["col".to_string()];
        let filter = RowFilter::new(&expr, &columns).unwrap();

        // Row with col = 10 (> 5)
        let row = Row::from(vec![Value::Integer(10)]);
        assert!(filter.matches(&row).unwrap());
    }

    #[test]
    fn test_row_filter_matches_false() {
        // col > 5
        let expr = make_infix(
            make_identifier("col"),
            InfixOperator::GreaterThan,
            make_int_literal(5),
        );
        let columns = vec!["col".to_string()];
        let filter = RowFilter::new(&expr, &columns).unwrap();

        // Row with col = 3 (not > 5)
        let row = Row::from(vec![Value::Integer(3)]);
        assert!(!filter.matches(&row).unwrap());
    }

    #[test]
    fn test_row_filter_evaluate() {
        // col + 10
        let expr = make_infix(
            make_identifier("col"),
            InfixOperator::Add,
            make_int_literal(10),
        );
        let columns = vec!["col".to_string()];
        let filter = RowFilter::new(&expr, &columns).unwrap();

        let row = Row::from(vec![Value::Integer(5)]);
        let result = filter.evaluate(&row).unwrap();
        assert_eq!(result, Value::Integer(15));
    }

    #[test]
    fn test_row_filter_clone() {
        let expr = make_infix(
            make_identifier("col"),
            InfixOperator::GreaterThan,
            make_int_literal(5),
        );
        let columns = vec!["col".to_string()];
        let filter = RowFilter::new(&expr, &columns).unwrap();
        let cloned = filter.clone();

        let row = Row::from(vec![Value::Integer(10)]);
        assert!(filter.matches(&row).unwrap());
        assert!(cloned.matches(&row).unwrap());
    }

    // =========================================================================
    // ExpressionEval tests
    // =========================================================================

    #[test]
    fn test_expression_eval_compile() {
        let expr = make_infix(
            make_identifier("col"),
            InfixOperator::GreaterThan,
            make_int_literal(5),
        );
        let columns = vec!["col".to_string()];
        let eval = ExpressionEval::compile(&expr, &columns);
        assert!(eval.is_ok());
    }

    #[test]
    fn test_expression_eval_eval() {
        // col + 10
        let expr = make_infix(
            make_identifier("col"),
            InfixOperator::Add,
            make_int_literal(10),
        );
        let columns = vec!["col".to_string()];
        let mut eval = ExpressionEval::compile(&expr, &columns).unwrap();

        let row = Row::from(vec![Value::Integer(5)]);
        let result = eval.eval(&row).unwrap();
        assert_eq!(result, Value::Integer(15));
    }

    #[test]
    fn test_expression_eval_eval_bool() {
        // col > 5
        let expr = make_infix(
            make_identifier("col"),
            InfixOperator::GreaterThan,
            make_int_literal(5),
        );
        let columns = vec!["col".to_string()];
        let mut eval = ExpressionEval::compile(&expr, &columns).unwrap();

        let row = Row::from(vec![Value::Integer(10)]);
        assert!(eval.eval_bool(&row).unwrap());

        let row = Row::from(vec![Value::Integer(3)]);
        assert!(!eval.eval_bool(&row).unwrap());
    }

    // =========================================================================
    // MultiExpressionEval tests
    // =========================================================================

    #[test]
    fn test_multi_expression_eval_compile() {
        let expr1 = make_infix(
            make_identifier("col"),
            InfixOperator::Add,
            make_int_literal(10),
        );
        let expr2 = make_infix(
            make_identifier("col"),
            InfixOperator::Multiply,
            make_int_literal(2),
        );
        let columns = vec!["col".to_string()];

        let eval = MultiExpressionEval::compile(&[expr1, expr2], &columns);
        assert!(eval.is_ok());
        assert_eq!(eval.unwrap().len(), 2);
    }

    #[test]
    fn test_multi_expression_eval_all() {
        let expr1 = make_infix(
            make_identifier("col"),
            InfixOperator::Add,
            make_int_literal(10),
        );
        let expr2 = make_infix(
            make_identifier("col"),
            InfixOperator::Multiply,
            make_int_literal(2),
        );
        let columns = vec!["col".to_string()];
        let mut eval = MultiExpressionEval::compile(&[expr1, expr2], &columns).unwrap();

        let row = Row::from(vec![Value::Integer(5)]);
        let results = eval.eval_all(&row).unwrap();
        assert_eq!(results.len(), 2);
        assert_eq!(results[0], Value::Integer(15)); // 5 + 10
        assert_eq!(results[1], Value::Integer(10)); // 5 * 2
    }

    // =========================================================================
    // CompiledEvaluator tests
    // =========================================================================

    #[test]
    fn test_compiled_evaluator_with_defaults() {
        let eval = CompiledEvaluator::with_defaults();
        assert!(eval.columns.is_empty());
    }

    #[test]
    fn test_compiled_evaluator_init_columns() {
        let mut eval = CompiledEvaluator::with_defaults();
        eval.init_columns(&["col1".to_string(), "col2".to_string()]);
        assert_eq!(eval.columns.len(), 2);
    }

    #[test]
    fn compiled_evaluator_with_row_binds_owned_row_for_value_and_bool_evaluation() {
        let columns = vec!["col".to_string()];
        let mut eval = CompiledEvaluator::with_defaults()
            .with_row(Row::from(vec![Value::Integer(10)]), &columns);

        assert_eq!(
            eval.evaluate(&make_identifier("col")).unwrap(),
            Value::Integer(10)
        );
        assert!(eval
            .evaluate_bool(&make_infix(
                make_identifier("col"),
                InfixOperator::GreaterThan,
                make_int_literal(5),
            ))
            .unwrap());
    }

    #[test]
    fn test_compiled_evaluator_evaluate_bool() {
        let mut eval = CompiledEvaluator::with_defaults();
        eval.init_columns(&["col".to_string()]);
        let row = Row::from(vec![Value::Integer(10)]);
        eval.set_row_array(&row);

        // col > 5
        let expr = make_infix(
            make_identifier("col"),
            InfixOperator::GreaterThan,
            make_int_literal(5),
        );

        let result = eval.evaluate_bool(&expr);
        assert!(result.is_ok());
        assert!(result.unwrap());
    }

    #[test]
    fn test_compiled_evaluator_evaluate() {
        let mut eval = CompiledEvaluator::with_defaults();
        eval.init_columns(&["col".to_string()]);
        let row = Row::from(vec![Value::Integer(5)]);
        eval.set_row_array(&row);

        // col + 10
        let expr = make_infix(
            make_identifier("col"),
            InfixOperator::Add,
            make_int_literal(10),
        );

        let result = eval.evaluate(&expr);
        assert!(result.is_ok());
        assert_eq!(result.unwrap(), Value::Integer(15));
    }

    #[test]
    fn test_compiled_evaluator_default() {
        let eval = CompiledEvaluator::default();
        assert!(eval.columns.is_empty());
    }

    #[test]
    fn compiled_evaluator_rebinds_semantically_changed_schemas() {
        let mut eval = CompiledEvaluator::with_defaults();
        let expr = make_identifier("a");

        eval.init_columns(&["a".to_string(), "b".to_string()]);
        eval.set_row_array(&Row::from(vec![Value::Integer(1), Value::Integer(2)]));
        assert_eq!(eval.evaluate(&expr).unwrap(), Value::Integer(1));

        eval.init_columns(&["b".to_string(), "a".to_string()]);
        eval.set_row_array(&Row::from(vec![Value::Integer(1), Value::Integer(2)]));
        assert_eq!(eval.evaluate(&expr).unwrap(), Value::Integer(2));
    }

    #[test]
    fn compiled_evaluator_rejects_wide_schema_without_losing_other_errors() {
        let mut eval = CompiledEvaluator::with_defaults();
        let columns = (0..=u16::MAX as usize + 1)
            .map(|index| format!("c{index}"))
            .collect::<Vec<_>>();
        eval.init_columns(&columns);
        assert!(eval.evaluate(&make_int_literal(1)).is_err());

        eval.init_columns(&["c".to_string()]);
        eval.set_row_array(&Row::from(vec![Value::Integer(1)]));
        assert_eq!(
            eval.evaluate(&make_int_literal(1)).unwrap(),
            Value::Integer(1)
        );
    }

    #[test]
    fn join_filter_reads_deferred_projection_without_materializing_it() {
        let expression = make_infix(
            make_identifier("left_id"),
            InfixOperator::Equal,
            make_identifier("right_id"),
        );
        let filter = JoinFilter::new(
            &expression,
            &["left_id".to_string()],
            &["right_id".to_string()],
            global_registry(),
        )
        .unwrap();
        let left = crate::operator::RowRef::projected(
            crate::operator::RowRef::owned(Row::from_values(vec![
                Value::Integer(99),
                Value::Integer(7),
            ])),
            crate::operator::RowRef::owned(Row::new()),
            CompactArc::from(vec![crate::operator::ColumnSource::Outer(1)]),
        );

        assert!(left.is_deferred());
        assert!(filter
            .matches_row_ref_checked(&left, &Row::from_values(vec![Value::Integer(7)]))
            .unwrap());
        assert!(left.is_deferred());
    }

    #[test]
    fn row_filter_reads_portable_deferred_projection_without_materializing_it() {
        let expression = make_infix(
            make_identifier("status"),
            InfixOperator::Equal,
            make_int_literal(7),
        );
        let filter = RowFilter::new(&expression, &["status".to_string()]).unwrap();
        let row = radixdb_storage::DeferredRow::projected(
            radixdb_storage::DeferredRow::owned(Row::from_values(vec![
                Value::Integer(99),
                Value::Integer(7),
            ])),
            radixdb_storage::DeferredRow::owned(Row::new()),
            CompactArc::from(vec![radixdb_storage::DeferredColumnSource::Left(1)]),
        );

        assert!(row.is_deferred());
        assert!(filter.matches_deferred_checked(&row).unwrap());
        assert!(row.is_deferred());
    }

    #[test]
    fn row_filter_reads_executor_projection_without_materializing_it() {
        let expression = make_infix(
            make_identifier("status"),
            InfixOperator::Equal,
            make_int_literal(7),
        );
        let filter = RowFilter::new(&expression, &["status".to_string()]).unwrap();
        let row = crate::operator::RowRef::projected(
            crate::operator::RowRef::owned(Row::from_values(vec![
                Value::Integer(99),
                Value::Integer(7),
            ])),
            crate::operator::RowRef::owned(Row::new()),
            CompactArc::from(vec![crate::operator::ColumnSource::Outer(1)]),
        );

        assert!(row.is_deferred());
        assert!(filter.matches_row_ref_checked(&row).unwrap());
        assert!(row.is_deferred());
    }

    #[test]
    fn registry_generation_invalidates_embedded_function_programs() {
        let registry = FunctionRegistry::new();
        registry.register_scalar::<MutableScalarOne>();
        let mut eval = CompiledEvaluator::new(&registry);
        eval.set_row_array(&Row::new());
        let expr = make_function("MUTABLE_TEST");
        assert_eq!(eval.evaluate(&expr).unwrap(), Value::Integer(1));

        registry.register_scalar::<MutableScalarTwo>();
        assert_eq!(eval.evaluate(&expr).unwrap(), Value::Integer(2));
    }
}