Update Sqlite driver

Author: joeydewaal

sqlx-sqlite/src/any.rs

@@ -10,6 +10,7 @@ use sqlx_core::any::{
     Any, AnyArguments, AnyColumn, AnyConnectOptions, AnyConnectionBackend, AnyQueryResult, AnyRow,
     AnyStatement, AnyTypeInfo, AnyTypeInfoKind, AnyValueKind,
 };
+use sqlx_core::sql_str::SqlStr;
 
 use crate::type_info::DataType;
 use sqlx_core::connection::{ConnectOptions, Connection};
@@ -75,7 +76,7 @@ impl AnyConnectionBackend for SqliteConnection {
 
     fn fetch_many<'q>(
         &'q mut self,
-        query: &'q str,
+        query: SqlStr,
         persistent: bool,
         arguments: Option<AnyArguments<'q>>,
     ) -> BoxStream<'q, sqlx_core::Result<Either<AnyQueryResult, AnyRow>>> {
@@ -98,7 +99,7 @@ impl AnyConnectionBackend for SqliteConnection {
 
     fn fetch_optional<'q>(
         &'q mut self,
-        query: &'q str,
+        query: SqlStr,
         persistent: bool,
         arguments: Option<AnyArguments<'q>>,
     ) -> BoxFuture<'q, sqlx_core::Result<Option<AnyRow>>> {
@@ -123,16 +124,16 @@ impl AnyConnectionBackend for SqliteConnection {
 
     fn prepare_with<'c, 'q: 'c>(
         &'c mut self,
-        sql: &'q str,
+        sql: SqlStr,
         _parameters: &[AnyTypeInfo],
     ) -> BoxFuture<'c, sqlx_core::Result<AnyStatement>> {
         Box::pin(async move {
-            let statement = Executor::prepare_with(self, sql, &[]).await?;
+            let statement = Executor::prepare_with(self, sql.clone(), &[]).await?;
             AnyStatement::try_from_statement(sql, &statement, statement.column_names.clone())
         })
     }
 
-    fn describe<'q>(&'q mut self, sql: &'q str) -> BoxFuture<'q, sqlx_core::Result<Describe<Any>>> {
+    fn describe<'q>(&'q mut self, sql: SqlStr) -> BoxFuture<'q, sqlx_core::Result<Describe<Any>>> {
         Box::pin(async move { Executor::describe(self, sql).await?.try_into_any() })
     }
 }

sqlx-sqlite/src/connection/describe.rs

@@ -5,14 +5,18 @@ use crate::error::Error;
 use crate::statement::VirtualStatement;
 use crate::type_info::DataType;
 use crate::{Sqlite, SqliteColumn};
+use sqlx_core::sql_str::SqlStr;
 use sqlx_core::Either;
 use std::convert::identity;
 
-pub(crate) fn describe(conn: &mut ConnectionState, query: &str) -> Result<Describe<Sqlite>, Error> {
+pub(crate) fn describe(
+    conn: &mut ConnectionState,
+    query: SqlStr,
+) -> Result<Describe<Sqlite>, Error> {
     // describing a statement from SQLite can be involved
     // each SQLx statement is comprised of multiple SQL statements
 
-    let mut statement = VirtualStatement::new(query, false)?;
+    let mut statement = VirtualStatement::new(query.as_str(), false)?;
 
     let mut columns = Vec::new();
     let mut nullable = Vec::new();

sqlx-sqlite/src/connection/execute.rs

@@ -3,12 +3,13 @@ use crate::error::Error;
 use crate::logger::QueryLogger;
 use crate::statement::{StatementHandle, VirtualStatement};
 use crate::{SqliteArguments, SqliteQueryResult, SqliteRow};
+use sqlx_core::sql_str::SqlSafeStr;
 use sqlx_core::Either;
 
 pub struct ExecuteIter<'a> {
     handle: &'a mut ConnectionHandle,
     statement: &'a mut VirtualStatement,
-    logger: QueryLogger<'a>,
+    logger: QueryLogger,
     args: Option<SqliteArguments<'a>>,
 
     /// since a `VirtualStatement` can encompass multiple actual statements,
@@ -20,12 +21,13 @@ pub struct ExecuteIter<'a> {
 
 pub(crate) fn iter<'a>(
     conn: &'a mut ConnectionState,
-    query: &'a str,
+    query: impl SqlSafeStr,
     args: Option<SqliteArguments<'a>>,
     persistent: bool,
 ) -> Result<ExecuteIter<'a>, Error> {
+    let query = query.into_sql_str();
     // fetch the cached statement or allocate a new one
-    let statement = conn.statements.get(query, persistent)?;
+    let statement = conn.statements.get(query.as_str(), persistent)?;
 
     let logger = QueryLogger::new(query, conn.log_settings.clone());
 

sqlx-sqlite/src/connection/executor.rs

@@ -7,7 +7,7 @@ use futures_util::{stream, FutureExt, StreamExt, TryFutureExt, TryStreamExt};
 use sqlx_core::describe::Describe;
 use sqlx_core::error::Error;
 use sqlx_core::executor::{Execute, Executor};
-use sqlx_core::sql_str::{AssertSqlSafe, SqlSafeStr};
+use sqlx_core::sql_str::SqlSafeStr;
 use sqlx_core::Either;
 use std::{future, pin::pin};
 
@@ -24,12 +24,12 @@ impl<'c> Executor<'c> for &'c mut SqliteConnection {
         'q: 'e,
         E: 'q,
     {
-        let sql = query.sql();
         let arguments = match query.take_arguments().map_err(Error::Encode) {
             Ok(arguments) => arguments,
             Err(error) => return stream::once(future::ready(Err(error))).boxed(),
         };
         let persistent = query.persistent() && arguments.is_some();
+        let sql = query.sql();
 
         Box::pin(
             self.worker
@@ -49,14 +49,14 @@ impl<'c> Executor<'c> for &'c mut SqliteConnection {
         'q: 'e,
         E: 'q,
     {
-        let sql = query.sql();
         let arguments = match query.take_arguments().map_err(Error::Encode) {
             Ok(arguments) => arguments,
             Err(error) => return future::ready(Err(error)).boxed(),
         };
         let persistent = query.persistent() && arguments.is_some();
 
         Box::pin(async move {
+            let sql = query.sql();
             let mut stream = pin!(self
                 .worker
                 .execute(sql, arguments, self.row_channel_size, persistent, Some(1))
@@ -73,29 +73,28 @@ impl<'c> Executor<'c> for &'c mut SqliteConnection {
         })
     }
 
-    fn prepare_with<'e, 'q: 'e>(
+    fn prepare_with<'e>(
         self,
-        sql: &'q str,
+        sql: impl SqlSafeStr,
         _parameters: &[SqliteTypeInfo],
     ) -> BoxFuture<'e, Result<SqliteStatement, Error>>
     where
         'c: 'e,
     {
+        let sql = sql.into_sql_str();
         Box::pin(async move {
             let statement = self.worker.prepare(sql).await?;
 
-            Ok(SqliteStatement {
-                sql: AssertSqlSafe(sql).into_sql_str(),
-                ..statement
-            })
+            Ok(statement)
         })
     }
 
     #[doc(hidden)]
-    fn describe<'e, 'q: 'e>(self, sql: &'q str) -> BoxFuture<'e, Result<Describe<Sqlite>, Error>>
+    fn describe<'e>(self, sql: impl SqlSafeStr) -> BoxFuture<'e, Result<Describe<Sqlite>, Error>>
     where
         'c: 'e,
     {
-        Box::pin(self.worker.describe(sql))
+        let sql = sql.into_sql_str();
+        Box::pin(async move { self.worker.describe(sql).await })
     }
 }

sqlx-sqlite/src/connection/explain.rs

@@ -12,6 +12,7 @@ use crate::from_row::FromRow;
 use crate::logger::{BranchParent, BranchResult, DebugDiff};
 use crate::type_info::DataType;
 use crate::SqliteTypeInfo;
+use sqlx_core::sql_str::AssertSqlSafe;
 use sqlx_core::{hash_map, HashMap};
 use std::fmt::Debug;
 use std::str::from_utf8;
@@ -567,7 +568,7 @@ pub(super) fn explain(
 ) -> Result<(Vec<SqliteTypeInfo>, Vec<Option<bool>>), Error> {
     let root_block_cols = root_block_columns(conn)?;
     let program: Vec<(i64, String, i64, i64, i64, Vec<u8>)> =
-        execute::iter(conn, &format!("EXPLAIN {query}"), None, false)?
+        execute::iter(conn, AssertSqlSafe(format!("EXPLAIN {query}")), None, false)?
             .filter_map(|res| res.map(|either| either.right()).transpose())
             .map(|row| FromRow::from_row(&row?))
             .collect::<Result<Vec<_>, Error>>()?;

sqlx-sqlite/src/connection/mod.rs

@@ -22,6 +22,7 @@ use sqlx_core::common::StatementCache;
 pub(crate) use sqlx_core::connection::*;
 use sqlx_core::error::Error;
 use sqlx_core::executor::Executor;
+use sqlx_core::sql_str::AssertSqlSafe;
 use sqlx_core::transaction::Transaction;
 
 use crate::connection::establish::EstablishParams;
@@ -226,7 +227,7 @@ impl Connection for SqliteConnection {
                     write!(pragma_string, "PRAGMA analysis_limit = {limit}; ").ok();
                 }
                 pragma_string.push_str("PRAGMA optimize;");
-                self.execute(&*pragma_string).await?;
+                self.execute(AssertSqlSafe(pragma_string)).await?;
             }
             let shutdown = self.worker.shutdown();
             // Drop the statement worker, which should

sqlx-sqlite/src/connection/worker.rs

@@ -5,7 +5,7 @@ use std::thread;
 
 use futures_channel::oneshot;
 use futures_intrusive::sync::{Mutex, MutexGuard};
-use sqlx_core::sql_str::{AssertSqlSafe, SqlSafeStr};
+use sqlx_core::sql_str::SqlStr;
 use tracing::span::Span;
 
 use sqlx_core::describe::Describe;
@@ -42,15 +42,15 @@ pub(crate) struct WorkerSharedState {
 
 enum Command {
     Prepare {
-        query: Box<str>,
+        query: SqlStr,
         tx: oneshot::Sender<Result<SqliteStatement, Error>>,
     },
     Describe {
-        query: Box<str>,
+        query: SqlStr,
         tx: oneshot::Sender<Result<Describe<Sqlite>, Error>>,
     },
     Execute {
-        query: Box<str>,
+        query: SqlStr,
         arguments: Option<SqliteArguments<'static>>,
         persistent: bool,
         tx: flume::Sender<Result<Either<SqliteQueryResult, SqliteRow>, Error>>,
@@ -132,7 +132,7 @@ impl ConnectionWorker {
                     let _guard = span.enter();
                     match cmd {
                         Command::Prepare { query, tx } => {
-                            tx.send(prepare(&mut conn, &query).map(|prepared| {
+                            tx.send(prepare(&mut conn, query).map(|prepared| {
                                 update_cached_statements_size(
                                     &conn,
                                     &shared.cached_statements_size,
@@ -142,7 +142,7 @@ impl ConnectionWorker {
                             .ok();
                         }
                         Command::Describe { query, tx } => {
-                            tx.send(describe(&mut conn, &query)).ok();
+                            tx.send(describe(&mut conn, query)).ok();
                         }
                         Command::Execute {
                             query,
@@ -151,7 +151,7 @@ impl ConnectionWorker {
                             tx,
                             limit
                         } => {
-                            let iter = match execute::iter(&mut conn, &query, arguments, persistent)
+                            let iter = match execute::iter(&mut conn, query, arguments, persistent)
                             {
                                 Ok(iter) => iter,
                                 Err(e) => {
@@ -198,7 +198,7 @@ impl ConnectionWorker {
                             let depth = conn.transaction_depth;
                             let res =
                                 conn.handle
-                                    .exec(begin_ansi_transaction_sql(depth))
+                                    .exec(begin_ansi_transaction_sql(depth).as_str())
                                     .map(|_| {
                                         conn.transaction_depth += 1;
                                     });
@@ -211,7 +211,7 @@ impl ConnectionWorker {
                                 // immediately otherwise it would remain started forever.
                                 if let Err(error) = conn
                                     .handle
-                                    .exec(rollback_ansi_transaction_sql(depth + 1))
+                                    .exec(rollback_ansi_transaction_sql(depth + 1).as_str())
                                     .map(|_| {
                                         conn.transaction_depth -= 1;
                                     })
@@ -229,7 +229,7 @@ impl ConnectionWorker {
 
                             let res = if depth > 0 {
                                 conn.handle
-                                    .exec(commit_ansi_transaction_sql(depth))
+                                    .exec(commit_ansi_transaction_sql(depth).as_str())
                                     .map(|_| {
                                         conn.transaction_depth -= 1;
                                     })
@@ -255,7 +255,7 @@ impl ConnectionWorker {
 
                             let res = if depth > 0 {
                                 conn.handle
-                                    .exec(rollback_ansi_transaction_sql(depth))
+                                    .exec(rollback_ansi_transaction_sql(depth).as_str())
                                     .map(|_| {
                                         conn.transaction_depth -= 1;
                                     })
@@ -308,25 +308,19 @@ impl ConnectionWorker {
         establish_rx.await.map_err(|_| Error::WorkerCrashed)?
     }
 
-    pub(crate) async fn prepare(&mut self, query: &str) -> Result<SqliteStatement, Error> {
-        self.oneshot_cmd(|tx| Command::Prepare {
-            query: query.into(),
-            tx,
-        })
-        .await?
+    pub(crate) async fn prepare(&mut self, query: SqlStr) -> Result<SqliteStatement, Error> {
+        self.oneshot_cmd(|tx| Command::Prepare { query, tx })
+            .await?
     }
 
-    pub(crate) async fn describe(&mut self, query: &str) -> Result<Describe<Sqlite>, Error> {
-        self.oneshot_cmd(|tx| Command::Describe {
-            query: query.into(),
-            tx,
-        })
-        .await?
+    pub(crate) async fn describe(&mut self, query: SqlStr) -> Result<Describe<Sqlite>, Error> {
+        self.oneshot_cmd(|tx| Command::Describe { query, tx })
+            .await?
     }
 
     pub(crate) async fn execute(
         &mut self,
-        query: &str,
+        query: SqlStr,
         args: Option<SqliteArguments<'_>>,
         chan_size: usize,
         persistent: bool,
@@ -337,7 +331,7 @@ impl ConnectionWorker {
         self.command_tx
             .send_async((
                 Command::Execute {
-                    query: query.into(),
+                    query,
                     arguments: args.map(SqliteArguments::into_static),
                     persistent,
                     tx,
@@ -465,9 +459,9 @@ impl ConnectionWorker {
     }
 }
 
-fn prepare(conn: &mut ConnectionState, query: &str) -> Result<SqliteStatement, Error> {
+fn prepare(conn: &mut ConnectionState, query: SqlStr) -> Result<SqliteStatement, Error> {
     // prepare statement object (or checkout from cache)
-    let statement = conn.statements.get(query, true)?;
+    let statement = conn.statements.get(query.as_str(), true)?;
 
     let mut parameters = 0;
     let mut columns = None;
@@ -484,7 +478,7 @@ fn prepare(conn: &mut ConnectionState, query: &str) -> Result<SqliteStatement, E
     }
 
     Ok(SqliteStatement {
-        sql: AssertSqlSafe(query).into_sql_str(),
+        sql: query,
         columns: columns.unwrap_or_default(),
         column_names: column_names.unwrap_or_default(),
         parameters,

sqlx-sqlite/src/lib.rs

@@ -57,6 +57,7 @@ pub use options::{
 };
 pub use query_result::SqliteQueryResult;
 pub use row::SqliteRow;
+use sqlx_core::sql_str::{AssertSqlSafe, SqlSafeStr};
 pub use statement::SqliteStatement;
 pub use transaction::SqliteTransactionManager;
 pub use type_info::SqliteTypeInfo;
@@ -132,9 +133,10 @@ pub fn describe_blocking(query: &str, database_url: &str) -> Result<Describe<Sql
     let mut conn = params.establish()?;
 
     // Execute any ancillary `PRAGMA`s
-    connection::execute::iter(&mut conn, &opts.pragma_string(), None, false)?.finish()?;
+    connection::execute::iter(&mut conn, AssertSqlSafe(opts.pragma_string()), None, false)?
+        .finish()?;
 
-    connection::describe::describe(&mut conn, query)
+    connection::describe::describe(&mut conn, AssertSqlSafe(query.to_string()).into_sql_str())
 
     // SQLite database is closed immediately when `conn` is dropped
 }