featherdb_query/

prepared.rs

//! Prepared statements with plan caching
//!
//! This module provides prepared statement support for FeatherDB, enabling
//! 5-10x speedup for repeated queries by caching query plans.
//!
//! # Example
//!
//! ```rust,ignore
//! use featherdb::Database;
//!
//! let db = Database::open("myapp.db").unwrap();
//!
//! // Prepare a statement once
//! let stmt = db.prepare("SELECT * FROM users WHERE id = ?").unwrap();
//!
//! // Execute multiple times with different parameters
//! let rows = db.execute_prepared(&stmt, &[1.into()]).unwrap();
//! let rows = db.execute_prepared(&stmt, &[2.into()]).unwrap();
//! ```

use crate::expr::Expr;
use crate::planner::LogicalPlan;
use featherdb_core::{ColumnType, Error, Result, Value};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::RwLock;
use std::time::Instant;

/// A prepared statement with a cached, optimized query plan
#[derive(Debug, Clone)]
pub struct PreparedStatement {
    /// The original SQL string
    sql: String,
    /// The optimized logical plan (with parameter placeholders)
    plan: LogicalPlan,
    /// Expected types for parameters (inferred from context)
    param_types: Vec<Option<ColumnType>>,
    /// Number of parameters in the query
    param_count: usize,
    /// Unique statement ID
    id: u64,
    /// Subquery plans associated with this statement (subquery_id -> LogicalPlan).
    /// Stored alongside the plan so that cache hits return matching subquery plans.
    subquery_plans: HashMap<usize, LogicalPlan>,
}

impl PreparedStatement {
    /// Create a new prepared statement
    pub fn new(
        sql: String,
        plan: LogicalPlan,
        param_types: Vec<Option<ColumnType>>,
        param_count: usize,
        id: u64,
    ) -> Self {
        PreparedStatement {
            sql,
            plan,
            param_types,
            param_count,
            id,
            subquery_plans: HashMap::new(),
        }
    }

    /// Create a new prepared statement with subquery plans
    pub fn with_subquery_plans(
        sql: String,
        plan: LogicalPlan,
        param_types: Vec<Option<ColumnType>>,
        param_count: usize,
        id: u64,
        subquery_plans: HashMap<usize, LogicalPlan>,
    ) -> Self {
        PreparedStatement {
            sql,
            plan,
            param_types,
            param_count,
            id,
            subquery_plans,
        }
    }

    /// Get the SQL string
    pub fn sql(&self) -> &str {
        &self.sql
    }

    /// Get the cached plan
    pub fn plan(&self) -> &LogicalPlan {
        &self.plan
    }

    /// Get the expected parameter types
    pub fn param_types(&self) -> &[Option<ColumnType>] {
        &self.param_types
    }

    /// Get the number of parameters
    pub fn param_count(&self) -> usize {
        self.param_count
    }

    /// Get the statement ID
    pub fn id(&self) -> u64 {
        self.id
    }

    /// Get the subquery plans
    pub fn subquery_plans(&self) -> &HashMap<usize, LogicalPlan> {
        &self.subquery_plans
    }

    /// Validate that provided parameters match expected count
    pub fn validate_params(&self, params: &[Value]) -> Result<()> {
        if params.len() != self.param_count {
            return Err(Error::InvalidQuery {
                message: format!(
                    "Expected {} parameters, got {}",
                    self.param_count,
                    params.len()
                ),
            });
        }

        // Optionally validate types
        for (i, (param, expected_type)) in params.iter().zip(self.param_types.iter()).enumerate() {
            if let Some(expected) = expected_type {
                if !param.is_null() && !expected.is_compatible(param) {
                    return Err(Error::TypeError {
                        expected: format!("{:?} for parameter ${}", expected, i + 1),
                        actual: format!("{:?}", param.column_type()),
                    });
                }
            }
        }

        Ok(())
    }
}

/// Cache entry with metadata
#[derive(Debug)]
struct CacheEntry {
    /// The prepared statement
    stmt: PreparedStatement,
    /// Last access time
    last_access: Instant,
    /// Access count for frequency tracking
    access_count: u64,
}

/// LRU cache for query plans
///
/// Maps SQL strings to optimized LogicalPlans with configurable max size.
/// Supports cache invalidation on schema changes (DDL).
#[derive(Debug)]
pub struct PlanCache {
    /// The cache mapping SQL to prepared statements
    cache: RwLock<HashMap<String, CacheEntry>>,
    /// Maximum number of entries
    max_size: usize,
    /// Counter for statement IDs
    next_id: AtomicU64,
    /// Cache statistics
    stats: CacheStats,
    /// Schema version for invalidation
    schema_version: AtomicU64,
}

/// Statistics for the plan cache
#[derive(Debug, Default)]
pub struct CacheStats {
    /// Number of cache hits
    hits: AtomicU64,
    /// Number of cache misses
    misses: AtomicU64,
    /// Number of evictions
    evictions: AtomicU64,
    /// Number of invalidations
    invalidations: AtomicU64,
}

impl CacheStats {
    /// Get the number of cache hits
    pub fn hits(&self) -> u64 {
        self.hits.load(Ordering::Relaxed)
    }

    /// Get the number of cache misses
    pub fn misses(&self) -> u64 {
        self.misses.load(Ordering::Relaxed)
    }

    /// Get the number of evictions
    pub fn evictions(&self) -> u64 {
        self.evictions.load(Ordering::Relaxed)
    }

    /// Get the number of invalidations
    pub fn invalidations(&self) -> u64 {
        self.invalidations.load(Ordering::Relaxed)
    }

    /// Get the cache hit ratio
    pub fn hit_ratio(&self) -> f64 {
        let hits = self.hits() as f64;
        let total = hits + self.misses() as f64;
        if total == 0.0 {
            1.0
        } else {
            hits / total
        }
    }
}

impl PlanCache {
    /// Create a new plan cache with the specified maximum size
    pub fn new(max_size: usize) -> Self {
        PlanCache {
            cache: RwLock::new(HashMap::new()),
            max_size,
            next_id: AtomicU64::new(1),
            stats: CacheStats::default(),
            schema_version: AtomicU64::new(0),
        }
    }

    /// Create a plan cache with default size (1000 entries)
    pub fn default_size() -> Self {
        Self::new(1000)
    }

    /// Get a cached plan, if available
    pub fn get(&self, sql: &str) -> Option<PreparedStatement> {
        let mut cache = self.cache.write().ok()?;

        if let Some(entry) = cache.get_mut(sql) {
            entry.last_access = Instant::now();
            entry.access_count += 1;
            self.stats.hits.fetch_add(1, Ordering::Relaxed);
            Some(entry.stmt.clone())
        } else {
            self.stats.misses.fetch_add(1, Ordering::Relaxed);
            None
        }
    }

    /// Insert a prepared statement into the cache
    pub fn insert(&self, sql: String, stmt: PreparedStatement) {
        let mut cache = match self.cache.write() {
            Ok(c) => c,
            Err(_) => return,
        };

        // Evict if necessary
        if cache.len() >= self.max_size && !cache.contains_key(&sql) {
            self.evict_lru(&mut cache);
        }

        cache.insert(
            sql,
            CacheEntry {
                stmt,
                last_access: Instant::now(),
                access_count: 1,
            },
        );
    }

    /// Evict the least recently used entry
    fn evict_lru(&self, cache: &mut HashMap<String, CacheEntry>) {
        let oldest = cache
            .iter()
            .min_by_key(|(_, entry)| entry.last_access)
            .map(|(k, _)| k.clone());

        if let Some(key) = oldest {
            cache.remove(&key);
            self.stats.evictions.fetch_add(1, Ordering::Relaxed);
        }
    }

    /// Invalidate all cached plans (called on schema changes)
    pub fn invalidate_all(&self) {
        if let Ok(mut cache) = self.cache.write() {
            let count = cache.len() as u64;
            cache.clear();
            self.stats.invalidations.fetch_add(count, Ordering::Relaxed);
            self.schema_version.fetch_add(1, Ordering::Release);
        }
    }

    /// Invalidate plans for a specific table
    pub fn invalidate_table(&self, table_name: &str) {
        if let Ok(mut cache) = self.cache.write() {
            // Remove entries that reference this table
            // This is a simple substring check; could be made more precise
            let table_lower = table_name.to_lowercase();
            let keys_to_remove: Vec<String> = cache
                .keys()
                .filter(|sql| sql.to_lowercase().contains(&table_lower))
                .cloned()
                .collect();

            for key in keys_to_remove {
                cache.remove(&key);
                self.stats.invalidations.fetch_add(1, Ordering::Relaxed);
            }
            self.schema_version.fetch_add(1, Ordering::Release);
        }
    }

    /// Get the next statement ID
    pub fn next_id(&self) -> u64 {
        self.next_id.fetch_add(1, Ordering::Relaxed)
    }

    /// Get the current schema version
    pub fn schema_version(&self) -> u64 {
        self.schema_version.load(Ordering::Acquire)
    }

    /// Get cache statistics
    pub fn stats(&self) -> &CacheStats {
        &self.stats
    }

    /// Get the number of cached entries
    pub fn len(&self) -> usize {
        self.cache.read().map(|c| c.len()).unwrap_or(0)
    }

    /// Check if the cache is empty
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Get the maximum cache size
    pub fn max_size(&self) -> usize {
        self.max_size
    }
}

impl Default for PlanCache {
    fn default() -> Self {
        Self::default_size()
    }
}

/// Substitute parameters in an expression
pub fn substitute_params(expr: &Expr, params: &[Value]) -> Result<Expr> {
    match expr {
        Expr::Parameter { index } => {
            if *index >= params.len() {
                return Err(Error::InvalidQuery {
                    message: format!(
                        "Parameter ${} referenced but only {} parameters provided",
                        index + 1,
                        params.len()
                    ),
                });
            }
            Ok(Expr::Literal(params[*index].clone()))
        }

        Expr::BinaryOp { left, op, right } => Ok(Expr::BinaryOp {
            left: Box::new(substitute_params(left, params)?),
            op: *op,
            right: Box::new(substitute_params(right, params)?),
        }),

        Expr::UnaryOp { op, expr } => Ok(Expr::UnaryOp {
            op: *op,
            expr: Box::new(substitute_params(expr, params)?),
        }),

        Expr::Function { name, args } => {
            let substituted_args: Vec<Expr> = args
                .iter()
                .map(|a| substitute_params(a, params))
                .collect::<Result<_>>()?;
            Ok(Expr::Function {
                name: name.clone(),
                args: substituted_args,
            })
        }

        Expr::Aggregate {
            func,
            arg,
            distinct,
        } => {
            let substituted_arg = arg
                .as_ref()
                .map(|a| substitute_params(a, params).map(Box::new))
                .transpose()?;
            Ok(Expr::Aggregate {
                func: func.clone(),
                arg: substituted_arg,
                distinct: *distinct,
            })
        }

        Expr::Case {
            operand,
            when_clauses,
            else_result,
        } => {
            let substituted_operand = operand
                .as_ref()
                .map(|o| substitute_params(o, params).map(Box::new))
                .transpose()?;

            let substituted_whens: Vec<(Expr, Expr)> = when_clauses
                .iter()
                .map(|(w, t)| Ok((substitute_params(w, params)?, substitute_params(t, params)?)))
                .collect::<Result<_>>()?;

            let substituted_else = else_result
                .as_ref()
                .map(|e| substitute_params(e, params).map(Box::new))
                .transpose()?;

            Ok(Expr::Case {
                operand: substituted_operand,
                when_clauses: substituted_whens,
                else_result: substituted_else,
            })
        }

        Expr::Between {
            expr,
            low,
            high,
            negated,
        } => Ok(Expr::Between {
            expr: Box::new(substitute_params(expr, params)?),
            low: Box::new(substitute_params(low, params)?),
            high: Box::new(substitute_params(high, params)?),
            negated: *negated,
        }),

        Expr::InList {
            expr,
            list,
            negated,
        } => {
            let substituted_list: Vec<Expr> = list
                .iter()
                .map(|i| substitute_params(i, params))
                .collect::<Result<_>>()?;
            Ok(Expr::InList {
                expr: Box::new(substitute_params(expr, params)?),
                list: substituted_list,
                negated: *negated,
            })
        }

        Expr::Like {
            expr,
            pattern,
            negated,
        } => Ok(Expr::Like {
            expr: Box::new(substitute_params(expr, params)?),
            pattern: pattern.clone(),
            negated: *negated,
        }),

        // Expressions without sub-expressions
        Expr::Column { .. } | Expr::Literal(_) | Expr::Wildcard | Expr::QualifiedWildcard(_) => {
            Ok(expr.clone())
        }

        // Window functions
        Expr::Window(wf) => {
            use crate::expr::WindowOrderByExpr;

            let substituted_partition_by = wf
                .partition_by
                .iter()
                .map(|e| substitute_params(e, params))
                .collect::<Result<Vec<_>>>()?;
            let substituted_order_by = wf
                .order_by
                .iter()
                .map(|window_order_expr| {
                    Ok(WindowOrderByExpr {
                        expr: substitute_params(&window_order_expr.expr, params)?,
                        order: window_order_expr.order,
                        nulls_first: window_order_expr.nulls_first,
                    })
                })
                .collect::<Result<Vec<_>>>()?;

            Ok(Expr::Window(crate::expr::WindowFunction {
                function: wf.function.clone(),
                partition_by: substituted_partition_by,
                order_by: substituted_order_by,
                frame: wf.frame.clone(),
            }))
        }

        // Subquery expressions - these reference subquery IDs, not inline subqueries
        Expr::ScalarSubquery {
            subquery_id,
            correlated,
        } => Ok(Expr::ScalarSubquery {
            subquery_id: *subquery_id,
            correlated: *correlated,
        }),

        Expr::Exists {
            subquery_id,
            negated,
        } => Ok(Expr::Exists {
            subquery_id: *subquery_id,
            negated: *negated,
        }),

        Expr::InSubquery {
            expr,
            subquery_id,
            negated,
        } => Ok(Expr::InSubquery {
            expr: Box::new(substitute_params(expr, params)?),
            subquery_id: *subquery_id,
            negated: *negated,
        }),

        Expr::AnySubquery {
            expr,
            op,
            subquery_id,
        } => Ok(Expr::AnySubquery {
            expr: Box::new(substitute_params(expr, params)?),
            op: *op,
            subquery_id: *subquery_id,
        }),

        Expr::AllSubquery {
            expr,
            op,
            subquery_id,
        } => Ok(Expr::AllSubquery {
            expr: Box::new(substitute_params(expr, params)?),
            op: *op,
            subquery_id: *subquery_id,
        }),
    }
}

/// Substitute parameters in a logical plan
pub fn substitute_plan_params(plan: &LogicalPlan, params: &[Value]) -> Result<LogicalPlan> {
    match plan {
        LogicalPlan::Scan {
            table,
            alias,
            projection,
            filter,
        } => {
            let substituted_filter = filter
                .as_ref()
                .map(|f| substitute_params(f, params))
                .transpose()?;
            Ok(LogicalPlan::Scan {
                table: table.clone(),
                alias: alias.clone(),
                projection: projection.clone(),
                filter: substituted_filter,
            })
        }

        LogicalPlan::IndexScan {
            table,
            alias,
            index,
            index_column,
            range,
            residual_filter,
            projection,
        } => {
            let substituted_filter = residual_filter
                .as_ref()
                .map(|f| substitute_params(f, params))
                .transpose()?;
            Ok(LogicalPlan::IndexScan {
                table: table.clone(),
                alias: alias.clone(),
                index: index.clone(),
                index_column: *index_column,
                range: range.clone(),
                residual_filter: substituted_filter,
                projection: projection.clone(),
            })
        }

        LogicalPlan::PkSeek {
            table,
            alias,
            key_values,
            residual_filter,
            projection,
        } => {
            let substituted_keys: Vec<Expr> = key_values
                .iter()
                .map(|e| substitute_params(e, params))
                .collect::<Result<_>>()?;
            let substituted_filter = residual_filter
                .as_ref()
                .map(|f| substitute_params(f, params))
                .transpose()?;
            Ok(LogicalPlan::PkSeek {
                table: table.clone(),
                alias: alias.clone(),
                key_values: substituted_keys,
                residual_filter: substituted_filter,
                projection: projection.clone(),
            })
        }

        LogicalPlan::PkRangeScan {
            table,
            alias,
            range,
            residual_filter,
            projection,
        } => {
            let substituted_filter = residual_filter
                .as_ref()
                .map(|f| substitute_params(f, params))
                .transpose()?;
            Ok(LogicalPlan::PkRangeScan {
                table: table.clone(),
                alias: alias.clone(),
                range: range.clone(),
                residual_filter: substituted_filter,
                projection: projection.clone(),
            })
        }

        LogicalPlan::Filter { input, predicate } => Ok(LogicalPlan::Filter {
            input: Box::new(substitute_plan_params(input, params)?),
            predicate: substitute_params(predicate, params)?,
        }),

        LogicalPlan::Project { input, exprs } => {
            let substituted_exprs: Vec<(Expr, String)> = exprs
                .iter()
                .map(|(e, name)| Ok((substitute_params(e, params)?, name.clone())))
                .collect::<Result<_>>()?;
            Ok(LogicalPlan::Project {
                input: Box::new(substitute_plan_params(input, params)?),
                exprs: substituted_exprs,
            })
        }

        LogicalPlan::Join {
            left,
            right,
            join_type,
            condition,
        } => {
            let substituted_condition = condition
                .as_ref()
                .map(|c| substitute_params(c, params))
                .transpose()?;
            Ok(LogicalPlan::Join {
                left: Box::new(substitute_plan_params(left, params)?),
                right: Box::new(substitute_plan_params(right, params)?),
                join_type: *join_type,
                condition: substituted_condition,
            })
        }

        LogicalPlan::Aggregate {
            input,
            group_by,
            aggregates,
        } => {
            let substituted_group_by: Vec<Expr> = group_by
                .iter()
                .map(|e| substitute_params(e, params))
                .collect::<Result<_>>()?;
            let substituted_aggregates: Vec<(Expr, String)> = aggregates
                .iter()
                .map(|(e, name)| Ok((substitute_params(e, params)?, name.clone())))
                .collect::<Result<_>>()?;
            Ok(LogicalPlan::Aggregate {
                input: Box::new(substitute_plan_params(input, params)?),
                group_by: substituted_group_by,
                aggregates: substituted_aggregates,
            })
        }

        LogicalPlan::Sort { input, order_by } => {
            let substituted_order_by: Vec<_> = order_by
                .iter()
                .map(|(e, order)| Ok((substitute_params(e, params)?, *order)))
                .collect::<Result<_>>()?;
            Ok(LogicalPlan::Sort {
                input: Box::new(substitute_plan_params(input, params)?),
                order_by: substituted_order_by,
            })
        }

        LogicalPlan::Limit {
            input,
            limit,
            offset,
        } => Ok(LogicalPlan::Limit {
            input: Box::new(substitute_plan_params(input, params)?),
            limit: *limit,
            offset: *offset,
        }),

        LogicalPlan::Distinct { input } => Ok(LogicalPlan::Distinct {
            input: Box::new(substitute_plan_params(input, params)?),
        }),

        LogicalPlan::Insert {
            table,
            columns,
            values,
        } => {
            let substituted_values: Vec<Vec<Expr>> = values
                .iter()
                .map(|row| {
                    row.iter()
                        .map(|e| substitute_params(e, params))
                        .collect::<Result<_>>()
                })
                .collect::<Result<_>>()?;
            Ok(LogicalPlan::Insert {
                table: table.clone(),
                columns: columns.clone(),
                values: substituted_values,
            })
        }

        LogicalPlan::Update {
            table,
            assignments,
            filter,
        } => {
            let substituted_assignments: Vec<(String, Expr)> = assignments
                .iter()
                .map(|(name, e)| Ok((name.clone(), substitute_params(e, params)?)))
                .collect::<Result<_>>()?;
            let substituted_filter = filter
                .as_ref()
                .map(|f| substitute_params(f, params))
                .transpose()?;
            Ok(LogicalPlan::Update {
                table: table.clone(),
                assignments: substituted_assignments,
                filter: substituted_filter,
            })
        }

        LogicalPlan::Delete { table, filter } => {
            let substituted_filter = filter
                .as_ref()
                .map(|f| substitute_params(f, params))
                .transpose()?;
            Ok(LogicalPlan::Delete {
                table: table.clone(),
                filter: substituted_filter,
            })
        }

        // DDL plans don't have parameters
        LogicalPlan::CreateTable { .. }
        | LogicalPlan::DropTable { .. }
        | LogicalPlan::AlterTable { .. }
        | LogicalPlan::EmptyRelation => Ok(plan.clone()),

        LogicalPlan::Explain {
            input,
            verbose,
            analyze,
        } => {
            let substituted_input = substitute_plan_params(input, params)?;
            Ok(LogicalPlan::Explain {
                input: Box::new(substituted_input),
                verbose: *verbose,
                analyze: *analyze,
            })
        }

        LogicalPlan::Window {
            input,
            window_exprs,
        } => {
            let substituted_input = substitute_plan_params(input, params)?;
            let substituted_exprs: Vec<(Expr, String)> = window_exprs
                .iter()
                .map(|(e, name)| Ok((substitute_params(e, params)?, name.clone())))
                .collect::<Result<_>>()?;
            Ok(LogicalPlan::Window {
                input: Box::new(substituted_input),
                window_exprs: substituted_exprs,
            })
        }

        LogicalPlan::WithCte { ctes, query } => {
            // Substitute params in CTE definitions and main query
            let substituted_ctes = ctes
                .iter()
                .map(|cte| {
                    Ok(crate::planner::CommonTableExpression {
                        name: cte.name.clone(),
                        columns: cte.columns.clone(),
                        query: Box::new(substitute_plan_params(&cte.query, params)?),
                        recursive: cte.recursive,
                    })
                })
                .collect::<Result<_>>()?;
            Ok(LogicalPlan::WithCte {
                ctes: substituted_ctes,
                query: Box::new(substitute_plan_params(query, params)?),
            })
        }

        LogicalPlan::CteScan {
            cte_name,
            alias,
            columns,
        } => Ok(LogicalPlan::CteScan {
            cte_name: cte_name.clone(),
            alias: alias.clone(),
            columns: columns.clone(),
        }),

        LogicalPlan::SubqueryScan {
            subquery,
            subquery_id,
            alias,
        } => Ok(LogicalPlan::SubqueryScan {
            subquery: Box::new(substitute_plan_params(subquery, params)?),
            subquery_id: *subquery_id,
            alias: alias.clone(),
        }),

        LogicalPlan::SemiJoin {
            left,
            right,
            condition,
            positive,
        } => {
            let substituted_condition = condition
                .as_ref()
                .map(|c| substitute_params(c, params))
                .transpose()?;
            Ok(LogicalPlan::SemiJoin {
                left: Box::new(substitute_plan_params(left, params)?),
                right: Box::new(substitute_plan_params(right, params)?),
                condition: substituted_condition,
                positive: *positive,
            })
        }

        LogicalPlan::AntiJoin {
            left,
            right,
            condition,
        } => {
            let substituted_condition = condition
                .as_ref()
                .map(|c| substitute_params(c, params))
                .transpose()?;
            Ok(LogicalPlan::AntiJoin {
                left: Box::new(substitute_plan_params(left, params)?),
                right: Box::new(substitute_plan_params(right, params)?),
                condition: substituted_condition,
            })
        }

        // Permission statements don't have parameters
        LogicalPlan::Grant {
            privileges,
            table,
            grantee,
        } => Ok(LogicalPlan::Grant {
            privileges: *privileges,
            table: table.clone(),
            grantee: grantee.clone(),
        }),

        LogicalPlan::Revoke {
            privileges,
            table,
            grantee,
        } => Ok(LogicalPlan::Revoke {
            privileges: *privileges,
            table: table.clone(),
            grantee: grantee.clone(),
        }),

        LogicalPlan::ShowGrants { target } => Ok(LogicalPlan::ShowGrants {
            target: target.clone(),
        }),

        LogicalPlan::Union { left, right, all } => Ok(LogicalPlan::Union {
            left: Box::new(substitute_plan_params(left, params)?),
            right: Box::new(substitute_plan_params(right, params)?),
            all: *all,
        }),

        LogicalPlan::Intersect { left, right, all } => Ok(LogicalPlan::Intersect {
            left: Box::new(substitute_plan_params(left, params)?),
            right: Box::new(substitute_plan_params(right, params)?),
            all: *all,
        }),

        LogicalPlan::Except { left, right, all } => Ok(LogicalPlan::Except {
            left: Box::new(substitute_plan_params(left, params)?),
            right: Box::new(substitute_plan_params(right, params)?),
            all: *all,
        }),

        LogicalPlan::CreateView {
            name,
            query,
            query_sql,
            columns,
        } => Ok(LogicalPlan::CreateView {
            name: name.clone(),
            query: Box::new(substitute_plan_params(query, params)?),
            query_sql: query_sql.clone(),
            columns: columns.clone(),
        }),

        LogicalPlan::DropView { name, if_exists } => Ok(LogicalPlan::DropView {
            name: name.clone(),
            if_exists: *if_exists,
        }),

        LogicalPlan::CreateIndex {
            index_name,
            table_name,
            columns,
            unique,
        } => Ok(LogicalPlan::CreateIndex {
            index_name: index_name.clone(),
            table_name: table_name.clone(),
            columns: columns.clone(),
            unique: *unique,
        }),
    }
}

/// Count parameters in an expression
pub fn count_params(expr: &Expr) -> usize {
    match expr {
        Expr::Parameter { index } => *index + 1,

        Expr::BinaryOp { left, right, .. } => count_params(left).max(count_params(right)),

        Expr::UnaryOp { expr, .. } => count_params(expr),

        Expr::Function { args, .. } => args.iter().map(count_params).max().unwrap_or(0),

        Expr::Aggregate { arg, .. } => arg.as_ref().map(|a| count_params(a)).unwrap_or(0),

        Expr::Case {
            operand,
            when_clauses,
            else_result,
        } => {
            let operand_count = operand.as_ref().map(|o| count_params(o)).unwrap_or(0);
            let when_count = when_clauses
                .iter()
                .map(|(w, t)| count_params(w).max(count_params(t)))
                .max()
                .unwrap_or(0);
            let else_count = else_result.as_ref().map(|e| count_params(e)).unwrap_or(0);
            operand_count.max(when_count).max(else_count)
        }

        Expr::Between {
            expr, low, high, ..
        } => count_params(expr)
            .max(count_params(low))
            .max(count_params(high)),

        Expr::InList { expr, list, .. } => {
            let list_max = list.iter().map(count_params).max().unwrap_or(0);
            count_params(expr).max(list_max)
        }

        Expr::Like { expr, .. } => count_params(expr),

        // Expressions without sub-expressions that contain parameters
        Expr::Column { .. } | Expr::Literal(_) | Expr::Wildcard | Expr::QualifiedWildcard(_) => 0,

        // Window function - count params in the window function expression
        Expr::Window(wf) => {
            // Count parameters in partition_by and order_by expressions
            let partition_max = wf.partition_by.iter().map(count_params).max().unwrap_or(0);
            let order_max = wf
                .order_by
                .iter()
                .map(|window_order_expr| count_params(&window_order_expr.expr))
                .max()
                .unwrap_or(0);
            partition_max.max(order_max)
        }

        // Subquery expressions - subqueries don't contain parameters in the expr itself
        // The actual subquery plan is handled separately
        Expr::ScalarSubquery { .. } | Expr::Exists { .. } => 0,

        Expr::InSubquery { expr, .. }
        | Expr::AnySubquery { expr, .. }
        | Expr::AllSubquery { expr, .. } => count_params(expr),
    }
}

/// Count parameters in a logical plan
pub fn count_plan_params(plan: &LogicalPlan) -> usize {
    match plan {
        LogicalPlan::Scan { filter, .. } => filter.as_ref().map(count_params).unwrap_or(0),

        LogicalPlan::IndexScan {
            residual_filter, ..
        } => residual_filter.as_ref().map(count_params).unwrap_or(0),

        LogicalPlan::PkSeek {
            key_values,
            residual_filter,
            ..
        } => {
            let key_max = key_values.iter().map(count_params).max().unwrap_or(0);
            let filter_max = residual_filter.as_ref().map(count_params).unwrap_or(0);
            key_max.max(filter_max)
        }

        LogicalPlan::PkRangeScan {
            residual_filter, ..
        } => residual_filter.as_ref().map(count_params).unwrap_or(0),

        LogicalPlan::Filter { input, predicate } => {
            count_plan_params(input).max(count_params(predicate))
        }

        LogicalPlan::Project { input, exprs } => {
            let expr_max = exprs
                .iter()
                .map(|(e, _)| count_params(e))
                .max()
                .unwrap_or(0);
            count_plan_params(input).max(expr_max)
        }

        LogicalPlan::Join {
            left,
            right,
            condition,
            ..
        } => {
            let cond_count = condition.as_ref().map(count_params).unwrap_or(0);
            count_plan_params(left)
                .max(count_plan_params(right))
                .max(cond_count)
        }

        LogicalPlan::Aggregate {
            input,
            group_by,
            aggregates,
        } => {
            let group_max = group_by.iter().map(count_params).max().unwrap_or(0);
            let agg_max = aggregates
                .iter()
                .map(|(e, _)| count_params(e))
                .max()
                .unwrap_or(0);
            count_plan_params(input).max(group_max).max(agg_max)
        }

        LogicalPlan::Sort { input, order_by } => {
            let order_max = order_by
                .iter()
                .map(|(e, _)| count_params(e))
                .max()
                .unwrap_or(0);
            count_plan_params(input).max(order_max)
        }

        LogicalPlan::Limit { input, .. } | LogicalPlan::Distinct { input } => {
            count_plan_params(input)
        }

        LogicalPlan::Insert { values, .. } => values
            .iter()
            .flat_map(|row| row.iter())
            .map(count_params)
            .max()
            .unwrap_or(0),

        LogicalPlan::Update {
            assignments,
            filter,
            ..
        } => {
            let assign_max = assignments
                .iter()
                .map(|(_, e)| count_params(e))
                .max()
                .unwrap_or(0);
            let filter_max = filter.as_ref().map(count_params).unwrap_or(0);
            assign_max.max(filter_max)
        }

        LogicalPlan::Delete { filter, .. } => filter.as_ref().map(count_params).unwrap_or(0),

        LogicalPlan::CreateTable { .. }
        | LogicalPlan::DropTable { .. }
        | LogicalPlan::AlterTable { .. }
        | LogicalPlan::EmptyRelation
        | LogicalPlan::CteScan { .. } => 0,

        LogicalPlan::Explain { input, .. } => count_plan_params(input),

        LogicalPlan::Window {
            input,
            window_exprs,
        } => {
            let expr_max = window_exprs
                .iter()
                .map(|(e, _)| count_params(e))
                .max()
                .unwrap_or(0);
            count_plan_params(input).max(expr_max)
        }

        LogicalPlan::WithCte { ctes, query } => {
            let cte_max = ctes
                .iter()
                .map(|cte| count_plan_params(&cte.query))
                .max()
                .unwrap_or(0);
            count_plan_params(query).max(cte_max)
        }

        LogicalPlan::SubqueryScan { subquery, .. } => count_plan_params(subquery),

        LogicalPlan::SemiJoin {
            left,
            right,
            condition,
            ..
        } => {
            let cond_count = condition.as_ref().map(count_params).unwrap_or(0);
            count_plan_params(left)
                .max(count_plan_params(right))
                .max(cond_count)
        }

        LogicalPlan::AntiJoin {
            left,
            right,
            condition,
        } => {
            let cond_count = condition.as_ref().map(count_params).unwrap_or(0);
            count_plan_params(left)
                .max(count_plan_params(right))
                .max(cond_count)
        }

        // Permission statements don't have parameters
        LogicalPlan::Grant { .. } | LogicalPlan::Revoke { .. } | LogicalPlan::ShowGrants { .. } => {
            0
        }

        LogicalPlan::Union { left, right, .. }
        | LogicalPlan::Intersect { left, right, .. }
        | LogicalPlan::Except { left, right, .. } => {
            count_plan_params(left).max(count_plan_params(right))
        }

        LogicalPlan::CreateView { query, .. } => count_plan_params(query),

        LogicalPlan::DropView { .. } => 0,
        LogicalPlan::CreateIndex { .. } => 0,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::expr::{AggregateFunction, BinaryOp, UnaryOp};

    #[test]
    fn test_plan_cache_basic() {
        let cache = PlanCache::new(10);

        let stmt = PreparedStatement::new(
            "SELECT 1".to_string(),
            LogicalPlan::EmptyRelation,
            vec![],
            0,
            1,
        );

        cache.insert("SELECT 1".to_string(), stmt.clone());

        let retrieved = cache.get("SELECT 1");
        assert!(retrieved.is_some());
        assert_eq!(retrieved.unwrap().sql(), "SELECT 1");
    }

    #[test]
    fn test_plan_cache_eviction() {
        let cache = PlanCache::new(2);

        for i in 0..3 {
            let sql = format!("SELECT {}", i);
            let stmt = PreparedStatement::new(
                sql.clone(),
                LogicalPlan::EmptyRelation,
                vec![],
                0,
                i as u64,
            );
            cache.insert(sql, stmt);
        }

        // Should have evicted the oldest entry
        assert_eq!(cache.len(), 2);
        assert_eq!(cache.stats().evictions(), 1);
    }

    #[test]
    fn test_plan_cache_invalidation() {
        let cache = PlanCache::new(10);

        cache.insert(
            "SELECT * FROM users".to_string(),
            PreparedStatement::new(
                "SELECT * FROM users".to_string(),
                LogicalPlan::EmptyRelation,
                vec![],
                0,
                1,
            ),
        );

        cache.insert(
            "SELECT * FROM posts".to_string(),
            PreparedStatement::new(
                "SELECT * FROM posts".to_string(),
                LogicalPlan::EmptyRelation,
                vec![],
                0,
                2,
            ),
        );

        assert_eq!(cache.len(), 2);

        // Invalidate plans for users table
        cache.invalidate_table("users");

        assert_eq!(cache.len(), 1);
        assert!(cache.get("SELECT * FROM users").is_none());
        assert!(cache.get("SELECT * FROM posts").is_some());
    }

    #[test]
    fn test_substitute_params() {
        let expr = Expr::BinaryOp {
            left: Box::new(Expr::Column {
                table: None,
                name: "id".to_string(),
                index: None,
            }),
            op: BinaryOp::Eq,
            right: Box::new(Expr::Parameter { index: 0 }),
        };

        let params = vec![Value::Integer(42)];
        let substituted = substitute_params(&expr, &params).unwrap();

        if let Expr::BinaryOp { right, .. } = substituted {
            assert!(matches!(*right, Expr::Literal(Value::Integer(42))));
        } else {
            panic!("Expected BinaryOp");
        }
    }

    #[test]
    fn test_count_params() {
        let expr = Expr::BinaryOp {
            left: Box::new(Expr::Parameter { index: 0 }),
            op: BinaryOp::And,
            right: Box::new(Expr::BinaryOp {
                left: Box::new(Expr::Column {
                    table: None,
                    name: "x".to_string(),
                    index: None,
                }),
                op: BinaryOp::Eq,
                right: Box::new(Expr::Parameter { index: 1 }),
            }),
        };

        assert_eq!(count_params(&expr), 2);
    }

    #[test]
    fn test_validate_params() {
        let stmt = PreparedStatement::new(
            "SELECT * FROM users WHERE id = ?".to_string(),
            LogicalPlan::EmptyRelation,
            vec![Some(ColumnType::Integer)],
            1,
            1,
        );

        // Valid params
        assert!(stmt.validate_params(&[Value::Integer(1)]).is_ok());

        // Wrong count
        assert!(stmt.validate_params(&[]).is_err());
        assert!(stmt
            .validate_params(&[Value::Integer(1), Value::Integer(2)])
            .is_err());

        // Wrong type (if type checking is enabled)
        assert!(stmt
            .validate_params(&[Value::Text("hello".to_string())])
            .is_err());
    }

    // ========== substitute_params for all expr types ==========

    #[test]
    fn test_substitute_unary_op() {
        // NOT $1
        let expr = Expr::UnaryOp {
            op: UnaryOp::Not,
            expr: Box::new(Expr::Parameter { index: 0 }),
        };

        let params = vec![Value::Boolean(true)];
        let substituted = substitute_params(&expr, &params).unwrap();

        match substituted {
            Expr::UnaryOp {
                op: UnaryOp::Not,
                expr,
            } => {
                assert!(matches!(*expr, Expr::Literal(Value::Boolean(true))));
            }
            _ => panic!("Expected UnaryOp"),
        }
    }

    #[test]
    fn test_substitute_function_params() {
        // UPPER($1)
        let expr = Expr::Function {
            name: "UPPER".to_string(),
            args: vec![Expr::Parameter { index: 0 }],
        };

        let params = vec![Value::Text("hello".to_string())];
        let substituted = substitute_params(&expr, &params).unwrap();

        match substituted {
            Expr::Function { name, args } => {
                assert_eq!(name, "UPPER");
                assert!(matches!(args[0], Expr::Literal(Value::Text(_))));
            }
            _ => panic!("Expected Function"),
        }
    }

    #[test]
    fn test_substitute_aggregate_params() {
        // SUM($1)
        let expr = Expr::Aggregate {
            func: AggregateFunction::Sum,
            arg: Some(Box::new(Expr::Parameter { index: 0 })),
            distinct: false,
        };

        let params = vec![Value::Integer(100)];
        let substituted = substitute_params(&expr, &params).unwrap();

        match substituted {
            Expr::Aggregate {
                func,
                arg,
                distinct,
            } => {
                assert_eq!(func, AggregateFunction::Sum);
                assert!(!distinct);
                assert!(matches!(*arg.unwrap(), Expr::Literal(Value::Integer(100))));
            }
            _ => panic!("Expected Aggregate"),
        }
    }

    #[test]
    fn test_substitute_case_expr() {
        // CASE WHEN $1 THEN $2 ELSE $3 END
        let expr = Expr::Case {
            operand: None,
            when_clauses: vec![(Expr::Parameter { index: 0 }, Expr::Parameter { index: 1 })],
            else_result: Some(Box::new(Expr::Parameter { index: 2 })),
        };

        let params = vec![Value::Boolean(true), Value::Integer(1), Value::Integer(0)];
        let substituted = substitute_params(&expr, &params).unwrap();

        match substituted {
            Expr::Case {
                when_clauses,
                else_result,
                ..
            } => {
                assert!(matches!(
                    when_clauses[0].0,
                    Expr::Literal(Value::Boolean(true))
                ));
                assert!(matches!(
                    when_clauses[0].1,
                    Expr::Literal(Value::Integer(1))
                ));
                assert!(matches!(
                    *else_result.unwrap(),
                    Expr::Literal(Value::Integer(0))
                ));
            }
            _ => panic!("Expected Case"),
        }
    }

    #[test]
    fn test_substitute_case_with_operand() {
        // CASE $1 WHEN 'active' THEN 1 END
        let expr = Expr::Case {
            operand: Some(Box::new(Expr::Parameter { index: 0 })),
            when_clauses: vec![(
                Expr::Literal(Value::Text("active".to_string())),
                Expr::Literal(Value::Integer(1)),
            )],
            else_result: None,
        };

        let params = vec![Value::Text("active".to_string())];
        let substituted = substitute_params(&expr, &params).unwrap();

        match substituted {
            Expr::Case { operand, .. } => {
                assert!(matches!(*operand.unwrap(), Expr::Literal(Value::Text(_))));
            }
            _ => panic!("Expected Case"),
        }
    }

    #[test]
    fn test_substitute_between_expr() {
        // x BETWEEN $1 AND $2
        let expr = Expr::Between {
            expr: Box::new(Expr::Column {
                table: None,
                name: "x".to_string(),
                index: None,
            }),
            low: Box::new(Expr::Parameter { index: 0 }),
            high: Box::new(Expr::Parameter { index: 1 }),
            negated: false,
        };

        let params = vec![Value::Integer(1), Value::Integer(10)];
        let substituted = substitute_params(&expr, &params).unwrap();

        match substituted {
            Expr::Between { low, high, .. } => {
                assert!(matches!(*low, Expr::Literal(Value::Integer(1))));
                assert!(matches!(*high, Expr::Literal(Value::Integer(10))));
            }
            _ => panic!("Expected Between"),
        }
    }

    #[test]
    fn test_substitute_in_list() {
        // x IN ($1, $2, $3)
        let expr = Expr::InList {
            expr: Box::new(Expr::Column {
                table: None,
                name: "x".to_string(),
                index: None,
            }),
            list: vec![
                Expr::Parameter { index: 0 },
                Expr::Parameter { index: 1 },
                Expr::Parameter { index: 2 },
            ],
            negated: false,
        };

        let params = vec![Value::Integer(1), Value::Integer(2), Value::Integer(3)];
        let substituted = substitute_params(&expr, &params).unwrap();

        match substituted {
            Expr::InList { list, .. } => {
                assert!(matches!(list[0], Expr::Literal(Value::Integer(1))));
                assert!(matches!(list[1], Expr::Literal(Value::Integer(2))));
                assert!(matches!(list[2], Expr::Literal(Value::Integer(3))));
            }
            _ => panic!("Expected InList"),
        }
    }

    #[test]
    fn test_substitute_like_expr() {
        // $1 LIKE 'pattern'
        let expr = Expr::Like {
            expr: Box::new(Expr::Parameter { index: 0 }),
            pattern: "%test%".to_string(),
            negated: false,
        };

        let params = vec![Value::Text("this is a test".to_string())];
        let substituted = substitute_params(&expr, &params).unwrap();

        match substituted {
            Expr::Like {
                expr,
                pattern,
                negated,
            } => {
                assert!(matches!(*expr, Expr::Literal(Value::Text(_))));
                assert_eq!(pattern, "%test%");
                assert!(!negated);
            }
            _ => panic!("Expected Like"),
        }
    }

    #[test]
    fn test_substitute_nested_expressions() {
        // ($1 + $2) * ($3 - $4)
        let expr = Expr::BinaryOp {
            left: Box::new(Expr::BinaryOp {
                left: Box::new(Expr::Parameter { index: 0 }),
                op: BinaryOp::Add,
                right: Box::new(Expr::Parameter { index: 1 }),
            }),
            op: BinaryOp::Mul,
            right: Box::new(Expr::BinaryOp {
                left: Box::new(Expr::Parameter { index: 2 }),
                op: BinaryOp::Sub,
                right: Box::new(Expr::Parameter { index: 3 }),
            }),
        };

        let params = vec![
            Value::Integer(1),
            Value::Integer(2),
            Value::Integer(10),
            Value::Integer(3),
        ];
        let substituted = substitute_params(&expr, &params).unwrap();

        // Verify structure is preserved
        match substituted {
            Expr::BinaryOp {
                op: BinaryOp::Mul, ..
            } => {}
            _ => panic!("Expected BinaryOp with Mul"),
        }
    }

    #[test]
    fn test_substitute_params_missing_param() {
        let expr = Expr::Parameter { index: 5 };
        let params = vec![Value::Integer(1)]; // Only 1 param, but index 5 requested

        let result = substitute_params(&expr, &params);
        assert!(result.is_err());
    }

    #[test]
    fn test_substitute_column_literal_unchanged() {
        let expr = Expr::Column {
            table: None,
            name: "x".to_string(),
            index: None,
        };
        let params = vec![];
        let substituted = substitute_params(&expr, &params).unwrap();
        assert!(matches!(substituted, Expr::Column { .. }));

        let expr = Expr::Literal(Value::Integer(42));
        let substituted = substitute_params(&expr, &params).unwrap();
        assert!(matches!(substituted, Expr::Literal(Value::Integer(42))));
    }

    // ========== substitute_plan_params tests ==========

    #[test]
    fn test_substitute_plan_filter() {
        let plan = LogicalPlan::Filter {
            input: Box::new(LogicalPlan::EmptyRelation),
            predicate: Expr::BinaryOp {
                left: Box::new(Expr::Column {
                    table: None,
                    name: "x".to_string(),
                    index: None,
                }),
                op: BinaryOp::Eq,
                right: Box::new(Expr::Parameter { index: 0 }),
            },
        };

        let params = vec![Value::Integer(42)];
        let substituted = substitute_plan_params(&plan, &params).unwrap();

        match substituted {
            LogicalPlan::Filter { predicate, .. } => match predicate {
                Expr::BinaryOp { right, .. } => {
                    assert!(matches!(*right, Expr::Literal(Value::Integer(42))));
                }
                _ => panic!("Expected BinaryOp predicate"),
            },
            _ => panic!("Expected Filter"),
        }
    }

    // Note: LogicalPlan::Scan requires Arc<Table> which is complex to construct in tests.
    // We test the substitution logic through Filter which wraps simpler plans.

    #[test]
    fn test_substitute_plan_project() {
        let plan = LogicalPlan::Project {
            input: Box::new(LogicalPlan::EmptyRelation),
            exprs: vec![(Expr::Parameter { index: 0 }, "col1".to_string())],
        };

        let params = vec![Value::Integer(100)];
        let substituted = substitute_plan_params(&plan, &params).unwrap();

        match substituted {
            LogicalPlan::Project { exprs, .. } => {
                assert!(matches!(exprs[0].0, Expr::Literal(Value::Integer(100))));
            }
            _ => panic!("Expected Project"),
        }
    }

    #[test]
    fn test_substitute_plan_join() {
        let plan = LogicalPlan::Join {
            left: Box::new(LogicalPlan::EmptyRelation),
            right: Box::new(LogicalPlan::EmptyRelation),
            join_type: crate::planner::JoinType::Inner,
            condition: Some(Expr::BinaryOp {
                left: Box::new(Expr::Column {
                    table: None,
                    name: "a".to_string(),
                    index: None,
                }),
                op: BinaryOp::Eq,
                right: Box::new(Expr::Parameter { index: 0 }),
            }),
        };

        let params = vec![Value::Integer(1)];
        let substituted = substitute_plan_params(&plan, &params).unwrap();

        match substituted {
            LogicalPlan::Join {
                condition: Some(c), ..
            } => match c {
                Expr::BinaryOp { right, .. } => {
                    assert!(matches!(*right, Expr::Literal(Value::Integer(1))));
                }
                _ => panic!("Expected BinaryOp"),
            },
            _ => panic!("Expected Join"),
        }
    }

    // Note: LogicalPlan::Insert/Update/Delete require Arc<Table> which is complex to construct.
    // These operations are tested through integration tests instead.

    #[test]
    fn test_substitute_plan_aggregate() {
        let plan = LogicalPlan::Aggregate {
            input: Box::new(LogicalPlan::EmptyRelation),
            group_by: vec![Expr::Parameter { index: 0 }],
            aggregates: vec![(Expr::Parameter { index: 1 }, "agg".to_string())],
        };

        let params = vec![Value::Integer(1), Value::Integer(2)];
        let substituted = substitute_plan_params(&plan, &params).unwrap();

        match substituted {
            LogicalPlan::Aggregate {
                group_by,
                aggregates,
                ..
            } => {
                assert!(matches!(group_by[0], Expr::Literal(Value::Integer(1))));
                assert!(matches!(aggregates[0].0, Expr::Literal(Value::Integer(2))));
            }
            _ => panic!("Expected Aggregate"),
        }
    }

    #[test]
    fn test_substitute_plan_sort() {
        use crate::planner::SortOrder;

        let plan = LogicalPlan::Sort {
            input: Box::new(LogicalPlan::EmptyRelation),
            order_by: vec![(Expr::Parameter { index: 0 }, SortOrder::Asc)],
        };

        let params = vec![Value::Integer(1)];
        let substituted = substitute_plan_params(&plan, &params).unwrap();

        match substituted {
            LogicalPlan::Sort { order_by, .. } => {
                assert!(matches!(order_by[0].0, Expr::Literal(Value::Integer(1))));
                assert_eq!(order_by[0].1, SortOrder::Asc);
            }
            _ => panic!("Expected Sort"),
        }
    }

    #[test]
    fn test_substitute_plan_limit_distinct() {
        let plan = LogicalPlan::Limit {
            input: Box::new(LogicalPlan::Distinct {
                input: Box::new(LogicalPlan::EmptyRelation),
            }),
            limit: Some(10),
            offset: 5,
        };

        let params = vec![];
        let substituted = substitute_plan_params(&plan, &params).unwrap();

        match substituted {
            LogicalPlan::Limit {
                input,
                limit,
                offset,
            } => {
                assert!(matches!(*input, LogicalPlan::Distinct { .. }));
                assert_eq!(limit, Some(10));
                assert_eq!(offset, 5);
            }
            _ => panic!("Expected Limit"),
        }
    }

    // ========== count_plan_params tests ==========

    #[test]
    fn test_count_params_in_plan_filter() {
        let plan = LogicalPlan::Filter {
            input: Box::new(LogicalPlan::EmptyRelation),
            predicate: Expr::BinaryOp {
                left: Box::new(Expr::Parameter { index: 0 }),
                op: BinaryOp::And,
                right: Box::new(Expr::Parameter { index: 2 }), // Gap in indices
            },
        };

        assert_eq!(count_plan_params(&plan), 3); // Max index + 1
    }

    // Note: count_plan_params for Insert is tested via integration tests
    // since LogicalPlan::Insert requires Arc<Table>.

    #[test]
    fn test_count_params_nested_plans() {
        let plan = LogicalPlan::Filter {
            input: Box::new(LogicalPlan::Project {
                input: Box::new(LogicalPlan::EmptyRelation),
                exprs: vec![(Expr::Parameter { index: 1 }, "col".to_string())],
            }),
            predicate: Expr::Parameter { index: 0 },
        };

        // Max should be 2 (index 1 + 1)
        assert_eq!(count_plan_params(&plan), 2);
    }

    // ========== PreparedStatement methods ==========

    #[test]
    fn test_prepared_statement_accessors() {
        let stmt = PreparedStatement::new(
            "SELECT * FROM users WHERE id = ?".to_string(),
            LogicalPlan::EmptyRelation,
            vec![Some(ColumnType::Integer)],
            1,
            42,
        );

        assert_eq!(stmt.sql(), "SELECT * FROM users WHERE id = ?");
        assert!(matches!(stmt.plan(), LogicalPlan::EmptyRelation));
        assert_eq!(stmt.param_types().len(), 1);
        assert_eq!(stmt.param_count(), 1);
        assert_eq!(stmt.id(), 42);
    }

    #[test]
    fn test_validate_params_with_nulls() {
        let stmt = PreparedStatement::new(
            "SELECT * FROM users WHERE id = ?".to_string(),
            LogicalPlan::EmptyRelation,
            vec![Some(ColumnType::Integer)],
            1,
            1,
        );

        // NULL should be accepted for any type
        assert!(stmt.validate_params(&[Value::Null]).is_ok());
    }

    #[test]
    fn test_validate_params_no_type_check() {
        let stmt = PreparedStatement::new(
            "SELECT * FROM users WHERE id = ?".to_string(),
            LogicalPlan::EmptyRelation,
            vec![None], // No type specified
            1,
            1,
        );

        // Any type should be accepted when no type constraint
        assert!(stmt.validate_params(&[Value::Integer(1)]).is_ok());
        assert!(stmt
            .validate_params(&[Value::Text("hello".to_string())])
            .is_ok());
    }

    // ========== PlanCache edge cases ==========

    #[test]
    fn test_cache_invalidate_all() {
        let cache = PlanCache::new(10);

        for i in 0..5 {
            let sql = format!("SELECT {}", i);
            let stmt = PreparedStatement::new(
                sql.clone(),
                LogicalPlan::EmptyRelation,
                vec![],
                0,
                i as u64,
            );
            cache.insert(sql, stmt);
        }

        assert_eq!(cache.len(), 5);

        cache.invalidate_all();

        assert_eq!(cache.len(), 0);
        assert!(cache.is_empty());
        assert_eq!(cache.stats().invalidations(), 5);
    }

    #[test]
    fn test_cache_stats_accuracy() {
        let cache = PlanCache::new(10);

        // Initial state
        assert_eq!(cache.stats().hits(), 0);
        assert_eq!(cache.stats().misses(), 0);
        assert_eq!(cache.stats().evictions(), 0);
        assert_eq!(cache.stats().invalidations(), 0);
        assert_eq!(cache.stats().hit_ratio(), 1.0); // No accesses yet = 100%

        // Miss
        assert!(cache.get("SELECT 1").is_none());
        assert_eq!(cache.stats().misses(), 1);

        // Insert and hit
        cache.insert(
            "SELECT 1".to_string(),
            PreparedStatement::new(
                "SELECT 1".to_string(),
                LogicalPlan::EmptyRelation,
                vec![],
                0,
                1,
            ),
        );
        assert!(cache.get("SELECT 1").is_some());
        assert_eq!(cache.stats().hits(), 1);

        // Hit ratio: 1 hit / (1 hit + 1 miss) = 0.5
        assert!((cache.stats().hit_ratio() - 0.5).abs() < 0.001);
    }

    #[test]
    fn test_cache_next_id() {
        let cache = PlanCache::new(10);

        let id1 = cache.next_id();
        let id2 = cache.next_id();
        let id3 = cache.next_id();

        assert_eq!(id1, 1);
        assert_eq!(id2, 2);
        assert_eq!(id3, 3);
    }

    #[test]
    fn test_cache_schema_version() {
        let cache = PlanCache::new(10);

        let v1 = cache.schema_version();
        cache.invalidate_all();
        let v2 = cache.schema_version();

        assert!(v2 > v1);
    }

    #[test]
    fn test_cache_max_size() {
        let cache = PlanCache::new(100);
        assert_eq!(cache.max_size(), 100);
    }

    #[test]
    fn test_cache_default_size() {
        let cache = PlanCache::default_size();
        assert_eq!(cache.max_size(), 1000);
    }

    #[test]
    fn test_cache_default_trait() {
        let cache = PlanCache::default();
        assert_eq!(cache.max_size(), 1000);
    }

    // ========== count_params for edge cases ==========

    #[test]
    fn test_count_params_case_expr() {
        let expr = Expr::Case {
            operand: Some(Box::new(Expr::Parameter { index: 0 })),
            when_clauses: vec![(Expr::Parameter { index: 1 }, Expr::Parameter { index: 2 })],
            else_result: Some(Box::new(Expr::Parameter { index: 3 })),
        };

        assert_eq!(count_params(&expr), 4);
    }

    #[test]
    fn test_count_params_between() {
        let expr = Expr::Between {
            expr: Box::new(Expr::Parameter { index: 0 }),
            low: Box::new(Expr::Parameter { index: 1 }),
            high: Box::new(Expr::Parameter { index: 2 }),
            negated: false,
        };

        assert_eq!(count_params(&expr), 3);
    }

    #[test]
    fn test_count_params_in_list() {
        let expr = Expr::InList {
            expr: Box::new(Expr::Column {
                table: None,
                name: "x".to_string(),
                index: None,
            }),
            list: vec![
                Expr::Parameter { index: 0 },
                Expr::Parameter { index: 3 }, // Gap
            ],
            negated: false,
        };

        assert_eq!(count_params(&expr), 4); // Max index + 1
    }

    #[test]
    fn test_count_params_aggregate() {
        let expr = Expr::Aggregate {
            func: AggregateFunction::Sum,
            arg: Some(Box::new(Expr::Parameter { index: 2 })),
            distinct: false,
        };

        assert_eq!(count_params(&expr), 3);
    }

    #[test]
    fn test_count_params_function() {
        let expr = Expr::Function {
            name: "CONCAT".to_string(),
            args: vec![Expr::Parameter { index: 0 }, Expr::Parameter { index: 1 }],
        };

        assert_eq!(count_params(&expr), 2);
    }

    #[test]
    fn test_count_params_like() {
        let expr = Expr::Like {
            expr: Box::new(Expr::Parameter { index: 0 }),
            pattern: "%test%".to_string(),
            negated: false,
        };

        assert_eq!(count_params(&expr), 1);
    }

    #[test]
    fn test_count_params_no_params() {
        let expr = Expr::Column {
            table: None,
            name: "x".to_string(),
            index: None,
        };
        assert_eq!(count_params(&expr), 0);

        let expr = Expr::Literal(Value::Integer(42));
        assert_eq!(count_params(&expr), 0);

        let expr = Expr::Wildcard;
        assert_eq!(count_params(&expr), 0);
    }
}