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::Sort { .. }
193            | LogicalPlan::Limit { .. } => self.execute_query(plan),
194        }
195    }
196
197    // ========================================================================
198    // DDL Operations (to be implemented in Phase 2)
199    // ========================================================================
200
201    fn execute_create_table(
202        &mut self,
203        table: crate::catalog::TableMetadata,
204        with_options: Vec<(String, String)>,
205        if_not_exists: bool,
206    ) -> Result<ExecutionResult> {
207        let mut catalog = self.catalog.write().expect("catalog lock poisoned");
208        self.run_in_write_txn(|txn| {
209            ddl::create_table::execute_create_table(
210                txn,
211                &mut *catalog,
212                table,
213                with_options,
214                if_not_exists,
215            )
216        })
217    }
218
219    fn execute_drop_table(&mut self, name: &str, if_exists: bool) -> Result<ExecutionResult> {
220        let mut catalog = self.catalog.write().expect("catalog lock poisoned");
221        self.run_in_write_txn(|txn| {
222            ddl::drop_table::execute_drop_table(txn, &mut *catalog, name, if_exists)
223        })
224    }
225
226    fn execute_create_index(
227        &mut self,
228        index: crate::catalog::IndexMetadata,
229        if_not_exists: bool,
230    ) -> Result<ExecutionResult> {
231        let mut catalog = self.catalog.write().expect("catalog lock poisoned");
232        self.run_in_write_txn(|txn| {
233            ddl::create_index::execute_create_index(txn, &mut *catalog, index, if_not_exists)
234        })
235    }
236
237    fn execute_drop_index(&mut self, name: &str, if_exists: bool) -> Result<ExecutionResult> {
238        let mut catalog = self.catalog.write().expect("catalog lock poisoned");
239        self.run_in_write_txn(|txn| {
240            ddl::drop_index::execute_drop_index(txn, &mut *catalog, name, if_exists)
241        })
242    }
243
244    // ========================================================================
245    // DML Operations (implemented in Phase 4)
246    // ========================================================================
247
248    fn execute_insert(
249        &mut self,
250        table: &str,
251        columns: Vec<String>,
252        values: Vec<Vec<crate::planner::TypedExpr>>,
253    ) -> Result<ExecutionResult> {
254        let catalog = self.catalog.read().expect("catalog lock poisoned");
255        self.run_in_write_txn(|txn| dml::execute_insert(txn, &*catalog, table, columns, values))
256    }
257
258    fn execute_insert_select(
259        &mut self,
260        table: &str,
261        columns: Vec<String>,
262        source: LogicalPlan,
263    ) -> Result<ExecutionResult> {
264        let catalog = self.catalog.read().expect("catalog lock poisoned");
265        self.run_in_write_txn(|txn| {
266            let ExecutionResult::Query(result) = query::execute_query(txn, &*catalog, source)?
267            else {
268                return Err(ExecutorError::InvalidOperation {
269                    operation: "INSERT ... SELECT".into(),
270                    reason: "SELECT source did not return query rows".into(),
271                });
272            };
273            dml::execute_insert_rows(txn, &*catalog, table, columns, result.rows)
274        })
275    }
276
277    fn execute_update(
278        &mut self,
279        table: &str,
280        assignments: Vec<crate::planner::TypedAssignment>,
281        filter: Option<crate::planner::TypedExpr>,
282    ) -> Result<ExecutionResult> {
283        let catalog = self.catalog.read().expect("catalog lock poisoned");
284        self.run_in_write_txn(|txn| dml::execute_update(txn, &*catalog, table, assignments, filter))
285    }
286
287    fn execute_delete(
288        &mut self,
289        table: &str,
290        filter: Option<crate::planner::TypedExpr>,
291    ) -> Result<ExecutionResult> {
292        let catalog = self.catalog.read().expect("catalog lock poisoned");
293        self.run_in_write_txn(|txn| dml::execute_delete(txn, &*catalog, table, filter))
294    }
295
296    // ========================================================================
297    // Query Operations (to be implemented in Phase 5)
298    // ========================================================================
299
300    fn execute_query(&mut self, plan: LogicalPlan) -> Result<ExecutionResult> {
301        if let Some(result) = system::try_execute(&self.bridge, &plan)? {
302            return Ok(result);
303        }
304        let catalog = self.catalog.read().expect("catalog lock poisoned");
305        self.run_in_write_txn(|txn| query::execute_query(txn, &*catalog, plan))
306    }
307}
308
309impl<S: KVStore> Executor<S, PersistentCatalog<S>> {
310    pub fn execute_in_txn<'a, 'b, 'c>(
311        &mut self,
312        plan: LogicalPlan,
313        txn: &mut BorrowedSqlTransaction<'a, 'b, 'c, S>,
314    ) -> Result<ExecutionResult> {
315        if txn.mode() == TxnMode::ReadOnly
316            && !matches!(
317                plan,
318                LogicalPlan::Scan { .. }
319                    | LogicalPlan::Filter { .. }
320                    | LogicalPlan::Project { .. }
321                    | LogicalPlan::Join { .. }
322                    | LogicalPlan::Aggregate { .. }
323                    | LogicalPlan::Sort { .. }
324                    | LogicalPlan::Limit { .. }
325            )
326        {
327            return Err(ExecutorError::ReadOnlyTransaction {
328                operation: plan.operation_name().to_string(),
329            });
330        }
331
332        let _statement_timestamp = evaluator::begin_statement();
333        let mut catalog = self.catalog.write().expect("catalog lock poisoned");
334        let (mut sql_txn, overlay) = txn.split_parts();
335
336        let result = match plan {
337            LogicalPlan::CreateTable {
338                table,
339                if_not_exists,
340                with_options,
341            } => self.execute_create_table_in_txn(
342                &mut *catalog,
343                &mut sql_txn,
344                overlay,
345                table,
346                with_options,
347                if_not_exists,
348            ),
349            LogicalPlan::DropTable { name, if_exists } => self.execute_drop_table_in_txn(
350                &mut *catalog,
351                &mut sql_txn,
352                overlay,
353                &name,
354                if_exists,
355            ),
356            LogicalPlan::CreateIndex {
357                index,
358                if_not_exists,
359            } => self.execute_create_index_in_txn(
360                &mut *catalog,
361                &mut sql_txn,
362                overlay,
363                index,
364                if_not_exists,
365            ),
366            LogicalPlan::DropIndex { name, if_exists } => self.execute_drop_index_in_txn(
367                &mut *catalog,
368                &mut sql_txn,
369                overlay,
370                &name,
371                if_exists,
372            ),
373            LogicalPlan::Pragma { .. } => Err(ExecutorError::UnsupportedOperation(
374                "PRAGMA is not available inside an external transaction".to_string(),
375            )),
376            LogicalPlan::Insert {
377                table,
378                columns,
379                values,
380            } => {
381                let view = TxnCatalogView::new(&*catalog, &*overlay);
382                dml::execute_insert(&mut sql_txn, &view, &table, columns, values)
383            }
384            LogicalPlan::InsertSelect {
385                table,
386                columns,
387                source,
388            } => {
389                let view = TxnCatalogView::new(&*catalog, &*overlay);
390                let ExecutionResult::Query(result) =
391                    query::execute_query(&mut sql_txn, &view, *source)?
392                else {
393                    return Err(ExecutorError::InvalidOperation {
394                        operation: "INSERT ... SELECT".into(),
395                        reason: "SELECT source did not return query rows".into(),
396                    });
397                };
398                dml::execute_insert_rows(&mut sql_txn, &view, &table, columns, result.rows)
399            }
400            LogicalPlan::Update {
401                table,
402                assignments,
403                filter,
404            } => {
405                let view = TxnCatalogView::new(&*catalog, &*overlay);
406                dml::execute_update(&mut sql_txn, &view, &table, assignments, filter)
407            }
408            LogicalPlan::Delete { table, filter } => {
409                let view = TxnCatalogView::new(&*catalog, &*overlay);
410                dml::execute_delete(&mut sql_txn, &view, &table, filter)
411            }
412            LogicalPlan::Scan { .. }
413            | LogicalPlan::Filter { .. }
414            | LogicalPlan::Project { .. }
415            | LogicalPlan::Join { .. }
416            | LogicalPlan::Aggregate { .. }
417            | LogicalPlan::Sort { .. }
418            | LogicalPlan::Limit { .. } => {
419                let view = TxnCatalogView::new(&*catalog, &*overlay);
420                query::execute_query(&mut sql_txn, &view, plan)
421            }
422        };
423
424        match result {
425            Ok(value) => {
426                sql_txn.flush_hnsw()?;
427                Ok(value)
428            }
429            Err(err) => {
430                let _ = sql_txn.abandon_hnsw();
431                Err(err)
432            }
433        }
434    }
435
436    fn map_catalog_error(err: CatalogError) -> ExecutorError {
437        match err {
438            CatalogError::Kv(e) => ExecutorError::Core(e),
439            CatalogError::Serialize(e) => ExecutorError::InvalidOperation {
440                operation: "CatalogPersistence".into(),
441                reason: e.to_string(),
442            },
443            CatalogError::InvalidKey(reason) => ExecutorError::InvalidOperation {
444                operation: "CatalogPersistence".into(),
445                reason,
446            },
447        }
448    }
449
450    fn execute_create_table_in_txn<'txn>(
451        &self,
452        catalog: &mut PersistentCatalog<S>,
453        txn: &mut impl crate::storage::SqlTxn<'txn, S>,
454        overlay: &mut CatalogOverlay,
455        mut table: crate::catalog::TableMetadata,
456        with_options: Vec<(String, String)>,
457        if_not_exists: bool,
458    ) -> Result<ExecutionResult>
459    where
460        S: 'txn,
461    {
462        if catalog.table_exists_in_txn(&table.name, overlay) {
463            return if if_not_exists {
464                Ok(ExecutionResult::Success)
465            } else {
466                Err(ExecutorError::TableAlreadyExists(table.name))
467            };
468        }
469
470        table.storage_options = ddl::create_table::parse_storage_options(&with_options)?;
471
472        let pk_index = if let Some(pk_columns) = table.primary_key.clone() {
473            let column_indices = pk_columns
474                .iter()
475                .map(|name| {
476                    table
477                        .get_column_index(name)
478                        .ok_or_else(|| ExecutorError::ColumnNotFound(name.clone()))
479                })
480                .collect::<Result<Vec<_>>>()?;
481            let index_id = catalog.next_index_id();
482            let index_name = ddl::create_pk_index_name(&table.name);
483            let mut index = crate::catalog::IndexMetadata::new(
484                index_id,
485                index_name,
486                table.name.clone(),
487                pk_columns,
488            )
489            .with_column_indices(column_indices)
490            .with_unique(true);
491            index.catalog_name = table.catalog_name.clone();
492            index.namespace_name = table.namespace_name.clone();
493            Some(index)
494        } else {
495            None
496        };
497
498        let table_id = catalog.next_table_id();
499        table = table.with_table_id(table_id);
500
501        // storage keyspace の初期化
502        txn.delete_prefix(&KeyEncoder::table_prefix(table_id))?;
503        txn.delete_prefix(&KeyEncoder::sequence_key(table_id))?;
504
505        // 永続化(同一 KV トランザクション内)
506        catalog
507            .persist_create_table(txn.inner_mut(), &table)
508            .map_err(Self::map_catalog_error)?;
509        if let Some(index) = &pk_index {
510            catalog
511                .persist_create_index(txn.inner_mut(), index)
512                .map_err(Self::map_catalog_error)?;
513        }
514
515        // オーバーレイに反映(ベースカタログはコミットまで不変)
516        overlay.add_table(TableFqn::from(&table), table);
517        if let Some(index) = pk_index {
518            overlay.add_index(IndexFqn::from(&index), index);
519        }
520
521        Ok(ExecutionResult::Success)
522    }
523
524    fn execute_drop_table_in_txn<'txn>(
525        &self,
526        catalog: &mut PersistentCatalog<S>,
527        txn: &mut impl crate::storage::SqlTxn<'txn, S>,
528        overlay: &mut CatalogOverlay,
529        table_name: &str,
530        if_exists: bool,
531    ) -> Result<ExecutionResult>
532    where
533        S: 'txn,
534    {
535        let table_meta = match catalog.get_table_in_txn(table_name, overlay) {
536            Some(table) => table.clone(),
537            None => {
538                return if if_exists {
539                    Ok(ExecutionResult::Success)
540                } else {
541                    Err(ExecutorError::TableNotFound(table_name.to_string()))
542                };
543            }
544        };
545        if table_meta.catalog_name != "default" || table_meta.namespace_name != "default" {
546            return if if_exists {
547                Ok(ExecutionResult::Success)
548            } else {
549                Err(ExecutorError::TableNotFound(table_name.to_string()))
550            };
551        }
552
553        let indexes = TxnCatalogView::new(catalog, overlay)
554            .get_indexes_for_table(table_name)
555            .into_iter()
556            .cloned()
557            .collect::<Vec<_>>();
558
559        for index in &indexes {
560            if matches!(index.method, Some(crate::ast::ddl::IndexMethod::Hnsw)) {
561                crate::executor::hnsw_bridge::HnswBridge::drop_index(txn, index, false)?;
562            } else {
563                txn.delete_prefix(&KeyEncoder::index_prefix(index.index_id))?;
564            }
565        }
566
567        txn.delete_prefix(&KeyEncoder::table_prefix(table_meta.table_id))?;
568        txn.delete_prefix(&KeyEncoder::sequence_key(table_meta.table_id))?;
569
570        catalog
571            .persist_drop_table(txn.inner_mut(), &TableFqn::from(&table_meta))
572            .map_err(Self::map_catalog_error)?;
573
574        overlay.drop_table(&TableFqn::from(&table_meta));
575
576        Ok(ExecutionResult::Success)
577    }
578
579    fn execute_create_index_in_txn<'txn>(
580        &self,
581        catalog: &mut PersistentCatalog<S>,
582        txn: &mut impl crate::storage::SqlTxn<'txn, S>,
583        overlay: &mut CatalogOverlay,
584        mut index: crate::catalog::IndexMetadata,
585        if_not_exists: bool,
586    ) -> Result<ExecutionResult>
587    where
588        S: 'txn,
589    {
590        if ddl::is_implicit_pk_index(&index.name) {
591            return Err(ExecutorError::InvalidIndexName {
592                name: index.name.clone(),
593                reason: "Index names starting with '__pk_' are reserved for PRIMARY KEY".into(),
594            });
595        }
596
597        if catalog.index_exists_in_txn(&index.name, overlay) {
598            return if if_not_exists {
599                Ok(ExecutionResult::Success)
600            } else {
601                Err(ExecutorError::IndexAlreadyExists(index.name))
602            };
603        }
604
605        let table = catalog
606            .get_table_in_txn(&index.table, overlay)
607            .ok_or_else(|| ExecutorError::TableNotFound(index.table.clone()))?
608            .clone();
609        index.catalog_name = table.catalog_name.clone();
610        index.namespace_name = table.namespace_name.clone();
611
612        let column_indices = index
613            .columns
614            .iter()
615            .map(|name| {
616                table
617                    .get_column_index(name)
618                    .ok_or_else(|| ExecutorError::ColumnNotFound(name.clone()))
619            })
620            .collect::<Result<Vec<_>>>()?;
621
622        let index_id = catalog.next_index_id();
623        index.index_id = index_id;
624        index.column_indices = column_indices.clone();
625
626        if matches!(index.method, Some(crate::ast::ddl::IndexMethod::Hnsw)) {
627            crate::executor::hnsw_bridge::HnswBridge::create_index(txn, &table, &index)?;
628        } else {
629            ddl::create_index::build_index_for_existing_rows(txn, &table, &index, column_indices)?;
630        }
631
632        catalog
633            .persist_create_index(txn.inner_mut(), &index)
634            .map_err(Self::map_catalog_error)?;
635
636        overlay.add_index(IndexFqn::from(&index), index);
637
638        Ok(ExecutionResult::Success)
639    }
640
641    fn execute_drop_index_in_txn<'txn>(
642        &self,
643        catalog: &mut PersistentCatalog<S>,
644        txn: &mut impl crate::storage::SqlTxn<'txn, S>,
645        overlay: &mut CatalogOverlay,
646        index_name: &str,
647        if_exists: bool,
648    ) -> Result<ExecutionResult>
649    where
650        S: 'txn,
651    {
652        if ddl::is_implicit_pk_index(index_name) {
653            return Err(ExecutorError::InvalidOperation {
654                operation: "DROP INDEX".into(),
655                reason: "Cannot drop implicit PRIMARY KEY index directly; use DROP TABLE".into(),
656            });
657        }
658
659        let index = match catalog.get_index_in_txn(index_name, overlay) {
660            Some(index) => index.clone(),
661            None => {
662                return if if_exists {
663                    Ok(ExecutionResult::Success)
664                } else {
665                    Err(ExecutorError::IndexNotFound(index_name.to_string()))
666                };
667            }
668        };
669        if index.catalog_name != "default" || index.namespace_name != "default" {
670            return if if_exists {
671                Ok(ExecutionResult::Success)
672            } else {
673                Err(ExecutorError::IndexNotFound(index_name.to_string()))
674            };
675        }
676
677        if matches!(index.method, Some(crate::ast::ddl::IndexMethod::Hnsw)) {
678            crate::executor::hnsw_bridge::HnswBridge::drop_index(txn, &index, if_exists)?;
679        } else {
680            txn.delete_prefix(&KeyEncoder::index_prefix(index.index_id))?;
681        }
682
683        catalog
684            .persist_drop_index(txn.inner_mut(), &IndexFqn::from(&index))
685            .map_err(Self::map_catalog_error)?;
686
687        overlay.drop_index(&IndexFqn::from(&index));
688
689        Ok(ExecutionResult::Success)
690    }
691}
692
693#[cfg(test)]
694mod tests {
695    use super::*;
696    use crate::catalog::MemoryCatalog;
697    use alopex_core::kv::memory::MemoryKV;
698
699    fn create_executor() -> Executor<MemoryKV, MemoryCatalog> {
700        let store = Arc::new(MemoryKV::new());
701        let catalog = Arc::new(RwLock::new(MemoryCatalog::new()));
702        Executor::new(store, catalog)
703    }
704
705    #[test]
706    fn test_executor_creation() {
707        let _executor = create_executor();
708        // Executor should be created without panic
709    }
710
711    #[test]
712    fn create_table_is_supported() {
713        let mut executor = create_executor();
714
715        use crate::catalog::{ColumnMetadata, TableMetadata};
716        use crate::planner::ResolvedType;
717
718        let table = TableMetadata::new(
719            "test",
720            vec![ColumnMetadata::new("id", ResolvedType::Integer)],
721        );
722
723        let result = executor.execute(LogicalPlan::CreateTable {
724            table,
725            if_not_exists: false,
726            with_options: vec![],
727        });
728        assert!(matches!(result, Ok(ExecutionResult::Success)));
729
730        let catalog = executor.catalog.read().unwrap();
731        assert!(catalog.table_exists("test"));
732    }
733
734    #[test]
735    fn insert_is_supported() {
736        use crate::Span;
737        use crate::catalog::{ColumnMetadata, TableMetadata};
738        use crate::planner::typed_expr::TypedExprKind;
739        use crate::planner::types::ResolvedType;
740
741        let mut executor = create_executor();
742
743        let table = TableMetadata::new("t", vec![ColumnMetadata::new("id", ResolvedType::Integer)])
744            .with_primary_key(vec!["id".into()]);
745
746        executor
747            .execute(LogicalPlan::CreateTable {
748                table,
749                if_not_exists: false,
750                with_options: vec![],
751            })
752            .unwrap();
753
754        let result = executor.execute(LogicalPlan::Insert {
755            table: "t".into(),
756            columns: vec!["id".into()],
757            values: vec![vec![crate::planner::typed_expr::TypedExpr {
758                kind: TypedExprKind::Literal(crate::ast::expr::Literal::Number("1".into())),
759                resolved_type: ResolvedType::Integer,
760                span: Span::default(),
761            }]],
762        });
763        assert!(matches!(result, Ok(ExecutionResult::RowsAffected(1))));
764    }
765
766    #[test]
767    fn system_pragma_and_stats_function_use_the_store() {
768        let mut executor = create_executor();
769        let catalog = MemoryCatalog::new();
770
771        let pragma = crate::Parser::parse_sql(&crate::AlopexDialect, "PRAGMA cache_size = 8")
772            .unwrap()
773            .pop()
774            .unwrap();
775        let plan = crate::Planner::new(&catalog).plan(&pragma).unwrap();
776        assert!(matches!(
777            executor.execute(plan),
778            Ok(ExecutionResult::Success)
779        ));
780
781        let select = crate::Parser::parse_sql(&crate::AlopexDialect, "SELECT memory_stats()")
782            .unwrap()
783            .pop()
784            .unwrap();
785        let plan = crate::Planner::new(&catalog).plan(&select).unwrap();
786        let result = executor.execute(plan).unwrap();
787        let ExecutionResult::Query(result) = result else {
788            panic!("expected query result");
789        };
790        assert_eq!(result.columns[0].name, "memory_stats");
791        assert!(
792            matches!(&result.rows[0][0], crate::SqlValue::Text(text) if text.contains("total_bytes"))
793        );
794    }
795}