vibesql-executor 0.2.0

use std::collections::HashMap;

use crate::select::window::WindowFunctionKey;

/// Project columns from a row based on SELECT list (combined schema version)
pub(crate) fn project_row_combined(
    row: &vibesql_storage::Row,
    columns: &[vibesql_ast::SelectItem],
    evaluator: &crate::evaluator::CombinedExpressionEvaluator,
    schema: &crate::schema::CombinedSchema,
    window_mapping: &Option<HashMap<WindowFunctionKey, usize>>,
    buffer_pool: &vibesql_storage::QueryBufferPool,
) -> Result<vibesql_storage::Row, crate::errors::ExecutorError> {
    // Use pooled buffer to reduce allocation overhead
    let mut values = buffer_pool.get_value_buffer(columns.len());

    for item in columns {
        match item {
            vibesql_ast::SelectItem::Wildcard { .. } => {
                // SELECT * - include all columns except hidden ones (from NATURAL JOIN
                // deduplication) When window functions are present, only include
                // base columns (not appended window values)
                //
                // SQL Standard Column Ordering for USING/NATURAL JOINs:
                // According to SQL:2011, SELECT * from JOIN USING should output:
                // 1. The common/USING columns (merged) - appear FIRST
                // 2. Remaining columns from the left table
                // 3. Remaining columns from the right table
                let max_col = if let Some(mapping) = window_mapping {
                    if !mapping.is_empty() {
                        // Find the minimum window column index to know where base columns end
                        mapping.values().min().copied().unwrap_or(row.values.len())
                    } else {
                        row.values.len()
                    }
                } else {
                    row.values.len()
                };

                // Build column indices for SELECT * expansion
                //
                // Issue #4916: SQLite's behavior for USING column reordering:
                // - Simple USING joins (no alias): columns stay in table order (a, b, c)
                // - Aliased USING joins ((t1 JOIN t2 USING(b)) AS j1): USING columns first (b, a, c)
                //
                // We detect aliased joins by checking if there's an alias table that shadows
                // all non-alias tables. In that case, we use the alias table's column order
                // (which already has USING columns first) and resolve actual row indices.
                let column_indices: Vec<usize> = {
                    // Check if there's an alias table covering all non-alias tables
                    // This indicates an aliased parenthesized join expression
                    let alias_covering_all = schema.alias_tables.iter().find(|alias_id| {
                        if let Some(shadowed) = schema.shadowed_tables.get(*alias_id) {
                            // Count non-alias tables
                            let non_alias_tables: Vec<_> = schema
                                .table_schemas
                                .keys()
                                .filter(|t| !schema.alias_tables.contains(*t))
                                .collect();
                            // Check if all non-alias tables are shadowed by this alias
                            non_alias_tables.iter().all(|t| shadowed.contains(*t))
                        } else {
                            false
                        }
                    });

                    if let Some(alias_id) = alias_covering_all {
                        // Aliased join: use alias table's column order and resolve actual indices
                        let mut indices = Vec::new();
                        if let Some((_, alias_schema)) = schema.table_schemas.get(alias_id) {
                            for col_schema in &alias_schema.columns {
                                // Resolve the actual row index for this column
                                let alias_name = alias_id.display().to_string();
                                if let Some(actual_idx) =
                                    schema.get_column_index(Some(&alias_name), &col_schema.name)
                                {
                                    if actual_idx < max_col {
                                        indices.push(actual_idx);
                                    }
                                }
                            }
                        }
                        indices
                    } else {
                        // No covering alias: use table column order (no USING reordering)
                        let mut other_cols: Vec<usize> = Vec::new();

                        // Sort table_schemas by start_index for deterministic iteration order
                        let mut sorted_tables: Vec<_> = schema.table_schemas.iter().collect();
                        sorted_tables.sort_by_key(|(_, (start_index, _))| *start_index);

                        for (table_id, (start_index, table_schema)) in sorted_tables {
                            // Skip alias tables - they have incorrect row indices
                            if schema.alias_tables.contains(table_id) {
                                continue;
                            }
                            for (col_idx, _col_schema) in table_schema.columns.iter().enumerate() {
                                let abs_idx = start_index + col_idx;
                                if abs_idx >= max_col {
                                    continue;
                                }

                                // Skip replacement targets (output via hidden column's position)
                                if schema.column_replacement_map.values().any(|&v| v == abs_idx) {
                                    continue;
                                }

                                // Skip right-side USING columns (output via left-side with COALESCE)
                                if schema.is_using_coalesce_right_side(abs_idx) {
                                    continue;
                                }

                                // Check if column should be included
                                let should_include = if schema.is_column_hidden(abs_idx) {
                                    schema.get_column_replacement(abs_idx).is_some()
                                        || schema
                                            .get_using_coalesce_right_for_left(abs_idx)
                                            .is_some()
                                } else {
                                    true
                                };

                                if should_include {
                                    other_cols.push(abs_idx);
                                }
                            }
                        }
                        other_cols
                    }
                };

                // Iterate through reordered columns, handling hidden columns with replacements
                for idx in column_indices {
                    if schema.is_column_hidden(idx) {
                        // For hidden USING columns in OUTER JOINs, check for coalesce chain first
                        // This is needed for 3+ table FULL JOINs where the replacement always
                        // points to the rightmost column, but we need N-way COALESCE semantics
                        // to find the first non-NULL from ANY table in the chain.
                        if let Some(all_indices) = schema.get_all_coalesce_indices_for_column(idx) {
                            // Apply N-way COALESCE: return first non-NULL from entire chain
                            let mut found = false;
                            for &chain_idx in all_indices {
                                if chain_idx < row.values.len() {
                                    let val = &row.values[chain_idx];
                                    if *val != vibesql_types::SqlValue::Null {
                                        values.push(val.clone());
                                        found = true;
                                        break;
                                    }
                                }
                            }
                            if !found {
                                values.push(vibesql_types::SqlValue::Null);
                            }
                        } else if let Some(replacement_idx) = schema.get_column_replacement(idx) {
                            // Simple replacement without coalesce chain (2-table case)
                            if replacement_idx < row.values.len() {
                                values.push(row.values[replacement_idx].clone());
                            }
                        }
                        // If neither coalesce chain nor replacement, skip this hidden column
                    } else {
                        // Check if this column is part of a USING coalesce chain
                        // In FULL OUTER JOIN with USING, the visible column should show
                        // N-way COALESCE to handle unmatched rows from any side in the chain
                        if let Some(all_indices) = schema.get_all_coalesce_indices_for_column(idx) {
                            // Apply N-way COALESCE: return first non-NULL from entire chain
                            let mut found = false;
                            for &chain_idx in all_indices {
                                if chain_idx < row.values.len() {
                                    let val = &row.values[chain_idx];
                                    if *val != vibesql_types::SqlValue::Null {
                                        values.push(val.clone());
                                        found = true;
                                        break;
                                    }
                                }
                            }
                            if !found {
                                values.push(vibesql_types::SqlValue::Null);
                            }
                        } else {
                            values.push(row.values[idx].clone());
                        }
                    }
                }
            }
            vibesql_ast::SelectItem::QualifiedWildcard { qualifier, .. } => {
                // SELECT table.* or SELECT alias.* - include columns from specific table/alias
                // TableKey lookup is case-insensitive
                let table_id = vibesql_ast::TableIdentifier::unquoted(qualifier);
                let result = schema.get_table(qualifier).cloned();

                if let Some((start_index, table_schema)) = result {
                    // Issue #4916: For alias tables (e.g., j1 for `(t1 JOIN t2) AS j1`),
                    // the start_index is 0 which doesn't correspond to actual row positions.
                    // We need to resolve each column individually using get_column_index.
                    if schema.alias_tables.contains(&table_id) {
                        // Alias table: resolve each column to its actual row index
                        for col_schema in &table_schema.columns {
                            if let Some(actual_idx) =
                                schema.get_column_index(Some(qualifier), &col_schema.name)
                            {
                                if actual_idx < row.values.len() {
                                    // Handle USING column coalescing
                                    if let Some(all_indices) =
                                        schema.get_all_coalesce_indices_for_column(actual_idx)
                                    {
                                        // Apply N-way COALESCE: return first non-NULL from chain
                                        let mut found = false;
                                        for &chain_idx in all_indices {
                                            if chain_idx < row.values.len() {
                                                let val = &row.values[chain_idx];
                                                if *val != vibesql_types::SqlValue::Null {
                                                    values.push(val.clone());
                                                    found = true;
                                                    break;
                                                }
                                            }
                                        }
                                        if !found {
                                            values.push(vibesql_types::SqlValue::Null);
                                        }
                                    } else {
                                        values.push(row.values[actual_idx].clone());
                                    }
                                }
                            }
                        }
                    } else {
                        // Regular table: use start_index directly
                        let num_columns = table_schema.columns.len();
                        let end_index = start_index + num_columns;

                        // When window functions are present, only include base columns
                        let effective_end = if let Some(mapping) = window_mapping {
                            if !mapping.is_empty() {
                                // Find the minimum window column index to know where base columns end
                                let min_window_col =
                                    mapping.values().min().copied().unwrap_or(row.values.len());
                                end_index.min(min_window_col)
                            } else {
                                end_index
                            }
                        } else {
                            end_index
                        };

                        // Extract the columns for this table
                        if start_index < effective_end && effective_end <= row.values.len() {
                            values.extend(row.values[start_index..effective_end].iter().cloned());
                        }
                        // If indices are out of bounds, this might be an error, but we'll be silent
                        // for now
                    }
                }
                // If table not found, skip silently (this should be caught during column name
                // derivation)
            }
            vibesql_ast::SelectItem::Expression { expr, alias: _, .. } => {
                // Check if this is a window function expression
                let value = if let Some(mapping) = window_mapping {
                    evaluate_expression_with_windows(expr, row, evaluator, mapping)?
                } else {
                    evaluator.eval(expr, row)?
                };
                values.push(value);
            }
        }
    }

    // Move data to result and return pooled buffer
    // This allows buffer capacity reuse while avoiding clone overhead
    let result_values = std::mem::take(&mut values);
    buffer_pool.return_value_buffer(values);

    // Issue #4954: Preserve row_id and row_ids from source row when projecting
    // This is needed for queries like `SELECT rowid, a FROM t WHERE b=1` to work
    // correctly when an index scan is used
    let mut result = vibesql_storage::Row::new(result_values);
    if let Some(ref row_ids) = row.row_ids {
        result.row_ids = Some(row_ids.clone());
    } else if let Some(row_id) = row.row_id {
        result.row_id = Some(row_id);
    }
    Ok(result)
}

/// Evaluate an expression, checking for window functions first.
///
/// Strategy: if the expression contains any window function (anywhere in its
/// tree), recursively substitute every `WindowFunction` node with the
/// pre-computed value (as a literal) and evaluate the rewritten expression.
/// This handles arbitrarily nested expressions like
/// `count() OVER () LIKE lead(x) OVER (...)` (issue #5095) without needing a
/// special case per Expression variant.
pub(super) fn evaluate_expression_with_windows(
    expr: &vibesql_ast::Expression,
    row: &vibesql_storage::Row,
    evaluator: &crate::evaluator::CombinedExpressionEvaluator,
    window_mapping: &HashMap<WindowFunctionKey, usize>,
) -> Result<vibesql_types::SqlValue, crate::errors::ExecutorError> {
    use vibesql_ast::Expression;

    // Fast path: top-level window function lookup.
    if let Expression::WindowFunction { function, over } = expr {
        let key = WindowFunctionKey::from_expression(function, over);
        if let Some(&col_idx) = window_mapping.get(&key) {
            let value = row.values.get(col_idx).cloned().ok_or({
                crate::errors::ExecutorError::ColumnIndexOutOfBounds { index: col_idx }
            })?;
            return Ok(value);
        } else {
            return Err(crate::errors::ExecutorError::UnsupportedExpression(format!(
                "Window function not found in mapping: {:?}",
                expr
            )));
        }
    }

    // General path: substitute any nested window functions with literals,
    // then evaluate the rewritten tree.
    let substituted = substitute_window_functions(expr, row, window_mapping)?;
    evaluator.eval(&substituted, row)
}

/// Substitute window function expressions with literal values from
/// pre-computed results.
///
/// Recurses through every variant that wraps a sub-expression in the *outer*
/// scope. Subquery-bearing variants (`ScalarSubquery`, `Exists`,
/// `In { subquery }`, `QuantifiedComparison`) are intentionally not
/// descended into: a window inside a subquery belongs to the subquery's own
/// scope, not the outer SELECT's mapping. The OVER clause of a
/// `WindowFunction` is also not traversed for the same reason.
///
/// If a variant is missed, the regular evaluator will eventually encounter a
/// `WindowFunction` node it cannot resolve and emit a false-positive
/// "misuse of window function" error (issue #5095).
fn substitute_window_functions(
    expr: &vibesql_ast::Expression,
    row: &vibesql_storage::Row,
    window_mapping: &HashMap<WindowFunctionKey, usize>,
) -> Result<vibesql_ast::Expression, crate::errors::ExecutorError> {
    use vibesql_ast::Expression;

    // Helper: substitute over an Option<Box<Expression>>.
    let sub_opt_box = |opt: &Option<Box<Expression>>,
                       row: &vibesql_storage::Row,
                       wm: &HashMap<WindowFunctionKey, usize>|
     -> Result<Option<Box<Expression>>, crate::errors::ExecutorError> {
        opt.as_ref()
            .map(|e| substitute_window_functions(e, row, wm).map(Box::new))
            .transpose()
    };

    match expr {
        Expression::WindowFunction { function, over } => {
            // Look up the pre-computed value and convert to a literal expression
            let key = WindowFunctionKey::from_expression(function, over);
            if let Some(&col_idx) = window_mapping.get(&key) {
                let value = row.values.get(col_idx).cloned().ok_or({
                    crate::errors::ExecutorError::ColumnIndexOutOfBounds { index: col_idx }
                })?;
                Ok(Expression::Literal(value))
            } else {
                Err(crate::errors::ExecutorError::UnsupportedExpression(format!(
                    "Window function not found in mapping: {:?}",
                    expr
                )))
            }
        }
        Expression::BinaryOp { left, right, op } => {
            let left_sub = substitute_window_functions(left, row, window_mapping)?;
            let right_sub = substitute_window_functions(right, row, window_mapping)?;
            Ok(Expression::BinaryOp {
                left: Box::new(left_sub),
                right: Box::new(right_sub),
                op: *op,
            })
        }
        Expression::Conjunction(exprs) => {
            let subs: Result<Vec<_>, _> = exprs
                .iter()
                .map(|e| substitute_window_functions(e, row, window_mapping))
                .collect();
            Ok(Expression::Conjunction(subs?))
        }
        Expression::Disjunction(exprs) => {
            let subs: Result<Vec<_>, _> = exprs
                .iter()
                .map(|e| substitute_window_functions(e, row, window_mapping))
                .collect();
            Ok(Expression::Disjunction(subs?))
        }
        Expression::UnaryOp { expr: inner, op } => {
            let inner_sub = substitute_window_functions(inner, row, window_mapping)?;
            Ok(Expression::UnaryOp { expr: Box::new(inner_sub), op: *op })
        }
        Expression::Function { name, args, character_unit } => {
            let substituted_args: Result<Vec<_>, _> = args
                .iter()
                .map(|arg| substitute_window_functions(arg, row, window_mapping))
                .collect();
            Ok(Expression::Function {
                name: name.clone(),
                args: substituted_args?,
                character_unit: character_unit.clone(),
            })
        }
        Expression::AggregateFunction { name, distinct, args, order_by, filter } => {
            let substituted_args: Result<Vec<_>, _> = args
                .iter()
                .map(|arg| substitute_window_functions(arg, row, window_mapping))
                .collect();
            let substituted_order_by = order_by
                .as_ref()
                .map(|items| {
                    items
                        .iter()
                        .map(|item| {
                            substitute_window_functions(&item.expr, row, window_mapping).map(
                                |e| vibesql_ast::OrderByItem {
                                    expr: e,
                                    direction: item.direction.clone(),
                                    nulls_order: item.nulls_order,
                                },
                            )
                        })
                        .collect::<Result<Vec<_>, _>>()
                })
                .transpose()?;
            let substituted_filter = sub_opt_box(filter, row, window_mapping)?;
            Ok(Expression::AggregateFunction {
                name: name.clone(),
                distinct: *distinct,
                args: substituted_args?,
                order_by: substituted_order_by,
                filter: substituted_filter,
            })
        }
        Expression::Case { operand, when_clauses, else_result } => {
            let subst_operand = operand
                .as_ref()
                .map(|op| substitute_window_functions(op, row, window_mapping))
                .transpose()?
                .map(Box::new);

            let subst_when: Result<Vec<vibesql_ast::CaseWhen>, crate::ExecutorError> = when_clauses
                .iter()
                .map(|when_clause| {
                    let subst_conditions: Result<
                        Vec<vibesql_ast::Expression>,
                        crate::ExecutorError,
                    > = when_clause
                        .conditions
                        .iter()
                        .map(|cond| substitute_window_functions(cond, row, window_mapping))
                        .collect();

                    Ok(vibesql_ast::CaseWhen {
                        conditions: subst_conditions?,
                        result: substitute_window_functions(
                            &when_clause.result,
                            row,
                            window_mapping,
                        )?,
                    })
                })
                .collect();

            let subst_else = else_result
                .as_ref()
                .map(|e| substitute_window_functions(e, row, window_mapping))
                .transpose()?
                .map(Box::new);

            Ok(Expression::Case {
                operand: subst_operand,
                when_clauses: subst_when?,
                else_result: subst_else,
            })
        }
        Expression::IsNull { expr: inner, negated } => {
            let inner_sub = substitute_window_functions(inner, row, window_mapping)?;
            Ok(Expression::IsNull { expr: Box::new(inner_sub), negated: *negated })
        }
        Expression::IsDistinctFrom { left, right, negated } => {
            let left_sub = substitute_window_functions(left, row, window_mapping)?;
            let right_sub = substitute_window_functions(right, row, window_mapping)?;
            Ok(Expression::IsDistinctFrom {
                left: Box::new(left_sub),
                right: Box::new(right_sub),
                negated: *negated,
            })
        }
        Expression::IsTruthValue { expr: inner, truth_value, negated } => {
            let inner_sub = substitute_window_functions(inner, row, window_mapping)?;
            Ok(Expression::IsTruthValue {
                expr: Box::new(inner_sub),
                truth_value: *truth_value, // TruthValue: Copy
                negated: *negated,
            })
        }
        Expression::Cast { expr: inner, data_type } => {
            let inner_sub = substitute_window_functions(inner, row, window_mapping)?;
            Ok(Expression::Cast { expr: Box::new(inner_sub), data_type: data_type.clone() })
        }
        Expression::Like { expr: inner, pattern, negated, escape } => {
            let inner_sub = substitute_window_functions(inner, row, window_mapping)?;
            let pattern_sub = substitute_window_functions(pattern, row, window_mapping)?;
            let escape_sub = sub_opt_box(escape, row, window_mapping)?;
            Ok(Expression::Like {
                expr: Box::new(inner_sub),
                pattern: Box::new(pattern_sub),
                negated: *negated,
                escape: escape_sub,
            })
        }
        Expression::Glob { expr: inner, pattern, negated, escape } => {
            let inner_sub = substitute_window_functions(inner, row, window_mapping)?;
            let pattern_sub = substitute_window_functions(pattern, row, window_mapping)?;
            let escape_sub = sub_opt_box(escape, row, window_mapping)?;
            Ok(Expression::Glob {
                expr: Box::new(inner_sub),
                pattern: Box::new(pattern_sub),
                negated: *negated,
                escape: escape_sub,
            })
        }
        Expression::InList { expr: inner, values, negated } => {
            let inner_sub = substitute_window_functions(inner, row, window_mapping)?;
            let value_subs: Result<Vec<_>, _> = values
                .iter()
                .map(|v| substitute_window_functions(v, row, window_mapping))
                .collect();
            Ok(Expression::InList {
                expr: Box::new(inner_sub),
                values: value_subs?,
                negated: *negated,
            })
        }
        Expression::In { expr: inner, subquery, negated } => {
            // Only the LHS is in the outer scope; the subquery has its own scope
            // and will be re-planned independently.
            let inner_sub = substitute_window_functions(inner, row, window_mapping)?;
            Ok(Expression::In {
                expr: Box::new(inner_sub),
                subquery: subquery.clone(),
                negated: *negated,
            })
        }
        Expression::QuantifiedComparison { expr: inner, op, quantifier, subquery } => {
            // Only the LHS is in the outer scope.
            let inner_sub = substitute_window_functions(inner, row, window_mapping)?;
            Ok(Expression::QuantifiedComparison {
                expr: Box::new(inner_sub),
                op: op.clone(),
                quantifier: quantifier.clone(),
                subquery: subquery.clone(),
            })
        }
        Expression::Between { expr: inner, low, high, negated, symmetric } => {
            let inner_sub = substitute_window_functions(inner, row, window_mapping)?;
            let low_sub = substitute_window_functions(low, row, window_mapping)?;
            let high_sub = substitute_window_functions(high, row, window_mapping)?;
            Ok(Expression::Between {
                expr: Box::new(inner_sub),
                low: Box::new(low_sub),
                high: Box::new(high_sub),
                negated: *negated,
                symmetric: *symmetric,
            })
        }
        Expression::Position { substring, string, character_unit } => {
            let substring_sub = substitute_window_functions(substring, row, window_mapping)?;
            let string_sub = substitute_window_functions(string, row, window_mapping)?;
            Ok(Expression::Position {
                substring: Box::new(substring_sub),
                string: Box::new(string_sub),
                character_unit: character_unit.clone(),
            })
        }
        Expression::Trim { position, removal_char, string } => {
            let removal_sub = sub_opt_box(removal_char, row, window_mapping)?;
            let string_sub = substitute_window_functions(string, row, window_mapping)?;
            Ok(Expression::Trim {
                position: position.clone(),
                removal_char: removal_sub,
                string: Box::new(string_sub),
            })
        }
        Expression::Extract { field, expr: inner } => {
            let inner_sub = substitute_window_functions(inner, row, window_mapping)?;
            Ok(Expression::Extract { field: field.clone(), expr: Box::new(inner_sub) })
        }
        Expression::Interval { value, unit, leading_precision, fractional_precision } => {
            let value_sub = substitute_window_functions(value, row, window_mapping)?;
            Ok(Expression::Interval {
                value: Box::new(value_sub),
                unit: unit.clone(),
                leading_precision: *leading_precision,
                fractional_precision: *fractional_precision,
            })
        }
        // Subquery-bearing and leaf expressions are returned unchanged. A
        // window inside a subquery belongs to the subquery's own scope.
        _ => Ok(expr.clone()),
    }
}

/// Iterator that lazily projects rows based on SELECT list
///
/// This iterator wraps a source iterator and applies projection on-demand,
/// only computing projected values for rows that are actually consumed.
/// This is more efficient than eagerly projecting all rows when LIMIT/OFFSET
/// is present, as it avoids projecting rows that will be discarded.
pub struct SelectProjectionIterator<
    'a,
    I: Iterator<Item = Result<vibesql_storage::Row, crate::errors::ExecutorError>>,
> {
    source: I,
    select_list: Vec<vibesql_ast::SelectItem>,
    evaluator: crate::evaluator::CombinedExpressionEvaluator<'a>,
    input_schema: crate::schema::CombinedSchema,
    window_mapping: Option<HashMap<WindowFunctionKey, usize>>,
    buffer_pool: vibesql_storage::QueryBufferPool,
}

impl<'a, I: Iterator<Item = Result<vibesql_storage::Row, crate::errors::ExecutorError>>>
    SelectProjectionIterator<'a, I>
{
    /// Creates a new SelectProjectionIterator
    ///
    /// # Arguments
    /// * `source` - The source iterator providing rows to project
    /// * `select_list` - The SELECT items to project
    /// * `evaluator` - Expression evaluator for computing projected values
    /// * `input_schema` - Schema of the input rows
    /// * `window_mapping` - Optional mapping of window functions to column indices
    pub fn new(
        source: I,
        select_list: Vec<vibesql_ast::SelectItem>,
        evaluator: crate::evaluator::CombinedExpressionEvaluator<'a>,
        input_schema: crate::schema::CombinedSchema,
        window_mapping: Option<HashMap<WindowFunctionKey, usize>>,
        buffer_pool: vibesql_storage::QueryBufferPool,
    ) -> Self {
        Self { source, select_list, evaluator, input_schema, window_mapping, buffer_pool }
    }
}

impl<'a, I: Iterator<Item = Result<vibesql_storage::Row, crate::errors::ExecutorError>>> Iterator
    for SelectProjectionIterator<'a, I>
{
    type Item = Result<vibesql_storage::Row, crate::errors::ExecutorError>;

    fn next(&mut self) -> Option<Self::Item> {
        // Get next row from source
        let row = match self.source.next()? {
            Ok(r) => r,
            Err(e) => return Some(Err(e)),
        };

        // Clear CSE cache before projecting this row to prevent values
        // from being incorrectly cached across different rows
        self.evaluator.clear_cse_cache();

        // Project the row using the existing projection function
        let projected = project_row_combined(
            &row,
            &self.select_list,
            &self.evaluator,
            &self.input_schema,
            &self.window_mapping,
            &self.buffer_pool,
        );

        Some(projected)
    }
}