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, JoinResultSet,
9 SqlError, SqlResult, 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) => join_rows(
168 db.select_join(&plan.table, plan.query)?,
169 plan.projection.as_deref(),
170 ),
171 SelectPlan::Aggregate(plan) => self
172 .schema
173 .aggregate(db, &plan.table, plan.query, &plan.aggregates)?
174 .into_iter()
175 .map(|row| aggregate_row(&row, &plan.outputs))
176 .collect(),
177 };
178 Ok(SqlResult::Rows(rows))
179 }
180 Statement::Insert(insert) => {
181 let plan = planner::insert(insert, &catalog, params)?;
182 let table = self.schema.static_table_name(&plan.table)?;
183 self.schema.insert(db, table, &plan.values)?;
184 Ok(SqlResult::RowsAffected(1))
185 }
186 Statement::Update(update) => {
187 let plan = planner::update(update, &catalog, params)?;
188 let table = self.schema.static_table_name(&plan.table)?;
189 let count = self
190 .schema
191 .update(db, table, &plan.values, Some(plan.filter))?;
192 Ok(SqlResult::RowsAffected(count))
193 }
194 Statement::Delete(delete) => {
195 let plan = planner::delete(delete, &catalog, params)?;
196 let table = self.schema.static_table_name(&plan.table)?;
197 let count = self
198 .schema
199 .delete(db, table, plan.behavior, Some(plan.filter))?;
200 Ok(SqlResult::RowsAffected(count))
201 }
202 Statement::Begin | Statement::Commit | Statement::Rollback => {
203 unreachable!("transaction statements are handled by `execute`")
204 }
205 }
206 }
207}
208
209impl<S> std::fmt::Debug for SqlEngine<S> {
210 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
211 f.debug_struct("SqlEngine").finish_non_exhaustive()
212 }
213}
214
215/// Shapes a row of a single-table query.
216fn table_row(row: Vec<(ColumnDef, Value)>, projection: Option<&[OutputColumn]>) -> SqlRow {
217 let Some(projection) = projection else {
218 return row
219 .into_iter()
220 .map(|(def, value)| (JoinColumnDef::from(def), value))
221 .collect();
222 };
223 projection
224 .iter()
225 .map(|output| {
226 let (def, value) = row
227 .iter()
228 .find(|(def, _)| def.name == output.column)
229 .expect("planned column must exist in the selected table row");
230 let mut def = JoinColumnDef::from(*def);
231 def.name = output.name.clone();
232 (def, value.clone())
233 })
234 .collect()
235}
236
237/// Shapes a join result into SQL rows, applying the select-list projection.
238///
239/// The projection is resolved once against the column list; every row is
240/// then built by index.
241fn join_rows(result: JoinResultSet, projection: Option<&[OutputColumn]>) -> Vec<SqlRow> {
242 let JoinResultSet { columns, rows } = result;
243 let outputs: Vec<(usize, JoinColumnDef)> = match projection {
244 None => columns.into_iter().enumerate().collect(),
245 Some(projection) => projection
246 .iter()
247 .map(|output| {
248 let index = columns
249 .iter()
250 .position(|def| def.table == output.table && def.name == output.column)
251 .expect("planned column must exist in the selected join row");
252 let mut def = columns[index].clone();
253 def.name = output.name.clone();
254 (index, def)
255 })
256 .collect(),
257 };
258
259 rows.into_iter()
260 .map(|row| {
261 outputs
262 .iter()
263 .map(|(index, def)| (def.clone(), row[*index].clone()))
264 .collect()
265 })
266 .collect()
267}
268
269/// Shapes an aggregated row into the columns of the select list.
270fn aggregate_row(row: &AggregatedRow, outputs: &[AggregateOutput]) -> SqlRow {
271 outputs
272 .iter()
273 .map(|output| {
274 let value = match output.source {
275 AggregateSource::GroupKey(index) => row
276 .group_keys
277 .get(index)
278 .cloned()
279 .expect("planned group key must exist in the aggregated row"),
280 AggregateSource::Aggregate(index) => row
281 .values
282 .get(index)
283 .map(|value| match value {
284 AggregatedValue::Count(count) => Value::from(*count),
285 AggregatedValue::Sum(value)
286 | AggregatedValue::Avg(value)
287 | AggregatedValue::Min(value)
288 | AggregatedValue::Max(value) => value.clone(),
289 })
290 .expect("planned aggregate must exist in the aggregated row"),
291 };
292 (output.def.clone(), value)
293 })
294 .collect()
295}