prax_query/

cache.rs

//! Query caching and prepared statement management.
//!
//! This module provides utilities for caching SQL queries and managing
//! prepared statements to improve performance.
//!
//! # Query Cache
//!
//! The `QueryCache` stores recently executed queries by their hash,
//! allowing fast lookup of previously built SQL strings.
//!
//! ```rust
//! use prax_query::cache::QueryCache;
//!
//! let cache = QueryCache::new(1000);
//!
//! // Cache a query
//! cache.insert("users_by_id", "SELECT * FROM users WHERE id = $1");
//!
//! // Retrieve later
//! if let Some(sql) = cache.get("users_by_id") {
//!     println!("Cached SQL: {}", sql);
//! }
//! ```

use std::borrow::Cow;
use std::collections::HashMap;
use std::hash::{Hash, Hasher};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use tracing::debug;

/// A thread-safe cache for SQL queries.
///
/// Uses a simple LRU-like eviction strategy when the cache is full.
#[derive(Debug)]
pub struct QueryCache {
    /// Maximum number of entries in the cache.
    max_size: usize,
    /// The cached queries.
    cache: RwLock<HashMap<QueryKey, CachedQuery>>,
    /// Statistics about cache usage.
    ///
    /// Atomic counters so the `get()` hot path can record hits/misses
    /// without taking the stats write lock while still holding the
    /// cache read lock (the previous nested-lock pattern was a
    /// contention hotspot under concurrent reads).
    stats: AtomicCacheStats,
}

/// Atomic backing for [`CacheStats`].
///
/// `record_*` methods use `Relaxed` because the counters are
/// strictly monotonic and don't synchronize with anything else; the
/// only consumer is `QueryCache::stats()` which snapshots them into a
/// regular [`CacheStats`] value.
#[derive(Debug, Default)]
struct AtomicCacheStats {
    hits: AtomicU64,
    misses: AtomicU64,
    evictions: AtomicU64,
    insertions: AtomicU64,
}

impl AtomicCacheStats {
    #[inline]
    fn record_hit(&self) {
        self.hits.fetch_add(1, Ordering::Relaxed);
    }

    #[inline]
    fn record_miss(&self) {
        self.misses.fetch_add(1, Ordering::Relaxed);
    }

    #[inline]
    fn record_eviction(&self) {
        self.evictions.fetch_add(1, Ordering::Relaxed);
    }

    #[inline]
    fn record_insertion(&self) {
        self.insertions.fetch_add(1, Ordering::Relaxed);
    }

    fn snapshot(&self) -> CacheStats {
        CacheStats {
            hits: self.hits.load(Ordering::Relaxed),
            misses: self.misses.load(Ordering::Relaxed),
            evictions: self.evictions.load(Ordering::Relaxed),
            insertions: self.insertions.load(Ordering::Relaxed),
        }
    }

    fn reset(&self) {
        self.hits.store(0, Ordering::Relaxed);
        self.misses.store(0, Ordering::Relaxed);
        self.evictions.store(0, Ordering::Relaxed);
        self.insertions.store(0, Ordering::Relaxed);
    }
}

/// A key for looking up cached queries.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct QueryKey {
    /// The unique identifier for this query type.
    key: Cow<'static, str>,
}

impl QueryKey {
    /// Create a new query key from a static string.
    #[inline]
    pub const fn new(key: &'static str) -> Self {
        Self {
            key: Cow::Borrowed(key),
        }
    }

    /// Create a new query key from an owned string.
    #[inline]
    pub fn owned(key: String) -> Self {
        Self {
            key: Cow::Owned(key),
        }
    }
}

impl From<&'static str> for QueryKey {
    fn from(s: &'static str) -> Self {
        Self::new(s)
    }
}

impl From<String> for QueryKey {
    fn from(s: String) -> Self {
        Self::owned(s)
    }
}

/// A cached SQL query.
#[derive(Debug)]
pub struct CachedQuery {
    /// The SQL string.
    pub sql: String,
    /// The number of parameters expected.
    pub param_count: usize,
    /// Number of times this query has been accessed.
    ///
    /// Atomic so hits can be recorded under the cache's read lock
    /// (interior mutability, same pattern as `SqlTemplate::touch`).
    access_count: AtomicU64,
}

impl Clone for CachedQuery {
    fn clone(&self) -> Self {
        Self {
            sql: self.sql.clone(),
            param_count: self.param_count,
            access_count: AtomicU64::new(self.access_count.load(Ordering::Relaxed)),
        }
    }
}

impl CachedQuery {
    /// Create a new cached query.
    pub fn new(sql: impl Into<String>, param_count: usize) -> Self {
        Self {
            sql: sql.into(),
            param_count,
            access_count: AtomicU64::new(0),
        }
    }

    /// Record a cache hit for LRU eviction.
    #[inline]
    fn record_access(&self) {
        self.access_count.fetch_add(1, Ordering::Relaxed);
    }

    /// Get the SQL string.
    #[inline]
    pub fn sql(&self) -> &str {
        &self.sql
    }

    /// Get the expected parameter count.
    #[inline]
    pub fn param_count(&self) -> usize {
        self.param_count
    }
}

/// Statistics about cache usage.
#[derive(Debug, Default, Clone)]
pub struct CacheStats {
    /// Number of cache hits.
    pub hits: u64,
    /// Number of cache misses.
    pub misses: u64,
    /// Number of evictions.
    pub evictions: u64,
    /// Number of insertions.
    pub insertions: u64,
}

impl CacheStats {
    /// Calculate the hit rate.
    #[inline]
    pub fn hit_rate(&self) -> f64 {
        let total = self.hits + self.misses;
        if total == 0 {
            0.0
        } else {
            self.hits as f64 / total as f64
        }
    }
}

impl QueryCache {
    /// Create a new query cache with the given maximum size.
    pub fn new(max_size: usize) -> Self {
        tracing::info!(max_size, "QueryCache initialized");
        Self {
            max_size,
            cache: RwLock::new(HashMap::with_capacity(max_size)),
            stats: AtomicCacheStats::default(),
        }
    }

    /// Insert a query into the cache.
    pub fn insert(&self, key: impl Into<QueryKey>, sql: impl Into<String>) {
        let key = key.into();
        let sql = sql.into();
        let param_count = count_placeholders(&sql);
        debug!(key = ?key.key, sql_len = sql.len(), param_count, "QueryCache::insert()");

        let mut cache = self.cache.write().unwrap();

        // Evict if full
        if cache.len() >= self.max_size && !cache.contains_key(&key) {
            self.evict_lru(&mut cache);
            self.stats.record_eviction();
            debug!("QueryCache evicted entry");
        }

        cache.insert(key, CachedQuery::new(sql, param_count));
        self.stats.record_insertion();
    }

    /// Insert a query with known parameter count.
    pub fn insert_with_params(
        &self,
        key: impl Into<QueryKey>,
        sql: impl Into<String>,
        param_count: usize,
    ) {
        let key = key.into();
        let sql = sql.into();

        let mut cache = self.cache.write().unwrap();

        // Evict if full
        if cache.len() >= self.max_size && !cache.contains_key(&key) {
            self.evict_lru(&mut cache);
            self.stats.record_eviction();
        }

        cache.insert(key, CachedQuery::new(sql, param_count));
        self.stats.record_insertion();
    }

    /// Get a query from the cache.
    pub fn get(&self, key: impl Into<QueryKey>) -> Option<String> {
        let key = key.into();

        let cache = self.cache.read().unwrap();
        if let Some(entry) = cache.get(&key) {
            // Atomic counters — no write-lock conflict with the read
            // lock we still hold on `cache`.
            entry.record_access();
            self.stats.record_hit();
            debug!(key = ?key.key, "QueryCache hit");
            return Some(entry.sql.clone());
        }
        drop(cache);

        self.stats.record_miss();
        debug!(key = ?key.key, "QueryCache miss");
        None
    }

    /// Get a cached query entry (includes metadata).
    pub fn get_entry(&self, key: impl Into<QueryKey>) -> Option<CachedQuery> {
        let key = key.into();

        let cache = self.cache.read().unwrap();
        if let Some(entry) = cache.get(&key) {
            entry.record_access();
            self.stats.record_hit();
            return Some(entry.clone());
        }
        drop(cache);

        self.stats.record_miss();
        None
    }

    /// Get or compute a query.
    ///
    /// If the query is cached, returns the cached version.
    /// Otherwise, computes it using the provided function and caches it.
    pub fn get_or_insert<F>(&self, key: impl Into<QueryKey>, f: F) -> String
    where
        F: FnOnce() -> String,
    {
        let key = key.into();

        // Try to get from cache
        if let Some(sql) = self.get(key.clone()) {
            return sql;
        }

        // Compute and insert
        let sql = f();
        self.insert(key, sql.clone());
        sql
    }

    /// Check if a key exists in the cache.
    pub fn contains(&self, key: impl Into<QueryKey>) -> bool {
        let key = key.into();
        let cache = self.cache.read().unwrap();
        cache.contains_key(&key)
    }

    /// Remove a query from the cache.
    pub fn remove(&self, key: impl Into<QueryKey>) -> Option<String> {
        let key = key.into();
        let mut cache = self.cache.write().unwrap();
        cache.remove(&key).map(|e| e.sql)
    }

    /// Clear the entire cache.
    pub fn clear(&self) {
        let mut cache = self.cache.write().unwrap();
        cache.clear();
    }

    /// Get the current number of cached queries.
    pub fn len(&self) -> usize {
        let cache = self.cache.read().unwrap();
        cache.len()
    }

    /// Check if the cache is empty.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Get the maximum cache size.
    pub fn max_size(&self) -> usize {
        self.max_size
    }

    /// Get cache statistics.
    pub fn stats(&self) -> CacheStats {
        self.stats.snapshot()
    }

    /// Reset cache statistics.
    pub fn reset_stats(&self) {
        self.stats.reset();
    }

    /// Evict the least recently used entries.
    fn evict_lru(&self, cache: &mut HashMap<QueryKey, CachedQuery>) {
        // Evict entries with the lowest access count: 25% of the cache,
        // but always at least one so a full cache never grows past
        // max_size (len < 4 would otherwise evict nothing).
        if cache.is_empty() {
            return;
        }
        let to_evict = (cache.len() / 4).max(1);

        // Snapshot (access_count, key) pairs — only the keys that are
        // actually evicted get cloned. A partial select partitions the
        // `to_evict` lowest-count entries up front (O(n) average
        // instead of an O(n log n) full sort).
        let mut entries: Vec<_> = cache
            .iter()
            .map(|(k, v)| (v.access_count.load(Ordering::Relaxed), k))
            .collect();
        entries.select_nth_unstable_by(to_evict - 1, |a, b| a.0.cmp(&b.0));

        let evict_keys: Vec<QueryKey> = entries[..to_evict]
            .iter()
            .map(|(_, k)| (*k).clone())
            .collect();
        for key in evict_keys {
            cache.remove(&key);
        }
    }
}

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

/// Count the number of parameter placeholders in a SQL string.
fn count_placeholders(sql: &str) -> usize {
    let mut count = 0;
    let mut chars = sql.chars().peekable();

    while let Some(c) = chars.next() {
        if c == '$' {
            // PostgreSQL-style: $1, $2, etc.
            let mut num = String::new();
            while let Some(&d) = chars.peek() {
                if d.is_ascii_digit() {
                    num.push(d);
                    chars.next();
                } else {
                    break;
                }
            }
            if !num.is_empty()
                && let Ok(n) = num.parse::<usize>()
            {
                count = count.max(n);
            }
        } else if c == '?' {
            // MySQL/SQLite-style
            count += 1;
        }
    }

    count
}

/// A query hash for fast lookup.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct QueryHash(u64);

impl QueryHash {
    /// Compute a hash for the given SQL query.
    pub fn new(sql: &str) -> Self {
        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        sql.hash(&mut hasher);
        Self(hasher.finish())
    }

    /// Get the raw hash value.
    #[inline]
    pub fn value(&self) -> u64 {
        self.0
    }
}

/// Common query patterns for caching.
pub mod patterns {
    use super::QueryKey;

    /// Query key for SELECT by ID.
    #[inline]
    pub fn select_by_id(table: &str) -> QueryKey {
        QueryKey::owned(format!("select_by_id:{}", table))
    }

    /// Query key for SELECT all.
    #[inline]
    pub fn select_all(table: &str) -> QueryKey {
        QueryKey::owned(format!("select_all:{}", table))
    }

    /// Query key for INSERT.
    #[inline]
    pub fn insert(table: &str, columns: usize) -> QueryKey {
        QueryKey::owned(format!("insert:{}:{}", table, columns))
    }

    /// Query key for UPDATE by ID.
    #[inline]
    pub fn update_by_id(table: &str, columns: usize) -> QueryKey {
        QueryKey::owned(format!("update_by_id:{}:{}", table, columns))
    }

    /// Query key for DELETE by ID.
    #[inline]
    pub fn delete_by_id(table: &str) -> QueryKey {
        QueryKey::owned(format!("delete_by_id:{}", table))
    }

    /// Query key for COUNT.
    #[inline]
    pub fn count(table: &str) -> QueryKey {
        QueryKey::owned(format!("count:{}", table))
    }

    /// Query key for COUNT with filter.
    #[inline]
    pub fn count_filtered(table: &str, filter_hash: u64) -> QueryKey {
        QueryKey::owned(format!("count:{}:{}", table, filter_hash))
    }
}

// =============================================================================
// High-Performance SQL Template Cache
// =============================================================================

/// A high-performance SQL template cache optimized for repeated queries.
///
/// Unlike `QueryCache` which stores full SQL strings, `SqlTemplateCache` stores
/// template structures with pre-computed placeholder positions for very fast
/// instantiation.
///
/// # Performance
///
/// - Cache lookup: O(1) hash lookup, ~5-10ns
/// - Template instantiation: O(n) where n is parameter count
/// - Thread-safe with minimal contention (parking_lot RwLock)
///
/// # Examples
///
/// ```rust
/// use prax_query::cache::SqlTemplateCache;
///
/// let cache = SqlTemplateCache::new(1000);
///
/// // Register a template
/// let template = cache.register("users_by_id", "SELECT * FROM users WHERE id = $1");
///
/// // Instant retrieval (~5ns)
/// let sql = cache.get("users_by_id");
/// ```
#[derive(Debug)]
pub struct SqlTemplateCache {
    /// Maximum number of templates.
    max_size: usize,
    /// Cached templates (using Arc for cheap cloning).
    templates: parking_lot::RwLock<HashMap<u64, Arc<SqlTemplate>>>,
    /// String key to hash lookup.
    key_index: parking_lot::RwLock<HashMap<Cow<'static, str>, u64>>,
    /// Statistics.
    stats: parking_lot::RwLock<CacheStats>,
}

/// A pre-parsed SQL template for fast instantiation.
#[derive(Debug)]
pub struct SqlTemplate {
    /// The complete SQL string (for direct use).
    pub sql: Arc<str>,
    /// Pre-computed hash for fast lookup.
    pub hash: u64,
    /// Number of parameters.
    pub param_count: usize,
    /// Access timestamp for LRU.
    last_access: std::sync::atomic::AtomicU64,
}

impl Clone for SqlTemplate {
    fn clone(&self) -> Self {
        use std::sync::atomic::Ordering;
        Self {
            sql: Arc::clone(&self.sql),
            hash: self.hash,
            param_count: self.param_count,
            last_access: std::sync::atomic::AtomicU64::new(
                self.last_access.load(Ordering::Relaxed),
            ),
        }
    }
}

impl SqlTemplate {
    /// Create a new SQL template.
    pub fn new(sql: impl AsRef<str>) -> Self {
        let sql_str = sql.as_ref();
        let param_count = count_placeholders(sql_str);
        let hash = {
            let mut hasher = std::collections::hash_map::DefaultHasher::new();
            sql_str.hash(&mut hasher);
            hasher.finish()
        };

        Self {
            sql: Arc::from(sql_str),
            hash,
            param_count,
            last_access: std::sync::atomic::AtomicU64::new(0),
        }
    }

    /// Get the SQL string as a reference.
    #[inline(always)]
    pub fn sql(&self) -> &str {
        &self.sql
    }

    /// Get the SQL string as an Arc (zero-copy clone).
    #[inline(always)]
    pub fn sql_arc(&self) -> Arc<str> {
        Arc::clone(&self.sql)
    }

    /// Touch the template to update LRU access time.
    #[inline]
    fn touch(&self) {
        use std::sync::atomic::Ordering;
        use std::time::{SystemTime, UNIX_EPOCH};
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);
        self.last_access.store(now, Ordering::Relaxed);
    }
}

impl SqlTemplateCache {
    /// Create a new template cache with the given maximum size.
    pub fn new(max_size: usize) -> Self {
        tracing::info!(max_size, "SqlTemplateCache initialized");
        Self {
            max_size,
            templates: parking_lot::RwLock::new(HashMap::with_capacity(max_size)),
            key_index: parking_lot::RwLock::new(HashMap::with_capacity(max_size)),
            stats: parking_lot::RwLock::new(CacheStats::default()),
        }
    }

    /// Register a SQL template with a string key.
    ///
    /// Returns the template for immediate use.
    #[inline]
    pub fn register(
        &self,
        key: impl Into<Cow<'static, str>>,
        sql: impl AsRef<str>,
    ) -> Arc<SqlTemplate> {
        let key = key.into();
        let template = Arc::new(SqlTemplate::new(sql));
        let hash = template.hash;

        let mut templates = self.templates.write();
        let mut key_index = self.key_index.write();
        let mut stats = self.stats.write();

        // Evict if full
        if templates.len() >= self.max_size {
            self.evict_lru_internal(&mut templates, &mut key_index);
            stats.evictions += 1;
        }

        key_index.insert(key, hash);
        templates.insert(hash, Arc::clone(&template));
        stats.insertions += 1;

        debug!(hash, "SqlTemplateCache::register()");
        template
    }

    /// Register a template by hash (for pre-computed hashes).
    #[inline]
    pub fn register_by_hash(&self, hash: u64, sql: impl AsRef<str>) -> Arc<SqlTemplate> {
        let template = Arc::new(SqlTemplate::new(sql));

        let mut templates = self.templates.write();
        let mut stats = self.stats.write();

        if templates.len() >= self.max_size {
            let mut key_index = self.key_index.write();
            self.evict_lru_internal(&mut templates, &mut key_index);
            stats.evictions += 1;
        }

        templates.insert(hash, Arc::clone(&template));
        stats.insertions += 1;

        template
    }

    /// Get a template by string key (returns Arc for zero-copy).
    ///
    /// # Performance
    ///
    /// This is the fastest way to get cached SQL:
    /// - Hash lookup: ~5ns
    /// - Returns Arc<SqlTemplate> (no allocation)
    #[inline]
    pub fn get(&self, key: &str) -> Option<Arc<SqlTemplate>> {
        let hash = {
            let key_index = self.key_index.read();
            match key_index.get(key) {
                Some(&h) => h,
                None => {
                    drop(key_index); // Release read lock before write
                    let mut stats = self.stats.write();
                    stats.misses += 1;
                    return None;
                }
            }
        };

        let templates = self.templates.read();
        if let Some(template) = templates.get(&hash) {
            template.touch();
            let mut stats = self.stats.write();
            stats.hits += 1;
            return Some(Arc::clone(template));
        }

        let mut stats = self.stats.write();
        stats.misses += 1;
        None
    }

    /// Get a template by pre-computed hash (fastest path).
    ///
    /// # Performance
    ///
    /// ~3-5ns for cache hit with pre-computed hash.
    #[inline(always)]
    pub fn get_by_hash(&self, hash: u64) -> Option<Arc<SqlTemplate>> {
        let templates = self.templates.read();
        if let Some(template) = templates.get(&hash) {
            template.touch();
            // Skip stats update for maximum performance
            return Some(Arc::clone(template));
        }
        None
    }

    /// Get the SQL string directly (convenience method).
    #[inline]
    pub fn get_sql(&self, key: &str) -> Option<Arc<str>> {
        self.get(key).map(|t| t.sql_arc())
    }

    /// Get or compute a template.
    #[inline]
    pub fn get_or_register<F>(&self, key: impl Into<Cow<'static, str>>, f: F) -> Arc<SqlTemplate>
    where
        F: FnOnce() -> String,
    {
        let key = key.into();

        // Fast path: check if exists
        if let Some(template) = self.get(&key) {
            return template;
        }

        // Slow path: compute and register
        let sql = f();
        self.register(key, sql)
    }

    /// Check if a key exists.
    #[inline]
    pub fn contains(&self, key: &str) -> bool {
        let key_index = self.key_index.read();
        key_index.contains_key(key)
    }

    /// Get cache statistics.
    pub fn stats(&self) -> CacheStats {
        self.stats.read().clone()
    }

    /// Get the number of cached templates.
    pub fn len(&self) -> usize {
        self.templates.read().len()
    }

    /// Check if the cache is empty.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Clear the cache.
    pub fn clear(&self) {
        self.templates.write().clear();
        self.key_index.write().clear();
    }

    /// Evict least recently used templates (internal, assumes locks held).
    fn evict_lru_internal(
        &self,
        templates: &mut HashMap<u64, Arc<SqlTemplate>>,
        key_index: &mut HashMap<Cow<'static, str>, u64>,
    ) {
        use std::sync::atomic::Ordering;

        let to_evict = templates.len() / 4;
        if to_evict == 0 {
            return;
        }

        // Find templates with oldest access times (partial select —
        // only the `to_evict` oldest need to be up front).
        let mut entries: Vec<_> = templates
            .iter()
            .map(|(&hash, t)| (hash, t.last_access.load(Ordering::Relaxed)))
            .collect();
        entries.select_nth_unstable_by(to_evict - 1, |a, b| a.1.cmp(&b.1));

        // Collect the set of hashes to evict, remove their templates, then
        // prune key_index in a single pass (avoids O(evicted * key_index.len())).
        let evicted: std::collections::HashSet<u64> = entries
            .into_iter()
            .take(to_evict)
            .map(|(hash, _)| {
                templates.remove(&hash);
                hash
            })
            .collect();
        key_index.retain(|_, h| !evicted.contains(h));
    }
}

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

// =============================================================================
// Global Template Cache (for zero-overhead repeated queries)
// =============================================================================

/// Global SQL template cache for maximum performance.
///
/// Use this for queries that are repeated many times with only parameter changes.
/// The global cache avoids the overhead of passing cache references around.
///
/// # Examples
///
/// ```rust
/// use prax_query::cache::{global_template_cache, register_global_template};
///
/// // Pre-register common queries at startup
/// register_global_template("users_by_id", "SELECT * FROM users WHERE id = $1");
///
/// // Later, get the cached SQL (~5ns)
/// if let Some(template) = global_template_cache().get("users_by_id") {
///     println!("SQL: {}", template.sql());
/// }
/// ```
static GLOBAL_TEMPLATE_CACHE: std::sync::OnceLock<SqlTemplateCache> = std::sync::OnceLock::new();

/// Get the global SQL template cache.
#[inline(always)]
pub fn global_template_cache() -> &'static SqlTemplateCache {
    GLOBAL_TEMPLATE_CACHE.get_or_init(|| SqlTemplateCache::new(10000))
}

/// Register a template in the global cache.
#[inline]
pub fn register_global_template(
    key: impl Into<Cow<'static, str>>,
    sql: impl AsRef<str>,
) -> Arc<SqlTemplate> {
    global_template_cache().register(key, sql)
}

/// Get a template from the global cache.
#[inline(always)]
pub fn get_global_template(key: &str) -> Option<Arc<SqlTemplate>> {
    global_template_cache().get(key)
}

/// Pre-compute a query hash for repeated lookups.
///
/// Use this when you have a query key that will be used many times.
/// Computing the hash once and using `get_by_hash` is faster than
/// string key lookups.
#[inline]
pub fn precompute_query_hash(key: &str) -> u64 {
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    key.hash(&mut hasher);
    hasher.finish()
}

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

    #[test]
    fn test_query_cache_basic() {
        let cache = QueryCache::new(10);

        cache.insert("users_by_id", "SELECT * FROM users WHERE id = $1");
        assert!(cache.contains("users_by_id"));

        let sql = cache.get("users_by_id");
        assert_eq!(sql, Some("SELECT * FROM users WHERE id = $1".to_string()));
    }

    #[test]
    fn test_query_cache_get_or_insert() {
        let cache = QueryCache::new(10);

        let sql1 = cache.get_or_insert("test", || "SELECT 1".to_string());
        assert_eq!(sql1, "SELECT 1");

        let sql2 = cache.get_or_insert("test", || "SELECT 2".to_string());
        assert_eq!(sql2, "SELECT 1"); // Should return cached value
    }

    #[test]
    fn test_query_cache_stats() {
        let cache = QueryCache::new(10);

        cache.insert("test", "SELECT 1");
        cache.get("test"); // Hit
        cache.get("test"); // Hit
        cache.get("missing"); // Miss

        let stats = cache.stats();
        assert_eq!(stats.hits, 2);
        assert_eq!(stats.misses, 1);
        assert_eq!(stats.insertions, 1);
    }

    #[test]
    fn test_query_cache_evicts_least_accessed() {
        let cache = QueryCache::new(3);

        cache.insert("a", "SELECT 'a'");
        cache.insert("b", "SELECT 'b'");
        cache.insert("c", "SELECT 'c'");

        // Access "a" repeatedly so it has the highest access count.
        for _ in 0..5 {
            cache.get("a");
        }

        // Cache is full; this insert must evict exactly one of the
        // never-accessed entries ("b" or "c") and retain "a".
        cache.insert("d", "SELECT 'd'");

        assert!(cache.len() <= cache.max_size());
        assert!(cache.contains("a"));
        assert!(cache.contains("d"));
        let retained = cache.contains("b") as usize + cache.contains("c") as usize;
        assert_eq!(retained, 1, "exactly one of b/c should be evicted");

        // Repeated inserts at capacity never grow beyond max_size.
        for i in 0..8 {
            cache.insert(format!("extra_{i}"), "SELECT 1");
            assert!(cache.len() <= cache.max_size());
        }
    }

    #[test]
    fn test_count_placeholders_postgres() {
        assert_eq!(count_placeholders("SELECT * FROM users WHERE id = $1"), 1);
        assert_eq!(
            count_placeholders("SELECT * FROM users WHERE id = $1 AND name = $2"),
            2
        );
        assert_eq!(count_placeholders("SELECT * FROM users WHERE id = $10"), 10);
    }

    #[test]
    fn test_count_placeholders_mysql() {
        assert_eq!(count_placeholders("SELECT * FROM users WHERE id = ?"), 1);
        assert_eq!(
            count_placeholders("SELECT * FROM users WHERE id = ? AND name = ?"),
            2
        );
    }

    #[test]
    fn test_query_hash() {
        let hash1 = QueryHash::new("SELECT * FROM users");
        let hash2 = QueryHash::new("SELECT * FROM users");
        let hash3 = QueryHash::new("SELECT * FROM posts");

        assert_eq!(hash1, hash2);
        assert_ne!(hash1, hash3);
    }

    #[test]
    fn test_patterns() {
        let key = patterns::select_by_id("users");
        assert!(key.key.starts_with("select_by_id:"));
    }

    // =========================================================================
    // SqlTemplateCache Tests
    // =========================================================================

    #[test]
    fn test_sql_template_cache_basic() {
        let cache = SqlTemplateCache::new(100);

        let template = cache.register("users_by_id", "SELECT * FROM users WHERE id = $1");
        assert_eq!(template.sql(), "SELECT * FROM users WHERE id = $1");
        assert_eq!(template.param_count, 1);
    }

    #[test]
    fn test_sql_template_cache_get() {
        let cache = SqlTemplateCache::new(100);

        cache.register("test_query", "SELECT * FROM test WHERE x = $1");

        let result = cache.get("test_query");
        assert!(result.is_some());
        assert_eq!(result.unwrap().sql(), "SELECT * FROM test WHERE x = $1");

        let missing = cache.get("nonexistent");
        assert!(missing.is_none());
    }

    #[test]
    fn test_sql_template_cache_get_by_hash() {
        let cache = SqlTemplateCache::new(100);

        let template = cache.register("fast_query", "SELECT 1");
        let hash = template.hash;

        // Get by hash should be very fast
        let result = cache.get_by_hash(hash);
        assert!(result.is_some());
        assert_eq!(result.unwrap().sql(), "SELECT 1");
    }

    #[test]
    fn test_sql_template_cache_get_or_register() {
        let cache = SqlTemplateCache::new(100);

        let t1 = cache.get_or_register("computed", || "SELECT * FROM computed".to_string());
        assert_eq!(t1.sql(), "SELECT * FROM computed");

        // Second call should return cached version
        let t2 = cache.get_or_register("computed", || panic!("Should not be called"));
        assert_eq!(t2.sql(), "SELECT * FROM computed");
        assert_eq!(t1.hash, t2.hash);
    }

    #[test]
    fn test_sql_template_cache_stats() {
        let cache = SqlTemplateCache::new(100);

        cache.register("q1", "SELECT 1");
        cache.get("q1"); // Hit
        cache.get("q1"); // Hit
        cache.get("missing"); // Miss

        let stats = cache.stats();
        assert_eq!(stats.hits, 2);
        assert_eq!(stats.misses, 1);
        assert_eq!(stats.insertions, 1);
    }

    #[test]
    fn test_global_template_cache() {
        // Register in global cache
        let template = register_global_template("global_test", "SELECT * FROM global");
        assert_eq!(template.sql(), "SELECT * FROM global");

        // Retrieve from global cache
        let result = get_global_template("global_test");
        assert!(result.is_some());
        assert_eq!(result.unwrap().sql(), "SELECT * FROM global");
    }

    #[test]
    fn test_precompute_query_hash() {
        let hash1 = precompute_query_hash("test_key");
        let hash2 = precompute_query_hash("test_key");
        let hash3 = precompute_query_hash("other_key");

        assert_eq!(hash1, hash2);
        assert_ne!(hash1, hash3);
    }

    #[test]
    fn test_execution_plan_cache() {
        let cache = ExecutionPlanCache::new(100);

        // Register a plan
        let plan = cache.register(
            "users_by_email",
            "SELECT * FROM users WHERE email = $1",
            PlanHint::IndexScan("users_email_idx".into()),
        );
        assert_eq!(plan.sql.as_ref(), "SELECT * FROM users WHERE email = $1");

        // Get cached plan
        let result = cache.get("users_by_email");
        assert!(result.is_some());
        assert!(matches!(result.unwrap().hint, PlanHint::IndexScan(_)));
    }
}

// ============================================================================
// Execution Plan Caching
// ============================================================================

/// Hints for query execution optimization.
///
/// These hints can be used by database engines to optimize query execution.
/// Different databases support different hints - the engine implementation
/// decides how to apply them.
#[derive(Debug, Clone, Default)]
pub enum PlanHint {
    /// No specific hint.
    #[default]
    None,
    /// Force use of a specific index.
    IndexScan(String),
    /// Force a sequential scan (for analytics queries).
    SeqScan,
    /// Enable parallel execution.
    Parallel(u32),
    /// Cache this query's execution plan.
    CachePlan,
    /// Set a timeout for this query.
    Timeout(std::time::Duration),
    /// Custom database-specific hint.
    Custom(String),
}

/// A cached execution plan with optimization hints.
#[derive(Debug)]
pub struct ExecutionPlan {
    /// The SQL query.
    pub sql: Arc<str>,
    /// Pre-computed hash for fast lookup.
    pub hash: u64,
    /// Execution hint.
    pub hint: PlanHint,
    /// Estimated cost (if available from EXPLAIN).
    pub estimated_cost: Option<f64>,
    /// Number of times this plan has been used.
    use_count: std::sync::atomic::AtomicU64,
    /// Average execution time in microseconds.
    avg_execution_us: std::sync::atomic::AtomicU64,
}

/// Compute a hash for a string.
fn compute_hash(s: &str) -> u64 {
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    s.hash(&mut hasher);
    hasher.finish()
}

impl ExecutionPlan {
    /// Create a new execution plan.
    pub fn new(sql: impl AsRef<str>, hint: PlanHint) -> Self {
        let sql_str = sql.as_ref();
        Self {
            sql: Arc::from(sql_str),
            hash: compute_hash(sql_str),
            hint,
            estimated_cost: None,
            use_count: std::sync::atomic::AtomicU64::new(0),
            avg_execution_us: std::sync::atomic::AtomicU64::new(0),
        }
    }

    /// Create with estimated cost.
    pub fn with_cost(sql: impl AsRef<str>, hint: PlanHint, cost: f64) -> Self {
        let sql_str = sql.as_ref();
        Self {
            sql: Arc::from(sql_str),
            hash: compute_hash(sql_str),
            hint,
            estimated_cost: Some(cost),
            use_count: std::sync::atomic::AtomicU64::new(0),
            avg_execution_us: std::sync::atomic::AtomicU64::new(0),
        }
    }

    /// Record an execution with timing.
    pub fn record_execution(&self, duration_us: u64) {
        let old_count = self
            .use_count
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let old_avg = self
            .avg_execution_us
            .load(std::sync::atomic::Ordering::Relaxed);

        // Update running average
        let new_avg = if old_count == 0 {
            duration_us
        } else {
            // Weighted average: (old_avg * old_count + new_value) / (old_count + 1)
            (old_avg * old_count + duration_us) / (old_count + 1)
        };

        self.avg_execution_us
            .store(new_avg, std::sync::atomic::Ordering::Relaxed);
    }

    /// Get the use count.
    pub fn use_count(&self) -> u64 {
        self.use_count.load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Get the average execution time in microseconds.
    pub fn avg_execution_us(&self) -> u64 {
        self.avg_execution_us
            .load(std::sync::atomic::Ordering::Relaxed)
    }
}

/// Cache for query execution plans.
///
/// This cache stores not just SQL strings but also execution hints and
/// performance metrics for each query, enabling adaptive optimization.
///
/// # Example
///
/// ```rust
/// use prax_query::cache::{ExecutionPlanCache, PlanHint};
///
/// let cache = ExecutionPlanCache::new(1000);
///
/// // Register a plan with an index hint
/// let plan = cache.register(
///     "find_user_by_email",
///     "SELECT * FROM users WHERE email = $1",
///     PlanHint::IndexScan("idx_users_email".into()),
/// );
///
/// // Get the plan later
/// if let Some(plan) = cache.get("find_user_by_email") {
///     println!("Using plan with hint: {:?}", plan.hint);
/// }
/// ```
#[derive(Debug)]
pub struct ExecutionPlanCache {
    /// Maximum number of plans to cache.
    max_size: usize,
    /// Cached plans.
    plans: parking_lot::RwLock<HashMap<u64, Arc<ExecutionPlan>>>,
    /// Key to hash lookup.
    key_index: parking_lot::RwLock<HashMap<Cow<'static, str>, u64>>,
}

impl ExecutionPlanCache {
    /// Create a new execution plan cache.
    pub fn new(max_size: usize) -> Self {
        Self {
            max_size,
            plans: parking_lot::RwLock::new(HashMap::with_capacity(max_size / 2)),
            key_index: parking_lot::RwLock::new(HashMap::with_capacity(max_size / 2)),
        }
    }

    /// Register a new execution plan.
    pub fn register(
        &self,
        key: impl Into<Cow<'static, str>>,
        sql: impl AsRef<str>,
        hint: PlanHint,
    ) -> Arc<ExecutionPlan> {
        let key = key.into();
        let plan = Arc::new(ExecutionPlan::new(sql, hint));
        let hash = plan.hash;

        let mut plans = self.plans.write();
        let mut key_index = self.key_index.write();

        // Evict if at capacity
        if plans.len() >= self.max_size && !plans.contains_key(&hash) {
            // Simple eviction: remove least used
            if let Some((&evict_hash, _)) = plans.iter().min_by_key(|(_, p)| p.use_count()) {
                plans.remove(&evict_hash);
                key_index.retain(|_, &mut v| v != evict_hash);
            }
        }

        plans.insert(hash, Arc::clone(&plan));
        key_index.insert(key, hash);

        plan
    }

    /// Register a plan with estimated cost.
    pub fn register_with_cost(
        &self,
        key: impl Into<Cow<'static, str>>,
        sql: impl AsRef<str>,
        hint: PlanHint,
        cost: f64,
    ) -> Arc<ExecutionPlan> {
        let key = key.into();
        let plan = Arc::new(ExecutionPlan::with_cost(sql, hint, cost));
        let hash = plan.hash;

        let mut plans = self.plans.write();
        let mut key_index = self.key_index.write();

        if plans.len() >= self.max_size
            && !plans.contains_key(&hash)
            && let Some((&evict_hash, _)) = plans.iter().min_by_key(|(_, p)| p.use_count())
        {
            plans.remove(&evict_hash);
            key_index.retain(|_, &mut v| v != evict_hash);
        }

        plans.insert(hash, Arc::clone(&plan));
        key_index.insert(key, hash);

        plan
    }

    /// Get a cached execution plan.
    pub fn get(&self, key: &str) -> Option<Arc<ExecutionPlan>> {
        let hash = {
            let key_index = self.key_index.read();
            *key_index.get(key)?
        };

        self.plans.read().get(&hash).cloned()
    }

    /// Get a plan by its hash.
    pub fn get_by_hash(&self, hash: u64) -> Option<Arc<ExecutionPlan>> {
        self.plans.read().get(&hash).cloned()
    }

    /// Get or create a plan.
    pub fn get_or_register<F>(
        &self,
        key: impl Into<Cow<'static, str>>,
        sql_fn: F,
        hint: PlanHint,
    ) -> Arc<ExecutionPlan>
    where
        F: FnOnce() -> String,
    {
        let key = key.into();

        // Fast path: check if exists
        if let Some(plan) = self.get(key.as_ref()) {
            return plan;
        }

        // Slow path: create and register
        self.register(key, sql_fn(), hint)
    }

    /// Record execution timing for a plan.
    pub fn record_execution(&self, key: &str, duration_us: u64) {
        if let Some(plan) = self.get(key) {
            plan.record_execution(duration_us);
        }
    }

    /// Get plans sorted by average execution time (slowest first).
    pub fn slowest_queries(&self, limit: usize) -> Vec<Arc<ExecutionPlan>> {
        let plans = self.plans.read();
        let mut sorted: Vec<_> = plans.values().cloned().collect();
        sorted.sort_by_key(|a| std::cmp::Reverse(a.avg_execution_us()));
        sorted.truncate(limit);
        sorted
    }

    /// Get plans sorted by use count (most used first).
    pub fn most_used(&self, limit: usize) -> Vec<Arc<ExecutionPlan>> {
        let plans = self.plans.read();
        let mut sorted: Vec<_> = plans.values().cloned().collect();
        sorted.sort_by_key(|a| std::cmp::Reverse(a.use_count()));
        sorted.truncate(limit);
        sorted
    }

    /// Clear all cached plans.
    pub fn clear(&self) {
        self.plans.write().clear();
        self.key_index.write().clear();
    }

    /// Get the number of cached plans.
    pub fn len(&self) -> usize {
        self.plans.read().len()
    }

    /// Check if the cache is empty.
    pub fn is_empty(&self) -> bool {
        self.plans.read().is_empty()
    }
}