siderite_backends/
sqlite.rs

1//! SQLite backend built on SQLx.
2//!
3//! Types without a native SQLite storage class are bound in their canonical
4//! text form ([`canonical_text`]), so they compare and sort correctly and
5//! decode through [`DbType::from_value`](siderite_orm::DbType::from_value).
6
7use crate::shared::{TxSlot, affected_result, map_error, returning_result, with_tx};
8use crate::sql::{CompiledQuery, Sqlite, compile, compile_write};
9use async_trait::async_trait;
10use siderite_orm::types::canonical_text;
11use siderite_orm::{
12    Backend, BackendCapabilities, BackendError, ExecResult, Executor, IsolationLevel, OrmError,
13    QueryError, QueryPlan, QueryResult, Row, Transaction, Value, WritePlan,
14};
15use sqlx::pool::PoolConnection;
16use sqlx::sqlite::{
17    SqliteArguments, SqliteConnectOptions, SqlitePool, SqlitePoolOptions, SqliteRow,
18    SqliteTransactionManager,
19};
20use sqlx::{Column as _, Row as _, TransactionManager, TypeInfo as _, ValueRef as _};
21use std::collections::HashSet;
22use std::time::Duration;
23
24mod gate;
25mod group;
26
27use gate::{WriteGate, WritePermit};
28use group::Committer;
29pub use group::GroupCommit;
30
31/// SQLite adapter executing compiled plans on a connection pool.
32///
33/// Writes from this backend and its clones (statements through
34/// [`execute`](Executor::execute), [`execute_raw`](Executor::execute_raw)
35/// and [`execute_script`](Executor::execute_script), transactions, and
36/// transactional schema changes) queue for one write slot instead of racing
37/// for SQLite's file lock, while reads keep using the whole pool. A write
38/// sent through [`fetch_raw`](Executor::fetch_raw) bypasses the queue.
39#[derive(Debug, Clone)]
40pub struct SqliteBackend {
41    pool: SqlitePool,
42    gate: WriteGate,
43    committer: Option<Committer>,
44}
45
46impl SqliteBackend {
47    /// Connect to `url` (e.g. `sqlite::memory:` or `sqlite://app.db?mode=rwc`).
48    ///
49    /// SQLx enables foreign-key enforcement on every SQLite connection.
50    ///
51    /// # Errors
52    /// Returns [`BackendError::Connection`] if the pool cannot be created.
53    pub async fn connect(url: &str) -> Result<Self, BackendError> {
54        // In-memory databases are per-connection, so a single connection keeps
55        // state consistent for `sqlite::memory:`.
56        let max = if url.contains(":memory:") { 1 } else { 10 };
57        let pool = SqlitePoolOptions::new()
58            .max_connections(max)
59            .connect(url)
60            .await
61            .map_err(|e| BackendError::Connection(e.to_string()))?;
62        Ok(Self::from_pool(pool))
63    }
64
65    /// Connect with explicit connection and pool options, for example to opt
66    /// in to write-ahead logging:
67    ///
68    /// ```no_run
69    /// # use siderite_backends::sqlite::SqliteBackend;
70    /// # use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
71    /// # async fn open() -> Result<(), Box<dyn std::error::Error>> {
72    /// let options: SqliteConnectOptions = "sqlite://app.db?mode=rwc".parse()?;
73    /// let db = SqliteBackend::connect_with(
74    ///     options.journal_mode(SqliteJournalMode::Wal),
75    ///     SqlitePoolOptions::new().max_connections(10),
76    /// )
77    /// .await?;
78    /// # Ok(())
79    /// # }
80    /// ```
81    ///
82    /// [`connect`](Self::connect) leaves the journal mode and `synchronous`
83    /// at the SQLite defaults (a rollback journal, synced on every commit).
84    /// WAL is recorded in the database file and does not work on network
85    /// filesystems; `synchronous = NORMAL` with WAL can lose the most recent
86    /// commits on power loss. An in-memory database needs a pool of one
87    /// connection.
88    ///
89    /// # Errors
90    /// Returns [`BackendError::Connection`] if the pool cannot be created.
91    pub async fn connect_with(
92        options: SqliteConnectOptions,
93        pool: SqlitePoolOptions,
94    ) -> Result<Self, BackendError> {
95        let pool = pool
96            .connect_with(options)
97            .await
98            .map_err(|e| BackendError::Connection(e.to_string()))?;
99        Ok(Self::from_pool(pool))
100    }
101
102    /// Wrap an existing pool.
103    pub fn from_pool(pool: SqlitePool) -> Self {
104        Self {
105            pool,
106            gate: WriteGate::new(gate::DEFAULT_WAIT),
107            committer: None,
108        }
109    }
110
111    /// Opt in to group commit: concurrent [`execute`](Executor::execute)
112    /// calls outside a transaction share one transaction and one commit
113    /// sync, each isolated by a savepoint and acknowledged only after the
114    /// commit. See [`GroupCommit`] for what changes compared with one
115    /// transaction per write. Raw statements, scripts and transactions are
116    /// unaffected.
117    ///
118    /// Call [`with_write_timeout`](Self::with_write_timeout) first; the
119    /// committer keeps the timeout it started with.
120    ///
121    /// # Panics
122    /// Outside a Tokio runtime, which runs the committer task.
123    #[must_use]
124    pub fn group_commit(mut self, config: GroupCommit) -> Self {
125        let committer = Committer::spawn(self.pool.clone(), self.gate.clone(), config);
126        self.committer = Some(committer);
127        self
128    }
129
130    /// How long a write waits for the write slot before failing with
131    /// "database is locked"; five seconds by default, like SQLx's busy
132    /// timeout. Match it when configuring a different busy timeout.
133    ///
134    /// The slot is shared with existing clones; only this handle's wait
135    /// changes.
136    #[must_use]
137    pub fn with_write_timeout(mut self, wait: Duration) -> Self {
138        self.gate = self.gate.with_wait(wait);
139        self
140    }
141}
142
143#[async_trait]
144impl Executor for SqliteBackend {
145    fn capabilities(&self) -> BackendCapabilities {
146        BackendCapabilities::sqlite()
147    }
148
149    async fn fetch(&self, plan: &QueryPlan) -> Result<QueryResult, OrmError> {
150        let compiled = compile(plan, &Sqlite)?;
151        fetch_rows(&self.pool, &compiled.sql, compiled.params).await
152    }
153
154    async fn execute(&self, plan: &WritePlan) -> Result<ExecResult, OrmError> {
155        let compiled = compile_write(plan, &Sqlite)?;
156        let returning = !plan.returning().is_empty();
157        if let Some(committer) = &self.committer {
158            return committer.submit(compiled, returning).await;
159        }
160        let _permit = self.gate.acquire().await?;
161        run_write(&self.pool, compiled, returning).await
162    }
163
164    async fn fetch_raw(&self, sql: &str, params: Vec<Value>) -> Result<QueryResult, OrmError> {
165        fetch_rows(&self.pool, sql, params).await
166    }
167
168    async fn execute_raw(&self, sql: &str, params: Vec<Value>) -> Result<u64, OrmError> {
169        let _permit = self.gate.acquire().await?;
170        execute_rows(&self.pool, sql, params).await
171    }
172
173    async fn execute_script(&self, sql: &str) -> Result<(), OrmError> {
174        let _permit = self.gate.acquire().await?;
175        script_result(sqlx::Executor::execute(&self.pool, sql).await)
176    }
177}
178
179#[async_trait]
180impl Backend for SqliteBackend {
181    fn read_parameter_count(&self, plan: &QueryPlan) -> Result<Option<usize>, OrmError> {
182        Ok(Some(compile(plan, &Sqlite)?.params.len()))
183    }
184
185    async fn begin(
186        &self,
187        isolation: Option<IsolationLevel>,
188    ) -> Result<Box<dyn Transaction>, OrmError> {
189        if let Some(level) = isolation {
190            self.capabilities()
191                .require(siderite_orm::Feature::Isolation(level))?;
192        }
193        // Held for the whole transaction: it may write at any point.
194        let permit = self.gate.acquire().await?;
195        let tx = self.pool.begin().await.map_err(map_error)?;
196        Ok(Box::new(SqliteTransaction {
197            slot: TxSlot::new(tx),
198            permit: Held::new(permit),
199        }))
200    }
201
202    async fn begin_schema(&self, transactional: bool) -> Result<Box<dyn Transaction>, OrmError> {
203        // `BEGIN IMMEDIATE` already excludes other writers. A non-transactional
204        // schema connection stays ungated: callers keep writing through the
205        // pool while it is open.
206        let permit = if transactional {
207            Some(self.gate.acquire().await?)
208        } else {
209            None
210        };
211        Ok(Box::new(
212            SqliteSchemaTransaction::open(&self.pool, transactional, permit).await?,
213        ))
214    }
215}
216
217/// The write slot of an open transaction, released once it finishes.
218///
219/// Dropping a transaction without commit releases the slot before SQLx's
220/// deferred rollback runs, so the next writer can briefly meet SQLite's busy
221/// handler; it does not fail because of it.
222struct Held(std::sync::Mutex<Option<WritePermit>>);
223
224impl Held {
225    fn new(permit: WritePermit) -> Self {
226        Self(std::sync::Mutex::new(Some(permit)))
227    }
228
229    fn none() -> Self {
230        Self(std::sync::Mutex::new(None))
231    }
232
233    fn release(&self) {
234        drop(self.0.lock().map(|mut held| held.take()));
235    }
236}
237
238/// An open SQLite transaction. Dropped without commit, it rolls back.
239struct SqliteTransaction {
240    slot: TxSlot<sqlx::Sqlite>,
241    permit: Held,
242}
243
244#[async_trait]
245impl Executor for SqliteTransaction {
246    fn capabilities(&self) -> BackendCapabilities {
247        BackendCapabilities::sqlite()
248    }
249
250    async fn fetch(&self, plan: &QueryPlan) -> Result<QueryResult, OrmError> {
251        let compiled = compile(plan, &Sqlite)?;
252        with_tx!(self.slot, conn => fetch_rows(conn, &compiled.sql, compiled.params).await)
253    }
254
255    async fn execute(&self, plan: &WritePlan) -> Result<ExecResult, OrmError> {
256        let compiled = compile_write(plan, &Sqlite)?;
257        let returning = !plan.returning().is_empty();
258        with_tx!(self.slot, conn => run_write(conn, compiled, returning).await)
259    }
260
261    async fn fetch_raw(&self, sql: &str, params: Vec<Value>) -> Result<QueryResult, OrmError> {
262        with_tx!(self.slot, conn => fetch_rows(conn, sql, params).await)
263    }
264
265    async fn execute_raw(&self, sql: &str, params: Vec<Value>) -> Result<u64, OrmError> {
266        with_tx!(self.slot, conn => execute_rows(conn, sql, params).await)
267    }
268
269    async fn execute_script(&self, sql: &str) -> Result<(), OrmError> {
270        with_tx!(self.slot, conn => script_result(sqlx::Executor::execute(conn, sql).await))
271    }
272}
273
274#[async_trait]
275impl Transaction for SqliteTransaction {
276    async fn commit(&self) -> Result<(), OrmError> {
277        let done = self.slot.commit().await;
278        self.permit.release();
279        done
280    }
281
282    async fn rollback(&self) -> Result<(), OrmError> {
283        let done = self.slot.rollback().await;
284        self.permit.release();
285        done
286    }
287}
288
289macro_rules! with_schema_conn {
290    ($this:expr, $conn:ident => $body:expr) => {{
291        let mut guard = $this.conn.lock().await;
292        let conn = guard
293            .as_mut()
294            .ok_or(::siderite_orm::QueryError::TransactionClosed)?;
295        let $conn = &mut **conn;
296        $body
297    }};
298}
299
300/// Dedicated connection with `PRAGMA foreign_keys` off around a schema change.
301///
302/// SQLite ignores `PRAGMA foreign_keys` inside a transaction, so the pragma
303/// is applied *before* `BEGIN` and restored after commit or rollback.
304///
305/// The pre-commit `foreign_key_check` is scoped to violations this schema
306/// change introduced: the rows at open are the baseline, and commit fails
307/// only on rows that are not in it. Pre-existing violations in unrelated
308/// tables must not block every future migration, and comparing rows (not
309/// counts) means a change that fixes one violation while adding another
310/// still fails. When foreign keys were already off (`restore_fk == 0`) the
311/// check is skipped entirely.
312struct SqliteSchemaTransaction {
313    conn: tokio::sync::Mutex<Option<PoolConnection<sqlx::Sqlite>>>,
314    restore_fk: i64,
315    in_txn: bool,
316    fk_baseline: Option<HashSet<FkViolation>>,
317    permit: Held,
318}
319
320/// One `PRAGMA foreign_key_check` row: the offending child row and the parent
321/// it points at. Identifies a violation independently of when it appeared.
322#[derive(Debug, Clone, PartialEq, Eq, Hash)]
323struct FkViolation {
324    table: String,
325    rowid: i64,
326    parent: String,
327}
328
329impl FkViolation {
330    /// A missing or wrong-typed column becomes a placeholder, so an unreadable
331    /// row still counts as a violation instead of being silently dropped.
332    fn from_row(row: &SqliteRow) -> Self {
333        let text = |key: &str| {
334            row.try_get::<String, _>(key)
335                .unwrap_or_else(|_| format!("<unreadable {key}>"))
336        };
337        let rowid = row.try_get("rowid").unwrap_or(-1);
338        Self {
339            table: text("table"),
340            rowid,
341            parent: text("parent"),
342        }
343    }
344
345    fn location(&self) -> String {
346        format!(
347            "{} row {} (missing parent {})",
348            self.table, self.rowid, self.parent
349        )
350    }
351}
352
353async fn fk_violations(
354    conn: &mut PoolConnection<sqlx::Sqlite>,
355) -> Result<Vec<SqliteRow>, OrmError> {
356    Ok(sqlx::query("PRAGMA foreign_key_check")
357        .fetch_all(&mut **conn)
358        .await
359        .map_err(map_error)?)
360}
361
362impl SqliteSchemaTransaction {
363    async fn open(
364        pool: &SqlitePool,
365        transactional: bool,
366        permit: Option<WritePermit>,
367    ) -> Result<Self, OrmError> {
368        let mut conn = pool.acquire().await.map_err(map_error)?;
369        let restore_fk: i64 = sqlx::query_scalar("PRAGMA foreign_keys")
370            .fetch_one(&mut *conn)
371            .await
372            .map_err(map_error)?;
373        // Baseline for the pre-commit check: only *new* violations fail the
374        // migration. Skipped when FKs were already off (nothing to enforce).
375        let fk_baseline: Option<HashSet<FkViolation>> = if restore_fk == 0 {
376            None
377        } else {
378            Some(
379                fk_violations(&mut conn)
380                    .await?
381                    .iter()
382                    .map(FkViolation::from_row)
383                    .collect(),
384            )
385        };
386        set_foreign_keys(&mut conn, 0).await?;
387        if transactional
388            && let Err(err) =
389                SqliteTransactionManager::begin(&mut *conn, Some("BEGIN IMMEDIATE".into())).await
390        {
391            drop(set_foreign_keys(&mut conn, restore_fk).await);
392            return Err(map_error(err).into());
393        }
394        Ok(Self {
395            conn: tokio::sync::Mutex::new(Some(conn)),
396            restore_fk,
397            in_txn: transactional,
398            fk_baseline,
399            permit: permit.map_or_else(Held::none, Held::new),
400        })
401    }
402
403    async fn take(&self) -> Result<PoolConnection<sqlx::Sqlite>, OrmError> {
404        self.conn
405            .lock()
406            .await
407            .take()
408            .ok_or_else(|| QueryError::TransactionClosed.into())
409    }
410}
411
412impl Drop for SqliteSchemaTransaction {
413    fn drop(&mut self) {
414        if let Some(mut conn) = self.conn.get_mut().take() {
415            if self.in_txn {
416                SqliteTransactionManager::start_rollback(&mut *conn);
417            }
418            // Do not return a connection whose foreign_keys pragma we changed.
419            conn.close_on_drop();
420        }
421    }
422}
423
424#[async_trait]
425impl Executor for SqliteSchemaTransaction {
426    fn capabilities(&self) -> BackendCapabilities {
427        BackendCapabilities::sqlite()
428    }
429
430    async fn fetch(&self, plan: &QueryPlan) -> Result<QueryResult, OrmError> {
431        let compiled = compile(plan, &Sqlite)?;
432        with_schema_conn!(self, conn => fetch_rows(conn, &compiled.sql, compiled.params).await)
433    }
434
435    async fn execute(&self, plan: &WritePlan) -> Result<ExecResult, OrmError> {
436        let compiled = compile_write(plan, &Sqlite)?;
437        let returning = !plan.returning().is_empty();
438        with_schema_conn!(self, conn => run_write(conn, compiled, returning).await)
439    }
440
441    async fn fetch_raw(&self, sql: &str, params: Vec<Value>) -> Result<QueryResult, OrmError> {
442        with_schema_conn!(self, conn => fetch_rows(conn, sql, params).await)
443    }
444
445    async fn execute_raw(&self, sql: &str, params: Vec<Value>) -> Result<u64, OrmError> {
446        with_schema_conn!(self, conn => execute_rows(conn, sql, params).await)
447    }
448
449    async fn execute_script(&self, sql: &str) -> Result<(), OrmError> {
450        with_schema_conn!(self, conn => script_result(sqlx::Executor::execute(conn, sql).await))
451    }
452}
453
454#[async_trait]
455impl Transaction for SqliteSchemaTransaction {
456    async fn commit(&self) -> Result<(), OrmError> {
457        let done = self.finish_commit().await;
458        self.permit.release();
459        done
460    }
461
462    async fn rollback(&self) -> Result<(), OrmError> {
463        let done = self.finish_rollback().await;
464        self.permit.release();
465        done
466    }
467}
468
469impl SqliteSchemaTransaction {
470    async fn finish_commit(&self) -> Result<(), OrmError> {
471        let mut conn = self.take().await?;
472        if let Some(baseline) = &self.fk_baseline {
473            // Row-level diff, not a count: a change that repairs one old
474            // violation while adding another keeps the count equal, and would
475            // wrongly pass.
476            let mut seen = HashSet::new();
477            let mut fresh = Vec::new();
478            for row in fk_violations(&mut conn).await? {
479                let violation = FkViolation::from_row(&row);
480                if !baseline.contains(&violation) && seen.insert(violation.clone()) {
481                    fresh.push(violation);
482                }
483            }
484            if !fresh.is_empty() {
485                if self.in_txn {
486                    drop(SqliteTransactionManager::rollback(&mut *conn).await);
487                }
488                drop(set_foreign_keys(&mut conn, self.restore_fk).await);
489                // Without a transaction the statements are already
490                // committed; say so rather than implying nothing happened.
491                let outcome = if self.in_txn {
492                    "rolled back"
493                } else {
494                    "already committed; fix the rows by hand"
495                };
496                return Err(BackendError::Constraint(format!(
497                    "SQLite foreign_key_check found {} new violation(s) after schema change ({outcome}): {}",
498                    fresh.len(),
499                    violation_summary(&fresh),
500                ))
501                .into());
502            }
503        }
504        if self.in_txn
505            && let Err(err) = SqliteTransactionManager::commit(&mut *conn).await
506        {
507            drop(set_foreign_keys(&mut conn, self.restore_fk).await);
508            return Err(map_error(err).into());
509        }
510        set_foreign_keys(&mut conn, self.restore_fk).await?;
511        Ok(())
512    }
513
514    async fn finish_rollback(&self) -> Result<(), OrmError> {
515        let mut conn = self.take().await?;
516        let rollback = if self.in_txn {
517            SqliteTransactionManager::rollback(&mut *conn)
518                .await
519                .map_err(map_error)
520        } else {
521            Ok(())
522        };
523        let restore = set_foreign_keys(&mut conn, self.restore_fk).await;
524        rollback?;
525        restore
526    }
527}
528
529/// One-line summary of new `PRAGMA foreign_key_check` violations.
530fn violation_summary(violations: &[FkViolation]) -> String {
531    let mut locations: Vec<String> = violations.iter().map(FkViolation::location).collect();
532    locations.sort();
533    locations.truncate(5);
534    if locations.is_empty() {
535        "see PRAGMA foreign_key_check".to_owned()
536    } else {
537        locations.join(", ")
538    }
539}
540
541async fn set_foreign_keys(
542    conn: &mut PoolConnection<sqlx::Sqlite>,
543    on: i64,
544) -> Result<(), OrmError> {
545    let sql = if on == 0 {
546        "PRAGMA foreign_keys = OFF"
547    } else {
548        "PRAGMA foreign_keys = ON"
549    };
550    sqlx::query(sql)
551        .execute(&mut **conn)
552        .await
553        .map(|_| ())
554        .map_err(|e| map_error(e).into())
555}
556
557fn bind_all(
558    sql: &str,
559    params: Vec<Value>,
560) -> sqlx::query::Query<'_, sqlx::Sqlite, SqliteArguments<'_>> {
561    params.into_iter().fold(sqlx::query(sql), |query, param| {
562        if let Some(text) = canonical_text(&param) {
563            return query.bind(text);
564        }
565        match param {
566            Value::Bool(b) => query.bind(b),
567            Value::Int(i) => query.bind(i),
568            Value::Float(f) => query.bind(f),
569            Value::Text(s) => query.bind(s),
570            Value::Bytes(b) => query.bind(b),
571            // `canonical_text` covered every other variant.
572            _ => query.bind(None::<i64>),
573        }
574    })
575}
576
577async fn fetch_rows<'c, E>(ex: E, sql: &str, params: Vec<Value>) -> Result<QueryResult, OrmError>
578where
579    E: sqlx::Executor<'c, Database = sqlx::Sqlite>,
580{
581    let rows = bind_all(sql, params)
582        .fetch_all(ex)
583        .await
584        .map_err(map_error)?;
585    let rows = rows.iter().map(decode_row).collect::<Result<_, _>>()?;
586    Ok(QueryResult { rows })
587}
588
589async fn execute_rows<'c, E>(ex: E, sql: &str, params: Vec<Value>) -> Result<u64, OrmError>
590where
591    E: sqlx::Executor<'c, Database = sqlx::Sqlite>,
592{
593    let done = bind_all(sql, params).execute(ex).await.map_err(map_error)?;
594    Ok(done.rows_affected())
595}
596
597async fn run_write<'c, E>(
598    ex: E,
599    compiled: CompiledQuery,
600    returning: bool,
601) -> Result<ExecResult, OrmError>
602where
603    E: sqlx::Executor<'c, Database = sqlx::Sqlite>,
604{
605    if returning {
606        let rows = fetch_rows(ex, &compiled.sql, compiled.params).await?.rows;
607        Ok(returning_result(rows))
608    } else {
609        Ok(affected_result(
610            execute_rows(ex, &compiled.sql, compiled.params).await?,
611        ))
612    }
613}
614
615fn script_result(
616    done: Result<sqlx::sqlite::SqliteQueryResult, sqlx::Error>,
617) -> Result<(), OrmError> {
618    done.map(|_| ()).map_err(|e| map_error(e).into())
619}
620
621fn decode_row(row: &SqliteRow) -> Result<Row, QueryError> {
622    row.columns()
623        .iter()
624        .enumerate()
625        .map(|(i, col)| {
626            let name = col.name().to_owned();
627            let decode_err = |e: sqlx::Error| QueryError::Decode {
628                column: name.clone(),
629                reason: e.to_string(),
630            };
631            let raw = row.try_get_raw(i).map_err(decode_err)?;
632            let value = if raw.is_null() {
633                Value::Null
634            } else {
635                match raw.type_info().name() {
636                    "INTEGER" | "BOOLEAN" => Value::Int(row.try_get(i).map_err(decode_err)?),
637                    "REAL" => Value::Float(row.try_get(i).map_err(decode_err)?),
638                    "BLOB" => Value::Bytes(row.try_get(i).map_err(decode_err)?),
639                    _ => Value::Text(row.try_get(i).map_err(decode_err)?),
640                }
641            };
642            Ok((name, value))
643        })
644        .collect::<Result<Vec<_>, _>>()
645        .map(Row::new)
646}
647
648#[cfg(test)]
649mod tests {
650    use super::*;
651    use siderite_orm::expr::Field;
652    use siderite_orm::{BackendCapabilityError, Expr, LockMode, OrderDirection};
653
654    struct User;
655    #[allow(non_upper_case_globals)]
656    impl User {
657        const name: Field<User, String> = Field::new("name");
658        const age: Field<User, i64> = Field::new("age");
659    }
660
661    async fn seeded() -> SqliteBackend {
662        let db = SqliteBackend::connect("sqlite::memory:").await.unwrap();
663        Executor::execute_script(&db,
664            "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL, age INTEGER);
665             INSERT INTO users (name, age) VALUES ('Alice', 30), ('bob', NULL), ('ALINA', 22), ('50%', 1);",
666        )
667        .await
668        .unwrap();
669        db
670    }
671
672    fn names(r: &QueryResult) -> Vec<String> {
673        r.rows
674            .iter()
675            .map(|row| match row.get("name") {
676                Some(Value::Text(s)) => s.clone(),
677                other => panic!("unexpected {other:?}"),
678            })
679            .collect()
680    }
681
682    #[tokio::test]
683    async fn executes_filters_ordering_and_limits() {
684        let db = seeded().await;
685        let plan = QueryPlan::from_table("users")
686            .filter(User::name.icontains("al"))
687            .order_by(User::age, OrderDirection::Asc);
688        assert_eq!(names(&db.fetch(&plan).await.unwrap()), ["ALINA", "Alice"]);
689
690        let cs = QueryPlan::from_table("users").filter(User::name.contains("Al"));
691        assert_eq!(names(&db.fetch(&cs).await.unwrap()), ["Alice"]);
692
693        let nulls = QueryPlan::from_table("users").filter(User::age.is_null());
694        assert_eq!(names(&db.fetch(&nulls).await.unwrap()), ["bob"]);
695
696        let wildcard = QueryPlan::from_table("users").filter(User::name.icontains("%"));
697        assert_eq!(names(&db.fetch(&wildcard).await.unwrap()), ["50%"]);
698
699        let page = QueryPlan::from_table("users")
700            .order_by(Expr::col("id"), OrderDirection::Asc)
701            .offset(3);
702        assert_eq!(names(&db.fetch(&page).await.unwrap()), ["50%"]);
703    }
704
705    #[tokio::test]
706    async fn decodes_nulls_and_integers() {
707        let db = seeded().await;
708        let r = db
709            .fetch(&QueryPlan::from_table("users").filter(User::name.eq("bob")))
710            .await
711            .unwrap();
712        assert_eq!(r.rows[0].get("age"), Some(&Value::Null));
713        assert_eq!(r.rows[0].get("id"), Some(&Value::Int(2)));
714    }
715
716    #[tokio::test]
717    async fn row_locking_is_rejected_before_io() {
718        let db = seeded().await;
719        let err = db
720            .fetch(&QueryPlan::from_table("users").lock(LockMode::ForUpdate))
721            .await;
722        assert!(matches!(
723            err,
724            Err(OrmError::Capability(
725                BackendCapabilityError::RowLockingUnsupported { .. }
726            ))
727        ));
728    }
729}
siderite_backends/sqlite.rs

siderite_backends/
sqlite.rs
