vibesql-executor 0.2.0

//! Index selection logic
//!
//! Determines when and which index to use for query optimization.
//! Supports both rule-based (simple) and cost-based (statistics-aware) selection.
//! Also includes skip-scan optimization for queries filtering on non-prefix columns.
//! Supports expression indexes (functional indexes) like CREATE INDEX idx ON t(lower(name)).

use vibesql_ast::{pretty_print::ToSql, Expression, IndexColumn};
use vibesql_catalog::TableSchema;
use vibesql_storage::{
    statistics::{AccessMethod, CostEstimator},
    Database,
};

use crate::evaluator::expression_hash::ExpressionHasher;

use crate::optimizer::index_planner::{IndexPlanner, SkipScanInfo};

/// Result of index selection, distinguishing between regular and skip-scan
#[derive(Debug, Clone)]
pub(crate) enum IndexScanChoice {
    /// Regular index scan using prefix columns
    Regular {
        index_name: String,
        sorted_columns: Option<Vec<(String, vibesql_ast::OrderDirection)>>,
    },
    /// Skip-scan using non-prefix column filter
    SkipScan { index_name: String, skip_scan_info: SkipScanInfo },
}

/// Check if any ORDER BY column is nullable in a way that prevents using the index
/// to satisfy the ORDER BY.
///
/// BTreeMap orders NULLs first (NULL < everything), but SQL default is:
/// - NULLS LAST for ASC
/// - NULLS FIRST for DESC
///
/// When a nullable column is used for ORDER BY without explicit NULLS FIRST/LAST,
/// the BTreeMap-iteration order would not match the SQL default for non-NULL
/// columns. To avoid producing rows in the wrong order, we normally reject the
/// index as a "pre-sort" — runtime path: `sorted_columns = None` → post-scan
/// `apply_order_by` runs.
///
/// **Pinned-column exception (per ORDER BY item)**: If an ORDER BY item is itself
/// one of the pinned leading index columns (e.g., `ORDER BY a, b` with
/// `WHERE a IN (1,2,3)` and pinned `a`), that ORDER BY item is constant within
/// each scan group and its NULL ordering is irrelevant — IN-lists/equality with
/// literal NULL never match NULL rows. The nullable check is skipped for such
/// items.
///
/// `index_columns` is the indexed-column list and `pinned_columns` is the count
/// of leading consecutive columns pinned by equality/IN predicates. Pass an
/// empty slice and `0` to apply the unconditional nullable rejection (callers
/// without index context should use this form).
fn any_order_by_column_nullable(
    order_items: &[vibesql_ast::OrderByItem],
    table_schema: &TableSchema,
    index_columns: &[IndexColumn],
    pinned_columns: usize,
) -> bool {
    let pinned_index_slice = &index_columns[..pinned_columns.min(index_columns.len())];
    for item in order_items {
        // Skip the per-item nullable check when this ORDER BY item refers to
        // one of the pinned leading index columns (constant within scan group).
        if !pinned_index_slice.is_empty()
            && order_item_matches_pinned_column(item, pinned_index_slice)
        {
            continue;
        }

        if let Expression::ColumnRef(col_id) = &item.expr {
            let column = col_id.column_canonical();
            // Look up column in schema (case-insensitive)
            if let Some(idx) = table_schema.get_column_index(column) {
                if table_schema.columns[idx].nullable {
                    return true;
                }
            }
        }
    }
    false
}

/// Determines if an index scan is beneficial for the given query
///
/// Returns Some((index_name, sorted_columns)) if an index should be used, None otherwise.
/// The sorted_columns vector indicates which columns are pre-sorted by the index scan.
#[allow(clippy::type_complexity)]
pub(crate) fn should_use_index_scan(
    table_name: &str,
    where_clause: Option<&Expression>,
    order_by: Option<&[vibesql_ast::OrderByItem]>,
    database: &Database,
) -> Option<(String, Option<Vec<(String, vibesql_ast::OrderDirection)>>)> {
    // We use indexes in three scenarios:
    // 1. WHERE clause references an indexed column (with or without ORDER BY)
    // 2. ORDER BY references an indexed column (even without WHERE)
    // 3. Both WHERE and ORDER BY use the same index
    //
    // Note: Index scans can provide partial optimization even for complex
    // predicates (including OR expressions). The full WHERE clause is always
    // applied as a post-filter in execute_index_scan() to ensure correctness.

    // Get all indexes for this table
    let table = database.get_table(table_name)?;
    let indexes = database.list_indexes_for_table(table_name);

    if indexes.is_empty() {
        return None;
    }

    // Find the best index (most pinned columns = better filtering)
    // We evaluate all applicable indexes and pick the one that covers the most WHERE columns
    let mut best_index: Option<(
        String,
        usize,
        bool,
        Option<Vec<(String, vibesql_ast::OrderDirection)>>,
    )> = None;
    // (index_name, pinned_count, can_use_for_order, sorted_columns)

    for index_name in &indexes {
        if let Some(index_metadata) = database.get_index(index_name) {
            // Partial indexes (CREATE INDEX ... WHERE expr) only cover the
            // subset of rows for which the predicate evaluates to TRUE. They
            // are usable only when the query's WHERE clause structurally
            // implies the index predicate (every index-predicate conjunct
            // appears among the query's top-level AND conjuncts). The full
            // WHERE clause is re-applied as a post-filter in
            // execute_index_scan(), so an over-inclusive index body cannot
            // produce wrong rows. The partial predicate lives on the
            // catalog-side `IndexMetadata`.
            if !crate::optimizer::predicate_implication::partial_index_usable(
                database,
                index_name,
                where_clause,
            ) {
                continue;
            }

            let first_indexed_column = index_metadata.columns.first()?;

            // Check if this index can be used for WHERE clause
            // Supports both column indexes and expression indexes
            let can_use_for_where = where_clause
                .map(|expr| index_column_can_filter(expr, first_indexed_column))
                .unwrap_or(false);

            // Count how many leading index columns are pinned by equality predicates
            // Note: For expression indexes, we count expression matches as pinned
            let pinned_columns = count_pinned_index_columns(where_clause, &index_metadata.columns);

            // Check if this index can be used for ORDER BY clause
            let can_use_for_order = if let Some(order_items) = order_by {
                // Check if ORDER BY columns match the index columns (after skipping pinned columns)
                let columns_match = can_use_index_for_order_by_with_pinned(
                    order_items,
                    &index_metadata.columns,
                    pinned_columns,
                );

                // Don't use index for ORDER BY if any non-pinned column is nullable.
                // BTreeMap orders NULLs first, but SQL default is NULLS LAST for ASC,
                // which would produce incorrect results for nullable columns. When
                // at least one leading index column is pinned by equality/IN, the
                // pinned-prefix exception in `any_order_by_column_nullable` allows
                // the index to be used regardless of trailing nullability.
                if columns_match
                    && any_order_by_column_nullable(
                        order_items,
                        &table.schema,
                        &index_metadata.columns,
                        pinned_columns,
                    )
                {
                    false
                } else {
                    columns_match
                }
            } else {
                false
            };

            // Skip if this index can't help with WHERE or ORDER BY
            if !can_use_for_where && !can_use_for_order {
                continue;
            }

            // Build sorted_columns metadata if ORDER BY can be satisfied
            let sorted_columns = if can_use_for_order {
                let order_items = order_by.unwrap();
                Some(
                    order_items
                        .iter()
                        .map(|item| {
                            // For expression indexes, ORDER BY may use expressions (e.g., length(a))
                            // We use to_sql() to convert the expression to a string representation
                            // The actual string is only used for metadata; the important part
                            // is the direction for determining scan order
                            let col_name = match &item.expr {
                                Expression::ColumnRef(col_id) => {
                                    col_id.column_canonical().to_string()
                                }
                                expr => expr.to_sql(),
                            };
                            (col_name, item.direction.clone())
                        })
                        .collect(),
                )
            } else {
                None
            };

            // Compare with best index so far
            // Prefer: more pinned columns > can satisfy ORDER BY > first found
            let is_better = match &best_index {
                None => true,
                Some((_, best_pinned, best_can_order, _)) => {
                    // More pinned columns = better (narrows down rows more)
                    if pinned_columns > *best_pinned {
                        true
                    } else if pinned_columns == *best_pinned {
                        // Same pinned count: prefer one that can satisfy ORDER BY
                        can_use_for_order && !*best_can_order
                    } else {
                        false
                    }
                }
            };

            if is_better {
                best_index =
                    Some((index_name.clone(), pinned_columns, can_use_for_order, sorted_columns));
            }
        }
    }

    // Return the best index if we found one
    if let Some((index_name, _, _, sorted_columns)) = best_index {
        return Some((index_name, sorted_columns));
    }

    None
}

/// Check if an expression filters a specific column
///
/// Returns true if the expression contains a predicate on the given column
/// For example: "WHERE age = 25" filters column "age"
pub(crate) fn expression_filters_column(expr: &Expression, column_name: &str) -> bool {
    match expr {
        Expression::BinaryOp { left, op, right } => {
            // Check for comparison operators
            match op {
                vibesql_ast::BinaryOperator::Equal
                | vibesql_ast::BinaryOperator::GreaterThan
                | vibesql_ast::BinaryOperator::GreaterThanOrEqual
                | vibesql_ast::BinaryOperator::LessThan
                | vibesql_ast::BinaryOperator::LessThanOrEqual => {
                    // Index can only filter when comparing column to a LITERAL value
                    // NOT when comparing column to another column (equijoin conditions)
                    // e.g., `l_shipdate > '1995-03-15'` CAN use index
                    // e.g., `l_orderkey = o_orderkey` CANNOT use index
                    let left_is_col = is_column_reference(left, column_name);
                    let right_is_col = is_column_reference(right, column_name);
                    let left_is_literal = is_literal(left);
                    let right_is_literal = is_literal(right);

                    // column op literal OR literal op column
                    if (left_is_col && right_is_literal) || (left_is_literal && right_is_col) {
                        return true;
                    }
                }
                vibesql_ast::BinaryOperator::And | vibesql_ast::BinaryOperator::Or => {
                    // Recursively check sub-expressions for AND/OR
                    return expression_filters_column(left, column_name)
                        || expression_filters_column(right, column_name);
                }
                _ => {}
            }
            false
        }
        // IS / IS NOT (NULL-safe comparison)
        // negated=true means "IS NOT DISTINCT FROM" which is equivalent to "IS" (NULL-safe equals)
        // negated=false means "IS DISTINCT FROM" which is equivalent to "IS NOT" (NULL-safe not-equals)
        // We only use the index for IS (negated=true) as it's equivalent to =
        Expression::IsDistinctFrom { left, right, negated: true } => {
            let left_is_col = is_column_reference(left, column_name);
            let right_is_col = is_column_reference(right, column_name);
            let left_is_literal = is_literal(left);
            let right_is_literal = is_literal(right);
            // column IS literal OR literal IS column
            (left_is_col && right_is_literal) || (left_is_literal && right_is_col)
        }
        // IN with value list: col IN (1, 2, 3)
        Expression::InList { expr, .. } => is_column_reference(expr, column_name),
        // IN with subquery: col IN (SELECT ...)
        Expression::In { expr, .. } => is_column_reference(expr, column_name),
        // BETWEEN: col BETWEEN low AND high
        Expression::Between { expr, .. } => is_column_reference(expr, column_name),
        // Conjunction: AND
        Expression::Conjunction(exprs) => {
            exprs.iter().any(|e| expression_filters_column(e, column_name))
        }
        // Disjunction: OR
        Expression::Disjunction(exprs) => {
            exprs.iter().any(|e| expression_filters_column(e, column_name))
        }
        _ => false,
    }
}

/// Check if an expression matches an expression index
///
/// For expression indexes like `CREATE INDEX idx ON t(lower(name))`, this function
/// checks if the WHERE clause contains a predicate on the indexed expression.
/// For example: `WHERE lower(name) = 'john'` matches the index.
///
/// Uses structural hashing to compare expressions - two expressions are considered
/// equivalent if they have the same structure (same operations on same columns).
pub(crate) fn expression_filters_index_expression(
    where_expr: &Expression,
    index_expr: &Expression,
) -> bool {
    let index_hash = ExpressionHasher::hash(index_expr);
    expression_contains_matching_predicate(where_expr, index_hash)
}

/// Check if an expression contains a predicate on a specific expression (by hash)
///
/// Recursively searches through AND/OR combinations to find predicates.
fn expression_contains_matching_predicate(expr: &Expression, target_hash: u64) -> bool {
    match expr {
        Expression::BinaryOp { left, op, right } => {
            match op {
                vibesql_ast::BinaryOperator::Equal
                | vibesql_ast::BinaryOperator::GreaterThan
                | vibesql_ast::BinaryOperator::GreaterThanOrEqual
                | vibesql_ast::BinaryOperator::LessThan
                | vibesql_ast::BinaryOperator::LessThanOrEqual => {
                    // Check if left or right side matches the indexed expression
                    let left_hash = ExpressionHasher::hash(left);
                    let right_hash = ExpressionHasher::hash(right);
                    let left_is_literal = is_literal(left);
                    let right_is_literal = is_literal(right);

                    // expr op literal OR literal op expr
                    if (left_hash == target_hash && right_is_literal)
                        || (right_hash == target_hash && left_is_literal)
                    {
                        return true;
                    }
                }
                vibesql_ast::BinaryOperator::And | vibesql_ast::BinaryOperator::Or => {
                    return expression_contains_matching_predicate(left, target_hash)
                        || expression_contains_matching_predicate(right, target_hash);
                }
                _ => {}
            }
            false
        }
        // IS / IS NOT (NULL-safe comparison)
        Expression::IsDistinctFrom { left, right, negated: true } => {
            let left_hash = ExpressionHasher::hash(left);
            let right_hash = ExpressionHasher::hash(right);
            let left_is_literal = is_literal(left);
            let right_is_literal = is_literal(right);
            (left_hash == target_hash && right_is_literal)
                || (right_hash == target_hash && left_is_literal)
        }
        // IN with value list: expr IN (1, 2, 3)
        Expression::InList { expr, .. } => ExpressionHasher::hash(expr) == target_hash,
        // IN with subquery: expr IN (SELECT ...)
        Expression::In { expr, .. } => ExpressionHasher::hash(expr) == target_hash,
        // BETWEEN: expr BETWEEN low AND high
        Expression::Between { expr, .. } => ExpressionHasher::hash(expr) == target_hash,
        // Conjunction: AND
        Expression::Conjunction(exprs) => {
            exprs.iter().any(|e| expression_contains_matching_predicate(e, target_hash))
        }
        // Disjunction: OR
        Expression::Disjunction(exprs) => {
            exprs.iter().any(|e| expression_contains_matching_predicate(e, target_hash))
        }
        _ => false,
    }
}

/// Check if an IndexColumn can be used for the given WHERE clause
///
/// This function handles both column indexes and expression indexes:
/// - For column indexes: delegates to `expression_filters_column`
/// - For expression indexes: uses `expression_filters_index_expression`
pub(crate) fn index_column_can_filter(where_expr: &Expression, index_col: &IndexColumn) -> bool {
    match index_col {
        IndexColumn::Column { column_name, .. } => {
            expression_filters_column(where_expr, column_name)
        }
        IndexColumn::Expression { expr, .. } => {
            expression_filters_index_expression(where_expr, expr)
        }
    }
}

/// Check if an expression is a reference to a specific column.
///
/// Uses case-insensitive comparison because:
/// - ColumnIdentifier.column_canonical() returns lowercase for unquoted identifiers
/// - Schema metadata (column_name) may not be consistently lowercased
/// - For SQL:1999 compliance with quoted identifiers, both sides should use canonical forms
///
/// TODO: Once schema metadata consistently uses canonical forms, change to direct equality
pub(super) fn is_column_reference(expr: &Expression, column_name: &str) -> bool {
    match expr {
        Expression::ColumnRef(col_id) => {
            col_id.column_canonical().eq_ignore_ascii_case(column_name)
        }
        _ => false,
    }
}

/// Check if an expression is a non-NULL literal value or parameter placeholder
///
/// Index scans can filter on columns compared to literal values or bound parameters,
/// not columns compared to other columns (which are equijoin conditions).
/// Parameter placeholders (?, $1, :name, etc.) are treated as literals for index selection
/// because they are resolved to concrete values at execution time.
///
/// **Important**: NULL literals are excluded because `col = NULL` can never return true
/// (per SQL three-valued logic), so using an index for this predicate is incorrect.
fn is_literal(expr: &Expression) -> bool {
    match expr {
        // Exclude NULL literals - col = NULL can never match rows
        Expression::Literal(vibesql_types::SqlValue::Null) => false,
        Expression::Literal(_) => true,
        Expression::Placeholder(_)
        | Expression::NumberedPlaceholder(_)
        | Expression::NamedPlaceholder(_) => true,
        _ => false,
    }
}

/// Check if two expressions are structurally equivalent for index matching purposes.
///
/// This performs a normalized comparison of expressions, suitable for matching
/// WHERE clause expressions against expression index definitions.
///
/// Normalization includes:
/// - Case-insensitive comparison of column names and function names
/// - Order-independent comparison for commutative operators where applicable
///
/// # Example
/// ```text
/// Index: CREATE INDEX idx ON t(LOWER(email))
/// Query: SELECT * FROM t WHERE LOWER(email) = 'x'
///
/// expressions_match(LOWER(email), LOWER(email)) -> true
/// ```
pub(crate) fn expressions_match(expr1: &Expression, expr2: &Expression) -> bool {
    match (expr1, expr2) {
        // Column references: case-insensitive comparison
        (Expression::ColumnRef(col1), Expression::ColumnRef(col2)) => {
            col1.column_canonical().eq_ignore_ascii_case(&col2.column_canonical())
        }

        // Function calls: compare function name and arguments
        (
            Expression::Function { name: name1, args: args1, .. },
            Expression::Function { name: name2, args: args2, .. },
        ) => {
            // Function names are case-insensitive in SQL
            name1.canonical().eq_ignore_ascii_case(name2.canonical())
                && args1.len() == args2.len()
                && args1.iter().zip(args2.iter()).all(|(a1, a2)| expressions_match(a1, a2))
        }

        // Binary operations: compare operator and operands
        (
            Expression::BinaryOp { left: l1, op: op1, right: r1 },
            Expression::BinaryOp { left: l2, op: op2, right: r2 },
        ) => op1 == op2 && expressions_match(l1, l2) && expressions_match(r1, r2),

        // Unary operations
        (Expression::UnaryOp { op: op1, expr: e1 }, Expression::UnaryOp { op: op2, expr: e2 }) => {
            op1 == op2 && expressions_match(e1, e2)
        }

        // Cast expressions
        (
            Expression::Cast { expr: e1, data_type: t1 },
            Expression::Cast { expr: e2, data_type: t2 },
        ) => t1 == t2 && expressions_match(e1, e2),

        // Literals: exact match
        (Expression::Literal(v1), Expression::Literal(v2)) => v1 == v2,

        // CASE expressions
        (
            Expression::Case { operand: op1, when_clauses: wc1, else_result: er1 },
            Expression::Case { operand: op2, when_clauses: wc2, else_result: er2 },
        ) => {
            // Compare operand
            let op_match = match (op1, op2) {
                (Some(o1), Some(o2)) => expressions_match(o1, o2),
                (None, None) => true,
                _ => false,
            };
            // Compare else
            let else_match = match (er1, er2) {
                (Some(e1), Some(e2)) => expressions_match(e1, e2),
                (None, None) => true,
                _ => false,
            };
            // Compare when clauses (conditions is a Vec, result is Expression)
            op_match
                && else_match
                && wc1.len() == wc2.len()
                && wc1.iter().zip(wc2.iter()).all(|(w1, w2)| {
                    w1.conditions.len() == w2.conditions.len()
                        && w1
                            .conditions
                            .iter()
                            .zip(w2.conditions.iter())
                            .all(|(c1, c2)| expressions_match(c1, c2))
                        && expressions_match(&w1.result, &w2.result)
                })
        }

        // NULL handling
        (
            Expression::IsNull { expr: e1, negated: n1 },
            Expression::IsNull { expr: e2, negated: n2 },
        ) => n1 == n2 && expressions_match(e1, e2),

        // Conjunction (AND chain)
        (Expression::Conjunction(exprs1), Expression::Conjunction(exprs2)) => {
            exprs1.len() == exprs2.len()
                && exprs1.iter().zip(exprs2.iter()).all(|(e1, e2)| expressions_match(e1, e2))
        }

        // Disjunction (OR chain)
        (Expression::Disjunction(exprs1), Expression::Disjunction(exprs2)) => {
            exprs1.len() == exprs2.len()
                && exprs1.iter().zip(exprs2.iter()).all(|(e1, e2)| expressions_match(e1, e2))
        }

        // For all other expression types, they must be exactly equal
        // This is conservative but safe
        _ => expr1 == expr2,
    }
}

/// Case-insensitive lookup of column statistics
///
/// Returns the ColumnStatistics for a column name, ignoring case differences.
/// This is necessary because schema column names may use different casing than
/// index column names or query column references.
fn get_column_stats_ignore_case<'a>(
    columns: &'a std::collections::HashMap<String, vibesql_storage::statistics::ColumnStatistics>,
    column_name: &str,
) -> Option<&'a vibesql_storage::statistics::ColumnStatistics> {
    // First try exact match for efficiency
    if let Some(stats) = columns.get(column_name) {
        return Some(stats);
    }
    // Fall back to case-insensitive search
    columns.iter().find(|(key, _)| key.eq_ignore_ascii_case(column_name)).map(|(_, stats)| stats)
}

/// Check if an index can be used to satisfy an ORDER BY clause
///
/// Returns true if the ORDER BY columns match the index columns (after skipping
/// any prefix columns pinned by equality predicates) and the sort directions
/// are compatible (either all matching or all reversed).
///
/// Examples:
/// - ORDER BY col0 ASC can use index (col0 ASC)
/// - ORDER BY col0 DESC can use index (col0 DESC) via reversal
/// - ORDER BY col0, col1 can use index (col0, col1)
/// - WHERE col0 = 1 ORDER BY col1 can use index (col0, col1) - col0 is pinned
pub(crate) fn can_use_index_for_order_by(
    order_items: &[vibesql_ast::OrderByItem],
    index_columns: &[vibesql_ast::IndexColumn],
) -> bool {
    can_use_index_for_order_by_with_pinned(order_items, index_columns, 0)
}

/// Returns true if `item` is a simple column-reference ORDER BY whose column name
/// matches one of the pinned leading index columns. Pinned columns are constants
/// within an index scan group (because they were filtered to a single value or
/// IN-list), so a leading ORDER BY reference to a pinned column is trivially
/// satisfied by the index regardless of nullability or direction.
fn order_item_matches_pinned_column(
    item: &vibesql_ast::OrderByItem,
    pinned_index_columns: &[vibesql_ast::IndexColumn],
) -> bool {
    let order_col_name = match &item.expr {
        Expression::ColumnRef(col_id)
            if col_id.schema_canonical().is_none() && col_id.table_canonical().is_none() =>
        {
            col_id.column_canonical()
        }
        _ => return false,
    };
    pinned_index_columns.iter().any(|ic| match ic {
        vibesql_ast::IndexColumn::Column { column_name, .. } => {
            order_col_name.eq_ignore_ascii_case(column_name)
        }
        vibesql_ast::IndexColumn::Expression { .. } => false,
    })
}

/// EQP-only: Determine if the ORDER BY can be served by an existing index in a
/// way that suppresses the `USE TEMP B-TREE FOR ORDER BY` step in EXPLAIN QUERY
/// PLAN output, even when a runtime post-sort would still be needed.
///
/// SQLite's EQP suppresses `USE TEMP B-TREE FOR ORDER BY` whenever an index's
/// natural traversal yields rows in (or close to) the requested order — even if
/// a final stabilization pass is required for NULL placement on nullable
/// trailing columns. We mirror that behavior: when an index has its leading
/// columns pinned by equality/IN predicates AND the ORDER BY structurally
/// aligns with the pinned-prefix-then-trailing-index layout, the EQP omits the
/// temp B-tree step. The runtime path still emits a post-scan sort when the
/// `sorted_columns` returned by `cost_based_index_selection` is `None` (which
/// happens when trailing nullable columns appear in the ORDER BY).
///
/// Returns true if a temp B-tree should be shown in EXPLAIN, false otherwise.
pub(crate) fn needs_temp_btree_for_order_by_eqp(
    table_name: &str,
    where_clause: Option<&Expression>,
    order_by: &[vibesql_ast::OrderByItem],
    database: &Database,
) -> bool {
    // No table → constant expression, no sort needed.
    let Some(table) = database.get_table(table_name) else {
        return true;
    };

    // #5375: ORDER BY led by the table's INTEGER PRIMARY KEY (the rowid
    // alias) over a plain sequential scan. The executor guarantees that a
    // sequential scan of a rowid-alias table delivers rows in rowid order
    // (`sort_rows_by_integer_primary_key`, #4926), so the requested order —
    // or its exact reverse, for DESC — IS the scan's natural output order
    // and the runtime ORDER BY sort is order-equivalent to the natural
    // traversal. sqlite3 3.51.0 shows a bare `SCAN t` with no temp line here
    // (its table B-tree is keyed by rowid); suppressing the line falls under
    // the same permissive EQP convention as the index stabilization-sort
    // suppression below. Trailing ORDER BY terms are ignored: the rowid is
    // unique, so they can never affect the order (verified live: sqlite3
    // suppresses for `ORDER BY id, y` and `ORDER BY id ASC, y DESC`).
    //
    // This only applies when the planner picks NO index for the scan: if the
    // WHERE clause rides an index (SEARCH/skip-scan), rows arrive in index
    // order, not rowid order, and the sort is real. It also does NOT apply
    // to the bare `rowid` pseudo-column on tables WITHOUT an INTEGER PRIMARY
    // KEY: their sequential scan yields physical insertion order, which
    // explicit-rowid INSERTs and `UPDATE ... SET rowid = ...` can decouple
    // from rowid order, so that temp line stays truthful (documented
    // divergence from sqlite3).
    if order_by_leads_with_rowid_alias(&table.schema, table_name, order_by)
        && cost_based_index_selection(table_name, where_clause, Some(order_by), database).is_none()
        && !has_skip_scan_plan(table_name, where_clause, database)
    {
        return false;
    }

    eqp_ordering_index(table_name, where_clause, order_by, database, false).is_none()
}

/// True when the planner would choose a skip-scan for this table + WHERE
/// (mirrors the skip-scan fallback in `ExplainExecutor::explain_table_scan`).
fn has_skip_scan_plan(
    table_name: &str,
    where_clause: Option<&Expression>,
    database: &Database,
) -> bool {
    let Some(where_expr) = where_clause else {
        return false;
    };
    IndexPlanner::new(database).plan_skip_scan(table_name, where_expr).is_some()
}

/// EQP-only (#5375): true when the FIRST ORDER BY term is a column reference
/// to the table's INTEGER PRIMARY KEY rowid alias — either by the declared
/// column name, or via the `rowid`/`_rowid_`/`oid` pseudo-column keywords
/// (which resolve to the alias column; real columns shadow the keywords, and
/// WITHOUT ROWID tables have no rowid pseudo-column, #4953).
///
/// Direction is ignored: ASC is the scan's natural rowid order and DESC its
/// exact reverse (the same reverse-traversal convention as index DESC
/// suppression; sqlite3 shows a bare `SCAN t` for both). Trailing terms are
/// ignored because the rowid is unique.
fn order_by_leads_with_rowid_alias(
    schema: &TableSchema,
    table_name: &str,
    order_by: &[vibesql_ast::OrderByItem],
) -> bool {
    let Some(ipk_idx) = schema.rowid_alias_column else {
        return false;
    };
    let Some(ipk_col) = schema.columns.get(ipk_idx) else {
        return false;
    };
    let Some(first) = order_by.first() else {
        return false;
    };
    let Expression::ColumnRef(col_id) = &first.expr else {
        return false;
    };
    // Schema-qualified references are out of scope; a table qualifier must
    // match the scanned table.
    if col_id.schema_canonical().is_some() {
        return false;
    }
    if let Some(qualifier) = col_id.table_canonical() {
        if !qualifier.eq_ignore_ascii_case(table_name) {
            return false;
        }
    }

    let col_name = col_id.column_canonical();
    if col_name.eq_ignore_ascii_case(&ipk_col.name) {
        return true;
    }

    // rowid/_rowid_/oid keywords: only when no real column shadows the
    // keyword (real columns take precedence, matching the evaluator) and the
    // table is not WITHOUT ROWID (no rowid pseudo-column there).
    let is_rowid_keyword = col_name.eq_ignore_ascii_case("rowid")
        || col_name.eq_ignore_ascii_case("_rowid_")
        || col_name.eq_ignore_ascii_case("oid");
    is_rowid_keyword && !schema.without_rowid && schema.get_column_index(col_name).is_none()
}

/// EQP-only: the index whose natural traversal delivers `order_by` for a
/// scan of `table_name`, if any.
///
/// Checks, in order:
/// 1. The planner's chosen index (`cost_based_index_selection`): when it
///    returns full `sorted_columns` the order is satisfied outright; even
///    when the runtime nullable-column guard withholds `sorted_columns`, a
///    structural match means the scan still delivers index order (e.g.
///    windowpushd.test 2.1.3.4: `SEARCH t1 USING INDEX i2 (b>?)` feeds
///    PARTITION BY b — SQLite shows no temp B-tree).
/// 2. Any other index with its leading columns pinned by equality/IN whose
///    remaining columns structurally align with the ORDER BY.
///
/// With a WHERE clause, no leading pinned column, and a competing access
/// path chosen by the planner, an unpinned structural match does NOT count
/// (the planner scans a different index, so ordering is not delivered).
/// `prefer_ordering_scan` relaxes that gate when the planner chose NO index:
/// the scan is then free to traverse the ordering index — SQLite picks the
/// index that delivers a window's PARTITION BY/ORDER BY order even without
/// any usable predicate (windowpushd.test 2.1.1.5, 2.1.3.6).
pub(crate) fn eqp_ordering_index(
    table_name: &str,
    where_clause: Option<&Expression>,
    order_by: &[vibesql_ast::OrderByItem],
    database: &Database,
    prefer_ordering_scan: bool,
) -> Option<String> {
    if order_by.is_empty() {
        return None;
    }

    let chosen = cost_based_index_selection(table_name, where_clause, Some(order_by), database);

    // The planner's chosen index delivers index order when the ORDER BY
    // structurally fits (pinned-prefix + trailing-index suffix), regardless
    // of the runtime nullable-column guard.
    if let Some((chosen_name, _)) = &chosen {
        if let Some(index_metadata) = database.get_index(chosen_name) {
            let pinned_columns = count_pinned_index_columns(where_clause, &index_metadata.columns);
            if can_use_index_for_order_by_with_pinned(
                order_by,
                &index_metadata.columns,
                pinned_columns,
            ) {
                return Some(chosen_name.clone());
            }
        }
    }

    // Fallback: check whether ANY index has its leading columns pinned by
    // equality/IN AND the ORDER BY structurally aligns with the index. If so,
    // SQLite's EQP would omit `USE TEMP B-TREE FOR ORDER BY`. The runtime
    // correctness of NULL placement is handled by the post-scan sort.
    let indexes = database.list_indexes_for_table(table_name);
    for index_name in &indexes {
        let Some(index_metadata) = database.get_index(index_name) else { continue };
        // Partial indexes participate in the EQP exemption only when the
        // query's WHERE clause structurally implies the index predicate —
        // otherwise the index cannot be relied on to satisfy ORDER BY.
        if !crate::optimizer::predicate_implication::partial_index_usable(
            database,
            index_name,
            where_clause,
        ) {
            continue;
        }
        if index_metadata.columns.is_empty() {
            continue;
        }
        let pinned_columns = count_pinned_index_columns(where_clause, &index_metadata.columns);
        // With a WHERE clause and no leading pinned column, the planner may
        // choose a different (filtering) index, so the EQP exception only
        // applies when leading columns are pinned. Without a WHERE clause
        // there is no competing access path: a structural prefix match alone
        // means the index's natural traversal yields the requested order
        // (e.g. window1.test 23.1: index t5ab(a, b) serves key (a, b) with
        // no predicate), so SQLite's EQP shows no temp B-tree even when the
        // columns are nullable and a runtime stabilization sort still runs.
        // `prefer_ordering_scan` extends the no-competing-path exception to
        // WHERE clauses the planner could not use any index for.
        if pinned_columns == 0
            && where_clause.is_some()
            && !(prefer_ordering_scan && chosen.is_none())
        {
            continue; // No leading pin → no EQP exception applies.
        }

        // ORDER BY must structurally fit (pinned-prefix + trailing-index suffix).
        if can_use_index_for_order_by_with_pinned(
            order_by,
            &index_metadata.columns,
            pinned_columns,
        ) {
            return Some(index_name.clone());
        }
    }

    None
}

/// Check if an index can be used for ORDER BY, accounting for pinned prefix columns
///
/// When a query has equality predicates on leading index columns (e.g., WHERE a = 1 AND b = 2),
/// those columns are "pinned" and the index is effectively sorted by the remaining columns.
/// This function skips over the pinned columns and checks if the ORDER BY matches.
///
/// **Pinned ORDER BY items**: ORDER BY can also include the pinned columns themselves —
/// e.g., `WHERE a IN (1,2,3) ORDER BY a, b`. Within each value of the IN-list, the
/// remaining index columns are sorted; the pinned ORDER BY items are constants within
/// the scan group, so they sort trivially. We skip leading ORDER BY items whose column
/// matches one of the pinned leading index columns and only require the remaining
/// ORDER BY items to align with the post-pinned index suffix.
///
/// For example, with index (a, b, c):
/// - WHERE a = 1 ORDER BY b, c → can use index (skip 1 pinned column)
/// - WHERE a = 1 AND b = 2 ORDER BY c → can use index (skip 2 pinned columns)
/// - WHERE a = 1 ORDER BY a, b, c → can use index (a is pinned, skip from ORDER BY)
/// - WHERE a IN (1,2,3) ORDER BY a, b → can use index (a is pinned)
/// - WHERE a = 1 ORDER BY c → cannot use index (b must come before c)
pub(crate) fn can_use_index_for_order_by_with_pinned(
    order_items: &[vibesql_ast::OrderByItem],
    index_columns: &[vibesql_ast::IndexColumn],
    pinned_columns: usize,
) -> bool {
    // Skip pinned columns at the head of the index — they're constants within the scan.
    let remaining_index_columns = &index_columns[pinned_columns..];

    // Skip leading ORDER BY items that refer to one of the pinned index columns.
    // These are constants within each scan group, so the index trivially satisfies
    // their ordering. We accept any prefix of pinned-column references in any order
    // (since each is a single value within the group, their relative order is moot).
    let pinned_index_slice = &index_columns[..pinned_columns];
    let pinned_skip = order_items
        .iter()
        .take_while(|item| order_item_matches_pinned_column(item, pinned_index_slice))
        .count();
    let remaining_order_items = &order_items[pinned_skip..];

    // If every ORDER BY item was a pinned column reference, the index trivially
    // satisfies the ORDER BY (each scan group has one effective value per pinned col).
    if remaining_order_items.is_empty() {
        return !order_items.is_empty();
    }

    // ORDER BY must not have more columns than remaining index columns
    if remaining_order_items.len() > remaining_index_columns.len() {
        return false;
    }

    // Check if ORDER BY matches index direction or is completely reversed
    // (allowing reverse scan to satisfy DESC ordering)
    let mut all_match = true;
    let mut all_reversed = true;

    // Check each ORDER BY column against corresponding index column
    for (order_item, index_col) in
        remaining_order_items.iter().zip(remaining_index_columns.iter())
    {
        // For expression indexes, we need to check if ORDER BY uses the same expression
        // For column indexes, ORDER BY must be a simple column reference
        match index_col {
            vibesql_ast::IndexColumn::Column { column_name, .. } => {
                // ORDER BY expression must be a simple column reference
                let order_col_name = match &order_item.expr {
                    Expression::ColumnRef(col_id)
                        if col_id.schema_canonical().is_none()
                            && col_id.table_canonical().is_none() =>
                    {
                        col_id.column_canonical()
                    }
                    _ => return false, // Complex expressions not supported for column indexes
                };

                // Column names must match (case-insensitive due to SQL identifier normalization)
                if !order_col_name.eq_ignore_ascii_case(column_name) {
                    return false;
                }
            }
            vibesql_ast::IndexColumn::Expression { expr: index_expr, .. } => {
                // For expression indexes, ORDER BY must use the exact same expression
                // e.g., ORDER BY LOWER(name) can use index on LOWER(name)
                if !expressions_match(&order_item.expr, index_expr) {
                    return false;
                }
            }
        }

        // Check sort directions
        let directions_match = order_item.direction == index_col.direction();
        let directions_opposite = matches!(
            (&order_item.direction, &index_col.direction()),
            (vibesql_ast::OrderDirection::Asc, vibesql_ast::OrderDirection::Desc)
                | (vibesql_ast::OrderDirection::Desc, vibesql_ast::OrderDirection::Asc)
        );

        if !directions_match {
            all_match = false;
        }
        if !directions_opposite {
            all_reversed = false;
        }
    }

    // Accept if all directions match OR all are reversed (reverse scan)
    all_match || all_reversed
}

/// Count how many leading index columns are pinned by equality predicates
///
/// A column is "pinned" if there's an equality predicate (col = value) in the WHERE clause.
/// For expression indexes, an expression is "pinned" if there's an equality predicate
/// (expr = value) where expr matches the indexed expression.
/// Returns the number of consecutive leading index columns that are pinned.
pub(crate) fn count_pinned_index_columns(
    where_clause: Option<&Expression>,
    index_columns: &[vibesql_ast::IndexColumn],
) -> usize {
    let where_clause = match where_clause {
        Some(expr) => expr,
        None => return 0,
    };

    // Collect all columns that have equality predicates
    let mut pinned_columns = std::collections::HashSet::new();
    collect_equality_columns(where_clause, &mut pinned_columns);

    // Collect all expressions that have equality predicates (for expression indexes)
    let mut pinned_expressions: Vec<&Expression> = Vec::new();
    collect_equality_expressions(where_clause, &mut pinned_expressions);

    // Count consecutive leading index columns that are pinned
    let mut count = 0;
    for index_col in index_columns {
        let is_pinned = match index_col {
            vibesql_ast::IndexColumn::Column { column_name, .. } => {
                // Check if this column is pinned (case-insensitive match)
                pinned_columns.iter().any(|c| c.eq_ignore_ascii_case(column_name))
            }
            vibesql_ast::IndexColumn::Expression { expr: index_expr, .. } => {
                // Check if any pinned expression matches the indexed expression
                pinned_expressions.iter().any(|e| expressions_match(e, index_expr))
            }
        };

        if is_pinned {
            count += 1;
        } else {
            break; // Stop at first non-pinned column
        }
    }
    count
}

/// Collect all column names that have equality predicates (col = literal or col IS literal)
///
/// Also recognizes positive IN-list (`col IN (literal-list)`) and IN-subquery
/// (`col IN (SELECT ...)`) as pinning the column. SQLite treats both as equality
/// predicates for index-selection purposes — they restrict the column to a finite
/// set of values, which is sufficient for the planner to skip the column when
/// matching ORDER BY against trailing index columns.
///
/// Negated IN (`NOT IN`) is NOT pinning. For an IN-list, all values must be
/// literals (or parameter placeholders) — a column reference inside the list
/// would make the IN equivalent to a join condition, not an equality.
fn collect_equality_columns(expr: &Expression, columns: &mut std::collections::HashSet<String>) {
    match expr {
        Expression::BinaryOp { left, op, right } => {
            match op {
                vibesql_ast::BinaryOperator::Equal => {
                    // Check for column = literal pattern
                    if let Expression::ColumnRef(col_id) = &**left {
                        if is_literal(right) {
                            columns.insert(col_id.column_canonical().to_uppercase());
                        }
                    }
                    if let Expression::ColumnRef(col_id) = &**right {
                        if is_literal(left) {
                            columns.insert(col_id.column_canonical().to_uppercase());
                        }
                    }
                }
                vibesql_ast::BinaryOperator::And => {
                    // Recurse into both sides of AND
                    collect_equality_columns(left, columns);
                    collect_equality_columns(right, columns);
                }
                _ => {}
            }
        }
        // Handle IS (NULL-safe equals): negated=true means "IS NOT DISTINCT FROM" = "IS"
        Expression::IsDistinctFrom { left, right, negated: true } => {
            // Check for column IS literal pattern
            if let Expression::ColumnRef(col_id) = &**left {
                if is_literal(right) {
                    columns.insert(col_id.column_canonical().to_uppercase());
                }
            }
            if let Expression::ColumnRef(col_id) = &**right {
                if is_literal(left) {
                    columns.insert(col_id.column_canonical().to_uppercase());
                }
            }
        }
        // IN-list with literal values: `col IN (1, 2, 3)` pins the column.
        // Empty lists, NOT IN, and lists containing non-literal expressions are
        // excluded since they don't behave as a finite-set equality.
        Expression::InList { expr: target, values, negated: false } => {
            if let Expression::ColumnRef(col_id) = &**target {
                if !values.is_empty() && values.iter().all(is_literal) {
                    columns.insert(col_id.column_canonical().to_uppercase());
                }
            }
        }
        // IN-subquery: `col IN (SELECT ...)` pins the column. The subquery is
        // resolved at execution time to a finite set of values, equivalent for
        // index-selection purposes to an IN-list.
        Expression::In { expr: target, negated: false, .. } => {
            if let Expression::ColumnRef(col_id) = &**target {
                columns.insert(col_id.column_canonical().to_uppercase());
            }
        }
        // Recurse into Conjunction (AND)
        Expression::Conjunction(exprs) => {
            for e in exprs {
                collect_equality_columns(e, columns);
            }
        }
        _ => {}
    }
}

/// Collect all expressions that have equality predicates (expr = literal)
///
/// This is used for expression index support. It collects non-column expressions
/// that are compared to literals in equality predicates.
///
/// Like `collect_equality_columns`, this also treats positive `expr IN (literal-list)`
/// and `expr IN (SELECT ...)` as pinning the expression for matching against
/// expression-index columns.
fn collect_equality_expressions<'a>(expr: &'a Expression, expressions: &mut Vec<&'a Expression>) {
    match expr {
        Expression::BinaryOp { left, op, right } => {
            match op {
                vibesql_ast::BinaryOperator::Equal => {
                    // Check for expr = literal pattern (where expr is not a simple column)
                    if is_literal(right) && !matches!(&**left, Expression::ColumnRef(_)) {
                        expressions.push(left);
                    }
                    if is_literal(left) && !matches!(&**right, Expression::ColumnRef(_)) {
                        expressions.push(right);
                    }
                }
                vibesql_ast::BinaryOperator::And => {
                    // Recurse into both sides of AND
                    collect_equality_expressions(left, expressions);
                    collect_equality_expressions(right, expressions);
                }
                _ => {}
            }
        }
        // Handle IS (NULL-safe equals): negated=true means "IS NOT DISTINCT FROM" = "IS"
        Expression::IsDistinctFrom { left, right, negated: true } => {
            // Check for expr IS literal pattern
            if is_literal(right) && !matches!(&**left, Expression::ColumnRef(_)) {
                expressions.push(left);
            }
            if is_literal(left) && !matches!(&**right, Expression::ColumnRef(_)) {
                expressions.push(right);
            }
        }
        // IN-list with literal values: `expr IN (1, 2, 3)` pins the expression.
        // Restricted to non-column expressions (column equality is collected via
        // `collect_equality_columns`) and lists where every element is a literal.
        Expression::InList { expr: target, values, negated: false } => {
            if !matches!(&**target, Expression::ColumnRef(_))
                && !values.is_empty()
                && values.iter().all(is_literal)
            {
                expressions.push(target);
            }
        }
        // IN-subquery: `expr IN (SELECT ...)` pins the expression for non-column targets.
        Expression::In { expr: target, negated: false, .. } => {
            if !matches!(&**target, Expression::ColumnRef(_)) {
                expressions.push(target);
            }
        }
        // Recurse into Conjunction (AND)
        Expression::Conjunction(exprs) => {
            for e in exprs {
                collect_equality_expressions(e, expressions);
            }
        }
        _ => {}
    }
}

/// Cost-based index selection using statistics
///
/// This function uses table and column statistics to make intelligent decisions
/// about whether to use an index scan or a table scan. It estimates the cost of
/// both access methods and chooses the cheaper one.
///
/// # Arguments
/// * `table_name` - Name of the table being queried
/// * `where_clause` - Optional WHERE clause predicate
/// * `order_by` - Optional ORDER BY clause
/// * `database` - Database reference for accessing statistics and indexes
///
/// # Returns
/// - `Some((index_name, sorted_columns))` if cost-based analysis suggests using an index
/// - `None` if table scan is cheaper or no statistics are available
///
/// # Fallback Behavior
/// If statistics are not available or stale, falls back to rule-based selection
/// using `should_use_index_scan()`.
#[allow(clippy::type_complexity)]
pub(crate) fn cost_based_index_selection(
    table_name: &str,
    where_clause: Option<&Expression>,
    order_by: Option<&[vibesql_ast::OrderByItem]>,
    database: &Database,
) -> Option<(String, Option<Vec<(String, vibesql_ast::OrderDirection)>>)> {
    // Get table statistics
    let table = database.get_table(table_name)?;
    let table_stats = table.get_statistics();

    // If no statistics or stale, fall back to rule-based selection
    if table_stats.is_none() || table_stats.as_ref().map(|s| s.needs_refresh()).unwrap_or(false) {
        return should_use_index_scan(table_name, where_clause, order_by, database);
    }

    let table_stats = table_stats.unwrap();
    let cost_estimator = CostEstimator::default();

    // Get all indexes for this table
    let indexes = database.list_indexes_for_table(table_name);
    if indexes.is_empty() {
        return None; // No indexes available
    }

    // Try each index and find the one with best score (pinned columns + cost)
    #[allow(clippy::type_complexity)]
    let mut best_index: Option<(
        String,
        AccessMethod,
        usize,
        Option<Vec<(String, vibesql_ast::OrderDirection)>>,
    )> = None;
    // (index_name, access_method, pinned_count, sorted_columns)
    let mut has_applicable_index_without_stats = false;

    for index_name in &indexes {
        if let Some(index_metadata) = database.get_index(index_name) {
            // Partial indexes are usable only when the query WHERE clause
            // structurally implies the index predicate — see
            // `should_use_index_scan` for the full rationale.
            if !crate::optimizer::predicate_implication::partial_index_usable(
                database,
                index_name,
                where_clause,
            ) {
                if std::env::var("INDEX_SELECT_DEBUG").is_ok() {
                    eprintln!(
                        "[INDEX_SELECT] skipping {} - partial index predicate not implied by query WHERE",
                        index_name
                    );
                }
                continue;
            }

            let first_indexed_column = index_metadata.columns.first()?;

            // Check if this index can be used for WHERE or ORDER BY
            // Supports both column indexes and expression indexes
            let can_use_for_where = where_clause
                .map(|expr| index_column_can_filter(expr, first_indexed_column))
                .unwrap_or(false);

            // Count how many leading index columns are pinned by equality predicates
            let pinned_columns = count_pinned_index_columns(where_clause, &index_metadata.columns);

            // Get column name for stats lookup (only for column indexes)
            let column_name = first_indexed_column.column_name();

            // Debug: trace index selection
            if std::env::var("INDEX_SELECT_DEBUG").is_ok() {
                let col_debug = column_name.unwrap_or("<expression>");
                eprintln!(
                    "[INDEX_SELECT] table={}, index={}, first_col={}, can_use_for_where={}",
                    table_name, index_name, col_debug, can_use_for_where
                );
            }

            let can_use_for_order = if let Some(order_items) = order_by {
                // Check if ORDER BY columns match the index columns (after skipping pinned columns)
                let columns_match = can_use_index_for_order_by_with_pinned(
                    order_items,
                    &index_metadata.columns,
                    pinned_columns,
                );

                // Don't use index for ORDER BY if any non-pinned column is nullable.
                // See `any_order_by_column_nullable` for the pinned-column exception.
                if columns_match
                    && any_order_by_column_nullable(
                        order_items,
                        &table.schema,
                        &index_metadata.columns,
                        pinned_columns,
                    )
                {
                    false
                } else {
                    columns_match
                }
            } else {
                false
            };

            // Skip this index if it can't help with WHERE or ORDER BY
            if !can_use_for_where && !can_use_for_order {
                if std::env::var("INDEX_SELECT_DEBUG").is_ok() {
                    eprintln!(
                        "[INDEX_SELECT] skipping {} - can't use for where or order",
                        index_name
                    );
                }
                continue;
            }

            // Debug: continue trace
            if std::env::var("INDEX_SELECT_DEBUG").is_ok() {
                eprintln!(
                    "[INDEX_SELECT] {} passed where/order check, checking stats...",
                    index_name
                );
            }

            // For expression indexes, we don't have column statistics
            // Fall back to rule-based selection for them
            if first_indexed_column.is_expression() {
                if std::env::var("INDEX_SELECT_DEBUG").is_ok() {
                    eprintln!(
                        "[INDEX_SELECT] {} is expression index, will use rule-based selection",
                        index_name
                    );
                }
                has_applicable_index_without_stats = true;
                continue;
            }

            // Get column statistics for the indexed column (case-insensitive lookup)
            // For expression indexes, we don't have direct column stats - fall back to rule-based
            let col_stats =
                column_name.and_then(|cn| get_column_stats_ignore_case(&table_stats.columns, cn));
            if col_stats.is_none() {
                // Track that we found an applicable index without column stats
                // We'll fall back to rule-based selection if cost-based fails
                if std::env::var("INDEX_SELECT_DEBUG").is_ok() {
                    let col_debug = column_name.unwrap_or("<expression>");
                    eprintln!(
                        "[INDEX_SELECT] {} no column stats for {}, will fallback",
                        index_name, col_debug
                    );
                }
                has_applicable_index_without_stats = true;
                continue; // No stats for this column, try next index
            }
            let col_stats = col_stats.unwrap();
            // At this point, column_name must be Some since we got col_stats
            let column_name = column_name.unwrap();

            if std::env::var("INDEX_SELECT_DEBUG").is_ok() {
                eprintln!("[INDEX_SELECT] {} has column stats for {}", index_name, column_name);
            }

            // Estimate selectivity based on WHERE clause
            let selectivity = if let Some(where_expr) = where_clause {
                estimate_selectivity(where_expr, column_name, col_stats)
            } else {
                1.0 // No WHERE clause means all rows
            };

            // Use cost estimator to decide
            let access_method =
                cost_estimator.choose_access_method(table_stats, Some(col_stats), selectivity);

            if std::env::var("INDEX_SELECT_DEBUG").is_ok() {
                eprintln!(
                    "[INDEX_SELECT] {} selectivity={:.4}, access_method={:?}, is_index_scan={}",
                    index_name,
                    selectivity,
                    access_method,
                    access_method.is_index_scan()
                );
            }

            // Build sorted_columns metadata if ORDER BY can be satisfied
            let sorted_columns = if can_use_for_order {
                let order_items = order_by.unwrap();
                Some(
                    order_items
                        .iter()
                        .map(|item| {
                            // For expression indexes, ORDER BY may use expressions (e.g., length(a))
                            // We use to_sql() to convert the expression to a string representation
                            // The actual string is only used for metadata; the important part
                            // is the direction for determining scan order
                            let col_name = match &item.expr {
                                Expression::ColumnRef(col_id) => {
                                    col_id.column_canonical().to_string()
                                }
                                expr => expr.to_sql(),
                            };
                            (col_name, item.direction.clone())
                        })
                        .collect(),
                )
            } else {
                None
            };

            // Track the best index
            // Priority: more pinned columns (better filtering) > lower cost
            if access_method.is_index_scan() {
                let is_better = match &best_index {
                    None => true,
                    Some((_, best_method, best_pinned, _)) => {
                        // More pinned columns is always better (filters more rows)
                        if pinned_columns > *best_pinned {
                            true
                        } else if pinned_columns == *best_pinned {
                            // Same pinned count: prefer lower cost
                            access_method.cost() < best_method.cost()
                        } else {
                            false
                        }
                    }
                };

                if is_better {
                    best_index =
                        Some((index_name.clone(), access_method, pinned_columns, sorted_columns));
                }
            } else if selectivity < 0.40 && can_use_for_where {
                // Cost-based chose table scan, but selectivity is good enough for index
                // The cost model may be too conservative for in-memory/prefix scans
                // Common fallback values: 0.33 (single predicate), 0.1089 (two predicates)
                // Fall back to rule-based selection for selective queries
                if std::env::var("INDEX_SELECT_DEBUG").is_ok() {
                    eprintln!(
                        "[INDEX_SELECT] {} selectivity={:.4} good, falling back to rule-based",
                        index_name, selectivity
                    );
                }
                return should_use_index_scan(table_name, where_clause, order_by, database);
            }
        }
    }

    // Return the best index if we found one
    if let Some((index_name, _, _, sorted_columns)) = best_index {
        if std::env::var("INDEX_SELECT_DEBUG").is_ok() {
            eprintln!("[INDEX_SELECT] selected best_index={} for table={}", index_name, table_name);
        }
        return Some((index_name, sorted_columns));
    }

    // If we have applicable indexes but no column stats, fall back to rule-based selection
    // This ensures we use indexes even when statistics are incomplete
    if has_applicable_index_without_stats {
        if std::env::var("INDEX_SELECT_DEBUG").is_ok() {
            eprintln!("[INDEX_SELECT] falling back to rule-based for table={}", table_name);
        }
        return should_use_index_scan(table_name, where_clause, order_by, database);
    }

    // No applicable indexes found
    if std::env::var("INDEX_SELECT_DEBUG").is_ok() {
        eprintln!("[INDEX_SELECT] no index selected for table={}", table_name);
    }
    None
}

/// Estimate selectivity of a predicate on a specific column
///
/// Uses column statistics to estimate what fraction of rows will match the predicate.
/// Returns a value between 0.0 (no rows) and 1.0 (all rows).
pub(crate) fn estimate_selectivity(
    expr: &Expression,
    column_name: &str,
    col_stats: &vibesql_storage::statistics::ColumnStatistics,
) -> f64 {
    match expr {
        Expression::BinaryOp { left, op, right } => {
            match op {
                vibesql_ast::BinaryOperator::Equal => {
                    // Check if this is a predicate on our column (case-insensitive)
                    // For literal values, use actual statistics
                    if let (Expression::ColumnRef(col_id), Expression::Literal(value)) =
                        (&**left, &**right)
                    {
                        if col_id.column_canonical().eq_ignore_ascii_case(column_name) {
                            return col_stats.estimate_eq_selectivity(value);
                        }
                    }
                    if let (Expression::Literal(value), Expression::ColumnRef(col_id)) =
                        (&**left, &**right)
                    {
                        if col_id.column_canonical().eq_ignore_ascii_case(column_name) {
                            return col_stats.estimate_eq_selectivity(value);
                        }
                    }
                    // For placeholder parameters, estimate using 1/n_distinct
                    // This is more accurate than the generic 0.33 fallback for equality predicates
                    let left_is_col = is_column_reference(left, column_name);
                    let right_is_col = is_column_reference(right, column_name);
                    let left_is_lit = is_literal(left);
                    let right_is_lit = is_literal(right);

                    if (left_is_col && right_is_lit) || (left_is_lit && right_is_col) {
                        // Use 1/n_distinct as selectivity estimate for equality with parameter
                        if col_stats.n_distinct > 0 {
                            return 1.0 / col_stats.n_distinct as f64;
                        }
                    }
                    0.33 // Default fallback
                }
                vibesql_ast::BinaryOperator::GreaterThan
                | vibesql_ast::BinaryOperator::GreaterThanOrEqual
                | vibesql_ast::BinaryOperator::LessThan
                | vibesql_ast::BinaryOperator::LessThanOrEqual => {
                    // Range predicates (case-insensitive column comparison)
                    if let (Expression::ColumnRef(col_id), Expression::Literal(value)) =
                        (&**left, &**right)
                    {
                        if col_id.column_canonical().eq_ignore_ascii_case(column_name) {
                            let op_str = match op {
                                vibesql_ast::BinaryOperator::GreaterThan => ">",
                                vibesql_ast::BinaryOperator::GreaterThanOrEqual => ">=",
                                vibesql_ast::BinaryOperator::LessThan => "<",
                                vibesql_ast::BinaryOperator::LessThanOrEqual => "<=",
                                _ => unreachable!(),
                            };
                            return col_stats.estimate_range_selectivity(value, op_str);
                        }
                    }
                    // For placeholder parameters, use a conservative 0.25 (assume filtering ~75% of
                    // rows)
                    if (is_column_reference(left, column_name) && is_literal(right))
                        || (is_literal(left) && is_column_reference(right, column_name))
                    {
                        return 0.25;
                    }
                    0.33 // Default fallback
                }
                vibesql_ast::BinaryOperator::And => {
                    // For AND, multiply selectivities (assuming independence)
                    let left_sel = estimate_selectivity(left, column_name, col_stats);
                    let right_sel = estimate_selectivity(right, column_name, col_stats);
                    left_sel * right_sel
                }
                vibesql_ast::BinaryOperator::Or => {
                    // For OR, use formula: P(A OR B) = P(A) + P(B) - P(A AND B)
                    // Assuming independence: P(A OR B) = P(A) + P(B) - P(A)*P(B)
                    let left_sel = estimate_selectivity(left, column_name, col_stats);
                    let right_sel = estimate_selectivity(right, column_name, col_stats);
                    left_sel + right_sel - (left_sel * right_sel)
                }
                _ => 0.33, // Default fallback for other operators
            }
        }
        Expression::Between { expr, low, high, negated: _, symmetric: _ } => {
            if let Expression::ColumnRef(col_id) = &**expr {
                if col_id.column_canonical().eq_ignore_ascii_case(column_name) {
                    // Estimate BETWEEN as: P(col >= low AND col <= high)
                    if let (Expression::Literal(low_val), Expression::Literal(high_val)) =
                        (&**low, &**high)
                    {
                        let low_sel = col_stats.estimate_range_selectivity(low_val, ">=");
                        let high_sel = col_stats.estimate_range_selectivity(high_val, "<=");
                        return low_sel * high_sel; // Assuming independence
                    }
                }
            }
            0.33 // Default fallback
        }
        // IS (NULL-safe equals): negated=true means "IS NOT DISTINCT FROM" = "IS"
        // Treat IS like = for selectivity estimation
        Expression::IsDistinctFrom { left, right, negated: true } => {
            // Check if this is a predicate on our column (case-insensitive)
            // For literal values, use actual statistics
            if let (Expression::ColumnRef(col_id), Expression::Literal(value)) = (&**left, &**right)
            {
                if col_id.column_canonical().eq_ignore_ascii_case(column_name) {
                    return col_stats.estimate_eq_selectivity(value);
                }
            }
            if let (Expression::Literal(value), Expression::ColumnRef(col_id)) = (&**left, &**right)
            {
                if col_id.column_canonical().eq_ignore_ascii_case(column_name) {
                    return col_stats.estimate_eq_selectivity(value);
                }
            }
            // For placeholder parameters, estimate using 1/n_distinct
            let left_is_col = is_column_reference(left, column_name);
            let right_is_col = is_column_reference(right, column_name);
            let left_is_lit = is_literal(left);
            let right_is_lit = is_literal(right);

            if (left_is_col && right_is_lit) || (left_is_lit && right_is_col) {
                if col_stats.n_distinct > 0 {
                    return 1.0 / col_stats.n_distinct as f64;
                }
            }
            0.33 // Default fallback
        }
        // Handle Conjunction (AND) similarly to BinaryOp::And
        Expression::Conjunction(exprs) => {
            let mut selectivity = 1.0;
            for e in exprs {
                selectivity *= estimate_selectivity(e, column_name, col_stats);
            }
            selectivity
        }
        // Handle Disjunction (OR) similarly to BinaryOp::Or
        Expression::Disjunction(exprs) => {
            let mut selectivity = 0.0;
            let mut product = 1.0;
            for e in exprs {
                let sel = estimate_selectivity(e, column_name, col_stats);
                selectivity += sel;
                product *= sel;
            }
            // P(A OR B OR C) ≈ P(A) + P(B) + P(C) - P(A)*P(B)*P(C) (simplified)
            (selectivity - product).min(1.0)
        }
        _ => 0.33, // Default fallback for unsupported expressions
    }
}

/// Unified index selection that returns IndexScanChoice
///
/// This function first tries regular index selection, and if no suitable index is found,
/// it attempts skip-scan optimization as a fallback. Skip-scan enables using composite
/// indexes when the WHERE clause filters on non-prefix columns.
///
/// # Arguments
/// * `table_name` - Name of the table being queried
/// * `where_clause` - Optional WHERE clause predicate
/// * `order_by` - Optional ORDER BY clause
/// * `database` - Database reference for accessing statistics and indexes
///
/// # Returns
/// - `Some(IndexScanChoice::Regular {...})` if a regular index scan should be used
/// - `Some(IndexScanChoice::SkipScan {...})` if skip-scan is beneficial
/// - `None` if table scan is more appropriate
///
/// # Example
/// ```text
/// // Query: SELECT * FROM sales WHERE date = '2024-01-01'
/// // Index: (region, date) - a composite index
/// //
/// // Regular index selection fails (no filter on 'region' prefix column)
/// // Skip-scan is considered: iterate through distinct 'region' values,
/// // for each region, seek to entries with date = '2024-01-01'
/// //
/// // If skip-scan cost < table scan cost, returns IndexScanChoice::SkipScan
/// ```
pub(crate) fn select_index_scan_method(
    table_name: &str,
    where_clause: Option<&Expression>,
    order_by: Option<&[vibesql_ast::OrderByItem]>,
    database: &Database,
) -> Option<IndexScanChoice> {
    // First, try regular cost-based index selection
    if let Some((index_name, sorted_columns)) =
        cost_based_index_selection(table_name, where_clause, order_by, database)
    {
        return Some(IndexScanChoice::Regular { index_name, sorted_columns });
    }

    // If regular index selection failed and we have a WHERE clause,
    // try skip-scan optimization
    if let Some(where_expr) = where_clause {
        let planner = IndexPlanner::new(database);
        if let Some(plan) = planner.plan_skip_scan(table_name, where_expr) {
            if plan.is_skip_scan {
                if let Some(skip_info) = plan.skip_scan_info {
                    if std::env::var("SKIP_SCAN_DEBUG").is_ok() {
                        eprintln!(
                            "[SKIP_SCAN] Selected skip-scan for table={}, index={}, filter_col={}, prefix_cardinality={}",
                            table_name,
                            plan.index_name,
                            skip_info.filter_column,
                            skip_info.prefix_cardinality
                        );
                    }
                    return Some(IndexScanChoice::SkipScan {
                        index_name: plan.index_name,
                        skip_scan_info: skip_info,
                    });
                }
            }
        }
    }

    // No index or skip-scan option found
    None
}

#[cfg(test)]
mod tests {
    use vibesql_ast::BinaryOperator;
    use vibesql_types::SqlValue;

    use super::*;

    #[test]
    fn test_expression_filters_column_simple() {
        let expr = Expression::BinaryOp {
            op: BinaryOperator::Equal,
            left: Box::new(Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
                "age", false,
            ))),
            right: Box::new(Expression::Literal(SqlValue::Integer(25))),
        };

        assert!(expression_filters_column(&expr, "age"));
        assert!(!expression_filters_column(&expr, "name"));
    }

    #[test]
    fn test_expression_filters_column_and() {
        let expr = Expression::BinaryOp {
            op: BinaryOperator::And,
            left: Box::new(Expression::BinaryOp {
                op: BinaryOperator::GreaterThan,
                left: Box::new(Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
                    "age", false,
                ))),
                right: Box::new(Expression::Literal(SqlValue::Integer(18))),
            }),
            right: Box::new(Expression::BinaryOp {
                op: BinaryOperator::Equal,
                left: Box::new(Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
                    "city", false,
                ))),
                right: Box::new(Expression::Literal(SqlValue::Varchar(arcstr::ArcStr::from(
                    "Boston",
                )))),
            }),
        };

        assert!(expression_filters_column(&expr, "age"));
        assert!(expression_filters_column(&expr, "city"));
        assert!(!expression_filters_column(&expr, "name"));
    }

    #[test]
    fn test_is_column_reference() {
        let expr = Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple("age", false));

        assert!(is_column_reference(&expr, "age"));
        assert!(!is_column_reference(&expr, "name"));
    }

    #[test]
    fn test_is_column_reference_case_insensitive() {
        // SQL parser normalizes unquoted identifiers to uppercase
        // but index columns might be lowercase
        let expr = Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple("I_ID", false)); // Uppercase from parser

        // Should match regardless of case
        assert!(is_column_reference(&expr, "i_id")); // lowercase
        assert!(is_column_reference(&expr, "I_ID")); // exact match
        assert!(is_column_reference(&expr, "I_id")); // mixed case
        assert!(!is_column_reference(&expr, "other_column"));
    }

    #[test]
    fn test_expression_filters_column_case_insensitive() {
        // WHERE I_ID = 42 (uppercase from parser)
        let expr = Expression::BinaryOp {
            op: BinaryOperator::Equal,
            left: Box::new(Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
                "I_ID", false,
            ))),
            right: Box::new(Expression::Literal(SqlValue::Integer(42))),
        };

        // Should match lowercase index column name
        assert!(expression_filters_column(&expr, "i_id"));
        assert!(expression_filters_column(&expr, "I_ID"));
        assert!(!expression_filters_column(&expr, "other"));
    }

    /// Helper for tests: build a column-only IndexColumn list from names.
    fn idx_cols(names: &[&str]) -> Vec<vibesql_ast::IndexColumn> {
        names
            .iter()
            .map(|n| vibesql_ast::IndexColumn::Column {
                column_name: n.to_string(),
                direction: vibesql_ast::OrderDirection::Asc,
                prefix_length: None,
            })
            .collect()
    }

    /// Helper: build `WHERE col IN (literal-list)` expression.
    fn in_list_expr(col: &str, values: Vec<SqlValue>) -> Expression {
        Expression::InList {
            expr: Box::new(Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
                col, false,
            ))),
            values: values.into_iter().map(Expression::Literal).collect(),
            negated: false,
        }
    }

    #[test]
    fn test_count_pinned_columns_in_list() {
        // WHERE a IN (1, 2, 3) — column `a` should be pinned for index (a, b).
        let where_expr =
            in_list_expr("a", vec![SqlValue::Integer(1), SqlValue::Integer(2), SqlValue::Integer(3)]);
        let index = idx_cols(&["a", "b"]);
        assert_eq!(count_pinned_index_columns(Some(&where_expr), &index), 1);
    }

    #[test]
    fn test_count_pinned_columns_not_in_list_excluded() {
        // WHERE a NOT IN (1, 2) — column `a` should NOT be pinned.
        let where_expr = Expression::InList {
            expr: Box::new(Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
                "a", false,
            ))),
            values: vec![
                Expression::Literal(SqlValue::Integer(1)),
                Expression::Literal(SqlValue::Integer(2)),
            ],
            negated: true,
        };
        let index = idx_cols(&["a", "b"]);
        assert_eq!(count_pinned_index_columns(Some(&where_expr), &index), 0);
    }

    #[test]
    fn test_count_pinned_columns_empty_in_list_excluded() {
        // WHERE a IN () — no values means no effective pinning.
        let where_expr = Expression::InList {
            expr: Box::new(Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
                "a", false,
            ))),
            values: vec![],
            negated: false,
        };
        let index = idx_cols(&["a", "b"]);
        assert_eq!(count_pinned_index_columns(Some(&where_expr), &index), 0);
    }

    #[test]
    fn test_count_pinned_columns_in_list_with_non_literal_excluded() {
        // WHERE a IN (1, b) — non-literal in list disqualifies pinning.
        let where_expr = Expression::InList {
            expr: Box::new(Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
                "a", false,
            ))),
            values: vec![
                Expression::Literal(SqlValue::Integer(1)),
                Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple("b", false)),
            ],
            negated: false,
        };
        let index = idx_cols(&["a", "b"]);
        assert_eq!(count_pinned_index_columns(Some(&where_expr), &index), 0);
    }

    #[test]
    fn test_count_pinned_columns_in_list_with_null_still_pins() {
        // WHERE a IN (1, NULL, 3) — list contains NULL but the IN-list still pins.
        // (NULL never compares true, so rows with a IS NULL never match — equivalent
        // to filtering them out, which preserves the pinning semantic.)
        let where_expr = Expression::InList {
            expr: Box::new(Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
                "a", false,
            ))),
            values: vec![
                Expression::Literal(SqlValue::Integer(1)),
                Expression::Literal(SqlValue::Null),
                Expression::Literal(SqlValue::Integer(3)),
            ],
            negated: false,
        };
        let index = idx_cols(&["a", "b"]);
        // Note: `is_literal` returns false for NULL, so the IN-list is rejected as
        // pinning. This is conservative but correct — current behavior.
        assert_eq!(count_pinned_index_columns(Some(&where_expr), &index), 0);
    }

    #[test]
    fn test_count_pinned_columns_in_list_combined_with_equality() {
        // WHERE a IN (1, 2) AND b = 5 — both `a` and `b` should be pinned.
        let in_a = in_list_expr("a", vec![SqlValue::Integer(1), SqlValue::Integer(2)]);
        let eq_b = Expression::BinaryOp {
            op: BinaryOperator::Equal,
            left: Box::new(Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple(
                "b", false,
            ))),
            right: Box::new(Expression::Literal(SqlValue::Integer(5))),
        };
        let where_expr = Expression::BinaryOp {
            op: BinaryOperator::And,
            left: Box::new(in_a),
            right: Box::new(eq_b),
        };
        let index = idx_cols(&["a", "b", "c"]);
        assert_eq!(count_pinned_index_columns(Some(&where_expr), &index), 2);
    }

    #[test]
    fn test_can_use_index_for_order_by_with_pinned_in_list_leading() {
        // Index (a, b), WHERE a IN (1,2,3), ORDER BY a, b — should be usable.
        // The leading ORDER BY item `a` is pinned; the trailing `b` aligns with
        // the remaining index column.
        let order_by = vec![
            vibesql_ast::OrderByItem {
                expr: Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple("a", false)),
                direction: vibesql_ast::OrderDirection::Asc,
                nulls_order: None,
            },
            vibesql_ast::OrderByItem {
                expr: Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple("b", false)),
                direction: vibesql_ast::OrderDirection::Asc,
                nulls_order: None,
            },
        ];
        let index = idx_cols(&["a", "b"]);
        // pinned_columns = 1 (a is pinned); ORDER BY [a, b] should still be matched.
        assert!(can_use_index_for_order_by_with_pinned(&order_by, &index, 1));
    }

    #[test]
    fn test_can_use_index_for_order_by_all_pinned() {
        // Index (a, b), WHERE a = 1 AND b = 2, ORDER BY a, b — every ORDER BY item
        // is pinned, so the index trivially satisfies the order.
        let order_by = vec![
            vibesql_ast::OrderByItem {
                expr: Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple("a", false)),
                direction: vibesql_ast::OrderDirection::Asc,
                nulls_order: None,
            },
            vibesql_ast::OrderByItem {
                expr: Expression::ColumnRef(vibesql_ast::ColumnIdentifier::simple("b", false)),
                direction: vibesql_ast::OrderDirection::Asc,
                nulls_order: None,
            },
        ];
        let index = idx_cols(&["a", "b"]);
        assert!(can_use_index_for_order_by_with_pinned(&order_by, &index, 2));
    }
}