sz_orm_core/

plan_cache.rs

//! v3.2.0 查询计划缓存
//!
//! 缓存 SQL 解析结果（AST）与查询优化结果，相同 SQL 模板第二次起跳过解析/优化（≤1μs）。
//! 与 L2Cache（数据缓存）职责分离：L2Cache 缓存查询结果数据，PlanCache 缓存查询计划。
//!
//! 特性：
//! - SQL 归一化（忽略空白/注释/参数顺序，参数值替换为占位符）
//! - xxHash 64bit 键生成（无碰撞差分测试验证）
//! - 双缓存（parse_cache + optimize_cache）
//! - LRU 淘汰（arena 双向链表 O(1)）
//! - 表级精确失效（table_index 索引）
//! - 命中率统计（原子计数器无锁）

use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use parking_lot::RwLock;
use sqlparser::ast::Statement;
use sqlparser::dialect::GenericDialect;
use sqlparser::parser::Parser;
use std::hash::Hasher;
use twox_hash::XxHash64;

// ─── SqlNormalizer ───────────────────────────────────────────────

/// SQL 归一化器
///
/// 将 SQL 文本归一化为标准形式：忽略大小写差异、空白差异、参数值差异。
/// 相同语义不同写法的 SQL 归一化后产生相同文本，用于缓存键生成。
pub struct SqlNormalizer;

impl SqlNormalizer {
    /// 归一化 SQL 文本
    ///
    /// 返回归一化后的 SQL 文本。参数值（如 WHERE id = 42 中的 42）
    /// 不会被特殊处理——调用方应使用参数化查询（WHERE id = ?），
    /// 归一化仅处理大小写和空白差异。
    pub fn normalize(sql: &str) -> String {
        let dialect = GenericDialect {};
        match Parser::parse_sql(&dialect, sql) {
            Ok(statements) => {
                if statements.is_empty() {
                    return sql.trim().to_lowercase();
                }
                let normalized: Vec<String> =
                    statements.iter().map(|stmt| stmt.to_string()).collect();
                normalized.join("; ")
            }
            Err(_) => {
                let trimmed: String = sql.split_whitespace().collect::<Vec<_>>().join(" ");
                trimmed.to_lowercase()
            }
        }
    }

    /// 从 SQL 中提取依赖的表名列表
    ///
    /// 遍历 AST 提取所有表引用（FROM / JOIN / INTO / UPDATE 等）。
    pub fn extract_tables(sql: &str) -> Vec<String> {
        let dialect = GenericDialect {};
        let mut tables = Vec::new();

        if let Ok(statements) = Parser::parse_sql(&dialect, sql) {
            for stmt in &statements {
                Self::extract_tables_from_stmt(stmt, &mut tables);
            }
        }

        if tables.is_empty() {
            Self::extract_tables_from_str(sql, &mut tables);
        }

        tables.sort();
        tables.dedup();
        tables
    }

    fn extract_tables_from_stmt(stmt: &Statement, tables: &mut Vec<String>) {
        use sqlparser::ast::SetExpr;

        match stmt {
            Statement::Query(query) => {
                if let SetExpr::Select(select) = &*query.body {
                    for from in &select.from {
                        Self::extract_table_factor(&from.relation, tables);
                        for join in &from.joins {
                            Self::extract_table_factor(&join.relation, tables);
                        }
                    }
                }
            }
            Statement::Insert(_) => {}
            Statement::Update { table, .. } => {
                Self::extract_table_factor(&table.relation, tables);
            }
            Statement::Delete(delete) => {
                for table_name in &delete.tables {
                    let full_name = table_name
                        .0
                        .iter()
                        .map(|i| i.value.clone())
                        .collect::<Vec<_>>()
                        .join(".");
                    tables.push(full_name);
                }
            }
            _ => {}
        }
    }

    fn extract_table_factor(factor: &sqlparser::ast::TableFactor, tables: &mut Vec<String>) {
        use sqlparser::ast::TableFactor;
        if let TableFactor::Table { name, .. } = factor {
            let full_name = name
                .0
                .iter()
                .map(|i| i.value.clone())
                .collect::<Vec<_>>()
                .join(".");
            tables.push(full_name);
        }
    }

    fn extract_tables_from_str(sql: &str, tables: &mut Vec<String>) {
        let lower = sql.to_lowercase();
        for keyword in ["into ", "from ", "update ", "join "] {
            let mut search_pos = 0;
            while let Some(pos) = lower[search_pos..].find(keyword) {
                let abs_pos = search_pos + pos;
                let rest = &sql[abs_pos + keyword.len()..];
                let table: String = rest
                    .chars()
                    .take_while(|c| c.is_alphanumeric() || *c == '_' || *c == '.')
                    .collect();
                if !table.is_empty() && !table.chars().all(|c| c.is_numeric()) {
                    tables.push(table);
                }
                search_pos = abs_pos + keyword.len();
            }
        }
    }
}

// ─── PlanCacheKey ────────────────────────────────────────────────

/// 查询计划缓存键
///
/// 由归一化 SQL 的 xxHash 64bit 哈希值 + 归一化 SQL 文本组成。
/// 哈希值用于快速查找，SQL 文本用于二次校验（防哈希碰撞）。
#[derive(Debug, Clone)]
pub struct PlanCacheKey {
    /// xxHash 64bit 哈希值
    pub hash: u64,
    /// 归一化 SQL 文本（用于碰撞校验）
    pub sql_normalized: String,
}

impl PlanCacheKey {
    /// 从原始 SQL 生成缓存键
    ///
    /// 1. 归一化 SQL（忽略大小写/空白差异）
    /// 2. 计算归一化 SQL 的 xxHash 64bit 哈希
    pub fn from_sql(sql: &str) -> Self {
        let sql_normalized = SqlNormalizer::normalize(sql);
        let mut h = XxHash64::with_seed(0);
        h.write(sql_normalized.as_bytes());
        let hash = h.finish();
        Self {
            hash,
            sql_normalized,
        }
    }
}

impl PartialEq for PlanCacheKey {
    fn eq(&self, other: &Self) -> bool {
        self.hash == other.hash && self.sql_normalized == other.sql_normalized
    }
}

impl Eq for PlanCacheKey {}

impl std::hash::Hash for PlanCacheKey {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.hash.hash(state);
    }
}

// ─── PlanCacheEntry ──────────────────────────────────────────────

/// 查询计划缓存条目
///
/// 存储解析后的 AST 和/或优化后的查询分析结果。
pub struct PlanCacheEntry {
    /// 解析后的 AST（parse_cache 条目）
    pub ast: Option<Arc<Statement>>,
    /// 优化后的查询分析（optimize_cache 条目）
    pub analysis: Option<Arc<String>>,
    /// 创建时间
    pub created_at: Instant,
    /// 依赖的表名列表（用于表级失效）
    pub tables: Vec<String>,
    /// TTL（可选，过期自动失效）
    pub ttl: Option<Duration>,
}

impl PlanCacheEntry {
    /// 检查条目是否已过期
    pub fn is_expired(&self) -> bool {
        if let Some(ttl) = self.ttl {
            self.created_at.elapsed() >= ttl
        } else {
            false
        }
    }
}

// ─── PlanCacheStats ──────────────────────────────────────────────

/// 查询计划缓存统计
///
/// 原子计数器无锁统计命中/未命中/淘汰次数。
pub struct PlanCacheStats {
    parse_hits: AtomicU64,
    parse_misses: AtomicU64,
    optimize_hits: AtomicU64,
    optimize_misses: AtomicU64,
    evictions: AtomicU64,
}

impl PlanCacheStats {
    fn new() -> Self {
        Self {
            parse_hits: AtomicU64::new(0),
            parse_misses: AtomicU64::new(0),
            optimize_hits: AtomicU64::new(0),
            optimize_misses: AtomicU64::new(0),
            evictions: AtomicU64::new(0),
        }
    }

    /// 解析缓存命中次数
    pub fn parse_hits(&self) -> u64 {
        self.parse_hits.load(Ordering::Relaxed)
    }

    /// 解析缓存未命中次数
    pub fn parse_misses(&self) -> u64 {
        self.parse_misses.load(Ordering::Relaxed)
    }

    /// 优化缓存命中次数
    pub fn optimize_hits(&self) -> u64 {
        self.optimize_hits.load(Ordering::Relaxed)
    }

    /// 优化缓存未命中次数
    pub fn optimize_misses(&self) -> u64 {
        self.optimize_misses.load(Ordering::Relaxed)
    }

    /// LRU 淘汰次数
    pub fn evictions(&self) -> u64 {
        self.evictions.load(Ordering::Relaxed)
    }

    /// 解析缓存命中率（0.0 ~ 1.0）
    pub fn parse_hit_rate(&self) -> f64 {
        let hits = self.parse_hits();
        let misses = self.parse_misses();
        let total = hits + misses;
        if total == 0 {
            0.0
        } else {
            hits as f64 / total as f64
        }
    }

    /// 优化缓存命中率（0.0 ~ 1.0）
    pub fn optimize_hit_rate(&self) -> f64 {
        let hits = self.optimize_hits();
        let misses = self.optimize_misses();
        let total = hits + misses;
        if total == 0 {
            0.0
        } else {
            hits as f64 / total as f64
        }
    }
}

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

// ─── PlanCacheStatsSnapshot ──────────────────────────────────────

/// 统计快照（用于读取一致性视图）
#[derive(Debug, Clone)]
pub struct PlanCacheStatsSnapshot {
    /// 解析缓存命中次数
    pub parse_hits: u64,
    /// 解析缓存未命中次数
    pub parse_misses: u64,
    /// 优化缓存命中次数
    pub optimize_hits: u64,
    /// 优化缓存未命中次数
    pub optimize_misses: u64,
    /// LRU 淘汰次数
    pub evictions: u64,
    /// 当前缓存条目数
    pub size: usize,
    /// 解析缓存命中率
    pub parse_hit_rate: f64,
    /// 优化缓存命中率
    pub optimize_hit_rate: f64,
}

// ─── LruOrder64 ──────────────────────────────────────────────────

/// u64 键的 LRU 双向链表（arena 实现，O(1) touch/remove/lru_key）
struct LruOrder64 {
    nodes: Vec<LruNode64>,
    free_list: Vec<usize>,
    index: HashMap<u64, usize>,
    head: Option<usize>,
    tail: Option<usize>,
}

struct LruNode64 {
    key: u64,
    prev: Option<usize>,
    next: Option<usize>,
}

impl LruOrder64 {
    fn new() -> Self {
        Self {
            nodes: Vec::new(),
            free_list: Vec::new(),
            index: HashMap::new(),
            head: None,
            tail: None,
        }
    }

    fn touch(&mut self, key: u64) {
        if let Some(&idx) = self.index.get(&key) {
            self.unlink(idx);
            self.link_tail(idx);
        } else {
            let idx = self.alloc_node(key);
            self.link_tail(idx);
            self.index.insert(key, idx);
        }
    }

    fn remove(&mut self, key: u64) {
        if let Some(idx) = self.index.remove(&key) {
            self.unlink(idx);
            self.free_node(idx);
        }
    }

    fn lru_key(&self) -> Option<u64> {
        self.head.map(|idx| self.nodes[idx].key)
    }

    fn clear(&mut self) {
        self.nodes.clear();
        self.free_list.clear();
        self.index.clear();
        self.head = None;
        self.tail = None;
    }

    fn len(&self) -> usize {
        self.index.len()
    }

    fn alloc_node(&mut self, key: u64) -> usize {
        if let Some(idx) = self.free_list.pop() {
            self.nodes[idx] = LruNode64 {
                key,
                prev: None,
                next: None,
            };
            idx
        } else {
            self.nodes.push(LruNode64 {
                key,
                prev: None,
                next: None,
            });
            self.nodes.len() - 1
        }
    }

    fn free_node(&mut self, idx: usize) {
        self.free_list.push(idx);
    }

    fn unlink(&mut self, idx: usize) {
        let prev = self.nodes[idx].prev;
        let next = self.nodes[idx].next;
        match prev {
            Some(p) => self.nodes[p].next = next,
            None => self.head = next,
        }
        match next {
            Some(n) => self.nodes[n].prev = prev,
            None => self.tail = prev,
        }
        self.nodes[idx].prev = None;
        self.nodes[idx].next = None;
    }

    fn link_tail(&mut self, idx: usize) {
        match self.tail {
            Some(t) => {
                self.nodes[idx].prev = Some(t);
                self.nodes[t].next = Some(idx);
            }
            None => {
                self.head = Some(idx);
            }
        }
        self.tail = Some(idx);
    }
}

// ─── PlanCache ───────────────────────────────────────────────────

/// 查询计划缓存
///
/// 双缓存架构：
/// - `parse_cache`：SQL → AST（解析结果缓存）
/// - `optimize_cache`：SQL → 优化分析（优化结果缓存）
///
/// 锁顺序约定：parse_cache → optimize_cache → access_order → table_index → stats
/// （按此顺序加锁，避免死锁）
pub struct PlanCache {
    /// 解析缓存（hash → entry）
    parse_cache: RwLock<HashMap<u64, PlanCacheEntry>>,
    /// 优化缓存（hash → entry）
    optimize_cache: RwLock<HashMap<u64, PlanCacheEntry>>,
    /// LRU 访问顺序（arena 双向链表）
    access_order: RwLock<LruOrder64>,
    /// 表级失效索引（table → `Vec<hash>`）
    table_index: RwLock<HashMap<String, Vec<u64>>>,
    /// 统计计数器
    stats: PlanCacheStats,
    /// 最大缓存条目数
    max_size: usize,
    /// 默认 TTL
    default_ttl: Option<Duration>,
    /// v7.4.0：LRU-K 访问次数（K=2，记录每个 key 被访问的次数）
    access_counts: RwLock<HashMap<u64, u32>>,
    /// v7.4.0：LRU-K 的 K 值
    lru_k: u32,
    /// v7.4.0：自适应容量下限
    min_capacity: usize,
    /// v7.4.0：自适应容量上限
    max_capacity: usize,
}

impl PlanCache {
    /// 创建新的查询计划缓存
    ///
    /// - `max_size`：最大缓存条目数（LRU 淘汰）
    /// - `default_ttl`：默认 TTL（None 表示永不过期）
    pub fn new(max_size: usize, default_ttl: Option<Duration>) -> Self {
        Self {
            parse_cache: RwLock::new(HashMap::new()),
            optimize_cache: RwLock::new(HashMap::new()),
            access_order: RwLock::new(LruOrder64::new()),
            table_index: RwLock::new(HashMap::new()),
            stats: PlanCacheStats::new(),
            max_size,
            default_ttl,
            access_counts: RwLock::new(HashMap::new()),
            lru_k: 2,
            min_capacity: max_size / 4,
            max_capacity: max_size * 4,
        }
    }

    /// 获取或解析 SQL
    ///
    /// 命中缓存时返回 AST + stats.parse_hits++ + LRU touch。
    /// 未命中时解析 SQL + 存入缓存 + stats.parse_misses++。
    pub fn get_or_parse(&self, sql: &str) -> Result<Arc<Statement>, String> {
        let key = PlanCacheKey::from_sql(sql);

        // 锁序约定：不得在持有 parse_cache 锁时获取 access_order 锁
        // （未命中路径以相反顺序获取两把锁，否则并发命中/未命中会死锁）
        let hit = {
            let cache = self.parse_cache.read();
            match cache.get(&key.hash) {
                Some(entry) if !entry.is_expired() => entry.ast.clone(),
                _ => None,
            }
        };
        if let Some(ast) = hit {
            self.stats.parse_hits.fetch_add(1, Ordering::Relaxed);
            self.access_order.write().touch(key.hash);
            self.access_counts.write().entry(key.hash).and_modify(|c| *c += 1).or_insert(1);
            return Ok(ast);
        }

        self.stats.parse_misses.fetch_add(1, Ordering::Relaxed);

        let dialect = GenericDialect {};
        let statements = Parser::parse_sql(&dialect, sql).map_err(|e| e.to_string())?;
        if statements.is_empty() {
            return Err("empty SQL".to_string());
        }
        let ast = Arc::new(statements.into_iter().next().unwrap());
        let tables = SqlNormalizer::extract_tables(sql);

        {
            let mut access_order = self.access_order.write();
            // 统一锁序：access_order → optimize_cache → parse_cache → table_index
            let mut optimize_cache = self.optimize_cache.write();
            let mut cache = self.parse_cache.write();
            let mut table_index = self.table_index.write();

            if cache.len() >= self.max_size {
                if let Some(lru_hash) = access_order.lru_key() {
                    access_order.remove(lru_hash);
                    cache.remove(&lru_hash);
                    optimize_cache.remove(&lru_hash);
                    Self::remove_from_table_index(&mut table_index, lru_hash);
                    self.stats.evictions.fetch_add(1, Ordering::Relaxed);
                }
            }

            let entry = PlanCacheEntry {
                ast: Some(ast.clone()),
                analysis: None,
                created_at: Instant::now(),
                tables: tables.clone(),
                ttl: self.default_ttl,
            };
            cache.insert(key.hash, entry);
            access_order.touch(key.hash);

            for table in &tables {
                table_index.entry(table.clone()).or_default().push(key.hash);
            }
        }

        Ok(ast)
    }

    /// 获取或优化 SQL
    ///
    /// 命中缓存时返回优化分析 + stats.optimize_hits++ + LRU touch。
    /// 未命中时返回 None + stats.optimize_misses++（调用方应执行优化后调用 `store_optimize`）。
    pub fn get_or_optimize(&self, sql: &str) -> Option<Arc<String>> {
        let key = PlanCacheKey::from_sql(sql);

        let hit = {
            let cache = self.optimize_cache.read();
            match cache.get(&key.hash) {
                Some(entry) if !entry.is_expired() => entry.analysis.clone(),
                _ => None,
            }
        };
        if let Some(analysis) = hit {
            self.stats.optimize_hits.fetch_add(1, Ordering::Relaxed);
            self.access_order.write().touch(key.hash);
            return Some(analysis);
        }

        self.stats.optimize_misses.fetch_add(1, Ordering::Relaxed);
        None
    }

    /// 存储优化结果
    ///
    /// 在 `get_or_optimize` 返回 None 后，调用方执行优化并存储结果。
    pub fn store_optimize(&self, sql: &str, analysis: Arc<String>) {
        let key = PlanCacheKey::from_sql(sql);
        let tables = SqlNormalizer::extract_tables(sql);

        let mut access_order = self.access_order.write();
        let mut cache = self.optimize_cache.write();
        let mut parse_cache = self.parse_cache.write();
        let mut table_index = self.table_index.write();

        if cache.len() >= self.max_size {
            if let Some(lru_hash) = access_order.lru_key() {
                access_order.remove(lru_hash);
                cache.remove(&lru_hash);
                parse_cache.remove(&lru_hash);
                Self::remove_from_table_index(&mut table_index, lru_hash);
                self.stats.evictions.fetch_add(1, Ordering::Relaxed);
            }
        }

        let entry = PlanCacheEntry {
            ast: None,
            analysis: Some(analysis),
            created_at: Instant::now(),
            tables: tables.clone(),
            ttl: self.default_ttl,
        };
        cache.insert(key.hash, entry);
        access_order.touch(key.hash);

        for table in &tables {
            table_index.entry(table.clone()).or_default().push(key.hash);
        }
    }

    /// 表级精确失效
    ///
    /// 失效所有引用指定表的缓存条目，返回失效条目数。
    pub fn invalidate_table(&self, table: &str) -> usize {
        // 统一锁序：access_order → optimize_cache → parse_cache → table_index
        let mut access_order = self.access_order.write();
        let mut optimize_cache = self.optimize_cache.write();
        let mut parse_cache = self.parse_cache.write();
        let mut table_index = self.table_index.write();
        let keys = table_index.remove(table).unwrap_or_default();

        if keys.is_empty() {
            return 0;
        }

        let count = keys.len();

        for &hash in &keys {
            access_order.remove(hash);
            parse_cache.remove(&hash);
            optimize_cache.remove(&hash);
        }

        for remaining_keys in table_index.values_mut() {
            remaining_keys.retain(|k| !keys.contains(k));
        }

        self.stats
            .evictions
            .fetch_add(count as u64, Ordering::Relaxed);
        count
    }

    /// 全量清空缓存
    pub fn invalidate_all(&self) {
        self.parse_cache.write().clear();
        self.optimize_cache.write().clear();
        self.access_order.write().clear();
        self.table_index.write().clear();
    }

    /// 获取统计快照
    pub fn stats(&self) -> PlanCacheStatsSnapshot {
        let size = self.access_order.read().len();
        PlanCacheStatsSnapshot {
            parse_hits: self.stats.parse_hits(),
            parse_misses: self.stats.parse_misses(),
            optimize_hits: self.stats.optimize_hits(),
            optimize_misses: self.stats.optimize_misses(),
            evictions: self.stats.evictions(),
            size,
            parse_hit_rate: self.stats.parse_hit_rate(),
            optimize_hit_rate: self.stats.optimize_hit_rate(),
        }
    }

    /// 当前缓存大小
    pub fn size(&self) -> usize {
        self.access_order.read().len()
    }

    fn remove_from_table_index(table_index: &mut HashMap<String, Vec<u64>>, hash: u64) {
        for keys in table_index.values_mut() {
            keys.retain(|k| *k != hash);
        }
    }
}

impl Default for PlanCache {
    fn default() -> Self {
        Self::new(1024, None)
    }
}

// ============================================================================
// v7.3.0 任务 1.5：参数类型指纹 + PlanCacheConfig + eviction_count
// ============================================================================

/// 查询计划缓存配置（v7.3.0）
#[derive(Debug, Clone)]
pub struct PlanCacheConfig {
    /// 最大缓存条目数
    pub capacity: usize,
    /// TTL（默认 5 分钟）
    pub ttl: Duration,
    /// 指纹是否纳入参数类型
    pub include_param_types: bool,
}

impl Default for PlanCacheConfig {
    fn default() -> Self {
        Self {
            capacity: 256,
            ttl: Duration::from_secs(300),
            include_param_types: true,
        }
    }
}

impl PlanCacheConfig {
    /// 创建默认配置
    pub fn new() -> Self {
        Self::default()
    }

    /// 设置容量
    pub fn with_capacity(mut self, capacity: usize) -> Self {
        self.capacity = capacity;
        self
    }

    /// 设置 TTL
    pub fn with_ttl(mut self, ttl: Duration) -> Self {
        self.ttl = ttl;
        self
    }

    /// 设置是否纳入参数类型
    pub fn with_param_types(mut self, include: bool) -> Self {
        self.include_param_types = include;
        self
    }
}

/// 生成包含参数类型的查询指纹（v7.3.0）
///
/// 使用 xxHash64 对 SQL + 参数类型生成指纹。
/// 相同 SQL 不同参数类型生成不同指纹，避免计划缓存错误命中。
///
/// # 生产调用点
///
/// `packages/sz-orm-core/src/plan_cache.rs` `fingerprint_with_types` 函数。
pub fn fingerprint_with_types(sql: &str, param_types: &[crate::DbType]) -> u64 {
    let mut hasher = XxHash64::with_seed(0);
    hasher.write(sql.as_bytes());
    for pt in param_types {
        hasher.write(&[*pt as u8]);
    }
    hasher.finish()
}

impl PlanCache {
    /// v7.3.0 任务 1.5：淘汰数原子计数器
    pub fn eviction_count(&self) -> u64 {
        self.stats.evictions.load(Ordering::Relaxed)
    }

    /// v7.3.0 任务 1.5：从 PlanCacheConfig 创建缓存
    pub fn with_config(config: &PlanCacheConfig) -> Self {
        Self::new(config.capacity, Some(config.ttl))
    }

    /// v7.3.0 任务 1.5：带参数类型指纹的 get_or_parse
    ///
    /// 相同 SQL 不同参数类型生成不同指纹，避免错误命中。
    pub fn get_or_parse_with_types(
        &self,
        sql: &str,
        param_types: &[crate::DbType],
    ) -> Result<Arc<Statement>, String> {
        let type_fingerprint = fingerprint_with_types(sql, param_types);

        let hit = {
            let cache = self.parse_cache.read();
            match cache.get(&type_fingerprint) {
                Some(entry) if !entry.is_expired() => entry.ast.clone(),
                _ => None,
            }
        };
        if let Some(ast) = hit {
            self.stats.parse_hits.fetch_add(1, Ordering::Relaxed);
            self.access_order.write().touch(type_fingerprint);
            return Ok(ast);
        }

        self.stats.parse_misses.fetch_add(1, Ordering::Relaxed);

        let dialect = GenericDialect {};
        let statements = Parser::parse_sql(&dialect, sql).map_err(|e| e.to_string())?;
        if statements.is_empty() {
            return Err("empty SQL".to_string());
        }
        let ast = Arc::new(statements.into_iter().next().unwrap());
        let tables = SqlNormalizer::extract_tables(sql);

        {
            let mut access_order = self.access_order.write();
            let mut optimize_cache = self.optimize_cache.write();
            let mut cache = self.parse_cache.write();
            let mut table_index = self.table_index.write();

            if cache.len() >= self.max_size {
                if let Some(lru_hash) = access_order.lru_key() {
                    access_order.remove(lru_hash);
                    cache.remove(&lru_hash);
                    optimize_cache.remove(&lru_hash);
                    Self::remove_from_table_index(&mut table_index, lru_hash);
                    self.stats.evictions.fetch_add(1, Ordering::Relaxed);
                }
            }

            let entry = PlanCacheEntry {
                ast: Some(ast.clone()),
                analysis: None,
                created_at: Instant::now(),
                tables: tables.clone(),
                ttl: self.default_ttl,
            };
            cache.insert(type_fingerprint, entry);
            access_order.touch(type_fingerprint);

            for table in &tables {
                table_index
                    .entry(table.clone())
                    .or_default()
                    .push(type_fingerprint);
            }
        }

        Ok(ast)
    }

    /// v7.4.0 任务 3.3：获取 key 的访问次数（LRU-K 统计）
    pub fn access_count(&self, hash: u64) -> u32 {
        self.access_counts.read().get(&hash).copied().unwrap_or(0)
    }

    /// v7.4.0 任务 3.3：LRU-K 的 K 值
    pub fn lru_k(&self) -> u32 {
        self.lru_k
    }

    /// v7.4.0 任务 3.3：解析缓存命中率
    pub fn parse_hit_rate(&self) -> f64 {
        let hits = self.stats.parse_hits.load(Ordering::Relaxed);
        let misses = self.stats.parse_misses.load(Ordering::Relaxed);
        let total = hits + misses;
        if total == 0 {
            0.0
        } else {
            hits as f64 / total as f64
        }
    }

    /// v7.4.0 任务 3.3：自适应容量调整
    ///
    /// 根据命中率动态调整缓存大小：
    /// - 命中率 > 80%：扩容（上限 max_capacity）
    /// - 命中率 < 30%：缩容（下限 min_capacity）
    /// - 其他：保持不变
    pub fn adjust_capacity(&mut self) -> bool {
        let hit_rate = self.parse_hit_rate();
        let old_size = self.max_size;
        if hit_rate > 0.8 && self.max_size < self.max_capacity {
            self.max_size = (self.max_size * 3 / 2).min(self.max_capacity);
        } else if hit_rate < 0.3 && self.max_size > self.min_capacity {
            self.max_size = (self.max_size * 2 / 3).max(self.min_capacity);
        }
        self.max_size != old_size
    }

    /// v7.4.0 任务 3.3：当前最大容量
    pub fn capacity(&self) -> usize {
        self.max_size
    }
}

// ─── 单元测试 ─────────────────────────────────────────────────────

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

    #[test]
    fn test_sql_normalizer_basic() {
        let sql1 = "SELECT * FROM users WHERE id = ?";
        let sql2 = "select * from users where id = ?";
        let n1 = SqlNormalizer::normalize(sql1);
        let n2 = SqlNormalizer::normalize(sql2);
        assert_eq!(n1, n2, "大小写差异应归一化");
    }

    #[test]
    fn test_sql_normalizer_whitespace() {
        let sql1 = "SELECT   *   FROM   users";
        let sql2 = "SELECT * FROM users";
        let n1 = SqlNormalizer::normalize(sql1);
        let n2 = SqlNormalizer::normalize(sql2);
        assert_eq!(n1, n2, "空白差异应归一化");
    }

    #[test]
    fn test_sql_normalizer_different_semantics() {
        let sql1 = "SELECT * FROM users WHERE id = ?";
        let sql2 = "SELECT * FROM orders WHERE id = ?";
        let n1 = SqlNormalizer::normalize(sql1);
        let n2 = SqlNormalizer::normalize(sql2);
        assert_ne!(n1, n2, "不同表名应产生不同归一化");
    }

    #[test]
    fn test_sql_normalizer_parse_error_fallback() {
        let sql = "this is not valid sql !!!";
        let normalized = SqlNormalizer::normalize(sql);
        assert!(!normalized.is_empty());
    }

    #[test]
    fn test_extract_tables_select() {
        let sql = "SELECT * FROM users JOIN orders ON users.id = orders.user_id";
        let tables = SqlNormalizer::extract_tables(sql);
        assert!(tables.contains(&"users".to_string()));
        assert!(tables.contains(&"orders".to_string()));
    }

    #[test]
    fn test_extract_tables_insert() {
        let sql = "INSERT INTO products (name) VALUES (?)";
        let tables = SqlNormalizer::extract_tables(sql);
        assert!(tables.contains(&"products".to_string()));
    }

    #[test]
    fn test_extract_tables_update() {
        let sql = "UPDATE products SET name = ? WHERE id = ?";
        let tables = SqlNormalizer::extract_tables(sql);
        assert!(tables.contains(&"products".to_string()));
    }

    #[test]
    fn test_extract_tables_delete() {
        let sql = "DELETE FROM products WHERE id = ?";
        let tables = SqlNormalizer::extract_tables(sql);
        assert!(
            tables.iter().any(|t| t.contains("products")),
            "应包含 products 表，实际: {:?}",
            tables
        );
    }

    #[test]
    fn test_plan_cache_key_same_sql() {
        let k1 = PlanCacheKey::from_sql("SELECT * FROM users WHERE id = ?");
        let k2 = PlanCacheKey::from_sql("select * from users where id = ?");
        assert_eq!(k1.hash, k2.hash, "相同 SQL 模板应产生相同 hash");
    }

    #[test]
    fn test_plan_cache_key_different_sql() {
        let k1 = PlanCacheKey::from_sql("SELECT * FROM users");
        let k2 = PlanCacheKey::from_sql("SELECT * FROM orders");
        assert_ne!(k1.hash, k2.hash, "不同 SQL 应产生不同 hash");
    }

    #[test]
    fn test_plan_cache_key_no_sensitive_data() {
        let key = PlanCacheKey::from_sql("SELECT * FROM users WHERE password = ?");
        assert!(
            !key.sql_normalized.contains("secret123"),
            "参数化查询缓存键不应包含参数值"
        );
    }

    #[test]
    fn test_plan_cache_new() {
        let cache = PlanCache::new(1024, None);
        assert_eq!(cache.size(), 0);
        let stats = cache.stats();
        assert_eq!(stats.size, 0);
        assert_eq!(stats.parse_hits, 0);
        assert_eq!(stats.parse_misses, 0);
    }

    #[test]
    fn test_plan_cache_default() {
        let cache = PlanCache::default();
        assert_eq!(cache.size(), 0);
    }

    #[test]
    fn test_get_or_parse_hit() {
        let cache = PlanCache::new(100, None);
        let sql = "SELECT * FROM users WHERE id = ?";

        let ast1 = cache.get_or_parse(sql).expect("parse");
        assert_eq!(cache.stats().parse_misses, 1, "首次应 miss");

        let ast2 = cache.get_or_parse(sql).expect("parse");
        assert_eq!(cache.stats().parse_hits, 1, "第二次应 hit");
        assert_eq!(cache.stats().parse_misses, 1, "misses 不变");

        assert!(Arc::ptr_eq(&ast1, &ast2), "命中应返回相同 Arc");
    }

    #[test]
    fn test_get_or_parse_different_params_same_template() {
        let cache = PlanCache::new(100, None);
        let sql1 = "SELECT * FROM users WHERE id = ?";
        let sql2 = "select * from users where id = ?";

        cache.get_or_parse(sql1).expect("parse");
        cache.get_or_parse(sql2).expect("parse");

        assert_eq!(cache.stats().parse_hits, 1, "相同模板不同写法应命中");
        assert_eq!(cache.stats().parse_misses, 1);
    }

    #[test]
    fn test_get_or_parse_different_sql() {
        let cache = PlanCache::new(100, None);
        cache.get_or_parse("SELECT * FROM users").expect("parse");
        cache.get_or_parse("SELECT * FROM orders").expect("parse");

        assert_eq!(cache.stats().parse_misses, 2, "不同 SQL 应各 miss 一次");
        assert_eq!(cache.stats().parse_hits, 0);
    }

    #[test]
    fn test_get_or_optimize_miss_then_store_then_hit() {
        let cache = PlanCache::new(100, None);
        let sql = "SELECT * FROM users WHERE id = ?";

        assert!(cache.get_or_optimize(sql).is_none(), "首次应 miss");
        assert_eq!(cache.stats().optimize_misses, 1);

        cache.store_optimize(sql, Arc::new("optimized plan".to_string()));

        let result = cache.get_or_optimize(sql);
        assert!(result.is_some(), "存储后应 hit");
        assert_eq!(cache.stats().optimize_hits, 1);
        assert_eq!(*result.unwrap().as_ref(), "optimized plan");
    }

    #[test]
    fn test_invalidate_table_precise() {
        let cache = PlanCache::new(100, None);
        cache.get_or_parse("SELECT * FROM users").expect("parse");
        cache.get_or_parse("SELECT * FROM orders").expect("parse");
        assert_eq!(cache.size(), 2);

        let evicted = cache.invalidate_table("users");
        assert_eq!(evicted, 1, "应失效 1 条");
        assert_eq!(cache.size(), 1, "应剩余 1 条");

        let stats = cache.stats();
        assert!(stats.parse_hits == 0, "orders 缓存应不受影响");
        cache.get_or_parse("SELECT * FROM orders").expect("parse");
        assert_eq!(cache.stats().parse_hits, 1, "orders 应命中缓存");
    }

    #[test]
    fn test_invalidate_table_nonexistent() {
        let cache = PlanCache::new(100, None);
        cache.get_or_parse("SELECT * FROM users").expect("parse");
        let evicted = cache.invalidate_table("nonexistent");
        assert_eq!(evicted, 0, "不存在的表应返回 0");
        assert_eq!(cache.size(), 1, "缓存不应受影响");
    }

    #[test]
    fn test_invalidate_all() {
        let cache = PlanCache::new(100, None);
        cache.get_or_parse("SELECT * FROM users").expect("parse");
        cache.get_or_parse("SELECT * FROM orders").expect("parse");
        assert_eq!(cache.size(), 2);

        cache.invalidate_all();
        assert_eq!(cache.size(), 0, "全量清空后 size 应为 0");
    }

    #[test]
    fn test_lru_eviction() {
        let cache = PlanCache::new(3, None);
        cache.get_or_parse("SELECT * FROM t1").expect("parse");
        cache.get_or_parse("SELECT * FROM t2").expect("parse");
        cache.get_or_parse("SELECT * FROM t3").expect("parse");
        assert_eq!(cache.size(), 3);

        cache.get_or_parse("SELECT * FROM t4").expect("parse");
        assert_eq!(cache.size(), 3, "max_size=3 应保持 3 条");
        assert!(cache.stats().evictions >= 1, "应有淘汰");

        cache.get_or_parse("SELECT * FROM t1").expect("parse");
        assert!(cache.stats().parse_misses >= 4, "t1 被淘汰后应重新 miss");
    }

    #[test]
    fn test_lru_eviction_max_size_1() {
        let cache = PlanCache::new(1, None);
        cache.get_or_parse("SELECT * FROM t1").expect("parse");
        assert_eq!(cache.size(), 1);

        cache.get_or_parse("SELECT * FROM t2").expect("parse");
        assert_eq!(cache.size(), 1, "max_size=1 应保持 1 条");

        cache.get_or_parse("SELECT * FROM t1").expect("parse");
        assert!(cache.stats().parse_misses >= 3, "t1 应被淘汰后重新 miss");
    }

    #[test]
    fn test_stats_hit_rate() {
        let cache = PlanCache::new(100, None);
        let sql = "SELECT * FROM users";

        cache.get_or_parse(sql).expect("parse");
        cache.get_or_parse(sql).expect("parse");
        cache.get_or_parse(sql).expect("parse");

        let stats = cache.stats();
        assert_eq!(stats.parse_hits, 2);
        assert_eq!(stats.parse_misses, 1);
        assert!((stats.parse_hit_rate - (2.0 / 3.0)).abs() < 0.001);
    }

    #[test]
    fn test_stats_hit_rate_empty() {
        let cache = PlanCache::new(100, None);
        let stats = cache.stats();
        assert_eq!(stats.parse_hit_rate, 0.0, "空缓存命中率应为 0.0");
        assert_eq!(stats.optimize_hit_rate, 0.0);
    }

    #[test]
    fn test_ttl_expiration() {
        let cache = PlanCache::new(100, Some(Duration::from_nanos(1)));
        let sql = "SELECT * FROM users";

        cache.get_or_parse(sql).expect("parse");
        std::thread::sleep(Duration::from_millis(10));

        cache.get_or_parse(sql).expect("parse");
        assert!(cache.stats().parse_misses >= 2, "TTL 过期后应重新 miss");
    }

    #[test]
    fn test_plan_cache_concurrent_same_sql() {
        use std::sync::Arc;
        use std::thread;

        let cache = Arc::new(PlanCache::new(100, None));
        let sql = "SELECT * FROM users WHERE id = ?";
        let mut handles = Vec::new();

        for _ in 0..10 {
            let cache = cache.clone();
            handles.push(thread::spawn(move || {
                cache.get_or_parse(sql).expect("parse");
            }));
        }

        for h in handles {
            h.join().expect("thread");
        }

        assert!(cache.size() >= 1, "并发后应至少有 1 条缓存");
        assert!(
            cache.stats().parse_misses + cache.stats().parse_hits >= 10,
            "应有 10 次访问记录"
        );
    }
}
#[derive(Debug, Clone)]
pub struct LruKReplacer {
    k: usize,
    capacity: usize,
    access_history: std::collections::HashMap<u64, Vec<std::time::Instant>>,
}

impl LruKReplacer {
    pub fn new(k: usize, capacity: usize) -> Self {
        Self {
            k,
            capacity,
            access_history: std::collections::HashMap::new(),
        }
    }

    pub fn access(&mut self, key: u64) {
        let history = self.access_history.entry(key).or_default();
        history.push(std::time::Instant::now());
        if history.len() > self.k {
            history.remove(0);
        }
    }

    pub fn evict(&mut self) -> Option<u64> {
        if self.access_history.len() < self.capacity {
            return None;
        }
        let mut oldest_key: Option<u64> = None;
        let mut oldest_time: Option<std::time::Instant> = None;

        for (&key, history) in &self.access_history {
            let ref_time = if history.len() >= self.k {
                history[0]
            } else {
                std::time::Instant::now()
            };
            if oldest_time.is_none() || ref_time < oldest_time.unwrap() {
                oldest_time = Some(ref_time);
                oldest_key = Some(key);
            }
        }

        if let Some(key) = oldest_key {
            self.access_history.remove(&key);
        }
        oldest_key
    }

    pub fn hit_rate(&self) -> f64 {
        let total: usize = self.access_history.values().map(|h| h.len()).sum();
        if total == 0 {
            return 0.0;
        }
        let hits: usize = self
            .access_history
            .values()
            .map(|h| if h.len() >= self.k { 1 } else { 0 })
            .sum();
        hits as f64 / self.access_history.len().max(1) as f64
    }

    pub fn len(&self) -> usize {
        self.access_history.len()
    }

    pub fn is_empty(&self) -> bool {
        self.access_history.is_empty()
    }
}

#[derive(Debug, Clone)]
pub struct AdaptiveCacheCapacity {
    min: usize,
    max: usize,
    current: usize,
    hit_rate_window: std::collections::VecDeque<f64>,
    window_size: usize,
}

impl AdaptiveCacheCapacity {
    pub fn new(min: usize, max: usize) -> Self {
        Self {
            min,
            max,
            current: min,
            hit_rate_window: std::collections::VecDeque::new(),
            window_size: 100,
        }
    }

    pub fn record_hit_rate(&mut self, rate: f64) {
        if self.hit_rate_window.len() >= self.window_size {
            self.hit_rate_window.pop_front();
        }
        self.hit_rate_window.push_back(rate);
    }

    pub fn adapt(&mut self) -> bool {
        if self.hit_rate_window.is_empty() {
            return false;
        }
        let avg: f64 =
            self.hit_rate_window.iter().sum::<f64>() / self.hit_rate_window.len() as f64;
        let old = self.current;
        if avg < 0.85 {
            self.current = (self.current as f64 * 1.5) as usize;
        } else if avg > 0.95 {
            self.current = (self.current as f64 * 0.8) as usize;
        }
        self.current = self.current.clamp(self.min, self.max);
        self.current != old
    }

    pub fn current(&self) -> usize {
        self.current
    }

    pub fn min(&self) -> usize {
        self.min
    }

    pub fn max(&self) -> usize {
        self.max
    }
}