siderite_backends/
mysql.rs

1//! MySQL backend built on SQLx (MySQL 8.0.31+, tested on 9.x).
2//!
3//! # Storage
4//!
5//! | [`Value`] | Column type | Notes |
6//! |---|---|---|
7//! | `Bool` | `TINYINT(1)` | read back as `Bool` (a bare `TINYINT` reads as `Int`) |
8//! | `Int` | `BIGINT` / `INT` / ... | unsigned columns above `i64::MAX` fail to decode |
9//! | `Float` | `DOUBLE` | |
10//! | `Decimal` | `DECIMAL(p, s)` | |
11//! | `Text` | `VARCHAR(n)` / `TEXT` | comparisons follow the column collation |
12//! | `Bytes` | `BLOB` / `VARBINARY` | |
13//! | `Uuid` | `CHAR(36)` | bound as hyphenated text, not as 16 raw bytes |
14//! | `Date` / `Time` | `DATE` / `TIME(6)` | |
15//! | `Timestamp` | `DATETIME(6)` | UTC; connections are pinned to `+00:00` |
16//! | `Json` | `JSON` | |
17//!
18//! # No `RETURNING`
19//!
20//! MySQL has no `RETURNING`, but the ORM's `save`, `create` and
21//! `bulk_create` need the stored rows back, so this adapter emulates it:
22//!
23//! * **`INSERT`**: a generated `AUTO_INCREMENT` key is recovered from the
24//!   OK packet's `LAST_INSERT_ID()` plus the consecutive values InnoDB hands
25//!   out to one multi-row `INSERT` (`innodb_autoinc_lock_mode` 0, 1 and 2 all
26//!   allocate the whole block of a plain `INSERT .. VALUES` at once; the step
27//!   is `@@auto_increment_increment`). Supplied keys are used as they are.
28//!   The rows are then re-read by key, in key order.
29//! * **`UPDATE` / `DELETE`**: the affected keys are read `FOR UPDATE` (an
30//!   `UPDATE` by primary-key equality skips that read), the statement runs,
31//!   and the rows are read back (after an `UPDATE`) or were read beforehand
32//!   (`DELETE`).
33//!
34//! Each emulated write that reads rows back, single-row ones included, runs
35//! in a transaction (or the caller's), so the rows read back are the ones
36//! written: no concurrent statement can change or delete them in between.
37//!
38//! A **single-row `INSERT`** skips the read-back, and with it the
39//! transaction, when the stored row is known without asking: the table has no
40//! `INSERT` trigger (which takes the `TRIGGER` privilege, granted directly
41//! and not through a role, to find out; without it the row is read back),
42//! and every returned column is either the omitted
43//! `AUTO_INCREMENT` column (taken from the OK packet) or a supplied value its
44//! column stores unchanged. That holds for integers within the range of a
45//! signed integer column, booleans in `TINYINT(1)`, and text that fits a
46//! `VARCHAR` / `TEXT` column (`utf8mb4`, or ASCII in any character set).
47//! Everything else (defaults, generated columns, `DECIMAL`, temporal, `JSON`,
48//! binary and `CHAR` columns) is read back as described above.
49//!
50//! The columns of a table and whether it has an `INSERT` trigger are looked up
51//! once per table in `information_schema` and cached;
52//! [`execute_script`](Executor::execute_script) clears the cache. A table or
53//! trigger changed by other means (another process, `execute_raw`) while the
54//! backend is in use is not noticed until then. A table without a primary key
55//! cannot use the emulation.
56//!
57//! # Session setup
58//!
59//! [`MySqlBackend::connect`] configures every connection: `time_zone` is
60//! `+00:00` (so `DATETIME`, `TIMESTAMP` and date parts agree with the UTC
61//! values the ORM writes), `NO_BACKSLASH_ESCAPES` is removed from `sql_mode`
62//! (the compiler writes `'\\'` string literals) and `group_concat_max_len` is
63//! raised (`GROUP_CONCAT` silently truncates at the default 1024 bytes).
64//! Pools built elsewhere and passed to [`MySqlBackend::from_pool`] should do
65//! the same.
66
67use crate::connection_init::ConnectionInit;
68use crate::shared::{ConnSlot, TxSlot, affected_result, map_error, with_conn, with_tx};
69use crate::sql::{MySql, compile, compile_write};
70use async_trait::async_trait;
71use siderite_orm::{
72    Backend, BackendCapabilities, BackendError, ExecResult, Executor, IsolationLevel, OrmError,
73    QueryPlan, QueryResult, Transaction, Value, WritePlan,
74};
75use sqlx::Executor as _;
76use sqlx::mysql::{MySqlPool, MySqlPoolOptions, MySqlQueryResult};
77use std::sync::Arc;
78
79mod io;
80mod metadata;
81#[cfg(feature = "mysql-native")]
82pub mod native;
83mod returning;
84mod sqlx_io;
85
86use metadata::TableCache;
87use returning::{prepare, synthesize, synthesized_result, table_of, write_returning};
88use sqlx_io::{execute_rows, fetch_rows, run};
89
90/// Session statements run on every new connection (see the module docs).
91const SESSION_SETUP: [&str; 3] = [
92    "SET time_zone = '+00:00'",
93    "SET SESSION sql_mode = REPLACE(@@sql_mode, 'NO_BACKSLASH_ESCAPES', '')",
94    "SET SESSION group_concat_max_len = 4294967295",
95];
96
97/// MySQL adapter executing compiled plans on a connection pool.
98#[derive(Debug, Clone)]
99pub struct MySqlBackend {
100    pool: MySqlPool,
101    tables: Arc<TableCache>,
102}
103
104impl MySqlBackend {
105    /// Connect to `url` (`mysql://user:pass@host/db`) with a pool of up to
106    /// ten connections, each configured as described in the module docs.
107    ///
108    /// A connection is **not** checked when it is taken from the pool: SQLx's
109    /// `test_before_acquire` costs a round trip per query. The price is that
110    /// the first query on a connection the server has closed (a restart,
111    /// `wait_timeout`) fails with a connection error instead of being retried
112    /// on a fresh connection. To have connections checked, use
113    /// [`connect_with`](Self::connect_with) with
114    /// `MySqlPoolOptions::new().test_before_acquire(true)`.
115    ///
116    /// # Errors
117    /// [`BackendError::Connection`] if the pool cannot be created.
118    pub async fn connect(url: &str) -> Result<Self, BackendError> {
119        let options = MySqlPoolOptions::new()
120            .max_connections(10)
121            .test_before_acquire(false);
122        Self::connect_with(url, options).await
123    }
124
125    /// Connect with pool settings and mandatory adapter initialization.
126    ///
127    /// Replaces any after_connect callback already stored in options.
128    /// SQLx does not expose that callback for chaining; pass custom setup to
129    /// [`Self::connect_with_init`] instead. Other pool options are retained.
130    ///
131    /// Args:
132    ///     url: Database URL.
133    ///     options: Pool capacity, timeouts and checkout policy.
134    ///
135    /// Returns:
136    ///     Adapter whose fresh sessions receive mandatory initialization.
137    ///
138    /// # Errors
139    /// Connection or initialization failure within the acquire deadline.
140    pub async fn connect_with(url: &str, options: MySqlPoolOptions) -> Result<Self, BackendError> {
141        let init: ConnectionInit<sqlx::MySql> =
142            std::sync::Arc::new(|_, _| Box::pin(async { Ok(()) }));
143        Self::connect_with_init(url, options, init).await
144    }
145
146    /// Compose explicit custom setup with mandatory session settings.
147    ///
148    /// Args:
149    ///     url: Database URL.
150    ///     options: Pool settings; its stored after_connect is replaced.
151    ///     init: Custom hook run first on every new/reconnected session.
152    ///
153    /// Returns:
154    ///     Adapter after custom setup and required settings succeed.
155    ///
156    /// # Errors
157    /// Connection or initialization failure within the acquire deadline.
158    pub async fn connect_with_init(
159        url: &str,
160        options: MySqlPoolOptions,
161        init: ConnectionInit<sqlx::MySql>,
162    ) -> Result<Self, BackendError> {
163        let pool = options
164            .after_connect(move |conn, meta| {
165                let init = init.clone();
166                Box::pin(async move {
167                    init(conn, meta).await?;
168                    for statement in SESSION_SETUP {
169                        conn.execute(statement).await?;
170                    }
171                    Ok(())
172                })
173            })
174            .connect(url)
175            .await
176            .map_err(|e| BackendError::Connection(e.to_string()))?;
177        Ok(Self::from_pool(pool))
178    }
179
180    /// Wrap an existing pool. Its connections should use the session setup
181    /// described in the module docs.
182    pub fn from_pool(pool: MySqlPool) -> Self {
183        Self {
184            pool,
185            tables: Arc::default(),
186        }
187    }
188}
189
190#[async_trait]
191impl Executor for MySqlBackend {
192    fn capabilities(&self) -> BackendCapabilities {
193        BackendCapabilities::mysql()
194    }
195
196    async fn fetch(&self, plan: &QueryPlan) -> Result<QueryResult, OrmError> {
197        let compiled = compile(plan, &MySql)?;
198        fetch_rows(&self.pool, &compiled.sql, compiled.params).await
199    }
200
201    async fn execute(&self, plan: &WritePlan) -> Result<ExecResult, OrmError> {
202        if plan.returning().is_empty() {
203            let compiled = compile_write(plan, &MySql)?;
204            let affected = execute_rows(&self.pool, &compiled.sql, compiled.params).await?;
205            return Ok(affected_result(affected));
206        }
207        let prepared = prepare(plan)?;
208        let info = self.tables.table(&mut &self.pool, table_of(plan)).await?;
209        if let WritePlan::Insert(p) = plan
210            && let Some(row) = synthesize(&info, p)
211        {
212            // One statement and nothing to read back: no transaction needed.
213            let write = prepared.write;
214            let done = run(&self.pool, &write.sql, write.params).await?;
215            return synthesized_result(&info, row, &done);
216        }
217        // A write that is read back runs in a transaction even for a single
218        // row: the write's row lock is held until the read-back, so a
219        // concurrent update or delete cannot change or remove the row in
220        // between.
221        let mut tx = self.pool.begin().await.map_err(db_error)?;
222        let result = write_returning(&mut *tx, &info, plan, prepared).await?;
223        tx.commit().await.map_err(db_error)?;
224        Ok(result)
225    }
226
227    async fn fetch_raw(&self, sql: &str, params: Vec<Value>) -> Result<QueryResult, OrmError> {
228        fetch_rows(&self.pool, sql, params).await
229    }
230
231    async fn execute_raw(&self, sql: &str, params: Vec<Value>) -> Result<u64, OrmError> {
232        execute_rows(&self.pool, sql, params).await
233    }
234
235    async fn execute_script(&self, sql: &str) -> Result<(), OrmError> {
236        let done = sqlx::Executor::execute(&self.pool, sql).await;
237        // The script may have changed keys or dropped tables.
238        self.tables.clear();
239        script_result(done)
240    }
241}
242
243#[async_trait]
244impl Backend for MySqlBackend {
245    fn read_parameter_count(&self, plan: &QueryPlan) -> Result<Option<usize>, OrmError> {
246        Ok(Some(compile(plan, &MySql)?.params.len()))
247    }
248
249    async fn begin(
250        &self,
251        isolation: Option<IsolationLevel>,
252    ) -> Result<Box<dyn Transaction>, OrmError> {
253        let tx = match isolation {
254            None => self.pool.begin().await,
255            Some(level) => {
256                self.capabilities()
257                    .require(siderite_orm::Feature::Isolation(level))?;
258                // `SET TRANSACTION` applies to the next transaction only and
259                // is an error once one has started, so it must come first.
260                let statement = format!(
261                    "SET TRANSACTION ISOLATION LEVEL {}; START TRANSACTION",
262                    isolation_sql(level)
263                );
264                self.pool.begin_with(statement).await
265            }
266        }
267        .map_err(db_error)?;
268        Ok(Box::new(MySqlTransaction {
269            slot: TxSlot::new(tx),
270            tables: Arc::clone(&self.tables),
271        }))
272    }
273
274    async fn begin_schema(&self, transactional: bool) -> Result<Box<dyn Transaction>, OrmError> {
275        if transactional {
276            return self.begin(None).await;
277        }
278        let conn = self.pool.acquire().await.map_err(db_error)?;
279        Ok(Box::new(MySqlHeld {
280            slot: ConnSlot::new(conn),
281            tables: Arc::clone(&self.tables),
282        }))
283    }
284}
285
286fn isolation_sql(level: IsolationLevel) -> &'static str {
287    match level {
288        IsolationLevel::ReadCommitted => "READ COMMITTED",
289        IsolationLevel::RepeatableRead => "REPEATABLE READ",
290        IsolationLevel::Serializable => "SERIALIZABLE",
291    }
292}
293
294/// An open MySQL transaction. Dropped without commit, it rolls back.
295struct MySqlTransaction {
296    slot: TxSlot<sqlx::MySql>,
297    tables: Arc<TableCache>,
298}
299
300#[async_trait]
301impl Executor for MySqlTransaction {
302    fn capabilities(&self) -> BackendCapabilities {
303        BackendCapabilities::mysql()
304    }
305
306    async fn fetch(&self, plan: &QueryPlan) -> Result<QueryResult, OrmError> {
307        let compiled = compile(plan, &MySql)?;
308        with_tx!(self.slot, conn => fetch_rows(conn, &compiled.sql, compiled.params).await)
309    }
310
311    async fn execute(&self, plan: &WritePlan) -> Result<ExecResult, OrmError> {
312        if plan.returning().is_empty() {
313            let compiled = compile_write(plan, &MySql)?;
314            let affected = with_tx!(self.slot, conn => execute_rows(conn, &compiled.sql, compiled.params).await)?;
315            return Ok(affected_result(affected));
316        }
317        let prepared = prepare(plan)?;
318        with_tx!(self.slot, conn => {
319            let info = self.tables.table(&mut *conn, table_of(plan)).await?;
320            write_returning(conn, &info, plan, prepared).await
321        })
322    }
323
324    async fn fetch_raw(&self, sql: &str, params: Vec<Value>) -> Result<QueryResult, OrmError> {
325        with_tx!(self.slot, conn => fetch_rows(conn, sql, params).await)
326    }
327
328    async fn execute_raw(&self, sql: &str, params: Vec<Value>) -> Result<u64, OrmError> {
329        with_tx!(self.slot, conn => execute_rows(conn, sql, params).await)
330    }
331
332    async fn execute_script(&self, sql: &str) -> Result<(), OrmError> {
333        let result =
334            with_tx!(self.slot, conn => script_result(sqlx::Executor::execute(conn, sql).await));
335        self.tables.clear();
336        result
337    }
338}
339
340#[async_trait]
341impl Transaction for MySqlTransaction {
342    async fn commit(&self) -> Result<(), OrmError> {
343        self.slot.commit().await
344    }
345
346    async fn rollback(&self) -> Result<(), OrmError> {
347        self.slot.rollback().await
348    }
349}
350
351/// Pooled connection held for a migration lock (no SQL transaction).
352///
353/// Like the Postgres held connection, commit and rollback close the
354/// connection instead of returning it to the pool so an unreleased
355/// `GET_LOCK` is never handed to another caller.
356struct MySqlHeld {
357    slot: ConnSlot<sqlx::MySql>,
358    tables: Arc<TableCache>,
359}
360
361#[async_trait]
362impl Executor for MySqlHeld {
363    fn capabilities(&self) -> BackendCapabilities {
364        BackendCapabilities::mysql()
365    }
366
367    async fn fetch(&self, plan: &QueryPlan) -> Result<QueryResult, OrmError> {
368        let compiled = compile(plan, &MySql)?;
369        with_conn!(self.slot, conn => fetch_rows(conn, &compiled.sql, compiled.params).await)
370    }
371
372    async fn execute(&self, plan: &WritePlan) -> Result<ExecResult, OrmError> {
373        if plan.returning().is_empty() {
374            let compiled = compile_write(plan, &MySql)?;
375            let affected = with_conn!(self.slot, conn => execute_rows(conn, &compiled.sql, compiled.params).await)?;
376            return Ok(affected_result(affected));
377        }
378        let prepared = prepare(plan)?;
379        with_conn!(self.slot, conn => {
380            let info = self.tables.table(&mut *conn, table_of(plan)).await?;
381            write_returning(conn, &info, plan, prepared).await
382        })
383    }
384
385    async fn fetch_raw(&self, sql: &str, params: Vec<Value>) -> Result<QueryResult, OrmError> {
386        with_conn!(self.slot, conn => fetch_rows(conn, sql, params).await)
387    }
388
389    async fn execute_raw(&self, sql: &str, params: Vec<Value>) -> Result<u64, OrmError> {
390        with_conn!(self.slot, conn => execute_rows(conn, sql, params).await)
391    }
392
393    async fn execute_script(&self, sql: &str) -> Result<(), OrmError> {
394        let result =
395            with_conn!(self.slot, conn => script_result(sqlx::Executor::execute(conn, sql).await));
396        self.tables.clear();
397        result
398    }
399}
400
401#[async_trait]
402impl Transaction for MySqlHeld {
403    async fn commit(&self) -> Result<(), OrmError> {
404        let mut conn = self.slot.take().await?;
405        conn.close_on_drop();
406        drop(conn);
407        Ok(())
408    }
409
410    async fn rollback(&self) -> Result<(), OrmError> {
411        let mut conn = self.slot.take().await?;
412        conn.close_on_drop();
413        drop(conn);
414        Ok(())
415    }
416}
417
418fn db_error(e: sqlx::Error) -> OrmError {
419    map_error(e).into()
420}
421
422fn script_result(done: Result<MySqlQueryResult, sqlx::Error>) -> Result<(), OrmError> {
423    done.map(|_| ()).map_err(db_error)
424}
425
426// ---- RETURNING emulation ----------------------------------------------------
siderite_backends/mysql.rs

siderite_backends/
mysql.rs
