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

cqlite_core/query/
engine.rs
