cqlite_core/query/
engine.rs

1//! Query engine implementation for CQLite
2//!
3//! This module provides the main query engine that coordinates between
4//! parsing, planning, and execution of CQL queries.
5
6// CQL (Cassandra Query Language) Reference:
7// https://cassandra.apache.org/doc/latest/cassandra/developing/cql/cql_singlefile.html
8//
9// This implements CQL v3.4.3+ for Apache Cassandra 5.0+
10// CQL is NOT SQL - it's a query language specifically designed for Cassandra's distributed architecture.
11
12use super::{
13    executor::{QueryExecutor, QueryResult},
14    parser::QueryParser,
15    planner::QueryPlanner,
16    prepared::PreparedQuery,
17    result::{QueryResultIterator, StreamingConfig},
18    QueryStats,
19};
20
21use super::engine_stats::AtomicQueryStats;
22#[cfg(feature = "state_machine")]
23use super::{
24    select_executor::SelectExecutor,
25    select_optimizer::{OptimizedQueryPlan, SelectOptimizer},
26    select_parser,
27};
28use crate::{
29    memory::MemoryManager, schema::SchemaManager, storage::StorageEngine, Config, Error, Result,
30    Value,
31};
32use dashmap::DashMap;
33use std::sync::atomic::{AtomicU64, Ordering};
34use std::sync::Arc;
35use std::time::Instant;
36
37/// Query cache entry
38///
39/// Crate-internal plan-cache entry with no external consumers, so the
40/// `hit_count: AtomicU64` field (issue #1595) is kept off the public semver
41/// surface by keeping this type `pub(crate)`.
42#[derive(Debug)]
43pub(crate) struct QueryCacheEntry {
44    /// Parsed query
45    pub parsed_query: super::ParsedQuery,
46    /// Query plan
47    pub plan: super::planner::QueryPlan,
48    /// Cache timestamp
49    pub cached_at: Instant,
50    /// Hit count. Lock-free (issue #1595): a plan-cache HIT bumps this via a
51    /// shard READ lock (`DashMap::get`) + relaxed `fetch_add`, so concurrent hits
52    /// to the same shard no longer contend on a write lock.
53    pub hit_count: AtomicU64,
54}
55
56// `AtomicU64` is not `Clone`, so the previous `#[derive(Clone)]` no longer
57// applies; preserve the public `Clone` surface by hand, snapshotting the counter
58// with a relaxed load (issue #1595).
59impl Clone for QueryCacheEntry {
60    fn clone(&self) -> Self {
61        Self {
62            parsed_query: self.parsed_query.clone(),
63            plan: self.plan.clone(),
64            cached_at: self.cached_at,
65            hit_count: AtomicU64::new(self.hit_count.load(Ordering::Relaxed)),
66        }
67    }
68}
69
70/// Schema availability status for diagnostic purposes
71#[derive(Debug, Clone)]
72pub enum SchemaStatus {
73    /// Schema is available and ready for queries
74    Available { keyspace: String, table: String },
75    /// Schema not found in registry
76    Missing { table: String, reason: String },
77    /// Schema extraction failed from SSTable
78    ExtractionFailed {
79        table: String,
80        cause: String,
81        suggestion: String,
82    },
83}
84
85/// Query engine with caching and statistics
86#[derive(Debug)]
87pub struct QueryEngine {
88    /// Query parser
89    parser: QueryParser,
90    /// Query planner
91    planner: QueryPlanner,
92    /// Query executor
93    executor: QueryExecutor,
94    /// Schema manager reference
95    schema_manager: Arc<SchemaManager>,
96    /// Advanced SELECT optimizer. `Arc` so prepared SELECTs can share the same
97    /// instance and reach the partition-targeted fast path (Issue #961).
98    #[cfg(feature = "state_machine")]
99    select_optimizer: Arc<SelectOptimizer>,
100    /// Advanced SELECT executor (shared with prepared SELECTs, Issue #961).
101    #[cfg(feature = "state_machine")]
102    select_executor: Arc<SelectExecutor>,
103    /// Prepared statement cache
104    prepared_cache: DashMap<String, Arc<PreparedQuery>>,
105    /// Query plan cache
106    plan_cache: DashMap<String, QueryCacheEntry>,
107    /// Issue #1587 (E5): plan cache for the modern (state_machine) SELECT path,
108    /// keyed by the exact CQL text. The optimized plan is a pure function of the
109    /// literal CQL, so a repeated identical ad-hoc SELECT reuses the plan instead
110    /// of re-parsing + re-optimizing. Kept separate from `plan_cache` (which
111    /// holds legacy `QueryPlan`s and feeds `CacheStats`) so neither disturbs the
112    /// other. Value is `(inserted_at, plan)` for the same simple-LRU eviction.
113    #[cfg(feature = "state_machine")]
114    select_plan_cache: DashMap<String, (Instant, Arc<OptimizedQueryPlan>)>,
115    /// Query statistics. Lock-free atomic counters (issue #1595): counters are
116    /// bumped with relaxed atomics instead of a `RwLock` taken 2–3× per query.
117    stats: AtomicQueryStats,
118    /// Configuration
119    config: Config,
120}
121
122impl QueryEngine {
123    /// Create a new query engine
124    pub fn new(
125        storage: Arc<StorageEngine>,
126        schema: Arc<SchemaManager>,
127        _memory: Arc<MemoryManager>,
128        config: &Config,
129    ) -> Result<Self> {
130        let parser = QueryParser::new(config);
131        let planner = QueryPlanner::new(schema.clone(), config);
132        let executor = QueryExecutor::new(storage.clone(), schema.clone(), config);
133
134        // Initialize advanced SELECT components
135        #[cfg(feature = "state_machine")]
136        let select_optimizer = Arc::new(SelectOptimizer::new(schema.clone(), storage.clone()));
137        // Issue #1582 (D6): wire the byte-bounded result budget from config so
138        // the `max_result_bytes` knob is load-bearing on the materializing path.
139        #[cfg(feature = "state_machine")]
140        let select_executor = Arc::new(
141            SelectExecutor::new(schema.clone(), storage)
142                .with_max_result_bytes(
143                    usize::try_from(config.query.max_result_bytes).unwrap_or(usize::MAX),
144                )
145                .with_max_result_rows(
146                    usize::try_from(config.query.max_result_rows).unwrap_or(usize::MAX),
147                ),
148        );
149
150        Ok(Self {
151            parser,
152            planner,
153            executor,
154            schema_manager: schema,
155            #[cfg(feature = "state_machine")]
156            select_optimizer,
157            #[cfg(feature = "state_machine")]
158            select_executor,
159            prepared_cache: DashMap::new(),
160            plan_cache: DashMap::new(),
161            #[cfg(feature = "state_machine")]
162            select_plan_cache: DashMap::new(),
163            stats: AtomicQueryStats::default(),
164            config: config.clone(),
165        })
166    }
167
168    /// Enforce the byte-bounded result budget (issue #1582 / D6) on a result
169    /// materialized by the LEGACY [`QueryExecutor`] — non-SELECT statements and
170    /// any cached legacy SELECT plan reused via the plan-cache HIT branch. Since
171    /// issue #1750 ad-hoc SELECTs route through the modern `SelectExecutor` (which
172    /// budgets itself); this guard covers the remaining legacy return sites,
173    /// INCLUDING a plan-cache HIT (roborev #1582 D6). Reuses the SAME estimator +
174    /// enforcement as the optimizer path. Applied exactly once at each site.
175    fn enforce_legacy_result_budget(&self, result: &QueryResult) -> Result<()> {
176        super::result_budget::enforce_materialized_rows(
177            &result.rows,
178            usize::try_from(self.config.query.max_result_bytes).unwrap_or(usize::MAX),
179            usize::try_from(self.config.query.max_result_rows).unwrap_or(usize::MAX),
180        )
181        .inspect_err(|e| {
182            self.inc_error_queries();
183            crate::observability::record_error(e, "query");
184        })
185    }
186
187    /// Increment the total queries counter (lock-free, issue #1595)
188    fn inc_total_queries(&self) {
189        self.stats.record_query();
190    }
191
192    /// Increment the error queries counter (lock-free, issue #1595)
193    fn inc_error_queries(&self) {
194        self.stats.record_error();
195    }
196
197    /// Record a plan-cache hit (lock-free, issue #1595). `cache_hit_ratio` is
198    /// derived at read time as `cache_hits / total_queries`.
199    fn record_cache_hit(&self) {
200        self.stats.record_cache_hit();
201    }
202
203    /// Execute a CQL query
204    ///
205    /// This is the parent of the query span tree (epic #1031, issue #1035): the
206    /// `query.execute` span created here is the context every read-path span
207    /// (issue #1034) and SELECT sub-span nests under. Bounded span attributes
208    /// (plan type, access path, rows returned) are recorded once the result is
209    /// known via [`Self::update_execution_stats`]; the query text is never
210    /// attached.
211    #[tracing::instrument(
212        name = "query.execute",
213        skip(self, cql),
214        fields(
215            cqlite.query.plan_type = tracing::field::Empty,
216            cqlite.query.access_path = tracing::field::Empty,
217            cqlite.query.rows = tracing::field::Empty,
218        )
219    )]
220    pub async fn execute(&self, cql: &str) -> Result<QueryResult> {
221        let start_time = Instant::now();
222        self.inc_total_queries();
223
224        // Route ALL SELECT statements through the modern optimizer +
225        // `SelectExecutor` (issue #1750). It populates `metadata.columns`, applies
226        // projection, reconstructs the partition-key column, and selects the
227        // partition-targeted fast path (#949/#956) from parsed query structure +
228        // schema — never from a substring/token-count guess. The retired
229        // `is_simple_id_lookup` fork (`WHERE id =` substring + ≤ 8-token check)
230        // diverted short point reads to the legacy `QueryExecutor`, returning
231        // column-less results and skipping projection — a no-heuristics mandate
232        // violation (#28) that flipped behavior with the PK name.
233        let trimmed_cql = cql.trim().to_uppercase();
234        if trimmed_cql.starts_with("SELECT") {
235            return self.execute_select_query(cql, start_time).await;
236        }
237
238        // Check plan cache first for non-SELECT queries. Issue #1595: serve the
239        // HIT under a shard READ lock (`DashMap::get`), bump the lock-free
240        // `hit_count`, clone the plan out, and DROP the guard before the `.await`
241        // so no shard lock is held across execution and concurrent hits do not
242        // contend on a write lock.
243        let cached_plan = self.plan_cache.get(cql).map(|entry| {
244            entry.hit_count.fetch_add(1, Ordering::Relaxed);
245            entry.plan.clone()
246        });
247        if let Some(plan) = cached_plan {
248            self.record_cache_hit();
249
250            let mut result =
251                crate::observability::record_result("query", self.executor.execute(&plan).await)?;
252            // Issue #1582 (D6): the LEGACY point-lookup path bypasses the
253            // SelectExecutor's byte budget; enforce it on the plan-cache HIT
254            // return too (roborev finding) so a single very wide row cannot
255            // exceed `max_result_bytes` without raising `ResultTooLarge`.
256            self.enforce_legacy_result_budget(&result)?;
257            self.update_execution_stats(&mut result, start_time);
258            return Ok(result);
259        }
260
261        let parsed_query = self.parser.parse(cql).inspect_err(|e| {
262            self.inc_error_queries();
263            crate::observability::record_error(e, "query");
264        })?;
265        let plan =
266            crate::observability::record_result("query", self.planner.plan(&parsed_query).await)?;
267
268        if self.config.query.query_cache_size.unwrap_or(0) > 0 {
269            self.cache_query_plan(cql, parsed_query, plan.clone());
270        }
271
272        let mut result =
273            crate::observability::record_result("query", self.executor.execute(&plan).await)?;
274        // Issue #1582 (D6): the LEGACY point-lookup path bypasses the
275        // SelectExecutor's byte budget; enforce it on the cold (cache-miss)
276        // return, matching the plan-cache-hit return above.
277        self.enforce_legacy_result_budget(&result)?;
278        self.update_execution_stats(&mut result, start_time);
279        Ok(result)
280    }
281
282    /// Execute a CQL query with streaming results (Issue #280)
283    ///
284    /// Returns a `QueryResultIterator` that yields rows incrementally via a bounded
285    /// channel, enabling memory-efficient processing of large result sets.
286    ///
287    /// # Arguments
288    ///
289    /// * `cql` - The CQL query string to execute (must be a SELECT statement)
290    /// * `config` - Streaming configuration (buffer size, chunk hints)
291    ///
292    /// # Errors
293    ///
294    /// Returns an error if:
295    /// - Query is not a SELECT statement
296    /// - SQL syntax is invalid
297    /// - Query execution fails
298    ///
299    /// # Memory Budget
300    ///
301    /// The streaming approach stays within the 128MB target by using bounded channels
302    /// and processing rows incrementally rather than materializing all results.
303    #[cfg(feature = "state_machine")]
304    pub async fn execute_streaming(
305        &self,
306        cql: &str,
307        config: StreamingConfig,
308    ) -> Result<QueryResultIterator> {
309        self.inc_total_queries();
310
311        if !cql.trim().to_uppercase().starts_with("SELECT") {
312            return Err(Error::query_execution(
313                "Streaming execution only supports SELECT queries",
314            ));
315        }
316
317        let select_statement =
318            select_parser::parse_select(cql).inspect_err(|_| self.inc_error_queries())?;
319        let optimized_plan = self.select_optimizer.optimize(select_statement).await?;
320
321        self.select_executor
322            .execute_streaming(optimized_plan, config)
323            .await
324    }
325
326    /// Execute a SELECT query using the advanced parser and optimizer
327    async fn execute_select_query(&self, cql: &str, start_time: Instant) -> Result<QueryResult> {
328        // Check plan cache first for SELECT queries too. Issue #1595: resolve the
329        // outcome under a shard READ lock (`DashMap::get`) — bumping the lock-free
330        // `hit_count` and cloning the plan for a reusable entry — then DROP the
331        // guard before acting, so the `.await` holds no shard lock and `remove`
332        // never runs while a `get` guard on the same shard is held.
333        enum HitOutcome {
334            Reuse(super::planner::QueryPlan),
335            Placeholder,
336            Miss,
337        }
338        let outcome = match self.plan_cache.get(cql) {
339            Some(entry) if entry.plan.table.is_some() => {
340                entry.hit_count.fetch_add(1, Ordering::Relaxed);
341                HitOutcome::Reuse(entry.plan.clone())
342            }
343            Some(_) => HitOutcome::Placeholder,
344            None => HitOutcome::Miss,
345        };
346        match outcome {
347            HitOutcome::Reuse(plan) => {
348                self.record_cache_hit();
349
350                let mut result = crate::observability::record_result(
351                    "query",
352                    self.executor.execute(&plan).await,
353                )?;
354                // Issue #1582 (D6): this cached SELECT plan is served by the
355                // LEGACY executor, which bypasses the SelectExecutor's byte
356                // budget; enforce it on this plan-cache HIT return too.
357                self.enforce_legacy_result_budget(&result)?;
358                self.update_execution_stats(&mut result, start_time);
359                return Ok(result);
360            }
361            HitOutcome::Placeholder => {
362                // Placeholder plans without table information are not reusable; drop them.
363                self.plan_cache.remove(cql);
364            }
365            HitOutcome::Miss => {}
366        }
367
368        #[cfg(not(feature = "state_machine"))]
369        return Err(Error::query_execution(
370            "Advanced SELECT parsing requires state_machine feature",
371        ));
372
373        #[cfg(feature = "state_machine")]
374        {
375            // Issue #1587 (E5): reuse a previously-optimized plan for the exact
376            // same CQL text. The optimized plan is a pure function of the literal
377            // CQL, so this is byte-for-byte equivalent to re-parsing +
378            // re-optimizing — it only skips the redundant work.
379            let optimized_plan: Arc<OptimizedQueryPlan> =
380                if let Some(entry) = self.select_plan_cache.get(cql) {
381                    self.record_cache_hit();
382                    Arc::clone(&entry.value().1)
383                } else {
384                    let select_statement = select_parser::parse_select(cql).inspect_err(|e| {
385                        self.inc_error_queries();
386                        crate::observability::record_error(e, "query");
387                    })?;
388                    let plan = Arc::new(crate::observability::record_result(
389                        "query",
390                        self.select_optimizer.optimize(select_statement).await,
391                    )?);
392                    self.cache_select_plan(cql, Arc::clone(&plan));
393                    plan
394                };
395            let mut result = crate::observability::record_result(
396                "query",
397                self.select_executor
398                    .execute((*optimized_plan).clone())
399                    .await,
400            )?;
401            self.update_execution_stats(&mut result, start_time);
402            Ok(result)
403        }
404    }
405
406    /// Insert an optimized SELECT plan into the modern plan cache, evicting the
407    /// oldest entry first when at capacity (simple LRU, matching the legacy
408    /// `cache_query_plan`). A configured `query_cache_size` of 0 disables it.
409    #[cfg(feature = "state_machine")]
410    fn cache_select_plan(&self, cql: &str, plan: Arc<OptimizedQueryPlan>) {
411        let cache_size = self.config.query.query_cache_size.unwrap_or(0);
412        if cache_size == 0 {
413            return;
414        }
415        if self.select_plan_cache.len() >= cache_size {
416            let oldest_key = self
417                .select_plan_cache
418                .iter()
419                .min_by_key(|entry| entry.value().0)
420                .map(|entry| entry.key().clone());
421            if let Some(key) = oldest_key {
422                self.select_plan_cache.remove(&key);
423            }
424        }
425        self.select_plan_cache
426            .insert(cql.to_string(), (Instant::now(), plan));
427    }
428
429    /// Execute a query with positional `?` parameters (Issue #961).
430    ///
431    /// The supplied `params` are bound, in source order, into the `?` placeholders
432    /// of the parsed statement *before* planning and execution, so the bound
433    /// values participate in partition-key classification, encoding, and typed
434    /// coercion. A `WHERE pk = ?` therefore engages the same partition-targeted
435    /// fast path (#949/#956) as the equivalent literal query.
436    ///
437    /// Binding is currently supported for SELECT statements only. A non-SELECT
438    /// CQL with parameters, or any use of named (`:name`) parameters, is rejected
439    /// with a clear error (named-parameter binding is intentionally out of scope:
440    /// the SELECT grammar only tokenizes positional `?`).
441    ///
442    /// # Routing parity with `execute` (Finding 1)
443    ///
444    /// When the parsed SELECT has **zero** bind markers and `params` is empty,
445    /// this delegates straight back to [`Self::execute`] so that a markerless
446    /// `execute_with_params(sql, &[])` is byte-for-byte equivalent to
447    /// `execute(sql)` — which since issue #1750 routes every literal SELECT
448    /// through the modern SELECT optimizer + executor. Only when markers are
449    /// present (`> 0`) is the statement bound and driven through that pipeline.
450    ///
451    /// Arity stays strict in both directions: markers `> 0` with a wrong
452    /// `params.len()` is an error, and markers `== 0` with a **non-empty**
453    /// `params` is also an error (a supplied parameter with no placeholder is a
454    /// caller bug). The latter matches [`SelectStatement::bind_parameters`]'s
455    /// contract and the documented strictness of this API.
456    pub async fn execute_with_params(&self, cql: &str, params: &[Value]) -> Result<QueryResult> {
457        let is_select = cql.trim().to_uppercase().starts_with("SELECT");
458
459        if !is_select {
460            // Non-SELECT: parameter binding is not supported. With no parameters
461            // this is just a normal statement, so defer to the regular path;
462            // with parameters it is an explicit, clear error.
463            if params.is_empty() {
464                return self.execute(cql).await;
465            }
466            self.inc_total_queries();
467            self.inc_error_queries();
468            return Err(Error::query_execution(
469                "Parameterized execution currently supports SELECT statements only",
470            ));
471        }
472
473        #[cfg(not(feature = "state_machine"))]
474        {
475            let _ = params;
476            self.inc_total_queries();
477            self.inc_error_queries();
478            return Err(Error::query_execution(
479                "Parameterized SELECT execution requires the state_machine feature",
480            ));
481        }
482
483        #[cfg(feature = "state_machine")]
484        {
485            // Parse once so we can count bind markers and decide routing. Parse
486            // failures here mirror `execute_select_query`, which would also fail.
487            let statement = select_parser::parse_select(cql).inspect_err(|_| {
488                self.inc_total_queries();
489                self.inc_error_queries();
490            })?;
491            let marker_count = statement.bind_marker_count();
492
493            // Finding 1: a markerless SELECT with no supplied params must route
494            // exactly like a literal `execute(cql)` (since #1750, the modern
495            // executor) so the two APIs cannot diverge. A marker-free statement
496            // with stray params is a caller bug: reject it for strict arity.
497            if marker_count == 0 {
498                if params.is_empty() {
499                    return self.execute(cql).await;
500                }
501                self.inc_total_queries();
502                self.inc_error_queries();
503                return Err(Error::query_execution(format!(
504                    "Parameter count mismatch: query has 0 bind marker(s), got {} parameter(s)",
505                    params.len()
506                )));
507            }
508
509            let start_time = Instant::now();
510            self.inc_total_queries();
511
512            // Markers present: bind through the SELECT pipeline. Arity is
513            // enforced by `bind_parameters` (too few / too many -> error). The
514            // bound statement reaches the same optimizer + executor as a literal
515            // `execute()`, so the partition-targeted fast path engages.
516            let mut statement = statement;
517            statement
518                .bind_parameters(params)
519                .inspect_err(|_| self.inc_error_queries())?;
520
521            let optimized_plan = self.select_optimizer.optimize(statement).await?;
522            let mut result = self.select_executor.execute(optimized_plan).await?;
523            self.update_execution_stats(&mut result, start_time);
524            Ok(result)
525        }
526    }
527
528    /// Prepare a query for repeated execution
529    pub async fn prepare(&self, cql: &str) -> Result<Arc<PreparedQuery>> {
530        if let Some(cached) = self.prepared_cache.get(cql) {
531            return Ok(cached.clone());
532        }
533
534        let parsed_query = self.parser.parse(cql)?;
535        let plan = self.planner.plan(&parsed_query).await?;
536
537        // Issue #961: when the prepared statement is a SELECT, attach the SELECT
538        // optimizer + executor pipeline so that `?` parameters are bound and the
539        // bound query reaches the partition-targeted fast path (#949/#956) — the
540        // same path a literal `execute()` takes. Non-SELECTs keep the legacy
541        // `QueryExecutor` plan path.
542        #[cfg(feature = "state_machine")]
543        let prepared = if cql.trim().to_uppercase().starts_with("SELECT") {
544            let statement = select_parser::parse_select(cql)?;
545            let marker_count = statement.bind_marker_count();
546            Arc::new(PreparedQuery::new_select(
547                parsed_query,
548                plan,
549                Arc::new(self.executor.clone()),
550                statement,
551                marker_count,
552                self.select_optimizer.clone(),
553                self.select_executor.clone(),
554            ))
555        } else {
556            Arc::new(PreparedQuery::new(
557                parsed_query,
558                plan,
559                Arc::new(self.executor.clone()),
560            ))
561        };
562
563        #[cfg(not(feature = "state_machine"))]
564        let prepared = Arc::new(PreparedQuery::new(
565            parsed_query,
566            plan,
567            Arc::new(self.executor.clone()),
568        ));
569
570        self.prepared_cache
571            .insert(cql.to_string(), prepared.clone());
572
573        Ok(prepared)
574    }
575
576    /// Execute a prepared query
577    pub async fn execute_prepared(
578        &self,
579        prepared: &PreparedQuery,
580        params: &[Value],
581    ) -> Result<QueryResult> {
582        let start_time = Instant::now();
583        self.inc_total_queries();
584
585        let mut result = prepared.execute(params).await?;
586        self.update_execution_stats(&mut result, start_time);
587        Ok(result)
588    }
589
590    /// Get query statistics
591    pub fn stats(&self) -> QueryStats {
592        self.stats.snapshot()
593    }
594
595    /// Clear all caches
596    pub fn clear_caches(&self) {
597        self.prepared_cache.clear();
598        self.plan_cache.clear();
599        #[cfg(feature = "state_machine")]
600        self.select_plan_cache.clear();
601    }
602
603    /// Clear prepared statement cache
604    pub fn clear_prepared_cache(&self) {
605        self.prepared_cache.clear();
606    }
607
608    /// Clear query plan cache
609    pub fn clear_plan_cache(&self) {
610        self.plan_cache.clear();
611        #[cfg(feature = "state_machine")]
612        self.select_plan_cache.clear();
613    }
614
615    /// Get cache statistics
616    pub fn cache_stats(&self) -> CacheStats {
617        CacheStats {
618            prepared_cache_size: self.prepared_cache.len(),
619            plan_cache_size: self.plan_cache.len(),
620            prepared_cache_hits: self.prepared_cache.len() as u64,
621            plan_cache_hits: self.plan_cache.len() as u64,
622        }
623    }
624
625    /// Optimize a query (return execution plan without executing)
626    pub async fn explain(&self, cql: &str) -> Result<ExplainResult> {
627        // Parse the query
628        let parsed_query = self.parser.parse(cql)?;
629
630        // Plan the query
631        let plan = self.planner.plan(&parsed_query).await?;
632
633        Ok(ExplainResult {
634            query_type: format!("{:?}", parsed_query.query_type),
635            plan_type: format!("{:?}", plan.plan_type),
636            estimated_cost: plan.estimated_cost,
637            estimated_rows: plan.estimated_rows,
638            selected_indexes: plan
639                .selected_indexes
640                .iter()
641                .map(|idx| format!("{} ({:?})", idx.index_name, idx.index_type))
642                .collect(),
643            execution_steps: plan
644                .steps
645                .iter()
646                .map(|step| {
647                    format!(
648                        "{:?}: {} (cost: {:.2})",
649                        step.step_type,
650                        step.columns.join(", "),
651                        step.cost
652                    )
653                })
654                .collect(),
655            parallelization_info: plan
656                .steps
657                .iter()
658                .filter(|step| step.parallelization.can_parallelize)
659                .map(|step| {
660                    format!(
661                        "Threads: {}, Partition: {:?}",
662                        step.parallelization.suggested_threads, step.parallelization.partition_key
663                    )
664                })
665                .collect(),
666        })
667    }
668
669    /// Analyze query performance
670    pub async fn analyze(&self, cql: &str) -> Result<AnalyzeResult> {
671        let start_time = Instant::now();
672
673        // Execute the query multiple times to get average performance
674        let mut execution_times = Vec::new();
675        let mut results = Vec::new();
676
677        for _ in 0..self.config.query.analyze_iterations.unwrap_or(5) {
678            let iter_start = Instant::now();
679            let result = self.execute(cql).await?;
680            execution_times.push(iter_start.elapsed());
681            results.push(result);
682        }
683
684        let total_time = start_time.elapsed();
685        let avg_time =
686            execution_times.iter().sum::<std::time::Duration>() / execution_times.len() as u32;
687        let no_times = || Error::query_execution("No execution times recorded for analysis");
688        let min_time = execution_times.iter().min().ok_or_else(no_times)?;
689        let max_time = execution_times.iter().max().ok_or_else(no_times)?;
690
691        // Calculate standard deviation
692        let variance = execution_times
693            .iter()
694            .map(|time| {
695                let diff = time.as_nanos() as f64 - avg_time.as_nanos() as f64;
696                diff * diff
697            })
698            .sum::<f64>()
699            / execution_times.len() as f64;
700        let std_dev = variance.sqrt();
701
702        Ok(AnalyzeResult {
703            iterations: execution_times.len(),
704            total_time_ms: total_time.as_millis() as u64,
705            avg_time_ms: avg_time.as_millis() as u64,
706            min_time_ms: min_time.as_millis() as u64,
707            max_time_ms: max_time.as_millis() as u64,
708            std_dev_ms: (std_dev / 1_000_000.0) as u64, // Convert from nanoseconds to milliseconds
709            avg_rows_returned: results.iter().map(|r| r.rows.len()).sum::<usize>() / results.len(),
710            cache_hit_ratio: self.stats().cache_hit_ratio,
711        })
712    }
713
714    /// Cache a query plan, evicting the oldest entry first if at capacity (simple LRU).
715    fn cache_query_plan(
716        &self,
717        cql: &str,
718        parsed_query: super::ParsedQuery,
719        plan: super::planner::QueryPlan,
720    ) {
721        let cache_size = self.config.query.query_cache_size.unwrap_or(0);
722        if cache_size == 0 {
723            return;
724        }
725
726        if self.plan_cache.len() >= cache_size {
727            let oldest_key = self
728                .plan_cache
729                .iter()
730                .min_by_key(|entry| entry.cached_at)
731                .map(|entry| entry.key().clone());
732            if let Some(key) = oldest_key {
733                self.plan_cache.remove(&key);
734            }
735        }
736
737        self.plan_cache.insert(
738            cql.to_string(),
739            QueryCacheEntry {
740                parsed_query,
741                plan,
742                cached_at: Instant::now(),
743                hit_count: AtomicU64::new(0),
744            },
745        );
746    }
747
748    /// Check if schema is available for a table
749    pub async fn has_schema_for_table(&self, table: &str) -> bool {
750        self.schema_manager.get_table_schema(table).await.is_ok()
751    }
752
753    /// Get detailed schema status for debugging
754    pub async fn schema_status(&self, table: &str) -> SchemaStatus {
755        match self.schema_manager.get_table_schema(table).await {
756            Ok(schema) => SchemaStatus::Available {
757                keyspace: schema.keyspace.clone(),
758                table: schema.table.clone(),
759            },
760            Err(Error::Schema(msg)) if msg.contains("not found") => {
761                SchemaStatus::Missing {
762                    table: table.to_string(),
763                    reason: msg,
764                }
765            }
766            Err(e) => SchemaStatus::ExtractionFailed {
767                table: table.to_string(),
768                cause: e.to_string(),
769                suggestion: "Verify SSTable files are valid Cassandra 5.0 format and Statistics.db contains SerializationHeader".to_string(),
770            },
771        }
772    }
773
774    /// Update execution statistics.
775    ///
776    /// This is the single chokepoint every materializing query path (cache hit,
777    /// parsed-plan, SELECT, parameterized, prepared) funnels through once its
778    /// result is known, so it is also where observability emits the end-to-end
779    /// query signals exactly once (issue #1035): the [`catalog::QUERY_DURATION`]
780    /// histogram and the [`catalog::QUERY_ROWS`] counter, both dimensioned by the
781    /// bounded access path the SELECT chose (when available). It also records the
782    /// bounded span attributes on the active `query.execute` span so the parent of
783    /// the read-path span tree is self-describing. Durations are reported in
784    /// seconds, per the catalog convention.
785    fn update_execution_stats(&self, result: &mut QueryResult, start_time: Instant) {
786        use crate::observability::{self as obs, catalog, AttrValue};
787
788        let execution_time = start_time.elapsed();
789        // Ensure any non-zero execution time is at least 1ms for reporting
790        result.execution_time_ms = if execution_time.is_zero() {
791            0
792        } else {
793            std::cmp::max(1, execution_time.as_millis() as u64)
794        };
795
796        // Bounded access-path dimension, sourced from the honest per-query signal
797        // the modern SelectExecutor attaches to the result (epic #951/#960). We
798        // CONSUME that existing label rather than reinventing it; `None` (legacy
799        // executor / non-SELECT) is reported as "unknown" to keep the dimension
800        // bounded without fabricating a path.
801        let access_path_label: &'static str = result
802            .metadata
803            .access_path
804            .as_ref()
805            .map(|p| p.label())
806            .unwrap_or("unknown");
807        let plan_type_label: &'static str = result
808            .metadata
809            .plan_info
810            .as_ref()
811            .map(|p| Self::plan_type_label(&p.plan_type))
812            .unwrap_or("unknown");
813
814        // Emit the end-to-end query metrics exactly once, here.
815        obs::record_histogram(
816            catalog::QUERY_DURATION,
817            execution_time.as_secs_f64(),
818            &[
819                (catalog::attr::SUBSYSTEM, AttrValue::StaticStr("query")),
820                (
821                    catalog::attr::ACCESS_PATH,
822                    AttrValue::StaticStr(access_path_label),
823                ),
824                (
825                    catalog::attr::PLAN_TYPE,
826                    AttrValue::StaticStr(plan_type_label),
827                ),
828            ],
829        );
830        obs::add_counter(
831            catalog::QUERY_ROWS,
832            result.rows.len() as u64,
833            &[
834                (
835                    catalog::attr::ACCESS_PATH,
836                    AttrValue::StaticStr(access_path_label),
837                ),
838                (
839                    catalog::attr::PLAN_TYPE,
840                    AttrValue::StaticStr(plan_type_label),
841                ),
842            ],
843        );
844
845        // Record bounded attributes on the active `query.execute` span (parent of
846        // the read-path span tree). Never attach the query text or key values.
847        let span = tracing::Span::current();
848        span.record(catalog::attr::PLAN_TYPE, plan_type_label);
849        span.record(catalog::attr::ACCESS_PATH, access_path_label);
850        span.record("cqlite.query.rows", result.rows.len());
851
852        // Lock-free (issue #1595): accumulate the execution time and rows into
853        // relaxed atomics. `avg_execution_time_us` is derived at read time as
854        // `exec_time_us_sum / total_queries`.
855        let new_time_us = execution_time.as_micros() as u64;
856        self.stats
857            .record_execution(new_time_us, result.rows_affected);
858    }
859
860    /// Map a `PlanInfo.plan_type` (an executor `Debug`-formatted plan family such
861    /// as `"TableScan"`) onto a bounded, lower-snake label suitable as a metric
862    /// dimension and span attribute (issue #1035). Falls back to `"other"` for
863    /// any value outside the known taxonomy so the dimension stays bounded.
864    fn plan_type_label(plan_type: &str) -> &'static str {
865        match plan_type {
866            "TableScan" => "table_scan",
867            "IndexScan" => "index_scan",
868            "PointLookup" => "point_lookup",
869            "RangeScan" => "range_scan",
870            "Join" => "join",
871            "Aggregation" => "aggregation",
872            "Subquery" => "subquery",
873            _ => "other",
874        }
875    }
876}
877
878/// Cache statistics
879#[derive(Debug, Clone)]
880pub struct CacheStats {
881    /// Number of prepared statements cached
882    pub prepared_cache_size: usize,
883    /// Number of query plans cached
884    pub plan_cache_size: usize,
885    /// Total prepared cache hits
886    pub prepared_cache_hits: u64,
887    /// Total plan cache hits
888    pub plan_cache_hits: u64,
889}
890
891/// Query explanation result
892#[derive(Debug, Clone)]
893pub struct ExplainResult {
894    /// Query type
895    pub query_type: String,
896    /// Plan type
897    pub plan_type: String,
898    /// Estimated cost
899    pub estimated_cost: f64,
900    /// Estimated rows
901    pub estimated_rows: u64,
902    /// Selected indexes
903    pub selected_indexes: Vec<String>,
904    /// Execution steps
905    pub execution_steps: Vec<String>,
906    /// Parallelization information
907    pub parallelization_info: Vec<String>,
908}
909
910/// Query analysis result
911#[derive(Debug, Clone)]
912pub struct AnalyzeResult {
913    /// Number of iterations
914    pub iterations: usize,
915    /// Total analysis time
916    pub total_time_ms: u64,
917    /// Average execution time
918    pub avg_time_ms: u64,
919    /// Minimum execution time
920    pub min_time_ms: u64,
921    /// Maximum execution time
922    pub max_time_ms: u64,
923    /// Standard deviation of execution times
924    pub std_dev_ms: u64,
925    /// Average rows returned
926    pub avg_rows_returned: usize,
927    /// Cache hit ratio
928    pub cache_hit_ratio: f64,
929}
930
931#[cfg(all(test, feature = "state_machine"))]
932mod tests {
933    use super::*;
934    use crate::Config;
935    use std::sync::Arc;
936    use tempfile::TempDir;
937
938    #[tokio::test]
939    async fn test_query_engine_creation() {
940        let temp_dir = TempDir::new().unwrap();
941        let config = Config::default();
942        let platform = Arc::new(crate::platform::Platform::new(&config).await.unwrap());
943
944        let storage = Arc::new(
945            crate::storage::StorageEngine::open(
946                temp_dir.path(),
947                &config,
948                platform,
949                #[cfg(feature = "state_machine")]
950                None,
951            )
952            .await
953            .unwrap(),
954        );
955        let schema = Arc::new(
956            crate::schema::SchemaManager::new(temp_dir.path())
957                .await
958                .unwrap(),
959        );
960        let memory = Arc::new(crate::memory::MemoryManager::new(&config).unwrap());
961
962        let query_engine = QueryEngine::new(storage, schema, memory, &config).unwrap();
963
964        assert_eq!(query_engine.stats().total_queries, 0);
965        assert_eq!(query_engine.cache_stats().prepared_cache_size, 0);
966        assert_eq!(query_engine.cache_stats().plan_cache_size, 0);
967    }
968
969    #[tokio::test]
970    #[ignore = "Hangs >60s; needs investigation - gated for M1"]
971    async fn test_query_caching() {
972        let temp_dir = TempDir::new().unwrap();
973        let mut config = Config::test_config();
974        config.query.query_cache_size = Some(10);
975
976        let platform = Arc::new(crate::platform::Platform::new(&config).await.unwrap());
977        let storage = Arc::new(
978            crate::storage::StorageEngine::open(
979                temp_dir.path(),
980                &config,
981                platform,
982                #[cfg(feature = "state_machine")]
983                None,
984            )
985            .await
986            .unwrap(),
987        );
988        let schema = Arc::new(
989            crate::schema::SchemaManager::new(temp_dir.path())
990                .await
991                .unwrap(),
992        );
993        let memory = Arc::new(crate::memory::MemoryManager::new(&config).unwrap());
994
995        let query_engine = QueryEngine::new(storage, schema, memory, &config).unwrap();
996
997        // Execute a query twice
998        let cql = "SELECT * FROM users WHERE id = 1";
999        let _ = query_engine.execute(cql).await;
1000        let _ = query_engine.execute(cql).await;
1001
1002        // Check that plan was cached
1003        assert_eq!(query_engine.cache_stats().plan_cache_size, 1);
1004
1005        // Check cache hit ratio
1006        let stats = query_engine.stats();
1007        assert!(stats.cache_hit_ratio > 0.0);
1008    }
1009
1010    #[tokio::test]
1011    #[cfg(feature = "state_machine")]
1012    async fn test_prepared_statements() {
1013        let temp_dir = TempDir::new().unwrap();
1014        let config = Config::default();
1015        let platform = Arc::new(crate::platform::Platform::new(&config).await.unwrap());
1016
1017        let storage = Arc::new(
1018            crate::storage::StorageEngine::open(
1019                temp_dir.path(),
1020                &config,
1021                platform,
1022                #[cfg(feature = "state_machine")]
1023                None,
1024            )
1025            .await
1026            .unwrap(),
1027        );
1028        let schema = Arc::new(
1029            crate::schema::SchemaManager::new(temp_dir.path())
1030                .await
1031                .unwrap(),
1032        );
1033        let memory = Arc::new(crate::memory::MemoryManager::new(&config).unwrap());
1034
1035        let query_engine = QueryEngine::new(storage, schema, memory, &config).unwrap();
1036
1037        // Prepare a statement
1038        let cql = "SELECT * FROM users WHERE id = ?";
1039        let prepared = query_engine.prepare(cql).await.unwrap();
1040
1041        // Execute it with parameters
1042        let params = vec![Value::Integer(1)];
1043        let result = query_engine
1044            .execute_prepared(&prepared, &params)
1045            .await
1046            .unwrap();
1047
1048        // Check that result was generated
1049        assert!(result.execution_time_ms > 0);
1050
1051        // Check that statement was cached
1052        assert_eq!(query_engine.cache_stats().prepared_cache_size, 1);
1053    }
1054
1055    #[tokio::test]
1056    async fn test_query_explain() {
1057        let temp_dir = TempDir::new().unwrap();
1058        let config = Config::default();
1059        let platform = Arc::new(crate::platform::Platform::new(&config).await.unwrap());
1060
1061        let storage = Arc::new(
1062            crate::storage::StorageEngine::open(
1063                temp_dir.path(),
1064                &config,
1065                platform,
1066                #[cfg(feature = "state_machine")]
1067                None,
1068            )
1069            .await
1070            .unwrap(),
1071        );
1072        let schema = Arc::new(
1073            crate::schema::SchemaManager::new(temp_dir.path())
1074                .await
1075                .unwrap(),
1076        );
1077        let memory = Arc::new(crate::memory::MemoryManager::new(&config).unwrap());
1078
1079        let query_engine = QueryEngine::new(storage, schema, memory, &config).unwrap();
1080
1081        // Explain a query
1082        let cql = "SELECT * FROM users WHERE id = 1";
1083        let explain_result = query_engine.explain(cql).await.unwrap();
1084
1085        assert_eq!(explain_result.query_type, "Select");
1086        assert!(explain_result.estimated_cost > 0.0);
1087        assert!(!explain_result.selected_indexes.is_empty());
1088        assert!(!explain_result.execution_steps.is_empty());
1089    }
1090
1091    #[tokio::test]
1092    #[cfg(feature = "state_machine")]
1093    async fn test_cache_eviction() {
1094        let temp_dir = TempDir::new().unwrap();
1095        let mut config = Config::default();
1096        config.query.query_cache_size = Some(2); // Very small cache
1097
1098        let platform = Arc::new(crate::platform::Platform::new(&config).await.unwrap());
1099        let storage = Arc::new(
1100            crate::storage::StorageEngine::open(
1101                temp_dir.path(),
1102                &config,
1103                platform,
1104                #[cfg(feature = "state_machine")]
1105                None,
1106            )
1107            .await
1108            .unwrap(),
1109        );
1110        let schema = Arc::new(
1111            crate::schema::SchemaManager::new(temp_dir.path())
1112                .await
1113                .unwrap(),
1114        );
1115        let memory = Arc::new(crate::memory::MemoryManager::new(&config).unwrap());
1116
1117        let query_engine = QueryEngine::new(storage, schema, memory, &config).unwrap();
1118
1119        // Execute 3 different queries. Since #1750 these route through the modern
1120        // SELECT executor; their plans live in `select_plan_cache`.
1121        let _ = query_engine
1122            .execute("SELECT * FROM users WHERE id = 1")
1123            .await;
1124        let _ = query_engine
1125            .execute("SELECT * FROM users WHERE id = 2")
1126            .await;
1127        let _ = query_engine
1128            .execute("SELECT * FROM users WHERE id = 3")
1129            .await;
1130
1131        // Cache should only have 2 entries due to eviction (simple LRU).
1132        assert_eq!(query_engine.select_plan_cache.len(), 2);
1133    }
1134
1135    #[tokio::test]
1136    #[cfg(feature = "state_machine")]
1137    async fn test_schema_validation_api() {
1138        let temp_dir = TempDir::new().unwrap();
1139        let config = Config::default();
1140        let platform = Arc::new(crate::platform::Platform::new(&config).await.unwrap());
1141
1142        let storage = Arc::new(
1143            crate::storage::StorageEngine::open(
1144                temp_dir.path(),
1145                &config,
1146                platform,
1147                #[cfg(feature = "state_machine")]
1148                None,
1149            )
1150            .await
1151            .unwrap(),
1152        );
1153        let schema = Arc::new(
1154            crate::schema::SchemaManager::new(temp_dir.path())
1155                .await
1156                .unwrap(),
1157        );
1158        let memory = Arc::new(crate::memory::MemoryManager::new(&config).unwrap());
1159
1160        let query_engine = QueryEngine::new(storage, schema, memory, &config).unwrap();
1161
1162        // Test has_schema_for_table with non-existent table
1163        let has_schema = query_engine.has_schema_for_table("nonexistent_table").await;
1164        assert!(!has_schema, "Should return false for non-existent table");
1165
1166        // Test schema_status with non-existent table
1167        let status = query_engine.schema_status("nonexistent_table").await;
1168        match status {
1169            SchemaStatus::Missing { .. } | SchemaStatus::ExtractionFailed { .. } => {
1170                // Expected - either missing or extraction failed is correct
1171            }
1172            SchemaStatus::Available { .. } => {
1173                panic!("Should not be Available for non-existent table");
1174            }
1175        }
1176    }
1177
1178    /// Issue #1587 (E5): the modern (state_machine) ad-hoc SELECT path caches the
1179    /// optimized plan keyed by the exact CQL text, so N identical ad-hoc executes
1180    /// of the same query optimize AT MOST ONCE. This is the `engine.rs` half of
1181    /// the plan-reuse optimization (the prepared-statement half is covered by
1182    /// `prepared_select_reuses_optimized_plan_across_executes`).
1183    ///
1184    /// `#[tokio::test]` runs on the current-thread runtime, so the optimizer's
1185    /// thread-local invocation counter reflects exactly this future's calls. The
1186    /// plan is cached BEFORE execution, so the (empty-store) execute result is
1187    /// irrelevant — reuse holds regardless.
1188    #[tokio::test]
1189    #[cfg(feature = "state_machine")]
1190    async fn adhoc_select_reuses_cached_optimized_plan() {
1191        use crate::query::select_optimizer::OPTIMIZE_INVOCATIONS;
1192
1193        let temp_dir = TempDir::new().unwrap();
1194        let mut config = Config::default();
1195        // Enable the modern SELECT plan cache (0 disables it).
1196        config.query.query_cache_size = Some(16);
1197        let platform = Arc::new(crate::platform::Platform::new(&config).await.unwrap());
1198        let storage = Arc::new(
1199            crate::storage::StorageEngine::open(
1200                temp_dir.path(),
1201                &config,
1202                platform,
1203                #[cfg(feature = "state_machine")]
1204                None,
1205            )
1206            .await
1207            .unwrap(),
1208        );
1209        let schema = Arc::new(
1210            crate::schema::SchemaManager::new(temp_dir.path())
1211                .await
1212                .unwrap(),
1213        );
1214        let memory = Arc::new(crate::memory::MemoryManager::new(&config).unwrap());
1215        let engine = QueryEngine::new(storage, schema, memory, &config).unwrap();
1216
1217        // Every SELECT routes through `execute_select_query` (issue #1750 retired
1218        // the `WHERE id =` text fork), which owns the modern `select_plan_cache`.
1219        let cql = "SELECT * FROM t WHERE age > 5";
1220
1221        OPTIMIZE_INVOCATIONS.with(|c| c.set(0));
1222        for _ in 0..75 {
1223            let _ = engine.execute(cql).await;
1224        }
1225        let after_same = OPTIMIZE_INVOCATIONS.with(|c| c.get());
1226        assert!(
1227            after_same <= 1,
1228            "issue #1587: an ad-hoc SELECT must optimize at most once across 75 \
1229             identical executes (modern plan cache), got {after_same}"
1230        );
1231
1232        // Clearing the cache must force exactly one re-optimize on the next run.
1233        engine.clear_plan_cache();
1234        let _ = engine.execute(cql).await;
1235        let after_clear = OPTIMIZE_INVOCATIONS.with(|c| c.get());
1236        assert_eq!(
1237            after_clear,
1238            after_same + 1,
1239            "clearing the plan cache must re-optimize once (no stale reuse)"
1240        );
1241    }
1242}
1243
1244// Plan-cache HIT path lock-hygiene coverage (issue #1595, spec Requirement 2:
1245// "Plan-cache hit path uses a shared (read) lock"). Kept in its own file to
1246// avoid growing this already-oversized module; declared here (a descendant of
1247// `engine`) so it can read the private `plan_cache`/`select_plan_cache` fields
1248// and the `pub` `hit_count` atomic directly, with no new public accessor.
1249#[cfg(test)]
1250#[path = "engine_lock_hygiene_tests.rs"]
1251mod engine_lock_hygiene_tests;
1252
1253#[cfg(test)]
1254#[cfg(all(feature = "experimental", feature = "state_machine"))]
1255mod plan_cache_tests {
1256    use super::*;
1257    use crate::{
1258        memory::MemoryManager, platform::Platform, schema::SchemaManager, storage::StorageEngine,
1259        Config,
1260    };
1261    use std::sync::Arc;
1262    use tempfile::TempDir;
1263
1264    async fn setup_query_engine(config: &Config) -> (QueryEngine, TempDir) {
1265        let temp_dir = TempDir::new().unwrap();
1266        let platform = Arc::new(Platform::new(config).await.unwrap());
1267        let storage = Arc::new(
1268            StorageEngine::open(
1269                temp_dir.path(),
1270                config,
1271                platform,
1272                #[cfg(feature = "state_machine")]
1273                None,
1274            )
1275            .await
1276            .unwrap(),
1277        );
1278        let schema = Arc::new(SchemaManager::new(temp_dir.path()).await.unwrap());
1279        let memory = Arc::new(MemoryManager::new(config).unwrap());
1280
1281        let engine = QueryEngine::new(storage, schema, memory, config).unwrap();
1282        (engine, temp_dir)
1283    }
1284
1285    /// Number of cached modern SELECT plans. Since issue #1750 ALL SELECTs
1286    /// (including `WHERE id = ?` point lookups) route through the modern
1287    /// `SelectExecutor`, whose plans live in `select_plan_cache`, not the legacy
1288    /// `plan_cache` surfaced by `cache_stats`; these LRU tests assert on it.
1289    fn select_plan_cache_len(engine: &QueryEngine) -> usize {
1290        engine.select_plan_cache.len()
1291    }
1292
1293    /// Register the table schema used by the plan-cache tests. Write path removed
1294    /// in Issue #175, so this only issues `CREATE TABLE` DDL — no row inserts. The
1295    /// assertions depend solely on planning (cache population/eviction), not
1296    /// stored rows, so an empty table exercises current behavior exactly.
1297    async fn create_sample_table(engine: &QueryEngine) {
1298        engine
1299            .execute(
1300                "CREATE TABLE plan_cache_test (
1301                    id INTEGER PRIMARY KEY,
1302                    value TEXT
1303                )",
1304            )
1305            .await
1306            .unwrap();
1307    }
1308
1309    #[tokio::test]
1310    async fn test_plan_cache_disabled() {
1311        let mut config = Config::default();
1312        config.query.query_cache_size = Some(0);
1313
1314        let (engine, _temp_dir) = setup_query_engine(&config).await;
1315        create_sample_table(&engine).await;
1316
1317        engine
1318            .execute("SELECT * FROM plan_cache_test WHERE id = 1")
1319            .await
1320            .unwrap();
1321
1322        assert_eq!(select_plan_cache_len(&engine), 0);
1323    }
1324
1325    #[tokio::test]
1326    async fn test_plan_cache_reuse_point_lookup() {
1327        let mut config = Config::default();
1328        config.query.query_cache_size = Some(4);
1329
1330        let (engine, _temp_dir) = setup_query_engine(&config).await;
1331        create_sample_table(&engine).await;
1332
1333        engine.clear_plan_cache();
1334
1335        engine
1336            .execute("SELECT * FROM plan_cache_test WHERE id = 1")
1337            .await
1338            .unwrap();
1339        engine
1340            .execute("SELECT * FROM plan_cache_test WHERE id = 1")
1341            .await
1342            .unwrap();
1343
1344        assert_eq!(select_plan_cache_len(&engine), 1);
1345        assert!(engine.stats().cache_hit_ratio > 0.0);
1346    }
1347
1348    #[tokio::test]
1349    async fn test_plan_cache_eviction_limit() {
1350        let mut config = Config::default();
1351        config.query.query_cache_size = Some(2);
1352
1353        let (engine, _temp_dir) = setup_query_engine(&config).await;
1354        create_sample_table(&engine).await;
1355
1356        engine.clear_plan_cache();
1357
1358        for id in 1..=3 {
1359            engine
1360                .execute(&format!("SELECT * FROM plan_cache_test WHERE id = {}", id))
1361                .await
1362                .unwrap();
1363        }
1364
1365        assert_eq!(select_plan_cache_len(&engine), 2);
1366    }
1367}
cqlite_core/query/engine.rs

cqlite_core/query/
engine.rs
