stoolap 0.4.0

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

//! Database struct and operations
//!
//! Provides a modern, ergonomic Rust API for database operations.
//!
//! # Examples
//!
//! ```ignore
//! use stoolap::{Database, params};
//!
//! let db = Database::open("memory://")?;
//!
//! // DDL - no params needed
//! db.execute("CREATE TABLE users (id INTEGER, name TEXT, age INTEGER)", ())?;
//!
//! // Insert with params - using tuple syntax
//! db.execute("INSERT INTO users VALUES ($1, $2, $3)", (1, "Alice", 30))?;
//!
//! // Insert with params! macro
//! db.execute("INSERT INTO users VALUES ($1, $2, $3)", params![2, "Bob", 25])?;
//!
//! // Query with iteration
//! for row in db.query("SELECT * FROM users WHERE age > $1", (20,))? {
//!     let row = row?;
//!     let name: String = row.get("name")?;
//!     println!("{}", name);
//! }
//!
//! // Query single value
//! let count: i64 = db.query_one("SELECT COUNT(*) FROM users", ())?;
//! ```

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

use crate::core::{DataType, Error, IsolationLevel, Result, Value};
use crate::executor::context::ExecutionContextBuilder;
use crate::executor::{CachedPlanRef, ExecutionContext, Executor};
use crate::storage::mvcc::engine::MVCCEngine;
use crate::storage::traits::Engine;
use crate::storage::{Config, SyncMode};

use super::params::{NamedParams, Params};
use super::rows::{FromRow, Rows};
use super::statement::Statement;
use super::transaction::Transaction;

/// Storage scheme constants
pub const MEMORY_SCHEME: &str = "memory";
pub const FILE_SCHEME: &str = "file";

/// Global database registry to ensure single instance per DSN
static DATABASE_REGISTRY: std::sync::LazyLock<RwLock<FxHashMap<String, Arc<DatabaseInner>>>> =
    std::sync::LazyLock::new(|| RwLock::new(FxHashMap::default()));

/// Inner database state (shared between Database instances with same DSN)
pub(crate) struct DatabaseInner {
    engine: Arc<MVCCEngine>,
    executor: Mutex<Executor>,
    dsn: String,
    /// Whether this DatabaseInner owns the engine (created it via open()).
    /// Cloned DatabaseInners share the engine but don't own it.
    owns_engine: bool,
    /// Temp directory for test-filedb feature. Deleted on drop.
    #[cfg(feature = "test-filedb")]
    _temp_dir: Option<tempfile::TempDir>,
}

/// Type alias for Statement to use (avoids exposing DatabaseInner directly)
pub(crate) type DatabaseInnerHandle = DatabaseInner;

impl Drop for DatabaseInner {
    fn drop(&mut self) {
        // Clear all thread-local caches to release references to engine internals
        // This prevents memory leaks from cached Arc<dyn Index> and closures
        crate::executor::clear_all_thread_local_caches();

        // Only close the engine if we own it (created via open(), not clone()).
        // Cloned databases share the engine but don't close it on drop.
        if self.owns_engine {
            let _ = self.engine.close_engine();
        }
    }
}

/// Database represents a Stoolap database connection
///
/// This is the main entry point for using Stoolap. It wraps the storage engine
/// and executor, providing a simple API for executing SQL queries.
///
/// # Thread Safety
///
/// Database is thread-safe and can be shared across threads via cloning.
/// Each clone shares the same underlying storage engine.
///
/// # Examples
///
/// ```ignore
/// use stoolap::{Database, params};
///
/// // Open in-memory database
/// let db = Database::open("memory://")?;
///
/// // Create table
/// db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)", ())?;
///
/// // Insert with parameters
/// db.execute("INSERT INTO users VALUES ($1, $2)", (1, "Alice"))?;
///
/// // Query
/// for row in db.query("SELECT * FROM users", ())? {
///     let row = row?;
///     println!("{}: {}", row.get::<i64>("id")?, row.get::<String>("name")?);
/// }
/// ```
pub struct Database {
    inner: Arc<DatabaseInner>,
}

#[cfg(feature = "ffi")]
impl Database {
    /// Returns an Arc reference to the inner state, preventing the engine
    /// from being closed while any keepalive handle exists.
    ///
    /// Used by the FFI layer to ensure cloned handles keep the original
    /// engine-owning DatabaseInner alive.
    pub(crate) fn keepalive(&self) -> Arc<DatabaseInner> {
        Arc::clone(&self.inner)
    }

    /// Returns a borrow of the inner Arc (no clone, no count change).
    pub(crate) fn inner_arc(&self) -> &Arc<DatabaseInner> {
        &self.inner
    }

    /// Try to remove `inner` from the global registry.
    ///
    /// Only removes the entry when:
    /// 1. The registry entry points to the same `DatabaseInner` (`ptr_eq`), and
    /// 2. `strong_count == 2`: the caller's reference + registry are the only
    ///    remaining references. After the caller drops, only the registry holds
    ///    a reference, so it is safe to clean up.
    ///
    /// This correctly handles both shared-DSN scenarios (two `stoolap_open()`
    /// calls with the same DSN) and clone scenarios (keepalive Arcs).
    pub(crate) fn try_unregister_arc(inner: &Arc<DatabaseInner>) {
        if let Ok(mut registry) = DATABASE_REGISTRY.write() {
            if let Some(entry) = registry.get(&inner.dsn) {
                if Arc::ptr_eq(entry, inner) && Arc::strong_count(inner) == 2 {
                    registry.remove(&inner.dsn);
                }
            }
        }
    }
}

impl Clone for Database {
    /// Clone the database handle.
    ///
    /// Each cloned handle has its own executor with independent transaction state,
    /// but shares the same underlying storage engine. This ensures proper transaction
    /// isolation - a BEGIN on one handle won't affect reads on another handle.
    fn clone(&self) -> Self {
        // Create a new executor with the same engine (shares data) but independent
        // transaction state (no dirty reads across handles)
        let engine = Arc::clone(&self.inner.engine);
        let executor = crate::executor::Executor::new(Arc::clone(&engine));

        let inner = Arc::new(DatabaseInner {
            engine,
            executor: Mutex::new(executor),
            dsn: self.inner.dsn.clone(),
            owns_engine: false,
            #[cfg(feature = "test-filedb")]
            _temp_dir: None, // Clones don't own the temp dir
        });

        Database { inner }
    }
}

impl Drop for Database {
    fn drop(&mut self) {
        // Only remove from registry if:
        // 1. The registry entry is OUR Arc (ptr_eq check) - prevents open_in_memory()
        //    from accidentally removing a different DatabaseInner with same DSN
        // 2. We're the last non-registry holder (strong_count == 2)
        if let Ok(mut registry) = DATABASE_REGISTRY.write() {
            if let Some(entry) = registry.get(&self.inner.dsn) {
                if Arc::ptr_eq(entry, &self.inner) && Arc::strong_count(&self.inner) == 2 {
                    registry.remove(&self.inner.dsn);
                }
            }
        }
        // Note: Thread-local cache clearing and engine closure happen in DatabaseInner::drop()
        // when the last Arc<DatabaseInner> is dropped.
    }
}

impl Database {
    /// Open a database connection
    ///
    /// The DSN (Data Source Name) specifies the database location:
    /// - `memory://` - In-memory database (data lost when closed)
    /// - `file:///path/to/db` - Persistent database at the specified path
    ///
    /// # Examples
    ///
    /// ```ignore
    /// // In-memory database
    /// let db = Database::open("memory://")?;
    ///
    /// // Persistent database
    /// let db = Database::open("file:///tmp/mydb")?;
    /// ```
    ///
    /// # Engine Reuse
    ///
    /// Opening the same DSN multiple times returns the same engine instance.
    /// This ensures consistency and prevents data corruption.
    pub fn open(dsn: &str) -> Result<Self> {
        // Check if we already have an engine for this DSN
        {
            let registry = DATABASE_REGISTRY
                .read()
                .map_err(|_| Error::LockAcquisitionFailed("registry read".to_string()))?;
            if let Some(inner) = registry.get(dsn) {
                return Ok(Database {
                    inner: Arc::clone(inner),
                });
            }
        }

        // Need to create a new engine - acquire write lock
        let mut registry = DATABASE_REGISTRY
            .write()
            .map_err(|_| Error::LockAcquisitionFailed("registry write".to_string()))?;

        // Double-check after acquiring write lock
        if let Some(inner) = registry.get(dsn) {
            return Ok(Database {
                inner: Arc::clone(inner),
            });
        }

        // Parse the DSN
        let (scheme, path) = Self::parse_dsn(dsn)?;

        // test-filedb: track temp dir so it lives as long as the engine
        #[cfg(feature = "test-filedb")]
        let mut _temp_dir_holder: Option<tempfile::TempDir> = None;

        // Create the engine based on scheme
        let engine = match scheme.as_str() {
            MEMORY_SCHEME => {
                #[cfg(feature = "test-filedb")]
                {
                    let tmp = tempfile::tempdir().map_err(|e| {
                        Error::internal(format!("failed to create temp dir: {}", e))
                    })?;
                    let file_dsn = format!("file://{}", tmp.path().display());
                    let (_clean_path, config) = Self::parse_file_config(&file_dsn[7..])?;
                    let engine = MVCCEngine::new(config);
                    engine.open_engine()?;
                    let engine = Arc::new(engine);
                    engine.start_cleanup();
                    _temp_dir_holder = Some(tmp);
                    engine
                }
                #[cfg(not(feature = "test-filedb"))]
                {
                    let engine = MVCCEngine::in_memory();
                    engine.open_engine()?;
                    let engine = Arc::new(engine);
                    engine.start_cleanup();
                    engine
                }
            }
            FILE_SCHEME => {
                // Parse optional query parameters
                let (_clean_path, config) = Self::parse_file_config(&path)?;

                let engine = MVCCEngine::new(config);
                engine.open_engine()?;
                let engine = Arc::new(engine);
                // Start background cleanup (uses config from engine)
                engine.start_cleanup();
                engine
            }
            _ => {
                return Err(Error::parse(format!(
                    "Unsupported scheme '{}'. Use 'memory://' or 'file://path'",
                    scheme
                )));
            }
        };

        // Create executor (uses shared default function registry)
        let executor = Executor::new(Arc::clone(&engine));

        let inner = Arc::new(DatabaseInner {
            engine,
            executor: Mutex::new(executor),
            dsn: dsn.to_string(),
            owns_engine: true,
            #[cfg(feature = "test-filedb")]
            _temp_dir: _temp_dir_holder,
        });

        // Store in registry
        registry.insert(dsn.to_string(), Arc::clone(&inner));

        Ok(Database { inner })
    }

    /// Open an in-memory database
    ///
    /// This is a convenience method that creates a new in-memory database.
    /// Each call creates a unique instance (unlike `open("memory://")` which
    /// would share the same instance).
    pub fn open_in_memory() -> Result<Self> {
        Self::create_in_memory_engine()
    }

    #[cfg(feature = "test-filedb")]
    fn create_in_memory_engine() -> Result<Self> {
        let tmp = tempfile::tempdir()
            .map_err(|e| Error::internal(format!("failed to create temp dir: {}", e)))?;
        let file_dsn = format!("file://{}", tmp.path().display());
        let (_clean_path, config) = Self::parse_file_config(&file_dsn[7..])?;
        let engine = MVCCEngine::new(config);
        engine.open_engine()?;
        let engine = Arc::new(engine);
        engine.start_cleanup();
        let executor = Executor::new(Arc::clone(&engine));
        let inner = Arc::new(DatabaseInner {
            engine,
            executor: Mutex::new(executor),
            dsn: "memory://".to_string(),
            owns_engine: true,
            _temp_dir: Some(tmp),
        });
        Ok(Database { inner })
    }

    #[cfg(not(feature = "test-filedb"))]
    fn create_in_memory_engine() -> Result<Self> {
        let engine = MVCCEngine::in_memory();
        engine.open_engine()?;
        let engine = Arc::new(engine);
        engine.start_cleanup();
        let executor = Executor::new(Arc::clone(&engine));
        let inner = Arc::new(DatabaseInner {
            engine,
            executor: Mutex::new(executor),
            dsn: "memory://".to_string(),
            owns_engine: true,
        });
        Ok(Database { inner })
    }

    /// Parse a DSN into scheme and path
    fn parse_dsn(dsn: &str) -> Result<(String, String)> {
        let idx = dsn
            .find("://")
            .ok_or_else(|| Error::parse("Invalid DSN format: expected scheme://path"))?;

        let scheme = dsn[..idx].to_lowercase();
        let path = dsn[idx + 3..].to_string();

        // Validate scheme
        match scheme.as_str() {
            MEMORY_SCHEME | FILE_SCHEME => {}
            _ => {
                return Err(Error::parse(format!(
                    "Unsupported scheme '{}'. Use 'memory://' or 'file://path'",
                    scheme
                )));
            }
        }

        // Validate file path
        if scheme == FILE_SCHEME {
            let clean_path = if path.contains('?') {
                &path[..path.find('?').unwrap()]
            } else {
                &path
            };

            if clean_path.is_empty() {
                return Err(Error::parse("file:// scheme requires a non-empty path"));
            }
        }

        Ok((scheme, path))
    }

    /// Parse file:// config from query parameters
    fn parse_file_config(path: &str) -> Result<(String, Config)> {
        let (clean_path, query) = if let Some(idx) = path.find('?') {
            (path[..idx].to_string(), Some(&path[idx + 1..]))
        } else {
            (path.to_string(), None)
        };

        let mut config = Config::with_path(&clean_path);

        // Parse query parameters
        if let Some(query) = query {
            for param in query.split('&') {
                let mut parts = param.splitn(2, '=');
                let key = parts.next().unwrap_or("");
                let value = parts.next().unwrap_or("");

                match key {
                    // Sync mode: sync=none|normal|full
                    "sync_mode" | "sync" => {
                        config.persistence.sync_mode = match value.to_lowercase().as_str() {
                            "none" | "off" | "0" => SyncMode::None,
                            "normal" | "1" => SyncMode::Normal,
                            "full" | "2" => SyncMode::Full,
                            _ => SyncMode::Normal,
                        };
                    }
                    // Checkpoint interval in seconds: checkpoint_interval=60
                    // Also accepts snapshot_interval for backward compatibility
                    "checkpoint_interval" | "snapshot_interval" => {
                        config.persistence.checkpoint_interval =
                            value.parse::<u32>().map_err(|_| {
                                Error::invalid_argument(format!(
                                    "invalid checkpoint_interval: '{}'",
                                    value
                                ))
                            })?;
                    }
                    // Compaction threshold: compact_threshold=4
                    "compact_threshold" => {
                        config.persistence.compact_threshold =
                            value.parse::<u32>().map_err(|_| {
                                Error::invalid_argument(format!(
                                    "invalid compact_threshold: '{}'",
                                    value
                                ))
                            })?;
                    }
                    // Number of backup snapshots to keep: keep_snapshots=3
                    "keep_snapshots" => {
                        config.persistence.keep_snapshots = value.parse::<u32>().map_err(|_| {
                            Error::invalid_argument(format!("invalid keep_snapshots: '{}'", value))
                        })?;
                    }
                    // WAL flush trigger in bytes: wal_flush_trigger=32768
                    "wal_flush_trigger" => {
                        config.persistence.wal_flush_trigger =
                            value.parse::<usize>().map_err(|_| {
                                Error::invalid_argument(format!(
                                    "invalid wal_flush_trigger: '{}'",
                                    value
                                ))
                            })?;
                    }
                    // WAL buffer size in bytes: wal_buffer_size=65536
                    "wal_buffer_size" => {
                        config.persistence.wal_buffer_size =
                            value.parse::<usize>().map_err(|_| {
                                Error::invalid_argument(format!(
                                    "invalid wal_buffer_size: '{}'",
                                    value
                                ))
                            })?;
                    }
                    // WAL max size in bytes: wal_max_size=67108864
                    "wal_max_size" => {
                        config.persistence.wal_max_size = value.parse::<usize>().map_err(|_| {
                            Error::invalid_argument(format!("invalid wal_max_size: '{}'", value))
                        })?;
                    }
                    // Commit batch size: commit_batch_size=100
                    "commit_batch_size" => {
                        config.persistence.commit_batch_size =
                            value.parse::<u32>().map_err(|_| {
                                Error::invalid_argument(format!(
                                    "invalid commit_batch_size: '{}'",
                                    value
                                ))
                            })?;
                    }
                    // Sync interval in ms: sync_interval_ms=10
                    "sync_interval_ms" | "sync_interval" => {
                        config.persistence.sync_interval_ms =
                            value.parse::<u32>().map_err(|_| {
                                Error::invalid_argument(format!(
                                    "invalid sync_interval_ms: '{}'",
                                    value
                                ))
                            })?;
                    }
                    // WAL compression: wal_compression=on|off
                    "wal_compression" => {
                        config.persistence.wal_compression =
                            matches!(value.to_lowercase().as_str(), "on" | "true" | "1" | "yes");
                    }
                    // Volume LZ4 compression: volume_compression=on|off
                    "volume_compression" => {
                        config.persistence.volume_compression =
                            matches!(value.to_lowercase().as_str(), "on" | "true" | "1" | "yes");
                    }
                    // All compressions (WAL + volume): compression=on|off
                    // Also accepts snapshot_compression for backward compatibility
                    "compression" | "snapshot_compression" => {
                        let enabled =
                            matches!(value.to_lowercase().as_str(), "on" | "true" | "1" | "yes");
                        config.persistence.wal_compression = enabled;
                        config.persistence.volume_compression = enabled;
                    }
                    // Compression threshold in bytes: compression_threshold=64
                    "compression_threshold" => {
                        config.persistence.compression_threshold =
                            value.parse::<usize>().map_err(|_| {
                                Error::invalid_argument(format!(
                                    "invalid compression_threshold: '{}'",
                                    value
                                ))
                            })?;
                    }
                    // Target rows per volume: target_volume_rows=1048576
                    "target_volume_rows" => {
                        let rows = value.parse::<usize>().map_err(|_| {
                            Error::invalid_argument(format!(
                                "invalid target_volume_rows: '{}'",
                                value
                            ))
                        })?;
                        config.persistence.target_volume_rows = rows.max(65_536);
                    }
                    // Checkpoint on close: checkpoint_on_close=off
                    // Set to off to simulate crashes in tests (WAL not truncated)
                    "checkpoint_on_close" => {
                        config.persistence.checkpoint_on_close =
                            matches!(value.to_lowercase().as_str(), "on" | "true" | "1" | "yes");
                    }
                    // Cleanup interval in seconds: cleanup_interval=60
                    "cleanup_interval" => {
                        config.cleanup.interval_secs = value.parse::<u64>().map_err(|_| {
                            Error::invalid_argument(format!(
                                "invalid cleanup_interval: '{}'",
                                value
                            ))
                        })?;
                    }
                    // Deleted row retention in seconds: deleted_row_retention=300
                    "deleted_row_retention" => {
                        config.cleanup.deleted_row_retention_secs =
                            value.parse::<u64>().map_err(|_| {
                                Error::invalid_argument(format!(
                                    "invalid deleted_row_retention: '{}'",
                                    value
                                ))
                            })?;
                    }
                    // Transaction retention in seconds: transaction_retention=3600
                    "transaction_retention" => {
                        config.cleanup.transaction_retention_secs =
                            value.parse::<u64>().map_err(|_| {
                                Error::invalid_argument(format!(
                                    "invalid transaction_retention: '{}'",
                                    value
                                ))
                            })?;
                    }
                    // Disable cleanup: cleanup=off
                    "cleanup" => {
                        config.cleanup.enabled =
                            matches!(value.to_lowercase().as_str(), "on" | "true" | "1" | "yes");
                    }
                    _ => {} // Ignore unknown parameters
                }
            }
        }

        Ok((clean_path, config))
    }

    /// Execute a SQL statement
    ///
    /// Use this for DDL (CREATE, DROP, ALTER) and DML (INSERT, UPDATE, DELETE) statements.
    ///
    /// # Parameters
    ///
    /// Parameters can be passed using:
    /// - Empty tuple `()` for no parameters
    /// - Tuple syntax `(1, "Alice", 30)` for multiple parameters
    /// - `params!` macro `params![1, "Alice", 30]`
    ///
    /// # Returns
    ///
    /// Returns the number of rows affected for DML statements, or 0 for DDL.
    ///
    /// # Examples
    ///
    /// ```ignore
    /// // DDL - no parameters
    /// db.execute("CREATE TABLE users (id INTEGER, name TEXT)", ())?;
    ///
    /// // DML with tuple parameters
    /// db.execute("INSERT INTO users VALUES ($1, $2)", (1, "Alice"))?;
    ///
    /// // DML with params! macro
    /// db.execute("INSERT INTO users VALUES ($1, $2)", params![2, "Bob"])?;
    ///
    /// // Update with mixed types
    /// let affected = db.execute(
    ///     "UPDATE users SET name = $1 WHERE id = $2",
    ///     ("Charlie", 1)
    /// )?;
    /// ```
    pub fn execute<P: Params>(&self, sql: &str, params: P) -> Result<i64> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;

        let param_values = params.into_params();
        let result = if param_values.is_empty() {
            executor.execute(sql)?
        } else if let Some(fast_result) = executor.try_fast_path_with_params(sql, &param_values) {
            fast_result?
        } else {
            executor.execute_with_params(sql, param_values)?
        };
        Ok(result.rows_affected())
    }

    /// Execute a query that returns rows
    ///
    /// # Parameters
    ///
    /// Parameters can be passed using:
    /// - Empty tuple `()` for no parameters
    /// - Tuple syntax `(value,)` for single parameter (note trailing comma)
    /// - Tuple syntax `(1, "Alice")` for multiple parameters
    /// - `params!` macro `params![1, "Alice"]`
    ///
    /// # Examples
    ///
    /// ```ignore
    /// // Query all rows
    /// for row in db.query("SELECT * FROM users", ())? {
    ///     let row = row?;
    ///     let id: i64 = row.get(0)?;
    ///     let name: String = row.get("name")?;
    /// }
    ///
    /// // Query with parameters
    /// for row in db.query("SELECT * FROM users WHERE age > $1", (18,))? {
    ///     // ...
    /// }
    ///
    /// // Collect into Vec
    /// let users: Vec<_> = db.query("SELECT * FROM users", ())?
    ///     .collect::<Result<Vec<_>, _>>()?;
    /// ```
    pub fn query<P: Params>(&self, sql: &str, params: P) -> Result<Rows> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;

        let param_values = params.into_params();
        let result = if param_values.is_empty() {
            executor.execute(sql)?
        } else if let Some(fast_result) = executor.try_fast_path_with_params(sql, &param_values) {
            fast_result?
        } else {
            executor.execute_with_params(sql, param_values)?
        };
        Ok(Rows::new(result))
    }

    /// Execute a query and return a single value
    ///
    /// This is a convenience method for queries that return a single row with a single column.
    /// Returns an error if the query returns no rows.
    ///
    /// # Examples
    ///
    /// ```ignore
    /// let count: i64 = db.query_one("SELECT COUNT(*) FROM users", ())?;
    /// let name: String = db.query_one("SELECT name FROM users WHERE id = $1", (1,))?;
    /// ```
    pub fn query_one<T: FromValue, P: Params>(&self, sql: &str, params: P) -> Result<T> {
        let row = self
            .query(sql, params)?
            .next()
            .ok_or(Error::NoRowsReturned)??;
        row.get(0)
    }

    /// Execute a query and return an optional single value
    ///
    /// Like `query_one`, but returns `None` if no rows are returned.
    ///
    /// # Examples
    ///
    /// ```ignore
    /// let name: Option<String> = db.query_opt("SELECT name FROM users WHERE id = $1", (999,))?;
    /// assert!(name.is_none());
    /// ```
    pub fn query_opt<T: FromValue, P: Params>(&self, sql: &str, params: P) -> Result<Option<T>> {
        match self.query(sql, params)?.next() {
            Some(row) => Ok(Some(row?.get(0)?)),
            None => Ok(None),
        }
    }

    /// Execute a write statement with a timeout
    ///
    /// Like `execute`, but cancels the query if it exceeds the timeout.
    /// Timeout is specified in milliseconds. Use 0 for no timeout.
    ///
    /// # Examples
    ///
    /// ```ignore
    /// // Execute with 5 second timeout
    /// db.execute_with_timeout("DELETE FROM large_table WHERE old = true", (), 5000)?;
    /// ```
    pub fn execute_with_timeout<P: Params>(
        &self,
        sql: &str,
        params: P,
        timeout_ms: u64,
    ) -> Result<i64> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;

        let param_values = params.into_params();
        let ctx = ExecutionContextBuilder::new()
            .params(param_values)
            .timeout_ms(timeout_ms)
            .build();

        let result = executor.execute_with_context(sql, &ctx)?;
        Ok(result.rows_affected())
    }

    /// Execute a query with a timeout
    ///
    /// Like `query`, but cancels the query if it exceeds the timeout.
    /// Timeout is specified in milliseconds. Use 0 for no timeout.
    ///
    /// # Examples
    ///
    /// ```ignore
    /// // Query with 10 second timeout
    /// for row in db.query_with_timeout("SELECT * FROM large_table", (), 10000)? {
    ///     // process row
    /// }
    /// ```
    pub fn query_with_timeout<P: Params>(
        &self,
        sql: &str,
        params: P,
        timeout_ms: u64,
    ) -> Result<Rows> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;

        let param_values = params.into_params();
        let ctx = ExecutionContextBuilder::new()
            .params(param_values)
            .timeout_ms(timeout_ms)
            .build();

        let result = executor.execute_with_context(sql, &ctx)?;
        Ok(Rows::new(result))
    }

    /// Prepare a SQL statement for repeated execution
    ///
    /// Prepared statements are more efficient when executing the same query
    /// multiple times with different parameters.
    ///
    /// # Examples
    ///
    /// ```ignore
    /// let stmt = db.prepare("SELECT * FROM users WHERE id = $1")?;
    ///
    /// // Execute multiple times with different parameters
    /// for id in 1..=10 {
    ///     for row in stmt.query((id,))? {
    ///         // ...
    ///     }
    /// }
    /// ```
    pub fn prepare(&self, sql: &str) -> Result<Statement> {
        Statement::new(Arc::downgrade(&self.inner), sql.to_string(), self)
    }

    /// Create a Database from an existing Arc<DatabaseInner>.
    /// Used by Statement to upgrade weak references.
    pub(crate) fn from_inner(inner: Arc<DatabaseInner>) -> Self {
        Database { inner }
    }

    /// Execute a statement with named parameters
    ///
    /// Named parameters use the `:name` syntax in SQL queries.
    ///
    /// # Examples
    ///
    /// ```ignore
    /// use stoolap::{Database, named_params};
    ///
    /// let db = Database::open("memory://")?;
    /// db.execute("CREATE TABLE users (id INTEGER, name TEXT, age INTEGER)", ())?;
    ///
    /// // Insert with named params
    /// db.execute_named(
    ///     "INSERT INTO users VALUES (:id, :name, :age)",
    ///     named_params!{ id: 1, name: "Alice", age: 30 }
    /// )?;
    ///
    /// // Update with named params
    /// db.execute_named(
    ///     "UPDATE users SET name = :name WHERE id = :id",
    ///     named_params!{ id: 1, name: "Alicia" }
    /// )?;
    /// ```
    pub fn execute_named(&self, sql: &str, params: NamedParams) -> Result<i64> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;

        let result = executor.execute_with_named_params(sql, params.into_inner())?;
        Ok(result.rows_affected())
    }

    /// Execute a query with named parameters
    ///
    /// Named parameters use the `:name` syntax in SQL queries.
    ///
    /// # Examples
    ///
    /// ```ignore
    /// use stoolap::{Database, named_params};
    ///
    /// let db = Database::open("memory://")?;
    /// db.execute("CREATE TABLE users (id INTEGER, name TEXT)", ())?;
    /// db.execute("INSERT INTO users VALUES (1, 'Alice'), (2, 'Bob')", ())?;
    ///
    /// // Query with named params
    /// for row in db.query_named(
    ///     "SELECT * FROM users WHERE name = :name",
    ///     named_params!{ name: "Alice" }
    /// )? {
    ///     let row = row?;
    ///     println!("Found user: id={}", row.get::<i64>(0)?);
    /// }
    /// ```
    pub fn query_named(&self, sql: &str, params: NamedParams) -> Result<Rows> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;

        let result = executor.execute_with_named_params(sql, params.into_inner())?;
        Ok(Rows::new(result))
    }

    /// Execute a query with named parameters and return a single value
    ///
    /// # Examples
    ///
    /// ```ignore
    /// use stoolap::{Database, named_params};
    ///
    /// let count: i64 = db.query_one_named(
    ///     "SELECT COUNT(*) FROM users WHERE age > :min_age",
    ///     named_params!{ min_age: 18 }
    /// )?;
    /// ```
    pub fn query_one_named<T: FromValue>(&self, sql: &str, params: NamedParams) -> Result<T> {
        let mut rows = self.query_named(sql, params)?;
        match rows.next() {
            Some(Ok(row)) => row.get(0),
            Some(Err(e)) => Err(e),
            None => Err(Error::NoRowsReturned),
        }
    }

    /// Execute a query and map results to structs
    ///
    /// This method executes a query and converts each row to a struct
    /// that implements the `FromRow` trait.
    ///
    /// # Examples
    ///
    /// ```ignore
    /// use stoolap::{Database, FromRow, ResultRow, Result};
    ///
    /// struct User {
    ///     id: i64,
    ///     name: String,
    /// }
    ///
    /// impl FromRow for User {
    ///     fn from_row(row: &ResultRow) -> Result<Self> {
    ///         Ok(User {
    ///             id: row.get(0)?,
    ///             name: row.get(1)?,
    ///         })
    ///     }
    /// }
    ///
    /// let db = Database::open("memory://")?;
    /// db.execute("CREATE TABLE users (id INTEGER, name TEXT)", ())?;
    /// db.execute("INSERT INTO users VALUES (1, 'Alice'), (2, 'Bob')", ())?;
    ///
    /// // Query and map to structs
    /// let users: Vec<User> = db.query_as("SELECT id, name FROM users", ())?;
    /// assert_eq!(users.len(), 2);
    /// assert_eq!(users[0].name, "Alice");
    /// ```
    pub fn query_as<T: FromRow, P: Params>(&self, sql: &str, params: P) -> Result<Vec<T>> {
        let rows = self.query(sql, params)?;
        rows.map(|r| r.and_then(|row| T::from_row(&row))).collect()
    }

    /// Execute a query with named parameters and map results to structs
    ///
    /// # Examples
    ///
    /// ```ignore
    /// use stoolap::{Database, FromRow, ResultRow, Result, named_params};
    ///
    /// struct Product {
    ///     id: i64,
    ///     name: String,
    ///     price: f64,
    /// }
    ///
    /// impl FromRow for Product {
    ///     fn from_row(row: &ResultRow) -> Result<Self> {
    ///         Ok(Product {
    ///             id: row.get(0)?,
    ///             name: row.get(1)?,
    ///             price: row.get(2)?,
    ///         })
    ///     }
    /// }
    ///
    /// let products: Vec<Product> = db.query_as_named(
    ///     "SELECT id, name, price FROM products WHERE price > :min_price",
    ///     named_params!{ min_price: 10.0 }
    /// )?;
    /// ```
    pub fn query_as_named<T: FromRow>(&self, sql: &str, params: NamedParams) -> Result<Vec<T>> {
        let rows = self.query_named(sql, params)?;
        rows.map(|r| r.and_then(|row| T::from_row(&row))).collect()
    }

    /// Begin a new transaction with default isolation level
    ///
    /// # Examples
    ///
    /// ```ignore
    /// let tx = db.begin()?;
    /// tx.execute("INSERT INTO users VALUES ($1, $2)", (1, "Alice"))?;
    /// tx.commit()?;
    /// ```
    pub fn begin(&self) -> Result<Transaction> {
        self.begin_with_isolation(IsolationLevel::ReadCommitted)
    }

    /// Begin a new transaction with a specific isolation level
    ///
    /// # Examples
    ///
    /// ```ignore
    /// use stoolap::IsolationLevel;
    ///
    /// let tx = db.begin_with_isolation(IsolationLevel::Snapshot)?;
    /// // All reads in this transaction see a consistent snapshot
    /// tx.execute("UPDATE users SET balance = balance - 100 WHERE id = $1", (1,))?;
    /// tx.commit()?;
    /// ```
    pub fn begin_with_isolation(&self, isolation: IsolationLevel) -> Result<Transaction> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;

        let tx = executor.begin_transaction_with_isolation(isolation)?;
        let engine = executor.engine().clone();
        Ok(Transaction::new(tx, engine))
    }

    /// Get the underlying storage engine
    ///
    /// This is primarily for advanced use cases and testing.
    pub fn engine(&self) -> &Arc<MVCCEngine> {
        &self.inner.engine
    }

    /// Close the database connection
    ///
    /// This removes the database from the global registry and closes the engine,
    /// releasing the file lock immediately so another process can open the database.
    ///
    /// Note: The engine is also closed automatically when all Database instances
    /// are dropped.
    pub fn close(&self) -> Result<()> {
        // Remove from registry
        let mut registry = DATABASE_REGISTRY
            .write()
            .map_err(|_| Error::LockAcquisitionFailed("registry write".to_string()))?;

        registry.remove(&self.inner.dsn);

        // Close the engine immediately to release the file lock
        // This is idempotent - calling close_engine() multiple times is safe
        self.inner.engine.close_engine()?;

        Ok(())
    }

    /// Get a cached plan for a SQL statement (parse once, execute many times).
    ///
    /// Returns a `CachedPlanRef` that can be stored and passed to
    /// `execute_plan()` / `query_plan()` for zero-lookup execution.
    pub fn cached_plan(&self, sql: &str) -> Result<CachedPlanRef> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
        executor.get_or_create_plan(sql)
    }

    /// Execute a pre-cached plan with positional parameters (no parsing, no cache lookup).
    pub fn execute_plan<P: Params>(&self, plan: &CachedPlanRef, params: P) -> Result<i64> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
        let param_values = params.into_params();
        let ctx = if param_values.is_empty() {
            ExecutionContext::new()
        } else {
            ExecutionContext::with_params(param_values)
        };
        let result = executor.execute_with_cached_plan(plan, &ctx)?;
        Ok(result.rows_affected())
    }

    /// Query using a pre-cached plan with positional parameters (no parsing, no cache lookup).
    pub fn query_plan<P: Params>(&self, plan: &CachedPlanRef, params: P) -> Result<Rows> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
        let param_values = params.into_params();
        let ctx = if param_values.is_empty() {
            ExecutionContext::new()
        } else {
            ExecutionContext::with_params(param_values)
        };
        let result = executor.execute_with_cached_plan(plan, &ctx)?;
        Ok(Rows::new(result))
    }

    /// Execute a pre-cached plan with named parameters (no parsing, no cache lookup).
    pub fn execute_named_plan(&self, plan: &CachedPlanRef, params: NamedParams) -> Result<i64> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
        let ctx = ExecutionContext::with_named_params(params.into_inner());
        let result = executor.execute_with_cached_plan(plan, &ctx)?;
        Ok(result.rows_affected())
    }

    /// Query using a pre-cached plan with named parameters (no parsing, no cache lookup).
    pub fn query_named_plan(&self, plan: &CachedPlanRef, params: NamedParams) -> Result<Rows> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
        let ctx = ExecutionContext::with_named_params(params.into_inner());
        let result = executor.execute_with_cached_plan(plan, &ctx)?;
        Ok(Rows::new(result))
    }

    /// Check if a table exists
    pub fn table_exists(&self, name: &str) -> Result<bool> {
        let engine = &self.inner.engine;
        let tx = engine.begin_transaction()?;
        Ok(tx.get_table(name).is_ok())
    }

    /// Get the DSN this database was opened with
    pub fn dsn(&self) -> &str {
        &self.inner.dsn
    }

    /// Set the default isolation level for new transactions
    pub fn set_default_isolation_level(&self, level: IsolationLevel) -> Result<()> {
        let mut executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
        executor.set_default_isolation_level(level);
        Ok(())
    }

    /// Create a backup snapshot of the database
    ///
    /// This creates snapshot files (.bin) for each table along with
    /// index/view definitions (ddl-{timestamp}.bin) for disaster recovery.
    /// Normal persistence uses the checkpoint cycle (seal to volumes + WAL).
    ///
    /// Note: This is a no-op for in-memory databases.
    pub fn create_snapshot(&self) -> Result<()> {
        use crate::storage::Engine;
        self.inner.engine.create_snapshot()
    }

    /// Restore the database from a backup snapshot.
    ///
    /// If no timestamp is provided, restores from the latest snapshot.
    /// If a timestamp is provided (format: "YYYYMMDD-HHMMSS.fff"),
    /// restores from that specific snapshot.
    ///
    /// This is a destructive operation that replaces all current data
    /// with the snapshot data. Indexes and views are restored from
    /// ddl-{timestamp}.bin or preserved from current in-memory state.
    pub fn restore_snapshot(&self, timestamp: Option<&str>) -> Result<String> {
        use crate::storage::Engine;
        let result = self.inner.engine.restore_snapshot(timestamp)?;
        // Clear all query caches since all data has changed.
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
        executor.clear_semantic_cache();
        crate::executor::context::clear_scalar_subquery_cache();
        crate::executor::context::clear_in_subquery_cache();
        crate::executor::context::clear_semi_join_cache();
        Ok(result)
    }

    /// Get the internal executor (for Statement use)
    pub(crate) fn executor(&self) -> &Mutex<Executor> {
        &self.inner.executor
    }

    /// Get semantic cache statistics
    ///
    /// Returns statistics about the semantic query cache including hit rates,
    /// exact matches, and subsumption matches.
    ///
    /// # Examples
    ///
    /// ```ignore
    /// let db = Database::open("memory://")?;
    /// // ... execute some queries ...
    /// let stats = db.semantic_cache_stats()?;
    /// println!("Cache hits: {}", stats.hits);
    /// println!("Subsumption hits: {}", stats.subsumption_hits);
    /// ```
    pub fn semantic_cache_stats(&self) -> Result<crate::executor::SemanticCacheStatsSnapshot> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
        Ok(executor.semantic_cache_stats())
    }

    /// Clear the semantic cache
    ///
    /// This clears all cached query results. Useful for testing or when
    /// you want to force queries to re-execute.
    pub fn clear_semantic_cache(&self) -> Result<()> {
        let executor = self
            .inner
            .executor
            .lock()
            .map_err(|_| Error::LockAcquisitionFailed("executor".to_string()))?;
        executor.clear_semantic_cache();
        Ok(())
    }

    /// Get the oldest snapshot timestamp loaded during startup.
    /// Returns None if no snapshots were loaded.
    pub fn oldest_loaded_snapshot_timestamp(&self) -> Option<String> {
        self.inner.engine.oldest_loaded_snapshot_timestamp()
    }
}

/// Trait for converting from Value to a Rust type
pub trait FromValue: Sized {
    /// Convert a Value to Self
    fn from_value(value: &Value) -> Result<Self>;
}

impl FromValue for i64 {
    fn from_value(value: &Value) -> Result<Self> {
        match value {
            Value::Integer(i) => Ok(*i),
            Value::Float(f) => Ok(*f as i64),
            _ => Err(Error::TypeConversion {
                from: format!("{:?}", value),
                to: "Integer".to_string(),
            }),
        }
    }
}

impl FromValue for i32 {
    fn from_value(value: &Value) -> Result<Self> {
        match value {
            Value::Integer(i) => Ok(*i as i32),
            Value::Float(f) => Ok(*f as i32),
            _ => Err(Error::TypeConversion {
                from: format!("{:?}", value),
                to: "Integer".to_string(),
            }),
        }
    }
}

impl FromValue for f64 {
    fn from_value(value: &Value) -> Result<Self> {
        match value {
            Value::Float(f) => Ok(*f),
            Value::Integer(i) => Ok(*i as f64),
            _ => Err(Error::TypeConversion {
                from: format!("{:?}", value),
                to: "Float".to_string(),
            }),
        }
    }
}

impl FromValue for String {
    fn from_value(value: &Value) -> Result<Self> {
        match value {
            Value::Text(s) => Ok(s.to_string()),
            Value::Extension(data) if data.first() == Some(&(DataType::Json as u8)) => {
                Ok(std::str::from_utf8(&data[1..]).unwrap_or("").to_string())
            }
            // Convert other types to string representation
            Value::Integer(i) => Ok(i.to_string()),
            Value::Float(f) => Ok(f.to_string()),
            Value::Boolean(b) => Ok(if *b {
                "true".to_string()
            } else {
                "false".to_string()
            }),
            Value::Timestamp(ts) => Ok(ts.format("%Y-%m-%dT%H:%M:%SZ").to_string()),
            Value::Extension(_) => value
                .as_string()
                .ok_or_else(|| Error::invalid_argument("Cannot convert extension to String")),
            Value::Null(_) => Ok(String::new()),
        }
    }
}

impl FromValue for bool {
    fn from_value(value: &Value) -> Result<Self> {
        match value {
            Value::Boolean(b) => Ok(*b),
            Value::Integer(i) => Ok(*i != 0),
            _ => Err(Error::TypeConversion {
                from: format!("{:?}", value),
                to: "Boolean".to_string(),
            }),
        }
    }
}

impl FromValue for Value {
    fn from_value(value: &Value) -> Result<Self> {
        Ok(value.clone())
    }
}

impl<T: FromValue> FromValue for Option<T> {
    fn from_value(value: &Value) -> Result<Self> {
        if value.is_null() {
            Ok(None)
        } else {
            Ok(Some(T::from_value(value)?))
        }
    }
}

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

    #[test]
    fn test_open_memory() {
        let db = Database::open("memory://").unwrap();
        assert_eq!(db.dsn(), "memory://");
    }

    #[test]
    fn test_open_in_memory() {
        let db = Database::open_in_memory().unwrap();
        db.execute("CREATE TABLE test (id INTEGER PRIMARY KEY)", ())
            .unwrap();
        db.execute("INSERT INTO test VALUES ($1)", (1,)).unwrap();

        for row in db.query("SELECT * FROM test", ()).unwrap() {
            let row = row.unwrap();
            let id: i64 = row.get(0).unwrap();
            assert_eq!(id, 1);
        }
    }

    #[test]
    fn test_execute_and_query_new_api() {
        let db = Database::open_in_memory().unwrap();

        // Create table - no params
        db.execute(
            "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)",
            (),
        )
        .unwrap();

        // Insert with tuple params
        let affected = db
            .execute(
                "INSERT INTO users VALUES ($1, $2, $3), ($4, $5, $6)",
                (1, "Alice", 30, 2, "Bob", 25),
            )
            .unwrap();
        assert_eq!(affected, 2);

        // Query with tuple params
        let rows: Vec<_> = db
            .query("SELECT * FROM users ORDER BY id", ())
            .unwrap()
            .collect::<std::result::Result<Vec<_>, _>>()
            .unwrap();

        assert_eq!(rows.len(), 2);
        assert_eq!(rows[0].get::<i64>(0).unwrap(), 1);
        assert_eq!(rows[0].get::<String>(1).unwrap(), "Alice");
        assert_eq!(rows[0].get::<i64>(2).unwrap(), 30);
    }

    #[test]
    fn test_query_one() {
        let db = Database::open_in_memory().unwrap();
        db.execute("CREATE TABLE test (id INTEGER PRIMARY KEY)", ())
            .unwrap();
        db.execute("INSERT INTO test VALUES ($1), ($2), ($3)", (1, 2, 3))
            .unwrap();

        let count: i64 = db.query_one("SELECT COUNT(*) FROM test", ()).unwrap();
        assert_eq!(count, 3);
    }

    #[test]
    fn test_query_opt() {
        let db = Database::open_in_memory().unwrap();
        db.execute("CREATE TABLE test (id INTEGER PRIMARY KEY)", ())
            .unwrap();
        db.execute("INSERT INTO test VALUES ($1)", (1,)).unwrap();

        // Found
        let result: Option<i64> = db
            .query_opt("SELECT id FROM test WHERE id = $1", (1,))
            .unwrap();
        assert_eq!(result, Some(1));

        // Not found
        let result: Option<i64> = db
            .query_opt("SELECT id FROM test WHERE id = $1", (999,))
            .unwrap();
        assert_eq!(result, None);
    }

    #[test]
    fn test_params_macro() {
        let db = Database::open_in_memory().unwrap();
        db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)", ())
            .unwrap();

        // Use params! macro
        db.execute(
            "INSERT INTO users VALUES ($1, $2)",
            crate::params![1, "Alice"],
        )
        .unwrap();

        let names: Vec<String> = db
            .query("SELECT name FROM users WHERE id = $1", crate::params![1])
            .unwrap()
            .map(|r| r.and_then(|row| row.get(0)))
            .collect::<std::result::Result<Vec<_>, _>>()
            .unwrap();

        assert_eq!(names, vec!["Alice"]);
    }

    #[test]
    fn test_parse_dsn() {
        // Memory
        let (scheme, path) = Database::parse_dsn("memory://").unwrap();
        assert_eq!(scheme, "memory");
        assert_eq!(path, "");

        // File
        let (scheme, path) = Database::parse_dsn("file:///tmp/test.db").unwrap();
        assert_eq!(scheme, "file");
        assert_eq!(path, "/tmp/test.db");

        // File with params
        let (scheme, path) = Database::parse_dsn("file:///tmp/test.db?sync=full").unwrap();
        assert_eq!(scheme, "file");
        assert_eq!(path, "/tmp/test.db?sync=full");

        // Invalid
        assert!(Database::parse_dsn("invalid").is_err());
        assert!(Database::parse_dsn("unknown://test").is_err());
    }

    #[test]
    fn test_from_value_types() {
        assert_eq!(i64::from_value(&Value::Integer(42)).unwrap(), 42);
        assert_eq!(f64::from_value(&Value::Float(3.5)).unwrap(), 3.5);
        assert_eq!(
            String::from_value(&Value::Text("hello".into())).unwrap(),
            "hello"
        );
        assert!(bool::from_value(&Value::Boolean(true)).unwrap());

        // Optional
        assert_eq!(
            Option::<i64>::from_value(&Value::Integer(42)).unwrap(),
            Some(42)
        );
        assert_eq!(
            Option::<i64>::from_value(&Value::null_unknown()).unwrap(),
            None
        );
    }

    #[test]
    fn test_cached_plan_insert_and_query() {
        let db = Database::open_in_memory().unwrap();
        db.execute(
            "CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT, score FLOAT)",
            (),
        )
        .unwrap();

        let insert_plan = db
            .cached_plan("INSERT INTO test VALUES ($1, $2, $3)")
            .unwrap();

        // Batch insert using cached plan
        db.execute_plan(&insert_plan, (1, "Alice", 95.5)).unwrap();
        db.execute_plan(&insert_plan, (2, "Bob", 82.0)).unwrap();
        db.execute_plan(&insert_plan, (3, "Charlie", 91.0)).unwrap();

        // Query using cached plan
        let query_plan = db
            .cached_plan("SELECT name FROM test WHERE id = $1")
            .unwrap();
        let mut rows = db.query_plan(&query_plan, (2,)).unwrap();
        let row = rows.next().unwrap().unwrap();
        assert_eq!(row.get::<String>(0).unwrap(), "Bob");
    }

    #[test]
    fn test_cached_plan_reuse() {
        let db = Database::open_in_memory().unwrap();
        db.execute(
            "CREATE TABLE test (id INTEGER PRIMARY KEY, value INTEGER)",
            (),
        )
        .unwrap();

        // Get the same plan twice — second call should hit the cache
        let plan1 = db.cached_plan("INSERT INTO test VALUES ($1, $2)").unwrap();
        let plan2 = db.cached_plan("INSERT INTO test VALUES ($1, $2)").unwrap();

        // Both should work independently
        db.execute_plan(&plan1, (1, 100)).unwrap();
        db.execute_plan(&plan2, (2, 200)).unwrap();

        let count: i64 = db.query_one("SELECT COUNT(*) FROM test", ()).unwrap();
        assert_eq!(count, 2);
    }

    #[test]
    fn test_cached_plan_update_delete() {
        let db = Database::open_in_memory().unwrap();
        db.execute(
            "CREATE TABLE test (id INTEGER PRIMARY KEY, value INTEGER)",
            (),
        )
        .unwrap();
        db.execute("INSERT INTO test VALUES (1, 100)", ()).unwrap();
        db.execute("INSERT INTO test VALUES (2, 200)", ()).unwrap();

        // Update via cached plan
        let update_plan = db
            .cached_plan("UPDATE test SET value = $1 WHERE id = $2")
            .unwrap();
        let affected = db.execute_plan(&update_plan, (999, 1)).unwrap();
        assert_eq!(affected, 1);

        let val: i64 = db
            .query_one("SELECT value FROM test WHERE id = 1", ())
            .unwrap();
        assert_eq!(val, 999);

        // Delete via cached plan
        let delete_plan = db.cached_plan("DELETE FROM test WHERE id = $1").unwrap();
        let affected = db.execute_plan(&delete_plan, (2,)).unwrap();
        assert_eq!(affected, 1);

        let count: i64 = db.query_one("SELECT COUNT(*) FROM test", ()).unwrap();
        assert_eq!(count, 1);
    }

    #[test]
    fn test_cached_plan_no_params() {
        let db = Database::open_in_memory().unwrap();
        db.execute(
            "CREATE TABLE test (id INTEGER PRIMARY KEY, value INTEGER)",
            (),
        )
        .unwrap();
        db.execute("INSERT INTO test VALUES (1, 10)", ()).unwrap();
        db.execute("INSERT INTO test VALUES (2, 20)", ()).unwrap();

        let plan = db.cached_plan("SELECT COUNT(*) FROM test").unwrap();
        let mut rows = db.query_plan(&plan, ()).unwrap();
        let row = rows.next().unwrap().unwrap();
        assert_eq!(row.get::<i64>(0).unwrap(), 2);
    }

    #[test]
    fn test_cached_plan_named_params() {
        let db = Database::open_in_memory().unwrap();
        db.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, name TEXT)", ())
            .unwrap();

        let plan = db
            .cached_plan("INSERT INTO test VALUES (:id, :name)")
            .unwrap();
        db.execute_named_plan(&plan, named_params! { id: 1, name: "Alice" })
            .unwrap();
        db.execute_named_plan(&plan, named_params! { id: 2, name: "Bob" })
            .unwrap();

        let query_plan = db
            .cached_plan("SELECT name FROM test WHERE id = :id")
            .unwrap();
        let mut rows = db
            .query_named_plan(&query_plan, named_params! { id: 1 })
            .unwrap();
        let row = rows.next().unwrap().unwrap();
        assert_eq!(row.get::<String>(0).unwrap(), "Alice");
    }

    #[test]
    fn test_cached_plan_multi_statement_error() {
        let db = Database::open_in_memory().unwrap();
        db.execute("CREATE TABLE test (id INTEGER PRIMARY KEY)", ())
            .unwrap();

        // Multiple statements should fail
        let result = db.cached_plan("INSERT INTO test VALUES (1); INSERT INTO test VALUES (2)");
        assert!(result.is_err());
    }
}