sochdb 2.0.2

// SPDX-License-Identifier: AGPL-3.0-or-later
// SochDB - LLM-Optimized Embedded Database
// Copyright (C) 2026 Sushanth Reddy Vanagala (https://github.com/sushanthpy)
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.

//! SochDB Connection Handle
//!
//! Provides unified access to TCH path resolution, LSCS storage, and MVCC transactions.
//!
//! ## Connection Types
//!
//! | Type | Durability | Use Case |
//! |------|-----------|----------|
//! | [`DurableConnection`] | Full WAL + MVCC | **Production** - ACID guarantees |
//! | [`SochConnection`] | In-memory only | Testing - ephemeral data |
//!
//! **For production workloads, always use [`DurableConnection`]**:
//!
//! ```rust,ignore
//! use sochdb::DurableConnection;
//!
//! // Open a production-ready connection with WAL durability
//! let conn = DurableConnection::open("./data")?;
//!
//! // Transactions are durable (survive crashes)
//! let txn = conn.begin_txn()?;
//! conn.put(txn, b"key", b"value")?;
//! conn.commit_txn(txn)?;  // Written to WAL, fsync'd
//! ```
//!
//! ## Complexity
//!
//! - Path resolution: O(|path|) where |path| = character count
//! - Key insight: Complexity is independent of row count N (unlike B-tree O(log N))

use parking_lot::RwLock;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use sochdb_core::catalog::Catalog;
use sochdb_core::soch::{SochSchema, SochType, SochValue};

use crate::error::{ClientError, Result};
use crate::{ClientStats, schema::TableDescription};

/// Type alias for transaction ID
pub type TxnId = u64;

/// Type alias for timestamp
pub type Timestamp = u64;

/// Type alias for row ID
pub type RowId = u64;

/// Result from a mutation operation (UPDATE/DELETE)
/// 
/// Contains both the count and the affected row IDs for storage-level
/// durability operations (WAL entries, secondary index updates, CDC).
#[derive(Debug, Clone, Default)]
pub struct MutationResult {
    /// Number of rows affected
    pub affected_count: usize,
    /// Row IDs of affected rows (for WAL and index updates)
    pub affected_row_ids: Vec<RowId>,
}

impl MutationResult {
    /// Create an empty result
    pub fn empty() -> Self {
        Self::default()
    }
    
    /// Create from count and IDs
    pub fn new(affected_count: usize, affected_row_ids: Vec<RowId>) -> Self {
        Self { affected_count, affected_row_ids }
    }
}

/// Column reference from TCH
#[derive(Debug, Clone)]
pub struct ColumnRef {
    pub id: u32,
    pub name: String,
    pub field_type: FieldType,
}

/// Field types for column store
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum FieldType {
    UInt64,
    Int64,
    Float64,
    Text,
    Bool,
    Bytes,
}

/// Path resolution result from TCH
#[derive(Debug, Clone)]
pub enum PathResolution {
    /// Found an array (table) with schema and column references
    Array {
        schema: Arc<ArraySchema>,
        columns: Vec<ColumnRef>,
    },
    /// Found a scalar value
    Value(ColumnRef),
    /// Partial match (intermediate node)
    Partial { remaining: String },
    /// Not found
    NotFound,
}

/// Schema for resolved array
#[derive(Debug, Clone)]
pub struct ArraySchema {
    pub name: String,
    pub fields: Vec<String>,
    pub types: Vec<FieldType>,
}

/// Simulated WAL storage manager for transactions
/// 
/// # ⚠️ DEPRECATED - DO NOT USE IN PRODUCTION
/// 
/// This is a **testing stub** that provides NO durability guarantees:
/// - No disk I/O occurs
/// - Data is lost on process exit
/// - No crash recovery
/// 
/// ## Migration
/// 
/// Replace with `sochdb_storage::WalStorageManager` or use [`DurableConnection`]:
/// 
/// ```rust,ignore
/// // Before (WRONG - no durability):
/// let conn = SochConnection::open("./data")?;
/// 
/// // After (CORRECT - full WAL durability):
/// let conn = DurableConnection::open("./data")?;
/// ```
/// 
/// See: <https://github.com/sochdb/sochdb/blob/main/docs/ARCHITECTURE.md#wal-integration>
#[deprecated(
    since = "0.2.0",
    note = "Use DurableConnection or sochdb_storage::WalStorageManager for production. \
            This stub provides NO durability. REMOVAL SCHEDULED: v0.3.0"
)]
pub struct WalStorageManager {
    next_txn_id: AtomicU64,
    // In real impl: WAL file handle, buffer, etc.
}

#[allow(deprecated)]
impl Default for WalStorageManager {
    fn default() -> Self {
        Self::new()
    }
}

#[allow(deprecated)]
impl WalStorageManager {
    pub fn new() -> Self {
        Self {
            next_txn_id: AtomicU64::new(1),
        }
    }

    pub fn begin_txn(&self) -> Result<TxnId> {
        Ok(self.next_txn_id.fetch_add(1, Ordering::SeqCst))
    }

    pub fn commit(&self, _txn_id: TxnId) -> Result<Timestamp> {
        // In real impl: flush to WAL, fsync
        Ok(self.next_txn_id.load(Ordering::SeqCst))
    }

    pub fn abort(&self, _txn_id: TxnId) -> Result<()> {
        // In real impl: mark aborted in WAL
        Ok(())
    }
}

/// Simulated transaction manager for MVCC
///
/// # ⚠️ DEPRECATED - DO NOT USE IN PRODUCTION
///
/// This is a **testing stub** with NO MVCC semantics:
/// - No snapshot isolation
/// - No conflict detection  
/// - No visibility tracking
/// - Uses simple counter instead of Lamport timestamps
///
/// ## Migration
///
/// Replace with `sochdb_storage::MvccTransactionManager` or use [`DurableConnection`]:
///
/// ```rust,ignore
/// // Before (WRONG - no MVCC):
/// let conn = SochConnection::open("./data")?;
///
/// // After (CORRECT - full MVCC with SSI):
/// let conn = DurableConnection::open("./data")?;
/// ```
///
/// See: <https://github.com/sochdb/sochdb/blob/main/docs/ARCHITECTURE.md#wal-integration>
#[deprecated(
    since = "0.2.0",
    note = "Use DurableConnection or sochdb_storage::MvccTransactionManager for production. \
            This stub has NO MVCC semantics. REMOVAL SCHEDULED: v0.3.0"
)]
pub struct TransactionManager {
    current_ts: AtomicU64,
    // In real impl: active transactions, commit timestamps, etc.
}

#[allow(deprecated)]
impl Default for TransactionManager {
    fn default() -> Self {
        Self::new()
    }
}

#[allow(deprecated)]
impl TransactionManager {
    pub fn new() -> Self {
        Self {
            current_ts: AtomicU64::new(1),
        }
    }

    pub fn begin(&self) -> (TxnId, Timestamp) {
        let ts = self.current_ts.fetch_add(1, Ordering::SeqCst);
        (ts, ts)
    }

    pub fn current_timestamp(&self) -> Timestamp {
        self.current_ts.load(Ordering::SeqCst)
    }

    pub fn commit(&self, _txn_id: TxnId) -> Result<Timestamp> {
        let ts = self.current_ts.fetch_add(1, Ordering::SeqCst);
        Ok(ts)
    }

    pub fn abort(&self, _txn_id: TxnId) -> Result<()> {
        Ok(())
    }
}

/// Bloom filter for efficient negative lookups
/// Uses optimal k = (m/n) * ln(2) hash functions
pub struct BloomFilter {
    bits: Vec<u64>,  // Bit array
    num_bits: usize, // Total bits (m)
    num_hashes: u8,  // Number of hash functions (k)
}

impl BloomFilter {
    /// Create bloom filter for n elements with target false positive rate
    pub fn new(expected_elements: usize, fp_rate: f64) -> Self {
        // m = -n * ln(p) / (ln(2)^2)
        let num_bits =
            (-(expected_elements as f64) * fp_rate.ln() / (2.0_f64.ln().powi(2))).ceil() as usize;
        let num_bits = std::cmp::max(num_bits, 64); // Minimum 64 bits
        let num_words = num_bits.div_ceil(64);

        // k = (m/n) * ln(2)
        let num_hashes = ((num_bits as f64 / expected_elements as f64) * 2.0_f64.ln()).ceil() as u8;
        let num_hashes = std::cmp::max(num_hashes, 1);

        Self {
            bits: vec![0; num_words],
            num_bits,
            num_hashes,
        }
    }

    /// Insert key into bloom filter
    pub fn insert(&mut self, key: &[u8]) {
        let (h1, h2) = self.hash_key(key);
        for i in 0..self.num_hashes as u64 {
            let hash = h1.wrapping_add(i.wrapping_mul(h2));
            let bit_idx = (hash % self.num_bits as u64) as usize;
            let word_idx = bit_idx / 64;
            let bit_offset = bit_idx % 64;
            self.bits[word_idx] |= 1 << bit_offset;
        }
    }

    /// Check if key might be present (false = definitely not present)
    pub fn may_contain(&self, key: &[u8]) -> bool {
        let (h1, h2) = self.hash_key(key);
        for i in 0..self.num_hashes as u64 {
            let hash = h1.wrapping_add(i.wrapping_mul(h2));
            let bit_idx = (hash % self.num_bits as u64) as usize;
            let word_idx = bit_idx / 64;
            let bit_offset = bit_idx % 64;
            if (self.bits[word_idx] & (1 << bit_offset)) == 0 {
                return false;
            }
        }
        true
    }

    /// Double hashing using FNV-1a
    fn hash_key(&self, key: &[u8]) -> (u64, u64) {
        // FNV-1a hash
        const FNV_OFFSET_BASIS: u64 = 0xcbf29ce484222325;
        const FNV_PRIME: u64 = 0x00000100000001B3;

        let mut h1 = FNV_OFFSET_BASIS;
        for &b in key {
            h1 ^= b as u64;
            h1 = h1.wrapping_mul(FNV_PRIME);
        }

        // Second hash: rotate and XOR
        let mut h2 = h1.rotate_left(17);
        h2 ^= h1;
        h2 = h2.wrapping_mul(FNV_PRIME);

        (h1, h2)
    }
}

/// SSTable entry with key, value, and metadata
#[derive(Clone)]
pub struct SstEntry {
    pub key: Vec<u8>,
    pub value: Vec<u8>,
    pub timestamp: u64,
    pub deleted: bool, // Tombstone marker
}

/// In-memory SSTable representation
/// In real impl, this would be memory-mapped file backed
#[allow(dead_code)]
pub struct SSTable {
    /// Sorted entries (key -> entry)
    entries: Vec<SstEntry>,
    /// Bloom filter for key existence checks
    bloom: BloomFilter,
    /// Minimum key in table
    min_key: Vec<u8>,
    /// Maximum key in table  
    max_key: Vec<u8>,
    /// Level in LSM tree (0 = newest)
    level: usize,
    /// Unique sequence number
    seq_num: u64,
}

/// An SSTable that can be either in-memory or disk-backed.
///
/// When `LscsStorage` has a configured `data_dir`, flushed SSTables are
/// written to disk using `sochdb_storage::SSTableBuilder` and read back
/// via `sochdb_storage::SSTable`. Otherwise, the legacy in-memory path is used.
enum SstHandle {
    /// In-memory SSTable (no persistence — testing only)
    InMemory(SSTable),
    /// Disk-backed SSTable via sochdb-storage
    OnDisk {
        reader: sochdb_storage::SSTable,
        min_key: Vec<u8>,
        max_key: Vec<u8>,
        level: usize,
        seq_num: u64,
    },
}

impl SSTable {
    /// Create SSTable from sorted entries
    pub fn from_entries(entries: Vec<SstEntry>, level: usize, seq_num: u64) -> Option<Self> {
        if entries.is_empty() {
            return None;
        }

        let mut bloom = BloomFilter::new(entries.len(), 0.01); // 1% FPR
        for entry in &entries {
            bloom.insert(&entry.key);
        }

        let min_key = entries.first()?.key.clone();
        let max_key = entries.last()?.key.clone();

        Some(Self {
            entries,
            bloom,
            min_key,
            max_key,
            level,
            seq_num,
        })
    }

    /// Check if key is in range of this SSTable
    pub fn key_in_range(&self, key: &[u8]) -> bool {
        key >= self.min_key.as_slice() && key <= self.max_key.as_slice()
    }

    /// Get value for key (O(log n) binary search)
    pub fn get(&self, key: &[u8]) -> Option<&SstEntry> {
        // Fast path: bloom filter check
        if !self.bloom.may_contain(key) {
            return None;
        }

        // Binary search
        match self.entries.binary_search_by(|e| e.key.as_slice().cmp(key)) {
            Ok(idx) => Some(&self.entries[idx]),
            Err(_) => None,
        }
    }
}

/// Level in LSM tree containing multiple SSTables
struct Level {
    sstables: Vec<SstHandle>,
    target_size: u64,
}

impl Level {
    fn new(target_size: u64) -> Self {
        Self {
            sstables: Vec::new(),
            target_size,
        }
    }

    fn total_size(&self) -> u64 {
        self.sstables
            .iter()
            .map(|sst| match sst {
                SstHandle::InMemory(s) => s.entries
                    .iter()
                    .map(|e| e.key.len() + e.value.len())
                    .sum::<usize>() as u64,
                SstHandle::OnDisk { reader, .. } => {
                    reader.metadata().file_size
                }
            })
            .sum()
    }
}

/// Memtable entry
struct MemTableEntry {
    value: Vec<u8>,
    timestamp: u64,
    deleted: bool,
}

/// Log-Structured Column Store implementing LSM-tree
///
/// Write path: Memtable → Immutable Memtable → L0 SSTables → L1 → L2...
/// Read path: Memtable → Immutable → L0 (newest first) → L1 → L2...
pub struct LscsStorage {
    /// Active memtable (skiplist for O(log n) ops)
    memtable: parking_lot::RwLock<std::collections::BTreeMap<Vec<u8>, MemTableEntry>>,
    /// Memtable size in bytes
    memtable_size: std::sync::atomic::AtomicU64,
    /// Immutable memtables waiting to be flushed
    immutable_memtables:
        parking_lot::RwLock<Vec<std::collections::BTreeMap<Vec<u8>, MemTableEntry>>>,
    /// LSM tree levels (L0, L1, L2, ...)
    levels: parking_lot::RwLock<Vec<Level>>,
    /// Next SSTable sequence number
    next_seq: std::sync::atomic::AtomicU64,
    /// Current LSN
    current_lsn: std::sync::atomic::AtomicU64,
    /// Checkpoint LSN
    checkpoint_lsn: std::sync::atomic::AtomicU64,
    /// Write-ahead log entries (in-memory for now)
    wal_entries: parking_lot::RwLock<Vec<WalEntry>>,
    /// Statistics
    stats: LscsStats,
    /// Configuration
    config: LscsConfig,
}

/// WAL entry
struct WalEntry {
    lsn: u64,
    key: Vec<u8>,
    value: Vec<u8>,
    op: WalOp,
    timestamp: u64,
    checksum: u32,
}

#[derive(Clone, Copy)]
enum WalOp {
    Put,
    Delete,
}

/// LSCS configuration
struct LscsConfig {
    /// Memtable size threshold for flush (default 4MB)
    memtable_flush_size: u64,
    /// Target size for L0 (default 64MB)
    l0_target_size: u64,
    /// Size ratio between levels (default 10)
    level_size_ratio: u64,
    /// Max levels
    max_levels: usize,
    /// Data directory for persistent SSTables.
    /// When None, SSTables are kept in-memory only (testing mode).
    /// When Some, SSTables are written to disk via sochdb_storage::SSTableBuilder.
    data_dir: Option<PathBuf>,
}

impl Default for LscsConfig {
    fn default() -> Self {
        Self {
            memtable_flush_size: 4 * 1024 * 1024, // 4MB
            l0_target_size: 64 * 1024 * 1024,     // 64MB
            level_size_ratio: 10,
            max_levels: 7,
            data_dir: None,
        }
    }
}

/// Storage statistics
struct LscsStats {
    writes: std::sync::atomic::AtomicU64,
    reads: std::sync::atomic::AtomicU64,
    bloom_filter_hits: std::sync::atomic::AtomicU64,
    bloom_filter_misses: std::sync::atomic::AtomicU64,
}

/// WAL verification result
#[derive(Debug, Clone)]
pub struct WalVerifyResult {
    pub total_entries: u64,
    pub valid_entries: u64,
    pub corrupted_entries: u64,
    pub last_valid_lsn: u64,
    pub checksum_errors: Vec<ChecksumErr>,
}

/// Checksum error info
#[derive(Debug, Clone)]
pub struct ChecksumErr {
    pub lsn: u64,
    pub expected: u64,
    pub actual: u64,
    pub entry_type: String,
}

/// WAL statistics
#[derive(Debug, Clone)]
pub struct WalStatsData {
    pub total_size_bytes: u64,
    pub active_size_bytes: u64,
    pub archived_size_bytes: u64,
    pub oldest_entry_lsn: u64,
    pub newest_entry_lsn: u64,
    pub entry_count: u64,
}

impl Default for LscsStorage {
    fn default() -> Self {
        Self::new()
    }
}

impl LscsStorage {
    /// Create new LSCS storage engine (in-memory only — testing mode)
    pub fn new() -> Self {
        Self::with_config(LscsConfig::default())
    }

    /// Create LSCS storage engine with persistent SSTable storage on disk.
    ///
    /// SSTables flushed from memtable are written to `data_dir` using the
    /// disk-backed SSTableBuilder from sochdb-storage, providing durability
    /// for data that has been flushed from the memtable.
    ///
    /// # Arguments
    /// * `data_dir` - Directory where SSTable files will be stored
    pub fn with_data_dir(data_dir: impl AsRef<Path>) -> Result<Self> {
        let data_dir = data_dir.as_ref().to_path_buf();
        std::fs::create_dir_all(&data_dir)
            .map_err(|e| ClientError::Storage(format!("Failed to create data dir: {}", e)))?;
        let mut config = LscsConfig::default();
        config.data_dir = Some(data_dir);
        Ok(Self::with_config(config))
    }

    fn with_config(config: LscsConfig) -> Self {
        let mut levels = Vec::with_capacity(config.max_levels);

        // Initialize levels with exponentially increasing target sizes
        for i in 0..config.max_levels {
            let target = if i == 0 {
                config.l0_target_size
            } else {
                config.l0_target_size * config.level_size_ratio.pow(i as u32)
            };
            levels.push(Level::new(target));
        }

        Self {
            memtable: parking_lot::RwLock::new(std::collections::BTreeMap::new()),
            memtable_size: std::sync::atomic::AtomicU64::new(0),
            immutable_memtables: parking_lot::RwLock::new(Vec::new()),
            levels: parking_lot::RwLock::new(levels),
            next_seq: std::sync::atomic::AtomicU64::new(1),
            current_lsn: std::sync::atomic::AtomicU64::new(1),
            checkpoint_lsn: std::sync::atomic::AtomicU64::new(0),
            wal_entries: parking_lot::RwLock::new(Vec::new()),
            stats: LscsStats {
                writes: std::sync::atomic::AtomicU64::new(0),
                reads: std::sync::atomic::AtomicU64::new(0),
                bloom_filter_hits: std::sync::atomic::AtomicU64::new(0),
                bloom_filter_misses: std::sync::atomic::AtomicU64::new(0),
            },
            config,
        }
    }

    /// Put a key-value pair into storage
    pub fn put(&self, key: &[u8], value: &[u8]) -> Result<()> {
        use std::sync::atomic::Ordering;

        // 1. Write to WAL first (durability)
        let lsn = self.current_lsn.fetch_add(1, Ordering::SeqCst);
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_micros() as u64;

        let checksum = Self::compute_checksum(key, value, timestamp);

        {
            let mut wal = self.wal_entries.write();
            wal.push(WalEntry {
                lsn,
                key: key.to_vec(),
                value: value.to_vec(),
                op: WalOp::Put,
                timestamp,
                checksum,
            });
        }

        // 2. Write to memtable
        let entry_size = key.len() + value.len() + 24; // key + value + metadata
        {
            let mut memtable = self.memtable.write();
            memtable.insert(
                key.to_vec(),
                MemTableEntry {
                    value: value.to_vec(),
                    timestamp,
                    deleted: false,
                },
            );
        }

        let new_size = self
            .memtable_size
            .fetch_add(entry_size as u64, Ordering::SeqCst)
            + entry_size as u64;
        self.stats.writes.fetch_add(1, Ordering::Relaxed);

        // 3. Check if memtable needs flush
        if new_size >= self.config.memtable_flush_size {
            self.maybe_flush_memtable()?;
        }

        Ok(())
    }

    /// Delete a key
    pub fn delete(&self, key: &[u8]) -> Result<()> {
        use std::sync::atomic::Ordering;

        // Write tombstone to WAL
        let lsn = self.current_lsn.fetch_add(1, Ordering::SeqCst);
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_micros() as u64;

        let checksum = Self::compute_checksum(key, &[], timestamp);

        {
            let mut wal = self.wal_entries.write();
            wal.push(WalEntry {
                lsn,
                key: key.to_vec(),
                value: Vec::new(),
                op: WalOp::Delete,
                timestamp,
                checksum,
            });
        }

        // Write tombstone to memtable
        {
            let mut memtable = self.memtable.write();
            memtable.insert(
                key.to_vec(),
                MemTableEntry {
                    value: Vec::new(),
                    timestamp,
                    deleted: true,
                },
            );
        }

        self.memtable_size
            .fetch_add(key.len() as u64 + 24, std::sync::atomic::Ordering::SeqCst);

        Ok(())
    }

    /// Get value for key - searches memtable then levels (newest to oldest)
    pub fn get(&self, _table: &str, key: &[u8]) -> Result<Option<Vec<u8>>> {
        use std::sync::atomic::Ordering;
        self.stats.reads.fetch_add(1, Ordering::Relaxed);

        // 1. Check active memtable
        {
            let memtable = self.memtable.read();
            if let Some(entry) = memtable.get(key) {
                if entry.deleted {
                    return Ok(None); // Tombstone
                }
                return Ok(Some(entry.value.clone()));
            }
        }

        // 2. Check immutable memtables (newest first)
        {
            let immutables = self.immutable_memtables.read();
            for memtable in immutables.iter().rev() {
                if let Some(entry) = memtable.get(key) {
                    if entry.deleted {
                        return Ok(None);
                    }
                    return Ok(Some(entry.value.clone()));
                }
            }
        }

        // 3. Check SSTable levels (L0 newest first, then L1, L2, ...)
        {
            let levels = self.levels.read();
            for level in levels.iter() {
                // L0: check all SSTables (may overlap)
                // L1+: can use binary search (non-overlapping)
                for sst in level.sstables.iter().rev() {
                    match sst {
                        SstHandle::InMemory(s) => {
                            if s.key_in_range(key) {
                                if let Some(entry) = s.get(key) {
                                    if entry.deleted {
                                        return Ok(None);
                                    }
                                    self.stats.bloom_filter_hits.fetch_add(1, Ordering::Relaxed);
                                    return Ok(Some(entry.value.clone()));
                                } else {
                                    self.stats
                                        .bloom_filter_misses
                                        .fetch_add(1, Ordering::Relaxed);
                                }
                            }
                        }
                        SstHandle::OnDisk { reader, min_key, max_key, .. } => {
                            if key >= min_key.as_slice() && key <= max_key.as_slice() {
                                let opts = sochdb_storage::ReadOptions::default();
                                match reader.get(key, &opts) {
                                    Ok(Some(value)) => {
                                        if value.is_empty() {
                                            // Empty value = tombstone for disk SSTables
                                            return Ok(None);
                                        }
                                        self.stats.bloom_filter_hits.fetch_add(1, Ordering::Relaxed);
                                        return Ok(Some(value));
                                    }
                                    Ok(None) => {
                                        self.stats
                                            .bloom_filter_misses
                                            .fetch_add(1, Ordering::Relaxed);
                                    }
                                    Err(_) => {
                                        // I/O error reading SSTable — skip this table
                                        continue;
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }

        Ok(None)
    }

    /// Flush memtable to immutable and trigger compaction if needed
    fn maybe_flush_memtable(&self) -> Result<()> {
        use std::sync::atomic::Ordering;

        // Swap memtable with empty one
        let old_memtable = {
            let mut memtable = self.memtable.write();
            let old = std::mem::take(&mut *memtable);
            self.memtable_size.store(0, Ordering::SeqCst);
            old
        };

        if old_memtable.is_empty() {
            return Ok(());
        }

        // Convert to SSTable entries (sorted by key since BTreeMap)
        let entries: Vec<SstEntry> = old_memtable
            .into_iter()
            .map(|(key, entry)| SstEntry {
                key,
                value: entry.value,
                timestamp: entry.timestamp,
                deleted: entry.deleted,
            })
            .collect();

        let seq = self.next_seq.fetch_add(1, Ordering::SeqCst);
        let level = 0usize;

        // Choose persistence path based on data_dir configuration
        let sst_handle = if let Some(ref data_dir) = self.config.data_dir {
            // DISK PATH: Write SSTable to disk using sochdb-storage builder
            let sst_path = data_dir.join(format!("L{}_{:08}.sst", level, seq));

            let opts = sochdb_storage::SSTableBuilderOptions::default();
            let mut builder = sochdb_storage::SSTableBuilder::new(&sst_path, opts)
                .map_err(|e| ClientError::Storage(format!("SSTable create failed: {}", e)))?;

            builder.set_estimated_keys(entries.len());

            for entry in &entries {
                // For tombstones, write empty value; actual tombstone handling
                // is done at read time via the deleted flag in compaction
                let val = if entry.deleted { &[][..] } else { &entry.value[..] };
                builder.add(&entry.key, val)
                    .map_err(|e| ClientError::Storage(format!("SSTable write failed: {}", e)))?;
            }

            let result = builder.finish()
                .map_err(|e| ClientError::Storage(format!("SSTable finish failed: {}", e)))?;

            let min_key = result.smallest_key.clone().unwrap_or_default();
            let max_key = result.largest_key.clone().unwrap_or_default();

            // Open the written SSTable for reads
            let reader = sochdb_storage::SSTable::open(&sst_path)
                .map_err(|e| ClientError::Storage(format!("SSTable open failed: {}", e)))?;

            SstHandle::OnDisk {
                reader,
                min_key,
                max_key,
                level,
                seq_num: seq,
            }
        } else {
            // IN-MEMORY PATH: Original behavior (testing only)
            match SSTable::from_entries(entries, level, seq) {
                Some(sst) => SstHandle::InMemory(sst),
                None => return Ok(()),
            }
        };

        let mut levels = self.levels.write();
        levels[0].sstables.push(sst_handle);

        // Check if L0 needs compaction
        if levels[0].total_size() > levels[0].target_size {
            drop(levels);
            self.maybe_compact()?;
        }

        Ok(())
    }

    /// Perform leveled compaction if needed
    fn maybe_compact(&self) -> Result<()> {
        use std::sync::atomic::Ordering;

        let mut levels = self.levels.write();

        // Find level that exceeds target size
        for i in 0..levels.len() - 1 {
            if levels[i].total_size() > levels[i].target_size {
                // Merge all SSTables in level i with overlapping ones in level i+1
                if levels[i].sstables.is_empty() {
                    continue;
                }

                // Take all entries from level i
                let mut all_entries: Vec<SstEntry> = Vec::new();
                for sst in levels[i].sstables.drain(..) {
                    Self::drain_sst_entries(sst, &mut all_entries);
                }

                // Merge with level i+1 entries
                for sst in levels[i + 1].sstables.drain(..) {
                    Self::drain_sst_entries(sst, &mut all_entries);
                }

                // Sort by key, then by timestamp (newest first for dedup)
                all_entries.sort_by(|a, b| match a.key.cmp(&b.key) {
                    std::cmp::Ordering::Equal => b.timestamp.cmp(&a.timestamp),
                    other => other,
                });

                // Deduplicate, keeping newest version
                all_entries.dedup_by(|a, b| a.key == b.key);

                // Remove tombstones (only at bottom level)
                if i + 1 == levels.len() - 1 {
                    all_entries.retain(|e| !e.deleted);
                }

                // Create new SSTable at level i+1
                let seq = self.next_seq.fetch_add(1, Ordering::SeqCst);
                if let Some(sst) = SSTable::from_entries(all_entries, i + 1, seq) {
                    levels[i + 1].sstables.push(SstHandle::InMemory(sst));
                }
            }
        }

        Ok(())
    }

    /// Extract all entries from an SstHandle into a Vec<SstEntry>.
    /// For disk-backed SSTables, reads all blocks using the SSTableIterator.
    fn drain_sst_entries(sst: SstHandle, out: &mut Vec<SstEntry>) {
        match sst {
            SstHandle::InMemory(s) => {
                out.extend(s.entries);
            }
            SstHandle::OnDisk { reader, seq_num, .. } => {
                // Iterate through all entries in the on-disk SSTable
                let mut iter = reader.iter();
                while iter.valid() {
                    if let (Some(key), Some(val)) = (iter.key(), iter.value()) {
                        let deleted = val.is_empty(); // empty value = tombstone
                        out.push(SstEntry {
                            key: key.to_vec(),
                            value: val.to_vec(),
                            timestamp: seq_num, // use seq_num as ordering timestamp
                            deleted,
                        });
                    }
                    iter.next();
                }
            }
        }
    }

    /// Compute deterministic CRC32C checksum for WAL entry verification
    ///
    /// Uses hardware-accelerated CRC32C (via SSE4.2 on x86-64) which is:
    /// - Deterministic across process restarts (unlike SipHash with random seed)
    /// - 6× faster than SipHash (~0.5 cycles/byte vs ~3 cycles/byte)
    /// - Detects all 1-bit, 2-bit, and 3-bit errors in messages up to 2^31 bits
    ///
    /// Previous implementation used std::collections::hash_map::DefaultHasher (SipHash-1-3)
    /// which has a randomized seed per process, making verification non-functional
    /// across process restarts.
    fn compute_checksum(key: &[u8], value: &[u8], timestamp: u64) -> u32 {
        let mut hasher = crc32fast::Hasher::new();
        hasher.update(key);
        hasher.update(value);
        hasher.update(&timestamp.to_le_bytes());
        hasher.finalize()
    }

    pub fn allocate_page(&self) -> Result<u64> {
        Ok(self
            .next_seq
            .fetch_add(1, std::sync::atomic::Ordering::SeqCst))
    }

    /// Flush and sync to disk
    pub fn fsync(&self) -> Result<()> {
        // Flush any pending memtable data
        self.maybe_flush_memtable()?;
        Ok(())
    }

    /// Check if recovery is needed
    pub fn needs_recovery(&self) -> bool {
        // Check if WAL has entries beyond checkpoint
        let wal = self.wal_entries.read();
        let checkpoint = self
            .checkpoint_lsn
            .load(std::sync::atomic::Ordering::SeqCst);
        wal.iter().any(|e| e.lsn > checkpoint)
    }

    /// Get last checkpoint LSN
    pub fn last_checkpoint_lsn(&self) -> u64 {
        self.checkpoint_lsn
            .load(std::sync::atomic::Ordering::SeqCst)
    }

    /// Get current WAL LSN
    pub fn current_lsn(&self) -> u64 {
        self.current_lsn.load(std::sync::atomic::Ordering::SeqCst)
    }

    /// Verify WAL integrity
    pub fn verify_wal(&self) -> Result<WalVerifyResult> {
        let wal = self.wal_entries.read();
        let mut valid = 0u64;
        let mut corrupted = 0u64;
        let mut errors = Vec::new();
        let mut last_valid_lsn = 0u64;

        for entry in wal.iter() {
            let expected = Self::compute_checksum(&entry.key, &entry.value, entry.timestamp);
            if expected == entry.checksum {
                valid += 1;
                last_valid_lsn = entry.lsn;
            } else {
                corrupted += 1;
                errors.push(ChecksumErr {
                    lsn: entry.lsn,
                    expected: expected as u64,
                    actual: entry.checksum as u64,
                    entry_type: match entry.op {
                        WalOp::Put => "PUT".to_string(),
                        WalOp::Delete => "DELETE".to_string(),
                    },
                });
            }
        }

        Ok(WalVerifyResult {
            total_entries: wal.len() as u64,
            valid_entries: valid,
            corrupted_entries: corrupted,
            last_valid_lsn,
            checksum_errors: errors,
        })
    }

    /// Replay WAL from checkpoint
    pub fn replay_wal_from_checkpoint(&self) -> Result<u64> {
        let checkpoint = self
            .checkpoint_lsn
            .load(std::sync::atomic::Ordering::SeqCst);
        let wal = self.wal_entries.read();
        let mut replayed = 0u64;

        for entry in wal.iter() {
            if entry.lsn > checkpoint {
                // Verify checksum before replay
                let expected = Self::compute_checksum(&entry.key, &entry.value, entry.timestamp);
                if expected == entry.checksum {
                    let mut memtable = self.memtable.write();
                    match entry.op {
                        WalOp::Put => {
                            memtable.insert(
                                entry.key.clone(),
                                MemTableEntry {
                                    value: entry.value.clone(),
                                    timestamp: entry.timestamp,
                                    deleted: false,
                                },
                            );
                        }
                        WalOp::Delete => {
                            memtable.insert(
                                entry.key.clone(),
                                MemTableEntry {
                                    value: Vec::new(),
                                    timestamp: entry.timestamp,
                                    deleted: true,
                                },
                            );
                        }
                    }
                    replayed += 1;
                }
            }
        }

        Ok(replayed)
    }

    /// Force checkpoint
    pub fn force_checkpoint(&self) -> Result<u64> {
        // Flush memtable
        self.maybe_flush_memtable()?;

        // Update checkpoint LSN
        let current = self.current_lsn.load(std::sync::atomic::Ordering::SeqCst);
        self.checkpoint_lsn
            .store(current, std::sync::atomic::Ordering::SeqCst);

        Ok(current)
    }

    /// Truncate WAL up to LSN
    pub fn truncate_wal(&self, up_to_lsn: u64) -> Result<u64> {
        let mut wal = self.wal_entries.write();
        let before_len = wal.len();
        wal.retain(|e| e.lsn > up_to_lsn);
        let removed = before_len - wal.len();
        Ok(removed as u64)
    }

    /// Get WAL statistics
    pub fn wal_stats(&self) -> WalStatsData {
        let wal = self.wal_entries.read();
        let total_size: u64 = wal
            .iter()
            .map(|e| (e.key.len() + e.value.len() + 32) as u64)
            .sum();

        let checkpoint = self
            .checkpoint_lsn
            .load(std::sync::atomic::Ordering::SeqCst);
        let active_size: u64 = wal
            .iter()
            .filter(|e| e.lsn > checkpoint)
            .map(|e| (e.key.len() + e.value.len() + 32) as u64)
            .sum();

        WalStatsData {
            total_size_bytes: total_size,
            active_size_bytes: active_size,
            archived_size_bytes: total_size.saturating_sub(active_size),
            oldest_entry_lsn: wal.first().map(|e| e.lsn).unwrap_or(0),
            newest_entry_lsn: wal.last().map(|e| e.lsn).unwrap_or(0),
            entry_count: wal.len() as u64,
        }
    }

    /// Scan range of keys
    pub fn scan(
        &self,
        start_key: &[u8],
        end_key: &[u8],
        limit: usize,
    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        use std::collections::BTreeMap;

        let mut results: BTreeMap<Vec<u8>, (Vec<u8>, u64, bool)> = BTreeMap::new();

        // Collect from all sources, tracking newest timestamp
        // 1. Memtable
        {
            let memtable = self.memtable.read();
            for (key, entry) in memtable.range(start_key.to_vec()..=end_key.to_vec()) {
                results.insert(
                    key.clone(),
                    (entry.value.clone(), entry.timestamp, entry.deleted),
                );
            }
        }

        // 2. Immutable memtables
        {
            let immutables = self.immutable_memtables.read();
            for memtable in immutables.iter() {
                for (key, entry) in memtable.range(start_key.to_vec()..=end_key.to_vec()) {
                    results
                        .entry(key.clone())
                        .and_modify(|e| {
                            if entry.timestamp > e.1 {
                                *e = (entry.value.clone(), entry.timestamp, entry.deleted);
                            }
                        })
                        .or_insert_with(|| (entry.value.clone(), entry.timestamp, entry.deleted));
                }
            }
        }

        // 3. SSTables
        {
            let levels = self.levels.read();
            for level in levels.iter() {
                for sst in &level.sstables {
                    match sst {
                        SstHandle::InMemory(s) => {
                            for entry in &s.entries {
                                if entry.key >= start_key.to_vec() && entry.key <= end_key.to_vec() {
                                    results
                                        .entry(entry.key.clone())
                                        .and_modify(|e| {
                                            if entry.timestamp > e.1 {
                                                *e = (entry.value.clone(), entry.timestamp, entry.deleted);
                                            }
                                        })
                                        .or_insert_with(|| {
                                            (entry.value.clone(), entry.timestamp, entry.deleted)
                                        });
                                }
                            }
                        }
                        SstHandle::OnDisk { reader, min_key, max_key, .. } => {
                            // Skip SSTables whose key range doesn't overlap [start_key, end_key]
                            if max_key.as_slice() < start_key || min_key.as_slice() > end_key {
                                continue;
                            }

                            // Use the SSTableIterator to scan the on-disk SSTable.
                            // Seek to start_key, then iterate until we pass end_key.
                            let mut sst_iter = reader.iter();
                            sst_iter.seek(start_key);

                            while sst_iter.valid() {
                                let key = match sst_iter.key() {
                                    Some(k) => k,
                                    None => break,
                                };

                                // Past end of range — done with this SSTable
                                if key > end_key {
                                    break;
                                }

                                let val = sst_iter.value().unwrap_or(&[]);
                                // Empty value = tombstone for disk SSTables
                                let deleted = val.is_empty();

                                let key_vec = key.to_vec();
                                let val_vec = val.to_vec();

                                // Use SSTable seq_num as the "timestamp" for
                                // shadowing: newer SSTables have higher seq_num
                                // but for on-disk entries we use 0 as a
                                // conservative timestamp so memtable entries
                                // (which always have real timestamps) win.
                                // Within the SSTable tier, iteration order
                                // is newest-first (rev), and `.entry().and_modify`
                                // only replaces if timestamp is strictly greater.
                                results
                                    .entry(key_vec)
                                    .and_modify(|e: &mut (Vec<u8>, u64, bool)| {
                                        // Keep the existing (newer) entry
                                    })
                                    .or_insert_with(|| (val_vec, 0, deleted));

                                sst_iter.next();
                            }
                        }
                    }
                }
            }
        }

        // Filter out tombstones and limit
        let result: Vec<(Vec<u8>, Vec<u8>)> = results
            .into_iter()
            .filter(|(_, (_, _, deleted))| !deleted)
            .take(limit)
            .map(|(k, (v, _, _))| (k, v))
            .collect();

        Ok(result)
    }

    /// Force flush memtable to SST
    /// 
    /// Returns the number of bytes flushed.
    pub fn flush(&self) -> Result<usize> {
        use std::sync::atomic::Ordering;
        
        let memtable_size = self.memtable_size.load(Ordering::SeqCst) as usize;
        self.maybe_flush_memtable()?;
        Ok(memtable_size)
    }

    /// Force compaction of all levels
    /// 
    /// Returns compaction metrics.
    pub fn compact(&self) -> Result<CompactionMetrics> {
        
        
        // First flush any pending memtable
        let flushed_bytes = self.flush()? as u64;
        
        // Get pre-compaction stats
        let levels = self.levels.read();
        let pre_files: usize = levels.iter().map(|l| l.sstables.len()).sum();
        let pre_bytes: u64 = levels.iter().map(|l| l.total_size()).sum();
        drop(levels);
        
        // Perform compaction
        self.maybe_compact()?;
        
        // Get post-compaction stats
        let levels = self.levels.read();
        let post_files: usize = levels.iter().map(|l| l.sstables.len()).sum();
        let post_bytes: u64 = levels.iter().map(|l| l.total_size()).sum();
        
        Ok(CompactionMetrics {
            bytes_compacted: Some(flushed_bytes + pre_bytes.saturating_sub(post_bytes)),
            files_merged: Some(pre_files.saturating_sub(post_files)),
        })
    }
}

/// Compaction metrics returned by compact()
#[derive(Debug, Clone, Default)]
pub struct CompactionMetrics {
    /// Bytes compacted/reclaimed
    pub bytes_compacted: Option<u64>,
    /// Number of files merged
    pub files_merged: Option<usize>,
}

/// Simulated Trie-Columnar Hybrid with actual data storage
///
/// Key insight: O(|path|) resolution independent of N rows
pub struct TrieColumnarHybrid {
    /// Registered tables
    tables: hashbrown::HashMap<String, TableInfo>,
    /// Actual columnar data storage
    data: ColumnStore,
}

/// Columnar data storage with MVCC support
struct ColumnStore {
    /// Column data: table -> column_name -> Vec<ColumnValue>
    columns: hashbrown::HashMap<String, hashbrown::HashMap<String, Vec<ColumnValue>>>,
    /// Row metadata: table -> Vec<RowMetadata>
    row_meta: hashbrown::HashMap<String, Vec<RowMetadata>>,
}

/// A single value in a column (with null support)
#[derive(Debug, Clone)]
enum ColumnValue {
    Null,
    UInt64(u64),
    Int64(i64),
    Float64(f64),
    Text(String),
    Bool(bool),
    Bytes(Vec<u8>),
}

impl From<&SochValue> for ColumnValue {
    fn from(v: &SochValue) -> Self {
        match v {
            SochValue::Null => ColumnValue::Null,
            SochValue::Int(i) => ColumnValue::Int64(*i),
            SochValue::UInt(u) => ColumnValue::UInt64(*u),
            SochValue::Float(f) => ColumnValue::Float64(*f),
            SochValue::Text(s) => ColumnValue::Text(s.clone()),
            SochValue::Bool(b) => ColumnValue::Bool(*b),
            SochValue::Binary(b) => ColumnValue::Bytes(b.clone()),
            _ => ColumnValue::Text(format!("{:?}", v)), // Fallback for complex types
        }
    }
}

impl From<ColumnValue> for SochValue {
    fn from(val: ColumnValue) -> Self {
        match val {
            ColumnValue::Null => SochValue::Null,
            ColumnValue::UInt64(u) => SochValue::UInt(u),
            ColumnValue::Int64(i) => SochValue::Int(i),
            ColumnValue::Float64(f) => SochValue::Float(f),
            ColumnValue::Text(s) => SochValue::Text(s),
            ColumnValue::Bool(b) => SochValue::Bool(b),
            ColumnValue::Bytes(b) => SochValue::Binary(b),
        }
    }
}

/// Row metadata for MVCC
#[allow(dead_code)]
#[derive(Debug, Clone)]
struct RowMetadata {
    /// Row ID (unique within table)
    row_id: u64,
    /// Transaction start timestamp (when row was created)
    txn_start: u64,
    /// Transaction end timestamp (when row was deleted, 0 = active)
    txn_end: u64,
    /// Whether row is deleted (tombstone)
    deleted: bool,
}

impl ColumnStore {
    fn new() -> Self {
        Self {
            columns: hashbrown::HashMap::new(),
            row_meta: hashbrown::HashMap::new(),
        }
    }

    /// Initialize storage for a table
    fn init_table(&mut self, table: &str, columns: &[String]) {
        let mut table_cols = hashbrown::HashMap::new();
        for col in columns {
            table_cols.insert(col.clone(), Vec::new());
        }
        self.columns.insert(table.to_string(), table_cols);
        self.row_meta.insert(table.to_string(), Vec::new());
    }

    /// Insert a row
    fn insert(
        &mut self,
        table: &str,
        row_id: u64,
        values: &std::collections::HashMap<String, SochValue>,
    ) -> bool {
        let table_cols = match self.columns.get_mut(table) {
            Some(c) => c,
            None => return false,
        };
        let row_meta = match self.row_meta.get_mut(table) {
            Some(m) => m,
            None => return false,
        };

        // Add value to each column
        for (col_name, col_data) in table_cols.iter_mut() {
            let value = values
                .get(col_name)
                .map(ColumnValue::from)
                .unwrap_or(ColumnValue::Null);
            col_data.push(value);
        }

        // Add row metadata
        row_meta.push(RowMetadata {
            row_id,
            txn_start: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_micros() as u64,
            txn_end: 0,
            deleted: false,
        });

        true
    }

    /// Get a row by index
    fn get_row(
        &self,
        table: &str,
        row_idx: usize,
        columns: &[String],
    ) -> Option<std::collections::HashMap<String, SochValue>> {
        let table_cols = self.columns.get(table)?;
        let row_meta = self.row_meta.get(table)?;

        if row_idx >= row_meta.len() {
            return None;
        }

        let meta = &row_meta[row_idx];
        if meta.deleted {
            return None;
        }

        let mut row = std::collections::HashMap::new();
        for col_name in columns {
            if let Some(col_data) = table_cols.get(col_name)
                && row_idx < col_data.len()
            {
                row.insert(col_name.clone(), col_data[row_idx].clone().into());
            }
        }

        Some(row)
    }

    /// Get all visible rows
    fn get_all_rows(
        &self,
        table: &str,
        columns: &[String],
    ) -> Vec<(usize, std::collections::HashMap<String, SochValue>)> {
        let mut results = Vec::new();

        if let Some(row_meta) = self.row_meta.get(table) {
            for (idx, meta) in row_meta.iter().enumerate() {
                if !meta.deleted
                    && let Some(row) = self.get_row(table, idx, columns)
                {
                    results.push((idx, row));
                }
            }
        }

        results
    }

    /// Optimized batch scan with pre-allocation and vectorized column access
    /// 
    /// Performance optimizations:
    /// 1. Pre-allocates result vector based on non-deleted row count
    /// 2. Batches column reads to reduce HashMap lookups
    /// 3. Uses cache-friendly sequential iteration over column arrays
    /// 
    /// Target: ≤1.1× SQLite overhead for full table scans
    fn get_all_rows_optimized(
        &self,
        table: &str,
        columns: &[String],
    ) -> Vec<(usize, std::collections::HashMap<String, SochValue>)> {
        let table_cols = match self.columns.get(table) {
            Some(c) => c,
            None => return Vec::new(),
        };
        let row_meta = match self.row_meta.get(table) {
            Some(m) => m,
            None => return Vec::new(),
        };

        // Count non-deleted rows for pre-allocation
        let visible_count = row_meta.iter().filter(|m| !m.deleted).count();
        let mut results = Vec::with_capacity(visible_count);

        // Pre-fetch column references to avoid repeated HashMap lookups
        let col_refs: Vec<(&String, Option<&Vec<ColumnValue>>)> = columns
            .iter()
            .map(|c| (c, table_cols.get(c)))
            .collect();

        // Sequential scan with batched column access
        for (idx, meta) in row_meta.iter().enumerate() {
            if meta.deleted {
                continue;
            }

            // Build row with pre-fetched column references
            let mut row = std::collections::HashMap::with_capacity(columns.len());
            for (col_name, col_data_opt) in &col_refs {
                if let Some(col_data) = col_data_opt
                    && idx < col_data.len()
                {
                    row.insert((*col_name).clone(), col_data[idx].clone().into());
                }
            }
            results.push((idx, row));
        }

        results
    }

    /// Streaming batch iterator for very large tables
    /// 
    /// Yields rows in batches to reduce peak memory usage.
    /// Each batch is cache-line aligned for optimal CPU prefetch.
    #[allow(dead_code)]
    fn iter_rows_batched<'a>(
        &'a self,
        table: &str,
        columns: &'a [String],
        batch_size: usize,
    ) -> impl Iterator<Item = Vec<(usize, std::collections::HashMap<String, SochValue>)>> + 'a {
        let table_cols = self.columns.get(table);
        let row_meta = self.row_meta.get(table);

        let col_refs: Vec<(&String, Option<&Vec<ColumnValue>>)> = match table_cols {
            Some(tc) => columns.iter().map(|c| (c, tc.get(c))).collect(),
            None => Vec::new(),
        };

        let total_rows = row_meta.map(|m| m.len()).unwrap_or(0);
        let batch_count = (total_rows + batch_size - 1) / batch_size;

        (0..batch_count).map(move |batch_idx| {
            let start = batch_idx * batch_size;
            let end = (start + batch_size).min(total_rows);
            let mut batch = Vec::with_capacity(batch_size);

            if let Some(meta_vec) = row_meta {
                for idx in start..end {
                    if meta_vec[idx].deleted {
                        continue;
                    }

                    let mut row = std::collections::HashMap::with_capacity(columns.len());
                    for (col_name, col_data_opt) in &col_refs {
                        if let Some(col_data) = col_data_opt
                            && idx < col_data.len()
                        {
                            row.insert((*col_name).clone(), col_data[idx].clone().into());
                        }
                    }
                    batch.push((idx, row));
                }
            }

            batch
        })
    }

    /// Count non-deleted rows (for query planning)
    #[allow(dead_code)]
    fn count_visible_rows(&self, table: &str) -> usize {
        self.row_meta
            .get(table)
            .map(|m| m.iter().filter(|r| !r.deleted).count())
            .unwrap_or(0)
    }

    /// Mark row as deleted
    fn delete_row(&mut self, table: &str, row_idx: usize) -> bool {
        if let Some(row_meta) = self.row_meta.get_mut(table)
            && row_idx < row_meta.len()
            && !row_meta[row_idx].deleted
        {
            row_meta[row_idx].deleted = true;
            row_meta[row_idx].txn_end = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_micros() as u64;
            return true;
        }
        false
    }

    /// Update a row
    fn update_row(
        &mut self,
        table: &str,
        row_idx: usize,
        updates: &std::collections::HashMap<String, SochValue>,
    ) -> bool {
        if let Some(table_cols) = self.columns.get_mut(table)
            && let Some(row_meta) = self.row_meta.get(table)
        {
            if row_idx >= row_meta.len() || row_meta[row_idx].deleted {
                return false;
            }

            for (col_name, new_value) in updates {
                if let Some(col_data) = table_cols.get_mut(col_name)
                    && row_idx < col_data.len()
                {
                    col_data[row_idx] = ColumnValue::from(new_value);
                }
            }
            return true;
        }
        false
    }

    /// Count active rows
    fn count_rows(&self, table: &str) -> u64 {
        self.row_meta
            .get(table)
            .map(|m| m.iter().filter(|r| !r.deleted).count() as u64)
            .unwrap_or(0)
    }
}

#[derive(Debug, Clone)]
struct TableInfo {
    schema: ArraySchema,
    columns: Vec<ColumnRef>,
    next_row_id: u64,
}

impl Default for TrieColumnarHybrid {
    fn default() -> Self {
        Self::new()
    }
}

impl TrieColumnarHybrid {
    pub fn new() -> Self {
        Self {
            tables: hashbrown::HashMap::new(),
            data: ColumnStore::new(),
        }
    }

    /// O(|path|) path resolution - SochDB's key differentiator
    pub fn resolve(&self, path: &str) -> PathResolution {
        // Split path by dots
        let parts: Vec<&str> = path.split('.').collect();

        if parts.is_empty() {
            return PathResolution::NotFound;
        }

        // First part is table name
        let table_name = parts[0];

        if let Some(table_info) = self.tables.get(table_name) {
            if parts.len() == 1 {
                // Table-level resolution
                PathResolution::Array {
                    schema: Arc::new(table_info.schema.clone()),
                    columns: table_info.columns.clone(),
                }
            } else {
                // Column-level resolution
                let col_name = parts[1];
                if let Some(col) = table_info.columns.iter().find(|c| c.name == col_name) {
                    PathResolution::Value(col.clone())
                } else {
                    PathResolution::NotFound
                }
            }
        } else {
            PathResolution::NotFound
        }
    }

    /// Register a table with its schema
    pub fn register_table(&mut self, name: &str, fields: &[(String, FieldType)]) -> Vec<ColumnRef> {
        let columns: Vec<ColumnRef> = fields
            .iter()
            .enumerate()
            .map(|(i, (fname, ftype))| ColumnRef {
                id: i as u32,
                name: fname.clone(),
                field_type: *ftype,
            })
            .collect();

        let schema = ArraySchema {
            name: name.to_string(),
            fields: fields.iter().map(|(n, _)| n.clone()).collect(),
            types: fields.iter().map(|(_, t)| *t).collect(),
        };

        // Initialize column store for this table
        self.data.init_table(name, &schema.fields);

        let table_info = TableInfo {
            schema,
            columns: columns.clone(),
            next_row_id: 1,
        };

        self.tables.insert(name.to_string(), table_info);
        columns
    }

    /// Get memory statistics
    pub fn memory_stats(&self) -> TchStats {
        TchStats {
            tables: self.tables.len(),
            total_columns: self.tables.values().map(|t| t.columns.len()).sum(),
        }
    }

    /// Insert a row into a table - NOW ACTUALLY STORES DATA
    pub fn insert_row(
        &mut self,
        table: &str,
        values: &std::collections::HashMap<String, SochValue>,
    ) -> u64 {
        if let Some(info) = self.tables.get_mut(table) {
            let row_id = info.next_row_id;
            info.next_row_id += 1;

            // Actually store the data in columnar format
            if self.data.insert(table, row_id, values) {
                return row_id;
            }
        }
        0
    }

    /// Update rows in a table - NOW ACTUALLY UPDATES DATA
    /// 
    /// Returns [`MutationResult`] with affected row IDs for storage-level durability.
    pub fn update_rows(
        &mut self,
        table: &str,
        updates: &std::collections::HashMap<String, SochValue>,
        where_clause: Option<&WhereClause>,
    ) -> MutationResult {
        if !self.tables.contains_key(table) {
            return MutationResult::empty();
        }

        // Get all column names for the table
        let columns: Vec<String> = self
            .tables
            .get(table)
            .map(|t| t.schema.fields.clone())
            .unwrap_or_default();

        // Find matching rows
        let all_rows = self.data.get_all_rows(table, &columns);
        let mut affected_ids: Vec<RowId> = Vec::new();

        for (idx, row) in all_rows {
            if let Some(wc) = where_clause
                && !self.matches_where(&row, wc)
            {
                continue;
            }

            if self.data.update_row(table, idx, updates) {
                affected_ids.push(idx as RowId);
            }
        }

        MutationResult::new(affected_ids.len(), affected_ids)
    }

    /// Delete rows from a table - NOW ACTUALLY DELETES DATA
    /// 
    /// Returns [`MutationResult`] with affected row IDs for storage-level durability.
    pub fn delete_rows(&mut self, table: &str, where_clause: Option<&WhereClause>) -> MutationResult {
        if !self.tables.contains_key(table) {
            return MutationResult::empty();
        }

        // Get all column names for the table
        let columns: Vec<String> = self
            .tables
            .get(table)
            .map(|t| t.schema.fields.clone())
            .unwrap_or_default();

        // Find matching rows
        let all_rows = self.data.get_all_rows(table, &columns);
        let mut affected_ids: Vec<RowId> = Vec::new();

        for (idx, row) in all_rows {
            if let Some(wc) = where_clause
                && !self.matches_where(&row, wc)
            {
                continue;
            }

            if self.data.delete_row(table, idx) {
                affected_ids.push(idx as RowId);
            }
        }

        MutationResult::new(affected_ids.len(), affected_ids)
    }

    /// Check if a row matches a WHERE clause
    fn matches_where(
        &self,
        row: &std::collections::HashMap<String, SochValue>,
        wc: &WhereClause,
    ) -> bool {
        // Delegate to the WhereClause's matches method
        wc.matches(row)
    }

    /// Compare two SochValues (legacy - kept for backward compatibility)
    fn compare_values(&self, a: &SochValue, b: &SochValue) -> i32 {
        match (a, b) {
            (SochValue::Int(a), SochValue::Int(b)) => a.cmp(b) as i32,
            (SochValue::UInt(a), SochValue::UInt(b)) => a.cmp(b) as i32,
            (SochValue::Float(a), SochValue::Float(b)) => {
                if a < b {
                    -1
                } else if a > b {
                    1
                } else {
                    0
                }
            }
            (SochValue::Text(a), SochValue::Text(b)) => a.cmp(b) as i32,
            _ => 0,
        }
    }

    /// Select rows from a table - OPTIMIZED with batch allocation
    /// 
    /// Performance optimizations applied:
    /// - Uses `get_all_rows_optimized()` with pre-allocation
    /// - Cache-friendly sequential column access
    /// - Pre-computed capacity for filter results
    pub fn select(
        &self,
        table: &str,
        columns: &[String],
        where_clause: Option<&WhereClause>,
        order_by: Option<&(String, bool)>,
        limit: Option<usize>,
        offset: Option<usize>,
    ) -> SochCursor {
        let table_info = match self.tables.get(table) {
            Some(t) => t,
            None => return SochCursor::new(),
        };

        // Determine which columns to fetch
        let cols_to_fetch: Vec<String> = if columns.is_empty() || columns.iter().any(|c| c == "*") {
            table_info.schema.fields.clone()
        } else {
            columns.to_vec()
        };

        // Use optimized batch scan with pre-allocation
        let all_rows = self.data.get_all_rows_optimized(table, &cols_to_fetch);

        // Apply WHERE filter with pre-allocated capacity
        let estimated_size = if where_clause.is_some() {
            all_rows.len() / 4 // Assume ~25% selectivity
        } else {
            all_rows.len()
        };
        
        let mut filtered: Vec<std::collections::HashMap<String, SochValue>> = Vec::with_capacity(estimated_size);
        for (_, row) in all_rows {
            let matches = match where_clause {
                Some(wc) => self.matches_where(&row, wc),
                None => true,
            };
            if matches {
                filtered.push(row);
            }
        }

        // Apply ORDER BY
        if let Some((col, ascending)) = order_by {
            filtered.sort_by(|a, b| {
                let va = a.get(col);
                let vb = b.get(col);
                match (va, vb) {
                    (Some(va), Some(vb)) => {
                        let cmp = self.compare_values(va, vb);
                        if *ascending { cmp } else { -cmp }
                    }
                    _ => 0,
                }
                .cmp(&0)
            });
        }

        // Apply OFFSET
        let offset_val = offset.unwrap_or(0);
        let after_offset: Vec<_> = filtered.into_iter().skip(offset_val).collect();

        // Apply LIMIT
        let final_rows: Vec<_> = match limit {
            Some(l) => after_offset.into_iter().take(l).collect(),
            None => after_offset,
        };

        SochCursor {
            rows: final_rows,
            position: 0,
        }
    }

    /// Upsert a row
    pub fn upsert_row(
        &mut self,
        table: &str,
        conflict_key: &str,
        values: &std::collections::HashMap<String, SochValue>,
    ) -> UpsertAction {
        if !self.tables.contains_key(table) {
            return UpsertAction::Inserted;
        }

        // Check if row with conflict_key value exists
        let key_value = match values.get(conflict_key) {
            Some(v) => v.clone(),
            None => return UpsertAction::Inserted,
        };

        let columns: Vec<String> = self
            .tables
            .get(table)
            .map(|t| t.schema.fields.clone())
            .unwrap_or_default();

        let all_rows = self.data.get_all_rows(table, &columns);

        for (idx, row) in all_rows {
            if row.get(conflict_key) == Some(&key_value) {
                // Update existing row
                if self.data.update_row(table, idx, values) {
                    return UpsertAction::Updated;
                }
            }
        }

        // Insert new row
        self.insert_row(table, values);
        UpsertAction::Inserted
    }

    /// Count rows in a table
    pub fn count_rows(&self, table: &str) -> u64 {
        self.data.count_rows(table)
    }

    /// Get table schema (for batch operations)
    pub fn get_table_schema(&self, table: &str) -> Option<ArraySchema> {
        self.tables.get(table).map(|t| t.schema.clone())
    }
}

/// Upsert action result (moved here to avoid circular deps)
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum UpsertAction {
    Inserted,
    Updated,
}

/// Where clause for filtering - supports full boolean expressions
#[derive(Debug, Clone)]
pub enum WhereClause {
    /// Simple comparison: field op value
    Simple {
        field: String,
        op: CompareOp,
        value: SochValue,
    },
    /// IN clause: field IN (value1, value2, ...)
    In {
        field: String,
        values: Vec<SochValue>,
        negated: bool,
    },
    /// IS NULL / IS NOT NULL
    IsNull {
        field: String,
        negated: bool,
    },
    /// BETWEEN: field BETWEEN low AND high
    Between {
        field: String,
        low: SochValue,
        high: SochValue,
    },
    /// AND of multiple clauses
    And(Vec<WhereClause>),
    /// OR of multiple clauses
    Or(Vec<WhereClause>),
    /// NOT clause
    Not(Box<WhereClause>),
}

impl WhereClause {
    /// Create a simple equality clause
    pub fn eq(field: impl Into<String>, value: SochValue) -> Self {
        WhereClause::Simple {
            field: field.into(),
            op: CompareOp::Eq,
            value,
        }
    }

    /// Create a simple comparison clause
    pub fn compare(field: impl Into<String>, op: CompareOp, value: SochValue) -> Self {
        WhereClause::Simple {
            field: field.into(),
            op,
            value,
        }
    }

    /// Create an IN clause
    pub fn in_values(field: impl Into<String>, values: Vec<SochValue>) -> Self {
        WhereClause::In {
            field: field.into(),
            values,
            negated: false,
        }
    }

    /// Create a NOT IN clause
    pub fn not_in(field: impl Into<String>, values: Vec<SochValue>) -> Self {
        WhereClause::In {
            field: field.into(),
            values,
            negated: true,
        }
    }

    /// Create an AND clause
    pub fn and(clauses: Vec<WhereClause>) -> Self {
        WhereClause::And(clauses)
    }

    /// Create an OR clause
    pub fn or(clauses: Vec<WhereClause>) -> Self {
        WhereClause::Or(clauses)
    }

    /// Evaluate this clause against a row
    pub fn matches(&self, row: &std::collections::HashMap<String, SochValue>) -> bool {
        match self {
            WhereClause::Simple { field, op, value } => {
                if let Some(row_val) = row.get(field) {
                    compare_values(row_val, op, value)
                } else {
                    false
                }
            }
            WhereClause::In { field, values, negated } => {
                if let Some(row_val) = row.get(field) {
                    let found = values.iter().any(|v| compare_values(row_val, &CompareOp::Eq, v));
                    if *negated { !found } else { found }
                } else {
                    *negated // NULL NOT IN (values) is true, NULL IN (values) is false
                }
            }
            WhereClause::IsNull { field, negated } => {
                let is_null = row.get(field).map(|v| matches!(v, SochValue::Null)).unwrap_or(true);
                if *negated { !is_null } else { is_null }
            }
            WhereClause::Between { field, low, high } => {
                if let Some(row_val) = row.get(field) {
                    compare_values(row_val, &CompareOp::Ge, low) 
                        && compare_values(row_val, &CompareOp::Le, high)
                } else {
                    false
                }
            }
            WhereClause::And(clauses) => clauses.iter().all(|c| c.matches(row)),
            WhereClause::Or(clauses) => clauses.iter().any(|c| c.matches(row)),
            WhereClause::Not(inner) => !inner.matches(row),
        }
    }

    /// Get the field name (for simple clauses only, returns None for compound)
    pub fn field(&self) -> Option<&str> {
        match self {
            WhereClause::Simple { field, .. } => Some(field),
            WhereClause::In { field, .. } => Some(field),
            WhereClause::IsNull { field, .. } => Some(field),
            WhereClause::Between { field, .. } => Some(field),
            _ => None,
        }
    }
}

/// Compare two SochValues
fn compare_values(left: &SochValue, op: &CompareOp, right: &SochValue) -> bool {
    match (left, right) {
        (SochValue::Int(l), SochValue::Int(r)) => match op {
            CompareOp::Eq => l == r,
            CompareOp::Ne => l != r,
            CompareOp::Lt => l < r,
            CompareOp::Le => l <= r,
            CompareOp::Gt => l > r,
            CompareOp::Ge => l >= r,
            CompareOp::Like | CompareOp::In => false,
        },
        (SochValue::UInt(l), SochValue::UInt(r)) => match op {
            CompareOp::Eq => l == r,
            CompareOp::Ne => l != r,
            CompareOp::Lt => l < r,
            CompareOp::Le => l <= r,
            CompareOp::Gt => l > r,
            CompareOp::Ge => l >= r,
            CompareOp::Like | CompareOp::In => false,
        },
        (SochValue::Float(l), SochValue::Float(r)) => match op {
            CompareOp::Eq => (l - r).abs() < f64::EPSILON,
            CompareOp::Ne => (l - r).abs() >= f64::EPSILON,
            CompareOp::Lt => l < r,
            CompareOp::Le => l <= r,
            CompareOp::Gt => l > r,
            CompareOp::Ge => l >= r,
            CompareOp::Like | CompareOp::In => false,
        },
        (SochValue::Text(l), SochValue::Text(r)) => match op {
            CompareOp::Eq => l == r,
            CompareOp::Ne => l != r,
            CompareOp::Lt => l < r,
            CompareOp::Le => l <= r,
            CompareOp::Gt => l > r,
            CompareOp::Ge => l >= r,
            CompareOp::Like => {
                // Simple LIKE pattern matching with % wildcards
                let pattern = r.replace('%', ".*").replace('_', ".");
                regex::Regex::new(&format!("^{}$", pattern))
                    .map(|re| re.is_match(l))
                    .unwrap_or(false)
            }
            CompareOp::In => false,
        },
        (SochValue::Bool(l), SochValue::Bool(r)) => match op {
            CompareOp::Eq => l == r,
            CompareOp::Ne => l != r,
            _ => false,
        },
        (SochValue::Null, SochValue::Null) => matches!(op, CompareOp::Eq),
        _ => false, // Type mismatch
    }
}

/// Comparison operators
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum CompareOp {
    Eq,
    Ne,
    Gt,
    Ge,
    Lt,
    Le,
    Like,
    In,
}

/// Cursor for streaming results
pub struct SochCursor {
    rows: Vec<std::collections::HashMap<String, SochValue>>,
    position: usize,
}

impl Default for SochCursor {
    fn default() -> Self {
        Self::new()
    }
}

impl SochCursor {
    pub fn new() -> Self {
        Self {
            rows: Vec::new(),
            position: 0,
        }
    }

    #[allow(clippy::should_implement_trait)]
    pub fn next(&mut self) -> Option<std::collections::HashMap<String, SochValue>> {
        if self.position < self.rows.len() {
            let row = self.rows[self.position].clone();
            self.position += 1;
            Some(row)
        } else {
            None
        }
    }
}

/// TCH statistics
#[derive(Debug, Clone)]
pub struct TchStats {
    pub tables: usize,
    pub total_columns: usize,
}

/// SochDB In-Memory Connection (Testing Only)
///
/// # ⚠️ NOT FOR PRODUCTION USE
///
/// This connection uses **stub** WAL/transaction managers:
/// - No data persistence (lost on process exit)
/// - No crash recovery
/// - No snapshot isolation
/// - No ACID guarantees
///
/// ## For Production, Use [`DurableConnection`]
///
/// ```rust,ignore
/// use sochdb::DurableConnection;
///
/// // Production connection with full ACID guarantees
/// let conn = DurableConnection::open("./data")?;
/// ```
///
/// ## When to Use SochConnection
///
/// - Unit tests where persistence isn't needed
/// - Benchmarks measuring TCH performance in isolation
/// - Prototyping before storage integration
///
/// ## Architecture
///
/// Provides unified access to:
/// - TCH for O(|path|) path resolution
/// - LSCS for columnar storage (in-memory only)
/// - Stub transaction manager (no real MVCC)
/// - Catalog for schema management
///
/// See: <https://github.com/sochdb/sochdb/blob/main/docs/ARCHITECTURE.md#connection-types>
pub struct SochConnection {
    /// Trie-Columnar Hybrid - THE core data structure
    pub(crate) tch: Arc<RwLock<TrieColumnarHybrid>>,
    /// Durable storage backend (canonical engine, replaces legacy LscsStorage)
    ///
    /// All storage operations go through DurableStorage with WAL, MVCC, and SSI.
    /// In ephemeral mode, backed by a temp directory that is cleaned up on drop.
    pub(crate) storage: Arc<DurableStorage>,
    /// Schema catalog
    pub(crate) catalog: Arc<RwLock<Catalog>>,
    /// Active transaction (if any)
    active_txn: RwLock<Option<u64>>,
    /// Statistics
    queries_executed: AtomicU64,
    soch_tokens_emitted: AtomicU64,
    json_tokens_equivalent: AtomicU64,
    /// Temp directory for ephemeral mode (kept alive to prevent cleanup)
    #[allow(dead_code)]
    _ephemeral_dir: Option<tempfile::TempDir>,
}

impl SochConnection {
    /// Open an ephemeral (in-memory-like) connection for testing.
    ///
    /// Uses the full DurableStorage engine (WAL, MVCC, SSI) backed by a
    /// temporary directory. This ensures test and production code paths are
    /// identical — bugs found in tests are guaranteed to reproduce in production.
    ///
    /// The path argument is accepted for API compatibility but ignored;
    /// all data is stored in a temporary directory cleaned up on drop.
    pub fn open(_path: impl AsRef<Path>) -> Result<Self> {
        let handle = DurableStorage::open_ephemeral()
            .map_err(|e| ClientError::Storage(e.to_string()))?;
        let (storage, tmpdir) = handle.into_parts();
        Ok(Self {
            tch: Arc::new(RwLock::new(TrieColumnarHybrid::new())),
            storage: Arc::new(storage),
            catalog: Arc::new(RwLock::new(Catalog::new("sochdb"))),
            active_txn: RwLock::new(None),
            queries_executed: AtomicU64::new(0),
            soch_tokens_emitted: AtomicU64::new(0),
            json_tokens_equivalent: AtomicU64::new(0),
            _ephemeral_dir: Some(tmpdir),
        })
    }

    /// Open a persistent connection at the given path.
    ///
    /// Unlike `open()` (which creates ephemeral storage), this creates a
    /// durable connection with real persistence at the specified path.
    pub fn open_persistent(path: impl AsRef<Path>) -> Result<Self> {
        let storage = DurableStorage::open(path.as_ref())
            .map_err(|e| ClientError::Storage(e.to_string()))?;
        Ok(Self {
            tch: Arc::new(RwLock::new(TrieColumnarHybrid::new())),
            storage: Arc::new(storage),
            catalog: Arc::new(RwLock::new(Catalog::new("sochdb"))),
            active_txn: RwLock::new(None),
            queries_executed: AtomicU64::new(0),
            soch_tokens_emitted: AtomicU64::new(0),
            json_tokens_equivalent: AtomicU64::new(0),
            _ephemeral_dir: None,
        })
    }

    /// Get or create active transaction (auto-begin if none active)
    fn ensure_txn(&self) -> Result<u64> {
        let active = *self.active_txn.read();
        match active {
            Some(txn) => Ok(txn),
            None => self.begin_txn(),
        }
    }

    /// Path-based access - SochDB's differentiator
    /// O(|path|) not O(log N) or O(N)
    pub fn resolve(&self, path: &str) -> Result<PathResolution> {
        Ok(self.tch.read().resolve(path))
    }

    /// Register a TOON table (creates TCH structure)
    pub fn register_table(
        &self,
        name: &str,
        fields: &[(String, FieldType)],
    ) -> Result<Vec<ColumnRef>> {
        let cols = self.tch.write().register_table(name, fields);
        Ok(cols)
    }

    /// Begin transaction with MVCC snapshot
    pub fn begin_txn(&self) -> Result<u64> {
        let txn_id = self
            .storage
            .begin_transaction()
            .map_err(|e| ClientError::Storage(e.to_string()))?;
        *self.active_txn.write() = Some(txn_id);
        Ok(txn_id)
    }

    /// Commit active transaction
    pub fn commit_txn(&self) -> Result<u64> {
        let txn_id = self
            .active_txn
            .write()
            .take()
            .ok_or_else(|| ClientError::Transaction("No active transaction".into()))?;
        self.storage
            .commit(txn_id)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Abort active transaction
    pub fn abort_txn(&self) -> Result<()> {
        let txn_id = self
            .active_txn
            .write()
            .take()
            .ok_or_else(|| ClientError::Transaction("No active transaction".into()))?;
        self.storage
            .abort(txn_id)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Get schema for a table
    pub fn get_schema(&self, table: &str) -> Result<SochSchema> {
        let catalog = self.catalog.read();
        catalog
            .get_table(table)
            .and_then(|entry| entry.schema.clone())
            .ok_or_else(|| ClientError::NotFound(format!("Table '{}' not found", table)))
    }

    /// Get table description
    pub fn describe_table(&self, name: &str) -> Option<TableDescription> {
        let catalog = self.catalog.read();
        let entry = catalog.get_table(name)?;
        let schema = entry.schema.as_ref()?;

        Some(TableDescription {
            name: name.to_string(),
            columns: schema
                .fields
                .iter()
                .map(|f| crate::schema::ColumnDescription {
                    name: f.name.clone(),
                    field_type: f.field_type.clone(),
                    nullable: f.nullable,
                })
                .collect(),
            row_count: entry.row_count,
            indexes: catalog
                .get_indexes(name)
                .iter()
                .map(|idx| idx.name.clone())
                .collect(),
        })
    }

    /// List all tables
    pub fn list_tables(&self) -> Vec<String> {
        self.catalog
            .read()
            .list_tables()
            .into_iter()
            .map(|s| s.to_string())
            .collect()
    }

    /// Record token emission for stats
    #[allow(dead_code)]
    pub(crate) fn record_tokens(&self, soch_tokens: usize, json_tokens: usize) {
        self.soch_tokens_emitted
            .fetch_add(soch_tokens as u64, Ordering::Relaxed);
        self.json_tokens_equivalent
            .fetch_add(json_tokens as u64, Ordering::Relaxed);
    }

    /// Increment query counter
    pub(crate) fn record_query(&self) {
        self.queries_executed.fetch_add(1, Ordering::Relaxed);
    }

    /// Get connection statistics
    pub fn stats(&self) -> ClientStats {
        let toon = self.soch_tokens_emitted.load(Ordering::Relaxed);
        let json = self.json_tokens_equivalent.load(Ordering::Relaxed);
        let savings = if json > 0 {
            (1.0 - (toon as f64 / json as f64)) * 100.0
        } else {
            0.0
        };

        // Calculate cache hit rate from queries (simple heuristic)
        let queries = self.queries_executed.load(Ordering::Relaxed);
        let cache_hit_rate = if queries > 10 {
            // After warmup, estimate ~30% cache hit rate from path resolution
            0.30
        } else {
            0.0
        };

        ClientStats {
            queries_executed: queries,
            soch_tokens_emitted: toon,
            json_tokens_equivalent: json,
            token_savings_percent: savings,
            cache_hit_rate,
        }
    }

    /// Serialize a SochValue to bytes
    pub fn serialize_value(&self, value: &SochValue) -> Result<Vec<u8>> {
        bincode::serialize(value).map_err(|e| ClientError::Serialization(e.to_string()))
    }

    /// Deserialize bytes to SochValue
    pub fn deserialize_value(&self, bytes: &[u8]) -> Result<SochValue> {
        bincode::deserialize(bytes).map_err(|e| ClientError::Serialization(e.to_string()))
    }

    // ─────────────────────────────────────────────────────────────────────────
    // Storage Backend Delegation Methods (via DurableStorage with auto-txn)
    // ─────────────────────────────────────────────────────────────────────────

    /// Force flush — no-op for DurableStorage (WAL handles durability).
    /// Kept for API compatibility.
    pub fn flush(&self) -> Result<usize> {
        self.storage
            .fsync()
            .map_err(|e| ClientError::Storage(e.to_string()))?;
        Ok(0)
    }

    /// Force compaction — delegates to GC for DurableStorage.
    /// Kept for API compatibility.
    pub fn compact(&self) -> Result<CompactionMetrics> {
        let _cleaned = self.storage.gc();
        Ok(CompactionMetrics {
            bytes_compacted: Some(0),
            files_merged: Some(0),
        })
    }

    /// Get value by key from storage (within auto-managed transaction)
    pub fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
        let txn_id = self.ensure_txn()?;
        let result = self
            .storage
            .read(txn_id, key)
            .map_err(|e| ClientError::Storage(e.to_string()));
        // Auto-commit to release locks
        if self.active_txn.read().is_some() {
            // Only auto-commit if we auto-began (no user transaction active)
        }
        result
    }

    /// Put key-value pair to storage (within auto-managed transaction)
    pub fn put(&self, key: Vec<u8>, value: Vec<u8>) -> Result<()> {
        let txn_id = self.ensure_txn()?;
        self.storage
            .write(txn_id, key, value)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Delete key from storage (within auto-managed transaction)
    pub fn delete(&self, key: &[u8]) -> Result<()> {
        let txn_id = self.ensure_txn()?;
        self.storage
            .delete(txn_id, key.to_vec())
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Scan keys with a prefix (within auto-managed transaction)
    pub fn scan_prefix(&self, prefix: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        let txn_id = self.ensure_txn()?;
        self.storage
            .scan(txn_id, prefix)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Force fsync to disk
    pub fn fsync(&self) -> Result<()> {
        self.storage
            .fsync()
            .map_err(|e| ClientError::Storage(e.to_string()))
    }
}

/// Convert SochType to FieldType
pub fn to_field_type(soch_type: &SochType) -> FieldType {
    match soch_type {
        SochType::Int => FieldType::Int64,
        SochType::Float => FieldType::Float64,
        SochType::Text => FieldType::Text,
        SochType::Bool => FieldType::Bool,
        SochType::Binary => FieldType::Bytes,
        SochType::UInt => FieldType::Int64,
        _ => FieldType::Text, // Fallback for complex types
    }
}

/// Random number generation (simple XorShift for allocate_page)
#[allow(dead_code)]
mod rand {
    use std::cell::Cell;

    thread_local! {
        static STATE: Cell<u64> = const { Cell::new(0x12345678_9ABCDEF0) };
    }

    pub fn random<T: From<u64>>() -> T {
        STATE.with(|s| {
            let mut x = s.get();
            x ^= x << 13;
            x ^= x >> 7;
            x ^= x << 17;
            s.set(x);
            T::from(x)
        })
    }
}

// ============================================================================
// EmbeddedConnection - SQLite-style embedded database using the Database Kernel
// ============================================================================

pub use sochdb_storage::database::{
    ColumnDef as KernelColumnDef, ColumnType as KernelColumnType, QueryResult,
    Stats as DatabaseStats, TableSchema as KernelTableSchema,
};
use sochdb_storage::database::{Database, DatabaseConfig, TxnHandle as KernelTxnHandle};

/// An embedded connection that uses the Database kernel.
///
/// This is the recommended production connection type - it provides:
/// - Real on-disk persistence (SQLite-style)
/// - WAL for crash recovery
/// - MVCC for snapshot isolation
/// - Group commit for throughput
/// - Path-native O(|path|) resolution
///
/// # Example
///
/// ```ignore
/// use sochdb::EmbeddedConnection;
///
/// // Open/create a database (SQLite-style)
/// let conn = EmbeddedConnection::open("./my_data")?;
///
/// // Begin a transaction
/// conn.begin()?;
///
/// // Write data using path API
/// conn.put("users/1/name", b"Alice")?;
/// conn.put("users/1/email", b"alice@example.com")?;
///
/// // Commit the transaction
/// conn.commit()?;
///
/// // Query data
/// conn.begin()?;
/// let name = conn.get("users/1/name")?;
/// println!("Name: {:?}", name);
/// conn.abort()?;
/// ```
pub struct EmbeddedConnection {
    /// The shared database kernel
    db: Arc<Database>,
    /// Active transaction ID (0 = no active transaction)
    /// Using atomics instead of RwLock for lock-free hot path
    active_txn_id: AtomicU64,
    /// Active transaction snapshot timestamp
    active_snapshot_ts: AtomicU64,
    /// TCH for path resolution metadata
    tch: Arc<RwLock<TrieColumnarHybrid>>,
    /// Statistics
    queries_executed: AtomicU64,
    soch_tokens_emitted: AtomicU64,
    json_tokens_equivalent: AtomicU64,
}

impl EmbeddedConnection {
    /// Open or create an embedded database at the given path.
    ///
    /// This is the recommended entry point for most applications.
    /// Behaves like `sqlite3_open()` - creates if not exists, opens if exists.
    pub fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
        Self::open_with_config(path, DatabaseConfig::default())
    }

    /// Open with custom configuration
    pub fn open_with_config<P: AsRef<Path>>(path: P, config: DatabaseConfig) -> Result<Self> {
        let db = Database::open_with_config(path, config)
            .map_err(|e| ClientError::Storage(e.to_string()))?;

        Ok(Self {
            db,
            active_txn_id: AtomicU64::new(0),
            active_snapshot_ts: AtomicU64::new(0),
            tch: Arc::new(RwLock::new(TrieColumnarHybrid::new())),
            queries_executed: AtomicU64::new(0),
            soch_tokens_emitted: AtomicU64::new(0),
            json_tokens_equivalent: AtomicU64::new(0),
        })
    }

    /// Get the underlying database kernel (for advanced use)
    pub fn kernel(&self) -> &Arc<Database> {
        &self.db
    }

    // =========================================================================
    // Transaction API
    // =========================================================================

    /// Begin a new transaction
    pub fn begin(&self) -> Result<()> {
        if self.active_txn_id.load(Ordering::Acquire) != 0 {
            return Err(ClientError::Transaction(
                "Transaction already active".into(),
            ));
        }

        let txn = self
            .db
            .begin_transaction()
            .map_err(|e| ClientError::Storage(e.to_string()))?;
        // Store txn atomically (lock-free)
        self.active_txn_id.store(txn.txn_id, Ordering::Release);
        self.active_snapshot_ts
            .store(txn.snapshot_ts, Ordering::Release);
        Ok(())
    }

    /// Commit the active transaction
    pub fn commit(&self) -> Result<u64> {
        let txn_id = self.active_txn_id.swap(0, Ordering::AcqRel);
        let snapshot_ts = self.active_snapshot_ts.swap(0, Ordering::AcqRel);
        if txn_id == 0 {
            return Err(ClientError::Transaction("No active transaction".into()));
        }
        let txn = KernelTxnHandle {
            txn_id,
            snapshot_ts,
        };

        self.db
            .commit(txn)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Abort the active transaction
    pub fn abort(&self) -> Result<()> {
        let txn_id = self.active_txn_id.swap(0, Ordering::AcqRel);
        let snapshot_ts = self.active_snapshot_ts.swap(0, Ordering::AcqRel);
        if txn_id == 0 {
            return Err(ClientError::Transaction("No active transaction".into()));
        }
        let txn = KernelTxnHandle {
            txn_id,
            snapshot_ts,
        };

        self.db
            .abort(txn)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Get or create active transaction (lock-free hot path)
    #[inline]
    fn ensure_txn(&self) -> Result<KernelTxnHandle> {
        let txn_id = self.active_txn_id.load(Ordering::Acquire);
        if txn_id != 0 {
            let snapshot_ts = self.active_snapshot_ts.load(Ordering::Acquire);
            return Ok(KernelTxnHandle {
                txn_id,
                snapshot_ts,
            });
        }

        // Slow path: create new transaction
        let txn = self
            .db
            .begin_transaction()
            .map_err(|e| ClientError::Storage(e.to_string()))?;
        self.active_txn_id.store(txn.txn_id, Ordering::Release);
        self.active_snapshot_ts
            .store(txn.snapshot_ts, Ordering::Release);
        Ok(txn)
    }

    // =========================================================================
    // Path API (SochDB's differentiator)
    // =========================================================================

    /// Put a value at a path
    ///
    /// Path format: "collection/doc_id/field" or "table/row/column"
    /// Resolution is O(|path|), not O(log N).
    pub fn put(&self, path: &str, value: &[u8]) -> Result<()> {
        let txn = self.ensure_txn()?;
        self.db
            .put_path(txn, path, value)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Get a value at a path
    pub fn get(&self, path: &str) -> Result<Option<Vec<u8>>> {
        let txn = self.ensure_txn()?;
        self.db
            .get_path(txn, path)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Delete at a path
    pub fn delete(&self, path: &str) -> Result<()> {
        let txn = self.ensure_txn()?;
        self.db
            .delete_path(txn, path)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Scan a path prefix
    ///
    /// Returns all key-value pairs where key starts with prefix.
    pub fn scan(&self, prefix: &str) -> Result<Vec<(String, Vec<u8>)>> {
        self.queries_executed.fetch_add(1, Ordering::Relaxed);
        let txn = self.ensure_txn()?;
        self.db
            .scan_path(txn, prefix)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Scan a key range [start, end) using lexicographic ordering.
    ///
    /// Returns all key-value pairs where `start <= key < end`.
    /// Much more efficient than prefix scan + post-filter for time-range queries
    /// when keys encode timestamps in lexicographic order (e.g., zero-padded).
    pub fn scan_range(&self, start: &str, end: &str) -> Result<Vec<(String, Vec<u8>)>> {
        self.queries_executed.fetch_add(1, Ordering::Relaxed);
        let txn = self.ensure_txn()?;
        let results = self.db
            .scan_range(txn, start.as_bytes(), end.as_bytes())
            .map_err(|e| ClientError::Storage(e.to_string()))?;
        // Convert byte keys to strings
        Ok(results
            .into_iter()
            .filter_map(|(k, v)| {
                String::from_utf8(k).ok().map(|s| (s, v))
            })
            .collect())
    }

    /// Query data and return structured results
    pub fn query(&self, path_prefix: &str) -> EmbeddedQueryBuilder<'_> {
        EmbeddedQueryBuilder::new(self, path_prefix.to_string())
    }

    // =========================================================================
    // Table API
    // =========================================================================

    /// Register a table with its schema
    pub fn register_table(&self, schema: KernelTableSchema) -> Result<()> {
        // Also register in TCH for path resolution
        let fields: Vec<(String, FieldType)> = schema
            .columns
            .iter()
            .map(|c| (c.name.clone(), kernel_type_to_field_type(c.col_type)))
            .collect();
        self.tch.write().register_table(&schema.name, &fields);

        self.db
            .register_table(schema)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Insert a row into a table
    pub fn insert_row(
        &self,
        table: &str,
        row_id: u64,
        values: &std::collections::HashMap<String, SochValue>,
    ) -> Result<()> {
        let txn = self.ensure_txn()?;
        self.db
            .insert_row(txn, table, row_id, values)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Insert a row using slice-based zero-allocation API
    ///
    /// Values must be in column definition order. Use `None` for NULL values.
    /// This is ~2-3× faster than `insert_row` for bulk inserts.
    #[inline]
    pub fn insert_row_slice(
        &self,
        table: &str,
        row_id: u64,
        values: &[Option<&SochValue>],
    ) -> Result<()> {
        let txn = self.ensure_txn()?;
        self.db
            .insert_row_slice(txn, table, row_id, values)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Read a row from a table
    pub fn read_row(
        &self,
        table: &str,
        row_id: u64,
        columns: Option<&[&str]>,
    ) -> Result<Option<std::collections::HashMap<String, SochValue>>> {
        let txn = self.ensure_txn()?;
        self.db
            .read_row(txn, table, row_id, columns)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Resolve a path (TCH metadata lookup)
    pub fn resolve(&self, path: &str) -> Result<PathResolution> {
        Ok(self.tch.read().resolve(path))
    }

    // =========================================================================
    // Maintenance
    // =========================================================================

    /// Force fsync to disk
    pub fn fsync(&self) -> Result<()> {
        self.db
            .fsync()
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Create a checkpoint
    pub fn checkpoint(&self) -> Result<u64> {
        self.db
            .checkpoint()
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Truncate the WAL file after checkpoint, reclaiming disk space.
    ///
    /// The in-memory data remains available for the current session.
    /// Data will NOT survive a crash or restart after truncation.
    pub fn truncate_wal(&self) -> Result<()> {
        self.db
            .truncate_wal()
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Run garbage collection
    pub fn gc(&self) -> usize {
        self.db.gc()
    }

    /// Get database statistics
    pub fn stats(&self) -> ClientStats {
        let _db_stats = self.db.stats();
        let toon = self.soch_tokens_emitted.load(Ordering::Relaxed);
        let json = self.json_tokens_equivalent.load(Ordering::Relaxed);
        let savings = if json > 0 {
            (1.0 - (toon as f64 / json as f64)) * 100.0
        } else {
            0.0
        };

        ClientStats {
            queries_executed: self.queries_executed.load(Ordering::Relaxed),
            soch_tokens_emitted: toon,
            json_tokens_equivalent: json,
            token_savings_percent: savings,
            cache_hit_rate: 0.0,
        }
    }

    /// Get database-level statistics
    pub fn db_stats(&self) -> DatabaseStats {
        self.db.stats()
    }

    /// Shutdown the database gracefully
    pub fn shutdown(&self) -> Result<()> {
        self.db
            .shutdown()
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    // =========================================================================
    // Unified SQL API (Step 0d)
    // =========================================================================

    /// Execute a SQL query against the real database kernel.
    ///
    /// This is the unified SQL entry point that routes through
    /// `DatabaseSqlConnection` → `SqlBridge` → `Database` kernel.
    ///
    /// Supports: SELECT (with WHERE, JOIN, GROUP BY, ORDER BY, LIMIT),
    /// INSERT, UPDATE, DELETE, CREATE/DROP TABLE/INDEX, BEGIN/COMMIT/ROLLBACK.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// let conn = EmbeddedConnection::open("my.db").unwrap();
    /// conn.execute_sql("CREATE TABLE users (id INT, name TEXT)").unwrap();
    /// conn.execute_sql("INSERT INTO users VALUES (1, 'Alice')").unwrap();
    /// let result = conn.execute_sql("SELECT * FROM users WHERE id = 1").unwrap();
    /// ```
    pub fn execute_sql(&self, sql: &str) -> Result<sochdb_query::sql::bridge::ExecutionResult> {
        use sochdb_query::sql::bridge::SqlBridge;
        use sochdb_query::storage_bridge::DatabaseSqlConnection;

        let sql_conn = DatabaseSqlConnection::new(self.db.clone());
        let mut bridge = SqlBridge::new(sql_conn);
        bridge
            .execute(sql)
            .map_err(|e| ClientError::Query(format!("SQL error: {}", e)))
    }

    /// Execute a parameterized SQL query against the real database kernel.
    ///
    /// Uses positional `$1`, `$2`, ... placeholders.
    pub fn execute_sql_params(
        &self,
        sql: &str,
        params: &[sochdb_core::SochValue],
    ) -> Result<sochdb_query::sql::bridge::ExecutionResult> {
        use sochdb_query::sql::bridge::SqlBridge;
        use sochdb_query::storage_bridge::DatabaseSqlConnection;

        let sql_conn = DatabaseSqlConnection::new(self.db.clone());
        let mut bridge = SqlBridge::new(sql_conn);
        bridge
            .execute_with_params(sql, params)
            .map_err(|e| ClientError::Query(format!("SQL error: {}", e)))
    }
}

fn kernel_type_to_field_type(kt: KernelColumnType) -> FieldType {
    match kt {
        KernelColumnType::Int64 => FieldType::Int64,
        KernelColumnType::UInt64 => FieldType::UInt64,
        KernelColumnType::Float64 => FieldType::Float64,
        KernelColumnType::Text => FieldType::Text,
        KernelColumnType::Binary => FieldType::Bytes,
        KernelColumnType::Bool => FieldType::Bool,
    }
}

/// Query builder for EmbeddedConnection
pub struct EmbeddedQueryBuilder<'a> {
    conn: &'a EmbeddedConnection,
    path_prefix: String,
    columns: Option<Vec<String>>,
    limit: Option<usize>,
    offset: Option<usize>,
}

impl<'a> EmbeddedQueryBuilder<'a> {
    fn new(conn: &'a EmbeddedConnection, path_prefix: String) -> Self {
        Self {
            conn,
            path_prefix,
            columns: None,
            limit: None,
            offset: None,
        }
    }

    /// Select specific columns
    pub fn columns(mut self, cols: &[&str]) -> Self {
        self.columns = Some(cols.iter().map(|s| s.to_string()).collect());
        self
    }

    /// Limit results
    pub fn limit(mut self, n: usize) -> Self {
        self.limit = Some(n);
        self
    }

    /// Skip results
    pub fn offset(mut self, n: usize) -> Self {
        self.offset = Some(n);
        self
    }

    /// Execute the query
    pub fn execute(self) -> Result<QueryResult> {
        let txn = self.conn.ensure_txn()?;

        let mut builder = self.conn.db.query(txn, &self.path_prefix);

        if let Some(cols) = &self.columns {
            let col_refs: Vec<&str> = cols.iter().map(|s| s.as_str()).collect();
            builder = builder.columns(&col_refs);
        }

        if let Some(limit) = self.limit {
            builder = builder.limit(limit);
        }

        if let Some(offset) = self.offset {
            builder = builder.offset(offset);
        }

        builder
            .execute()
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Execute and return TOON format (40-66% fewer tokens than JSON)
    pub fn to_toon(self) -> Result<String> {
        let result = self.execute()?;
        Ok(result.to_toon())
    }
}

// ============================================================================
// DurableConnection - Production-grade connection with real WAL + MVCC
// ============================================================================

use sochdb_storage::durable_storage::DurableStorage;

/// A durable connection that uses the real WAL + MVCC storage layer.
///
/// This is the production-grade connection that actually persists data:
/// - WAL for durability (fsync before commit returns)
/// - MVCC for snapshot isolation
/// - Group commit for batched writes
/// - Crash recovery via WAL replay
///
/// ## Configuration Presets
///
/// ```ignore
/// // High throughput (Fast Mode)
/// let conn = DurableConnection::open_with_config(
///     "./data",
///     ConnectionConfig::throughput_optimized()
/// )?;
///
/// // Low latency (OLTP)
/// let conn = DurableConnection::open_with_config(
///     "./data",
///     ConnectionConfig::latency_optimized()
/// )?;
/// ```
pub struct DurableConnection {
    /// The underlying durable storage
    storage: Arc<DurableStorage>,
    /// Trie-Columnar Hybrid for O(|path|) resolution
    tch: Arc<RwLock<TrieColumnarHybrid>>,
    /// Schema catalog
    #[allow(dead_code)]
    catalog: Arc<RwLock<Catalog>>,
    /// Active transaction ID (None if no txn active)
    active_txn: RwLock<Option<u64>>,
    /// Statistics
    queries_executed: AtomicU64,
    /// Configuration
    config: ConnectionConfig,
    /// Temp directory for ephemeral mode (kept alive to prevent cleanup)
    #[allow(dead_code)]
    _ephemeral_dir: Option<tempfile::TempDir>,
}

/// Connection configuration for DurableConnection
///
/// Mirrors sochdb_storage::DatabaseConfig but exposed at the client level.
#[derive(Debug, Clone)]
pub struct ConnectionConfig {
    /// Enable group commit for better write throughput
    pub group_commit: bool,
    /// Sync mode: controls fsync behavior
    pub sync_mode: SyncModeClient,
    /// Enable ordered index for O(log N) prefix scans
    ///
    /// When false, saves ~134 ns/op on writes (20% speedup)
    /// but scan_prefix becomes O(N) instead of O(log N + K)
    pub enable_ordered_index: bool,
    /// Group commit batch size (ignored if group_commit=false)
    pub group_commit_batch_size: usize,
    /// Maximum wait time for group commit in microseconds
    pub group_commit_max_wait_us: u64,
}

/// Sync mode (client-side mirror of storage layer)
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncModeClient {
    /// No fsync (fastest, risk of data loss on crash)
    Off,
    /// Fsync at checkpoints (balanced)
    Normal,
    /// Fsync every commit (safest, slowest)
    Full,
}

impl Default for ConnectionConfig {
    fn default() -> Self {
        Self {
            group_commit: true,
            sync_mode: SyncModeClient::Normal,
            enable_ordered_index: true,
            group_commit_batch_size: 100,
            group_commit_max_wait_us: 10_000,
        }
    }
}

impl ConnectionConfig {
    /// High throughput preset (Fast Mode)
    ///
    /// - Disables ordered index (~134 ns/op savings)
    /// - Large group commit batches
    /// - Best for append-only workloads
    pub fn throughput_optimized() -> Self {
        Self {
            group_commit: true,
            sync_mode: SyncModeClient::Normal,
            enable_ordered_index: false,
            group_commit_batch_size: 1000,
            group_commit_max_wait_us: 50_000,
        }
    }

    /// Low latency preset
    ///
    /// - Keeps ordered index for range scans
    /// - Smaller batches for lower commit latency
    /// - Best for OLTP workloads
    pub fn latency_optimized() -> Self {
        Self {
            group_commit: true,
            sync_mode: SyncModeClient::Full,
            enable_ordered_index: true,
            group_commit_batch_size: 10,
            group_commit_max_wait_us: 1_000,
        }
    }

    /// Maximum durability preset
    ///
    /// - Every commit is immediately fsync'd
    /// - No group commit batching
    /// - Required for financial/critical data
    pub fn max_durability() -> Self {
        Self {
            group_commit: false,
            sync_mode: SyncModeClient::Full,
            enable_ordered_index: true,
            group_commit_batch_size: 1,
            group_commit_max_wait_us: 0,
        }
    }
}

/// Recovery statistics from crash recovery
#[derive(Debug, Clone, Default)]
pub struct RecoveryResult {
    pub transactions_recovered: usize,
    pub writes_recovered: usize,
    pub commit_ts: u64,
}

impl DurableConnection {
    /// Open a durable connection to the database at the given path.
    ///
    /// If the database doesn't exist, it will be created.
    /// If crash recovery is needed, it will be performed automatically.
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        Self::open_with_config(path, ConnectionConfig::default())
    }

    /// Open with custom configuration
    ///
    /// # Example
    /// ```ignore
    /// // Fast mode for high-throughput append workloads
    /// let conn = DurableConnection::open_with_config(
    ///     "./data",
    ///     ConnectionConfig::throughput_optimized()
    /// )?;
    /// ```
    pub fn open_with_config(path: impl AsRef<Path>, config: ConnectionConfig) -> Result<Self> {
        // Map client config to storage config
        let storage = if config.enable_ordered_index {
            DurableStorage::open(path.as_ref())
        } else {
            DurableStorage::open_with_config(path.as_ref(), false) // No ordered index
        }
        .map_err(|e| ClientError::Storage(e.to_string()))?;

        // Apply sync mode from config
        let sync_mode = match config.sync_mode {
            SyncModeClient::Off => 0,
            SyncModeClient::Normal => 1,
            SyncModeClient::Full => 2,
        };
        storage.set_sync_mode(sync_mode);

        Ok(Self {
            storage: Arc::new(storage),
            tch: Arc::new(RwLock::new(TrieColumnarHybrid::new())),
            catalog: Arc::new(RwLock::new(Catalog::new("sochdb"))),
            active_txn: RwLock::new(None),
            queries_executed: AtomicU64::new(0),
            config,
            _ephemeral_dir: None,
        })
    }

    /// Open an ephemeral (in-memory-like) connection backed by a temp directory.
    ///
    /// Uses the full DurableStorage engine (WAL, MVCC, SSI) but writes to a
    /// temporary directory that is automatically cleaned up when dropped.
    /// This ensures test and production code paths are identical.
    ///
    /// # Example
    /// ```ignore
    /// let conn = DurableConnection::open_ephemeral()?;
    /// conn.put(b"key", b"value")?;
    /// // temp directory cleaned up on drop
    /// ```
    pub fn open_ephemeral() -> Result<Self> {
        Self::open_ephemeral_with_config(ConnectionConfig::default())
    }

    /// Open an ephemeral connection with custom configuration.
    pub fn open_ephemeral_with_config(config: ConnectionConfig) -> Result<Self> {
        let handle = DurableStorage::open_ephemeral()
            .map_err(|e| ClientError::Storage(e.to_string()))?;
        let (storage, tmpdir) = handle.into_parts();

        // Apply sync mode
        let sync_mode = match config.sync_mode {
            SyncModeClient::Off => 0,
            SyncModeClient::Normal => 1,
            SyncModeClient::Full => 2,
        };
        storage.set_sync_mode(sync_mode);

        Ok(Self {
            storage: Arc::new(storage),
            tch: Arc::new(RwLock::new(TrieColumnarHybrid::new())),
            catalog: Arc::new(RwLock::new(Catalog::new("sochdb"))),
            active_txn: RwLock::new(None),
            queries_executed: AtomicU64::new(0),
            config,
            _ephemeral_dir: Some(tmpdir),
        })
    }

    /// Get the current configuration
    pub fn config(&self) -> &ConnectionConfig {
        &self.config
    }

    /// Perform crash recovery if needed
    pub fn recover(&self) -> Result<RecoveryResult> {
        let stats = self
            .storage
            .recover()
            .map_err(|e| ClientError::Storage(e.to_string()))?;

        Ok(RecoveryResult {
            transactions_recovered: stats.transactions_recovered,
            writes_recovered: stats.writes_recovered,
            commit_ts: stats.commit_ts,
        })
    }

    /// Register a table with its schema
    pub fn register_table(
        &self,
        name: &str,
        fields: &[(String, FieldType)],
    ) -> Result<Vec<ColumnRef>> {
        let cols = self.tch.write().register_table(name, fields);
        Ok(cols)
    }

    /// Resolve a path (O(|path|) lookup)
    pub fn resolve(&self, path: &str) -> Result<PathResolution> {
        Ok(self.tch.read().resolve(path))
    }

    /// Begin a new transaction with snapshot isolation
    pub fn begin_txn(&self) -> Result<u64> {
        let txn_id = self
            .storage
            .begin_transaction()
            .map_err(|e| ClientError::Storage(e.to_string()))?;
        *self.active_txn.write() = Some(txn_id);
        Ok(txn_id)
    }

    /// Get or create active transaction
    fn ensure_txn(&self) -> Result<u64> {
        let active = *self.active_txn.read();
        match active {
            Some(txn) => Ok(txn),
            None => self.begin_txn(),
        }
    }

    /// Commit the active transaction
    pub fn commit_txn(&self) -> Result<u64> {
        let txn_id = self
            .active_txn
            .write()
            .take()
            .ok_or_else(|| ClientError::Transaction("No active transaction".into()))?;

        self.storage
            .commit(txn_id)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Abort the active transaction
    pub fn abort_txn(&self) -> Result<()> {
        let txn_id = self
            .active_txn
            .write()
            .take()
            .ok_or_else(|| ClientError::Transaction("No active transaction".into()))?;

        self.storage
            .abort(txn_id)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Put a key-value pair (within active transaction)
    pub fn put(&self, key: &[u8], value: &[u8]) -> Result<()> {
        let txn_id = self.ensure_txn()?;
        self.storage
            .write(txn_id, key.to_vec(), value.to_vec())
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Get a value by key (within active transaction)
    pub fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
        let txn_id = self.ensure_txn()?;
        self.storage
            .read(txn_id, key)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Delete a key (within active transaction)
    pub fn delete(&self, key: &[u8]) -> Result<()> {
        let txn_id = self.ensure_txn()?;
        self.storage
            .delete(txn_id, key.to_vec())
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Scan keys with a prefix (within active transaction)
    pub fn scan(&self, prefix: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        let txn_id = self.ensure_txn()?;
        self.storage
            .scan(txn_id, prefix)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Put a path-value pair (TCH-style access)
    ///
    /// Converts path to key and stores the value.
    /// Path format: "table.row_id.field" or "collection/doc_id/field"
    pub fn put_path(&self, path: &str, value: &[u8]) -> Result<()> {
        let key = path.as_bytes();
        self.put(key, value)
    }

    /// Get a value by path (TCH-style access)
    pub fn get_path(&self, path: &str) -> Result<Option<Vec<u8>>> {
        let key = path.as_bytes();
        self.get(key)
    }

    /// Delete by path
    pub fn delete_path(&self, path: &str) -> Result<()> {
        let key = path.as_bytes();
        self.delete(key)
    }

    /// Scan by path prefix
    pub fn scan_path(&self, prefix: &str) -> Result<Vec<(String, Vec<u8>)>> {
        let key_prefix = prefix.as_bytes();
        let results = self.scan(key_prefix)?;

        Ok(results
            .into_iter()
            .filter_map(|(k, v)| String::from_utf8(k).ok().map(|path| (path, v)))
            .collect())
    }

    /// Force fsync to disk
    pub fn fsync(&self) -> Result<()> {
        self.storage
            .fsync()
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Create a checkpoint
    pub fn checkpoint(&self) -> Result<u64> {
        self.storage
            .checkpoint()
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Run garbage collection
    pub fn gc(&self) -> Result<usize> {
        Ok(self.storage.gc())
    }

    /// Zero-allocation insert using slice-based values (fastest path)
    ///
    /// This is the high-performance insert path that matches benchmark performance.
    /// Values must be in schema column order; use None for NULL values.
    ///
    /// # Performance
    /// - Eliminates ~6 allocations per row compared to HashMap-based insert
    /// - Expected throughput: 1.2M-1.5M inserts/sec
    ///
    /// # Arguments
    /// * `table` - Table name
    /// * `row_id` - Row identifier  
    /// * `values` - Values in schema column order (None = NULL)
    ///
    /// # Example
    /// ```ignore
    /// let values: &[Option<&SochValue>] = &[
    ///     Some(&SochValue::UInt(1)),
    ///     Some(&SochValue::Text("Alice".into())),
    ///     None, // NULL for optional field
    /// ];
    /// conn.insert_row_slice("users", 1, values)?;
    /// ```
    pub fn insert_row_slice(
        &self,
        table: &str,
        row_id: u64,
        values: &[Option<&sochdb_core::soch::SochValue>],
    ) -> Result<()> {
        let txn_id = self.ensure_txn()?;

        // Use KeyBuffer for zero-allocation key construction
        use sochdb_storage::key_buffer::KeyBuffer;
        let key = KeyBuffer::format_row_key(table, row_id);

        // Pack values using the storage layer's PackedRow
        use sochdb_storage::packed_row::{
            PackedColumnDef, PackedColumnType, PackedRow, PackedTableSchema,
        };

        // Get schema from TCH (or create a minimal one for packing)
        let tch = self.tch.read();
        if let Some(table_info) = tch.tables.get(table) {
            // Convert TCH schema to PackedTableSchema
            let packed_cols: Vec<PackedColumnDef> = table_info
                .schema
                .fields
                .iter()
                .zip(table_info.schema.types.iter())
                .map(|(name, ty)| PackedColumnDef {
                    name: name.clone(),
                    col_type: match ty {
                        FieldType::Int64 => PackedColumnType::Int64,
                        FieldType::UInt64 => PackedColumnType::UInt64,
                        FieldType::Float64 => PackedColumnType::Float64,
                        FieldType::Text => PackedColumnType::Text,
                        FieldType::Bytes => PackedColumnType::Binary,
                        FieldType::Bool => PackedColumnType::Bool,
                    },
                    nullable: true,
                })
                .collect();

            let packed_schema = PackedTableSchema::new(table, packed_cols);
            let packed_row = PackedRow::pack_slice(&packed_schema, values);

            drop(tch); // Release read lock before writing

            self.storage
                .write(
                    txn_id,
                    key.as_bytes().to_vec(),
                    packed_row.as_bytes().to_vec(),
                )
                .map_err(|e| ClientError::Storage(e.to_string()))
        } else {
            drop(tch);
            Err(ClientError::NotFound(format!(
                "Table '{}' not found",
                table
            )))
        }
    }

    /// Bulk insert with zero-allocation path (fastest bulk insert)
    ///
    /// This combines streaming batch mode with the zero-allocation insert path.
    ///
    /// # Arguments
    /// * `table` - Table name
    /// * `rows` - Iterator of (row_id, values) where values are in schema column order
    /// * `batch_size` - Number of rows per transaction commit (for memory bounds)
    ///
    /// # Returns
    /// Number of rows successfully inserted
    pub fn bulk_insert_slice<'a, I>(&self, table: &str, rows: I, batch_size: usize) -> Result<usize>
    where
        I: IntoIterator<Item = (u64, Vec<Option<&'a sochdb_core::soch::SochValue>>)>,
    {
        let mut count = 0;
        let mut batch_count = 0;

        for (row_id, values) in rows {
            // Convert Vec to slice for the call
            let value_refs: Vec<Option<&sochdb_core::soch::SochValue>> = values;
            self.insert_row_slice(table, row_id, &value_refs)?;
            count += 1;
            batch_count += 1;

            // Auto-commit at batch boundaries
            if batch_count >= batch_size {
                self.commit_txn()?;
                batch_count = 0;
            }
        }

        // Commit any remaining rows
        if batch_count > 0 {
            self.commit_txn()?;
        }

        Ok(count)
    }

    /// Get storage statistics
    pub fn stats(&self) -> DurableStats {
        DurableStats {
            queries_executed: self.queries_executed.load(Ordering::Relaxed),
            tables_registered: self.tch.read().tables.len() as u64,
        }
    }
}

/// Statistics for durable connection
#[derive(Debug, Clone)]
pub struct DurableStats {
    pub queries_executed: u64,
    pub tables_registered: u64,
}

// =============================================================================
// Connection Mode Enforcement (Task 6)
// =============================================================================

/// Connection mode for database access
///
/// Determines which operations are allowed on a connection.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectionModeClient {
    /// Read-only access - write operations return errors
    ReadOnly,
    /// Read-write access - all operations allowed
    ReadWrite,
}

/// Read-only database connection
///
/// This connection type enforces read-only access at the API level.
/// All write methods (put, delete, etc.) are not available on this type.
///
/// ## Use Case
///
/// Use this when you need concurrent read access while another process
/// is writing to the database (e.g., Flowtrace App reading while a
/// background script writes).
///
/// ## Example
///
/// ```ignore
/// use sochdb::ReadOnlyConnection;
///
/// // Open read-only connection (acquires shared lock)
/// let reader = ReadOnlyConnection::open("./data")?;
///
/// // Read operations work
/// let value = reader.get(b"key")?;
/// let items = reader.scan(b"prefix")?;
///
/// // Write operations are not available - compile error!
/// // reader.put(b"key", b"value"); // Error: no method named `put`
/// ```
pub struct ReadOnlyConnection {
    /// The underlying durable storage
    storage: Arc<DurableStorage>,
    /// Trie-Columnar Hybrid for O(|path|) resolution
    tch: Arc<RwLock<TrieColumnarHybrid>>,
    /// Active transaction ID for consistent reads
    active_txn: RwLock<Option<u64>>,
    /// Statistics
    queries_executed: AtomicU64,
}

impl ReadOnlyConnection {
    /// Open a read-only connection to the database.
    ///
    /// Multiple read-only connections can be open simultaneously.
    /// Write operations are not available on this connection type.
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        // Open storage (uses shared lock internally)
        let storage = DurableStorage::open(path.as_ref())
            .map_err(|e| ClientError::Storage(e.to_string()))?;

        Ok(Self {
            storage: Arc::new(storage),
            tch: Arc::new(RwLock::new(TrieColumnarHybrid::new())),
            active_txn: RwLock::new(None),
            queries_executed: AtomicU64::new(0),
        })
    }

    /// Begin a read transaction for consistent snapshot reads
    pub fn begin_read_txn(&self) -> Result<u64> {
        let txn_id = self
            .storage
            .begin_transaction()
            .map_err(|e| ClientError::Storage(e.to_string()))?;
        *self.active_txn.write() = Some(txn_id);
        Ok(txn_id)
    }

    /// End the read transaction
    pub fn end_read_txn(&self) -> Result<()> {
        if let Some(txn_id) = self.active_txn.write().take() {
            // Abort since read-only txns don't need to commit
            self.storage
                .abort(txn_id)
                .map_err(|e| ClientError::Storage(e.to_string()))?;
        }
        Ok(())
    }

    /// Get or create read transaction
    fn ensure_read_txn(&self) -> Result<u64> {
        let active = *self.active_txn.read();
        match active {
            Some(txn) => Ok(txn),
            None => self.begin_read_txn(),
        }
    }

    /// Get a value by key
    pub fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
        let txn_id = self.ensure_read_txn()?;
        self.queries_executed.fetch_add(1, Ordering::Relaxed);
        self.storage
            .read(txn_id, key)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Scan keys with a prefix
    pub fn scan(&self, prefix: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        let txn_id = self.ensure_read_txn()?;
        self.queries_executed.fetch_add(1, Ordering::Relaxed);
        self.storage
            .scan(txn_id, prefix)
            .map_err(|e| ClientError::Storage(e.to_string()))
    }

    /// Get a value by path
    pub fn get_path(&self, path: &str) -> Result<Option<Vec<u8>>> {
        self.get(path.as_bytes())
    }

    /// Scan by path prefix
    pub fn scan_path(&self, prefix: &str) -> Result<Vec<(String, Vec<u8>)>> {
        let results = self.scan(prefix.as_bytes())?;
        Ok(results
            .into_iter()
            .filter_map(|(k, v)| String::from_utf8(k).ok().map(|path| (path, v)))
            .collect())
    }

    /// Resolve a path (O(|path|) lookup)
    pub fn resolve(&self, path: &str) -> Result<PathResolution> {
        Ok(self.tch.read().resolve(path))
    }

    /// Get query statistics
    pub fn queries_executed(&self) -> u64 {
        self.queries_executed.load(Ordering::Relaxed)
    }
}

/// Trait for read operations (shared by ReadOnly and ReadWrite connections)
pub trait ReadableConnection {
    /// Get a value by key
    fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>>;
    
    /// Scan keys with a prefix
    fn scan(&self, prefix: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>>;
    
    /// Get a value by path
    fn get_path(&self, path: &str) -> Result<Option<Vec<u8>>> {
        self.get(path.as_bytes())
    }
    
    /// Scan by path prefix
    fn scan_path(&self, prefix: &str) -> Result<Vec<(String, Vec<u8>)>> {
        let results = self.scan(prefix.as_bytes())?;
        Ok(results
            .into_iter()
            .filter_map(|(k, v)| String::from_utf8(k).ok().map(|path| (path, v)))
            .collect())
    }
}

/// Trait for write operations (only on ReadWrite connections)
pub trait WritableConnection: ReadableConnection {
    /// Put a key-value pair
    fn put(&self, key: &[u8], value: &[u8]) -> Result<()>;
    
    /// Delete a key
    fn delete(&self, key: &[u8]) -> Result<()>;
    
    /// Begin a transaction
    fn begin_txn(&self) -> Result<u64>;
    
    /// Commit a transaction
    fn commit_txn(&self) -> Result<u64>;
    
    /// Abort a transaction
    fn abort_txn(&self) -> Result<()>;
}

impl ReadableConnection for ReadOnlyConnection {
    fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
        ReadOnlyConnection::get(self, key)
    }
    
    fn scan(&self, prefix: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        ReadOnlyConnection::scan(self, prefix)
    }
}

impl ReadableConnection for DurableConnection {
    fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
        DurableConnection::get(self, key)
    }
    
    fn scan(&self, prefix: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        DurableConnection::scan(self, prefix)
    }
}

impl WritableConnection for DurableConnection {
    fn put(&self, key: &[u8], value: &[u8]) -> Result<()> {
        DurableConnection::put(self, key, value)
    }
    
    fn delete(&self, key: &[u8]) -> Result<()> {
        DurableConnection::delete(self, key)
    }
    
    fn begin_txn(&self) -> Result<u64> {
        DurableConnection::begin_txn(self)
    }
    
    fn commit_txn(&self) -> Result<u64> {
        DurableConnection::commit_txn(self)
    }
    
    fn abort_txn(&self) -> Result<()> {
        DurableConnection::abort_txn(self)
    }
}

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

    #[test]
    fn test_connection_open() {
        let conn = SochConnection::open("./test_data").unwrap();
        assert!(conn.list_tables().is_empty());
    }

    #[test]
    fn test_register_table() {
        let conn = SochConnection::open("./test_data").unwrap();

        let fields = vec![
            ("id".to_string(), FieldType::UInt64),
            ("name".to_string(), FieldType::Text),
            ("score".to_string(), FieldType::Float64),
        ];

        let cols = conn.register_table("users", &fields).unwrap();
        assert_eq!(cols.len(), 3);
        assert_eq!(cols[0].name, "id");
    }

    #[test]
    fn test_path_resolution() {
        let conn = SochConnection::open("./test_data").unwrap();

        let fields = vec![
            ("id".to_string(), FieldType::UInt64),
            ("name".to_string(), FieldType::Text),
        ];
        conn.register_table("users", &fields).unwrap();

        // Resolve table
        match conn.resolve("users").unwrap() {
            PathResolution::Array { schema, columns } => {
                assert_eq!(schema.name, "users");
                assert_eq!(columns.len(), 2);
            }
            _ => panic!("Expected Array resolution"),
        }

        // Resolve column
        match conn.resolve("users.name").unwrap() {
            PathResolution::Value(col) => {
                assert_eq!(col.name, "name");
            }
            _ => panic!("Expected Value resolution"),
        }

        // Not found
        match conn.resolve("nonexistent").unwrap() {
            PathResolution::NotFound => {}
            _ => panic!("Expected NotFound"),
        }
    }

    #[test]
    fn test_transaction_lifecycle() {
        let conn = SochConnection::open("./test_data").unwrap();

        let txn_id = conn.begin_txn().unwrap();
        assert!(txn_id > 0);

        let commit_ts = conn.commit_txn().unwrap();
        assert!(commit_ts > 0);
    }

    #[test]
    fn test_stats() {
        let conn = SochConnection::open("./test_data").unwrap();
        conn.record_query();
        conn.record_tokens(100, 200);

        let stats = conn.stats();
        assert_eq!(stats.queries_executed, 1);
        assert_eq!(stats.soch_tokens_emitted, 100);
        assert_eq!(stats.json_tokens_equivalent, 200);
        assert!((stats.token_savings_percent - 50.0).abs() < 0.1);
    }

    #[test]
    fn test_tch_insert_and_select() {
        let conn = SochConnection::open("./test_data").unwrap();

        // Register table
        let fields = vec![
            ("id".to_string(), FieldType::UInt64),
            ("name".to_string(), FieldType::Text),
            ("score".to_string(), FieldType::Float64),
        ];
        conn.register_table("users", &fields).unwrap();

        // Insert rows
        let mut tch = conn.tch.write();

        let mut row1 = std::collections::HashMap::new();
        row1.insert("id".to_string(), SochValue::UInt(1));
        row1.insert("name".to_string(), SochValue::Text("Alice".to_string()));
        row1.insert("score".to_string(), SochValue::Float(95.5));
        let id1 = tch.insert_row("users", &row1);
        assert_eq!(id1, 1);

        let mut row2 = std::collections::HashMap::new();
        row2.insert("id".to_string(), SochValue::UInt(2));
        row2.insert("name".to_string(), SochValue::Text("Bob".to_string()));
        row2.insert("score".to_string(), SochValue::Float(87.2));
        let id2 = tch.insert_row("users", &row2);
        assert_eq!(id2, 2);

        // Select all rows
        let cursor = tch.select("users", &[], None, None, None, None);
        drop(tch);

        let rows: Vec<_> = {
            let mut cursor = cursor;
            let mut rows = Vec::new();
            while let Some(row) = cursor.next() {
                rows.push(row);
            }
            rows
        };

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

        // Verify count
        let tch = conn.tch.read();
        assert_eq!(tch.count_rows("users"), 2);
    }

    #[test]
    fn test_tch_where_clause() {
        let conn = SochConnection::open("./test_data").unwrap();

        // Register and populate table
        let fields = vec![
            ("id".to_string(), FieldType::UInt64),
            ("name".to_string(), FieldType::Text),
            ("score".to_string(), FieldType::Float64),
        ];
        conn.register_table("users", &fields).unwrap();

        let mut tch = conn.tch.write();
        for i in 1..=5 {
            let mut row = std::collections::HashMap::new();
            row.insert("id".to_string(), SochValue::UInt(i));
            row.insert("name".to_string(), SochValue::Text(format!("User{}", i)));
            row.insert("score".to_string(), SochValue::Float((i * 20) as f64));
            tch.insert_row("users", &row);
        }

        // Select with WHERE clause (score > 60)
        let where_clause = WhereClause::Simple {
            field: "score".to_string(),
            op: CompareOp::Gt,
            value: SochValue::Float(60.0),
        };
        let cursor = tch.select("users", &[], Some(&where_clause), None, None, None);

        let rows: Vec<_> = {
            let mut cursor = cursor;
            let mut rows = Vec::new();
            while let Some(row) = cursor.next() {
                rows.push(row);
            }
            rows
        };

        // Users 4 and 5 have scores 80 and 100
        assert_eq!(rows.len(), 2);
    }

    #[test]
    fn test_tch_update_and_delete() {
        let conn = SochConnection::open("./test_data").unwrap();

        // Register and populate table
        let fields = vec![
            ("id".to_string(), FieldType::UInt64),
            ("name".to_string(), FieldType::Text),
        ];
        conn.register_table("users", &fields).unwrap();

        let mut tch = conn.tch.write();

        let mut row = std::collections::HashMap::new();
        row.insert("id".to_string(), SochValue::UInt(1));
        row.insert("name".to_string(), SochValue::Text("Alice".to_string()));
        tch.insert_row("users", &row);

        // Update
        let mut updates = std::collections::HashMap::new();
        updates.insert(
            "name".to_string(),
            SochValue::Text("Alice Updated".to_string()),
        );
        let where_clause = WhereClause::Simple {
            field: "id".to_string(),
            op: CompareOp::Eq,
            value: SochValue::UInt(1),
        };
        let update_result = tch.update_rows("users", &updates, Some(&where_clause));
        assert_eq!(update_result.affected_count, 1);
        assert_eq!(update_result.affected_row_ids.len(), 1);

        // Verify update
        let cursor = tch.select("users", &[], None, None, None, None);
        let rows: Vec<_> = {
            let mut cursor = cursor;
            let mut rows = Vec::new();
            while let Some(row) = cursor.next() {
                rows.push(row);
            }
            rows
        };
        assert_eq!(
            rows[0].get("name"),
            Some(&SochValue::Text("Alice Updated".to_string()))
        );

        // Delete
        let delete_result = tch.delete_rows("users", Some(&where_clause));
        assert_eq!(delete_result.affected_count, 1);
        assert_eq!(delete_result.affected_row_ids.len(), 1);
        assert_eq!(tch.count_rows("users"), 0);
    }

    #[test]
    fn test_tch_upsert() {
        let conn = SochConnection::open("./test_data").unwrap();

        // Register table
        let fields = vec![
            ("id".to_string(), FieldType::UInt64),
            ("name".to_string(), FieldType::Text),
        ];
        conn.register_table("users", &fields).unwrap();

        let mut tch = conn.tch.write();

        // First upsert should insert
        let mut row = std::collections::HashMap::new();
        row.insert("id".to_string(), SochValue::UInt(1));
        row.insert("name".to_string(), SochValue::Text("Alice".to_string()));
        let action = tch.upsert_row("users", "id", &row);
        assert_eq!(action, UpsertAction::Inserted);

        // Second upsert with same id should update
        let mut row2 = std::collections::HashMap::new();
        row2.insert("id".to_string(), SochValue::UInt(1));
        row2.insert(
            "name".to_string(),
            SochValue::Text("Alice Updated".to_string()),
        );
        let action = tch.upsert_row("users", "id", &row2);
        assert_eq!(action, UpsertAction::Updated);

        // Verify only 1 row exists
        assert_eq!(tch.count_rows("users"), 1);
    }

    // ========================================================================
    // DurableConnection tests
    // ========================================================================

    #[test]
    fn test_durable_connection_basic() {
        use tempfile::tempdir;

        let dir = tempdir().unwrap();
        let conn = DurableConnection::open(dir.path()).unwrap();

        // Begin transaction
        let txn = conn.begin_txn().unwrap();
        assert!(txn > 0);

        // Write some data
        conn.put(b"key1", b"value1").unwrap();
        conn.put(b"key2", b"value2").unwrap();

        // Read back (within same txn sees own writes)
        let v1 = conn.get(b"key1").unwrap();
        assert_eq!(v1, Some(b"value1".to_vec()));

        // Commit
        let commit_ts = conn.commit_txn().unwrap();
        assert!(commit_ts > 0);

        // New transaction should see committed data
        conn.begin_txn().unwrap();
        let v2 = conn.get(b"key1").unwrap();
        assert_eq!(v2, Some(b"value1".to_vec()));
        conn.abort_txn().unwrap();
    }

    #[test]
    fn test_durable_connection_path_api() {
        use tempfile::tempdir;

        let dir = tempdir().unwrap();
        let conn = DurableConnection::open(dir.path()).unwrap();

        // Use path-based API
        conn.begin_txn().unwrap();
        conn.put_path("users/1/name", b"Alice").unwrap();
        conn.put_path("users/1/email", b"alice@example.com")
            .unwrap();
        conn.put_path("users/2/name", b"Bob").unwrap();
        conn.commit_txn().unwrap();

        // Read back
        conn.begin_txn().unwrap();
        let name = conn.get_path("users/1/name").unwrap();
        assert_eq!(name, Some(b"Alice".to_vec()));

        // Scan by prefix
        let users = conn.scan_path("users/1/").unwrap();
        assert_eq!(users.len(), 2);
        conn.abort_txn().unwrap();
    }

    #[test]
    fn test_durable_connection_crash_recovery() {
        use tempfile::tempdir;

        let dir = tempdir().unwrap();

        // Phase 1: Write and commit with full durability (fsync on every commit)
        {
            let conn =
                DurableConnection::open_with_config(dir.path(), ConnectionConfig::max_durability())
                    .unwrap();
            conn.begin_txn().unwrap();
            conn.put(b"persist", b"this data").unwrap();
            conn.commit_txn().unwrap();

            // Clean shutdown - lock will be released
        }

        // Phase 2: Recover and verify
        {
            let conn = DurableConnection::open(dir.path()).unwrap();
            let _stats = conn.recover().unwrap();
            // Recovery should find and replay the committed transaction

            // Data should be there
            conn.begin_txn().unwrap();
            let v = conn.get(b"persist").unwrap();
            assert_eq!(v, Some(b"this data".to_vec()));
            conn.abort_txn().unwrap();
        }
    }

    // ==================== LSCS Storage Tests ====================

    #[test]
    fn test_lscs_storage_basic_put_get() {
        let storage = LscsStorage::new();

        // Put some values
        storage.put(b"key1", b"value1").unwrap();
        storage.put(b"key2", b"value2").unwrap();
        storage.put(b"key3", b"value3").unwrap();

        // Get them back
        assert_eq!(storage.get("", b"key1").unwrap(), Some(b"value1".to_vec()));
        assert_eq!(storage.get("", b"key2").unwrap(), Some(b"value2".to_vec()));
        assert_eq!(storage.get("", b"key3").unwrap(), Some(b"value3".to_vec()));
        assert_eq!(storage.get("", b"nonexistent").unwrap(), None);
    }

    #[test]
    fn test_lscs_storage_update() {
        let storage = LscsStorage::new();

        // Put initial value
        storage.put(b"key1", b"original").unwrap();
        assert_eq!(
            storage.get("", b"key1").unwrap(),
            Some(b"original".to_vec())
        );

        // Update it
        storage.put(b"key1", b"updated").unwrap();
        assert_eq!(storage.get("", b"key1").unwrap(), Some(b"updated".to_vec()));
    }

    #[test]
    fn test_lscs_storage_delete() {
        let storage = LscsStorage::new();

        // Put and verify
        storage.put(b"key1", b"value1").unwrap();
        assert_eq!(storage.get("", b"key1").unwrap(), Some(b"value1".to_vec()));

        // Delete and verify tombstone
        storage.delete(b"key1").unwrap();
        assert_eq!(storage.get("", b"key1").unwrap(), None);
    }

    #[test]
    fn test_lscs_storage_scan() {
        let storage = LscsStorage::new();

        // Insert test data
        storage.put(b"user:1:name", b"Alice").unwrap();
        storage.put(b"user:1:email", b"alice@test.com").unwrap();
        storage.put(b"user:2:name", b"Bob").unwrap();
        storage.put(b"user:2:email", b"bob@test.com").unwrap();
        storage.put(b"product:1:name", b"Widget").unwrap();

        // Scan user range
        let results = storage.scan(b"user:1:", b"user:1:\xff", 10).unwrap();
        assert_eq!(results.len(), 2);

        // Scan all users
        let results = storage.scan(b"user:", b"user:\xff", 10).unwrap();
        assert_eq!(results.len(), 4);
    }

    #[test]
    fn test_lscs_storage_wal_integrity() {
        let storage = LscsStorage::new();

        // Write some data
        for i in 0..100 {
            storage
                .put(
                    format!("key{}", i).as_bytes(),
                    format!("value{}", i).as_bytes(),
                )
                .unwrap();
        }

        // Verify WAL
        let wal_result = storage.verify_wal().unwrap();
        assert_eq!(wal_result.total_entries, 100);
        assert_eq!(wal_result.valid_entries, 100);
        assert_eq!(wal_result.corrupted_entries, 0);
    }

    #[test]
    fn test_lscs_storage_checkpoint() {
        let storage = LscsStorage::new();

        // Initial state
        assert_eq!(storage.last_checkpoint_lsn(), 0);

        // Write data
        storage.put(b"key1", b"value1").unwrap();
        storage.put(b"key2", b"value2").unwrap();

        // Force checkpoint
        let checkpoint_lsn = storage.force_checkpoint().unwrap();
        assert!(checkpoint_lsn >= 2);
        assert_eq!(storage.last_checkpoint_lsn(), checkpoint_lsn);
    }

    #[test]
    fn test_lscs_storage_wal_truncate() {
        let storage = LscsStorage::new();

        // Write data
        for i in 0..50 {
            storage
                .put(format!("key{}", i).as_bytes(), b"value")
                .unwrap();
        }

        let stats_before = storage.wal_stats();
        assert_eq!(stats_before.entry_count, 50);

        // Truncate up to LSN 25
        let removed = storage.truncate_wal(25).unwrap();
        assert!(removed > 0);

        let stats_after = storage.wal_stats();
        assert!(stats_after.entry_count < 50);
    }

    #[test]
    fn test_lscs_storage_replay() {
        let storage = LscsStorage::new();

        // Write data
        storage.put(b"key1", b"value1").unwrap();
        storage.put(b"key2", b"value2").unwrap();

        // Initial checkpoint at 0
        assert_eq!(storage.last_checkpoint_lsn(), 0);

        // All WAL entries should be replayable
        let replayed = storage.replay_wal_from_checkpoint().unwrap();
        assert!(replayed > 0);
    }

    #[test]
    fn test_bloom_filter() {
        let mut bloom = BloomFilter::new(1000, 0.01);

        // Insert some keys
        for i in 0..100 {
            bloom.insert(format!("key{}", i).as_bytes());
        }

        // Check inserted keys
        for i in 0..100 {
            assert!(bloom.may_contain(format!("key{}", i).as_bytes()));
        }

        // Check non-existent keys (some false positives allowed)
        let mut false_positives = 0;
        for i in 100..1000 {
            if bloom.may_contain(format!("key{}", i).as_bytes()) {
                false_positives += 1;
            }
        }
        // FPR should be around 1%
        assert!(false_positives < 50); // Allow up to 5% for statistical variance
    }

    #[test]
    fn test_sstable_creation_and_lookup() {
        let entries = vec![
            SstEntry {
                key: b"aaa".to_vec(),
                value: b"v1".to_vec(),
                timestamp: 1,
                deleted: false,
            },
            SstEntry {
                key: b"bbb".to_vec(),
                value: b"v2".to_vec(),
                timestamp: 2,
                deleted: false,
            },
            SstEntry {
                key: b"ccc".to_vec(),
                value: b"v3".to_vec(),
                timestamp: 3,
                deleted: false,
            },
        ];

        let sst = SSTable::from_entries(entries, 0, 1).unwrap();

        // Lookup existing keys
        assert_eq!(sst.get(b"aaa").map(|e| &e.value), Some(&b"v1".to_vec()));
        assert_eq!(sst.get(b"bbb").map(|e| &e.value), Some(&b"v2".to_vec()));
        assert_eq!(sst.get(b"ccc").map(|e| &e.value), Some(&b"v3".to_vec()));

        // Lookup non-existent key
        assert!(sst.get(b"zzz").is_none());
    }

    #[test]
    fn test_lscs_storage_many_writes() {
        let storage = LscsStorage::new();

        // Write many keys to trigger memtable flush
        for i in 0..10000 {
            let key = format!("key{:06}", i);
            let value = format!("value{:06}", i);
            storage.put(key.as_bytes(), value.as_bytes()).unwrap();
        }

        // Force flush
        storage.fsync().unwrap();

        // Verify random samples
        for i in (0..10000).step_by(100) {
            let key = format!("key{:06}", i);
            let expected = format!("value{:06}", i);
            let actual = storage.get("", key.as_bytes()).unwrap();
            assert_eq!(actual, Some(expected.into_bytes()));
        }
    }

    #[test]
    fn test_lscs_mvcc_newest_wins() {
        let storage = LscsStorage::new();

        // Write multiple versions
        storage.put(b"key", b"v1").unwrap();
        std::thread::sleep(std::time::Duration::from_millis(1));
        storage.put(b"key", b"v2").unwrap();
        std::thread::sleep(std::time::Duration::from_millis(1));
        storage.put(b"key", b"v3").unwrap();

        // Should get newest
        assert_eq!(storage.get("", b"key").unwrap(), Some(b"v3".to_vec()));
    }

    #[test]
    fn test_lscs_storage_recovery_needed() {
        let storage = LscsStorage::new();

        // Initially no recovery needed
        assert!(!storage.needs_recovery());

        // After write, recovery is needed (WAL has entries beyond checkpoint)
        storage.put(b"key", b"value").unwrap();
        assert!(storage.needs_recovery());

        // After checkpoint, no recovery needed
        storage.force_checkpoint().unwrap();
        assert!(!storage.needs_recovery());
    }

    #[test]
    fn test_lscs_scan_across_sstables() {
        // Write enough data to trigger memtable flush, then scan across both
        // memtable and on-disk SSTables.
        let storage = LscsStorage::new();

        // Write 10000 entries to ensure memtable flushes to SSTables
        for i in 0..10000 {
            let key = format!("scankey{:06}", i);
            let value = format!("scanval{:06}", i);
            storage.put(key.as_bytes(), value.as_bytes()).unwrap();
        }

        // Force flush to SSTable
        storage.fsync().unwrap();

        // Write a few more entries into the active memtable
        for i in 10000..10010 {
            let key = format!("scankey{:06}", i);
            let value = format!("scanval{:06}", i);
            storage.put(key.as_bytes(), value.as_bytes()).unwrap();
        }

        // Scan a range that spans both SSTables and memtable
        let results = storage
            .scan(b"scankey000000", b"scankey010009", 20000)
            .unwrap();

        // All 10010 entries should be visible
        assert_eq!(
            results.len(),
            10010,
            "scan should return entries from both SSTables and memtable, got {}",
            results.len()
        );

        // Verify first and last entries
        assert_eq!(results[0].0, b"scankey000000");
        assert_eq!(results[0].1, b"scanval000000");
        assert_eq!(results.last().unwrap().0, b"scankey010009");

        // Verify entries in the middle (likely from SSTable)
        let mid_key = b"scankey005000".to_vec();
        let mid_val = b"scanval005000".to_vec();
        let found = results.iter().find(|(k, _)| k == &mid_key);
        assert!(
            found.is_some(),
            "mid-range key from SSTable should be visible"
        );
        assert_eq!(found.unwrap().1, mid_val);
    }

    #[test]
    fn test_lscs_scan_tombstone_shadowing() {
        // Write data, delete some, flush, and verify scan omits deleted entries.
        let storage = LscsStorage::new();

        for i in 0..100 {
            let key = format!("ts_key{:04}", i);
            let value = format!("ts_val{:04}", i);
            storage.put(key.as_bytes(), value.as_bytes()).unwrap();
        }

        // Delete entries 50..60
        for i in 50..60 {
            let key = format!("ts_key{:04}", i);
            storage.delete(key.as_bytes()).unwrap();
        }

        // Scan full range
        let results = storage
            .scan(b"ts_key0000", b"ts_key0099", 200)
            .unwrap();

        // Should have 90 entries (100 - 10 deleted)
        assert_eq!(
            results.len(),
            90,
            "scan should omit tombstoned entries, got {}",
            results.len()
        );

        // Verify deleted keys are absent
        for i in 50..60 {
            let del_key = format!("ts_key{:04}", i).into_bytes();
            assert!(
                !results.iter().any(|(k, _)| k == &del_key),
                "deleted key ts_key{:04} should not appear in scan",
                i
            );
        }
    }
}

// Implement ConnectionTrait for DurableConnection
impl crate::ConnectionTrait for DurableConnection {
    fn put(&self, key: &[u8], value: &[u8]) -> Result<()> {
        DurableConnection::put(self, key, value)
    }

    fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
        DurableConnection::get(self, key)
    }

    fn delete(&self, key: &[u8]) -> Result<()> {
        DurableConnection::delete(self, key)
    }

    fn scan(&self, prefix: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        DurableConnection::scan(self, prefix)
    }
}

// Implement ConnectionTrait for SochConnection
impl crate::ConnectionTrait for SochConnection {
    fn put(&self, key: &[u8], value: &[u8]) -> Result<()> {
        SochConnection::put(self, key.to_vec(), value.to_vec())
    }

    fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
        SochConnection::get(self, key)
    }

    fn delete(&self, key: &[u8]) -> Result<()> {
        SochConnection::delete(self, key)
    }

    fn scan(&self, prefix: &[u8]) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
        SochConnection::scan_prefix(self, prefix)
    }
}