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::Window { .. }
193 | LogicalPlan::SetOperation { .. }
194 | LogicalPlan::Sort { .. }
195 | LogicalPlan::Limit { .. } => self.execute_query(plan),
196 }
197 }
198
199 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 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 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 txn.delete_prefix(&KeyEncoder::table_prefix(table_id))?;
509 txn.delete_prefix(&KeyEncoder::sequence_key(table_id))?;
510
511 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 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 }
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}