Skip to main content

alopex_sql/executor/
mod.rs

1//! SQL Executor module for Alopex SQL.
2//!
3//! This module provides the execution engine for SQL statements.
4//!
5//! # Overview
6//!
7//! The Executor takes a [`LogicalPlan`] from the Planner and executes it
8//! against the storage layer. It supports DDL, DML, and Query operations.
9//!
10//! Query execution currently materializes intermediate results per stage;
11//! future versions may add streaming pipelines as requirements grow.
12
13//! # Components
14//!
15//! - [`Executor`]: Main executor struct
16//! - [`ExecutorError`]: Error types for execution
17//! - [`ExecutionResult`]: Execution result types
18//!
19//! # Example
20//!
21//! ```ignore
22//! use std::sync::{Arc, RwLock};
23//! use alopex_core::kv::memory::MemoryKV;
24//! use alopex_sql::executor::Executor;
25//! use alopex_sql::catalog::MemoryCatalog;
26//! use alopex_sql::planner::LogicalPlan;
27//!
28//! // Create storage and catalog
29//! let store = Arc::new(MemoryKV::new());
30//! let catalog = Arc::new(RwLock::new(MemoryCatalog::new()));
31//!
32//! // Create executor
33//! let mut executor = Executor::new(store, catalog);
34//!
35//! // Execute a plan
36//! let result = executor.execute(plan)?;
37//! ```
38
39#[cfg(feature = "tokio")]
40pub mod async_executor;
41pub mod bulk;
42pub(crate) mod ddl;
43pub(crate) mod dml;
44mod error;
45pub mod evaluator;
46mod hnsw_bridge;
47pub mod memory;
48pub mod query;
49mod result;
50mod system;
51
52#[cfg(feature = "tokio")]
53pub use async_executor::AsyncExecutor;
54pub use error::{ConstraintViolation, EvaluationError, ExecutorError, Result};
55pub use memory::{MemoryPolicy, SpillPolicy};
56pub use query::{RowIterator, ScanIterator, build_streaming_pipeline};
57pub use result::{ColumnInfo, ExecutionResult, QueryResult, QueryRowIterator, Row};
58
59/// Returns whether a plan requires direct access to the backing KV store.
60pub fn is_store_direct_plan(plan: &LogicalPlan) -> bool {
61    system::is_store_direct_plan(plan)
62}
63
64use std::sync::{Arc, RwLock};
65
66use alopex_core::kv::KVStore;
67use alopex_core::types::TxnMode;
68
69use crate::catalog::Catalog;
70use crate::catalog::persistent::{IndexFqn, TableFqn};
71use crate::catalog::{CatalogError, CatalogOverlay, PersistentCatalog, TxnCatalogView};
72use crate::planner::LogicalPlan;
73use crate::storage::{BorrowedSqlTransaction, KeyEncoder, SqlTransaction, SqlTxn as _, TxnBridge};
74
75/// SQL statement executor.
76///
77/// The Executor takes a [`LogicalPlan`] and executes it against the storage layer.
78/// It manages transactions and coordinates between DDL, DML, and Query operations.
79///
80/// # Type Parameters
81///
82/// - `S`: The underlying KV store type (must implement [`KVStore`])
83/// - `C`: The catalog type (must implement [`Catalog`])
84pub struct Executor<S: KVStore, C: Catalog> {
85    /// Transaction bridge for storage operations.
86    bridge: TxnBridge<S>,
87
88    /// Catalog for metadata operations.
89    catalog: Arc<RwLock<C>>,
90}
91
92impl<S: KVStore, C: Catalog> Executor<S, C> {
93    fn run_in_write_txn<R, F>(&self, f: F) -> Result<R>
94    where
95        F: FnOnce(&mut SqlTransaction<'_, S>) -> Result<R>,
96    {
97        let mut txn = self.bridge.begin_write().map_err(ExecutorError::from)?;
98        match f(&mut txn) {
99            Ok(result) => {
100                txn.commit().map_err(ExecutorError::from)?;
101                Ok(result)
102            }
103            Err(err) => {
104                txn.rollback().map_err(ExecutorError::from)?;
105                Err(err)
106            }
107        }
108    }
109
110    /// Create a new Executor with the given store and catalog.
111    ///
112    /// # Arguments
113    ///
114    /// - `store`: The underlying KV store
115    /// - `catalog`: The catalog for metadata operations
116    pub fn new(store: Arc<S>, catalog: Arc<RwLock<C>>) -> Self {
117        Self {
118            bridge: TxnBridge::new(store),
119            catalog,
120        }
121    }
122
123    /// Execute a logical plan and return the result.
124    ///
125    /// # Arguments
126    ///
127    /// - `plan`: The logical plan to execute
128    ///
129    /// # Returns
130    ///
131    /// Returns an [`ExecutionResult`] on success, or an [`ExecutorError`] on failure.
132    ///
133    /// # DDL Operations
134    ///
135    /// - `CreateTable`: Creates a new table with optional PK index
136    /// - `DropTable`: Drops a table and its associated indexes
137    /// - `CreateIndex`: Creates a new index
138    /// - `DropIndex`: Drops an index
139    ///
140    /// # DML Operations
141    ///
142    /// - `Insert`: Inserts rows into a table
143    /// - `Update`: Updates rows in a table
144    /// - `Delete`: Deletes rows from a table
145    ///
146    /// # Query Operations
147    ///
148    /// - `Scan`, `Filter`, `Sort`, `Limit`: SELECT query execution
149    pub fn execute(&mut self, plan: LogicalPlan) -> Result<ExecutionResult> {
150        let _statement_timestamp = evaluator::begin_statement();
151        match plan {
152            LogicalPlan::Pragma { name, value } => {
153                system::execute_pragma(&self.bridge, &name, value.as_ref())
154            }
155            // DDL Operations
156            LogicalPlan::CreateTable {
157                table,
158                if_not_exists,
159                with_options,
160            } => self.execute_create_table(table, with_options, if_not_exists),
161            LogicalPlan::DropTable { name, if_exists } => self.execute_drop_table(&name, if_exists),
162            LogicalPlan::CreateIndex {
163                index,
164                if_not_exists,
165            } => self.execute_create_index(index, if_not_exists),
166            LogicalPlan::DropIndex { name, if_exists } => self.execute_drop_index(&name, if_exists),
167
168            // DML Operations
169            LogicalPlan::Insert {
170                table,
171                columns,
172                values,
173            } => self.execute_insert(&table, columns, values),
174            LogicalPlan::InsertSelect {
175                table,
176                columns,
177                source,
178            } => self.execute_insert_select(&table, columns, *source),
179            LogicalPlan::Update {
180                table,
181                assignments,
182                filter,
183            } => self.execute_update(&table, assignments, filter),
184            LogicalPlan::Delete { table, filter } => self.execute_delete(&table, filter),
185
186            // Query Operations
187            LogicalPlan::Scan { .. }
188            | LogicalPlan::Filter { .. }
189            | LogicalPlan::Project { .. }
190            | LogicalPlan::Join { .. }
191            | LogicalPlan::Aggregate { .. }
192            | LogicalPlan::Window { .. }
193            | LogicalPlan::SetOperation { .. }
194            | LogicalPlan::Sort { .. }
195            | LogicalPlan::Limit { .. } => self.execute_query(plan),
196        }
197    }
198
199    // ========================================================================
200    // DDL Operations (to be implemented in Phase 2)
201    // ========================================================================
202
203    fn execute_create_table(
204        &mut self,
205        table: crate::catalog::TableMetadata,
206        with_options: Vec<(String, String)>,
207        if_not_exists: bool,
208    ) -> Result<ExecutionResult> {
209        let mut catalog = self.catalog.write().expect("catalog lock poisoned");
210        self.run_in_write_txn(|txn| {
211            ddl::create_table::execute_create_table(
212                txn,
213                &mut *catalog,
214                table,
215                with_options,
216                if_not_exists,
217            )
218        })
219    }
220
221    fn execute_drop_table(&mut self, name: &str, if_exists: bool) -> Result<ExecutionResult> {
222        let mut catalog = self.catalog.write().expect("catalog lock poisoned");
223        self.run_in_write_txn(|txn| {
224            ddl::drop_table::execute_drop_table(txn, &mut *catalog, name, if_exists)
225        })
226    }
227
228    fn execute_create_index(
229        &mut self,
230        index: crate::catalog::IndexMetadata,
231        if_not_exists: bool,
232    ) -> Result<ExecutionResult> {
233        let mut catalog = self.catalog.write().expect("catalog lock poisoned");
234        self.run_in_write_txn(|txn| {
235            ddl::create_index::execute_create_index(txn, &mut *catalog, index, if_not_exists)
236        })
237    }
238
239    fn execute_drop_index(&mut self, name: &str, if_exists: bool) -> Result<ExecutionResult> {
240        let mut catalog = self.catalog.write().expect("catalog lock poisoned");
241        self.run_in_write_txn(|txn| {
242            ddl::drop_index::execute_drop_index(txn, &mut *catalog, name, if_exists)
243        })
244    }
245
246    // ========================================================================
247    // DML Operations (implemented in Phase 4)
248    // ========================================================================
249
250    fn execute_insert(
251        &mut self,
252        table: &str,
253        columns: Vec<String>,
254        values: Vec<Vec<crate::planner::TypedExpr>>,
255    ) -> Result<ExecutionResult> {
256        let catalog = self.catalog.read().expect("catalog lock poisoned");
257        self.run_in_write_txn(|txn| dml::execute_insert(txn, &*catalog, table, columns, values))
258    }
259
260    fn execute_insert_select(
261        &mut self,
262        table: &str,
263        columns: Vec<String>,
264        source: LogicalPlan,
265    ) -> Result<ExecutionResult> {
266        let catalog = self.catalog.read().expect("catalog lock poisoned");
267        self.run_in_write_txn(|txn| {
268            let ExecutionResult::Query(result) = query::execute_query(txn, &*catalog, source)?
269            else {
270                return Err(ExecutorError::InvalidOperation {
271                    operation: "INSERT ... SELECT".into(),
272                    reason: "SELECT source did not return query rows".into(),
273                });
274            };
275            dml::execute_insert_rows(txn, &*catalog, table, columns, result.rows)
276        })
277    }
278
279    fn execute_update(
280        &mut self,
281        table: &str,
282        assignments: Vec<crate::planner::TypedAssignment>,
283        filter: Option<crate::planner::TypedExpr>,
284    ) -> Result<ExecutionResult> {
285        let catalog = self.catalog.read().expect("catalog lock poisoned");
286        self.run_in_write_txn(|txn| dml::execute_update(txn, &*catalog, table, assignments, filter))
287    }
288
289    fn execute_delete(
290        &mut self,
291        table: &str,
292        filter: Option<crate::planner::TypedExpr>,
293    ) -> Result<ExecutionResult> {
294        let catalog = self.catalog.read().expect("catalog lock poisoned");
295        self.run_in_write_txn(|txn| dml::execute_delete(txn, &*catalog, table, filter))
296    }
297
298    // ========================================================================
299    // Query Operations (to be implemented in Phase 5)
300    // ========================================================================
301
302    fn execute_query(&mut self, plan: LogicalPlan) -> Result<ExecutionResult> {
303        if let Some(result) = system::try_execute(&self.bridge, &plan)? {
304            return Ok(result);
305        }
306        let catalog = self.catalog.read().expect("catalog lock poisoned");
307        self.run_in_write_txn(|txn| query::execute_query(txn, &*catalog, plan))
308    }
309}
310
311impl<S: KVStore> Executor<S, PersistentCatalog<S>> {
312    pub fn execute_in_txn<'a, 'b, 'c>(
313        &mut self,
314        plan: LogicalPlan,
315        txn: &mut BorrowedSqlTransaction<'a, 'b, 'c, S>,
316    ) -> Result<ExecutionResult> {
317        if txn.mode() == TxnMode::ReadOnly
318            && !matches!(
319                plan,
320                LogicalPlan::Scan { .. }
321                    | LogicalPlan::Filter { .. }
322                    | LogicalPlan::Project { .. }
323                    | LogicalPlan::Join { .. }
324                    | LogicalPlan::Aggregate { .. }
325                    | LogicalPlan::Window { .. }
326                    | LogicalPlan::SetOperation { .. }
327                    | LogicalPlan::Sort { .. }
328                    | LogicalPlan::Limit { .. }
329            )
330        {
331            return Err(ExecutorError::ReadOnlyTransaction {
332                operation: plan.operation_name().to_string(),
333            });
334        }
335
336        let _statement_timestamp = evaluator::begin_statement();
337        let mut catalog = self.catalog.write().expect("catalog lock poisoned");
338        let (mut sql_txn, overlay) = txn.split_parts();
339
340        let result = match plan {
341            LogicalPlan::CreateTable {
342                table,
343                if_not_exists,
344                with_options,
345            } => self.execute_create_table_in_txn(
346                &mut *catalog,
347                &mut sql_txn,
348                overlay,
349                table,
350                with_options,
351                if_not_exists,
352            ),
353            LogicalPlan::DropTable { name, if_exists } => self.execute_drop_table_in_txn(
354                &mut *catalog,
355                &mut sql_txn,
356                overlay,
357                &name,
358                if_exists,
359            ),
360            LogicalPlan::CreateIndex {
361                index,
362                if_not_exists,
363            } => self.execute_create_index_in_txn(
364                &mut *catalog,
365                &mut sql_txn,
366                overlay,
367                index,
368                if_not_exists,
369            ),
370            LogicalPlan::DropIndex { name, if_exists } => self.execute_drop_index_in_txn(
371                &mut *catalog,
372                &mut sql_txn,
373                overlay,
374                &name,
375                if_exists,
376            ),
377            LogicalPlan::Pragma { .. } => Err(ExecutorError::UnsupportedOperation(
378                "PRAGMA is not available inside an external transaction".to_string(),
379            )),
380            LogicalPlan::Insert {
381                table,
382                columns,
383                values,
384            } => {
385                let view = TxnCatalogView::new(&*catalog, &*overlay);
386                dml::execute_insert(&mut sql_txn, &view, &table, columns, values)
387            }
388            LogicalPlan::InsertSelect {
389                table,
390                columns,
391                source,
392            } => {
393                let view = TxnCatalogView::new(&*catalog, &*overlay);
394                let ExecutionResult::Query(result) =
395                    query::execute_query(&mut sql_txn, &view, *source)?
396                else {
397                    return Err(ExecutorError::InvalidOperation {
398                        operation: "INSERT ... SELECT".into(),
399                        reason: "SELECT source did not return query rows".into(),
400                    });
401                };
402                dml::execute_insert_rows(&mut sql_txn, &view, &table, columns, result.rows)
403            }
404            LogicalPlan::Update {
405                table,
406                assignments,
407                filter,
408            } => {
409                let view = TxnCatalogView::new(&*catalog, &*overlay);
410                dml::execute_update(&mut sql_txn, &view, &table, assignments, filter)
411            }
412            LogicalPlan::Delete { table, filter } => {
413                let view = TxnCatalogView::new(&*catalog, &*overlay);
414                dml::execute_delete(&mut sql_txn, &view, &table, filter)
415            }
416            LogicalPlan::Scan { .. }
417            | LogicalPlan::Filter { .. }
418            | LogicalPlan::Project { .. }
419            | LogicalPlan::Join { .. }
420            | LogicalPlan::Aggregate { .. }
421            | LogicalPlan::Window { .. }
422            | LogicalPlan::SetOperation { .. }
423            | LogicalPlan::Sort { .. }
424            | LogicalPlan::Limit { .. } => {
425                let view = TxnCatalogView::new(&*catalog, &*overlay);
426                query::execute_query(&mut sql_txn, &view, plan)
427            }
428        };
429
430        match result {
431            Ok(value) => {
432                sql_txn.flush_hnsw()?;
433                Ok(value)
434            }
435            Err(err) => {
436                let _ = sql_txn.abandon_hnsw();
437                Err(err)
438            }
439        }
440    }
441
442    fn map_catalog_error(err: CatalogError) -> ExecutorError {
443        match err {
444            CatalogError::Kv(e) => ExecutorError::Core(e),
445            CatalogError::Serialize(e) => ExecutorError::InvalidOperation {
446                operation: "CatalogPersistence".into(),
447                reason: e.to_string(),
448            },
449            CatalogError::InvalidKey(reason) => ExecutorError::InvalidOperation {
450                operation: "CatalogPersistence".into(),
451                reason,
452            },
453        }
454    }
455
456    fn execute_create_table_in_txn<'txn>(
457        &self,
458        catalog: &mut PersistentCatalog<S>,
459        txn: &mut impl crate::storage::SqlTxn<'txn, S>,
460        overlay: &mut CatalogOverlay,
461        mut table: crate::catalog::TableMetadata,
462        with_options: Vec<(String, String)>,
463        if_not_exists: bool,
464    ) -> Result<ExecutionResult>
465    where
466        S: 'txn,
467    {
468        if catalog.table_exists_in_txn(&table.name, overlay) {
469            return if if_not_exists {
470                Ok(ExecutionResult::Success)
471            } else {
472                Err(ExecutorError::TableAlreadyExists(table.name))
473            };
474        }
475
476        table.storage_options = ddl::create_table::parse_storage_options(&with_options)?;
477
478        let pk_index = if let Some(pk_columns) = table.primary_key.clone() {
479            let column_indices = pk_columns
480                .iter()
481                .map(|name| {
482                    table
483                        .get_column_index(name)
484                        .ok_or_else(|| ExecutorError::ColumnNotFound(name.clone()))
485                })
486                .collect::<Result<Vec<_>>>()?;
487            let index_id = catalog.next_index_id();
488            let index_name = ddl::create_pk_index_name(&table.name);
489            let mut index = crate::catalog::IndexMetadata::new(
490                index_id,
491                index_name,
492                table.name.clone(),
493                pk_columns,
494            )
495            .with_column_indices(column_indices)
496            .with_unique(true);
497            index.catalog_name = table.catalog_name.clone();
498            index.namespace_name = table.namespace_name.clone();
499            Some(index)
500        } else {
501            None
502        };
503
504        let table_id = catalog.next_table_id();
505        table = table.with_table_id(table_id);
506
507        // storage keyspace の初期化
508        txn.delete_prefix(&KeyEncoder::table_prefix(table_id))?;
509        txn.delete_prefix(&KeyEncoder::sequence_key(table_id))?;
510
511        // 永続化(同一 KV トランザクション内)
512        catalog
513            .persist_create_table(txn.inner_mut(), &table)
514            .map_err(Self::map_catalog_error)?;
515        if let Some(index) = &pk_index {
516            catalog
517                .persist_create_index(txn.inner_mut(), index)
518                .map_err(Self::map_catalog_error)?;
519        }
520
521        // オーバーレイに反映(ベースカタログはコミットまで不変)
522        overlay.add_table(TableFqn::from(&table), table);
523        if let Some(index) = pk_index {
524            overlay.add_index(IndexFqn::from(&index), index);
525        }
526
527        Ok(ExecutionResult::Success)
528    }
529
530    fn execute_drop_table_in_txn<'txn>(
531        &self,
532        catalog: &mut PersistentCatalog<S>,
533        txn: &mut impl crate::storage::SqlTxn<'txn, S>,
534        overlay: &mut CatalogOverlay,
535        table_name: &str,
536        if_exists: bool,
537    ) -> Result<ExecutionResult>
538    where
539        S: 'txn,
540    {
541        let table_meta = match catalog.get_table_in_txn(table_name, overlay) {
542            Some(table) => table.clone(),
543            None => {
544                return if if_exists {
545                    Ok(ExecutionResult::Success)
546                } else {
547                    Err(ExecutorError::TableNotFound(table_name.to_string()))
548                };
549            }
550        };
551        if table_meta.catalog_name != "default" || table_meta.namespace_name != "default" {
552            return if if_exists {
553                Ok(ExecutionResult::Success)
554            } else {
555                Err(ExecutorError::TableNotFound(table_name.to_string()))
556            };
557        }
558
559        let indexes = TxnCatalogView::new(catalog, overlay)
560            .get_indexes_for_table(table_name)
561            .into_iter()
562            .cloned()
563            .collect::<Vec<_>>();
564
565        for index in &indexes {
566            if matches!(index.method, Some(crate::ast::ddl::IndexMethod::Hnsw)) {
567                crate::executor::hnsw_bridge::HnswBridge::drop_index(txn, index, false)?;
568            } else {
569                txn.delete_prefix(&KeyEncoder::index_prefix(index.index_id))?;
570            }
571        }
572
573        txn.delete_prefix(&KeyEncoder::table_prefix(table_meta.table_id))?;
574        txn.delete_prefix(&KeyEncoder::sequence_key(table_meta.table_id))?;
575
576        catalog
577            .persist_drop_table(txn.inner_mut(), &TableFqn::from(&table_meta))
578            .map_err(Self::map_catalog_error)?;
579
580        overlay.drop_table(&TableFqn::from(&table_meta));
581
582        Ok(ExecutionResult::Success)
583    }
584
585    fn execute_create_index_in_txn<'txn>(
586        &self,
587        catalog: &mut PersistentCatalog<S>,
588        txn: &mut impl crate::storage::SqlTxn<'txn, S>,
589        overlay: &mut CatalogOverlay,
590        mut index: crate::catalog::IndexMetadata,
591        if_not_exists: bool,
592    ) -> Result<ExecutionResult>
593    where
594        S: 'txn,
595    {
596        if ddl::is_implicit_pk_index(&index.name) {
597            return Err(ExecutorError::InvalidIndexName {
598                name: index.name.clone(),
599                reason: "Index names starting with '__pk_' are reserved for PRIMARY KEY".into(),
600            });
601        }
602
603        if catalog.index_exists_in_txn(&index.name, overlay) {
604            return if if_not_exists {
605                Ok(ExecutionResult::Success)
606            } else {
607                Err(ExecutorError::IndexAlreadyExists(index.name))
608            };
609        }
610
611        let table = catalog
612            .get_table_in_txn(&index.table, overlay)
613            .ok_or_else(|| ExecutorError::TableNotFound(index.table.clone()))?
614            .clone();
615        index.catalog_name = table.catalog_name.clone();
616        index.namespace_name = table.namespace_name.clone();
617
618        let column_indices = index
619            .columns
620            .iter()
621            .map(|name| {
622                table
623                    .get_column_index(name)
624                    .ok_or_else(|| ExecutorError::ColumnNotFound(name.clone()))
625            })
626            .collect::<Result<Vec<_>>>()?;
627
628        let index_id = catalog.next_index_id();
629        index.index_id = index_id;
630        index.column_indices = column_indices.clone();
631
632        if matches!(index.method, Some(crate::ast::ddl::IndexMethod::Hnsw)) {
633            crate::executor::hnsw_bridge::HnswBridge::create_index(txn, &table, &index)?;
634        } else {
635            ddl::create_index::build_index_for_existing_rows(txn, &table, &index, column_indices)?;
636        }
637
638        catalog
639            .persist_create_index(txn.inner_mut(), &index)
640            .map_err(Self::map_catalog_error)?;
641
642        overlay.add_index(IndexFqn::from(&index), index);
643
644        Ok(ExecutionResult::Success)
645    }
646
647    fn execute_drop_index_in_txn<'txn>(
648        &self,
649        catalog: &mut PersistentCatalog<S>,
650        txn: &mut impl crate::storage::SqlTxn<'txn, S>,
651        overlay: &mut CatalogOverlay,
652        index_name: &str,
653        if_exists: bool,
654    ) -> Result<ExecutionResult>
655    where
656        S: 'txn,
657    {
658        if ddl::is_implicit_pk_index(index_name) {
659            return Err(ExecutorError::InvalidOperation {
660                operation: "DROP INDEX".into(),
661                reason: "Cannot drop implicit PRIMARY KEY index directly; use DROP TABLE".into(),
662            });
663        }
664
665        let index = match catalog.get_index_in_txn(index_name, overlay) {
666            Some(index) => index.clone(),
667            None => {
668                return if if_exists {
669                    Ok(ExecutionResult::Success)
670                } else {
671                    Err(ExecutorError::IndexNotFound(index_name.to_string()))
672                };
673            }
674        };
675        if index.catalog_name != "default" || index.namespace_name != "default" {
676            return if if_exists {
677                Ok(ExecutionResult::Success)
678            } else {
679                Err(ExecutorError::IndexNotFound(index_name.to_string()))
680            };
681        }
682
683        if matches!(index.method, Some(crate::ast::ddl::IndexMethod::Hnsw)) {
684            crate::executor::hnsw_bridge::HnswBridge::drop_index(txn, &index, if_exists)?;
685        } else {
686            txn.delete_prefix(&KeyEncoder::index_prefix(index.index_id))?;
687        }
688
689        catalog
690            .persist_drop_index(txn.inner_mut(), &IndexFqn::from(&index))
691            .map_err(Self::map_catalog_error)?;
692
693        overlay.drop_index(&IndexFqn::from(&index));
694
695        Ok(ExecutionResult::Success)
696    }
697}
698
699#[cfg(test)]
700mod tests {
701    use super::*;
702    use crate::catalog::MemoryCatalog;
703    use alopex_core::kv::memory::MemoryKV;
704
705    fn create_executor() -> Executor<MemoryKV, MemoryCatalog> {
706        let store = Arc::new(MemoryKV::new());
707        let catalog = Arc::new(RwLock::new(MemoryCatalog::new()));
708        Executor::new(store, catalog)
709    }
710
711    #[test]
712    fn test_executor_creation() {
713        let _executor = create_executor();
714        // Executor should be created without panic
715    }
716
717    #[test]
718    fn create_table_is_supported() {
719        let mut executor = create_executor();
720
721        use crate::catalog::{ColumnMetadata, TableMetadata};
722        use crate::planner::ResolvedType;
723
724        let table = TableMetadata::new(
725            "test",
726            vec![ColumnMetadata::new("id", ResolvedType::Integer)],
727        );
728
729        let result = executor.execute(LogicalPlan::CreateTable {
730            table,
731            if_not_exists: false,
732            with_options: vec![],
733        });
734        assert!(matches!(result, Ok(ExecutionResult::Success)));
735
736        let catalog = executor.catalog.read().unwrap();
737        assert!(catalog.table_exists("test"));
738    }
739
740    #[test]
741    fn insert_is_supported() {
742        use crate::Span;
743        use crate::catalog::{ColumnMetadata, TableMetadata};
744        use crate::planner::typed_expr::TypedExprKind;
745        use crate::planner::types::ResolvedType;
746
747        let mut executor = create_executor();
748
749        let table = TableMetadata::new("t", vec![ColumnMetadata::new("id", ResolvedType::Integer)])
750            .with_primary_key(vec!["id".into()]);
751
752        executor
753            .execute(LogicalPlan::CreateTable {
754                table,
755                if_not_exists: false,
756                with_options: vec![],
757            })
758            .unwrap();
759
760        let result = executor.execute(LogicalPlan::Insert {
761            table: "t".into(),
762            columns: vec!["id".into()],
763            values: vec![vec![crate::planner::typed_expr::TypedExpr {
764                kind: TypedExprKind::Literal(crate::ast::expr::Literal::Number("1".into())),
765                resolved_type: ResolvedType::Integer,
766                span: Span::default(),
767            }]],
768        });
769        assert!(matches!(result, Ok(ExecutionResult::RowsAffected(1))));
770    }
771
772    #[test]
773    fn system_pragma_and_stats_function_use_the_store() {
774        let mut executor = create_executor();
775        let catalog = MemoryCatalog::new();
776
777        let pragma = crate::Parser::parse_sql(&crate::AlopexDialect, "PRAGMA cache_size = 8")
778            .unwrap()
779            .pop()
780            .unwrap();
781        let plan = crate::Planner::new(&catalog).plan(&pragma).unwrap();
782        assert!(matches!(
783            executor.execute(plan),
784            Ok(ExecutionResult::Success)
785        ));
786
787        let select = crate::Parser::parse_sql(&crate::AlopexDialect, "SELECT memory_stats()")
788            .unwrap()
789            .pop()
790            .unwrap();
791        let plan = crate::Planner::new(&catalog).plan(&select).unwrap();
792        let result = executor.execute(plan).unwrap();
793        let ExecutionResult::Query(result) = result else {
794            panic!("expected query result");
795        };
796        assert_eq!(result.columns[0].name, "memory_stats");
797        assert!(
798            matches!(&result.rows[0][0], crate::SqlValue::Text(text) if text.contains("total_bytes"))
799        );
800    }
801}