wasm_dbms_sql/engine.rs
1//! SQL execution against a `wasm-dbms` database.
2
3#[cfg(test)]
4mod tests;
5
6use wasm_dbms::prelude::{DatabaseSchema, DbmsContext, WasmDbmsDatabase};
7use wasm_dbms_api::prelude::{
8 AggregatedRow, AggregatedValue, ColumnDef, Database as _, JoinColumnDef, SqlError, SqlResult,
9 SqlRow, TransactionId, Value,
10};
11use wasm_dbms_memory::prelude::MemoryProvider;
12
13use crate::ast::Statement;
14use crate::parser::parse;
15use crate::planner::{self, AggregateOutput, AggregateSource, OutputColumn, SelectPlan};
16
17/// Runs SQL statements against a `wasm-dbms` database.
18///
19/// The engine holds the database schema and nothing else. It does not borrow
20/// the [`DbmsContext`]: the context is passed to every
21/// [`execute`](Self::execute) call, so one engine can serve several contexts
22/// and can be stored next to its context, for example in a `thread_local!`.
23///
24/// # Transactions
25///
26/// Transactions are addressed by their [`TransactionId`], the same ids that
27/// [`DbmsContext::begin_transaction`] returns. `BEGIN` opens a transaction and
28/// returns its id in [`SqlResult::TxBegin`]; passing that id to later calls
29/// runs them inside the transaction, where they see its uncommitted writes.
30/// `COMMIT` and `ROLLBACK` close the transaction whose id they receive. A
31/// statement that fails leaves the transaction open. Calls without an id are
32/// applied immediately and do not see open transactions.
33///
34/// The engine does not check who holds an id. An application layer that
35/// serves several identities keeps its own ledger from id to identity and
36/// checks it before calling `execute` with that id.
37///
38/// # Examples
39///
40/// ```rust,ignore
41/// use wasm_dbms::prelude::*;
42/// use wasm_dbms_api::prelude::*;
43/// use wasm_dbms_memory::prelude::HeapMemoryProvider;
44/// use wasm_dbms_sql::SqlEngine;
45///
46/// #[derive(Clone, DatabaseSchema)]
47/// #[tables(User = "users")]
48/// pub struct MySchema;
49///
50/// let ctx = DbmsContext::new(HeapMemoryProvider::default());
51/// MySchema::register_tables(&ctx)?;
52/// let engine = SqlEngine::new(MySchema);
53///
54/// engine.execute(
55/// &ctx,
56/// None,
57/// "INSERT INTO users (id, name) VALUES (?, ?)",
58/// &[Value::from(1u32), Value::from("Alice")],
59/// )?;
60///
61/// let SqlResult::TxBegin(tx) = engine.execute(&ctx, None, "BEGIN", &[])? else {
62/// unreachable!("BEGIN returns TxBegin");
63/// };
64/// engine.execute(&ctx, Some(tx), "UPDATE users SET name = 'Alicia' WHERE id = 1", &[])?;
65/// engine.execute(&ctx, Some(tx), "COMMIT", &[])?;
66/// ```
67pub struct SqlEngine<S> {
68 schema: S,
69}
70
71impl<S> SqlEngine<S> {
72 /// Creates an engine for the tables of `schema`.
73 pub fn new(schema: S) -> Self {
74 Self { schema }
75 }
76
77 /// Parses and runs one SQL statement, outside or inside a transaction.
78 ///
79 /// `tx` selects where the statement runs: `None` applies it immediately
80 /// through [`WasmDbmsDatabase::oneshot`], `Some(id)` runs it through
81 /// [`WasmDbmsDatabase::from_transaction`]. `BEGIN` requires `None` and
82 /// returns the new id in [`SqlResult::TxBegin`]; `COMMIT` and `ROLLBACK`
83 /// require `Some(id)`. `params` supplies one value per `?` placeholder, in
84 /// order.
85 ///
86 /// # Errors
87 ///
88 /// - [`SqlError::Parse`] and [`SqlError::MissingWhereClause`] when `sql`
89 /// is not a valid statement.
90 /// - [`SqlError::ParameterCountMismatch`] when `params` does not have one
91 /// value per placeholder.
92 /// - [`SqlError::UnknownTable`], [`SqlError::UnknownColumn`], and
93 /// [`SqlError::AmbiguousColumn`] when a name does not resolve.
94 /// - [`SqlError::TypeMismatch`] and [`SqlError::InvalidLiteral`] when a
95 /// value cannot be converted to the type of its column.
96 /// - [`SqlError::Unsupported`] when the statement combines features the
97 /// DBMS cannot run together.
98 /// - [`SqlError::TransactionAlreadyActive`] for `BEGIN` with `Some(id)`.
99 /// - [`SqlError::NoActiveTransaction`] for `COMMIT` or `ROLLBACK` with
100 /// `None`.
101 /// - [`SqlError::Runtime`] when the DBMS rejects the operation, for
102 /// example on a constraint violation, or when `id` does not name an
103 /// open transaction.
104 pub fn execute<'ctx, M>(
105 &self,
106 ctx: &'ctx DbmsContext<M>,
107 tx: Option<TransactionId>,
108 sql: &str,
109 params: &[Value],
110 ) -> Result<SqlResult, SqlError>
111 where
112 M: MemoryProvider,
113 S: DatabaseSchema<M> + Clone + 'ctx,
114 {
115 let parsed = parse(sql)?;
116 if parsed.parameter_count != params.len() {
117 return Err(SqlError::ParameterCountMismatch {
118 expected: parsed.parameter_count,
119 got: params.len(),
120 });
121 }
122 match (&parsed.statement, tx) {
123 (Statement::Begin, None) => Ok(SqlResult::TxBegin(ctx.begin_transaction())),
124 (Statement::Begin, Some(_)) => Err(SqlError::TransactionAlreadyActive),
125 (Statement::Commit | Statement::Rollback, None) => Err(SqlError::NoActiveTransaction),
126 (Statement::Commit, Some(id)) => {
127 WasmDbmsDatabase::from_transaction(ctx, self.schema.clone(), id).commit()?;
128 Ok(SqlResult::TxCommit)
129 }
130 (Statement::Rollback, Some(id)) => {
131 WasmDbmsDatabase::from_transaction(ctx, self.schema.clone(), id).rollback()?;
132 Ok(SqlResult::TxRollback)
133 }
134 (statement, Some(id)) => self.run(
135 &WasmDbmsDatabase::from_transaction(ctx, self.schema.clone(), id),
136 statement,
137 params,
138 ),
139 (statement, None) => self.run(
140 &WasmDbmsDatabase::oneshot(ctx, self.schema.clone()),
141 statement,
142 params,
143 ),
144 }
145 }
146
147 /// Plans and runs a data statement on `db`.
148 fn run<M>(
149 &self,
150 db: &WasmDbmsDatabase<'_, M>,
151 statement: &Statement,
152 params: &[Value],
153 ) -> Result<SqlResult, SqlError>
154 where
155 M: MemoryProvider,
156 S: DatabaseSchema<M>,
157 {
158 let catalog = |table: &str| self.schema.table_columns(table).ok();
159 match statement {
160 Statement::Select(select) => {
161 let rows = match planner::select(select, &catalog, params)? {
162 SelectPlan::Table(plan) => db
163 .select_raw(&plan.table, plan.query)?
164 .into_iter()
165 .map(|row| table_row(row, plan.projection.as_deref()))
166 .collect(),
167 SelectPlan::Join(plan) => db
168 .select_join(&plan.table, plan.query)?
169 .into_iter()
170 .map(|row| join_row(row, plan.projection.as_deref()))
171 .collect(),
172 SelectPlan::Aggregate(plan) => self
173 .schema
174 .aggregate(db, &plan.table, plan.query, &plan.aggregates)?
175 .into_iter()
176 .map(|row| aggregate_row(&row, &plan.outputs))
177 .collect(),
178 };
179 Ok(SqlResult::Rows(rows))
180 }
181 Statement::Insert(insert) => {
182 let plan = planner::insert(insert, &catalog, params)?;
183 let table = self.schema.static_table_name(&plan.table)?;
184 self.schema.insert(db, table, &plan.values)?;
185 Ok(SqlResult::RowsAffected(1))
186 }
187 Statement::Update(update) => {
188 let plan = planner::update(update, &catalog, params)?;
189 let table = self.schema.static_table_name(&plan.table)?;
190 let count = self
191 .schema
192 .update(db, table, &plan.values, Some(plan.filter))?;
193 Ok(SqlResult::RowsAffected(count))
194 }
195 Statement::Delete(delete) => {
196 let plan = planner::delete(delete, &catalog, params)?;
197 let table = self.schema.static_table_name(&plan.table)?;
198 let count = self
199 .schema
200 .delete(db, table, plan.behavior, Some(plan.filter))?;
201 Ok(SqlResult::RowsAffected(count))
202 }
203 Statement::Begin | Statement::Commit | Statement::Rollback => {
204 unreachable!("transaction statements are handled by `execute`")
205 }
206 }
207 }
208}
209
210impl<S> std::fmt::Debug for SqlEngine<S> {
211 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
212 f.debug_struct("SqlEngine").finish_non_exhaustive()
213 }
214}
215
216/// Shapes a row of a single-table query.
217fn table_row(row: Vec<(ColumnDef, Value)>, projection: Option<&[OutputColumn]>) -> SqlRow {
218 let Some(projection) = projection else {
219 return row
220 .into_iter()
221 .map(|(def, value)| (JoinColumnDef::from(def), value))
222 .collect();
223 };
224 projection
225 .iter()
226 .map(|output| {
227 let (def, value) = row
228 .iter()
229 .find(|(def, _)| def.name == output.column)
230 .expect("planned column must exist in the selected table row");
231 let mut def = JoinColumnDef::from(*def);
232 def.name = output.name.clone();
233 (def, value.clone())
234 })
235 .collect()
236}
237
238/// Shapes a row of a join query, whose columns carry their table name.
239fn join_row(row: SqlRow, projection: Option<&[OutputColumn]>) -> SqlRow {
240 let Some(projection) = projection else {
241 return row;
242 };
243 projection
244 .iter()
245 .map(|output| {
246 let (def, value) = row
247 .iter()
248 .find(|(def, _)| def.table == output.table && def.name == output.column)
249 .expect("planned column must exist in the selected join row");
250 let mut def = def.clone();
251 def.name = output.name.clone();
252 (def, value.clone())
253 })
254 .collect()
255}
256
257/// Shapes an aggregated row into the columns of the select list.
258fn aggregate_row(row: &AggregatedRow, outputs: &[AggregateOutput]) -> SqlRow {
259 outputs
260 .iter()
261 .map(|output| {
262 let value = match output.source {
263 AggregateSource::GroupKey(index) => row
264 .group_keys
265 .get(index)
266 .cloned()
267 .expect("planned group key must exist in the aggregated row"),
268 AggregateSource::Aggregate(index) => row
269 .values
270 .get(index)
271 .map(|value| match value {
272 AggregatedValue::Count(count) => Value::from(*count),
273 AggregatedValue::Sum(value)
274 | AggregatedValue::Avg(value)
275 | AggregatedValue::Min(value)
276 | AggregatedValue::Max(value) => value.clone(),
277 })
278 .expect("planned aggregate must exist in the aggregated row"),
279 };
280 (output.def.clone(), value)
281 })
282 .collect()
283}