stoolap 0.4.0

// Copyright 2025 Stoolap Contributors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! MVCC Table implementation
//!
//! Provides MVCC isolation for table operations.
//!

use rustc_hash::{FxHashMap, FxHashSet};
use std::sync::{Arc, RwLock};

use crate::common::{CompactArc, I64Set};
use crate::core::{
    DataType, Error, IndexType, Result, Row, RowIdVec, RowVec, Schema, SchemaColumn, Value,
};
use crate::storage::expression::logical::OrExpr;
use crate::storage::expression::Expression;
use crate::storage::index::{BTreeIndex, BitmapIndex, HashIndex, HnswIndex, MultiColumnIndex};
use crate::storage::mvcc::scanner::MVCCScanner;
use crate::storage::mvcc::{TransactionVersionStore, VersionStore};
use crate::storage::traits::{Index, QueryResult, ScanPlan, Scanner, Table};
use crate::storage::MemoryResult;

/// MVCC Table wrapper that provides MVCC isolation for tables
pub struct MVCCTable {
    /// Transaction ID
    txn_id: i64,
    /// Reference to the version store
    version_store: Arc<VersionStore>,
    /// Transaction-local version store (shared between multiple MVCCTable instances for same txn+table)
    txn_versions: Arc<RwLock<TransactionVersionStore>>,
    /// Cached schema for returning references (Arc clone from version_store - O(1) instead of cloning)
    cached_schema: CompactArc<Schema>,
}

impl MVCCTable {
    /// Creates a new MVCC table with an owned transaction version store
    /// (wraps it in Arc<RwLock> internally)
    pub fn new(
        txn_id: i64,
        version_store: Arc<VersionStore>,
        txn_versions: TransactionVersionStore,
    ) -> Self {
        // CompactArc clone - O(1) reference count increment, not full schema clone
        let cached_schema = version_store.schema().clone();
        Self {
            txn_id,
            version_store,
            txn_versions: Arc::new(RwLock::new(txn_versions)),
            cached_schema,
        }
    }

    /// Creates a new MVCC table with a shared transaction version store
    /// (used by the engine's get_table_for_transaction to share stores)
    pub fn new_with_shared_store(
        txn_id: i64,
        version_store: Arc<VersionStore>,
        txn_versions: Arc<RwLock<TransactionVersionStore>>,
    ) -> Self {
        // CompactArc clone - O(1) reference count increment, not full schema clone
        let cached_schema = version_store.schema().clone();
        Self {
            txn_id,
            version_store,
            txn_versions,
            cached_schema,
        }
    }

    /// Returns the transaction ID
    pub fn txn_id(&self) -> i64 {
        self.txn_id
    }

    /// Returns a reference to the version store
    pub fn version_store(&self) -> &Arc<VersionStore> {
        &self.version_store
    }

    /// Returns a reference to the shared transaction version store
    pub fn txn_versions(&self) -> &Arc<RwLock<TransactionVersionStore>> {
        &self.txn_versions
    }

    /// Auto-selects the optimal index type based on column data types
    ///
    /// # Type-Based Index Selection Rules:
    /// - TEXT/JSON columns → Hash index (avoids O(strlen) comparisons per B-tree node)
    /// - BOOLEAN columns → Bitmap index (only 2 values, fast AND/OR operations)
    /// - INTEGER/FLOAT/TIMESTAMP columns → BTree index (supports range queries)
    /// - Mixed types → BTree as safe default
    ///
    /// For multi-column indexes, the first column's type determines the index type
    /// unless there's a BOOLEAN (which always gets Bitmap for AND/OR optimization).
    fn auto_select_index_type(data_types: &[DataType]) -> IndexType {
        if data_types.is_empty() {
            return IndexType::BTree;
        }

        // Check if any column is BOOLEAN - use Bitmap for fast AND/OR
        let has_boolean = data_types.iter().any(|dt| matches!(dt, DataType::Boolean));
        if has_boolean && data_types.len() == 1 {
            return IndexType::Bitmap;
        }

        // Check the primary (first) column type
        match data_types[0] {
            // TEXT/JSON - use Hash for O(1) lookups, avoid O(strlen) comparisons
            DataType::Text | DataType::Json => IndexType::Hash,

            // BOOLEAN - use Bitmap for fast AND/OR/NOT operations
            DataType::Boolean => IndexType::Bitmap,

            // Numeric types - use BTree for range query support
            DataType::Integer | DataType::Float | DataType::Timestamp => IndexType::BTree,

            // Vector columns - use HNSW for approximate nearest neighbor search
            DataType::Vector => IndexType::Hnsw,

            // NULL type - use BTree as safe default
            DataType::Null => IndexType::BTree,
        }
    }

    /// Gets the current auto-increment value
    pub fn get_current_auto_increment_value(&self) -> i64 {
        self.version_store.get_auto_increment_counter()
    }

    /// Normalize a row to match the current schema
    ///
    /// This handles schema evolution (ALTER TABLE ADD/DROP COLUMN):
    /// - If row has fewer columns than schema, append default values (or NULLs) for missing columns
    /// - If row has more columns than schema, truncate the row
    fn normalize_row_to_schema(&self, mut row: Row, schema: &Schema) -> Row {
        let schema_cols = schema.columns.len();
        let row_cols = row.len();

        if row_cols < schema_cols {
            // Row has fewer columns - add default values (or NULLs) for new columns
            for i in row_cols..schema_cols {
                let col = &schema.columns[i];
                // Use pre-computed default value if available, otherwise use NULL
                if let Some(ref default_val) = col.default_value {
                    row.push(default_val.clone());
                } else {
                    row.push(Value::null(col.data_type));
                }
            }
        } else if row_cols > schema_cols {
            // Row has more columns - truncate (columns were dropped)
            row.truncate(schema_cols);
        }

        row
    }

    /// Try to extract a primary key lookup from the expression
    ///
    /// Returns Some(row_id) if the expression is a simple equality on the PK column
    fn try_pk_lookup(&self, expr: &dyn Expression, schema: &Schema) -> Option<i64> {
        use crate::core::Operator;

        // Get PK column info
        let pk_indices = schema.primary_key_indices();
        if pk_indices.len() != 1 {
            return None; // Only support single-column PK for now
        }
        let pk_col_idx = pk_indices[0];
        let pk_col = &schema.columns[pk_col_idx];

        // Use the new get_comparison_info method (no downcasting required)
        let (col_name, operator, value) = expr.get_comparison_info()?;

        // Check if it's an equality on the PK column (case-insensitive comparison)
        if !col_name.eq_ignore_ascii_case(&pk_col.name) || operator != Operator::Eq {
            return None;
        }

        // Get the integer value (PKs are always integers in our system)
        match value {
            Value::Integer(i) => Some(*i),
            _ => None,
        }
    }

    /// Try to identify a PK range lookup (WHERE id >= X AND id < Y)
    /// Returns Some(RowIdVec) if this is a PK range query (pooled for efficient memory reuse)
    fn try_pk_range_lookup(&self, expr: &dyn Expression, schema: &Schema) -> Option<RowIdVec> {
        use crate::core::Operator;

        // Get PK column info
        let pk_indices = schema.primary_key_indices();
        if pk_indices.len() != 1 {
            return None;
        }
        let pk_col_idx = pk_indices[0];
        let pk_col_name = &schema.columns[pk_col_idx].name;

        // Check for AND with two comparisons on PK
        let and_operands = expr.get_and_operands()?;
        if and_operands.len() != 2 {
            return None;
        }

        let (info1, info2) = (
            and_operands[0].get_comparison_info(),
            and_operands[1].get_comparison_info(),
        );

        let ((col1, op1, val1), (col2, op2, val2)) = match (info1, info2) {
            (Some(a), Some(b)) => (a, b),
            _ => return None,
        };

        // Both must be on PK column
        if !col1.eq_ignore_ascii_case(pk_col_name) || !col2.eq_ignore_ascii_case(pk_col_name) {
            return None;
        }

        // Extract integer values
        let (v1, v2) = match (val1, val2) {
            (Value::Integer(a), Value::Integer(b)) => (*a, *b),
            _ => return None,
        };

        // Identify lower and upper bounds
        let (min_id, min_inclusive, max_id, max_inclusive) = match (op1, op2) {
            (Operator::Gte, Operator::Lt) => (v1, true, v2, false),
            (Operator::Gte, Operator::Lte) => (v1, true, v2, true),
            (Operator::Gt, Operator::Lt) => (v1, false, v2, false),
            (Operator::Gt, Operator::Lte) => (v1, false, v2, true),
            (Operator::Lt, Operator::Gte) => (v2, true, v1, false),
            (Operator::Lt, Operator::Gt) => (v2, false, v1, false),
            (Operator::Lte, Operator::Gte) => (v2, true, v1, true),
            (Operator::Lte, Operator::Gt) => (v2, false, v1, true),
            _ => return None,
        };

        // Adjust for inclusive/exclusive bounds using saturating arithmetic to prevent overflow
        let start = if min_inclusive {
            min_id
        } else {
            min_id.saturating_add(1)
        };
        let end = if max_inclusive {
            max_id
        } else {
            max_id.saturating_sub(1)
        };

        // Check for invalid range (start > end means empty result)
        if start > end {
            return Some(RowIdVec::new());
        }

        // SAFETY: Limit range size to prevent memory explosion
        // For ranges larger than threshold, return None to fall back to index/full scan
        const MAX_PK_RANGE_SIZE: i64 = 100_000;
        let range_size = end.saturating_sub(start).saturating_add(1);
        if range_size > MAX_PK_RANGE_SIZE {
            return None; // Fall back to index scan or full scan
        }

        // Generate row_ids directly (no index needed - PK IS the row_id)
        let mut row_ids = RowIdVec::with_capacity(range_size as usize);
        for id in start..=end {
            row_ids.push(id);
        }
        Some(row_ids)
    }

    /// Try to use an index to filter row IDs
    ///
    /// Returns Some(row_ids) if an index can be used, None otherwise
    /// Returns a pooled RowIdVec for efficient memory reuse.
    #[allow(clippy::only_used_in_recursion)]
    fn try_index_lookup(&self, expr: &dyn Expression, schema: &Schema) -> Option<RowIdVec> {
        use crate::core::Operator;
        use crate::storage::index::intersect_sorted_ids;

        // First, try simple comparison on a single column
        if let Some((col_name, operator, value)) = expr.get_comparison_info() {
            // Skip index for boolean equality - low cardinality (2 values) means ~50% selectivity
            // which makes full scan faster than index lookup + row fetch
            if matches!(value, Value::Boolean(_)) && matches!(operator, Operator::Eq | Operator::Ne)
            {
                return None;
            }

            if let Some(index) = self.version_store.get_index_by_column(col_name) {
                return self.query_index_with_operator(&*index, operator, value);
            }
        }

        // OPTIMIZATION: Handle OR expressions with HYBRID index optimization
        // For (indexed_col = 'a' OR non_indexed_col = 'b'):
        // - Use index for indexed_col operands
        // - Return None only if ALL operands are non-indexed (full scan needed)
        // - If at least one operand uses index but others don't, we still return
        //   the indexed row_ids (the executor will handle memory filtering for others)
        if let Some(or_operands) = expr.get_or_operands() {
            let mut indexed_row_ids: Vec<RowIdVec> = Vec::with_capacity(or_operands.len());
            let mut has_unindexed_operand = false;

            for operand in or_operands {
                // Recursively try index lookup for each OR operand
                if let Some(row_ids) = self.try_index_lookup(operand.as_ref(), schema) {
                    indexed_row_ids.push(row_ids);
                } else {
                    // This operand can't use an index
                    has_unindexed_operand = true;
                }
            }

            // HYBRID OPTIMIZATION: If some operands use indexes but not all,
            // we can't use pure index lookup (would miss rows from unindexed operands).
            // For now, fall back to full scan - the memory filter will handle it.
            // Future: Could return indexed row_ids + flag for partial optimization
            if has_unindexed_operand || indexed_row_ids.is_empty() {
                return None;
            }

            // All operands indexed - return the union of all row IDs
            if indexed_row_ids.len() == 1 {
                return Some(indexed_row_ids.into_iter().next().unwrap());
            }

            // Bitset union: build a bitset for deduplication, collect unique row IDs
            // Cap at 131072 words (~1MB) to prevent OOM with large/sparse row IDs.
            // Negative row IDs (from user-specified PKs) cannot use the bitset path
            // because `id as usize` wraps to a huge value.
            const MAX_BITSET_WORDS: usize = 131_072;
            let mut max_id = 0i64;
            let mut min_id = 0i64;
            for ids in &indexed_row_ids {
                for &id in ids.iter() {
                    if id > max_id {
                        max_id = id;
                    }
                    if id < min_id {
                        min_id = id;
                    }
                }
            }
            let num_words = (max_id as usize / 64) + 1;
            let total_len: usize = indexed_row_ids.iter().map(|v| v.len()).sum();
            if min_id >= 0 && num_words <= MAX_BITSET_WORDS {
                // Fast bitset path
                let mut bits = vec![0u64; num_words];
                let mut result = RowIdVec::with_capacity(total_len);
                for ids in &indexed_row_ids {
                    for &id in ids.iter() {
                        let idx = id as usize;
                        let word = idx / 64;
                        let mask = 1u64 << (idx % 64);
                        if (bits[word] & mask) == 0 {
                            bits[word] |= mask;
                            result.push(id);
                        }
                    }
                }
                return Some(result);
            } else {
                // Fallback: I64Set-based union for large/sparse row IDs
                let mut seen = I64Set::with_capacity(total_len);
                let mut has_i64_min = false;
                let mut result = RowIdVec::with_capacity(total_len);
                for ids in &indexed_row_ids {
                    for &id in ids.iter() {
                        if id == i64::MIN {
                            if !has_i64_min {
                                has_i64_min = true;
                                result.push(id);
                            }
                        } else if seen.insert(id) {
                            result.push(id);
                        }
                    }
                }
                return Some(result);
            }
        }

        // OPTIMIZATION: Detect BETWEEN/range pattern (col >= X AND col <= Y) for single range scan
        // This avoids two separate index scans + sort + intersection
        if let Some(and_operands) = expr.get_and_operands() {
            // Check for range pattern: exactly 2 comparisons on the same indexed column
            if and_operands.len() == 2 {
                if let (Some((col1, op1, val1)), Some((col2, op2, val2))) = (
                    and_operands[0].get_comparison_info(),
                    and_operands[1].get_comparison_info(),
                ) {
                    // Same column?
                    if col1 == col2 {
                        // Check for range pattern: one lower bound + one upper bound
                        let (lower_op, lower_val, upper_op, upper_val) = match (op1, op2) {
                            (Operator::Gt | Operator::Gte, Operator::Lt | Operator::Lte) => {
                                (op1, val1, op2, val2)
                            }
                            (Operator::Lt | Operator::Lte, Operator::Gt | Operator::Gte) => {
                                (op2, val2, op1, val1)
                            }
                            _ => (Operator::Eq, val1, Operator::Eq, val2), // Not a range
                        };

                        // If we have a valid range pattern, use single range scan
                        if matches!(lower_op, Operator::Gt | Operator::Gte)
                            && matches!(upper_op, Operator::Lt | Operator::Lte)
                        {
                            if let Some(index) = self.version_store.get_index_by_column(col1) {
                                // Only B-tree and PrimaryKey indexes support range queries
                                // Hash and Bitmap indexes return empty for range queries,
                                // which would incorrectly indicate "no matches" instead of
                                // "can't handle this query"
                                if !matches!(
                                    index.index_type(),
                                    IndexType::BTree | IndexType::PrimaryKey
                                ) {
                                    return None; // Fall back to full scan
                                }
                                let min_inclusive = matches!(lower_op, Operator::Gte);
                                let max_inclusive = matches!(upper_op, Operator::Lte);
                                let row_ids = index.get_row_ids_in_range(
                                    std::slice::from_ref(lower_val),
                                    std::slice::from_ref(upper_val),
                                    min_inclusive,
                                    max_inclusive,
                                );
                                // Return result even if empty - we successfully used the index
                                // Empty result is valid (no rows match the range)
                                return Some(row_ids);
                            }
                        }
                    }
                }
            }
        }

        // OPTIMIZATION: Handle AND expressions with recursive index lookup
        // For '(col1 IN (...)) AND (col2 IN (...))', process each operand and intersect results
        // This is critical for semi-join optimization where we combine right_filter with IN clause
        if let Some(and_operands) = expr.get_and_operands() {
            let mut indexed_row_ids: Vec<RowIdVec> = Vec::with_capacity(and_operands.len());

            for operand in and_operands {
                // Recursively try index lookup for each AND operand
                // OPTIMIZATION: Don't sort here - defer sorting until we know intersection is needed
                if let Some(row_ids) = self.try_index_lookup(operand.as_ref(), schema) {
                    indexed_row_ids.push(row_ids);
                }
                // If an operand can't use an index, we'll still proceed with available indexes
                // and the executor will handle memory filtering for the rest
            }

            // If at least one operand used an index, intersect and return the most restrictive result
            // The executor will handle memory filtering for operands without index support
            if !indexed_row_ids.is_empty() {
                if indexed_row_ids.len() == 1 {
                    // OPTIMIZATION: Single operand - no intersection needed
                    return Some(indexed_row_ids.into_iter().next().unwrap());
                }

                // Bitset intersection: build a bitset from the larger list for O(1) lookups,
                // then retain only matching elements from the smaller list.
                // Cap at MAX_BITSET_WORDS (~1MB) to prevent OOM with large/sparse row IDs.
                // Negative row IDs skip the bitset path (id as usize wraps).
                const MAX_BITSET_WORDS: usize = 131_072;
                indexed_row_ids.sort_by_key(|v| v.len());
                let mut result = indexed_row_ids.swap_remove(0); // smallest
                for other in &indexed_row_ids {
                    let mut max_id = 0i64;
                    let mut min_id = 0i64;
                    for &id in other.iter() {
                        if id > max_id {
                            max_id = id;
                        }
                        if id < min_id {
                            min_id = id;
                        }
                    }
                    let num_words = (max_id as usize / 64) + 1;
                    if min_id >= 0 && num_words <= MAX_BITSET_WORDS {
                        // Fast bitset path
                        let mut bits = vec![0u64; num_words];
                        for &id in other.iter() {
                            let idx = id as usize;
                            bits[idx / 64] |= 1u64 << (idx % 64);
                        }
                        result.retain(|id| {
                            let idx = *id as usize;
                            idx < num_words * 64 && (bits[idx / 64] & (1u64 << (idx % 64))) != 0
                        });
                    } else {
                        // Fallback: I64Set-based intersection for large/sparse row IDs
                        let mut set = I64Set::with_capacity(other.len());
                        let mut has_i64_min = false;
                        for &id in other.iter() {
                            if id == i64::MIN {
                                has_i64_min = true;
                            } else {
                                set.insert(id);
                            }
                        }
                        result.retain(|id| {
                            if *id == i64::MIN {
                                has_i64_min
                            } else {
                                set.contains(*id)
                            }
                        });
                    }
                    if result.is_empty() {
                        return Some(RowIdVec::new());
                    }
                }
                return Some(result);
            }

            // No operands could use indexes - fall through to other strategies
            // Note: We don't return None here because the subsequent code might
            // still be able to extract simple comparisons from the AND expression
        }

        // OPTIMIZATION: Handle IN list expressions with direct index lookup
        // For 'col IN (a, b, c)', use get_row_ids_in for efficient multi-value lookup
        if let Some(in_list) = expr
            .as_any()
            .downcast_ref::<crate::storage::expression::InListExpr>()
        {
            // Only handle positive IN (not NOT IN)
            if !in_list.is_not() {
                if let Some(col_name) = in_list.get_column_name() {
                    if let Some(index) = self.version_store.get_index_by_column(col_name) {
                        // Use the efficient get_row_ids_in method
                        let values = in_list.get_values();
                        let row_ids = index.get_row_ids_in(values);
                        return Some(row_ids);
                    }
                }
            }
        }

        // OPTIMIZATION: Handle LIKE prefix patterns with index range scan
        // For 'name LIKE 'John%'', use index range scan from 'John' to 'John\xff'
        if let Some((col_name, prefix, negated)) = expr.get_like_prefix_info() {
            // Don't optimize NOT LIKE (would need complement of range)
            if negated {
                return None;
            }

            if let Some(index) = self.version_store.get_index_by_column(col_name) {
                // Create range from prefix to prefix + '\xff' (highest byte)
                // This captures all strings starting with the prefix
                let min_value = Value::text(&prefix);
                let mut max_prefix = prefix.clone();
                max_prefix.push('\u{FFFF}'); // Highest unicode char
                let max_value = Value::text(&max_prefix);

                // Use index range query
                if let Ok(entries) = index.find_range(
                    &[min_value],
                    &[max_value],
                    true,  // include min
                    false, // exclude max
                ) {
                    let mut row_ids = RowIdVec::with_capacity(entries.len());
                    for entry in entries {
                        row_ids.push(entry.row_id);
                    }
                    return Some(row_ids);
                }
            }
        }

        // Try to extract comparisons from AND expressions
        let comparisons = expr.collect_comparisons();
        if comparisons.is_empty() {
            return None;
        }

        // Group comparisons by column name
        let mut column_comparisons: FxHashMap<&str, Vec<(Operator, &Value)>> = FxHashMap::default();
        for (col_name, op, val) in &comparisons {
            column_comparisons
                .entry(*col_name)
                .or_default()
                .push((*op, *val));
        }

        // OPTIMIZATION: Try multi-column index first
        // Collect columns that have equality predicates (can be used with multi-column index)
        let eq_columns: Vec<&str> = column_comparisons
            .iter()
            .filter_map(|(col_name, ops)| {
                // Check if this column has an equality predicate
                if ops.iter().any(|(op, _)| matches!(op, Operator::Eq)) {
                    Some(*col_name)
                } else {
                    None
                }
            })
            .collect();

        // Try to find a multi-column index that covers these equality columns (prefix rule)
        if !eq_columns.is_empty() {
            if let Some((multi_idx, matched_count)) =
                self.version_store.get_multi_column_index(&eq_columns)
            {
                // Build the values array in index column order
                let index_columns = multi_idx.column_names();
                let mut values: Vec<Value> = Vec::with_capacity(matched_count);
                let mut all_columns_matched = true;

                for idx_col in index_columns.iter().take(matched_count) {
                    if let Some(ops) = column_comparisons.get(idx_col.as_str()) {
                        // Find the equality value for this column
                        if let Some((_, val)) =
                            ops.iter().find(|(op, _)| matches!(op, Operator::Eq))
                        {
                            values.push((*val).clone());
                        } else {
                            all_columns_matched = false;
                            break;
                        }
                    } else {
                        all_columns_matched = false;
                        break;
                    }
                }

                // If we matched all columns in the prefix, use the multi-column index
                if all_columns_matched && values.len() == matched_count {
                    // Check if the next index column has range predicates — use
                    // BTree range scan instead of hash prefix lookup for better selectivity
                    let trailing_range = if matched_count < index_columns.len() {
                        let next_col = &index_columns[matched_count];
                        column_comparisons.get(next_col.as_str()).and_then(|ops| {
                            // Collect range bounds for this column
                            let mut min_val: Option<(&Value, bool)> = None;
                            let mut max_val: Option<(&Value, bool)> = None;
                            for (op, val) in ops {
                                match op {
                                    Operator::Gt => {
                                        min_val = Some((*val, false));
                                    }
                                    Operator::Gte => {
                                        min_val = Some((*val, true));
                                    }
                                    Operator::Lt => {
                                        max_val = Some((*val, false));
                                    }
                                    Operator::Lte => {
                                        max_val = Some((*val, true));
                                    }
                                    _ => {}
                                }
                            }
                            // Need an upper bound for a valid BTree range scan.
                            // Lower-bound-only (b > 10) builds min_key=[prefix, 10]
                            // but max_key=[prefix] which is SHORTER, and CompositeKey
                            // ordering makes shorter keys less — creating an inverted
                            // range that panics in BTreeMap::range.
                            // Upper-bound-only (b < 10) is safe: min_key=[prefix] <
                            // max_key=[prefix, 10].
                            if max_val.is_some() {
                                Some((min_val, max_val))
                            } else {
                                None
                            }
                        })
                    } else {
                        None
                    };

                    let row_ids = if let Some((min_bound, max_bound)) = trailing_range {
                        // Build composite min/max keys for BTree range scan
                        let mut min_key = values.clone();
                        let mut max_key = values.clone();
                        let (min_inclusive, max_inclusive) = match (min_bound, max_bound) {
                            (Some((min_v, min_inc)), Some((max_v, max_inc))) => {
                                min_key.push(min_v.clone());
                                max_key.push(max_v.clone());
                                (min_inc, max_inc)
                            }
                            // Lower-bound-only is filtered out above (inverted range)
                            (Some(_), None) => unreachable!(),
                            (None, Some((max_v, max_inc))) => {
                                // min_key stays as equality prefix
                                max_key.push(max_v.clone());
                                (true, max_inc)
                            }
                            (None, None) => unreachable!(),
                        };
                        if let Ok(entries) =
                            multi_idx.find_range(&min_key, &max_key, min_inclusive, max_inclusive)
                        {
                            let mut ids = RowIdVec::with_capacity(entries.len());
                            for entry in entries {
                                ids.push(entry.row_id);
                            }
                            ids
                        } else {
                            multi_idx.get_row_ids_equal(&values)
                        }
                    } else {
                        multi_idx.get_row_ids_equal(&values)
                    };

                    if !row_ids.is_empty() {
                        // Multi-column index gave us results - check if we need to apply
                        // additional filters for columns not covered by the index
                        let mut covered_columns: FxHashSet<&str> = index_columns
                            .iter()
                            .take(matched_count)
                            .map(|s| s.as_str())
                            .collect();
                        // The trailing range column is also covered if used
                        if trailing_range.is_some() && matched_count < index_columns.len() {
                            covered_columns.insert(index_columns[matched_count].as_str());
                        }

                        // Check if there are non-covered columns with predicates
                        let uncovered_columns: Vec<&str> = column_comparisons
                            .keys()
                            .filter(|c| !covered_columns.contains(*c))
                            .copied()
                            .collect();

                        if uncovered_columns.is_empty() {
                            // All predicate columns are covered by multi-column index
                            return Some(row_ids);
                        }
                        // There are uncovered columns - continue to check single-column indexes
                        // and intersect with multi-column index results
                        let mut all_row_ids: Vec<RowIdVec> = vec![row_ids];

                        // Check single-column indexes for uncovered columns
                        for col_name in uncovered_columns {
                            if let Some(single_idx) =
                                self.version_store.get_index_by_column(col_name)
                            {
                                if let Some(ops) = column_comparisons.get(col_name) {
                                    // Try equality first
                                    if let Some((_, val)) =
                                        ops.iter().find(|(op, _)| matches!(op, Operator::Eq))
                                    {
                                        let ids =
                                            single_idx.get_row_ids_equal(std::slice::from_ref(val));
                                        if ids.is_empty() {
                                            return Some(RowIdVec::new());
                                        }
                                        // Don't sort yet - only sort when intersection is needed
                                        all_row_ids.push(ids);
                                    }
                                }
                            }
                        }

                        // Intersect all results
                        if all_row_ids.len() == 1 {
                            return Some(all_row_ids.into_iter().next().unwrap());
                        }
                        // Row IDs are already sorted by index (sorted insertion)
                        let mut result = all_row_ids.swap_remove(0);
                        for other in &all_row_ids {
                            result = intersect_sorted_ids(&result, other);
                            if result.is_empty() {
                                return Some(RowIdVec::new());
                            }
                        }
                        return Some(result);
                    }
                }
            }
        }

        // Fall back to single-column index strategy
        // Collect row IDs from all indexed columns
        let mut all_row_ids: Vec<RowIdVec> = Vec::new();

        for (col_name, ops) in &column_comparisons {
            if let Some(index) = self.version_store.get_index_by_column(col_name) {
                // Check for range pattern: col >= min AND col <= max
                let mut min_val: Option<(&Value, bool)> = None; // (value, inclusive)
                let mut max_val: Option<(&Value, bool)> = None;
                let mut eq_val: Option<&Value> = None;

                for (op, val) in ops {
                    match op {
                        Operator::Eq => eq_val = Some(val),
                        Operator::Gt => min_val = Some((val, false)),
                        Operator::Gte => min_val = Some((val, true)),
                        Operator::Lt => max_val = Some((val, false)),
                        Operator::Lte => max_val = Some((val, true)),
                        _ => {}
                    }
                }

                // Equality takes precedence - but skip boolean (low cardinality)
                if let Some(val) = eq_val {
                    // Skip boolean equality - ~50% selectivity makes full scan faster
                    if matches!(val, Value::Boolean(_)) {
                        continue;
                    }
                    // OPTIMIZATION: Use from_ref to avoid clone
                    let row_ids = index.get_row_ids_equal(std::slice::from_ref(val));
                    if row_ids.is_empty() {
                        // If any index returns empty, the AND result is empty
                        return Some(RowIdVec::new());
                    }
                    // Don't sort yet - only sort when intersection is needed
                    all_row_ids.push(row_ids);
                    continue;
                }

                // Range query - but skip Hash indexes (they don't support range queries)
                if min_val.is_some() || max_val.is_some() {
                    // Hash indexes don't support range queries - skip them
                    // and let the query fall back to a full scan
                    if matches!(index.index_type(), IndexType::Hash) {
                        continue;
                    }

                    let row_ids =
                        if let (Some((min, min_inc)), Some((max, max_inc))) = (min_val, max_val) {
                            // OPTIMIZATION: Use from_ref to avoid clone
                            index.get_row_ids_in_range(
                                std::slice::from_ref(min),
                                std::slice::from_ref(max),
                                min_inc,
                                max_inc,
                            )
                        } else if let Some((val, inclusive)) = min_val {
                            let op = if inclusive {
                                Operator::Gte
                            } else {
                                Operator::Gt
                            };
                            self.query_index_with_operator(&*index, op, val)
                                .unwrap_or_default()
                        } else if let Some((val, inclusive)) = max_val {
                            let op = if inclusive {
                                Operator::Lte
                            } else {
                                Operator::Lt
                            };
                            self.query_index_with_operator(&*index, op, val)
                                .unwrap_or_default()
                        } else {
                            RowIdVec::new()
                        };

                    if row_ids.is_empty() {
                        // If any index returns empty, the AND result is empty
                        return Some(RowIdVec::new());
                    }
                    // Don't sort yet - only sort when intersection is needed
                    all_row_ids.push(row_ids);
                }
            }
        }

        // If we have no indexed results, return None
        if all_row_ids.is_empty() {
            return None;
        }

        // If we have only one index, return its results (no sort needed)
        if all_row_ids.len() == 1 {
            return Some(all_row_ids.into_iter().next().unwrap());
        }

        // Range query results from BTree are in VALUE order, not row_id order.
        // intersect_sorted_ids requires row_id-sorted input (uses binary_search).
        // Sort all sets before intersection.
        for ids in &mut all_row_ids {
            ids.sort_unstable();
        }

        // Intersect all row ID sets for multi-column filtering
        let mut result = all_row_ids.swap_remove(0);
        for other in &all_row_ids {
            result = intersect_sorted_ids(&result, other);
            if result.is_empty() {
                return Some(RowIdVec::new());
            }
        }

        Some(result)
    }

    /// Query an index with a specific operator
    /// Returns a pooled RowIdVec for efficient memory reuse.
    fn query_index_with_operator(
        &self,
        index: &dyn crate::storage::traits::Index,
        operator: crate::core::Operator,
        value: &Value,
    ) -> Option<RowIdVec> {
        use crate::core::Operator;

        match operator {
            Operator::Eq => {
                let row_ids = index.get_row_ids_equal(std::slice::from_ref(value));
                if row_ids.is_empty() {
                    None
                } else {
                    Some(row_ids)
                }
            }
            Operator::Gt | Operator::Gte | Operator::Lt | Operator::Lte => {
                // Use find_with_operator for range queries
                let entries = index
                    .find_with_operator(operator, std::slice::from_ref(value))
                    .ok()?;
                let mut row_ids = RowIdVec::with_capacity(entries.len());
                for entry in entries {
                    row_ids.push(entry.row_id);
                }
                if row_ids.is_empty() {
                    None
                } else {
                    Some(row_ids)
                }
            }
            _ => None,
        }
    }

    /// Fast path for index-based row fetching when there are no local transaction changes.
    /// Returns Some(rows) if we can serve from index, None if we should fall through to full path.
    /// Takes RowIdVec for efficient pooled memory reuse.
    #[inline]
    fn try_fetch_from_index(
        &self,
        row_ids: RowIdVec,
        expr: &dyn Expression,
        schema: &Schema,
        limit: usize,
        offset: usize,
    ) -> Option<RowVec> {
        if limit == 0 {
            return Some(RowVec::new());
        }

        let txn_versions = self.txn_versions.read().unwrap();
        if txn_versions.has_local_changes() {
            return None; // Fall through to full path
        }

        // No local changes - use for_each_visible for single lock acquisition + early termination
        let mut result = RowVec::with_capacity(limit.min(row_ids.len()));
        let mut skipped = 0usize;

        self.version_store
            .for_each_visible(&row_ids, self.txn_id, |row_id, row_data| {
                // Re-apply filter (index may return superset for complex expressions)
                let row = self.normalize_row_to_schema(row_data, schema);
                if expr.evaluate_fast(&row) {
                    if skipped < offset {
                        skipped += 1;
                    } else {
                        result.push((row_id, row));
                        if result.len() >= limit {
                            return false; // Stop: LIMIT reached
                        }
                    }
                }
                true // Continue
            });
        Some(result)
    }

    /// Hybrid optimization for mixed OR predicates (some indexed, some not).
    ///
    /// For `(indexed_col = X OR non_indexed_col = Y)`:
    /// 1. Fetch rows matching indexed branches via index lookup
    /// 2. Scan all rows evaluating only the non-indexed branches (skip already-found rows)
    /// 3. Union deduplicated results
    ///
    /// Returns `Some(rows)` when we can handle this, `None` to fall through.
    fn try_mixed_or_fetch(
        &self,
        expr: &dyn Expression,
        schema: &Schema,
        limit: usize,
        offset: usize,
    ) -> Option<RowVec> {
        if limit == 0 {
            return Some(RowVec::new());
        }

        let or_operands = expr.get_or_operands()?;

        // Partition operands into indexed and non-indexed
        let mut indexed_row_ids: Vec<RowIdVec> = Vec::new();
        let mut unindexed_operands: Vec<Box<dyn Expression>> = Vec::new();

        for operand in or_operands {
            if let Some(row_ids) = self.try_index_lookup(operand.as_ref(), schema) {
                indexed_row_ids.push(row_ids);
            } else {
                unindexed_operands.push(operand.clone_box());
            }
        }

        // Only useful when we have BOTH indexed and non-indexed operands
        if indexed_row_ids.is_empty() || unindexed_operands.is_empty() {
            return None;
        }

        // Bail if there are local transaction changes (complex merge needed)
        let txn_versions = self.txn_versions.read().unwrap();
        if txn_versions.has_local_changes() {
            return None;
        }
        drop(txn_versions);

        // Phase 1: Fetch rows matching indexed branches
        // Build a bitset/set of indexed row IDs for dedup
        const MAX_BITSET_WORDS: usize = 131_072;
        let mut max_id = 0i64;
        let mut min_id = 0i64;
        let total_indexed: usize = indexed_row_ids.iter().map(|v| v.len()).sum();
        for ids in &indexed_row_ids {
            for &id in ids.iter() {
                if id > max_id {
                    max_id = id;
                }
                if id < min_id {
                    min_id = id;
                }
            }
        }

        // Deduplicate indexed row IDs into a single RowIdVec + build lookup set
        let num_words = if max_id >= 0 {
            (max_id as usize / 64) + 1
        } else {
            0
        };
        let use_bitset = min_id >= 0 && num_words <= MAX_BITSET_WORDS;

        let mut deduped_ids = RowIdVec::with_capacity(total_indexed);
        let mut bitset = if use_bitset {
            vec![0u64; num_words]
        } else {
            vec![]
        };
        let mut id_set = if !use_bitset {
            I64Set::with_capacity(total_indexed)
        } else {
            I64Set::new()
        };

        for ids in &indexed_row_ids {
            for &id in ids.iter() {
                if use_bitset {
                    let idx = id as usize;
                    let word = idx / 64;
                    let mask = 1u64 << (idx % 64);
                    if (bitset[word] & mask) == 0 {
                        bitset[word] |= mask;
                        deduped_ids.push(id);
                    }
                } else if id_set.insert(id) {
                    deduped_ids.push(id);
                }
            }
        }

        // Fetch indexed rows
        let mut result = RowVec::with_capacity(limit.min(total_indexed + 64));
        let mut skipped = 0usize;

        self.version_store
            .for_each_visible(&deduped_ids, self.txn_id, |row_id, row_data| {
                let row = self.normalize_row_to_schema(row_data, schema);
                // Re-apply full OR filter to ensure correctness (index may return superset)
                if expr.evaluate_fast(&row) {
                    if skipped < offset {
                        skipped += 1;
                    } else {
                        result.push((row_id, row));
                        if result.len() >= limit {
                            return false;
                        }
                    }
                }
                true
            });

        if result.len() >= limit {
            return Some(result);
        }

        // Phase 2: Scan for rows matching non-indexed branches (skip already-found rows)
        // Build filter from non-indexed operands only
        let mut unindexed_filter: Box<dyn Expression> = if unindexed_operands.len() == 1 {
            unindexed_operands.into_iter().next().unwrap()
        } else {
            Box::new(OrExpr::new(unindexed_operands))
        };
        unindexed_filter.prepare_for_schema(schema);

        let remaining_offset = offset.saturating_sub(skipped);
        let mut scan_skipped = 0usize;

        // Streaming scan: filter + dedup + limit inline, no bulk materialization
        self.version_store.for_each_visible_filtered(
            self.txn_id,
            unindexed_filter.as_ref(),
            |row_id, row_data| {
                // Skip rows already found by index
                let already_found = if use_bitset {
                    let idx = row_id as usize;
                    if idx / 64 < bitset.len() {
                        (bitset[idx / 64] & (1u64 << (idx % 64))) != 0
                    } else {
                        false
                    }
                } else {
                    id_set.contains(row_id)
                };
                if already_found {
                    return true; // continue
                }

                let row = self.normalize_row_to_schema(row_data, schema);
                if scan_skipped < remaining_offset {
                    scan_skipped += 1;
                } else {
                    result.push((row_id, row));
                    if result.len() >= limit {
                        return false; // stop
                    }
                }
                true // continue
            },
        );

        Some(result)
    }

    /// Validates and coerces a row against the schema
    /// Returns the coerced row if successful
    fn validate_and_coerce_row(&self, row: &mut Row) -> Result<()> {
        // OPTIMIZATION: Use cached_schema instead of version_store.schema() to avoid clone
        let schema = &self.cached_schema;

        // Check column count
        if row.len() != schema.columns.len() {
            return Err(Error::internal(format!(
                "invalid column count: expected {}, got {}",
                schema.columns.len(),
                row.len()
            )));
        }

        // Validate and coerce each column
        for (i, col) in schema.columns.iter().enumerate() {
            let value = row.get(i).ok_or_else(|| {
                Error::internal(format!("nil value at index {} (column '{}')", i, col.name))
            })?;

            // Check NULL constraint
            if !col.nullable && value.is_null() {
                return Err(Error::internal(format!(
                    "NULL value in non-nullable column '{}'",
                    col.name
                )));
            }

            // Check type compatibility for non-NULL values
            if !value.is_null() {
                let actual_type = value.data_type();
                if actual_type != col.data_type {
                    // Allow implicit type coercions (standard SQL behavior)
                    if actual_type == DataType::Text && col.data_type == DataType::Json {
                        // Coerce Text to Json
                        if let Some(text_str) = value.as_str() {
                            let _ = row.set(i, Value::json(text_str));
                        }
                    } else if actual_type == DataType::Integer && col.data_type == DataType::Float {
                        // Coerce Integer to Float
                        if let Value::Integer(n) = value {
                            let _ = row.set(i, Value::Float(*n as f64));
                        }
                    } else if actual_type == DataType::Float && col.data_type == DataType::Integer {
                        // Coerce Float to Integer (truncate)
                        if let Value::Float(f) = value {
                            let _ = row.set(i, Value::Integer(*f as i64));
                        }
                    } else if actual_type == DataType::Integer && col.data_type == DataType::Boolean
                    {
                        // Coerce Integer to Boolean (0 = false, non-zero = true)
                        if let Value::Integer(n) = value {
                            let _ = row.set(i, Value::Boolean(*n != 0));
                        }
                    } else if actual_type == DataType::Text && col.data_type == DataType::Vector {
                        // Coerce Text to Extension(Vector) binary format
                        if let Value::Text(s) = value {
                            match crate::core::value::parse_vector_str(s.as_ref()) {
                                Some(floats) => {
                                    let _ = row.set(i, Value::vector(floats));
                                }
                                None => {
                                    return Err(Error::internal(format!(
                                        "cannot parse vector value in column '{}': '{}'",
                                        col.name, s
                                    )));
                                }
                            }
                        }
                    } else {
                        return Err(Error::internal(format!(
                            "type mismatch in column '{}': expected {:?}, got {:?}",
                            col.name, col.data_type, actual_type
                        )));
                    }
                }
            }
        }

        Ok(())
    }

    /// Extracts the primary key value from a row
    /// OPTIMIZATION: Uses cached_schema and pk_column_index to avoid iteration
    fn extract_row_pk(&self, row: &Row) -> i64 {
        // Fast path: use cached PK index if available
        if let Some(pk_idx) = self.cached_schema.pk_column_index() {
            if let Some(value) = row.get(pk_idx) {
                if let Some(pk) = value.as_int64() {
                    return pk;
                }
            }
        }

        // Fallback: If no primary key or not an integer, generate a synthetic row ID
        self.version_store.get_next_auto_increment_id()
    }

    /// Finds the primary key column index
    /// OPTIMIZATION: Uses cached pk_column_index from schema
    #[inline]
    fn find_pk_column_index(&self) -> Option<usize> {
        self.cached_schema.pk_column_index()
    }

    /// Check unique index constraints for a row being inserted.
    fn check_unique_constraints(&self, row: &Row, _row_id: i64) -> Result<()> {
        let schema = &self.cached_schema;
        self.version_store
            .for_each_unique_index(|index_name, index| {
                let column_ids = index.column_ids();
                if column_ids.is_empty() {
                    return Ok(());
                }
                let values: Vec<Value> = column_ids
                    .iter()
                    .filter_map(|&col_id| row.get(col_id as usize).cloned())
                    .collect();
                if values.len() != column_ids.len() {
                    return Ok(());
                }
                if values.iter().any(|v| v.is_null()) {
                    return Ok(());
                }
                let entries = index.find(&values)?;
                if let Some(entry) = entries.first() {
                    let col_names: Vec<&str> = column_ids
                        .iter()
                        .map(|&col_id| {
                            schema
                                .columns
                                .get(col_id as usize)
                                .map(|c| c.name.as_str())
                                .unwrap_or("unknown")
                        })
                        .collect();
                    return Err(Error::UniqueConstraint {
                        index: index_name.to_string(),
                        column: col_names.join(", "),
                        value: format!("{:?}", values),
                        row_id: entry.row_id,
                    });
                }
                Ok(())
            })
    }

    /// Prepares a row for insertion: handles auto-increment, validates, and checks constraints.
    /// Returns the row_id on success. The row is modified in-place with auto-increment values.
    #[inline]
    fn prepare_insert(&mut self, row: &mut Row) -> Result<i64> {
        // Handle auto-increment for primary key BEFORE validation
        // This allows inserting without specifying the primary key (only if AUTO_INCREMENT)
        if let Some(pk_idx) = self.find_pk_column_index() {
            let pk_col = &self.cached_schema.columns[pk_idx];
            let is_auto_increment = pk_col.auto_increment;

            if let Some(value) = row.get(pk_idx) {
                if value.is_null() {
                    if is_auto_increment {
                        // Generate new ID for NULL primary key
                        let next_id = self.version_store.get_next_auto_increment_id();
                        let _ = row.set(pk_idx, Value::Integer(next_id));
                    } else {
                        // Non-INTEGER PRIMARY KEY without AUTO_INCREMENT cannot be NULL
                        return Err(Error::internal(format!(
                            "NULL value not allowed for PRIMARY KEY column '{}'.",
                            pk_col.name
                        )));
                    }
                } else if let Some(pk_val) = value.as_int64() {
                    // Track max row_id for ORDER BY contiguous iteration optimization.
                    // Uses the auto-increment counter as a convenient max-row-id tracker.
                    let current = self.version_store.get_auto_increment_counter();
                    if pk_val > current {
                        self.version_store.set_auto_increment_counter(pk_val);
                    }
                }
            }
        }

        // Validate and coerce row AFTER auto-increment has filled in the primary key
        self.validate_and_coerce_row(row)?;

        // Extract row ID
        let row_id = self.extract_row_pk(row);

        // Check if row already exists in local versions
        {
            let txn_versions = self.txn_versions.read().unwrap();
            if txn_versions.has_locally_seen(row_id) && txn_versions.get(row_id).is_some() {
                return Err(Error::primary_key_constraint(row_id));
            }
        }

        // Check if row exists in global store
        if self.version_store.quick_check_row_existence(row_id) {
            if let Some(version) = self.version_store.get_visible_version(row_id, self.txn_id) {
                if !version.is_deleted() {
                    return Err(Error::primary_key_constraint(row_id));
                }
            }
        }

        // Check unique index constraints (READ lock).
        self.check_unique_constraints(row, row_id)?;

        Ok(row_id)
    }

    /// Commits the transaction's local changes
    ///
    /// Index updates are handled by TransactionVersionStore::commit() using batch
    /// operations to reduce lock acquisitions from O(rows × indexes) to O(indexes).
    pub fn commit(&mut self) -> Result<()> {
        // Check if there are local changes before committing
        let has_changes = {
            let txn_versions = self.txn_versions.read().unwrap();
            txn_versions.has_local_changes()
        };

        // Commit versions to the version store (this also updates indexes)
        self.txn_versions.write().unwrap().commit()?;

        // Mark zone maps as stale if we had any data changes
        // This ensures the optimizer won't use outdated pruning info
        if has_changes {
            self.version_store.mark_zone_maps_stale();
        }

        Ok(())
    }

    /// Rolls back the transaction's local changes
    pub fn rollback(&mut self) {
        self.txn_versions.write().unwrap().rollback();
    }

    /// Returns the row count visible to this transaction
    ///
    /// OPTIMIZATION: Uses single-pass counting instead of per-row visibility checks.
    /// Reduces lock acquisitions from O(N) to O(1) for the global store.
    pub fn row_count(&self) -> usize {
        // Count global visible versions in single pass (O(1) lock instead of O(N))
        let mut count = self.version_store.count_visible_rows(self.txn_id);

        // Adjust for local changes (uncommitted in this transaction)
        let txn_versions = self.txn_versions.read().unwrap();
        for (row_id, version) in txn_versions.iter_local() {
            // Check if this row exists in global store
            let exists_in_global = self.version_store.quick_check_row_existence(row_id);

            if version.is_deleted() {
                // If deleted locally and existed in global, subtract from count
                if exists_in_global {
                    count = count.saturating_sub(1);
                }
            } else {
                // If inserted locally and not in global, add to count
                if !exists_in_global {
                    count += 1;
                }
            }
        }

        count
    }

    /// Fast O(1) row count for COUNT(*) queries without local changes
    ///
    /// Returns Some(count) if the fast path can be used (no local changes),
    /// Returns None if the caller should fall back to the full row_count() method.
    ///
    /// OPTIMIZATION: This uses the pre-computed committed_row_count which is O(1)
    /// instead of iterating all rows with visibility checks.
    #[inline]
    pub fn fast_row_count(&self) -> Option<usize> {
        // Check if there are any local changes
        let txn_versions = self.txn_versions.read().unwrap();
        if txn_versions.has_local_changes() {
            return None;
        }
        // Under snapshot isolation, committed_row_count includes rows committed
        // after our snapshot point, so the O(1) count would be wrong.
        if self.version_store.needs_snapshot_isolation(self.txn_id) {
            return None;
        }
        // No local changes, read-committed - use the O(1) committed row count
        Some(self.version_store.committed_row_count())
    }

    /// Collect all visible rows, optionally filtered
    ///
    /// Optimized to use batch fetch even when there are local changes,
    /// then merge the results.
    /// Returns RowVec for zero-allocation reuse across queries.
    #[inline]
    fn collect_visible_rows(&self, filter: Option<&dyn Expression>) -> RowVec {
        let txn_versions = self.txn_versions.read().unwrap();
        let schema = &self.cached_schema;

        // Check if we have local versions (uncommitted changes in this transaction)
        let has_local = txn_versions.has_local_changes();

        if !has_local {
            // No local versions - use thread-local cached Vec for zero-allocation reuse
            // Arena storage provides 50x+ faster scans via contiguous memory access
            if let Some(expr) = filter {
                // Use filtered version - returns RowVec directly
                return self
                    .version_store
                    .get_all_visible_rows_filtered(self.txn_id, expr);
            }

            // Use cached version for unfiltered scans (main optimization)
            let raw_rows = self.version_store.get_all_visible_rows_cached(self.txn_id);

            // OPTIMIZATION: Skip normalization if first row matches schema column count
            // This is the common case when no ALTER TABLE ADD/DROP COLUMN has occurred
            let schema_cols = schema.columns.len();
            if raw_rows
                .first()
                .is_none_or(|(_, row)| row.len() == schema_cols)
            {
                // Fast path: all rows should have same column count, skip normalization
                return raw_rows;
            }

            // Slow path: need to normalize rows for schema evolution
            // Iterate over cached vec (drains it), collect into RowVec
            return raw_rows
                .into_iter()
                .map(|(row_id, row)| (row_id, self.normalize_row_to_schema(row, schema)))
                .collect();
        }

        // Has local versions - use batch fetch then merge
        // Step 1: Get all global rows in one batch (single lock acquisition)
        // Uses thread-local cached Vec for zero-allocation reuse
        let global_rows = self.version_store.get_all_visible_rows_cached(self.txn_id);

        // Step 2: Build set of local row IDs for quick lookup (I64Set for fast i64 lookups)
        let mut local_row_ids = I64Set::new();
        let mut local_has_i64_min = false;
        for (row_id, _) in txn_versions.iter_local() {
            if row_id == i64::MIN {
                local_has_i64_min = true;
            } else {
                local_row_ids.insert(row_id);
            }
        }

        // Step 3: Pre-allocate result
        let mut rows = RowVec::with_capacity(global_rows.len() + local_row_ids.len());

        // Step 4: Add global rows that don't have local overrides
        for (row_id, row) in global_rows {
            if (row_id == i64::MIN && local_has_i64_min) || local_row_ids.contains(row_id) {
                continue; // Local version takes precedence
            }
            // Normalize row to match current schema (handles ALTER TABLE ADD/DROP COLUMN)
            let row = self.normalize_row_to_schema(row, schema);
            if let Some(expr) = filter {
                if !expr.evaluate_fast(&row) {
                    continue;
                }
            }
            rows.push((row_id, row));
        }

        // Step 5: Add local versions (both updates and inserts)
        for (row_id, version) in txn_versions.iter_local() {
            if version.is_deleted() {
                continue;
            }
            // Normalize row to match current schema (handles ALTER TABLE ADD/DROP COLUMN)
            let row = self.normalize_row_to_schema(version.data.clone(), schema);
            if let Some(expr) = filter {
                if !expr.evaluate_fast(&row) {
                    continue;
                }
            }
            rows.push((row_id, row));
        }

        rows
    }

    /// Collect visible rows WITHOUT sorting for GROUP BY optimization.
    ///
    /// This skips the O(n log n) sort since GROUP BY doesn't care about row order.
    /// Returns rows in version store iteration order.
    #[inline]
    fn collect_visible_rows_unsorted(&self) -> RowVec {
        // For GROUP BY, order doesn't matter - use same cached path
        self.collect_visible_rows(None)
    }

    /// Collect visible rows with early termination when limit is reached
    /// This is the LIMIT pushdown optimization
    fn collect_visible_rows_with_limit(
        &self,
        filter: Option<&dyn Expression>,
        limit: usize,
        offset: usize,
    ) -> RowVec {
        let schema = &self.cached_schema;

        // FAST PATH: Check if this is a primary key equality lookup (WHERE id = X)
        // This is O(1) and skips all scanning
        if let Some(expr) = filter {
            if let Some(pk_id) = self.try_pk_lookup(expr, schema) {
                // Direct O(1) lookup by primary key
                let txn_versions = self.txn_versions.read().unwrap();
                // Check local versions first via get_local_version (preserves
                // delete signal). txn_versions.get() swallows deletes as None,
                // which would cause a fallback to the committed store and miss
                // uncommitted deletes in the current transaction.
                let row = if let Some(local) = txn_versions.get_local_version(pk_id) {
                    if local.is_deleted() {
                        None // Locally deleted in this transaction
                    } else {
                        Some((pk_id, local.data.clone()))
                    }
                } else if let Some(version) =
                    self.version_store.get_visible_version(pk_id, self.txn_id)
                {
                    if !version.is_deleted() {
                        Some((pk_id, version.data.clone()))
                    } else {
                        None
                    }
                } else {
                    None
                };

                return match row {
                    Some((row_id, r)) if offset == 0 && limit >= 1 => {
                        let mut rv = RowVec::with_capacity(1);
                        rv.push((row_id, self.normalize_row_to_schema(r, schema)));
                        rv
                    }
                    _ => RowVec::new(),
                };
            }

            // OPTIMIZATION: Try secondary index lookup for OR/AND expressions on indexed columns
            // This handles queries like: WHERE age = 25 OR age = 50 OR age = 75
            if let Some(row_ids) = self.try_index_lookup(expr, schema) {
                if let Some(result) =
                    self.try_fetch_from_index(row_ids, expr, schema, limit, offset)
                {
                    return result;
                }
                // Has local changes - fall through to full path
            }

            // OPTIMIZATION: Mixed OR (indexed OR non-indexed) hybrid scan
            if let Some(result) = self.try_mixed_or_fetch(expr, schema, limit, offset) {
                return result;
            }
        }

        let txn_versions = self.txn_versions.read().unwrap();

        // Check if we have local versions (uncommitted changes in this transaction)
        let has_local = txn_versions.has_local_changes();

        if !has_local {
            // No local versions - use optimized path with true LIMIT pushdown
            let raw_rows = if let Some(expr) = filter {
                // With filter + limit: use filtered limit pushdown
                // This early-terminates when limit is reached after filtering
                self.version_store.get_visible_rows_filtered_with_limit(
                    self.txn_id,
                    expr,
                    limit,
                    offset,
                )
            } else {
                // No filter: use simple LIMIT pushdown
                // This avoids scanning all 10K rows for LIMIT 10 queries - ~30x speedup
                self.version_store
                    .get_visible_rows_with_limit(self.txn_id, limit, offset)
            };

            // Keep row IDs in RowVec
            return raw_rows
                .into_iter()
                .map(|(row_id, row)| (row_id, self.normalize_row_to_schema(row, schema)))
                .collect();
        }

        // Has local versions - use batch fetch then merge with early termination
        // Uses thread-local cached Vec for zero-allocation reuse
        let global_rows = self.version_store.get_all_visible_rows_cached(self.txn_id);

        // Build set of local row IDs for quick lookup
        let mut local_row_ids = I64Set::new();
        let mut local_has_i64_min = false;
        for (row_id, _) in txn_versions.iter_local() {
            if row_id == i64::MIN {
                local_has_i64_min = true;
            } else {
                local_row_ids.insert(row_id);
            }
        }

        let mut result = RowVec::with_capacity(limit);
        let mut count = 0;

        // Add global rows that don't have local overrides
        for (row_id, row) in global_rows {
            if (row_id == i64::MIN && local_has_i64_min) || local_row_ids.contains(row_id) {
                continue; // Local version takes precedence
            }
            let row = self.normalize_row_to_schema(row, schema);
            if let Some(expr) = filter {
                if !expr.evaluate_fast(&row) {
                    continue;
                }
            }
            if count >= offset {
                result.push((row_id, row));
                if result.len() >= limit {
                    return result;
                }
            }
            count += 1;
        }

        // Add local versions (both updates and inserts)
        for (row_id, version) in txn_versions.iter_local() {
            if version.is_deleted() {
                continue;
            }
            let row = self.normalize_row_to_schema(version.data.clone(), schema);
            if let Some(expr) = filter {
                if !expr.evaluate_fast(&row) {
                    continue;
                }
            }
            if count >= offset {
                result.push((row_id, row));
                if result.len() >= limit {
                    return result;
                }
            }
            count += 1;
        }

        result
    }

    /// Collect visible rows with LIMIT without guaranteeing deterministic order.
    /// This is an optimization for queries with LIMIT but without ORDER BY.
    /// Since SQL doesn't guarantee order for LIMIT without ORDER BY, we can
    /// skip sorting and return rows in arbitrary order, enabling true early termination.
    fn collect_visible_rows_with_limit_unordered(
        &self,
        filter: Option<&dyn Expression>,
        limit: usize,
        offset: usize,
    ) -> RowVec {
        let schema = &self.cached_schema;

        // FAST PATH: Check if this is a primary key equality lookup (WHERE id = X)
        // This is O(1) and skips all scanning
        if let Some(expr) = filter {
            if let Some(pk_id) = self.try_pk_lookup(expr, schema) {
                // Direct O(1) lookup by primary key
                let txn_versions = self.txn_versions.read().unwrap();
                // Check local versions first via get_local_version (preserves
                // delete signal). txn_versions.get() swallows deletes as None,
                // which would cause a fallback to the committed store and miss
                // uncommitted deletes in the current transaction.
                let row = if let Some(local) = txn_versions.get_local_version(pk_id) {
                    if local.is_deleted() {
                        None // Locally deleted in this transaction
                    } else {
                        Some((pk_id, local.data.clone()))
                    }
                } else if let Some(version) =
                    self.version_store.get_visible_version(pk_id, self.txn_id)
                {
                    if !version.is_deleted() {
                        Some((pk_id, version.data.clone()))
                    } else {
                        None
                    }
                } else {
                    None
                };

                return match row {
                    Some((row_id, r)) if offset == 0 && limit >= 1 => {
                        let mut rv = RowVec::with_capacity(1);
                        rv.push((row_id, self.normalize_row_to_schema(r, schema)));
                        rv
                    }
                    _ => RowVec::new(),
                };
            }

            // OPTIMIZATION: Try secondary index lookup for OR/AND expressions on indexed columns
            // This handles queries like: WHERE age = 25 OR age = 50 OR age = 75
            if let Some(row_ids) = self.try_index_lookup(expr, schema) {
                if let Some(result) =
                    self.try_fetch_from_index(row_ids, expr, schema, limit, offset)
                {
                    return result;
                }
                // Has local changes - fall through to full path
            }

            // OPTIMIZATION: Mixed OR (indexed OR non-indexed) hybrid scan
            if let Some(result) = self.try_mixed_or_fetch(expr, schema, limit, offset) {
                return result;
            }
        }

        let txn_versions = self.txn_versions.read().unwrap();

        // Check if we have local versions (uncommitted changes in this transaction)
        let has_local = txn_versions.has_local_changes();

        if !has_local {
            // No local versions - use optimized unordered path with true early termination
            let raw_rows = if let Some(expr) = filter {
                self.version_store
                    .get_visible_rows_filtered_with_limit_unordered(
                        self.txn_id,
                        expr,
                        limit,
                        offset,
                    )
            } else {
                self.version_store
                    .get_visible_rows_with_limit_unordered(self.txn_id, limit, offset)
            };

            // Keep row IDs in RowVec
            return raw_rows
                .into_iter()
                .map(|(row_id, row)| (row_id, self.normalize_row_to_schema(row, schema)))
                .collect();
        }

        // Has local versions - use same path as ordered (local changes are rare)
        // Early termination is already implemented in collect_visible_rows_with_limit
        // when there are local changes
        drop(txn_versions);
        self.collect_visible_rows_with_limit(filter, limit, offset)
    }
}

impl Table for MVCCTable {
    fn name(&self) -> &str {
        self.version_store.table_name()
    }

    fn schema(&self) -> &Schema {
        &self.cached_schema
    }

    fn txn_id(&self) -> i64 {
        self.txn_id
    }

    /// Fetch rows by their IDs, applying filter
    ///
    /// Fetch rows into a reusable RowVec buffer.
    /// Optimized to use batch fetch for global store rows,
    /// reducing lock contention from O(n) to O(1).
    fn fetch_rows_by_ids_into(&self, row_ids: &[i64], filter: &dyn Expression, rows: &mut RowVec) {
        let txn_versions = self.txn_versions.read().unwrap();
        let schema = &self.cached_schema;

        // Don't pre-allocate based on row_ids.len() — for volume-backed tables,
        // indexes may contain row_ids that exist in frozen volumes but not in
        // the version store. Most lookups will be misses, so pre-allocating
        // for all of them wastes memory.
        let mut global_row_ids = Vec::with_capacity(row_ids.len().min(4096));

        // Step 1: Check local versions first (uncommitted changes in this transaction)
        for &row_id in row_ids {
            if let Some(version) = txn_versions.get_local_version(row_id) {
                if !version.is_deleted() {
                    // Normalize row to match current schema (handles ALTER TABLE ADD/DROP COLUMN)
                    let row = self.normalize_row_to_schema(version.data.clone(), schema);
                    if filter.evaluate_fast(&row) {
                        rows.push((row_id, row));
                    }
                }
                // Skip global lookup - local version takes precedence
            } else {
                // Need to fetch from global store
                global_row_ids.push(row_id);
            }
        }

        // Step 2: Batch fetch from global store (single lock acquisition)
        if !global_row_ids.is_empty() {
            let global_rows = self
                .version_store
                .get_visible_versions_batch(&global_row_ids, self.txn_id);

            // Apply filter to fetched rows
            for (row_id, row) in global_rows {
                // Normalize row to match current schema (handles ALTER TABLE ADD/DROP COLUMN)
                let row = self.normalize_row_to_schema(row, schema);
                if filter.evaluate_fast(&row) {
                    rows.push((row_id, row));
                }
            }
        }
    }

    fn create_column(&mut self, name: &str, column_type: DataType, nullable: bool) -> Result<()> {
        self.create_column_with_default(name, column_type, nullable, None)
    }

    fn create_column_with_default(
        &mut self,
        name: &str,
        column_type: DataType,
        nullable: bool,
        default_expr: Option<String>,
    ) -> Result<()> {
        self.create_column_with_default_value(name, column_type, nullable, default_expr, None)
    }

    fn create_column_with_default_value(
        &mut self,
        name: &str,
        column_type: DataType,
        nullable: bool,
        default_expr: Option<String>,
        default_value: Option<Value>,
    ) -> Result<()> {
        // Create a SchemaColumn and add to both version store and cached schema
        // Get the next column ID
        let next_id = self.cached_schema.columns.len();
        let column = SchemaColumn::with_default_value(
            next_id,
            name,
            column_type,
            nullable,
            false, // primary_key
            false, // auto_increment
            default_expr,
            default_value,
            None, // check_expr
        );
        {
            let mut schema_guard = self.version_store.schema_mut();
            CompactArc::make_mut(&mut *schema_guard).add_column(column.clone())?;
        }
        CompactArc::make_mut(&mut self.cached_schema).add_column(column)?;
        Ok(())
    }

    fn drop_column(&mut self, name: &str) -> Result<()> {
        // Remove column from both version store and cached schema
        {
            let mut schema_guard = self.version_store.schema_mut();
            CompactArc::make_mut(&mut *schema_guard).remove_column(name)?;
        }
        CompactArc::make_mut(&mut self.cached_schema).remove_column(name)?;
        Ok(())
    }

    fn insert(&mut self, mut row: Row) -> Result<Row> {
        let row_id = self.prepare_insert(&mut row)?;
        let inserted_row = row.clone();
        self.txn_versions.write().unwrap().put(row_id, row, false)?;
        Ok(inserted_row)
    }

    fn insert_discard(&mut self, mut row: Row) -> Result<()> {
        let row_id = self.prepare_insert(&mut row)?;
        self.txn_versions.write().unwrap().put(row_id, row, false)?;
        Ok(())
    }

    fn insert_batch(&mut self, rows: Vec<Row>) -> Result<()> {
        // Use insert_discard since we don't need returned rows
        for row in rows {
            self.insert_discard(row)?;
        }
        Ok(())
    }

    fn update(
        &mut self,
        where_expr: Option<&dyn Expression>,
        setter: &mut dyn FnMut(Row) -> Result<(Row, bool)>,
    ) -> Result<i32> {
        // OPTIMIZATION: Borrow schema instead of cloning - saves allocation per update
        let schema = &self.cached_schema;

        // Fast path: Check if this is a primary key equality lookup (WHERE id = X)
        if let Some(expr) = where_expr {
            if let Some(pk_id) = self.try_pk_lookup(expr, schema) {
                // Direct O(1) lookup by primary key
                // Use get_local_version to preserve delete signal. txn_versions.get()
                // swallows deletes as None, causing fallback to committed store and
                // missing uncommitted deletes in the current transaction.
                let row_with_original = {
                    let txn_versions = self.txn_versions.read().unwrap();
                    if let Some(local) = txn_versions.get_local_version(pk_id) {
                        if local.is_deleted() {
                            None // Locally deleted — can't update
                        } else {
                            // Local version - no need to track original (already in write-set)
                            Some((local.data.clone(), None))
                        }
                    } else if let Some(version) =
                        self.version_store.get_visible_version(pk_id, self.txn_id)
                    {
                        if !version.is_deleted() {
                            let data = version.data.clone();
                            Some((data, Some(version)))
                        } else {
                            None
                        }
                    } else {
                        None
                    }
                };

                if let Some((row, original_version)) = row_with_original {
                    // Normalize row to match current schema (handles ALTER TABLE ADD/DROP COLUMN)
                    let row = self.normalize_row_to_schema(row, schema);
                    let (updated_row, changed) = setter(row)?;
                    if !changed {
                        return Ok(0);
                    }
                    if let Some(orig) = original_version {
                        // Use optimized put that skips redundant get_visible_version
                        self.txn_versions.write().unwrap().put_with_original(
                            pk_id,
                            updated_row,
                            orig,
                            false,
                        )?;
                    } else {
                        // Local version - use regular put
                        self.txn_versions
                            .write()
                            .unwrap()
                            .put(pk_id, updated_row, false)?;
                    }
                    return Ok(1);
                }
                return Ok(0);
            }

            // Fast path: PK range lookup (WHERE id >= X AND id < Y)
            if let Some(pk_range_ids) = self.try_pk_range_lookup(expr, schema) {
                // OPTIMIZATION: Track original versions to avoid redundant lookups
                let mut local_rows = RowVec::with_capacity(pk_range_ids.len() / 4);
                let mut rows_with_originals: Vec<(i64, Row, crate::storage::mvcc::RowVersion)> =
                    Vec::with_capacity(pk_range_ids.len());

                // Step 1: Check local versions first (single lock acquisition)
                // Use get_local_version to distinguish "no local version" from "locally deleted"
                let mut remaining_row_ids: Vec<i64> = Vec::with_capacity(pk_range_ids.len());
                {
                    let txn_versions = self.txn_versions.read().unwrap();
                    for row_id in pk_range_ids {
                        if let Some(local) = txn_versions.get_local_version(row_id) {
                            if !local.is_deleted() {
                                let row = self.normalize_row_to_schema(local.data.clone(), schema);
                                let (updated_row, changed) = setter(row)?;
                                if changed {
                                    local_rows.push((row_id, updated_row));
                                }
                            }
                            // Locally deleted — skip, don't fall through
                        } else {
                            remaining_row_ids.push(row_id);
                        }
                    }
                }

                // Step 2: Batch fetch remaining from version store (1 lock for N rows)
                if !remaining_row_ids.is_empty() {
                    let batch_rows = self
                        .version_store
                        .get_visible_versions_for_update(&remaining_row_ids, self.txn_id);
                    for (row_id, row, version) in batch_rows {
                        let row = self.normalize_row_to_schema(row, schema);
                        let (updated_row, changed) = setter(row)?;
                        if changed {
                            rows_with_originals.push((row_id, updated_row, version));
                        }
                    }
                }

                let update_count = (local_rows.len() + rows_with_originals.len()) as i32;
                if !local_rows.is_empty() || !rows_with_originals.is_empty() {
                    let mut txn_versions = self.txn_versions.write().unwrap();
                    if !local_rows.is_empty() {
                        txn_versions.put_batch_for_update(local_rows)?;
                    }
                    if !rows_with_originals.is_empty() {
                        txn_versions.put_batch_with_originals(rows_with_originals)?;
                    }
                }
                return Ok(update_count);
            }

            // Try index lookup for non-PK columns
            if let Some(filtered_row_ids) = self.try_index_lookup(expr, schema) {
                // Step 1: Check local versions first (these don't need write-set tracking)
                // Use get_local_version to distinguish "no local version" from "locally deleted"
                let mut local_rows_to_update = RowVec::with_capacity(filtered_row_ids.len() / 4);
                let mut remaining_row_ids: Vec<i64> = Vec::with_capacity(filtered_row_ids.len());

                {
                    let txn_versions = self.txn_versions.read().unwrap();
                    for &row_id in &filtered_row_ids {
                        if let Some(local) = txn_versions.get_local_version(row_id) {
                            if !local.is_deleted() {
                                let row = self.normalize_row_to_schema(local.data.clone(), schema);
                                // Re-apply filter
                                if expr.evaluate(&row).unwrap_or(false) {
                                    local_rows_to_update.push((row_id, row));
                                }
                            }
                            // Locally deleted — skip, don't fall through
                        } else {
                            remaining_row_ids.push(row_id);
                        }
                    }
                }

                // Step 2: Batch fetch remaining rows from version store WITH original versions
                // This avoids redundant get_visible_version() calls during put
                // OPTIMIZATION: Pre-allocate with known capacity
                let mut rows_with_originals: Vec<(i64, Row, crate::storage::mvcc::RowVersion)> =
                    Vec::with_capacity(remaining_row_ids.len());
                if !remaining_row_ids.is_empty() {
                    let batch_rows = self
                        .version_store
                        .get_visible_versions_for_update(&remaining_row_ids, self.txn_id);
                    for (row_id, row, original) in batch_rows {
                        // Normalize row to match current schema (handles ALTER TABLE ADD/DROP COLUMN)
                        let row = self.normalize_row_to_schema(row, schema);
                        // Re-apply filter (index may be partial match)
                        if expr.evaluate(&row).unwrap_or(false) {
                            rows_with_originals.push((row_id, row, original));
                        }
                    }
                }

                // Step 3: Apply setter to all rows, filtering out unchanged rows
                // Setter returns Result — on error, abort BEFORE batch put
                // to guarantee statement-level atomicity.
                let mut setter_error: Option<crate::core::Error> = None;
                local_rows_to_update.retain_mut(|(_, row)| {
                    if setter_error.is_some() {
                        return false;
                    }
                    match setter(std::mem::take(row)) {
                        Ok((updated_row, changed)) => {
                            *row = updated_row;
                            changed
                        }
                        Err(e) => {
                            setter_error = Some(e);
                            false
                        }
                    }
                });

                // Update rows from version store with pre-fetched originals
                if setter_error.is_none() {
                    rows_with_originals.retain_mut(|(_, row, _)| {
                        if setter_error.is_some() {
                            return false;
                        }
                        match setter(std::mem::take(row)) {
                            Ok((updated_row, changed)) => {
                                *row = updated_row;
                                changed
                            }
                            Err(e) => {
                                setter_error = Some(e);
                                false
                            }
                        }
                    });
                }

                // Abort before batch put if setter reported an error
                if let Some(err) = setter_error {
                    return Err(err);
                }

                let update_count = local_rows_to_update.len() + rows_with_originals.len();

                // Batch put - first the local rows (use regular put)
                {
                    let mut txn_versions = self.txn_versions.write().unwrap();
                    txn_versions.put_batch_for_update(local_rows_to_update)?;
                    // Then the rows with originals (use optimized put)
                    txn_versions.put_batch_with_originals(rows_with_originals)?;
                }
                return Ok(update_count as i32);
            }
        }

        // Fall back to full scan - use batch fetch WITH original versions for O(1) lock
        // OPTIMIZATION: When WHERE clause exists, push filter to storage layer
        // This avoids allocating Row objects for non-matching rows
        let mut rows_with_originals: Vec<(i64, Row, crate::storage::mvcc::RowVersion)> =
            if let Some(expr) = where_expr {
                // Use filtered scan - filter is applied BEFORE cloning rows
                self.version_store
                    .get_all_visible_rows_for_update_filtered(self.txn_id, expr)
                    .into_iter()
                    .map(|(row_id, row, orig)| {
                        // Normalize row to match current schema (handles ALTER TABLE ADD/DROP COLUMN)
                        (row_id, self.normalize_row_to_schema(row, schema), orig)
                    })
                    .collect()
            } else {
                // No filter - get all rows
                self.version_store
                    .get_all_visible_rows_for_update(self.txn_id)
                    .into_iter()
                    .map(|(row_id, row, orig)| {
                        // Normalize row to match current schema (handles ALTER TABLE ADD/DROP COLUMN)
                        (row_id, self.normalize_row_to_schema(row, schema), orig)
                    })
                    .collect()
            };

        // Overlay local transaction changes onto the global rows:
        // - Rows locally updated: use local data instead of stale global data
        // - Rows locally deleted: remove from update set
        // - Rows locally inserted (not in global store): add to update set
        let local_rows_to_update: RowVec = {
            let txn_versions = self.txn_versions.read().unwrap();
            if txn_versions.has_local_changes() {
                // Fix up global rows that have local modifications
                rows_with_originals.retain_mut(|(row_id, row, _orig)| {
                    if let Some(local_version) = txn_versions.get_latest_local(*row_id) {
                        if local_version.is_deleted() {
                            // Locally deleted — exclude from update
                            return false;
                        }
                        // Locally modified — use local data instead of stale global data
                        *row = self.normalize_row_to_schema(local_version.data.clone(), schema);
                        // Re-check filter against local data
                        if let Some(expr) = where_expr {
                            if !expr.evaluate(row).unwrap_or(false) {
                                return false;
                            }
                        }
                    }
                    true
                });
            }

            // Collect local-only inserts (not in global store)
            txn_versions
                .iter_local()
                .filter_map(|(row_id, version)| {
                    // Skip if already in global store (processed above)
                    if self.version_store.quick_check_row_existence(row_id) {
                        return None;
                    }
                    if version.is_deleted() {
                        return None;
                    }
                    let row = self.normalize_row_to_schema(version.data.clone(), schema);
                    if let Some(expr) = where_expr {
                        if !expr.evaluate(&row).unwrap_or(false) {
                            return None;
                        }
                    }
                    Some((row_id, row))
                })
                .collect()
        };

        // Apply setter to rows with originals (from version store), filtering out unchanged
        // Setter returns Result — on error, abort BEFORE batch put for statement atomicity.
        let mut setter_error: Option<crate::core::Error> = None;
        rows_with_originals.retain_mut(|(_, row, _)| {
            if setter_error.is_some() {
                return false;
            }
            match setter(std::mem::take(row)) {
                Ok((updated_row, changed)) => {
                    *row = updated_row;
                    changed
                }
                Err(e) => {
                    setter_error = Some(e);
                    false
                }
            }
        });

        // Apply setter to local rows, filtering out unchanged
        let mut local_updated: RowVec = RowVec::new();
        if setter_error.is_none() {
            for (row_id, row) in local_rows_to_update {
                match setter(row) {
                    Ok((updated_row, changed)) => {
                        if changed {
                            local_updated.push((row_id, updated_row));
                        }
                    }
                    Err(e) => {
                        setter_error = Some(e);
                        break;
                    }
                }
            }
        }

        // Abort before batch put if setter reported an error
        if let Some(err) = setter_error {
            return Err(err);
        }

        // Batch update all rows at once
        let update_count = rows_with_originals.len() + local_updated.len();
        {
            let mut txn_versions = self.txn_versions.write().unwrap();
            // Use optimized put for rows from version store (avoids O(N) get_visible_version calls)
            txn_versions.put_batch_with_originals(rows_with_originals)?;
            // Use regular put for local rows (already tracked in local store)
            txn_versions.put_batch_for_update(local_updated)?;
        }

        Ok(update_count as i32)
    }

    fn update_by_row_ids(
        &mut self,
        row_ids: &[i64],
        setter: &mut dyn FnMut(Row) -> Result<(Row, bool)>,
    ) -> Result<i32> {
        let schema = &self.cached_schema;

        // Step 1: Check local versions first (single lock acquisition)
        // Use get_local_version to distinguish "no local version" from "locally deleted"
        let mut local_rows = RowVec::with_capacity(row_ids.len() / 4);
        let mut remaining_row_ids: Vec<i64> = Vec::with_capacity(row_ids.len());

        {
            let txn_versions = self.txn_versions.read().unwrap();
            for &row_id in row_ids {
                if let Some(local) = txn_versions.get_local_version(row_id) {
                    if !local.is_deleted() {
                        let row = self.normalize_row_to_schema(local.data.clone(), schema);
                        let (updated_row, changed) = setter(row)?;
                        if changed {
                            local_rows.push((row_id, updated_row));
                        }
                    }
                    // Locally deleted — skip, don't fall through
                } else {
                    remaining_row_ids.push(row_id);
                }
            }
        }

        // Step 2: Batch fetch remaining from version store with original versions
        let mut rows_with_originals: Vec<(i64, Row, crate::storage::mvcc::RowVersion)> =
            Vec::with_capacity(remaining_row_ids.len());
        if !remaining_row_ids.is_empty() {
            let batch_rows = self
                .version_store
                .get_visible_versions_for_update(&remaining_row_ids, self.txn_id);
            for (row_id, row, version) in batch_rows {
                let row = self.normalize_row_to_schema(row, schema);
                let (updated_row, changed) = setter(row)?;
                if changed {
                    rows_with_originals.push((row_id, updated_row, version));
                }
            }
        }

        // Step 3: Batch put all updates
        let update_count = (local_rows.len() + rows_with_originals.len()) as i32;
        if !local_rows.is_empty() || !rows_with_originals.is_empty() {
            let mut txn_versions = self.txn_versions.write().unwrap();
            if !local_rows.is_empty() {
                txn_versions.put_batch_for_update(local_rows)?;
            }
            if !rows_with_originals.is_empty() {
                txn_versions.put_batch_with_originals(rows_with_originals)?;
            }
        }

        Ok(update_count)
    }

    fn delete_by_row_ids(&mut self, row_ids: &[i64]) -> Result<i32> {
        let schema = &self.cached_schema;

        // Step 1: Check local versions first
        // Use get_local_version to distinguish "no local version" from "locally deleted"
        let mut local_deletes = RowVec::with_capacity(row_ids.len() / 4);
        let mut remaining_row_ids: Vec<i64> = Vec::with_capacity(row_ids.len());

        {
            let txn_versions = self.txn_versions.read().unwrap();
            for &row_id in row_ids {
                if let Some(local) = txn_versions.get_local_version(row_id) {
                    if !local.is_deleted() {
                        let row = self.normalize_row_to_schema(local.data.clone(), schema);
                        local_deletes.push((row_id, row));
                    }
                    // Already locally deleted — skip, don't fall through
                } else {
                    remaining_row_ids.push(row_id);
                }
            }
        }

        // Step 2: Batch fetch remaining from version store
        let mut rows_with_originals: Vec<(i64, Row, crate::storage::mvcc::RowVersion)> =
            Vec::with_capacity(remaining_row_ids.len());
        // Track which remaining row_ids have hot versions (cold-only rows won't)
        let mut found_ids = rustc_hash::FxHashSet::default();
        if !remaining_row_ids.is_empty() {
            let batch_rows = self
                .version_store
                .get_visible_versions_for_update(&remaining_row_ids, self.txn_id);
            for (row_id, row, version) in batch_rows {
                found_ids.insert(row_id);
                let row = self.normalize_row_to_schema(row, schema);
                rows_with_originals.push((row_id, row, version));
            }
        }

        // Cold-only row_ids: not in local versions and not in the hot version store.
        // Create phantom delete markers so the WAL records the deletion, enabling
        // tombstone recovery on restart.
        let cold_only_ids: Vec<i64> = remaining_row_ids
            .iter()
            .filter(|id| !found_ids.contains(id))
            .copied()
            .collect();

        // Step 3: Claim cold-only rows before acquiring txn_versions lock.
        // Prevents concurrent DELETE + UPDATE from both committing (lost update).
        // The claim is tracked in write_set via track_external_claim.
        for row_id in &cold_only_ids {
            self.try_claim_row(*row_id)?;
        }

        // Step 4: Batch delete all rows
        let delete_count =
            (local_deletes.len() + rows_with_originals.len() + cold_only_ids.len()) as i32;
        if !local_deletes.is_empty() || !rows_with_originals.is_empty() || !cold_only_ids.is_empty()
        {
            let mut txn_versions = self.txn_versions.write().unwrap();
            for (row_id, row) in local_deletes {
                txn_versions.put(row_id, row, true)?;
            }
            for (row_id, row, orig) in rows_with_originals {
                txn_versions.put_with_original(row_id, row, orig, true)?;
            }
            for row_id in cold_only_ids {
                txn_versions.put(row_id, Row::new(), true)?;
            }
        }

        Ok(delete_count)
    }

    fn get_active_row_ids(&self) -> Vec<i64> {
        self.version_store.get_all_row_ids()
    }

    fn collect_hot_row_ids_into(&self, dest: &mut rustc_hash::FxHashSet<i64>) {
        self.version_store.collect_row_ids_into(dest);
    }

    fn has_row_id(&self, row_id: i64) -> bool {
        self.version_store.has_committed_row(row_id)
    }

    fn try_claim_row(&self, row_id: i64) -> Result<()> {
        self.version_store
            .try_claim_row(row_id, self.txn_id)
            .map_err(|e| Error::internal(e.to_string()))?;
        // Track this claim in TransactionVersionStore's write_set so that
        // commit/rollback releases it. Without this, claims made directly
        // on VersionStore (for cold row UPDATE/DELETE) are never released
        // because TransactionVersionStore::commit() only drains write_set.
        self.txn_versions
            .write()
            .unwrap()
            .track_external_claim(row_id);
        Ok(())
    }

    fn delete(&mut self, where_expr: Option<&dyn Expression>) -> Result<i32> {
        // OPTIMIZATION: Borrow schema instead of cloning - saves allocation per delete
        let schema = &self.cached_schema;

        // Fast path: Check if this is a primary key equality lookup (WHERE id = X)
        if let Some(expr) = where_expr {
            if let Some(pk_id) = self.try_pk_lookup(expr, schema) {
                // Direct O(1) lookup by primary key
                // Use get_local_version to preserve delete signal. txn_versions.get()
                // swallows deletes as None, causing fallback to committed store and
                // missing uncommitted deletes in the current transaction.
                let row_with_original = {
                    let txn_versions = self.txn_versions.read().unwrap();
                    if let Some(local) = txn_versions.get_local_version(pk_id) {
                        if local.is_deleted() {
                            None // Already locally deleted — don't double-count
                        } else {
                            // Local version - no need to track original (already in write-set)
                            Some((local.data.clone(), None))
                        }
                    } else if let Some(version) =
                        self.version_store.get_visible_version(pk_id, self.txn_id)
                    {
                        if !version.is_deleted() {
                            let data = version.data.clone();
                            Some((data, Some(version)))
                        } else {
                            None
                        }
                    } else {
                        None
                    }
                };

                if let Some((row, original_version)) = row_with_original {
                    if let Some(orig) = original_version {
                        // Use optimized put that skips redundant get_visible_version
                        self.txn_versions
                            .write()
                            .unwrap()
                            .put_with_original(pk_id, row, orig, true)?;
                    } else {
                        // Local version - use regular put
                        self.txn_versions.write().unwrap().put(pk_id, row, true)?;
                    }
                    return Ok(1);
                }
                return Ok(0);
            }

            // Fast path: PK range lookup (WHERE id >= X AND id < Y)
            // PK IS the row_id, so we can generate the range directly
            if let Some(pk_range_ids) = self.try_pk_range_lookup(expr, schema) {
                // OPTIMIZATION: Track original versions to avoid redundant lookups
                let mut local_rows = RowVec::with_capacity(pk_range_ids.len() / 4);
                let mut rows_with_originals: Vec<(i64, Row, crate::storage::mvcc::RowVersion)> =
                    Vec::with_capacity(pk_range_ids.len());

                // Step 1: Check local versions first (single lock acquisition)
                // Use get_local_version to distinguish "no local version" from "locally deleted"
                let mut remaining_row_ids: Vec<i64> = Vec::with_capacity(pk_range_ids.len());
                {
                    let txn_versions = self.txn_versions.read().unwrap();
                    for row_id in pk_range_ids {
                        if let Some(local) = txn_versions.get_local_version(row_id) {
                            if !local.is_deleted() {
                                local_rows.push((row_id, local.data.clone()));
                            }
                            // Already locally deleted — skip, don't fall through
                        } else {
                            remaining_row_ids.push(row_id);
                        }
                    }
                }

                // Step 2: Batch fetch remaining from version store (1 lock for N rows)
                if !remaining_row_ids.is_empty() {
                    let batch_rows = self
                        .version_store
                        .get_visible_versions_for_update(&remaining_row_ids, self.txn_id);
                    for (row_id, row, version) in batch_rows {
                        rows_with_originals.push((row_id, row, version));
                    }
                }

                let delete_count = (local_rows.len() + rows_with_originals.len()) as i32;
                if !local_rows.is_empty() || !rows_with_originals.is_empty() {
                    let mut txn_versions = self.txn_versions.write().unwrap();
                    if !local_rows.is_empty() {
                        txn_versions.put_batch_deleted(local_rows)?;
                    }
                    if !rows_with_originals.is_empty() {
                        txn_versions.put_batch_deleted_with_originals(rows_with_originals)?;
                    }
                }
                return Ok(delete_count);
            }

            // Try index lookup for non-PK columns
            if let Some(filtered_row_ids) = self.try_index_lookup(expr, schema) {
                // OPTIMIZATION: Batch collect rows to delete, then batch put
                // This reduces lock contention from 2N locks to 2 locks for N rows
                let mut rows_to_delete = RowVec::with_capacity(filtered_row_ids.len());

                // Step 1: Check local versions first (single lock acquisition)
                // Use get_local_version to distinguish "no local version" from "locally deleted"
                let mut remaining_row_ids: Vec<i64> = Vec::with_capacity(filtered_row_ids.len());
                {
                    let txn_versions = self.txn_versions.read().unwrap();
                    for row_id in filtered_row_ids {
                        if let Some(local) = txn_versions.get_local_version(row_id) {
                            if !local.is_deleted() {
                                let row = local.data.clone();
                                // Re-apply filter (index may be partial match)
                                if expr.evaluate(&row).unwrap_or(false) {
                                    rows_to_delete.push((row_id, row));
                                }
                            }
                            // Already locally deleted — skip, don't fall through
                        } else {
                            remaining_row_ids.push(row_id);
                        }
                    }
                }

                // Step 2: Batch fetch remaining from version store (1 lock for N rows)
                if !remaining_row_ids.is_empty() {
                    let batch_rows = self
                        .version_store
                        .get_visible_versions_batch(&remaining_row_ids, self.txn_id);
                    for (row_id, row) in batch_rows {
                        // Re-apply filter (index may be partial match)
                        if expr.evaluate(&row).unwrap_or(false) {
                            rows_to_delete.push((row_id, row));
                        }
                    }
                }

                // Single batch write for all deletes
                let delete_count = rows_to_delete.len() as i32;
                if !rows_to_delete.is_empty() {
                    self.txn_versions
                        .write()
                        .unwrap()
                        .put_batch_deleted(rows_to_delete)?;
                }
                return Ok(delete_count);
            }
        }

        // Fall back to full scan
        let mut delete_count = 0;

        // Get all visible rows
        let row_ids = self.version_store.get_all_row_ids();

        for row_id in row_ids {
            // OPTIMIZATION: Check filter BEFORE cloning to avoid wasted allocations
            // For DELETE with selective WHERE, this can save 90%+ of clones

            // First, check local versions
            // Use get_local_version to distinguish "no local version" from "locally deleted"
            let local_version = {
                let txn_versions = self.txn_versions.read().unwrap();
                txn_versions
                    .get_local_version(row_id)
                    .map(|v| (v.is_deleted(), v.data.clone()))
            };

            if let Some((is_deleted, row)) = local_version {
                if is_deleted {
                    continue; // Already locally deleted — skip
                }
                // Apply filter on local row
                if let Some(expr) = where_expr {
                    match expr.evaluate(&row) {
                        Ok(true) => {}
                        Ok(false) => continue,
                        Err(_) => continue,
                    }
                }
                // Mark as deleted
                self.txn_versions.write().unwrap().put(row_id, row, true)?;
                delete_count += 1;
            } else if let Some(version) =
                self.version_store.get_visible_version(row_id, self.txn_id)
            {
                if version.is_deleted() {
                    continue;
                }

                // Apply filter BEFORE cloning - evaluate on reference
                if let Some(expr) = where_expr {
                    match expr.evaluate(&version.data) {
                        Ok(true) => {}
                        Ok(false) => continue, // Skip clone entirely!
                        Err(_) => continue,
                    }
                }

                // Only clone AFTER filter passes
                self.txn_versions
                    .write()
                    .unwrap()
                    .put(row_id, version.data.clone(), true)?;
                delete_count += 1;
            }
        }

        // Also check local inserts that might not be in global store
        let local_ids: Vec<i64> = {
            let txn_versions = self.txn_versions.read().unwrap();
            txn_versions.iter_local().map(|(id, _)| id).collect()
        };
        for row_id in local_ids {
            // Skip if already processed
            if self.version_store.quick_check_row_existence(row_id) {
                continue;
            }

            let row = {
                let txn_versions = self.txn_versions.read().unwrap();
                txn_versions.get(row_id)
            };
            if let Some(row) = row {
                // Apply filter
                if let Some(expr) = where_expr {
                    match expr.evaluate(&row) {
                        Ok(true) => {}
                        Ok(false) => continue,
                        Err(_) => continue,
                    }
                }

                // Row is already owned from txn_versions.get(), no extra clone needed
                self.txn_versions.write().unwrap().put(row_id, row, true)?;
                delete_count += 1;
            }
        }

        Ok(delete_count)
    }

    fn truncate(&mut self) -> Result<i32> {
        // Fast path: drop all storage directly, bypassing per-row MVCC versioning.
        // This is O(1) instead of O(N) for delete-all.
        // Fails if other transactions have uncommitted writes on this table.
        self.version_store.truncate_all()
    }

    fn scan(
        &self,
        column_indices: &[usize],
        where_expr: Option<&dyn Expression>,
    ) -> Result<Box<dyn Scanner>> {
        // CompactArc clone - O(1) reference count increment instead of full Schema clone
        let schema = self.cached_schema.clone();

        // Fast path: Check if this is a primary key equality lookup (WHERE id = X)
        if let Some(expr) = where_expr {
            if let Some(pk_lookup) = self.try_pk_lookup(expr, &schema) {
                // Direct O(1) lookup by primary key
                // Use get_local_version to preserve delete signal. txn_versions.get()
                // swallows deletes as None, causing fallback to committed store and
                // missing uncommitted deletes in the current transaction.
                let row = {
                    let txn_versions = self.txn_versions.read().unwrap();
                    if let Some(local) = txn_versions.get_local_version(pk_lookup) {
                        if local.is_deleted() {
                            None // Locally deleted in this transaction
                        } else {
                            Some(local.data.clone())
                        }
                    } else if let Some(version) = self
                        .version_store
                        .get_visible_version(pk_lookup, self.txn_id)
                    {
                        if !version.is_deleted() {
                            Some(version.data.clone())
                        } else {
                            None
                        }
                    } else {
                        None
                    }
                };

                if let Some(row) = row {
                    // Normalize row to match current schema (handles ALTER TABLE ADD/DROP COLUMN)
                    let row = self.normalize_row_to_schema(row, &schema);
                    let mut rows = RowVec::with_capacity(1);
                    rows.push((pk_lookup, row));
                    let scanner = MVCCScanner::from_rows(rows, schema, column_indices.to_vec());
                    return Ok(Box::new(scanner));
                } else {
                    // Row not found - return empty scanner
                    let scanner = MVCCScanner::empty(schema, column_indices.to_vec());
                    return Ok(Box::new(scanner));
                }
            }

            // Try index lookup for non-PK columns
            if let Some(filtered_row_ids) = self.try_index_lookup(expr, &schema) {
                // Use index-based scan - much more efficient
                let rows = self.fetch_rows_by_ids(&filtered_row_ids, expr);
                let scanner = MVCCScanner::from_rows(rows, schema, column_indices.to_vec());
                return Ok(Box::new(scanner));
            }
        }

        // Fall back to full scan - use MVCCScanner with RowVec for cache reuse
        let rows = self.collect_visible_rows(where_expr);
        let scanner = MVCCScanner::from_rows(rows, schema, column_indices.to_vec());
        Ok(Box::new(scanner))
    }

    fn collect_all_rows(&self, where_expr: Option<&dyn Expression>) -> Result<RowVec> {
        // Return cached row vector directly - caller iterates (i64, Row) tuples
        Ok(self.collect_visible_rows(where_expr))
    }

    fn collect_all_rows_unsorted(&self) -> Result<RowVec> {
        Ok(self.collect_visible_rows_unsorted())
    }

    fn collect_rows_by_ids(&self, row_ids: &[i64]) -> Result<RowVec> {
        let mut rows = RowVec::with_capacity(row_ids.len());
        for &row_id in row_ids {
            if let Some(version) = self.version_store.get_visible_version(row_id, self.txn_id) {
                if !version.is_deleted() {
                    rows.push((row_id, version.data.clone()));
                }
            }
        }
        Ok(rows)
    }

    fn collect_rows_with_limit(
        &self,
        where_expr: Option<&dyn Expression>,
        limit: usize,
        offset: usize,
    ) -> Result<RowVec> {
        // Use the optimized version with limit/offset
        Ok(self.collect_visible_rows_with_limit(where_expr, limit, offset))
    }

    fn collect_rows_with_limit_unordered(
        &self,
        where_expr: Option<&dyn Expression>,
        limit: usize,
        offset: usize,
    ) -> Result<RowVec> {
        // Use the optimized unordered version with true early termination
        Ok(self.collect_visible_rows_with_limit_unordered(where_expr, limit, offset))
    }

    fn collect_rows_sorted_with_limit(
        &self,
        sort_col_idx: usize,
        ascending: bool,
        limit: usize,
        offset: usize,
    ) -> Result<Vec<Row>> {
        // DEFERRED MATERIALIZATION OPTIMIZATION
        // Instead of cloning all rows, sorting, and taking limit:
        // 1. Get row indices (no cloning)
        // 2. Load only sort column values
        // 3. Sort indices by values
        // 4. Take top N indices
        // 5. Materialize only N rows
        //
        // Performance: For 100K rows with 20 columns, LIMIT 10:
        // - Old: Clone 2M values, sort, take 10
        // - New: Load 100K sort values, sort indices, clone 200 values

        // Check for local versions - if present, fall back to default implementation
        let has_local = self.txn_versions.read().unwrap().has_local_changes();
        if has_local {
            // Local changes exist - use slower but correct path
            let mut rows = self.collect_visible_rows(None);
            rows.sort_by(|(_, a), (_, b)| {
                let va = a.get(sort_col_idx);
                let vb = b.get(sort_col_idx);
                let cmp = match (va, vb) {
                    (None, None) => std::cmp::Ordering::Equal,
                    (None, Some(_)) => std::cmp::Ordering::Less,
                    (Some(_), None) => std::cmp::Ordering::Greater,
                    (Some(va), Some(vb)) => va.compare(vb).unwrap_or(std::cmp::Ordering::Equal),
                };
                if ascending {
                    cmp
                } else {
                    cmp.reverse()
                }
            });
            return Ok(rows
                .into_iter()
                .skip(offset)
                .take(limit)
                .map(|(_, row)| row)
                .collect());
        }

        // FAST PATH: Use deferred materialization from version_store
        let rows = self.version_store.get_visible_rows_sorted_limit(
            self.txn_id,
            sort_col_idx,
            ascending,
            limit,
            offset,
        );

        // Normalize rows to schema and return
        let schema = &self.cached_schema;
        Ok(rows
            .into_iter()
            .map(|(_, row)| self.normalize_row_to_schema(row, schema))
            .collect())
    }

    fn close(&mut self) -> Result<()> {
        // Rollback any uncommitted changes
        self.txn_versions.write().unwrap().rollback();
        Ok(())
    }

    fn commit(&mut self) -> Result<()> {
        // Call inherent method which handles index updates
        MVCCTable::commit(self)
    }

    fn rollback(&mut self) {
        self.txn_versions.write().unwrap().rollback();
    }

    fn rollback_to_timestamp(&self, timestamp: i64) {
        self.txn_versions
            .write()
            .unwrap()
            .rollback_to_timestamp(timestamp);
    }

    fn has_local_changes(&self) -> bool {
        self.txn_versions.read().unwrap().has_local_changes()
    }

    fn get_pending_versions(&self) -> Vec<(i64, Row, bool, i64)> {
        let txn_versions = self.txn_versions.read().unwrap();
        txn_versions
            .iter_local()
            .map(|(row_id, version)| {
                (
                    row_id,
                    version.data.clone(),
                    version.is_deleted(),
                    version.txn_id,
                )
            })
            .collect()
    }

    fn create_index(&self, name: &str, columns: &[&str], is_unique: bool) -> Result<()> {
        // Delegate to create_index_with_type with auto-selection
        self.create_index_with_type(name, columns, is_unique, None)
    }

    fn create_index_with_type(
        &self,
        name: &str,
        columns: &[&str],
        is_unique: bool,
        index_type: Option<IndexType>,
    ) -> Result<()> {
        if columns.is_empty() {
            return Err(Error::internal("index must have at least one column"));
        }

        let schema = self.version_store.schema();

        // Collect column info
        let mut column_names = Vec::with_capacity(columns.len());
        let mut column_ids = Vec::with_capacity(columns.len());
        let mut data_types = Vec::with_capacity(columns.len());
        let mut col_indices = Vec::with_capacity(columns.len());

        for col_name in columns {
            let (col_idx, col) = schema
                .find_column(col_name)
                .ok_or(Error::ColumnNotFound(col_name.to_string()))?;
            column_names.push(col.name.clone());
            column_ids.push(col.id as i32);
            data_types.push(col.data_type);
            col_indices.push(col_idx);
        }

        // Determine index type: use explicit type or auto-select based on column types
        // For multi-column indexes, always use MultiColumn (hash+btree hybrid)
        let chosen_type = if columns.len() > 1 {
            IndexType::MultiColumn
        } else {
            index_type.unwrap_or_else(|| Self::auto_select_index_type(&data_types))
        };

        // Check if index with same name already exists
        if self.version_store.index_exists(name) {
            return Err(Error::IndexAlreadyExists(name.to_string()));
        }

        // Check if an index already exists on the same column(s)
        // For single-column indexes, use get_index_by_column
        // For multi-column indexes, check all existing indexes
        if columns.len() == 1 {
            if let Some(existing_idx) = self.version_store.get_index_by_column(columns[0]) {
                // If the existing index is a PkIndex, silently skip -
                // the PK column is already covered
                if existing_idx.index_type() == IndexType::PrimaryKey {
                    return Ok(());
                }
                if is_unique && !existing_idx.is_unique() {
                    return Err(Error::internal(format!(
                        "cannot create unique index on column '{}': a non-unique index already exists",
                        columns[0]
                    )));
                } else if !is_unique && existing_idx.is_unique() {
                    return Err(Error::internal(format!(
                        "cannot create non-unique index on column '{}': a unique index already exists",
                        columns[0]
                    )));
                }
                // Same type of index on same column - also reject
                return Err(Error::internal(format!(
                    "an index already exists on column '{}'",
                    columns[0]
                )));
            }
        } else {
            // For multi-column indexes, check if any existing index has the exact same columns
            for existing_idx in self.version_store.get_all_indexes() {
                let existing_cols = existing_idx.column_names();
                if existing_cols.len() == columns.len() {
                    let same_cols = existing_cols
                        .iter()
                        .zip(columns.iter())
                        .all(|(a, b)| a == *b);
                    if same_cols {
                        if is_unique && !existing_idx.is_unique() {
                            return Err(Error::internal(format!(
                                "cannot create unique index on columns {:?}: a non-unique index already exists",
                                columns
                            )));
                        } else if !is_unique && existing_idx.is_unique() {
                            return Err(Error::internal(format!(
                                "cannot create non-unique index on columns {:?}: a unique index already exists",
                                columns
                            )));
                        }
                        return Err(Error::internal(format!(
                            "an index already exists on columns {:?}",
                            columns
                        )));
                    }
                }
            }
        }

        // Get row count for capacity hint
        let expected_rows = self.version_store.row_count();

        // Create the appropriate index type with capacity hint
        let index: Arc<dyn Index> = match chosen_type {
            IndexType::Hash => Arc::new(HashIndex::new(
                name.to_string(),
                self.name().to_string(),
                column_names,
                column_ids,
                data_types,
                is_unique,
                expected_rows,
            )),
            IndexType::Bitmap => Arc::new(BitmapIndex::new(
                name.to_string(),
                self.name().to_string(),
                column_names,
                column_ids,
                data_types,
                is_unique,
                expected_rows,
            )),
            IndexType::BTree => {
                // For single-column BTree, use BTreeIndex
                // For multi-column, use MultiColumnIndex
                if columns.len() == 1 {
                    Arc::new(BTreeIndex::new(
                        name.to_string(),
                        self.name().to_string(),
                        column_ids[0],
                        column_names[0].clone(),
                        data_types[0],
                        is_unique,
                        expected_rows,
                    ))
                } else {
                    Arc::new(MultiColumnIndex::new(
                        name.to_string(),
                        self.name().to_string(),
                        column_names,
                        column_ids,
                        data_types,
                        is_unique,
                        expected_rows,
                    ))
                }
            }
            IndexType::MultiColumn => {
                // MultiColumn always uses MultiColumnIndex
                Arc::new(MultiColumnIndex::new(
                    name.to_string(),
                    self.name().to_string(),
                    column_names,
                    column_ids,
                    data_types,
                    is_unique,
                    expected_rows,
                ))
            }
            IndexType::PrimaryKey => {
                // PrimaryKey indexes are auto-created, users cannot CREATE INDEX with this type
                return Err(Error::internal(
                    "cannot explicitly create a primary key index; use PRIMARY KEY constraint instead".to_string(),
                ));
            }
            IndexType::Hnsw => {
                // HNSW only supports single-column vector indexes
                if columns.len() != 1 {
                    return Err(Error::internal(
                        "HNSW index must be on a single vector column".to_string(),
                    ));
                }
                if data_types[0] != DataType::Vector {
                    return Err(Error::internal(format!(
                        "HNSW index requires a VECTOR column, got {:?}",
                        data_types[0]
                    )));
                }
                // Get vector dimensions from schema column
                let (_, col) = schema
                    .find_column(columns[0])
                    .ok_or(Error::ColumnNotFound(columns[0].to_string()))?;
                let dims = col.vector_dimensions as usize;
                if dims == 0 {
                    return Err(Error::internal(
                        "HNSW index requires a VECTOR column with specified dimensions".to_string(),
                    ));
                }
                let default_m = crate::storage::index::default_m_for_dims(dims);
                let mut hnsw = HnswIndex::new(
                    name.to_string(),
                    self.name().to_string(),
                    column_names[0].clone(),
                    column_ids[0],
                    dims,
                    default_m,
                    crate::storage::index::default_ef_construction(default_m),
                    crate::storage::index::default_ef_search(default_m),
                    crate::storage::index::HnswDistanceMetric::L2,
                );
                hnsw.set_unique(is_unique);
                Arc::new(hnsw)
            }
        };

        // Populate the index with existing data using batch_slice for better performance
        // Collect all entries first, then add in a single batch operation
        let mut entries: Vec<(i64, Vec<Value>)> = Vec::new();

        for row_id in self.version_store.get_all_visible_row_ids(self.txn_id) {
            if let Some(version) = self.version_store.get_visible_version(row_id, self.txn_id) {
                if !version.is_deleted() {
                    let values: Vec<Value> = col_indices
                        .iter()
                        .map(|&idx| {
                            version
                                .data
                                .get(idx)
                                .cloned()
                                .unwrap_or(Value::Null(DataType::Null))
                        })
                        .collect();
                    entries.push((row_id, values));
                }
            }
        }

        // Also collect any local uncommitted data
        {
            let txn_versions = self.txn_versions.read().unwrap();
            for (row_id, version) in txn_versions.iter_local() {
                if !version.is_deleted() && !self.version_store.quick_check_row_existence(row_id) {
                    let values: Vec<Value> = col_indices
                        .iter()
                        .map(|&idx| {
                            version
                                .data
                                .get(idx)
                                .cloned()
                                .unwrap_or(Value::Null(DataType::Null))
                        })
                        .collect();
                    entries.push((row_id, values));
                }
            }
        }

        // Add all entries in a single batch operation
        if !entries.is_empty() {
            let entry_refs: Vec<(i64, &[Value])> = entries
                .iter()
                .map(|(row_id, values)| (*row_id, values.as_slice()))
                .collect();
            index.add_batch_slice(&entry_refs)?;
        }

        // Add to version store
        self.version_store.add_index(name.to_string(), index);

        Ok(())
    }

    fn create_hnsw_index(
        &self,
        name: &str,
        column: &str,
        is_unique: bool,
        m: usize,
        ef_construction: usize,
        ef_search: usize,
        metric: crate::storage::index::HnswDistanceMetric,
    ) -> Result<()> {
        let schema = self.version_store.schema();
        let (col_idx, col) = schema
            .find_column(column)
            .ok_or(Error::ColumnNotFound(column.to_string()))?;

        if col.data_type != DataType::Vector {
            return Err(Error::internal(format!(
                "HNSW index requires a VECTOR column, got {:?}",
                col.data_type
            )));
        }
        let dims = col.vector_dimensions as usize;
        if dims == 0 {
            return Err(Error::internal(
                "HNSW index requires a VECTOR column with specified dimensions".to_string(),
            ));
        }

        // Check for duplicate
        if self.version_store.index_exists(name) {
            return Err(Error::IndexAlreadyExists(name.to_string()));
        }
        if let Some(existing_idx) = self.version_store.get_index_by_column(column) {
            if existing_idx.index_type() == IndexType::PrimaryKey {
                return Ok(());
            }
            return Err(Error::internal(format!(
                "an index already exists on column '{}'",
                column
            )));
        }

        let mut hnsw = HnswIndex::new(
            name.to_string(),
            self.name().to_string(),
            col.name.clone(),
            col.id as i32,
            dims,
            m,
            ef_construction,
            ef_search,
            metric,
        );
        hnsw.set_unique(is_unique);
        let index: Arc<dyn Index> = Arc::new(hnsw);

        // Populate with existing data
        let mut entries: Vec<(i64, Vec<Value>)> = Vec::new();
        for row_id in self.version_store.get_all_visible_row_ids(self.txn_id) {
            if let Some(version) = self.version_store.get_visible_version(row_id, self.txn_id) {
                if !version.is_deleted() {
                    let val = version
                        .data
                        .get(col_idx)
                        .cloned()
                        .unwrap_or(Value::Null(DataType::Null));
                    entries.push((row_id, vec![val]));
                }
            }
        }
        {
            let txn_versions = self.txn_versions.read().unwrap();
            for (row_id, version) in txn_versions.iter_local() {
                if !version.is_deleted() && !self.version_store.quick_check_row_existence(row_id) {
                    let val = version
                        .data
                        .get(col_idx)
                        .cloned()
                        .unwrap_or(Value::Null(DataType::Null));
                    entries.push((row_id, vec![val]));
                }
            }
        }
        if !entries.is_empty() {
            let entry_refs: Vec<(i64, &[Value])> = entries
                .iter()
                .map(|(row_id, values)| (*row_id, values.as_slice()))
                .collect();
            index.add_batch_slice(&entry_refs)?;
        }

        self.version_store.add_index(name.to_string(), index);
        Ok(())
    }

    fn drop_index(&self, name: &str) -> Result<()> {
        // Check if index exists
        if !self.version_store.index_exists(name) {
            return Err(Error::IndexNotFound(name.to_string()));
        }

        // Remove from version store
        self.version_store.remove_index(name);
        Ok(())
    }

    fn has_index_on_column(&self, column_name: &str) -> bool {
        self.version_store
            .get_index_by_column(column_name)
            .is_some()
    }

    fn get_index_on_column(&self, column_name: &str) -> Option<std::sync::Arc<dyn Index>> {
        self.version_store.get_index_by_column(column_name)
    }

    fn get_index(&self, name: &str) -> Option<std::sync::Arc<dyn Index>> {
        self.version_store.get_index(name)
    }

    fn get_unique_indexes(&self) -> Vec<(String, Vec<String>)> {
        self.version_store
            .get_all_indexes()
            .into_iter()
            .filter(|idx| idx.is_unique())
            .map(|idx| (idx.name().to_string(), idx.column_names().to_vec()))
            .collect()
    }

    fn for_each_unique_non_pk_index(
        &self,
        f: &mut dyn FnMut(&str, &[String]) -> Result<()>,
    ) -> Result<()> {
        let pk_col = self
            .cached_schema
            .pk_column_index()
            .map(|i| &self.cached_schema.columns[i].name_lower);
        let indexes = self.version_store.indexes_read();
        for idx in indexes.values() {
            if !idx.is_unique() {
                continue;
            }
            let names = idx.column_names();
            // Skip single-column indexes that match the PK column
            if names.len() == 1 {
                if let Some(pk) = pk_col {
                    if names[0].eq_ignore_ascii_case(pk) {
                        continue;
                    }
                }
            }
            f(idx.name(), names)?;
        }
        Ok(())
    }

    fn has_unique_non_pk_indexes(&self) -> bool {
        let pk_col = self
            .cached_schema
            .pk_column_index()
            .map(|i| &self.cached_schema.columns[i].name_lower);
        let indexes = self.version_store.indexes_read();
        indexes.values().any(|idx| {
            if !idx.is_unique() {
                return false;
            }
            let names = idx.column_names();
            if names.len() == 1 {
                if let Some(pk) = pk_col {
                    return !names[0].eq_ignore_ascii_case(pk);
                }
            }
            true
        })
    }

    fn acquire_upsert_lock(&self) -> Option<Box<dyn std::any::Any>> {
        Some(Box::new(self.version_store.acquire_upsert_lock()))
    }

    fn get_multi_column_index(
        &self,
        predicate_columns: &[&str],
    ) -> Option<(std::sync::Arc<dyn Index>, usize)> {
        self.version_store.get_multi_column_index(predicate_columns)
    }

    fn create_btree_index(
        &self,
        column_name: &str,
        is_unique: bool,
        custom_name: Option<&str>,
    ) -> Result<()> {
        // Find the column
        let schema = self.version_store.schema();
        let (col_idx, col) = schema
            .find_column(column_name)
            .ok_or_else(|| Error::ColumnNotFound(column_name.to_string()))?;

        // Generate index name
        let index_name = custom_name
            .map(|s| s.to_string())
            .unwrap_or_else(|| format!("idx_{}_{}_btree", self.name(), column_name));

        // Check if index with same name already exists
        if self.version_store.index_exists(&index_name) {
            return Err(Error::internal(format!(
                "index already exists: {}",
                index_name
            )));
        }

        // Check if an index already exists on this column
        if let Some(existing_index) = self.version_store.get_index_by_column(column_name) {
            let existing_is_unique = existing_index.is_unique();
            if existing_is_unique && !is_unique {
                return Err(Error::internal(format!(
                    "cannot create non-unique index on column '{}': a unique index already exists",
                    column_name
                )));
            } else if !existing_is_unique && is_unique {
                return Err(Error::internal(format!(
                    "cannot create unique index on column '{}': a non-unique index already exists",
                    column_name
                )));
            } else {
                return Err(Error::internal(format!(
                    "an index already exists on column '{}'",
                    column_name
                )));
            }
        }

        // Create the btree index with capacity hint
        let expected_rows = self.version_store.row_count();
        let index = BTreeIndex::new(
            index_name.clone(),
            self.name().to_string(),
            col.id as i32,
            column_name.to_string(),
            col.data_type,
            is_unique,
            expected_rows,
        );

        // Populate the index with existing data using batch_slice
        let mut entries: Vec<(i64, Vec<Value>)> = Vec::new();

        // Scan all visible rows
        for row_id in self.version_store.get_all_visible_row_ids(self.txn_id) {
            if let Some(version) = self.version_store.get_visible_version(row_id, self.txn_id) {
                if !version.is_deleted() {
                    if let Some(value) = version.data.get(col_idx) {
                        entries.push((row_id, vec![value.clone()]));
                    }
                }
            }
        }

        // Also collect any local uncommitted data
        {
            let txn_versions = self.txn_versions.read().unwrap();
            for (row_id, version) in txn_versions.iter_local() {
                if !version.is_deleted() {
                    if let Some(value) = version.data.get(col_idx) {
                        // Skip if already added from global store
                        if !self.version_store.quick_check_row_existence(row_id) {
                            entries.push((row_id, vec![value.clone()]));
                        }
                    }
                }
            }
        }

        // Add all entries in a single batch operation
        if !entries.is_empty() {
            let entry_refs: Vec<(i64, &[Value])> = entries
                .iter()
                .map(|(row_id, values)| (*row_id, values.as_slice()))
                .collect();
            index.add_batch_slice(&entry_refs)?;
        }

        // Add to version store
        self.version_store.add_index(index_name, Arc::new(index));

        Ok(())
    }

    /// Create a multi-column index using MultiColumnIndex
    fn create_multi_column_index(
        &self,
        name: &str,
        columns: &[&str],
        is_unique: bool,
    ) -> Result<()> {
        let schema = self.version_store.schema();

        // Validate and collect column info
        let mut column_names = Vec::with_capacity(columns.len());
        let mut column_ids = Vec::with_capacity(columns.len());
        let mut data_types = Vec::with_capacity(columns.len());
        let mut col_indices = Vec::with_capacity(columns.len());

        for col_name in columns {
            let (col_idx, col) = schema
                .find_column(col_name)
                .ok_or(Error::ColumnNotFound(col_name.to_string()))?;
            column_names.push(col.name.clone());
            column_ids.push(col.id as i32);
            data_types.push(col.data_type);
            col_indices.push(col_idx);
        }

        // Check if index with same name already exists
        if self.version_store.index_exists(name) {
            return Err(Error::IndexAlreadyExists(name.to_string()));
        }

        // Create the multi-column index with capacity hint
        let expected_rows = self.version_store.row_count();
        let index = MultiColumnIndex::new(
            name.to_string(),
            self.name().to_string(),
            column_names,
            column_ids,
            data_types,
            is_unique,
            expected_rows,
        );

        // Populate the index with existing data using batch_slice
        let mut entries: Vec<(i64, Vec<Value>)> = Vec::new();

        for row_id in self.version_store.get_all_visible_row_ids(self.txn_id) {
            if let Some(version) = self.version_store.get_visible_version(row_id, self.txn_id) {
                if !version.is_deleted() {
                    // Collect values for all columns in the index
                    let values: Vec<Value> = col_indices
                        .iter()
                        .map(|&idx| {
                            version
                                .data
                                .get(idx)
                                .cloned()
                                .unwrap_or(Value::Null(DataType::Null))
                        })
                        .collect();
                    entries.push((row_id, values));
                }
            }
        }

        // Also collect any local uncommitted data
        {
            let txn_versions = self.txn_versions.read().unwrap();
            for (row_id, version) in txn_versions.iter_local() {
                if !version.is_deleted() {
                    // Skip if already added from global store
                    if !self.version_store.quick_check_row_existence(row_id) {
                        let values: Vec<Value> = col_indices
                            .iter()
                            .map(|&idx| {
                                version
                                    .data
                                    .get(idx)
                                    .cloned()
                                    .unwrap_or(Value::Null(DataType::Null))
                            })
                            .collect();
                        entries.push((row_id, values));
                    }
                }
            }
        }

        // Add all entries in a single batch operation
        if !entries.is_empty() {
            let entry_refs: Vec<(i64, &[Value])> = entries
                .iter()
                .map(|(row_id, values)| (*row_id, values.as_slice()))
                .collect();
            index.add_batch_slice(&entry_refs)?;
        }

        // Add to version store
        self.version_store
            .add_index(name.to_string(), Arc::new(index));

        Ok(())
    }

    fn drop_btree_index(&self, column_name: &str) -> Result<()> {
        // Generate default index name
        let index_name = format!("idx_{}_{}_btree", self.name(), column_name);

        // Check if index exists
        if !self.version_store.index_exists(&index_name) {
            return Err(Error::internal(format!(
                "btree index not found for column: {}",
                column_name
            )));
        }

        // Remove from version store
        self.version_store.remove_index(&index_name);
        Ok(())
    }

    fn get_index_min_value(&self, column_name: &str) -> Option<Value> {
        // Try to find an index on this column and get its minimum value
        if let Some(index) = self.version_store.get_index_by_column(column_name) {
            return index.get_min_value();
        }
        None
    }

    fn get_index_max_value(&self, column_name: &str) -> Option<Value> {
        // Try to find an index on this column and get its maximum value
        if let Some(index) = self.version_store.get_index_by_column(column_name) {
            return index.get_max_value();
        }
        None
    }

    fn row_count(&self) -> usize {
        // Try O(1) fast path first
        if let Some(count) = MVCCTable::fast_row_count(self) {
            return count;
        }
        // Fall back to optimized single-pass counting
        MVCCTable::row_count(self)
    }

    fn row_count_hint(&self) -> usize {
        // O(1) - just return the committed row count without any lock checks
        self.version_store.committed_row_count()
    }

    fn fast_row_count(&self) -> Option<usize> {
        MVCCTable::fast_row_count(self)
    }

    fn collect_rows_ordered_by_index(
        &self,
        column_name: &str,
        ascending: bool,
        limit: usize,
        offset: usize,
    ) -> Option<RowVec> {
        // If transaction has local changes (inserts/updates/deletes), fall back to
        // the regular path which correctly merges local changes via collect_visible_rows.
        // This optimization only reads from the global version store and indexes.
        {
            let txn_versions = self.txn_versions.read().unwrap();
            if txn_versions.has_local_changes() {
                return None;
            }
        }

        // OPTIMIZATION: Handle PRIMARY KEY column specially
        // For INTEGER PRIMARY KEY, the row_id IS the value, so we can iterate
        // directly in order without any sorting using skip/take semantics.
        if let Some(pk_idx) = self.cached_schema.pk_column_index() {
            let pk_col = &self.cached_schema.columns[pk_idx];
            if pk_col.name_lower == column_name.to_lowercase() {
                return self.version_store.collect_rows_pk_ordered(
                    self.txn_id,
                    ascending,
                    limit,
                    offset,
                );
            }
        }

        // Check if column has an index
        let index = self.version_store.get_index_by_column(column_name)?;

        // Try using the efficient ordered iteration method (available in B-tree indexes)
        // We request more row IDs than needed to account for invisible rows
        let batch_size = (limit + offset) * 2 + 100; // Request extra to handle filtered rows

        if let Some(ordered_row_ids) = index.get_row_ids_ordered(ascending, batch_size, 0) {
            // Fast path: B-tree index supports ordered iteration
            let mut rows = RowVec::with_capacity(limit.min(100));
            let mut skipped = 0;

            for row_id in ordered_row_ids {
                // Check visibility and get row
                if let Some(version) = self.version_store.get_visible_version(row_id, self.txn_id) {
                    if version.is_deleted() {
                        continue;
                    }

                    // Handle offset
                    if skipped < offset {
                        skipped += 1;
                        continue;
                    }

                    rows.push((row_id, version.data.clone()));

                    // Check if we've reached the limit
                    if rows.len() >= limit {
                        return Some(rows);
                    }
                }
            }

            // If we got all needed rows, return them
            // If not, we may need to fetch more (rare case with many invisible rows)
            if !rows.is_empty() {
                return Some(rows);
            }
        }

        // Fallback path: Get all values and sort (for non-B-tree indexes)
        let all_values = index.get_all_values();

        // If the index doesn't support get_all_values (returns empty), return None
        // to let the regular query execution path handle ORDER BY + LIMIT
        if all_values.is_empty() {
            return None;
        }

        // Sort values using partial_cmp (Value implements PartialOrd)
        let mut sorted_values = all_values;
        sorted_values.sort_by(|a, b| {
            if ascending {
                a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)
            } else {
                b.partial_cmp(a).unwrap_or(std::cmp::Ordering::Equal)
            }
        });

        // Collect rows by iterating through sorted values
        let mut rows = RowVec::with_capacity(limit.min(100));
        let mut skipped = 0;
        let mut row_ids = Vec::new();

        for value in sorted_values {
            // Get all row IDs for this value - reuse buffer to avoid allocation per iteration
            row_ids.clear();
            index.get_row_ids_equal_into(&[value], &mut row_ids);

            for row_id in &row_ids {
                let row_id = *row_id;
                // Check visibility and get row
                if let Some(version) = self.version_store.get_visible_version(row_id, self.txn_id) {
                    if version.is_deleted() {
                        continue;
                    }

                    // Handle offset
                    if skipped < offset {
                        skipped += 1;
                        continue;
                    }

                    rows.push((row_id, version.data.clone()));

                    // Check if we've reached the limit
                    if rows.len() >= limit {
                        return Some(rows);
                    }
                }
            }
        }

        Some(rows)
    }

    fn collect_rows_pk_keyset(
        &self,
        start_after: Option<i64>,
        start_from: Option<i64>,
        ascending: bool,
        limit: usize,
    ) -> Option<RowVec> {
        // Only works if table has a single-column INTEGER PRIMARY KEY
        self.cached_schema.pk_column_index()?;

        // If transaction has local changes, fall back to regular path
        {
            let txn_versions = self.txn_versions.read().unwrap();
            if txn_versions.has_local_changes() {
                return None;
            }
        }

        // Use the efficient keyset iteration from version store
        // Returns RowVec with (row_id, Row) tuples
        Some(self.version_store.collect_rows_keyset(
            self.txn_id,
            start_after,
            start_from,
            ascending,
            limit,
        ))
    }

    fn collect_rows_grouped_by_partition(&self, column_name: &str) -> Option<Vec<(Value, RowVec)>> {
        // If transaction has local changes, fall back to regular path
        {
            let txn_versions = self.txn_versions.read().unwrap();
            if txn_versions.has_local_changes() {
                return None;
            }
        }

        // Check if column has an index
        let index = self.version_store.get_index_by_column(column_name)?;

        // Get all unique values from the index (partition keys)
        let all_values = index.get_all_values();
        if all_values.is_empty() {
            return Some(Vec::new());
        }

        // Collect rows grouped by partition value
        let mut result: Vec<(Value, RowVec)> = Vec::with_capacity(all_values.len());
        let mut row_ids = Vec::new();

        for partition_value in all_values {
            // Get all row IDs for this partition value - reuse buffer to avoid allocation per iteration
            row_ids.clear();
            index.get_row_ids_equal_into(std::slice::from_ref(&partition_value), &mut row_ids);

            // Collect visible rows for this partition
            let mut partition_rows = RowVec::with_capacity(row_ids.len());
            for &row_id in &row_ids {
                if let Some(version) = self.version_store.get_visible_version(row_id, self.txn_id) {
                    if !version.is_deleted() {
                        partition_rows.push((row_id, version.data.clone()));
                    }
                }
            }

            if !partition_rows.is_empty() {
                result.push((partition_value, partition_rows));
            }
        }

        Some(result)
    }

    fn get_partition_values(&self, column_name: &str) -> Option<Vec<Value>> {
        // Only use index-based distinct values if no uncommitted local changes
        // (local changes are in txn_versions, not reflected in the index)
        let txn_versions = self.txn_versions.read().unwrap();
        if txn_versions.has_local_changes() {
            return None;
        }
        drop(txn_versions);

        // Get index for the column
        let index = self.version_store.get_index_by_column(column_name)?;
        // Return all distinct values from the index
        Some(index.get_all_values())
    }

    fn get_partition_count(&self, column_name: &str) -> Option<usize> {
        // Only use index-based count if no uncommitted local changes
        let txn_versions = self.txn_versions.read().unwrap();
        if txn_versions.has_local_changes() {
            return None;
        }
        drop(txn_versions);

        // Get index for the column
        let index = self.version_store.get_index_by_column(column_name)?;
        // Return count of distinct non-null values without cloning
        index.get_distinct_count_excluding_null()
    }

    fn get_rows_for_partition_value(
        &self,
        column_name: &str,
        partition_value: &Value,
    ) -> Option<RowVec> {
        // If transaction has local changes, fall back to regular path
        {
            let txn_versions = self.txn_versions.read().unwrap();
            if txn_versions.has_local_changes() {
                return None;
            }
        }

        // Get index for the column
        let index = self.version_store.get_index_by_column(column_name)?;

        // Get row IDs for this partition value
        let row_ids = index.get_row_ids_equal(std::slice::from_ref(partition_value));

        // Collect visible rows for this partition
        let mut rows = RowVec::with_capacity(row_ids.len());
        for &row_id in row_ids.iter() {
            if let Some(version) = self.version_store.get_visible_version(row_id, self.txn_id) {
                if !version.is_deleted() {
                    rows.push((row_id, version.data.clone()));
                }
            }
        }

        Some(rows)
    }

    fn rename_column(&mut self, old_name: &str, new_name: &str) -> Result<()> {
        // Rename column in both version store and cached schema
        {
            let mut schema_guard = self.version_store.schema_mut();
            CompactArc::make_mut(&mut *schema_guard).rename_column(old_name, new_name)?;
        }
        CompactArc::make_mut(&mut self.cached_schema).rename_column(old_name, new_name)?;
        Ok(())
    }

    fn modify_column(&mut self, name: &str, column_type: DataType, nullable: bool) -> Result<()> {
        // Modify column in both version store and cached schema
        {
            let mut schema_guard = self.version_store.schema_mut();
            CompactArc::make_mut(&mut *schema_guard).modify_column(
                name,
                Some(column_type),
                Some(nullable),
            )?;
        }
        CompactArc::make_mut(&mut self.cached_schema).modify_column(
            name,
            Some(column_type),
            Some(nullable),
        )?;
        Ok(())
    }

    fn select(
        &self,
        columns: &[&str],
        expr: Option<&dyn Expression>,
    ) -> Result<Box<dyn QueryResult>> {
        // Convert column names to indices
        let column_indices: Vec<usize> = columns
            .iter()
            .filter_map(|name| self.cached_schema.find_column(name).map(|(idx, _)| idx))
            .collect();

        // Scan and collect results
        let mut scanner = self.scan(&column_indices, expr)?;
        let mut rows = Vec::new();

        while scanner.next() {
            rows.push(scanner.take_row());
        }

        scanner.close()?;

        // Create result
        let result_columns: Vec<String> = columns.iter().map(|s| s.to_string()).collect();
        let result = MemoryResult::with_rows(result_columns, rows);

        Ok(Box::new(result))
    }

    fn select_with_aliases(
        &self,
        columns: &[&str],
        expr: Option<&dyn Expression>,
        aliases: &FxHashMap<String, String>,
    ) -> Result<Box<dyn QueryResult>> {
        // Get base result
        let result = self.select(columns, expr)?;

        // Apply aliases
        Ok(result.with_aliases(aliases.clone()))
    }

    fn select_as_of(
        &self,
        columns: &[&str],
        expr: Option<&dyn Expression>,
        temporal_type: &str,
        temporal_value: i64,
    ) -> Result<Box<dyn QueryResult>> {
        // Convert column names to indices
        let column_indices: Vec<usize> = columns
            .iter()
            .filter_map(|name| self.cached_schema.find_column(name).map(|(idx, _)| idx))
            .collect();

        // Get all row IDs
        let row_ids = self.version_store.get_all_row_ids();

        // Collect temporal rows
        let mut rows = Vec::new();
        for row_id in row_ids {
            let version = match temporal_type.to_uppercase().as_str() {
                "TRANSACTION" => self
                    .version_store
                    .get_visible_version_as_of_transaction(row_id, temporal_value),
                "TIMESTAMP" => self
                    .version_store
                    .get_visible_version_as_of_timestamp(row_id, temporal_value),
                _ => {
                    return Err(Error::internal(format!(
                        "unsupported temporal type: {}",
                        temporal_type
                    )))
                }
            };

            if let Some(v) = version {
                if !v.is_deleted() {
                    // Apply filter
                    if let Some(e) = expr {
                        match e.evaluate(&v.data) {
                            Ok(true) => {}
                            Ok(false) => continue,
                            Err(_) => continue,
                        }
                    }

                    // Project columns
                    let projected: Vec<Value> = column_indices
                        .iter()
                        .map(|&idx| v.data.get(idx).cloned().unwrap_or(Value::null_unknown()))
                        .collect();
                    rows.push(Row::from_values(projected));
                }
            }
        }

        // Create result
        let result_columns: Vec<String> = columns.iter().map(|s| s.to_string()).collect();
        let result = MemoryResult::with_rows(result_columns, rows);

        Ok(Box::new(result))
    }

    fn explain_scan(&self, where_expr: Option<&dyn Expression>) -> ScanPlan {
        use crate::core::Operator;

        let table_name = self.cached_schema.table_name.clone();
        let schema = &self.cached_schema;

        // No WHERE clause - always Seq Scan
        let Some(expr) = where_expr else {
            return ScanPlan::SeqScan {
                table: table_name,
                filter: None,
            };
        };

        // Check for PK lookup
        let pk_indices = schema.primary_key_indices();
        if pk_indices.len() == 1 {
            let pk_col_idx = pk_indices[0];
            let pk_col = &schema.columns[pk_col_idx];

            if let Some((col_name, operator, value)) = expr.get_comparison_info() {
                if col_name.eq_ignore_ascii_case(&pk_col.name) && operator == Operator::Eq {
                    return ScanPlan::PkLookup {
                        table: table_name,
                        pk_column: pk_col.name.clone(),
                        pk_value: format!("{}", value),
                    };
                }
            }
        }

        // Check for single column index lookup
        if let Some((col_name, operator, value)) = expr.get_comparison_info() {
            // Skip boolean index (low cardinality)
            if !matches!(value, Value::Boolean(_))
                || !matches!(operator, Operator::Eq | Operator::Ne)
            {
                if let Some(index) = self.version_store.get_index_by_column(col_name) {
                    let condition = format!("{} {}", operator_to_string(operator), value);
                    return ScanPlan::IndexScan {
                        table: table_name,
                        index_name: index.name().to_string(),
                        column: col_name.to_string(),
                        condition,
                        filter: None,
                    };
                }
            }
        }

        // Check for LIKE prefix pattern
        if let Some((col_name, prefix, negated)) = expr.get_like_prefix_info() {
            if !negated {
                if let Some(index) = self.version_store.get_index_by_column(col_name) {
                    return ScanPlan::IndexScan {
                        table: table_name,
                        index_name: index.name().to_string(),
                        column: col_name.to_string(),
                        condition: format!("LIKE '{}%'", prefix),
                        filter: None,
                    };
                }
            }
        }

        // Check for OR expressions (union of indexes)
        if let Some(or_operands) = expr.get_or_operands() {
            let mut indexed_info: Vec<(String, String, String)> = Vec::new();
            let mut all_indexed = true;

            for operand in or_operands {
                if let Some((col_name, operator, value)) = operand.get_comparison_info() {
                    if let Some(index) = self.version_store.get_index_by_column(col_name) {
                        let condition = format!("{} {}", operator_to_string(operator), value);
                        indexed_info.push((
                            index.name().to_string(),
                            col_name.to_string(),
                            condition,
                        ));
                    } else {
                        all_indexed = false;
                    }
                } else {
                    all_indexed = false;
                }
            }

            if !indexed_info.is_empty() {
                if all_indexed {
                    if indexed_info.len() == 1 {
                        let (idx_name, col, cond) = indexed_info.into_iter().next().unwrap();
                        return ScanPlan::IndexScan {
                            table: table_name,
                            index_name: idx_name,
                            column: col,
                            condition: cond,
                            filter: None,
                        };
                    }
                    return ScanPlan::MultiIndexScan {
                        table: table_name,
                        indexes: indexed_info,
                        operation: "OR".to_string(),
                        filter: None,
                    };
                }
                // Mixed OR: some operands indexed, some not — hybrid scan
                // Show as Multi-Index Scan with a Filter for the non-indexed operands
                let non_indexed_parts: Vec<String> = or_operands
                    .iter()
                    .filter(|op| {
                        if let Some((col_name, _, _)) = op.get_comparison_info() {
                            self.version_store.get_index_by_column(col_name).is_none()
                        } else {
                            true
                        }
                    })
                    .map(|op| {
                        if let Some((col, operator, val)) = op.get_comparison_info() {
                            format!("{} {} {}", col, operator_to_string(operator), val)
                        } else {
                            format!("{:?}", op)
                        }
                    })
                    .collect();
                let filter_str = if non_indexed_parts.len() == 1 {
                    non_indexed_parts.into_iter().next().unwrap()
                } else {
                    non_indexed_parts.join(" OR ")
                };
                return ScanPlan::MultiIndexScan {
                    table: table_name,
                    indexes: indexed_info,
                    operation: "OR".to_string(),
                    filter: Some(filter_str),
                };
            }
        }

        // Check for multi-column (composite) index before single-column index intersection
        let comparisons = expr.collect_comparisons();
        if !comparisons.is_empty() {
            // Group by column - needed for both multi-column and single-column index checks
            let mut column_conditions: FxHashMap<&str, Vec<(Operator, &Value)>> =
                FxHashMap::default();
            for (col_name, op, val) in &comparisons {
                column_conditions
                    .entry(*col_name)
                    .or_default()
                    .push((*op, *val));
            }

            // Collect columns with equality predicates (best for multi-column index)
            let eq_columns: Vec<&str> = column_conditions
                .iter()
                .filter_map(|(col, ops)| {
                    if ops.iter().any(|(op, _)| *op == Operator::Eq) {
                        Some(*col)
                    } else {
                        None
                    }
                })
                .collect();

            // Try multi-column index if we have equality predicates matching a prefix
            if !eq_columns.is_empty() {
                if let Some((multi_idx, matched_count)) =
                    self.version_store.get_multi_column_index(&eq_columns)
                {
                    let index_columns = multi_idx.column_names();
                    let mut columns = Vec::new();
                    let mut conditions = Vec::new();

                    // Build columns and conditions in index column order
                    for col in index_columns.iter().take(matched_count) {
                        if let Some(ops) = column_conditions.get(col.as_str()) {
                            columns.push(col.clone());
                            // Find the equality condition
                            for (op, val) in ops {
                                if *op == Operator::Eq {
                                    conditions.push(format!("= {}", val));
                                    break;
                                }
                            }
                        }
                    }

                    // Check if the next index column (after the equality prefix) has
                    // range predicates — the composite index BTree can serve these too
                    if matched_count < index_columns.len() {
                        let next_col = &index_columns[matched_count];
                        if let Some(ops) = column_conditions.get(next_col.as_str()) {
                            let range_ops: Vec<String> = ops
                                .iter()
                                .filter(|(op, _)| !matches!(op, Operator::Eq))
                                .map(|(op, val)| format!("{} {}", operator_to_string(*op), val))
                                .collect();
                            if !range_ops.is_empty() {
                                columns.push(next_col.clone());
                                conditions.push(range_ops.join(" AND "));
                            }
                        }
                    }

                    if !columns.is_empty() {
                        // Collect residual predicates: columns not covered by the index
                        let covered: rustc_hash::FxHashSet<&str> =
                            columns.iter().map(|c| c.as_str()).collect();
                        let residual: Vec<String> = column_conditions
                            .iter()
                            .filter(|(col, _)| !covered.contains(*col))
                            .map(|(col, ops)| {
                                ops.iter()
                                    .map(|(op, val)| {
                                        format!("{} {} {}", col, operator_to_string(*op), val)
                                    })
                                    .collect::<Vec<_>>()
                                    .join(" AND ")
                            })
                            .collect();
                        let filter = if residual.is_empty() {
                            None
                        } else {
                            Some(residual.join(" AND "))
                        };

                        return ScanPlan::CompositeIndexScan {
                            table: table_name,
                            index_name: multi_idx.name().to_string(),
                            columns,
                            conditions,
                            filter,
                        };
                    }
                }
            }
        }

        // Check for AND expressions (intersection of indexes)
        if !comparisons.is_empty() {
            let mut indexed_info: Vec<(String, String, String)> = Vec::new();

            // Re-group by column for single-column index checks
            let mut column_conditions: FxHashMap<&str, Vec<(Operator, &Value)>> =
                FxHashMap::default();
            for (col_name, op, val) in &comparisons {
                column_conditions
                    .entry(*col_name)
                    .or_default()
                    .push((*op, *val));
            }

            let mut indexed_columns: rustc_hash::FxHashSet<&str> = rustc_hash::FxHashSet::default();
            for (col_name, ops) in &column_conditions {
                if let Some(index) = self.version_store.get_index_by_column(col_name) {
                    // Simplify to a single condition string
                    let condition = if ops.len() == 1 {
                        let (op, val) = ops[0];
                        format!("{} {}", operator_to_string(op), val)
                    } else {
                        // Multiple conditions on same column (e.g., col >= 5 AND col <= 10)
                        let parts: Vec<String> = ops
                            .iter()
                            .map(|(op, val)| format!("{} {}", operator_to_string(*op), val))
                            .collect();
                        parts.join(" AND ")
                    };
                    indexed_info.push((index.name().to_string(), col_name.to_string(), condition));
                    indexed_columns.insert(col_name);
                }
            }

            if !indexed_info.is_empty() {
                // Collect residual predicates for non-indexed columns
                let residual: Vec<String> = column_conditions
                    .iter()
                    .filter(|(col, _)| !indexed_columns.contains(*col))
                    .map(|(col, ops)| {
                        ops.iter()
                            .map(|(op, val)| format!("{} {} {}", col, operator_to_string(*op), val))
                            .collect::<Vec<_>>()
                            .join(" AND ")
                    })
                    .collect();
                let filter = if residual.is_empty() {
                    None
                } else {
                    Some(residual.join(" AND "))
                };

                if indexed_info.len() == 1 {
                    let (idx_name, col, cond) = indexed_info.into_iter().next().unwrap();
                    return ScanPlan::IndexScan {
                        table: table_name,
                        index_name: idx_name,
                        column: col,
                        condition: cond,
                        filter,
                    };
                }
                return ScanPlan::MultiIndexScan {
                    table: table_name,
                    indexes: indexed_info,
                    operation: "AND".to_string(),
                    filter,
                };
            }
        }

        // Default: Seq Scan with filter
        ScanPlan::SeqScan {
            table: table_name,
            filter: Some(format!("{:?}", expr)),
        }
    }

    fn set_zone_maps(&self, zone_maps: crate::storage::mvcc::zonemap::TableZoneMap) {
        self.version_store.set_zone_maps(zone_maps);
    }

    fn get_zone_maps(&self) -> Option<std::sync::Arc<crate::storage::mvcc::zonemap::TableZoneMap>> {
        self.version_store.get_zone_maps()
    }

    fn get_segments_to_scan(
        &self,
        column: &str,
        operator: crate::core::Operator,
        value: &Value,
    ) -> Option<Vec<u32>> {
        self.version_store
            .get_segments_to_scan(column, operator, value)
    }

    fn sum_column(&self, col_idx: usize) -> Option<(f64, usize)> {
        // Only use deferred aggregation if no uncommitted local changes
        // (local changes are in txn_versions, not the main version_store)
        let txn_versions = self.txn_versions.read().unwrap();
        if txn_versions.has_local_changes() {
            return None;
        }
        drop(txn_versions);

        Some(self.version_store.sum_column(self.txn_id, col_idx))
    }

    fn min_column(&self, col_idx: usize) -> Option<Option<Value>> {
        // Only use deferred aggregation if no uncommitted local changes
        let txn_versions = self.txn_versions.read().unwrap();
        if txn_versions.has_local_changes() {
            return None;
        }
        drop(txn_versions);

        Some(self.version_store.min_column(self.txn_id, col_idx))
    }

    fn max_column(&self, col_idx: usize) -> Option<Option<Value>> {
        // Only use deferred aggregation if no uncommitted local changes
        let txn_versions = self.txn_versions.read().unwrap();
        if txn_versions.has_local_changes() {
            return None;
        }
        drop(txn_versions);

        Some(self.version_store.max_column(self.txn_id, col_idx))
    }

    fn compute_grouped_aggregates(
        &self,
        group_by_indices: &[usize],
        aggregates: &[(crate::storage::mvcc::version_store::AggregateOp, usize)],
    ) -> Option<Vec<crate::storage::mvcc::version_store::GroupedAggregateResult>> {
        // Only use storage-level aggregation if no uncommitted local changes
        let txn_versions = self.txn_versions.read().unwrap();
        if txn_versions.has_local_changes() {
            return None;
        }
        drop(txn_versions);

        self.version_store
            .compute_grouped_aggregates(self.txn_id, group_by_indices, aggregates)
    }
}

/// Helper function to convert Operator to string for display
fn operator_to_string(op: crate::core::Operator) -> &'static str {
    use crate::core::Operator;
    match op {
        Operator::Eq => "=",
        Operator::Ne => "!=",
        Operator::Lt => "<",
        Operator::Lte => "<=",
        Operator::Gt => ">",
        Operator::Gte => ">=",
        Operator::Like => "LIKE",
        Operator::In => "IN",
        Operator::NotIn => "NOT IN",
        Operator::IsNull => "IS NULL",
        Operator::IsNotNull => "IS NOT NULL",
    }
}

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

    fn test_schema() -> Schema {
        SchemaBuilder::new("test_table")
            .column("id", DataType::Integer, false, true)
            .column("name", DataType::Text, true, false)
            .build()
    }

    fn simple_schema() -> Schema {
        SchemaBuilder::new("test_table")
            .column("id", DataType::Integer, false, true)
            .build()
    }

    #[test]
    fn test_mvcc_table_creation() {
        let schema = test_schema();

        let version_store = Arc::new(VersionStore::new("test_table".to_string(), schema));
        let txn_id = 1;

        let txn_versions = TransactionVersionStore::new(Arc::clone(&version_store), txn_id);

        let table = MVCCTable::new(txn_id, version_store, txn_versions);

        assert_eq!(table.name(), "test_table");
        assert_eq!(table.schema().columns.len(), 2);
    }

    #[test]
    fn test_mvcc_table_insert_and_scan() {
        let schema = test_schema();

        let version_store = Arc::new(VersionStore::new("test_table".to_string(), schema));
        let txn_id = 1;

        let txn_versions = TransactionVersionStore::new(Arc::clone(&version_store), txn_id);

        let mut table = MVCCTable::new(txn_id, version_store, txn_versions);

        // Insert a row
        let row = Row::from_values(vec![Value::Integer(1), Value::text("Alice")]);
        table.insert(row).unwrap();

        // Scan to verify
        let mut scanner = table.scan(&[0, 1], None).unwrap();

        assert!(scanner.next());
        let row = scanner.row();
        assert_eq!(row.get(0), Some(&Value::Integer(1)));
        assert_eq!(row.get(1), Some(&Value::text("Alice")));

        assert!(!scanner.next());
        scanner.close().unwrap();
    }

    #[test]
    fn test_mvcc_table_duplicate_key_error() {
        let schema = simple_schema();

        let version_store = Arc::new(VersionStore::new("test_table".to_string(), schema));
        let txn_id = 1;

        let txn_versions = TransactionVersionStore::new(Arc::clone(&version_store), txn_id);

        let mut table = MVCCTable::new(txn_id, version_store, txn_versions);

        // Insert first row
        let row1 = Row::from_values(vec![Value::Integer(1)]);
        table.insert(row1).unwrap();

        // Try to insert duplicate
        let row2 = Row::from_values(vec![Value::Integer(1)]);
        let result = table.insert(row2);

        assert!(result.is_err());
    }

    #[test]
    fn test_mvcc_table_delete() {
        let schema = simple_schema();

        let version_store = Arc::new(VersionStore::new("test_table".to_string(), schema));
        let txn_id = 1;

        let txn_versions = TransactionVersionStore::new(Arc::clone(&version_store), txn_id);

        let mut table = MVCCTable::new(txn_id, version_store, txn_versions);

        // Insert rows
        table
            .insert(Row::from_values(vec![Value::Integer(1)]))
            .unwrap();
        table
            .insert(Row::from_values(vec![Value::Integer(2)]))
            .unwrap();
        table
            .insert(Row::from_values(vec![Value::Integer(3)]))
            .unwrap();

        // Delete all rows (no filter)
        let deleted = table.delete(None).unwrap();
        assert_eq!(deleted, 3);

        // Verify no rows remain
        let mut scanner = table.scan(&[0], None).unwrap();
        assert!(!scanner.next());
        scanner.close().unwrap();
    }

    #[test]
    fn test_mvcc_table_update() {
        let schema = SchemaBuilder::new("test_table")
            .column("id", DataType::Integer, false, true)
            .column("value", DataType::Integer, true, false)
            .build();

        let version_store = Arc::new(VersionStore::new("test_table".to_string(), schema));
        let txn_id = 1;

        let txn_versions = TransactionVersionStore::new(Arc::clone(&version_store), txn_id);

        let mut table = MVCCTable::new(txn_id, version_store, txn_versions);

        // Insert row
        table
            .insert(Row::from_values(vec![
                Value::Integer(1),
                Value::Integer(10),
            ]))
            .unwrap();

        // Update the row
        let updated = table
            .update(None, &mut |row| {
                let mut new_row = row.clone();
                let _ = new_row.set(1, Value::Integer(20));
                Ok((new_row, true))
            })
            .unwrap();

        assert_eq!(updated, 1);

        // Verify update
        let mut scanner = table.scan(&[0, 1], None).unwrap();
        assert!(scanner.next());
        let row = scanner.row();
        assert_eq!(row.get(1), Some(&Value::Integer(20)));
        scanner.close().unwrap();
    }

    #[test]
    fn test_mvcc_table_validation_error() {
        let schema = SchemaBuilder::new("test_table")
            .column("id", DataType::Integer, false, false)
            .column("name", DataType::Text, false, false) // NOT NULL
            .build();

        let version_store = Arc::new(VersionStore::new("test_table".to_string(), schema));
        let txn_id = 1;

        let txn_versions = TransactionVersionStore::new(Arc::clone(&version_store), txn_id);

        let mut table = MVCCTable::new(txn_id, version_store, txn_versions);

        // Try to insert with NULL in non-nullable column
        let row = Row::from_values(vec![Value::Integer(1), Value::Null(DataType::Text)]);
        let result = table.insert(row);

        assert!(result.is_err());
    }

    #[test]
    fn test_mvcc_table_row_count() {
        let schema = simple_schema();

        let version_store = Arc::new(VersionStore::new("test_table".to_string(), schema));
        let txn_id = 1;

        let txn_versions = TransactionVersionStore::new(Arc::clone(&version_store), txn_id);

        let mut table = MVCCTable::new(txn_id, version_store, txn_versions);

        assert_eq!(table.row_count(), 0);

        table
            .insert(Row::from_values(vec![Value::Integer(1)]))
            .unwrap();
        table
            .insert(Row::from_values(vec![Value::Integer(2)]))
            .unwrap();

        assert_eq!(table.row_count(), 2);
    }

    #[test]
    fn test_validate_coerce_integer_to_float() {
        let schema = SchemaBuilder::new("test_table")
            .column("id", DataType::Integer, false, true)
            .column("score", DataType::Float, true, false)
            .build();

        let version_store = Arc::new(VersionStore::new("test_table".to_string(), schema));
        let txn_id = 1;
        let txn_versions = TransactionVersionStore::new(Arc::clone(&version_store), txn_id);
        let mut table = MVCCTable::new(txn_id, version_store, txn_versions);

        // Insert Integer into Float column — should coerce
        let row = Row::from_values(vec![Value::Integer(1), Value::Integer(42)]);
        table.insert(row).unwrap();

        let mut scanner = table.scan(&[0, 1], None).unwrap();
        assert!(scanner.next());
        assert_eq!(scanner.row().get(1), Some(&Value::Float(42.0)));
        scanner.close().unwrap();
    }

    #[test]
    fn test_validate_coerce_float_to_integer() {
        let schema = SchemaBuilder::new("test_table")
            .column("id", DataType::Integer, false, true)
            .column("count", DataType::Integer, true, false)
            .build();

        let version_store = Arc::new(VersionStore::new("test_table".to_string(), schema));
        let txn_id = 1;
        let txn_versions = TransactionVersionStore::new(Arc::clone(&version_store), txn_id);
        let mut table = MVCCTable::new(txn_id, version_store, txn_versions);

        // Insert Float into Integer column — should truncate
        let row = Row::from_values(vec![Value::Integer(1), Value::Float(9.7)]);
        table.insert(row).unwrap();

        let mut scanner = table.scan(&[0, 1], None).unwrap();
        assert!(scanner.next());
        assert_eq!(scanner.row().get(1), Some(&Value::Integer(9)));
        scanner.close().unwrap();
    }

    #[test]
    fn test_validate_coerce_integer_to_boolean() {
        let schema = SchemaBuilder::new("test_table")
            .column("id", DataType::Integer, false, true)
            .column("active", DataType::Boolean, true, false)
            .build();

        let version_store = Arc::new(VersionStore::new("test_table".to_string(), schema));
        let txn_id = 1;
        let txn_versions = TransactionVersionStore::new(Arc::clone(&version_store), txn_id);
        let mut table = MVCCTable::new(txn_id, version_store, txn_versions);

        // 0 -> false
        let row = Row::from_values(vec![Value::Integer(1), Value::Integer(0)]);
        table.insert(row).unwrap();

        // non-zero -> true
        let row = Row::from_values(vec![Value::Integer(2), Value::Integer(5)]);
        table.insert(row).unwrap();

        let mut results: Vec<(i64, bool)> = Vec::new();
        let mut scanner = table.scan(&[0, 1], None).unwrap();
        while scanner.next() {
            let id = match scanner.row().get(0) {
                Some(Value::Integer(v)) => *v,
                _ => panic!("expected integer id"),
            };
            let active = match scanner.row().get(1) {
                Some(Value::Boolean(v)) => *v,
                _ => panic!("expected boolean active"),
            };
            results.push((id, active));
        }
        scanner.close().unwrap();

        results.sort_by_key(|(id, _)| *id);
        assert_eq!(results, vec![(1, false), (2, true)]);
    }

    #[test]
    fn test_validate_coerce_type_mismatch_error() {
        let schema = SchemaBuilder::new("test_table")
            .column("id", DataType::Integer, false, true)
            .column("name", DataType::Text, true, false)
            .build();

        let version_store = Arc::new(VersionStore::new("test_table".to_string(), schema));
        let txn_id = 1;
        let txn_versions = TransactionVersionStore::new(Arc::clone(&version_store), txn_id);
        let mut table = MVCCTable::new(txn_id, version_store, txn_versions);

        // Boolean into Text column — not a supported coercion, should error
        let row = Row::from_values(vec![Value::Integer(1), Value::Boolean(true)]);
        assert!(table.insert(row).is_err());
    }
}