vibesql-executor 0.2.0

//! Combined schema expression evaluator
//!
//! This module provides the CombinedExpressionEvaluator for evaluating expressions
//! in the context of combined schemas (e.g., JOINs with multiple tables).

use std::{
    cell::RefCell,
    collections::HashMap,
    hash::{Hash, Hasher},
    rc::Rc,
};

use ahash::AHasher;
use lru::LruCache;

use crate::{errors::ExecutorError, schema::CombinedSchema, select::WindowFunctionKey};

/// Evaluates expressions with combined schema (for JOINs)
pub struct CombinedExpressionEvaluator<'a> {
    pub(super) schema: &'a CombinedSchema,
    pub(super) database: Option<&'a vibesql_storage::Database>,
    pub(super) outer_row: Option<&'a vibesql_storage::Row>,
    pub(super) outer_schema: Option<&'a CombinedSchema>,
    /// All outer rows for outer-correlated aggregates (issue #4930)
    /// When an aggregate in a scalar subquery references only outer columns,
    /// it should aggregate over ALL outer rows, not just the current one.
    pub(super) outer_rows: Option<&'a [vibesql_storage::Row]>,
    /// Outer context for chained column resolution (SQLite-style context chaining)
    /// This enables proper scope shadowing for deeply nested subqueries (issue #4493)
    pub(super) outer_context: Option<&'a CombinedExpressionEvaluator<'a>>,
    pub(super) window_mapping: Option<&'a HashMap<WindowFunctionKey, usize>>,
    /// Procedural context for stored procedure/function variable resolution
    pub(super) procedural_context: Option<&'a crate::procedural::ExecutionContext>,
    /// Trigger context for OLD/NEW pseudo-variable resolution inside trigger bodies
    /// (e.g. UPDATE...FROM in a trigger body referencing NEW.x / OLD.x)
    pub(super) trigger_context: Option<&'a crate::trigger_execution::TriggerContext<'a>>,
    /// CTE (Common Table Expression) context for accessing WITH clause results
    pub(super) cte_context: Option<&'a HashMap<String, crate::select::cte::CteResult>>,
    /// Cache for column lookups to avoid repeated schema traversals
    /// Uses pre-computed hash of (table, column) as key to avoid string allocations
    column_cache: RefCell<HashMap<u64, usize>>,
    /// Cache for non-correlated subquery results with LRU eviction (key = subquery hash, value =
    /// result rows) Shared via Rc across child evaluators within a single statement execution.
    /// Cache lifetime is tied to the evaluator instance - each new evaluator gets a fresh cache.
    pub(super) subquery_cache: Rc<RefCell<LruCache<u64, Vec<vibesql_storage::Row>>>>,
    /// Current depth in expression tree (for preventing stack overflow)
    pub(super) depth: usize,
    /// CSE cache for common sub-expression elimination with LRU eviction (shared via Rc across
    /// depth levels)
    pub(super) cse_cache: Rc<RefCell<LruCache<u64, vibesql_types::SqlValue>>>,
    /// Whether CSE is enabled (can be disabled for debugging)
    pub(super) enable_cse: bool,
    /// Cache for subquery correlation analysis (key = subquery pointer address, value =
    /// is_correlated) Avoids expensive AST traversal for every row evaluation (issue #4142)
    pub(super) correlation_cache: Rc<RefCell<HashMap<usize, bool>>>,
    /// Cache for subquery hash values (key = subquery pointer address, value = hash)
    /// Avoids expensive Debug format + hash computation for every row evaluation (issue #4142)
    pub(super) subquery_hash_cache: Rc<RefCell<HashMap<usize, u64>>>,
}

impl<'a> CombinedExpressionEvaluator<'a> {
    /// Create a new combined expression evaluator
    /// Note: Currently unused as all callers use with_database(), but kept for API completeness
    #[allow(dead_code)]
    pub(crate) fn new(schema: &'a CombinedSchema) -> Self {
        CombinedExpressionEvaluator {
            schema,
            database: None,
            outer_row: None,
            outer_schema: None,
            outer_rows: None,
            outer_context: None,
            window_mapping: None,
            procedural_context: None,
            trigger_context: None,
            cte_context: None,
            column_cache: RefCell::new(HashMap::new()),
            subquery_cache: Rc::new(RefCell::new(super::caching::create_subquery_cache())),
            depth: 0,
            cse_cache: Rc::new(RefCell::new(super::caching::create_cse_cache())),
            enable_cse: super::caching::is_cse_enabled(),
            correlation_cache: Rc::new(RefCell::new(HashMap::new())),
            subquery_hash_cache: Rc::new(RefCell::new(HashMap::new())),
        }
    }

    /// Create a new combined expression evaluator with database reference
    pub(crate) fn with_database(
        schema: &'a CombinedSchema,
        database: &'a vibesql_storage::Database,
    ) -> Self {
        CombinedExpressionEvaluator {
            schema,
            database: Some(database),
            outer_row: None,
            outer_schema: None,
            outer_rows: None,
            outer_context: None,
            window_mapping: None,
            procedural_context: None,
            trigger_context: None,
            cte_context: None,
            column_cache: RefCell::new(HashMap::new()),
            subquery_cache: Rc::new(RefCell::new(super::caching::create_subquery_cache())),
            depth: 0,
            cse_cache: Rc::new(RefCell::new(super::caching::create_cse_cache())),
            enable_cse: super::caching::is_cse_enabled(),
            correlation_cache: Rc::new(RefCell::new(HashMap::new())),
            subquery_hash_cache: Rc::new(RefCell::new(HashMap::new())),
        }
    }

    /// Create a new combined expression evaluator with database and outer context for correlated
    /// subqueries
    pub(crate) fn with_database_and_outer_context(
        schema: &'a CombinedSchema,
        database: &'a vibesql_storage::Database,
        outer_row: &'a vibesql_storage::Row,
        outer_schema: &'a CombinedSchema,
    ) -> Self {
        CombinedExpressionEvaluator {
            schema,
            database: Some(database),
            outer_row: Some(outer_row),
            outer_schema: Some(outer_schema),
            outer_rows: None,
            outer_context: None,
            window_mapping: None,
            procedural_context: None,
            trigger_context: None,
            cte_context: None,
            column_cache: RefCell::new(HashMap::new()),
            subquery_cache: Rc::new(RefCell::new(super::caching::create_subquery_cache())),
            depth: 0,
            cse_cache: Rc::new(RefCell::new(super::caching::create_cse_cache())),
            enable_cse: super::caching::is_cse_enabled(),
            correlation_cache: Rc::new(RefCell::new(HashMap::new())),
            subquery_hash_cache: Rc::new(RefCell::new(HashMap::new())),
        }
    }

    /// Create a new combined expression evaluator with database and window mapping
    pub(crate) fn with_database_and_windows(
        schema: &'a CombinedSchema,
        database: &'a vibesql_storage::Database,
        window_mapping: &'a HashMap<WindowFunctionKey, usize>,
    ) -> Self {
        CombinedExpressionEvaluator {
            schema,
            database: Some(database),
            outer_row: None,
            outer_schema: None,
            outer_rows: None,
            outer_context: None,
            window_mapping: Some(window_mapping),
            procedural_context: None,
            trigger_context: None,
            cte_context: None,
            column_cache: RefCell::new(HashMap::new()),
            subquery_cache: Rc::new(RefCell::new(super::caching::create_subquery_cache())),
            depth: 0,
            cse_cache: Rc::new(RefCell::new(super::caching::create_cse_cache())),
            enable_cse: super::caching::is_cse_enabled(),
            correlation_cache: Rc::new(RefCell::new(HashMap::new())),
            subquery_hash_cache: Rc::new(RefCell::new(HashMap::new())),
        }
    }

    /// Create a new combined expression evaluator with database, outer context, and window mapping
    /// Used for window functions in correlated subqueries without FROM clause
    pub(crate) fn with_database_outer_and_windows(
        schema: &'a CombinedSchema,
        database: &'a vibesql_storage::Database,
        outer_row: &'a vibesql_storage::Row,
        outer_schema: &'a CombinedSchema,
        window_mapping: &'a HashMap<WindowFunctionKey, usize>,
    ) -> Self {
        CombinedExpressionEvaluator {
            schema,
            database: Some(database),
            outer_row: Some(outer_row),
            outer_schema: Some(outer_schema),
            outer_rows: None,
            outer_context: None,
            window_mapping: Some(window_mapping),
            procedural_context: None,
            trigger_context: None,
            cte_context: None,
            column_cache: RefCell::new(HashMap::new()),
            subquery_cache: Rc::new(RefCell::new(super::caching::create_subquery_cache())),
            depth: 0,
            cse_cache: Rc::new(RefCell::new(super::caching::create_cse_cache())),
            enable_cse: super::caching::is_cse_enabled(),
            correlation_cache: Rc::new(RefCell::new(HashMap::new())),
            subquery_hash_cache: Rc::new(RefCell::new(HashMap::new())),
        }
    }

    /// Create a new combined expression evaluator with database, window mapping, and CTE context
    pub(crate) fn with_database_and_windows_and_cte(
        schema: &'a CombinedSchema,
        database: &'a vibesql_storage::Database,
        window_mapping: &'a HashMap<WindowFunctionKey, usize>,
        cte_context: &'a HashMap<String, crate::select::cte::CteResult>,
    ) -> Self {
        CombinedExpressionEvaluator {
            schema,
            database: Some(database),
            outer_row: None,
            outer_schema: None,
            outer_rows: None,
            outer_context: None,
            window_mapping: Some(window_mapping),
            procedural_context: None,
            trigger_context: None,
            cte_context: Some(cte_context),
            column_cache: RefCell::new(HashMap::new()),
            subquery_cache: Rc::new(RefCell::new(super::caching::create_subquery_cache())),
            depth: 0,
            cse_cache: Rc::new(RefCell::new(super::caching::create_cse_cache())),
            enable_cse: super::caching::is_cse_enabled(),
            correlation_cache: Rc::new(RefCell::new(HashMap::new())),
            subquery_hash_cache: Rc::new(RefCell::new(HashMap::new())),
        }
    }

    /// Create a new combined expression evaluator with database and procedural context
    pub(crate) fn with_database_and_procedural_context(
        schema: &'a CombinedSchema,
        database: &'a vibesql_storage::Database,
        procedural_context: &'a crate::procedural::ExecutionContext,
    ) -> Self {
        CombinedExpressionEvaluator {
            schema,
            database: Some(database),
            outer_row: None,
            outer_schema: None,
            outer_rows: None,
            outer_context: None,
            window_mapping: None,
            procedural_context: Some(procedural_context),
            trigger_context: None,
            cte_context: None,
            column_cache: RefCell::new(HashMap::new()),
            subquery_cache: Rc::new(RefCell::new(super::caching::create_subquery_cache())),
            depth: 0,
            cse_cache: Rc::new(RefCell::new(super::caching::create_cse_cache())),
            enable_cse: super::caching::is_cse_enabled(),
            correlation_cache: Rc::new(RefCell::new(HashMap::new())),
            subquery_hash_cache: Rc::new(RefCell::new(HashMap::new())),
        }
    }

    /// Create a new combined expression evaluator with database and trigger context
    ///
    /// Used by the synthetic SELECT inside `UPDATE … FROM …` when the UPDATE is
    /// running inside a trigger body, so that `OLD.col` / `NEW.col` references in
    /// the WHERE clause and SET expressions can be resolved against the firing row.
    pub(crate) fn with_database_and_trigger_context(
        schema: &'a CombinedSchema,
        database: &'a vibesql_storage::Database,
        trigger_context: &'a crate::trigger_execution::TriggerContext<'a>,
    ) -> Self {
        CombinedExpressionEvaluator {
            schema,
            database: Some(database),
            outer_row: None,
            outer_schema: None,
            outer_rows: None,
            outer_context: None,
            window_mapping: None,
            procedural_context: None,
            trigger_context: Some(trigger_context),
            cte_context: None,
            column_cache: RefCell::new(HashMap::new()),
            subquery_cache: Rc::new(RefCell::new(super::caching::create_subquery_cache())),
            depth: 0,
            cse_cache: Rc::new(RefCell::new(super::caching::create_cse_cache())),
            enable_cse: super::caching::is_cse_enabled(),
            correlation_cache: Rc::new(RefCell::new(HashMap::new())),
            subquery_hash_cache: Rc::new(RefCell::new(HashMap::new())),
        }
    }

    /// Create a new combined expression evaluator with database and CTE context
    pub(crate) fn with_database_and_cte(
        schema: &'a CombinedSchema,
        database: &'a vibesql_storage::Database,
        cte_context: &'a HashMap<String, crate::select::cte::CteResult>,
    ) -> Self {
        CombinedExpressionEvaluator {
            schema,
            database: Some(database),
            outer_row: None,
            outer_schema: None,
            outer_rows: None,
            outer_context: None,
            window_mapping: None,
            procedural_context: None,
            trigger_context: None,
            cte_context: Some(cte_context),
            column_cache: RefCell::new(HashMap::new()),
            subquery_cache: Rc::new(RefCell::new(super::caching::create_subquery_cache())),
            depth: 0,
            cse_cache: Rc::new(RefCell::new(super::caching::create_cse_cache())),
            enable_cse: super::caching::is_cse_enabled(),
            correlation_cache: Rc::new(RefCell::new(HashMap::new())),
            subquery_hash_cache: Rc::new(RefCell::new(HashMap::new())),
        }
    }

    /// Create a new combined expression evaluator with database, outer context, and CTE context
    pub(crate) fn with_database_and_outer_context_and_cte(
        schema: &'a CombinedSchema,
        database: &'a vibesql_storage::Database,
        outer_row: &'a vibesql_storage::Row,
        outer_schema: &'a CombinedSchema,
        cte_context: &'a HashMap<String, crate::select::cte::CteResult>,
    ) -> Self {
        CombinedExpressionEvaluator {
            schema,
            database: Some(database),
            outer_row: Some(outer_row),
            outer_schema: Some(outer_schema),
            outer_rows: None,
            outer_context: None,
            window_mapping: None,
            procedural_context: None,
            trigger_context: None,
            cte_context: Some(cte_context),
            column_cache: RefCell::new(HashMap::new()),
            subquery_cache: Rc::new(RefCell::new(super::caching::create_subquery_cache())),
            depth: 0,
            cse_cache: Rc::new(RefCell::new(super::caching::create_cse_cache())),
            enable_cse: super::caching::is_cse_enabled(),
            correlation_cache: Rc::new(RefCell::new(HashMap::new())),
            subquery_hash_cache: Rc::new(RefCell::new(HashMap::new())),
        }
    }

    /// Create a new combined expression evaluator with database, procedural context, and CTE
    /// context
    pub(crate) fn with_database_and_procedural_context_and_cte(
        schema: &'a CombinedSchema,
        database: &'a vibesql_storage::Database,
        procedural_context: &'a crate::procedural::ExecutionContext,
        cte_context: &'a HashMap<String, crate::select::cte::CteResult>,
    ) -> Self {
        CombinedExpressionEvaluator {
            schema,
            database: Some(database),
            outer_row: None,
            outer_schema: None,
            outer_rows: None,
            outer_context: None,
            window_mapping: None,
            procedural_context: Some(procedural_context),
            trigger_context: None,
            cte_context: Some(cte_context),
            column_cache: RefCell::new(HashMap::new()),
            subquery_cache: Rc::new(RefCell::new(super::caching::create_subquery_cache())),
            depth: 0,
            cse_cache: Rc::new(RefCell::new(super::caching::create_cse_cache())),
            enable_cse: super::caching::is_cse_enabled(),
            correlation_cache: Rc::new(RefCell::new(HashMap::new())),
            subquery_hash_cache: Rc::new(RefCell::new(HashMap::new())),
        }
    }

    /// Attach an outer CTE context to this evaluator (builder style).
    ///
    /// Used by execution paths whose constructor combination has no
    /// dedicated `*_and_cte` variant (e.g. outer context + window mapping),
    /// so that subqueries evaluated in expression position resolve names
    /// bound by an enclosing WITH clause before falling back to catalog
    /// tables/views (issue #5350).
    pub(crate) fn with_cte_context(
        mut self,
        cte_context: &'a HashMap<String, crate::select::cte::CteResult>,
    ) -> Self {
        self.cte_context = Some(cte_context);
        self
    }

    /// Clear the CSE cache
    /// Should be called before evaluating expressions for a new row in multi-row contexts
    pub(crate) fn clear_cse_cache(&self) {
        self.cse_cache.borrow_mut().clear();
    }

    /// Set all outer rows for outer-correlated aggregates (issue #4930).
    ///
    /// When an aggregate function in a scalar subquery references only outer columns,
    /// it should aggregate over ALL outer rows, not just the current one.
    pub fn set_outer_rows(&mut self, outer_rows: &'a [vibesql_storage::Row]) {
        self.outer_rows = Some(outer_rows);
    }

    /// Get the outer rows if set
    pub(crate) fn get_outer_rows(&self) -> Option<&'a [vibesql_storage::Row]> {
        self.outer_rows
    }

    /// Get the outer schema if set
    pub(crate) fn get_outer_schema(&self) -> Option<&'a CombinedSchema> {
        self.outer_schema
    }

    /// Get the inner schema
    pub(crate) fn get_schema(&self) -> &'a CombinedSchema {
        self.schema
    }

    /// Compute hash key for column cache without allocating strings
    #[inline]
    fn column_cache_key(table: Option<&str>, column: &str) -> u64 {
        let mut hasher = AHasher::default();
        table.hash(&mut hasher);
        column.hash(&mut hasher);
        hasher.finish()
    }

    /// Get column index with caching to avoid repeated schema lookups
    pub(crate) fn get_column_index_cached(
        &self,
        table: Option<&str>,
        column: &str,
    ) -> Option<usize> {
        let key = Self::column_cache_key(table, column);

        // Check cache first
        if let Some(&idx) = self.column_cache.borrow().get(&key) {
            return Some(idx);
        }

        // Cache miss: lookup and store
        if let Some(idx) = self.schema.get_column_index(table, column) {
            self.column_cache.borrow_mut().insert(key, idx);
            Some(idx)
        } else {
            None
        }
    }

    /// Helper to execute a closure with incremented depth
    pub(super) fn with_incremented_depth<F, T>(&self, f: F) -> Result<T, ExecutorError>
    where
        F: FnOnce(&Self) -> Result<T, ExecutorError>,
    {
        // Create a new evaluator with incremented depth
        // Share caches between parent and child evaluators
        let evaluator = CombinedExpressionEvaluator {
            schema: self.schema,
            database: self.database,
            outer_row: self.outer_row,
            outer_schema: self.outer_schema,
            outer_rows: self.outer_rows,
            outer_context: self.outer_context,
            window_mapping: self.window_mapping,
            procedural_context: self.procedural_context,
            trigger_context: self.trigger_context,
            cte_context: self.cte_context,
            // Share the column cache between parent and child evaluators
            column_cache: RefCell::new(self.column_cache.borrow().clone()),
            // Share the subquery cache - subqueries can be reused across depths
            subquery_cache: self.subquery_cache.clone(),
            depth: self.depth + 1,
            cse_cache: self.cse_cache.clone(),
            enable_cse: self.enable_cse,
            // Share correlation and hash caches - they're keyed by AST pointer address
            correlation_cache: self.correlation_cache.clone(),
            subquery_hash_cache: self.subquery_hash_cache.clone(),
        };
        f(&evaluator)
    }

    /// Clone the evaluator for evaluating a different expression
    ///
    /// Shares the subquery cache (safe because non-correlated subqueries produce
    /// the same results regardless of the current row) but creates a fresh CSE cache
    /// (necessary because CSE results depend on row values).
    /// Also shares correlation and hash caches (keyed by AST pointer, not row-dependent).
    pub fn clone_for_new_expression(&self) -> Self {
        CombinedExpressionEvaluator {
            schema: self.schema,
            database: self.database,
            outer_row: self.outer_row,
            outer_schema: self.outer_schema,
            outer_rows: self.outer_rows,
            outer_context: self.outer_context,
            window_mapping: self.window_mapping,
            procedural_context: self.procedural_context,
            trigger_context: self.trigger_context,
            cte_context: self.cte_context,
            column_cache: RefCell::new(HashMap::new()),
            subquery_cache: self.subquery_cache.clone(),
            depth: self.depth,
            cse_cache: Rc::new(RefCell::new(super::caching::create_cse_cache())),
            enable_cse: self.enable_cse,
            // Share correlation and hash caches - they're keyed by AST pointer address
            correlation_cache: self.correlation_cache.clone(),
            subquery_hash_cache: self.subquery_hash_cache.clone(),
        }
    }

    /// Get the combined schema for this evaluator
    pub(crate) fn schema(&self) -> &'a CombinedSchema {
        self.schema
    }

    /// Get the database reference (if available)
    pub(crate) fn database(&self) -> Option<&'a vibesql_storage::Database> {
        self.database
    }

    /// Check if a subquery is correlated, with caching by AST pointer address.
    ///
    /// This avoids expensive AST traversal for every row evaluation.
    /// The correlation status of a subquery is determined by its AST structure,
    /// which doesn't change during query execution.
    #[inline]
    pub(crate) fn is_correlated_cached(&self, subquery: &vibesql_ast::SelectStmt) -> bool {
        let ptr = subquery as *const _ as usize;

        // Check cache first
        if let Some(&is_correlated) = self.correlation_cache.borrow().get(&ptr) {
            return is_correlated;
        }

        // Cache miss: compute and store
        let is_correlated =
            crate::optimizer::subquery_rewrite::correlation::is_correlated(subquery);
        self.correlation_cache.borrow_mut().insert(ptr, is_correlated);
        is_correlated
    }

    /// Compute subquery hash with caching by AST pointer address.
    ///
    /// This avoids expensive Debug format + hash computation for every row evaluation.
    /// The hash of a subquery is determined by its AST structure,
    /// which doesn't change during query execution.
    #[inline]
    pub(crate) fn compute_subquery_hash_cached(&self, subquery: &vibesql_ast::SelectStmt) -> u64 {
        let ptr = subquery as *const _ as usize;

        // Check cache first
        if let Some(&hash) = self.subquery_hash_cache.borrow().get(&ptr) {
            return hash;
        }

        // Cache miss: compute and store
        let hash = super::caching::compute_subquery_hash(subquery);
        self.subquery_hash_cache.borrow_mut().insert(ptr, hash);
        hash
    }

    /// Get evaluator components for parallel execution
    /// Returns (schema, database, outer_row, outer_schema, window_mapping, cte_context, enable_cse)
    ///
    /// Issue #3562: Now includes cte_context to enable IN subqueries referencing CTEs
    /// during parallel predicate evaluation.
    #[cfg(feature = "parallel")]
    pub(crate) fn get_parallel_components(&self) -> super::parallel::ParallelComponents<'a> {
        (
            self.schema,
            self.database,
            self.outer_row,
            self.outer_schema,
            self.window_mapping,
            self.cte_context,
            self.enable_cse,
        )
    }

    /// Create evaluator from parallel components
    /// Creates a fresh evaluator with independent caches for thread-safe parallel execution
    ///
    /// Issue #3562: Now accepts cte_context to enable IN subqueries referencing CTEs
    /// during parallel predicate evaluation.
    #[cfg(feature = "parallel")]
    pub(crate) fn from_parallel_components(
        schema: &'a CombinedSchema,
        database: Option<&'a vibesql_storage::Database>,
        outer_row: Option<&'a vibesql_storage::Row>,
        outer_schema: Option<&'a CombinedSchema>,
        window_mapping: Option<&'a HashMap<WindowFunctionKey, usize>>,
        cte_context: Option<
            &'a std::collections::HashMap<String, super::super::select::cte::CteResult>,
        >,
        enable_cse: bool,
    ) -> Self {
        CombinedExpressionEvaluator {
            schema,
            database,
            outer_row,
            outer_schema,
            outer_rows: None,
            outer_context: None,
            window_mapping,
            procedural_context: None,
            trigger_context: None,
            cte_context,
            column_cache: RefCell::new(HashMap::new()),
            subquery_cache: Rc::new(RefCell::new(super::caching::create_subquery_cache())),
            depth: 0,
            cse_cache: Rc::new(RefCell::new(super::caching::create_cse_cache())),
            correlation_cache: Rc::new(RefCell::new(HashMap::new())),
            subquery_hash_cache: Rc::new(RefCell::new(HashMap::new())),
            enable_cse,
        }
    }
}