cqlite-core 0.15.0

//! Plan-cache HIT path lock-hygiene tests (issue #1595, Epic F/F5).
//!
//! These cover spec Requirement 2 ("Plan-cache hit path uses a shared (read)
//! lock") of the `lock-hygiene-querystats` change — the three scenarios the
//! spec-auditor (C) found unmet on test evidence:
//!
//! 1. A cache hit bumps the per-entry counter via a shard READ lock
//!    (`DashMap::get`) + relaxed `fetch_add` — never a shard write lock.
//! 2. Concurrent hits to the SAME cached plan lose no increments (exact counts,
//!    no wall-clock window).
//! 3. A non-reusable placeholder plan (`plan.table.is_none()`) is still evicted
//!    and the query proceeds via the cold path.
//!
//! This module is declared inside `engine.rs` (via `#[path]`), so it is a
//! descendant of the `engine` module and reads the private
//! `plan_cache`/`select_plan_cache` fields and the `pub hit_count` atomic
//! directly — no new public/`pub(crate)` accessor is introduced.

use super::{QueryCacheEntry, QueryEngine};
use crate::query::planner::{PlanType, QueryHints, QueryPlan};
use crate::query::{ParsedQuery, QueryType};
use crate::{Config, TableId};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Instant;
use tempfile::TempDir;

/// Build a query engine over an empty temp store. The returned `TempDir` must
/// be kept alive for the duration of the test (it backs the storage).
async fn new_test_engine(config: &Config) -> (QueryEngine, TempDir) {
    let temp_dir = TempDir::new().unwrap();
    let platform = Arc::new(crate::platform::Platform::new(config).await.unwrap());
    let storage = Arc::new(
        crate::storage::StorageEngine::open(
            temp_dir.path(),
            config,
            platform,
            #[cfg(feature = "state_machine")]
            None,
        )
        .await
        .unwrap(),
    );
    let schema = Arc::new(
        crate::schema::SchemaManager::new(temp_dir.path())
            .await
            .unwrap(),
    );
    let memory = Arc::new(crate::memory::MemoryManager::new(config).unwrap());
    let engine = QueryEngine::new(storage, schema, memory, config).unwrap();
    (engine, temp_dir)
}

/// A minimal legacy `QueryPlan`; `table` controls reusability
/// (`Some` => reusable, `None` => placeholder).
fn minimal_plan(table: Option<TableId>) -> QueryPlan {
    QueryPlan {
        plan_type: PlanType::PointLookup,
        table,
        estimated_cost: 1.0,
        estimated_rows: 0,
        selected_indexes: Vec::new(),
        steps: Vec::new(),
        hints: QueryHints::default(),
    }
}

fn seed_parsed(cql: &str, table: Option<TableId>) -> ParsedQuery {
    ParsedQuery {
        query_type: QueryType::Select,
        table,
        columns: vec!["*".to_string()],
        where_clause: None,
        values: Vec::new(),
        set_clause: HashMap::new(),
        order_by: Vec::new(),
        limit: None,
        cql: cql.to_string(),
    }
}

/// Insert a `QueryCacheEntry` directly into the legacy plan cache so a
/// subsequent `execute(cql)` is a genuine HIT of the read-lock path.
fn seed_plan_cache(engine: &QueryEngine, cql: &str, table: Option<TableId>) {
    engine.plan_cache.insert(
        cql.to_string(),
        QueryCacheEntry {
            parsed_query: seed_parsed(cql, table.clone()),
            plan: minimal_plan(table),
            cached_at: Instant::now(),
            hit_count: AtomicU64::new(0),
        },
    );
}

/// Scenario: "Cache hit bumps the counter without a write lock".
///
/// A cached simple-`WHERE id =` lookup is served by `execute`'s legacy
/// plan-cache HIT path, which does `plan_cache.get` (read lock) +
/// `hit_count.fetch_add(1, Relaxed)`. We read the counter while HOLDING a
/// shared `get` guard (a WRITE-lock hit path could not coexist with this live
/// read guard), then run the query and observe the bump.
#[tokio::test]
async fn plan_cache_hit_bumps_counter_via_read_lock() {
    let mut config = Config::default();
    config.query.query_cache_size = Some(8);
    let (engine, _temp) = new_test_engine(&config).await;

    let cql = "SELECT * FROM t WHERE id = 1";
    seed_plan_cache(&engine, cql, Some(TableId::new("t")));

    // Read the counter while holding a shared read guard: only possible because
    // the HIT path takes a shared (read) lock, not a write lock.
    {
        let guard = engine.plan_cache.get(cql).expect("seeded entry present");
        assert_eq!(guard.hit_count.load(Ordering::Relaxed), 0);
    }

    // The seeded plan has no lookup step, so the executor returns an error AFTER
    // the HIT path has already bumped the counters; the lock-hygiene behavior
    // under test is unaffected by the (ignored) execution outcome.
    let _ = engine.execute(cql).await;

    let hits = engine
        .plan_cache
        .get(cql)
        .expect("entry still cached after a hit")
        .hit_count
        .load(Ordering::Relaxed);
    assert_eq!(
        hits, 1,
        "a cache hit must bump the entry hit_count exactly once"
    );

    let stats = engine.stats();
    assert_eq!(stats.total_queries, 1);
    assert!(
        stats.cache_hit_ratio > 0.0,
        "cache_hit_ratio must be > 0 after a plan-cache hit"
    );
}

/// Scenario: "Concurrent hits to the same cached plan do not serialize on a
/// write lock" — and lose no increments.
///
/// N tasks on a multi-thread runtime hit the SAME cached plan concurrently.
/// After joining ALL tasks (no wall-clock window), the exact counter sums must
/// equal the issued counts: `total_queries == N`, per-entry `hit_count == N`,
/// and `cache_hit_ratio == 1.0` (every query was a hit).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrent_plan_cache_hits_lose_no_increments() {
    const TASKS: u64 = 64;

    let mut config = Config::default();
    config.query.query_cache_size = Some(8);
    let (engine, _temp) = new_test_engine(&config).await;
    let engine = Arc::new(engine);

    let cql = "SELECT * FROM t WHERE id = 1";
    seed_plan_cache(&engine, cql, Some(TableId::new("t")));

    let mut handles = Vec::with_capacity(TASKS as usize);
    for _ in 0..TASKS {
        let engine = Arc::clone(&engine);
        handles.push(tokio::spawn(async move {
            let _ = engine.execute(cql).await;
        }));
    }
    for h in handles {
        h.await.expect("query task panicked");
    }

    // Only after every task has joined do we snapshot — no timing boundary.
    let stats = engine.stats();
    assert_eq!(
        stats.total_queries, TASKS,
        "every issued query must be counted exactly once"
    );
    let hits = engine
        .plan_cache
        .get(cql)
        .expect("entry stays cached (all executes were hits, none re-cached)")
        .hit_count
        .load(Ordering::Relaxed);
    assert_eq!(
        hits, TASKS,
        "concurrent hits must not lose increments (exact per-entry hit_count)"
    );
    assert!(
        (stats.cache_hit_ratio - 1.0).abs() < f64::EPSILON,
        "all queries hit the cache => cache_hit_ratio == 1.0 (cache_hits == total_queries)"
    );
}

/// Scenario: "A non-reusable placeholder plan is still evicted".
///
/// A cached entry whose plan has no resolved table (`table.is_none()`) is a
/// placeholder. When a SELECT for that key runs, `execute_select_query`
/// resolves `HitOutcome::Placeholder` under the read guard, drops the guard,
/// then removes the entry and proceeds via the cold (modern) path — which
/// caches the OPTIMIZED plan in `select_plan_cache`, never back into the legacy
/// `plan_cache`.
#[tokio::test]
#[cfg(feature = "state_machine")]
async fn placeholder_plan_is_evicted_and_query_proceeds_cold() {
    let mut config = Config::default();
    config.query.query_cache_size = Some(8);
    let (engine, _temp) = new_test_engine(&config).await;

    // NOT a simple `WHERE id =` lookup, so `execute` routes it to
    // `execute_select_query`, which owns the placeholder-eviction branch.
    let cql = "SELECT * FROM t WHERE age > 5";
    seed_plan_cache(&engine, cql, None); // table.is_none() => placeholder
    assert!(
        engine.plan_cache.get(cql).is_some(),
        "placeholder entry seeded into the legacy plan cache"
    );

    let _ = engine.execute(cql).await;

    assert!(
        engine.plan_cache.get(cql).is_none(),
        "a non-reusable placeholder must be evicted from the legacy plan cache"
    );
    assert!(
        engine.select_plan_cache.get(cql).is_some(),
        "the query proceeded cold via the modern path (optimized plan now cached)"
    );
}