motedb 0.1.2

//! Database Core - MoteDB Structure and Initialization
//!
//! Extracted from database_legacy.rs (4,798 lines) as part of modularization
//! This module contains:
//! - MoteDB struct definition
//! - create() / create_with_config()
//! - open() with WAL recovery
//! - Index loading helpers

use crate::config::DBConfig;
use crate::index::btree::{BTree, BTreeConfig};
use crate::index::vamana::{DiskANNIndex, VamanaConfig};
use crate::index::{SpatialHybridIndex, SpatialHybridConfig, BoundingBoxF32};
use crate::index::text_fts::TextFTSIndex;
use crate::index::column_value::ColumnValueIndex;
use crate::storage::{LSMEngine, LSMConfig};
use crate::txn::coordinator::TransactionCoordinator;
use crate::txn::version_store::VersionStore;
use crate::txn::wal::{WALManager, WALRecord};
use crate::types::RowId;
use crate::catalog::TableRegistry;
use crate::cache::RowCache;
use crate::{Result, StorageError, MoteDBError};
use dashmap::DashMap;
use parking_lot::RwLock;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::sync::atomic::AtomicBool;

/// Database statistics
#[derive(Debug, Clone)]
pub struct DatabaseStats {
    pub total_rows: RowId,
    pub num_partitions: u8,
}

/// Vector index statistics
#[derive(Debug, Clone)]
pub struct VectorIndexStats {
    pub total_vectors: usize,
    pub dimension: usize,
    pub cache_hit_rate: f32,
    pub memory_usage: usize,
    pub disk_usage: usize,
}

/// Spatial index statistics
#[derive(Debug, Clone)]
pub struct SpatialIndexStats {
    pub total_entries: usize,
    pub memory_usage: usize,
    pub bytes_per_entry: usize,
}

/// MoteDB instance
pub struct MoteDB {
    /// Database file path
    pub(crate) path: PathBuf,

    /// WAL manager
    pub(crate) wal: Arc<WALManager>,
    
    /// LSM-Tree storage engine (main data storage)
    pub(crate) lsm_engine: Arc<LSMEngine>,

    /// Timestamp index (using BTree for persistent storage)
    pub(crate) timestamp_index: Arc<RwLock<BTree>>,

    /// Next row ID
    pub(crate) next_row_id: Arc<RwLock<RowId>>,
    
    /// 🚀 Phase 4: Per-table AUTO_INCREMENT counters
    /// Format: table_name → next_id
    pub(crate) table_auto_increment: Arc<DashMap<String, Arc<RwLock<i64>>>>,

    /// Number of partitions
    pub(crate) num_partitions: u8,

    /// Transaction coordinator
    pub(crate) txn_coordinator: Arc<TransactionCoordinator>,

    /// Version store for MVCC
    pub(crate) version_store: Arc<VersionStore>,
    
    /// Pending index updates counter (for triggering background flush)
    /// 🚀 P0 CRITICAL FIX: 使用 AtomicUsize 避免锁竞争，解决 CPU 飙升问题
    pub(crate) pending_updates: Arc<std::sync::atomic::AtomicUsize>,
    
    /// Pending spatial index updates counter
    /// 🚀 P0 CRITICAL FIX: 使用 AtomicUsize 避免锁竞争
    pub(crate) pending_spatial_updates: Arc<std::sync::atomic::AtomicUsize>,
    
    /// 🚀 Vector indexes (DiskANN) - 使用 DashMap 提升并发性能
    pub(crate) vector_indexes: Arc<DashMap<String, Arc<RwLock<DiskANNIndex>>>>,
    
    /// 🚀 Spatial indexes (Hybrid Grid+RTree) - 使用 DashMap 提升并发性能
    pub(crate) spatial_indexes: Arc<DashMap<String, Arc<RwLock<SpatialHybridIndex>>>>,
    
    /// 🚀 Text indexes (FTS with single-file B-Tree) - 使用 DashMap 提升并发性能
    pub text_indexes: Arc<DashMap<String, Arc<RwLock<TextFTSIndex>>>>,
    
    /// 🚀 Column value indexes (for WHERE optimization) - 使用 DashMap 提升并发性能
    pub column_indexes: Arc<DashMap<String, Arc<RwLock<ColumnValueIndex>>>>,
    
    /// Table registry (catalog)
    pub(crate) table_registry: Arc<TableRegistry>,
    
    /// 🆕 Index metadata registry
    pub(crate) index_registry: Arc<crate::database::index_metadata::IndexRegistry>,
    
    /// 🚀 P1: Row cache (hot data cache)
    pub(crate) row_cache: Arc<RowCache>,

    /// 🚀 Phase 3+: Index update strategy
    pub(crate) index_update_strategy: crate::config::IndexUpdateStrategy,
    
    /// 🚀 P0: Query timeout (seconds)
    pub(crate) query_timeout_secs: Option<u64>,
    
    /// 🆕 防止递归 flush 的标志
    pub(crate) is_flushing: Arc<AtomicBool>,

    /// 🔒 Checkpoint mutex: prevents concurrent checkpoints (auto + manual)
    /// which can cause deadlock via timestamp_index write lock contention
    pub(crate) checkpoint_mutex: Arc<Mutex<()>>,

    /// Auto-checkpoint thread (if enabled)
    auto_checkpoint_thread: Option<AutoCheckpointThread>,
}

/// Auto-checkpoint background thread
struct AutoCheckpointThread {
    /// Thread handle
    handle: Option<std::thread::JoinHandle<()>>,
    
    /// Stop signal
    should_stop: Arc<AtomicBool>,
}

impl MoteDB {
    /// Create a new database
    pub fn create<P: AsRef<Path>>(path: P) -> Result<Self> {
        Self::create_with_config(path, DBConfig::default())
    }
    
    /// Create a new database with custom configuration
    pub fn create_with_config<P: AsRef<Path>>(path: P, config: DBConfig) -> Result<Self> {
        let path = path.as_ref();
        let db_path = path.with_extension("mote");
        
        // 🎯 统一目录结构：所有文件放在 {name}.mote/ 目录下
        std::fs::create_dir_all(&db_path)?;
        
        let wal_path = db_path.join("wal");
        let lsm_dir = db_path.join("lsm");
        let indexes_dir = db_path.join("indexes");

        let num_partitions = config.num_partitions;

        // Create WAL directory with config
        std::fs::create_dir_all(&wal_path)?;
        let wal_config = crate::txn::wal::WALConfig::from(config.wal_config);
        let wal = Arc::new(WALManager::create_with_config(&wal_path, num_partitions, wal_config)?);

        // Create timestamp index with BTree storage (放在 indexes/ 目录)
        std::fs::create_dir_all(&indexes_dir)?;
        let timestamp_storage = indexes_dir.join("timestamp.idx");
        let btree_config = BTreeConfig {
            unique_keys: false,  // Allow duplicate timestamps
            allow_updates: true,
            ..Default::default()
        };
        let timestamp_index = Arc::new(RwLock::new(BTree::with_config(timestamp_storage, btree_config)?));
        
        // Create LSM-Tree storage engine
        std::fs::create_dir_all(&lsm_dir)?;
        let lsm_engine = Arc::new(LSMEngine::new(lsm_dir, LSMConfig::default())?);

        // Create version store and transaction coordinator
        let version_store = Arc::new(VersionStore::new());
        let txn_coordinator = Arc::new(TransactionCoordinator::new(version_store.clone()));
        
        // Create table registry (catalog)
        let table_registry = Arc::new(TableRegistry::new(&db_path)?);
        
        // 🆕 Create index metadata registry
        let index_registry = Arc::new(crate::database::index_metadata::IndexRegistry::new(&db_path));

        // 🚀 P1: Create row cache (default 10000 rows ≈ 10MB)
        let row_cache = Arc::new(RowCache::new(config.row_cache_size.unwrap_or(10000)));

        // Ensure "_default" table has a stable table_id (= 0)
        table_registry.ensure_default_table_id()?;

        let db = Self {
            path: db_path,
            wal,
            lsm_engine: lsm_engine.clone(),
            timestamp_index,
            next_row_id: Arc::new(RwLock::new(0)),
            table_auto_increment: Arc::new(DashMap::new()),
            num_partitions,
            txn_coordinator,
            version_store,
            pending_updates: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            pending_spatial_updates: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            vector_indexes: Arc::new(DashMap::new()),
            spatial_indexes: Arc::new(DashMap::new()),
            text_indexes: Arc::new(DashMap::new()),
            column_indexes: Arc::new(DashMap::new()),
            table_registry,
            index_registry,
            row_cache,
            index_update_strategy: config.index_update_strategy.clone(),
            query_timeout_secs: config.query_timeout_secs,
            is_flushing: Arc::new(AtomicBool::new(false)),
            checkpoint_mutex: Arc::new(Mutex::new(())),
            auto_checkpoint_thread: None,
        };
        
        // 🚀 Unified Flush Callback: DISABLED for v0.1.2 release.
        //
        // The callback blocks the background flush thread during index building,
        // which can cause deadlocks when checkpoint() waits for queue drain
        // while the background thread is stuck in the callback.
        //
        // Index building is handled by checkpoint() → rebuild_timestamp_index() + flush_all_indexes()
        // This is safe because checkpoint() is always called before close/drop.
        //
        // TODO: Re-enable with async callback or separate index-builder thread pool in v0.2.0
        //
        // let db_clone = db.clone_for_callback();
        // lsm_engine.set_flush_callback(move |memtable| {
        //     db_clone.batch_build_indexes_from_flush(memtable)
        // })?;
        
        // 🚀 Start auto-checkpoint thread if enabled
        let auto_checkpoint_thread = if let Some(auto_config) = config.auto_checkpoint {
            Some(Self::start_auto_checkpoint_thread(
                db.clone_for_callback(),
                auto_config,
            ))
        } else {
            None
        };
        
        // Update db with the thread handle
        let mut db = db;
        db.auto_checkpoint_thread = auto_checkpoint_thread;
        
        Ok(db)
    }
    
    /// Clone self for callback (only what's needed)
    pub(crate) fn clone_for_callback(&self) -> Self {
        Self {
            path: self.path.clone(),
            wal: self.wal.clone(),
            lsm_engine: self.lsm_engine.clone(),
            timestamp_index: self.timestamp_index.clone(),
            next_row_id: self.next_row_id.clone(),
            table_auto_increment: self.table_auto_increment.clone(),  // 🚀 Phase 4
            num_partitions: self.num_partitions,
            txn_coordinator: self.txn_coordinator.clone(),
            version_store: self.version_store.clone(),
            pending_updates: self.pending_updates.clone(),
            pending_spatial_updates: self.pending_spatial_updates.clone(),
            vector_indexes: self.vector_indexes.clone(),
            spatial_indexes: self.spatial_indexes.clone(),
            text_indexes: self.text_indexes.clone(),
            column_indexes: self.column_indexes.clone(),
            table_registry: self.table_registry.clone(),
            index_registry: self.index_registry.clone(),  // 🆕
            row_cache: self.row_cache.clone(),
            index_update_strategy: self.index_update_strategy.clone(),
            query_timeout_secs: self.query_timeout_secs,  // 🚀 P0
            is_flushing: self.is_flushing.clone(),  // 🆕 共享 flush 标志
            checkpoint_mutex: self.checkpoint_mutex.clone(),
            auto_checkpoint_thread: None,  // Don't clone thread (only owned by original)
        }
    }

    /// Open an existing database
    pub fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
        let path = path.as_ref();
        let db_path = path.with_extension("mote");
        
        // 🎯 统一目录结构：从 {name}.mote/ 目录读取
        let wal_path = db_path.join("wal");
        let lsm_dir = db_path.join("lsm");
        let indexes_dir = db_path.join("indexes");

        // Default number of partitions
        let num_partitions = 4;

        // Open or create WAL
        let wal = if wal_path.exists() {
            Arc::new(WALManager::open(&wal_path, num_partitions)?)
        } else {
            std::fs::create_dir_all(&wal_path)?;
            Arc::new(WALManager::create(&wal_path, num_partitions)?)
        };

        // Recover from WAL
        let recovered_records = wal.recover()?;
        
        // Open timestamp index with BTree storage (从 indexes/ 目录)
        std::fs::create_dir_all(&indexes_dir)?;
        let timestamp_storage = indexes_dir.join("timestamp.idx");
        let btree_config = BTreeConfig {
            unique_keys: false,
            allow_updates: true,
            ..Default::default()
        };
        let mut timestamp_idx = BTree::with_config(timestamp_storage, btree_config)?;
        
        // Get total entries from timestamp index (already persisted data)
        let persisted_count = timestamp_idx.len();
        
        let mut max_row_id = if persisted_count > 0 {
            // Estimate max_row_id from persisted count
            // Since row_ids are sequential starting from 0, max is count-1
            (persisted_count - 1) as u64
        } else {
            0
        };

        // Replay WAL records (if any uncommitted changes after last checkpoint)
        for records in recovered_records.values() {
            for record in records {
                if let WALRecord::Insert { row_id, data, .. } = record {
                    max_row_id = max_row_id.max(*row_id);
                    
                    // Also insert into timestamp index
                    if let Some(crate::types::Value::Timestamp(ts)) = data.first() {
                        let _ = timestamp_idx.insert(ts.as_micros() as u64, *row_id);
                    }
                }
            }
        }

        let timestamp_index = Arc::new(RwLock::new(timestamp_idx));

        // Open LSM-Tree storage engine
        std::fs::create_dir_all(&lsm_dir)?;
        let lsm_engine = Arc::new(LSMEngine::new(lsm_dir, LSMConfig::default())?);

        // Load table registry BEFORE WAL replay so we can resolve table_name → table_id
        // for correct composite key construction.
        let table_registry = Arc::new(TableRegistry::new(&db_path)?);
        table_registry.ensure_default_table_id()?;

        // Replay WAL records into LSM Engine using stable table_id
        debug_log!("[database] 恢复 WAL 记录到 LSM Engine...");
        let mut recovered_count = 0;
        for records in recovered_records.values() {
            for record in records {
                match record {
                    WALRecord::Insert { table_name, row_id, data, .. } => {
                        // Use table_id for composite_key (same as make_composite_key)
                        let table_id = table_registry.get_table_id(table_name)
                            .unwrap_or(0);
                        let composite_key = ((table_id as u64) << 32) | (*row_id & 0xFFFFFFFF);

                        let row_data = bincode::serialize(data)?;
                        let value = crate::storage::lsm::Value::new(row_data, composite_key);
                        lsm_engine.put(composite_key, value)?;
                        recovered_count += 1;
                    }
                    WALRecord::Update { table_name, row_id, new_data, .. } => {
                        let table_id = table_registry.get_table_id(table_name)
                            .unwrap_or(0);
                        let composite_key = ((table_id as u64) << 32) | (*row_id & 0xFFFFFFFF);

                        let row_data = bincode::serialize(new_data)?;
                        let value = crate::storage::lsm::Value::new(row_data, composite_key);
                        lsm_engine.put(composite_key, value)?;
                        recovered_count += 1;
                    }
                    WALRecord::Delete { table_name, row_id, timestamp, .. } => {
                        let table_id = table_registry.get_table_id(table_name)
                            .unwrap_or(0);
                        let composite_key = ((table_id as u64) << 32) | (*row_id & 0xFFFFFFFF);

                        lsm_engine.delete(composite_key, *timestamp)?;
                        recovered_count += 1;
                    }
                    _ => {}
                }
            }
        }
        debug_log!("[database] WAL 恢复完成，恢复了 {} 条记录", recovered_count);

        // Create version store and transaction coordinator
        let version_store = Arc::new(VersionStore::new());
        let txn_coordinator = Arc::new(TransactionCoordinator::new(version_store.clone()));

        // Load existing vector indexes
        let vector_indexes = Self::load_vector_indexes(&db_path)?;
        
        // Load existing spatial indexes
        let spatial_indexes = Self::load_spatial_indexes(&db_path)?;
        
        // Load existing text indexes
        let text_indexes = Self::load_text_indexes(&db_path)?;
        
        // 🆕 Load index metadata registry
        // (table_registry already loaded above, before WAL replay)
        let index_registry = Arc::new(crate::database::index_metadata::IndexRegistry::new(&db_path));
        let _ = index_registry.load();  // Ignore error if file doesn't exist
        
        // 🚀 P1: Create row cache (default 10000 rows)
        let row_cache = Arc::new(RowCache::new(10000));

        let db = Self {
            path: db_path,
            wal,
            lsm_engine: lsm_engine.clone(),
            timestamp_index,
            next_row_id: Arc::new(RwLock::new(max_row_id + 1)),
            table_auto_increment: Arc::new(DashMap::new()),
            num_partitions,
            txn_coordinator,
            version_store,
            pending_updates: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            pending_spatial_updates: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            vector_indexes: Arc::new(Self::hashmap_to_dashmap(vector_indexes)),
            spatial_indexes: Arc::new(Self::hashmap_to_dashmap(spatial_indexes)),
            text_indexes: Arc::new(Self::hashmap_to_dashmap(text_indexes)),
            column_indexes: Arc::new(DashMap::new()),
            table_registry,
            index_registry,
            row_cache,
            index_update_strategy: crate::config::IndexUpdateStrategy::default(),
            query_timeout_secs: None,
            is_flushing: Arc::new(AtomicBool::new(false)),
            checkpoint_mutex: Arc::new(Mutex::new(())),
            auto_checkpoint_thread: None,
        };

        // ⚠️ Flush callback DISABLED — it blocks the background flush thread,
        // causing deadlocks when checkpoint() waits for queue drain.
        // Index building is handled by checkpoint() → rebuild_timestamp_index() instead.
        // See: helpers.rs batch_build_indexes_from_flush (kept for future use)
        //
        // let db_clone = db.clone_for_callback();
        // lsm_engine.set_flush_callback(move |memtable| {
        //     db_clone.batch_build_indexes_from_flush(memtable)
        // })?;

        // 🚀 Start auto-checkpoint thread (using default config on open)
        let auto_checkpoint_thread = Some(Self::start_auto_checkpoint_thread(
            db.clone_for_callback(),
            crate::config::AutoCheckpointConfig::default(),
        ));
        
        // Update db with the thread handle
        let mut db = db;
        db.auto_checkpoint_thread = auto_checkpoint_thread;
        
        // 🚀 Phase 5: Recover AUTO_INCREMENT counters (B3: Crash Recovery)
        // For each table with AUTO_INCREMENT, find max ID from LSM and initialize counter
        for table_name in db.table_registry.list_tables()? {
            let schema = db.table_registry.get_table(&table_name)?;
            if schema.is_primary_key_auto_increment() {
                let max_id = db.recover_auto_increment_counter(&table_name, &schema)?;
                debug_log!("[database] 🔄 Recovered AUTO_INCREMENT counter for '{}': next_id = {}", 
                    table_name, max_id + 1);
                
                db.table_auto_increment.insert(
                    table_name,
                    Arc::new(RwLock::new(max_id + 1))
                );
            }
        }
        
        Ok(db)
    }
    
    /// 🚀 Helper: Convert HashMap to DashMap
    fn hashmap_to_dashmap<K: std::hash::Hash + Eq, V>(map: HashMap<K, V>) -> DashMap<K, V> {
        let dashmap = DashMap::new();
        for (k, v) in map {
            dashmap.insert(k, v);
        }
        dashmap
    }
    
    /// 🆕 Set AUTO_INCREMENT value for a table
    /// 
    /// # Arguments
    /// * `table_name` - Table name
    /// * `new_value` - New AUTO_INCREMENT starting value
    /// 
    /// # Errors
    /// Returns error if table doesn't exist or doesn't have AUTO_INCREMENT
    pub fn set_auto_increment_value(&self, table_name: &str, new_value: i64) -> Result<()> {
        // Verify table has AUTO_INCREMENT
        if !self.table_auto_increment.contains_key(table_name) {
            return Err(MoteDBError::InvalidArgument(
                format!("Table {} does not have AUTO_INCREMENT", table_name)
            ));
        }
        
        // Update counter
        if let Some(counter_ref) = self.table_auto_increment.get(table_name) {
            let mut counter = counter_ref.write();
            *counter = new_value;
            debug_log!("[database] ✓ Set AUTO_INCREMENT for '{}' to {}", table_name, new_value);
        }
        
        Ok(())
    }
    
    /// Load existing vector indexes from disk
    fn load_vector_indexes(db_path: &Path) -> Result<HashMap<String, Arc<RwLock<DiskANNIndex>>>> {
        let mut indexes = HashMap::new();
        
        // 🎯 从统一目录加载：{db}.mote/indexes/vector_*/
        let indexes_dir = db_path.join("indexes");
        if indexes_dir.exists() {
            if let Ok(entries) = std::fs::read_dir(&indexes_dir) {
                for entry in entries.flatten() {
                    if let Ok(name) = entry.file_name().into_string() {
                        if name.starts_with("vector_") {
                            let index_name = name.strip_prefix("vector_").unwrap();
                            let index_path = entry.path();
                            
                            // Try to load the index
                            let config = VamanaConfig::default();
                            if let Ok(index) = DiskANNIndex::load(&index_path, config) {
                                indexes.insert(
                                    index_name.to_string(), 
                                    Arc::new(RwLock::new(index))
                                );
                                println!("[MoteDB] Loaded vector index: {}", index_name);
                            }
                        }
                    }
                }
            }
        }
        
        Ok(indexes)
    }
    
    /// Load existing spatial indexes from disk
    fn load_spatial_indexes(db_path: &Path) -> Result<HashMap<String, Arc<RwLock<SpatialHybridIndex>>>> {
        let mut indexes = HashMap::new();
        
        // 🎯 从统一目录加载：{db}.mote/indexes/spatial_*/
        let indexes_dir = db_path.join("indexes");
        if indexes_dir.exists() {
            if let Ok(entries) = std::fs::read_dir(&indexes_dir) {
                for entry in entries.flatten() {
                    if let Ok(name) = entry.file_name().into_string() {
                        if name.starts_with("spatial_") {
                            let index_name = name.strip_prefix("spatial_").unwrap();
                            let index_path = entry.path();
                            
                            // Try to load with default config (will use saved config from metadata)
                            let default_config = SpatialHybridConfig::new(
                                BoundingBoxF32::new(0.0, 0.0, 1000.0, 1000.0)
                            ).with_mmap(true, Some(index_path.clone()));
                            
                            if let Ok(index) = SpatialHybridIndex::load(&index_path, default_config) {
                                indexes.insert(
                                    index_name.to_string(),
                                    Arc::new(RwLock::new(index))
                                );
                                println!("[MoteDB] Loaded spatial index: {}", index_name);
                            }
                        }
                    }
                }
            }
        }
        
        Ok(indexes)
    }
    
    /// Load existing text indexes from disk
    fn load_text_indexes(db_path: &Path) -> Result<HashMap<String, Arc<RwLock<TextFTSIndex>>>> {
        let mut indexes = HashMap::new();
        
        // 🧹 Clean up legacy text_indexes_metadata.bin (no longer used)
        let legacy_metadata_path = db_path.join("text_indexes_metadata.bin");
        if legacy_metadata_path.exists() {
            if let Err(e) = std::fs::remove_file(&legacy_metadata_path) {
                eprintln!("⚠️ Failed to remove legacy text_indexes_metadata.bin: {}", e);
            } else {
                println!("[MoteDB] 🧹 Removed legacy text_indexes_metadata.bin (replaced by index_metadata.bin)");
            }
        }
        
        // 🎯 从统一目录加载：{db}.mote/indexes/text_*/
        let indexes_dir = db_path.join("indexes");
        if indexes_dir.exists() {
            if let Ok(entries) = std::fs::read_dir(&indexes_dir) {
                for entry in entries.flatten() {
                    if let Ok(name) = entry.file_name().into_string() {
                        if name.starts_with("text_") {
                            let index_name = name.strip_prefix("text_").unwrap();
                            let index_path = entry.path();
                            
                            // Try to load the index
                            if let Ok(index) = TextFTSIndex::new(index_path) {
                                indexes.insert(
                                    index_name.to_string(),
                                    Arc::new(RwLock::new(index))
                                );
                                println!("[MoteDB] Loaded text index: {}", index_name);
                            }
                        }
                    }
                }
            }
        }
        
        Ok(indexes)
    }
    
    /// Start auto-checkpoint background thread
    /// 
    /// 🚀 Optimized for embedded environments:
    /// 1. Lazy-checking: Only checks WAL size when interval reached (no unnecessary fs calls)
    /// 2. Adaptive sleep: Longer intervals in low-activity periods
    /// 3. Zero allocation in hot path
    /// 4. Minimal CPU usage: < 0.1% CPU overhead
    fn start_auto_checkpoint_thread(
        db: Self,
        config: crate::config::AutoCheckpointConfig,
    ) -> AutoCheckpointThread {
        use std::time::{Duration, Instant};
        
        let should_stop = Arc::new(AtomicBool::new(false));
        let should_stop_clone = should_stop.clone();
        
        let handle = std::thread::spawn(move || {
            let mut last_checkpoint = Instant::now();
            
            // 🚀 Adaptive check interval:
            // - Start with min_interval (avoid too-frequent checks)
            // - Only check WAL size when interval reached
            let check_interval = Duration::from_secs(config.min_interval_secs.max(10));
            
            debug_log!("[AutoCheckpoint] 🚀 Background thread started (embedded-optimized)");
            debug_log!("[AutoCheckpoint] Config: max_wal={}MB, interval={}s, check_every={}s",
                config.max_wal_size_bytes / 1024 / 1024, 
                config.min_interval_secs,
                check_interval.as_secs());
            
            while !should_stop_clone.load(std::sync::atomic::Ordering::Relaxed) {
                // 🚀 **CRITICAL FIX**: Use interruptible sleep (check every 1s)
                // This allows fast shutdown when Drop is called
                // 
                // Before: sleep(60s) -> Drop waits 60s
                // After: sleep(1s) × 60 -> Drop waits max 1s
                let mut remaining = check_interval;
                while remaining > Duration::ZERO {
                    if should_stop_clone.load(std::sync::atomic::Ordering::Relaxed) {
                        debug_log!("[AutoCheckpoint] 🛑 Shutdown signal received during sleep");
                        break;
                    }
                    
                    let sleep_chunk = Duration::from_secs(1).min(remaining);
                    std::thread::sleep(sleep_chunk);
                    remaining = remaining.saturating_sub(sleep_chunk);
                }
                
                // Check if stop signal was set during sleep
                if should_stop_clone.load(std::sync::atomic::Ordering::Relaxed) {
                    break;
                }
                
                // 🚀 Only check WAL size when enough time has passed
                // (avoids unnecessary filesystem calls)
                let elapsed = last_checkpoint.elapsed();
                if elapsed.as_secs() < config.min_interval_secs {
                    continue;
                }
                
                // 🚀 Lazy WAL size check - only when needed
                let wal_dir = db.path.join("wal");
                match get_directory_size(&wal_dir) {
                    Ok(wal_size) if wal_size >= config.max_wal_size_bytes => {
                        debug_log!("[AutoCheckpoint] 🔔 Trigger: WAL {}MB >= {}MB",
                            wal_size / 1024 / 1024, config.max_wal_size_bytes / 1024 / 1024);
                        
                        // Trigger checkpoint
                        if let Err(e) = db.checkpoint() {
                            eprintln!("[AutoCheckpoint] ⚠️  Checkpoint failed: {:?}", e);
                        } else {
                            debug_log!("[AutoCheckpoint] ✅ Checkpoint complete");
                            last_checkpoint = Instant::now();
                        }
                    }
                    Ok(_) => {
                        // WAL size below threshold, skip checkpoint
                    }
                    Err(e) => {
                        debug_log!("[AutoCheckpoint] ⚠️  Failed to check WAL size: {:?}", e);
                    }
                }
            }
            
            debug_log!("[AutoCheckpoint] 👋 Background thread stopped");
        });
        
        AutoCheckpointThread {
            handle: Some(handle),
            should_stop,
        }
    }
    
    /// 🚀 Phase 5: Recover AUTO_INCREMENT counter from existing data (B3)
    /// 
    /// Scans table rows to find maximum AUTO_INCREMENT ID value.
    /// This is called during database open() to restore counter state after crash.
    fn recover_auto_increment_counter(
        &self,
        table_name: &str,
        schema: &crate::types::TableSchema,
    ) -> Result<i64> {
        use crate::types::Value;
        
        // Get primary key column position
        let pk_col_name = schema.primary_key()
            .ok_or_else(|| StorageError::InvalidData(
                format!("Table '{}' has no primary key", table_name)
            ))?;
        let pk_col = schema.get_column(pk_col_name)
            .ok_or_else(|| StorageError::ColumnNotFound(pk_col_name.to_string()))?;
        
        // Start from schema's start value - 1 (will be incremented to start value if no data)
        let mut max_id = schema.get_auto_increment_start() - 1;
        
        // Scan all rows in the table (streaming to avoid loading entire table into memory)
        match self.scan_table_rows_streaming(table_name) {
            Ok(iter) => {
                for result in iter {
                    match result {
                        Ok((_row_id, row)) => {
                            if let Some(Value::Integer(id)) = row.get(pk_col.position) {
                                max_id = max_id.max(*id);
                            }
                        }
                        Err(e) => {
                            debug_log!("[database] Warning: Error during AUTO_INCREMENT scan: {:?}", e);
                            break;
                        }
                    }
                }
            }
            Err(e) => {
                // Table might be empty or not exist yet (first time opening)
                debug_log!("[database] Warning: Failed to scan table '{}' for AUTO_INCREMENT recovery: {:?}",
                    table_name, e);
            }
        }
        
        Ok(max_id)
    }
}

/// Get total size of all files in a directory
fn get_directory_size(dir: &std::path::Path) -> Result<u64> {
    let mut total = 0;
    
    if !dir.exists() {
        return Ok(0);
    }
    
    for entry in std::fs::read_dir(dir)? {
        let entry = entry?;
        let metadata = entry.metadata()?;
        if metadata.is_file() {
            total += metadata.len();
        }
    }
    
    Ok(total)
}

/// Automatic cleanup when database is dropped
/// 
/// This ensures proper shutdown:
/// 1. Flush all in-memory data (MemTable → SSTable)
/// 2. Persist all indexes
/// 3. Checkpoint WAL (truncate log files)
/// 
/// This prevents WAL files from accumulating indefinitely and ensures
/// clean shutdown even if user forgets to call checkpoint().
impl Drop for MoteDB {
    fn drop(&mut self) {
        // 🛑 Step 1: Stop auto-checkpoint thread first
        if let Some(mut thread) = self.auto_checkpoint_thread.take() {
            debug_log!("[MoteDB::Drop] 🛑 Stopping auto-checkpoint thread...");
            thread.should_stop.store(true, std::sync::atomic::Ordering::Relaxed);
            if let Some(handle) = thread.handle.take() {
                let _ = handle.join();
            }
            debug_log!("[MoteDB::Drop] ✅ Auto-checkpoint thread stopped");
        }
        
        // ⚠️ CRITICAL: Always checkpoint on drop to:
        // 1. Persist all data safely
        // 2. Truncate WAL files (prevent accumulation)
        // 3. Ensure clean shutdown
        
        debug_log!("[MoteDB::Drop] 🚪 Database closing, performing final checkpoint...");
        
        // Ignore errors during drop (logging only)
        // We're shutting down anyway, and panic in drop() is dangerous
        if let Err(e) = self.checkpoint() {
            eprintln!("[MoteDB::Drop] ⚠️  Failed to checkpoint during drop: {:?}", e);
            eprintln!("[MoteDB::Drop] ⚠️  WAL files may not be cleaned up");
        } else {
            debug_log!("[MoteDB::Drop] ✅ Final checkpoint complete, WAL cleaned");
        }
        
        debug_log!("[MoteDB::Drop] 👋 Database closed cleanly");
    }
}