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, UpdatePlan,
67};
68use llkv_result::{Error, Result};
69use llkv_table::{CatalogDdl, canonical_table_name};
70pub use llkv_transaction::{
71    TXN_ID_AUTO_COMMIT, TXN_ID_NONE, TransactionContext, TransactionKind, TransactionManager,
72    TransactionResult, TransactionSession, TransactionSnapshot, TxnId, TxnIdManager,
73};
74use sqlparser::ast::Select;
75
76fn is_index_not_found_error(err: &Error) -> bool {
77    matches!(err, Error::CatalogError(message) if message.contains("does not exist"))
78}
79
80fn is_table_missing_error(err: &Error) -> bool {
81    matches!(err, Error::CatalogError(message) if message.contains("does not exist"))
82}
83
84pub fn statement_table_name(statement: &PlanStatement) -> Option<&str> {
85    match statement {
86        PlanStatement::CreateTable(plan) => Some(plan.name.as_str()),
87        PlanStatement::DropTable(plan) => Some(plan.name.as_str()),
88        PlanStatement::AlterTable(plan) => Some(plan.table_name.as_str()),
89        PlanStatement::CreateIndex(plan) => Some(plan.table.as_str()),
90        PlanStatement::Insert(plan) => Some(plan.table.as_str()),
91        PlanStatement::Update(plan) => Some(plan.table.as_str()),
92        PlanStatement::Delete(plan) => Some(plan.table.as_str()),
93        PlanStatement::Select(plan) => plan
94            .tables
95            .first()
96            .map(|table_ref| table_ref.table.as_str()),
97        PlanStatement::DropIndex(_) => None,
98        PlanStatement::BeginTransaction
99        | PlanStatement::CommitTransaction
100        | PlanStatement::RollbackTransaction => None,
101    }
102}
103
104mod runtime_context;
105pub use runtime_context::RuntimeContext;
106
107// Re-export range SELECT parsing from llkv-plan
108// llkv-sql talks to llkv-runtime, which delegates to llkv-plan
109pub use llkv_plan::RangeSelectRows as RuntimeRangeSelectRows;
110
111/// Extract rows from a range() SELECT statement.
112///
113/// This is a thin wrapper around llkv_plan::extract_rows_from_range that maintains
114/// the llkv-runtime API boundary. llkv-sql should call this function, not directly
115/// access llkv-plan.
116///
117/// # Examples
118///
119/// ```ignore
120/// use llkv_runtime::extract_rows_from_range;
121/// let select = /* parse SELECT */;
122/// if let Some(rows) = extract_rows_from_range(&select)? {
123///     // Handle range query
124/// }
125/// ```
126pub fn extract_rows_from_range(select: &Select) -> Result<Option<RuntimeRangeSelectRows>> {
127    llkv_plan::extract_rows_from_range(select)
128}
129
130#[cfg(test)]
131mod tests {
132    use super::*;
133    use arrow::array::{Array, Int64Array, StringArray};
134    use arrow::datatypes::DataType;
135    use llkv_plan::{NotNull, Nullable};
136    use llkv_storage::pager::MemPager;
137    use std::sync::Arc;
138
139    #[test]
140    fn create_insert_select_roundtrip() {
141        let pager = Arc::new(MemPager::default());
142        let context = Arc::new(RuntimeContext::new(pager));
143
144        let table = context
145            .create_table(
146                "people",
147                [
148                    ("id", DataType::Int64, NotNull),
149                    ("name", DataType::Utf8, Nullable),
150                ],
151            )
152            .expect("create table");
153        table
154            .insert_rows([(1_i64, "alice"), (2_i64, "bob")])
155            .expect("insert rows");
156
157        let execution = table.lazy().expect("lazy scan");
158        let select = execution.collect().expect("build select execution");
159        let batches = select.collect().expect("collect batches");
160        assert_eq!(batches.len(), 1);
161        let column = batches[0]
162            .column(1)
163            .as_any()
164            .downcast_ref::<StringArray>()
165            .expect("string column");
166        assert_eq!(column.len(), 2);
167    }
168
169    #[test]
170    fn aggregate_count_nulls() {
171        let pager = Arc::new(MemPager::default());
172        let context = Arc::new(RuntimeContext::new(pager));
173
174        let table = context
175            .create_table("ints", [("i", DataType::Int64)])
176            .expect("create table");
177        table
178            .insert_rows([
179                (PlanValue::Null,),
180                (PlanValue::Integer(1),),
181                (PlanValue::Null,),
182            ])
183            .expect("insert rows");
184
185        let plan =
186            SelectPlan::new("ints").with_aggregates(vec![AggregateExpr::count_nulls("i", "nulls")]);
187        let snapshot = context.default_snapshot();
188        let execution = context.execute_select(plan, snapshot).expect("select");
189        let batches = execution.collect().expect("collect batches");
190        let column = batches[0]
191            .column(0)
192            .as_any()
193            .downcast_ref::<Int64Array>()
194            .expect("int column");
195        assert_eq!(column.value(0), 2);
196    }
197}
llkv_runtime/lib.rs

llkv_runtime/
lib.rs
