llkv_runtime/
lib.rs

1//! Query execution runtime for LLKV.
2//!
3//! This crate provides the runtime API (see [`RuntimeEngine`]) for executing SQL plans with full
4//! transaction support. It coordinates between the transaction layer, storage layer,
5//! and query executor to provide a complete database runtime.
6//!
7//! # Key Components
8//!
9//! - **[`RuntimeEngine`]**: Main execution engine for SQL operations
10//! - **[`RuntimeSession`]**: Session-level interface with transaction management
11//! - **[`TransactionContext`]**: Single-transaction execution context
12//! - **Table Provider**: Integration with the query executor for table access
13//!
14//! # Transaction Support
15//!
16//! The runtime supports both:
17//! - **Auto-commit**: Single-statement transactions (uses `TXN_ID_AUTO_COMMIT`)
18//! - **Multi-statement**: Explicit BEGIN/COMMIT/ROLLBACK transactions
19//!
20//! # MVCC Integration
21//!
22//! All data modifications automatically include MVCC metadata:
23//! - `row_id`: Unique row identifier
24//! - `created_by`: Transaction ID that created the row
25//! - `deleted_by`: Transaction ID that deleted the row (or `TXN_ID_NONE`)
26//!
27//! The runtime ensures these columns are injected and managed consistently.
28
29#![forbid(unsafe_code)]
30
31pub mod runtime_storage_namespace;
32
33pub use runtime_storage_namespace::{
34    PERSISTENT_NAMESPACE_ID, PersistentRuntimeNamespace, RuntimeNamespaceId,
35    RuntimeStorageNamespace, RuntimeStorageNamespaceOps, RuntimeStorageNamespaceRegistry,
36    TEMPORARY_NAMESPACE_ID, TemporaryRuntimeNamespace,
37};
38
39mod runtime_statement_result;
40pub use runtime_statement_result::RuntimeStatementResult;
41
42mod runtime_transaction_context;
43pub use runtime_transaction_context::RuntimeTransactionContext;
44
45mod runtime_session;
46pub use runtime_session::RuntimeSession;
47
48mod runtime_engine;
49pub use runtime_engine::RuntimeEngine;
50
51mod runtime_lazy_frame;
52pub use runtime_lazy_frame::RuntimeLazyFrame;
53
54mod runtime_table;
55pub use runtime_table::{
56    IntoInsertRow, RuntimeCreateTableBuilder, RuntimeInsertRowKind, RuntimeRow, RuntimeTableHandle,
57    row,
58};
59
60pub use llkv_executor::SelectExecution;
61pub use llkv_plan::{
62    AggregateExpr, AlterTablePlan, AssignmentValue, ColumnAssignment, CreateIndexPlan,
63    CreateTablePlan, CreateTableSource, DeletePlan, DropIndexPlan, DropTablePlan, ForeignKeyAction,
64    ForeignKeySpec, IndexColumnPlan, InsertPlan, InsertSource, IntoPlanColumnSpec,
65    MultiColumnUniqueSpec, OrderByPlan, OrderSortType, OrderTarget, PlanColumnSpec, PlanOperation,
66    PlanStatement, PlanValue, RenameTablePlan, SelectPlan, SelectProjection, TruncatePlan,
67    UpdatePlan,
68};
69use llkv_result::{Error, Result};
70use llkv_table::{CatalogDdl, canonical_table_name};
71pub use llkv_transaction::{
72    TXN_ID_AUTO_COMMIT, TXN_ID_NONE, TransactionContext, TransactionKind, TransactionManager,
73    TransactionResult, TransactionSession, TransactionSnapshot, TxnId, TxnIdManager,
74};
75use sqlparser::ast::Select;
76
77fn is_index_not_found_error(err: &Error) -> bool {
78    matches!(err, Error::CatalogError(message) if message.contains("does not exist"))
79}
80
81fn is_table_missing_error(err: &Error) -> bool {
82    matches!(err, Error::CatalogError(message) if message.contains("does not exist"))
83}
84
85pub fn statement_table_name(statement: &PlanStatement) -> Option<&str> {
86    match statement {
87        PlanStatement::CreateTable(plan) => Some(plan.name.as_str()),
88        PlanStatement::DropTable(plan) => Some(plan.name.as_str()),
89        PlanStatement::AlterTable(plan) => Some(plan.table_name.as_str()),
90        PlanStatement::CreateIndex(plan) => Some(plan.table.as_str()),
91        PlanStatement::Insert(plan) => Some(plan.table.as_str()),
92        PlanStatement::Update(plan) => Some(plan.table.as_str()),
93        PlanStatement::Delete(plan) => Some(plan.table.as_str()),
94        PlanStatement::Truncate(plan) => Some(plan.table.as_str()),
95        PlanStatement::Select(plan) => plan
96            .tables
97            .first()
98            .map(|table_ref| table_ref.table.as_str()),
99        PlanStatement::DropIndex(_) => None,
100        PlanStatement::BeginTransaction
101        | PlanStatement::CommitTransaction
102        | PlanStatement::RollbackTransaction => None,
103    }
104}
105
106mod runtime_context;
107pub use runtime_context::RuntimeContext;
108
109// Re-export range SELECT parsing from llkv-plan
110// llkv-sql talks to llkv-runtime, which delegates to llkv-plan
111pub use llkv_plan::RangeSelectRows as RuntimeRangeSelectRows;
112
113/// Extract rows from a range() SELECT statement.
114///
115/// This is a thin wrapper around llkv_plan::extract_rows_from_range that maintains
116/// the llkv-runtime API boundary. llkv-sql should call this function, not directly
117/// access llkv-plan.
118///
119/// # Examples
120///
121/// ```ignore
122/// use llkv_runtime::extract_rows_from_range;
123/// let select = /* parse SELECT */;
124/// if let Some(rows) = extract_rows_from_range(&select)? {
125///     // Handle range query
126/// }
127/// ```
128pub fn extract_rows_from_range(select: &Select) -> Result<Option<RuntimeRangeSelectRows>> {
129    llkv_plan::extract_rows_from_range(select)
130}
131
132#[cfg(test)]
133mod tests {
134    use super::*;
135    use arrow::array::{Array, Int64Array, StringArray};
136    use arrow::datatypes::DataType;
137    use llkv_plan::{NotNull, Nullable};
138    use llkv_storage::pager::MemPager;
139    use std::sync::Arc;
140
141    #[test]
142    fn create_insert_select_roundtrip() {
143        let pager = Arc::new(MemPager::default());
144        let context = Arc::new(RuntimeContext::new(pager));
145
146        let table = context
147            .create_table(
148                "people",
149                [
150                    ("id", DataType::Int64, NotNull),
151                    ("name", DataType::Utf8, Nullable),
152                ],
153            )
154            .expect("create table");
155        table
156            .insert_rows([(1_i64, "alice"), (2_i64, "bob")])
157            .expect("insert rows");
158
159        let execution = table.lazy().expect("lazy scan");
160        let select = execution.collect().expect("build select execution");
161        let batches = select.collect().expect("collect batches");
162        assert_eq!(batches.len(), 1);
163        let column = batches[0]
164            .column(1)
165            .as_any()
166            .downcast_ref::<StringArray>()
167            .expect("string column");
168        assert_eq!(column.len(), 2);
169    }
170
171    #[test]
172    fn aggregate_count_nulls() {
173        let pager = Arc::new(MemPager::default());
174        let context = Arc::new(RuntimeContext::new(pager));
175
176        let table = context
177            .create_table("ints", [("i", DataType::Int64)])
178            .expect("create table");
179        table
180            .insert_rows([
181                (PlanValue::Null,),
182                (PlanValue::Integer(1),),
183                (PlanValue::Null,),
184            ])
185            .expect("insert rows");
186
187        let plan =
188            SelectPlan::new("ints").with_aggregates(vec![AggregateExpr::count_nulls("i", "nulls")]);
189        let snapshot = context.default_snapshot();
190        let execution = context.execute_select(plan, snapshot).expect("select");
191        let batches = execution.collect().expect("collect batches");
192        let column = batches[0]
193            .column(0)
194            .as_any()
195            .downcast_ref::<Int64Array>()
196            .expect("int column");
197        assert_eq!(column.value(0), 2);
198    }
199}
llkv_runtime/lib.rs

llkv_runtime/
lib.rs
