featherdb_query/expr/

mod.rs

//! Expression representation and evaluation

mod aggregate;
mod functions;
pub mod helpers;
mod ops;
mod subquery;
mod window;

// Re-export all public types
pub use aggregate::AggregateFunction;
pub use functions::eval_function;
pub use ops::{BinaryOp, UnaryOp};
pub use subquery::{SubqueryCompareOp, SubqueryType};
pub use window::{
    FrameBound, FrameUnit, WindowFrame, WindowFunction, WindowFunctionType, WindowOrderByExpr,
    WindowSortOrder,
};

use featherdb_core::{Error, Result, Value};
use std::collections::HashMap;

/// Expression tree
#[derive(Debug, Clone, PartialEq)]
pub enum Expr {
    /// Column reference
    Column {
        table: Option<String>,
        name: String,
        index: Option<usize>,
    },

    /// Literal value
    Literal(Value),

    /// Parameter placeholder for prepared statements
    /// Index is 0-based (? is index 0, $1 maps to index 0, $2 to index 1, etc.)
    Parameter { index: usize },

    /// Binary operation
    BinaryOp {
        left: Box<Expr>,
        op: BinaryOp,
        right: Box<Expr>,
    },

    /// Unary operation
    UnaryOp { op: UnaryOp, expr: Box<Expr> },

    /// Function call
    Function { name: String, args: Vec<Expr> },

    /// Aggregate function
    Aggregate {
        func: AggregateFunction,
        arg: Option<Box<Expr>>,
        distinct: bool,
    },

    /// CASE WHEN ... THEN ... ELSE ... END
    Case {
        operand: Option<Box<Expr>>,
        when_clauses: Vec<(Expr, Expr)>,
        else_result: Option<Box<Expr>>,
    },

    /// BETWEEN
    Between {
        expr: Box<Expr>,
        low: Box<Expr>,
        high: Box<Expr>,
        negated: bool,
    },

    /// IN (list)
    InList {
        expr: Box<Expr>,
        list: Vec<Expr>,
        negated: bool,
    },

    /// LIKE pattern
    Like {
        expr: Box<Expr>,
        pattern: String,
        negated: bool,
    },

    /// Wildcard (*) for SELECT *
    Wildcard,

    /// Qualified wildcard (table.*) for SELECT table.*
    QualifiedWildcard(String),

    /// Window function expression
    Window(WindowFunction),

    /// Scalar subquery - returns a single value
    /// e.g., SELECT name, (SELECT COUNT(*) FROM orders WHERE user_id = users.id)
    ScalarSubquery {
        /// Unique identifier for the subquery (for correlation tracking)
        subquery_id: usize,
        /// Whether this subquery is correlated (references outer query)
        correlated: bool,
    },

    /// EXISTS subquery
    /// e.g., WHERE EXISTS (SELECT 1 FROM orders WHERE user_id = users.id)
    Exists { subquery_id: usize, negated: bool },

    /// IN subquery
    /// e.g., WHERE id IN (SELECT user_id FROM orders)
    InSubquery {
        expr: Box<Expr>,
        subquery_id: usize,
        negated: bool,
    },

    /// ANY/SOME subquery comparison
    /// e.g., WHERE x > ANY (SELECT value FROM t)
    AnySubquery {
        expr: Box<Expr>,
        op: SubqueryCompareOp,
        subquery_id: usize,
    },

    /// ALL subquery comparison
    /// e.g., WHERE x > ALL (SELECT value FROM t)
    AllSubquery {
        expr: Box<Expr>,
        op: SubqueryCompareOp,
        subquery_id: usize,
    },
}

impl Expr {
    /// Create a column reference
    pub fn column(name: impl Into<String>) -> Self {
        Expr::Column {
            table: None,
            name: name.into(),
            index: None,
        }
    }

    /// Create a qualified column reference
    pub fn qualified_column(table: impl Into<String>, name: impl Into<String>) -> Self {
        Expr::Column {
            table: Some(table.into()),
            name: name.into(),
            index: None,
        }
    }

    /// Create a literal
    pub fn literal(value: impl Into<Value>) -> Self {
        Expr::Literal(value.into())
    }

    /// Create a binary operation
    pub fn binary(left: Expr, op: BinaryOp, right: Expr) -> Self {
        Expr::BinaryOp {
            left: Box::new(left),
            op,
            right: Box::new(right),
        }
    }

    /// Create an AND expression
    pub fn and(left: Expr, right: Expr) -> Self {
        Expr::binary(left, BinaryOp::And, right)
    }

    /// Create an OR expression
    pub fn or(left: Expr, right: Expr) -> Self {
        Expr::binary(left, BinaryOp::Or, right)
    }

    /// Create a parameter placeholder
    pub fn parameter(index: usize) -> Self {
        Expr::Parameter { index }
    }

    /// Pre-resolve column indices from a column map so that eval() can use
    /// direct index access (O(1)) instead of HashMap lookups per row.
    /// Call once before a row loop, then use the resolved expression for all rows.
    pub fn resolve_indices(&mut self, column_map: &HashMap<String, usize>) {
        match self {
            Expr::Column {
                name, table, index, ..
            } => {
                if index.is_some() {
                    return; // Already resolved
                }
                let resolved = if let Some(t) = table {
                    let key = format!("{}.{}", t, name);
                    column_map
                        .get(&key)
                        .or_else(|| column_map.get(name.as_str()))
                        .or_else(|| column_map.get(&key.to_ascii_lowercase()))
                        .or_else(|| column_map.get(&name.to_ascii_lowercase()))
                        .copied()
                } else {
                    column_map
                        .get(name.as_str())
                        .or_else(|| {
                            column_map
                                .iter()
                                .find(|(k, _)| {
                                    k.strip_suffix(name.as_str())
                                        .is_some_and(|prefix| prefix.ends_with('.'))
                                })
                                .map(|(_, v)| v)
                        })
                        .or_else(|| column_map.get(&name.to_ascii_lowercase()))
                        .copied()
                };
                *index = resolved;
            }
            Expr::BinaryOp { left, right, .. } => {
                left.resolve_indices(column_map);
                right.resolve_indices(column_map);
            }
            Expr::UnaryOp { expr, .. } => expr.resolve_indices(column_map),
            Expr::Function { args, .. } => {
                for arg in args {
                    arg.resolve_indices(column_map);
                }
            }
            Expr::Aggregate { arg: Some(e), .. } => e.resolve_indices(column_map),
            Expr::Case {
                operand,
                when_clauses,
                else_result,
            } => {
                if let Some(e) = operand {
                    e.resolve_indices(column_map);
                }
                for (w, t) in when_clauses {
                    w.resolve_indices(column_map);
                    t.resolve_indices(column_map);
                }
                if let Some(e) = else_result {
                    e.resolve_indices(column_map);
                }
            }
            Expr::Between {
                expr, low, high, ..
            } => {
                expr.resolve_indices(column_map);
                low.resolve_indices(column_map);
                high.resolve_indices(column_map);
            }
            Expr::InList { expr, list, .. } => {
                expr.resolve_indices(column_map);
                for e in list {
                    e.resolve_indices(column_map);
                }
            }
            Expr::Like { expr, .. } => expr.resolve_indices(column_map),
            Expr::InSubquery { expr, .. } => expr.resolve_indices(column_map),
            Expr::AnySubquery { expr, .. } => expr.resolve_indices(column_map),
            Expr::AllSubquery { expr, .. } => expr.resolve_indices(column_map),
            _ => {} // Literals, parameters, etc. — nothing to resolve
        }
    }

    /// Collect all resolved column indices referenced by this expression.
    pub fn referenced_columns(&self, out: &mut Vec<usize>) {
        match self {
            Expr::Column { index: Some(i), .. } => out.push(*i),
            Expr::BinaryOp { left, right, .. } => {
                left.referenced_columns(out);
                right.referenced_columns(out);
            }
            Expr::UnaryOp { expr, .. } => expr.referenced_columns(out),
            Expr::Function { args, .. } => {
                for arg in args {
                    arg.referenced_columns(out);
                }
            }
            Expr::Aggregate { arg: Some(e), .. } => e.referenced_columns(out),
            Expr::Case {
                operand,
                when_clauses,
                else_result,
            } => {
                if let Some(e) = operand {
                    e.referenced_columns(out);
                }
                for (w, t) in when_clauses {
                    w.referenced_columns(out);
                    t.referenced_columns(out);
                }
                if let Some(e) = else_result {
                    e.referenced_columns(out);
                }
            }
            Expr::Between {
                expr, low, high, ..
            } => {
                expr.referenced_columns(out);
                low.referenced_columns(out);
                high.referenced_columns(out);
            }
            Expr::InList { expr, list, .. } => {
                expr.referenced_columns(out);
                for e in list {
                    e.referenced_columns(out);
                }
            }
            Expr::Like { expr, .. } => expr.referenced_columns(out),
            Expr::InSubquery { expr, .. } => expr.referenced_columns(out),
            Expr::AnySubquery { expr, .. } => expr.referenced_columns(out),
            Expr::AllSubquery { expr, .. } => expr.referenced_columns(out),
            _ => {} // Literals, parameters, subqueries without expr, etc.
        }
    }

    /// Evaluate expression against a row
    pub fn eval(&self, row: &[Value], column_map: &HashMap<String, usize>) -> Result<Value> {
        match self {
            Expr::Column { name, index, table } => {
                if let Some(idx) = index {
                    Ok(row.get(*idx).cloned().unwrap_or(Value::Null))
                } else if let Some(t) = table {
                    // Qualified name: try "table.name" first
                    let key = format!("{}.{}", t, name);
                    let idx = column_map
                        .get(&key)
                        .or_else(|| column_map.get(name.as_str()))
                        .or_else(|| column_map.get(&key.to_ascii_lowercase()))
                        .or_else(|| column_map.get(&name.to_ascii_lowercase()));
                    match idx {
                        Some(&i) => Ok(row.get(i).cloned().unwrap_or(Value::Null)),
                        None => Err(Error::ColumnNotFound {
                            column: name.clone(),
                            table: t.clone(),
                            suggestion: None,
                        }),
                    }
                } else {
                    // Unqualified name (most common after predicate pushdown):
                    // Try direct O(1) lookup first, then suffix match for
                    // col_maps that only have qualified "table.col" entries
                    let idx = column_map
                        .get(name.as_str())
                        .or_else(|| {
                            column_map
                                .iter()
                                .find(|(k, _)| {
                                    k.strip_suffix(name.as_str())
                                        .is_some_and(|prefix| prefix.ends_with('.'))
                                })
                                .map(|(_, v)| v)
                        })
                        .or_else(|| {
                            let lower = name.to_ascii_lowercase();
                            column_map.get(lower.as_str()).or_else(|| {
                                column_map
                                    .iter()
                                    .find(|(k, _)| {
                                        k.to_ascii_lowercase()
                                            .strip_suffix(lower.as_str())
                                            .is_some_and(|prefix| prefix.ends_with('.'))
                                    })
                                    .map(|(_, v)| v)
                            })
                        });
                    match idx {
                        Some(&i) => Ok(row.get(i).cloned().unwrap_or(Value::Null)),
                        None => Err(Error::ColumnNotFound {
                            column: name.clone(),
                            table: String::new(),
                            suggestion: None,
                        }),
                    }
                }
            }

            Expr::Literal(v) => Ok(v.clone()),

            Expr::BinaryOp { left, op, right } => {
                let l = left.eval(row, column_map)?;
                let r = right.eval(row, column_map)?;
                op.eval(&l, &r)
            }

            Expr::UnaryOp { op, expr } => {
                let v = expr.eval(row, column_map)?;
                op.eval(&v)
            }

            Expr::Case {
                operand,
                when_clauses,
                else_result,
            } => {
                let operand_val = operand
                    .as_ref()
                    .map(|e| e.eval(row, column_map))
                    .transpose()?;

                for (when_expr, then_expr) in when_clauses {
                    let when_val = when_expr.eval(row, column_map)?;
                    let matches = if let Some(ref op_val) = operand_val {
                        when_val == *op_val
                    } else {
                        when_val.as_bool().unwrap_or(false)
                    };

                    if matches {
                        return then_expr.eval(row, column_map);
                    }
                }

                if let Some(else_expr) = else_result {
                    else_expr.eval(row, column_map)
                } else {
                    Ok(Value::Null)
                }
            }

            Expr::Between {
                expr,
                low,
                high,
                negated,
            } => {
                let v = expr.eval(row, column_map)?;
                let l = low.eval(row, column_map)?;
                let h = high.eval(row, column_map)?;

                let in_range = v >= l && v <= h;
                Ok(Value::Boolean(if *negated { !in_range } else { in_range }))
            }

            Expr::InList {
                expr,
                list,
                negated,
            } => {
                let v = expr.eval(row, column_map)?;
                let mut found = false;

                for item in list {
                    let item_val = item.eval(row, column_map)?;
                    if v == item_val {
                        found = true;
                        break;
                    }
                }

                Ok(Value::Boolean(if *negated { !found } else { found }))
            }

            Expr::Like {
                expr,
                pattern,
                negated,
            } => {
                let v = expr.eval(row, column_map)?;
                let matches = match v {
                    Value::Text(s) => helpers::like_match(&s, pattern),
                    _ => false,
                };
                Ok(Value::Boolean(if *negated { !matches } else { matches }))
            }

            Expr::Function { name, args } => {
                let arg_values: Vec<Value> = args
                    .iter()
                    .map(|a| a.eval(row, column_map))
                    .collect::<Result<_>>()?;

                eval_function(name, &arg_values)
            }

            Expr::Aggregate { .. } => {
                // Aggregates are handled specially by the executor
                Err(Error::Internal(
                    "Aggregate should be handled by executor".into(),
                ))
            }

            Expr::Parameter { index } => {
                // Parameters should be substituted before execution
                Err(Error::Internal(format!(
                    "Parameter ${} was not substituted before execution",
                    index + 1
                )))
            }

            Expr::Wildcard | Expr::QualifiedWildcard(_) => {
                Err(Error::Internal("Wildcard should be expanded".into()))
            }

            Expr::Window(_) => {
                // Window functions are handled specially by the executor
                Err(Error::Internal(
                    "Window function should be handled by executor".into(),
                ))
            }

            Expr::ScalarSubquery { .. } => {
                // Scalar subqueries are handled specially by the executor
                Err(Error::Internal(
                    "Scalar subquery should be handled by executor".into(),
                ))
            }

            Expr::Exists { .. } => {
                // EXISTS subqueries are handled specially by the executor
                Err(Error::Internal(
                    "EXISTS subquery should be handled by executor".into(),
                ))
            }

            Expr::InSubquery { .. } => {
                // IN subqueries are handled specially by the executor
                Err(Error::Internal(
                    "IN subquery should be handled by executor".into(),
                ))
            }

            Expr::AnySubquery { .. } => {
                // ANY subqueries are handled specially by the executor
                Err(Error::Internal(
                    "ANY subquery should be handled by executor".into(),
                ))
            }

            Expr::AllSubquery { .. } => {
                // ALL subqueries are handled specially by the executor
                Err(Error::Internal(
                    "ALL subquery should be handled by executor".into(),
                ))
            }
        }
    }

    /// Check if expression contains aggregates
    pub fn contains_aggregate(&self) -> bool {
        match self {
            Expr::Aggregate { .. } => true,
            Expr::BinaryOp { left, right, .. } => {
                left.contains_aggregate() || right.contains_aggregate()
            }
            Expr::UnaryOp { expr, .. } => expr.contains_aggregate(),
            Expr::Function { args, .. } => args.iter().any(|a| a.contains_aggregate()),
            Expr::Case {
                operand,
                when_clauses,
                else_result,
            } => {
                operand
                    .as_ref()
                    .map(|e| e.contains_aggregate())
                    .unwrap_or(false)
                    || when_clauses
                        .iter()
                        .any(|(w, t)| w.contains_aggregate() || t.contains_aggregate())
                    || else_result
                        .as_ref()
                        .map(|e| e.contains_aggregate())
                        .unwrap_or(false)
            }
            Expr::Between {
                expr, low, high, ..
            } => expr.contains_aggregate() || low.contains_aggregate() || high.contains_aggregate(),
            Expr::InList { expr, list, .. } => {
                expr.contains_aggregate() || list.iter().any(|e| e.contains_aggregate())
            }
            Expr::Like { expr, .. } => expr.contains_aggregate(),
            Expr::InSubquery { expr, .. } => expr.contains_aggregate(),
            Expr::AnySubquery { expr, .. } => expr.contains_aggregate(),
            Expr::AllSubquery { expr, .. } => expr.contains_aggregate(),
            _ => false,
        }
    }

    /// Collect all column references in this expression
    /// Returns tuples of (table, column_name)
    pub fn collect_column_references(&self) -> Vec<(Option<String>, String)> {
        let mut refs = Vec::new();
        self.collect_column_references_recursive(&mut refs);
        refs
    }

    fn collect_column_references_recursive(&self, refs: &mut Vec<(Option<String>, String)>) {
        match self {
            Expr::Column { table, name, .. } => {
                refs.push((table.clone(), name.clone()));
            }
            Expr::BinaryOp { left, right, .. } => {
                left.collect_column_references_recursive(refs);
                right.collect_column_references_recursive(refs);
            }
            Expr::UnaryOp { expr, .. } => {
                expr.collect_column_references_recursive(refs);
            }
            Expr::Function { args, .. } => {
                for arg in args {
                    arg.collect_column_references_recursive(refs);
                }
            }
            Expr::Aggregate { arg: Some(e), .. } => {
                e.collect_column_references_recursive(refs);
            }
            Expr::Aggregate { arg: None, .. } => {}
            Expr::Case {
                operand,
                when_clauses,
                else_result,
            } => {
                if let Some(e) = operand {
                    e.collect_column_references_recursive(refs);
                }
                for (when, then) in when_clauses {
                    when.collect_column_references_recursive(refs);
                    then.collect_column_references_recursive(refs);
                }
                if let Some(e) = else_result {
                    e.collect_column_references_recursive(refs);
                }
            }
            Expr::Between {
                expr, low, high, ..
            } => {
                expr.collect_column_references_recursive(refs);
                low.collect_column_references_recursive(refs);
                high.collect_column_references_recursive(refs);
            }
            Expr::InList { expr, list, .. } => {
                expr.collect_column_references_recursive(refs);
                for item in list {
                    item.collect_column_references_recursive(refs);
                }
            }
            Expr::Like { expr, .. } => {
                expr.collect_column_references_recursive(refs);
            }
            Expr::InSubquery { expr, .. } => {
                expr.collect_column_references_recursive(refs);
            }
            Expr::AnySubquery { expr, .. } => {
                expr.collect_column_references_recursive(refs);
            }
            Expr::AllSubquery { expr, .. } => {
                expr.collect_column_references_recursive(refs);
            }
            Expr::Window(window) => {
                for expr in &window.partition_by {
                    expr.collect_column_references_recursive(refs);
                }
                for order_expr in &window.order_by {
                    order_expr.expr.collect_column_references_recursive(refs);
                }
                // Also collect from window function arguments
                match &window.function {
                    WindowFunctionType::Lag { expr, default, .. }
                    | WindowFunctionType::Lead { expr, default, .. } => {
                        expr.collect_column_references_recursive(refs);
                        if let Some(d) = default {
                            d.collect_column_references_recursive(refs);
                        }
                    }
                    WindowFunctionType::FirstValue(e)
                    | WindowFunctionType::LastValue(e)
                    | WindowFunctionType::Sum(e)
                    | WindowFunctionType::Avg(e)
                    | WindowFunctionType::Min(e)
                    | WindowFunctionType::Max(e) => {
                        e.collect_column_references_recursive(refs);
                    }
                    WindowFunctionType::NthValue(e, _) => {
                        e.collect_column_references_recursive(refs);
                    }
                    WindowFunctionType::Count(Some(e)) => {
                        e.collect_column_references_recursive(refs);
                    }
                    _ => {}
                }
            }
            // Literals, parameters, wildcards don't reference columns
            _ => {}
        }
    }

    /// Check if expression contains window functions
    pub fn contains_window_function(&self) -> bool {
        match self {
            Expr::Window(_) => true,
            Expr::BinaryOp { left, right, .. } => {
                left.contains_window_function() || right.contains_window_function()
            }
            Expr::UnaryOp { expr, .. } => expr.contains_window_function(),
            Expr::Function { args, .. } => args.iter().any(|a| a.contains_window_function()),
            Expr::Case {
                operand,
                when_clauses,
                else_result,
            } => {
                operand
                    .as_ref()
                    .map(|e| e.contains_window_function())
                    .unwrap_or(false)
                    || when_clauses
                        .iter()
                        .any(|(w, t)| w.contains_window_function() || t.contains_window_function())
                    || else_result
                        .as_ref()
                        .map(|e| e.contains_window_function())
                        .unwrap_or(false)
            }
            Expr::Between {
                expr, low, high, ..
            } => {
                expr.contains_window_function()
                    || low.contains_window_function()
                    || high.contains_window_function()
            }
            Expr::InList { expr, list, .. } => {
                expr.contains_window_function() || list.iter().any(|e| e.contains_window_function())
            }
            Expr::Like { expr, .. } => expr.contains_window_function(),
            Expr::InSubquery { expr, .. } => expr.contains_window_function(),
            Expr::AnySubquery { expr, .. } => expr.contains_window_function(),
            Expr::AllSubquery { expr, .. } => expr.contains_window_function(),
            _ => false,
        }
    }

    /// Check if expression contains subqueries
    pub fn contains_subquery(&self) -> bool {
        match self {
            Expr::ScalarSubquery { .. }
            | Expr::Exists { .. }
            | Expr::InSubquery { .. }
            | Expr::AnySubquery { .. }
            | Expr::AllSubquery { .. } => true,
            Expr::BinaryOp { left, right, .. } => {
                left.contains_subquery() || right.contains_subquery()
            }
            Expr::UnaryOp { expr, .. } => expr.contains_subquery(),
            Expr::Function { args, .. } => args.iter().any(|a| a.contains_subquery()),
            Expr::Case {
                operand,
                when_clauses,
                else_result,
            } => {
                operand
                    .as_ref()
                    .map(|e| e.contains_subquery())
                    .unwrap_or(false)
                    || when_clauses
                        .iter()
                        .any(|(w, t)| w.contains_subquery() || t.contains_subquery())
                    || else_result
                        .as_ref()
                        .map(|e| e.contains_subquery())
                        .unwrap_or(false)
            }
            Expr::Between {
                expr, low, high, ..
            } => expr.contains_subquery() || low.contains_subquery() || high.contains_subquery(),
            Expr::InList { expr, list, .. } => {
                expr.contains_subquery() || list.iter().any(|e| e.contains_subquery())
            }
            Expr::Like { expr, .. } => expr.contains_subquery(),
            _ => false,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_expr_column() {
        let expr = Expr::column("name");
        assert!(matches!(expr, Expr::Column { .. }));
    }

    #[test]
    fn test_expr_qualified_column() {
        let expr = Expr::qualified_column("users", "name");
        match expr {
            Expr::Column { table, name, .. } => {
                assert_eq!(table, Some("users".to_string()));
                assert_eq!(name, "name");
            }
            _ => panic!("Expected Column variant"),
        }
    }

    #[test]
    fn test_expr_literal() {
        let expr = Expr::literal(Value::Integer(42));
        assert_eq!(expr, Expr::Literal(Value::Integer(42)));
    }

    #[test]
    fn test_expr_binary() {
        let expr = Expr::binary(
            Expr::literal(Value::Integer(5)),
            BinaryOp::Add,
            Expr::literal(Value::Integer(3)),
        );
        assert!(matches!(expr, Expr::BinaryOp { .. }));
    }

    #[test]
    fn test_expr_and() {
        let expr = Expr::and(
            Expr::literal(Value::Boolean(true)),
            Expr::literal(Value::Boolean(false)),
        );
        match expr {
            Expr::BinaryOp { op, .. } => assert_eq!(op, BinaryOp::And),
            _ => panic!("Expected BinaryOp"),
        }
    }

    #[test]
    fn test_expr_or() {
        let expr = Expr::or(
            Expr::literal(Value::Boolean(true)),
            Expr::literal(Value::Boolean(false)),
        );
        match expr {
            Expr::BinaryOp { op, .. } => assert_eq!(op, BinaryOp::Or),
            _ => panic!("Expected BinaryOp"),
        }
    }

    #[test]
    fn test_expr_parameter() {
        let expr = Expr::parameter(0);
        assert_eq!(expr, Expr::Parameter { index: 0 });
    }

    #[test]
    fn test_expr_eval_literal() {
        let expr = Expr::literal(Value::Integer(42));
        let row = vec![];
        let column_map = HashMap::new();
        let result = expr.eval(&row, &column_map).unwrap();
        assert_eq!(result, Value::Integer(42));
    }

    #[test]
    fn test_expr_eval_binary_op() {
        let expr = Expr::binary(
            Expr::literal(Value::Integer(5)),
            BinaryOp::Add,
            Expr::literal(Value::Integer(3)),
        );
        let row = vec![];
        let column_map = HashMap::new();
        let result = expr.eval(&row, &column_map).unwrap();
        assert_eq!(result, Value::Integer(8));
    }

    #[test]
    fn test_contains_aggregate() {
        let agg_expr = Expr::Aggregate {
            func: AggregateFunction::Count,
            arg: None,
            distinct: false,
        };
        assert!(agg_expr.contains_aggregate());

        let non_agg = Expr::literal(Value::Integer(42));
        assert!(!non_agg.contains_aggregate());
    }

    #[test]
    fn test_contains_window_function() {
        let window_expr = Expr::Window(WindowFunction {
            function: WindowFunctionType::RowNumber,
            partition_by: vec![],
            order_by: vec![],
            frame: None,
        });
        assert!(window_expr.contains_window_function());

        let non_window = Expr::literal(Value::Integer(42));
        assert!(!non_window.contains_window_function());
    }

    #[test]
    fn test_contains_subquery() {
        let subquery_expr = Expr::ScalarSubquery {
            subquery_id: 0,
            correlated: false,
        };
        assert!(subquery_expr.contains_subquery());

        let non_subquery = Expr::literal(Value::Integer(42));
        assert!(!non_subquery.contains_subquery());
    }

    #[test]
    fn test_collect_column_references() {
        let expr = Expr::BinaryOp {
            left: Box::new(Expr::Column {
                table: Some("users".to_string()),
                name: "age".to_string(),
                index: None,
            }),
            op: BinaryOp::Gt,
            right: Box::new(Expr::Column {
                table: None,
                name: "salary".to_string(),
                index: None,
            }),
        };

        let refs = expr.collect_column_references();
        assert_eq!(refs.len(), 2);
        assert!(refs.contains(&(Some("users".to_string()), "age".to_string())));
        assert!(refs.contains(&(None, "salary".to_string())));
    }

    // =============================================================================
    // Additional eval() tests
    // =============================================================================

    #[test]
    fn test_eval_column_lookup() {
        let row = vec![
            Value::Integer(1),
            Value::Text("Alice".into()),
            Value::Integer(30),
        ];
        let col_map: HashMap<String, usize> = [("id", 0), ("name", 1), ("age", 2)]
            .iter()
            .map(|(k, v)| (k.to_string(), *v))
            .collect();

        let expr = Expr::column("name");
        let result = expr.eval(&row, &col_map).unwrap();
        assert_eq!(result, Value::Text("Alice".into()));

        let expr = Expr::column("age");
        let result = expr.eval(&row, &col_map).unwrap();
        assert_eq!(result, Value::Integer(30));
    }

    #[test]
    fn test_eval_qualified_column() {
        let row = vec![Value::Integer(100)];
        let mut col_map = HashMap::new();
        col_map.insert("users.salary".to_string(), 0);

        let expr = Expr::qualified_column("users", "salary");
        let result = expr.eval(&row, &col_map).unwrap();
        assert_eq!(result, Value::Integer(100));
    }

    #[test]
    fn test_eval_column_not_found() {
        let row = vec![Value::Integer(1)];
        let col_map = HashMap::new();

        let expr = Expr::column("missing");
        let result = expr.eval(&row, &col_map);
        assert!(result.is_err());
        assert!(matches!(result.unwrap_err(), Error::ColumnNotFound { .. }));
    }

    #[test]
    fn test_eval_between_inclusive() {
        let row = vec![];
        let col_map = HashMap::new();

        // Test value = low boundary
        let expr = Expr::Between {
            expr: Box::new(Expr::literal(Value::Integer(5))),
            low: Box::new(Expr::literal(Value::Integer(5))),
            high: Box::new(Expr::literal(Value::Integer(10))),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));

        // Test value = high boundary
        let expr = Expr::Between {
            expr: Box::new(Expr::literal(Value::Integer(10))),
            low: Box::new(Expr::literal(Value::Integer(5))),
            high: Box::new(Expr::literal(Value::Integer(10))),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));

        // Test value in range
        let expr = Expr::Between {
            expr: Box::new(Expr::literal(Value::Integer(7))),
            low: Box::new(Expr::literal(Value::Integer(5))),
            high: Box::new(Expr::literal(Value::Integer(10))),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));

        // Test value out of range (below)
        let expr = Expr::Between {
            expr: Box::new(Expr::literal(Value::Integer(3))),
            low: Box::new(Expr::literal(Value::Integer(5))),
            high: Box::new(Expr::literal(Value::Integer(10))),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(false));

        // Test value out of range (above)
        let expr = Expr::Between {
            expr: Box::new(Expr::literal(Value::Integer(15))),
            low: Box::new(Expr::literal(Value::Integer(5))),
            high: Box::new(Expr::literal(Value::Integer(10))),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(false));
    }

    #[test]
    fn test_eval_between_negated() {
        let row = vec![];
        let col_map = HashMap::new();

        // NOT BETWEEN - value in range should return false
        let expr = Expr::Between {
            expr: Box::new(Expr::literal(Value::Integer(7))),
            low: Box::new(Expr::literal(Value::Integer(5))),
            high: Box::new(Expr::literal(Value::Integer(10))),
            negated: true,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(false));

        // NOT BETWEEN - value out of range should return true
        let expr = Expr::Between {
            expr: Box::new(Expr::literal(Value::Integer(15))),
            low: Box::new(Expr::literal(Value::Integer(5))),
            high: Box::new(Expr::literal(Value::Integer(10))),
            negated: true,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));
    }

    #[test]
    fn test_eval_in_list_found() {
        let row = vec![];
        let col_map = HashMap::new();

        let expr = Expr::InList {
            expr: Box::new(Expr::literal(Value::Integer(5))),
            list: vec![
                Expr::literal(Value::Integer(1)),
                Expr::literal(Value::Integer(5)),
                Expr::literal(Value::Integer(10)),
            ],
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));
    }

    #[test]
    fn test_eval_in_list_not_found() {
        let row = vec![];
        let col_map = HashMap::new();

        let expr = Expr::InList {
            expr: Box::new(Expr::literal(Value::Integer(7))),
            list: vec![
                Expr::literal(Value::Integer(1)),
                Expr::literal(Value::Integer(5)),
                Expr::literal(Value::Integer(10)),
            ],
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(false));
    }

    #[test]
    fn test_eval_in_list_negated() {
        let row = vec![];
        let col_map = HashMap::new();

        // NOT IN - value not in list should return true
        let expr = Expr::InList {
            expr: Box::new(Expr::literal(Value::Integer(7))),
            list: vec![
                Expr::literal(Value::Integer(1)),
                Expr::literal(Value::Integer(5)),
                Expr::literal(Value::Integer(10)),
            ],
            negated: true,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));

        // NOT IN - value in list should return false
        let expr = Expr::InList {
            expr: Box::new(Expr::literal(Value::Integer(5))),
            list: vec![
                Expr::literal(Value::Integer(1)),
                Expr::literal(Value::Integer(5)),
                Expr::literal(Value::Integer(10)),
            ],
            negated: true,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(false));
    }

    #[test]
    fn test_eval_in_list_with_null() {
        let row = vec![];
        let col_map = HashMap::new();

        // NULL in list - In FeatherDB, NULL == NULL uses Rust's PartialEq which returns true
        let expr = Expr::InList {
            expr: Box::new(Expr::literal(Value::Null)),
            list: vec![
                Expr::literal(Value::Integer(1)),
                Expr::literal(Value::Null),
                Expr::literal(Value::Integer(5)),
            ],
            negated: false,
        };
        // FeatherDB uses Rust's == operator, so NULL == NULL returns true
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));

        // Search for value with NULL in list (should not match NULL)
        let expr = Expr::InList {
            expr: Box::new(Expr::literal(Value::Integer(5))),
            list: vec![Expr::literal(Value::Null), Expr::literal(Value::Integer(5))],
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));
    }

    #[test]
    fn test_eval_like_percent() {
        let row = vec![];
        let col_map = HashMap::new();

        // Starts with pattern
        let expr = Expr::Like {
            expr: Box::new(Expr::literal(Value::Text("hello".into()))),
            pattern: "h%".to_string(),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));

        // Ends with pattern
        let expr = Expr::Like {
            expr: Box::new(Expr::literal(Value::Text("hello".into()))),
            pattern: "%lo".to_string(),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));

        // Contains pattern
        let expr = Expr::Like {
            expr: Box::new(Expr::literal(Value::Text("hello".into()))),
            pattern: "%ll%".to_string(),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));

        // No match
        let expr = Expr::Like {
            expr: Box::new(Expr::literal(Value::Text("hello".into()))),
            pattern: "%world%".to_string(),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(false));
    }

    #[test]
    fn test_eval_like_underscore() {
        let row = vec![];
        let col_map = HashMap::new();

        // Single character wildcard
        let expr = Expr::Like {
            expr: Box::new(Expr::literal(Value::Text("hello".into()))),
            pattern: "h_llo".to_string(),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));

        // Multiple single character wildcards
        let expr = Expr::Like {
            expr: Box::new(Expr::literal(Value::Text("hello".into()))),
            pattern: "h__lo".to_string(),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));

        // Wrong number of characters
        let expr = Expr::Like {
            expr: Box::new(Expr::literal(Value::Text("hello".into()))),
            pattern: "h_lo".to_string(),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(false));
    }

    #[test]
    fn test_eval_like_exact() {
        let row = vec![];
        let col_map = HashMap::new();

        // Exact match (no wildcards)
        let expr = Expr::Like {
            expr: Box::new(Expr::literal(Value::Text("hello".into()))),
            pattern: "hello".to_string(),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));

        // Not exact match
        let expr = Expr::Like {
            expr: Box::new(Expr::literal(Value::Text("hello".into()))),
            pattern: "world".to_string(),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(false));
    }

    #[test]
    fn test_eval_like_negated() {
        let row = vec![];
        let col_map = HashMap::new();

        // NOT LIKE with match should return false
        let expr = Expr::Like {
            expr: Box::new(Expr::literal(Value::Text("hello".into()))),
            pattern: "h%".to_string(),
            negated: true,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(false));

        // NOT LIKE with no match should return true
        let expr = Expr::Like {
            expr: Box::new(Expr::literal(Value::Text("hello".into()))),
            pattern: "%world%".to_string(),
            negated: true,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(true));
    }

    #[test]
    fn test_eval_like_non_text() {
        let row = vec![];
        let col_map = HashMap::new();

        // LIKE on non-text should return false (not error)
        let expr = Expr::Like {
            expr: Box::new(Expr::literal(Value::Integer(123))),
            pattern: "1%".to_string(),
            negated: false,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Boolean(false));
    }

    #[test]
    fn test_eval_case_simple() {
        let row = vec![];
        let col_map = HashMap::new();

        // CASE WHEN condition matches
        let expr = Expr::Case {
            operand: None,
            when_clauses: vec![
                (
                    Expr::literal(Value::Boolean(false)),
                    Expr::literal(Value::Text("first".into())),
                ),
                (
                    Expr::literal(Value::Boolean(true)),
                    Expr::literal(Value::Text("second".into())),
                ),
            ],
            else_result: Some(Box::new(Expr::literal(Value::Text("else".into())))),
        };
        assert_eq!(
            expr.eval(&row, &col_map).unwrap(),
            Value::Text("second".into())
        );
    }

    #[test]
    fn test_eval_case_else() {
        let row = vec![];
        let col_map = HashMap::new();

        // CASE with no match, falls through to ELSE
        let expr = Expr::Case {
            operand: None,
            when_clauses: vec![
                (
                    Expr::literal(Value::Boolean(false)),
                    Expr::literal(Value::Text("first".into())),
                ),
                (
                    Expr::literal(Value::Boolean(false)),
                    Expr::literal(Value::Text("second".into())),
                ),
            ],
            else_result: Some(Box::new(Expr::literal(Value::Text("else".into())))),
        };
        assert_eq!(
            expr.eval(&row, &col_map).unwrap(),
            Value::Text("else".into())
        );
    }

    #[test]
    fn test_eval_case_no_else() {
        let row = vec![];
        let col_map = HashMap::new();

        // CASE with no match and no ELSE returns NULL
        let expr = Expr::Case {
            operand: None,
            when_clauses: vec![
                (
                    Expr::literal(Value::Boolean(false)),
                    Expr::literal(Value::Text("first".into())),
                ),
                (
                    Expr::literal(Value::Boolean(false)),
                    Expr::literal(Value::Text("second".into())),
                ),
            ],
            else_result: None,
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Null);
    }

    #[test]
    fn test_eval_case_with_operand() {
        let row = vec![];
        let col_map = HashMap::new();

        // CASE operand WHEN value THEN result
        let expr = Expr::Case {
            operand: Some(Box::new(Expr::literal(Value::Integer(2)))),
            when_clauses: vec![
                (
                    Expr::literal(Value::Integer(1)),
                    Expr::literal(Value::Text("one".into())),
                ),
                (
                    Expr::literal(Value::Integer(2)),
                    Expr::literal(Value::Text("two".into())),
                ),
                (
                    Expr::literal(Value::Integer(3)),
                    Expr::literal(Value::Text("three".into())),
                ),
            ],
            else_result: Some(Box::new(Expr::literal(Value::Text("other".into())))),
        };
        assert_eq!(
            expr.eval(&row, &col_map).unwrap(),
            Value::Text("two".into())
        );

        // CASE operand with no match and ELSE
        let expr = Expr::Case {
            operand: Some(Box::new(Expr::literal(Value::Integer(5)))),
            when_clauses: vec![
                (
                    Expr::literal(Value::Integer(1)),
                    Expr::literal(Value::Text("one".into())),
                ),
                (
                    Expr::literal(Value::Integer(2)),
                    Expr::literal(Value::Text("two".into())),
                ),
            ],
            else_result: Some(Box::new(Expr::literal(Value::Text("other".into())))),
        };
        assert_eq!(
            expr.eval(&row, &col_map).unwrap(),
            Value::Text("other".into())
        );
    }

    #[test]
    fn test_eval_binary_null_arithmetic() {
        let row = vec![];
        let col_map = HashMap::new();

        // NULL + Integer should propagate NULL (SQL standard)
        let expr = Expr::binary(
            Expr::literal(Value::Null),
            BinaryOp::Add,
            Expr::literal(Value::Integer(1)),
        );
        let result = expr.eval(&row, &col_map).unwrap();
        assert_eq!(result, Value::Null);

        // Integer + NULL should propagate NULL
        let expr = Expr::binary(
            Expr::literal(Value::Integer(5)),
            BinaryOp::Add,
            Expr::literal(Value::Null),
        );
        let result = expr.eval(&row, &col_map).unwrap();
        assert_eq!(result, Value::Null);

        // NULL * Integer should propagate NULL
        let expr = Expr::binary(
            Expr::literal(Value::Null),
            BinaryOp::Mul,
            Expr::literal(Value::Integer(5)),
        );
        let result = expr.eval(&row, &col_map).unwrap();
        assert_eq!(result, Value::Null);
    }

    #[test]
    fn test_eval_nested_binary() {
        let row = vec![];
        let col_map = HashMap::new();

        // (1 + 2) * 3 = 9
        let expr = Expr::binary(
            Expr::binary(
                Expr::literal(Value::Integer(1)),
                BinaryOp::Add,
                Expr::literal(Value::Integer(2)),
            ),
            BinaryOp::Mul,
            Expr::literal(Value::Integer(3)),
        );
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Integer(9));

        // 10 - (3 + 2) = 5
        let expr = Expr::binary(
            Expr::literal(Value::Integer(10)),
            BinaryOp::Sub,
            Expr::binary(
                Expr::literal(Value::Integer(3)),
                BinaryOp::Add,
                Expr::literal(Value::Integer(2)),
            ),
        );
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Integer(5));
    }

    #[test]
    fn test_eval_function_upper() {
        let row = vec![];
        let col_map = HashMap::new();

        let expr = Expr::Function {
            name: "UPPER".to_string(),
            args: vec![Expr::literal(Value::Text("hello".into()))],
        };
        assert_eq!(
            expr.eval(&row, &col_map).unwrap(),
            Value::Text("HELLO".into())
        );
    }

    #[test]
    fn test_eval_function_lower() {
        let row = vec![];
        let col_map = HashMap::new();

        let expr = Expr::Function {
            name: "LOWER".to_string(),
            args: vec![Expr::literal(Value::Text("HELLO".into()))],
        };
        assert_eq!(
            expr.eval(&row, &col_map).unwrap(),
            Value::Text("hello".into())
        );
    }

    #[test]
    fn test_eval_function_length() {
        let row = vec![];
        let col_map = HashMap::new();

        let expr = Expr::Function {
            name: "LENGTH".to_string(),
            args: vec![Expr::literal(Value::Text("hello".into()))],
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Integer(5));
    }

    #[test]
    fn test_eval_function_coalesce() {
        let row = vec![];
        let col_map = HashMap::new();

        // COALESCE returns first non-NULL value
        let expr = Expr::Function {
            name: "COALESCE".to_string(),
            args: vec![
                Expr::literal(Value::Null),
                Expr::literal(Value::Integer(5)),
                Expr::literal(Value::Integer(10)),
            ],
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Integer(5));

        // COALESCE with all NULL returns NULL
        let expr = Expr::Function {
            name: "COALESCE".to_string(),
            args: vec![Expr::literal(Value::Null), Expr::literal(Value::Null)],
        };
        assert_eq!(expr.eval(&row, &col_map).unwrap(), Value::Null);
    }

    #[test]
    fn test_eval_function_with_column() {
        let row = vec![Value::Text("world".into())];
        let mut col_map = HashMap::new();
        col_map.insert("greeting".to_string(), 0);

        // UPPER(column)
        let expr = Expr::Function {
            name: "UPPER".to_string(),
            args: vec![Expr::column("greeting")],
        };
        assert_eq!(
            expr.eval(&row, &col_map).unwrap(),
            Value::Text("WORLD".into())
        );
    }
}