1#[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
59pub 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
75pub struct Executor<S: KVStore, C: Catalog> {
85 bridge: TxnBridge<S>,
87
88 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 pub fn new(store: Arc<S>, catalog: Arc<RwLock<C>>) -> Self {
117 Self {
118 bridge: TxnBridge::new(store),
119 catalog,
120 }
121 }
122
123 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 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 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 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 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 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 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 txn.delete_prefix(&KeyEncoder::table_prefix(table_id))?;
503 txn.delete_prefix(&KeyEncoder::sequence_key(table_id))?;
504
505 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 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 }
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}