cqlite_core/query/
mod.rs

1//! Query execution engine for CQLite
2//!
3//! This module provides the core query execution engine that bridges between
4//! CQL parsing and the storage engine. It includes:
5//!
6//! - CQL query parsing and validation
7//! - Query planning and optimization
8//! - Query execution with storage integration
9//! - Prepared statement support
10//! - Result set management
11//! - REVOLUTIONARY SELECT parser for direct SSTable access
12
13// CQL (Cassandra Query Language) Reference:
14// https://cassandra.apache.org/doc/latest/cassandra/developing/cql/cql_singlefile.html
15//
16// This implements CQL v3.4.3+ for Apache Cassandra 5.0+
17// CQL is NOT SQL - it's a query language specifically designed for Cassandra's distributed architecture.
18
19pub mod access_path;
20pub mod agg_stream_probe;
21pub mod engine;
22pub(crate) mod engine_stats;
23pub mod executor;
24pub mod m2_select_validator;
25pub mod parser;
26pub mod planner;
27pub mod prepared;
28pub mod result;
29pub(crate) mod result_budget;
30pub mod writetime_ttl_validator;
31
32// Advanced SELECT query components
33#[cfg(feature = "state_machine")]
34pub mod select_ast;
35#[cfg(feature = "state_machine")]
36pub mod select_executor;
37#[cfg(feature = "state_machine")]
38pub mod select_integration_tests;
39#[cfg(feature = "state_machine")]
40pub mod select_naming;
41#[cfg(feature = "state_machine")]
42pub mod select_optimizer;
43#[cfg(feature = "state_machine")]
44pub mod select_parser;
45
46pub use access_path::{AccessPath, FallbackReason};
47pub use engine::{
48    AnalyzeResult, CacheStats, ExplainResult, QueryEngine as AdvancedQueryEngine, SchemaStatus,
49};
50pub use executor::{
51    QueryExecutor, QueryResult as ExecutorQueryResult, QueryRow as ExecutorQueryRow,
52};
53pub use m2_select_validator::{M2SelectValidator, SelectValidationResult, UnsupportedFeature};
54pub use parser::QueryParser;
55pub use planner::{ExecutionStep, IndexSelection, PlanType, QueryHints, QueryPlan, QueryPlanner};
56pub use prepared::{
57    ExecutionHints, ParameterMetadata, PreparedContext, PreparedQuery, PreparedQueryBuilder,
58    PreparedQueryStats,
59};
60pub use result::{
61    cql_type_to_data_type, ColumnInfo, PerformanceMetrics, QueryMetadata, QueryResult,
62    QueryResultIterator, QueryRow, RowMetadata, StreamingConfig,
63};
64pub use writetime_ttl_validator::{
65    descriptors_from_table_schema, validate_all_writetime_ttl_calls, validate_writetime_ttl_call,
66    ColumnDescriptor as WriteTimeTtlColumnDescriptor, WriteTimeTtlKind,
67};
68
69// Re-export advanced SELECT components.
70// Only the symbols with known callers are re-exported here. `select_ast` is
71// intentionally not glob-re-exported: a `pub use select_ast::*` would shadow the
72// inline `ComparisonOperator`, `OrderByClause`, and `SortDirection` types below.
73// Other `select_ast` items remain reachable via `cqlite_core::query::select_ast::*`.
74#[cfg(feature = "state_machine")]
75pub use select_ast::SelectStatement;
76#[cfg(feature = "state_machine")]
77pub use select_executor::{
78    build_row_from_scan, build_row_from_scan_cached, evaluate_leaf, evaluate_predicates,
79    LeafOutcome, PartitionKeyCache, SelectExecutor,
80};
81#[cfg(feature = "state_machine")]
82pub use select_optimizer::{
83    OptimizedQueryPlan, SSTableFilterOp, SSTablePredicate, SelectOptimizer,
84};
85#[cfg(feature = "state_machine")]
86pub use select_parser::{parse_select, SelectParser};
87
88use std::collections::HashMap;
89use std::sync::Arc;
90
91use crate::{
92    memory::MemoryManager, schema::SchemaManager, storage::StorageEngine, Config, Result, TableId,
93    Value,
94};
95
96/// Query execution statistics
97#[derive(Debug, Clone, Default)]
98pub struct QueryStats {
99    /// Total queries executed
100    pub total_queries: u64,
101    /// Queries that resulted in errors
102    pub error_queries: u64,
103    /// Average execution time in microseconds
104    pub avg_execution_time_us: u64,
105    /// Cache hit ratio
106    pub cache_hit_ratio: f64,
107    /// Total rows affected by write operations
108    pub rows_affected: u64,
109}
110
111/// Legacy query engine wrapper for backward compatibility.
112///
113/// Delegates to [`AdvancedQueryEngine`] (re-exported from [`engine`]); kept as a
114/// stable public surface used by [`crate::Database`] and the CLI's REPL.
115#[derive(Debug)]
116pub struct QueryEngine {
117    advanced_engine: AdvancedQueryEngine,
118}
119
120impl QueryEngine {
121    /// Create a new query engine.
122    pub fn new(
123        storage: Arc<StorageEngine>,
124        schema: Arc<SchemaManager>,
125        memory: Arc<MemoryManager>,
126        config: &Config,
127    ) -> Result<Self> {
128        Ok(Self {
129            advanced_engine: AdvancedQueryEngine::new(storage, schema, memory, config)?,
130        })
131    }
132
133    /// Execute a CQL query
134    pub async fn execute(&self, cql: &str) -> Result<QueryResult> {
135        self.advanced_engine.execute(cql).await
136    }
137
138    /// Execute a SELECT with positional `?` parameters (Issue #961).
139    pub async fn execute_with_params(&self, cql: &str, params: &[Value]) -> Result<QueryResult> {
140        self.advanced_engine.execute_with_params(cql, params).await
141    }
142
143    /// Prepare a query for repeated execution
144    pub async fn prepare(&self, cql: &str) -> Result<Arc<PreparedQuery>> {
145        self.advanced_engine.prepare(cql).await
146    }
147
148    /// Get query statistics
149    pub fn stats(&self) -> QueryStats {
150        self.advanced_engine.stats()
151    }
152
153    /// Clear prepared statement cache
154    pub fn clear_cache(&self) {
155        self.advanced_engine.clear_prepared_cache();
156    }
157
158    /// Explain a query
159    pub async fn explain(&self, cql: &str) -> Result<ExplainResult> {
160        self.advanced_engine.explain(cql).await
161    }
162
163    /// Analyze query performance
164    pub async fn analyze(&self, cql: &str) -> Result<AnalyzeResult> {
165        self.advanced_engine.analyze(cql).await
166    }
167
168    /// Execute a CQL query with streaming results (Issue #280)
169    ///
170    /// Returns a `QueryResultIterator` that yields rows incrementally via a bounded
171    /// channel, enabling memory-efficient processing of large result sets.
172    #[cfg(feature = "state_machine")]
173    pub async fn execute_streaming(
174        &self,
175        cql: &str,
176        config: StreamingConfig,
177    ) -> Result<QueryResultIterator> {
178        self.advanced_engine.execute_streaming(cql, config).await
179    }
180
181    /// Get cache statistics
182    pub fn cache_stats(&self) -> CacheStats {
183        self.advanced_engine.cache_stats()
184    }
185
186    /// Check if schema is available for a table
187    pub async fn has_schema_for_table(&self, table: &str) -> bool {
188        self.advanced_engine.has_schema_for_table(table).await
189    }
190
191    /// Get detailed schema status for debugging
192    pub async fn schema_status(&self, table: &str) -> SchemaStatus {
193        self.advanced_engine.schema_status(table).await
194    }
195}
196
197/// CQL query types
198#[derive(Debug, Clone, PartialEq)]
199pub enum QueryType {
200    /// SELECT statement
201    Select,
202    /// INSERT statement
203    Insert,
204    /// UPDATE statement
205    Update,
206    /// DELETE statement
207    Delete,
208    /// CREATE TABLE statement
209    CreateTable,
210    /// DROP TABLE statement
211    DropTable,
212    /// CREATE INDEX statement
213    CreateIndex,
214    /// DROP INDEX statement
215    DropIndex,
216    /// DESCRIBE statement
217    Describe,
218    /// USE statement (for keyspace)
219    Use,
220}
221
222/// Parsed CQL query representation
223#[derive(Debug, Clone)]
224pub struct ParsedQuery {
225    /// Query type
226    pub query_type: QueryType,
227    /// Target table (if applicable)
228    pub table: Option<TableId>,
229    /// Column selections (for SELECT)
230    pub columns: Vec<String>,
231    /// WHERE clause conditions
232    pub where_clause: Option<WhereClause>,
233    /// VALUES for INSERT
234    pub values: Vec<Value>,
235    /// SET clause for UPDATE
236    pub set_clause: HashMap<String, Value>,
237    /// ORDER BY clause
238    pub order_by: Vec<OrderByClause>,
239    /// LIMIT clause
240    pub limit: Option<usize>,
241    /// Original CQL text
242    pub cql: String,
243}
244
245/// WHERE clause representation
246#[derive(Debug, Clone)]
247pub struct WhereClause {
248    /// Conditions in the WHERE clause
249    pub conditions: Vec<Condition>,
250}
251
252/// Individual condition in WHERE clause
253#[derive(Debug, Clone)]
254pub struct Condition {
255    /// Column name
256    pub column: String,
257    /// Comparison operator
258    pub operator: ComparisonOperator,
259    /// Value to compare against
260    pub value: Value,
261}
262
263/// Comparison operators
264#[derive(Debug, Clone, PartialEq)]
265pub enum ComparisonOperator {
266    /// Equal (=)
267    Equal,
268    /// Not equal (<>)
269    NotEqual,
270    /// Less than (<)
271    LessThan,
272    /// Less than or equal (<=)
273    LessThanOrEqual,
274    /// Greater than (>)
275    GreaterThan,
276    /// Greater than or equal (>=)
277    GreaterThanOrEqual,
278    /// IN operator
279    In,
280    /// NOT IN operator
281    NotIn,
282    /// LIKE operator
283    Like,
284    /// NOT LIKE operator
285    NotLike,
286}
287
288/// ORDER BY clause
289#[derive(Debug, Clone)]
290pub struct OrderByClause {
291    /// Column name
292    pub column: String,
293    /// Sort direction
294    pub direction: SortDirection,
295}
296
297/// Sort direction
298#[derive(Debug, Clone, PartialEq)]
299pub enum SortDirection {
300    /// Ascending
301    Asc,
302    /// Descending
303    Desc,
304}
305
306#[cfg(all(test, feature = "state_machine"))]
307mod tests {
308    use super::*;
309    use crate::platform::Platform;
310    use tempfile::TempDir;
311
312    #[tokio::test]
313    async fn test_query_engine_creation() {
314        let temp_dir = TempDir::new().unwrap();
315        let config = Config::default();
316        let platform = Arc::new(Platform::new(&config).await.unwrap());
317
318        let storage = Arc::new(
319            StorageEngine::open(
320                temp_dir.path(),
321                &config,
322                platform,
323                #[cfg(feature = "state_machine")]
324                None,
325            )
326            .await
327            .unwrap(),
328        );
329        let schema = Arc::new(SchemaManager::new(temp_dir.path()).await.unwrap());
330        let memory = Arc::new(MemoryManager::new(&config).unwrap());
331
332        let query_engine = QueryEngine::new(storage, schema, memory, &config).unwrap();
333
334        assert_eq!(query_engine.stats().total_queries, 0);
335    }
336}
cqlite_core/query/mod.rs

cqlite_core/query/
mod.rs
