fix: respect isolation level parameter and defaults (#5056)

* fix(tests): add test for transaction isolation level on mssql

* fix: fix isolation level issue in mssql

* chore: address PR comments, extend the fix to PooledConnection

* chore: fix a clippy lint

* chore: feature gate several things to resolve warnings in WASM build

* chore: make rustfmt happy

* fix: update the query count for MSSQL now that it sets the isolation settings

* fix: add sqlite to feature gates

* fix: default to READ UNCOMMITTED in Quaint test

---------

Co-authored-by: Lucian Buzzo <[email protected]>

.github/workflows/test-quaint.yml

@@ -22,7 +22,7 @@ jobs:
       TEST_MYSQL8: "mysql://root:prisma@localhost:3307/prisma"
       TEST_MYSQL_MARIADB: "mysql://root:prisma@localhost:3308/prisma"
       TEST_PSQL: "postgres://postgres:prisma@localhost:5432/postgres"
-      TEST_MSSQL: "jdbc:sqlserver://localhost:1433;database=master;user=SA;password=<YourStrong@Passw0rd>;trustServerCertificate=true"
+      TEST_MSSQL: "jdbc:sqlserver://localhost:1433;database=master;user=SA;password=<YourStrong@Passw0rd>;trustServerCertificate=true;isolationLevel=READ UNCOMMITTED"
       TEST_CRDB: "postgresql://[email protected]:26259/postgres"
 
     steps:

quaint/src/connector/queryable.rs

@@ -112,6 +112,12 @@ pub trait TransactionCapable: Queryable {
     ) -> crate::Result<Box<dyn Transaction + 'a>>;
 }
 
+#[cfg(any(
+    feature = "sqlite-native",
+    feature = "mssql-native",
+    feature = "postgresql-native",
+    feature = "mysql-native"
+))]
 macro_rules! impl_default_TransactionCapable {
     ($t:ty) => {
         #[async_trait]
@@ -130,4 +136,10 @@ macro_rules! impl_default_TransactionCapable {
     };
 }
 
+#[cfg(any(
+    feature = "sqlite-native",
+    feature = "mssql-native",
+    feature = "postgresql-native",
+    feature = "mysql-native"
+))]
 pub(crate) use impl_default_TransactionCapable;

quaint/src/connector/transaction.rs

@@ -21,6 +21,12 @@ pub trait Transaction: Queryable {
     fn as_queryable(&self) -> &dyn Queryable;
 }
 
+#[cfg(any(
+    feature = "sqlite-native",
+    feature = "mssql-native",
+    feature = "postgresql-native",
+    feature = "mysql-native"
+))]
 pub(crate) struct TransactionOptions {
     /// The isolation level to use.
     pub(crate) isolation_level: Option<IsolationLevel>,
@@ -29,6 +35,21 @@ pub(crate) struct TransactionOptions {
     pub(crate) isolation_first: bool,
 }
 
+#[cfg(any(
+    feature = "sqlite-native",
+    feature = "mssql-native",
+    feature = "postgresql-native",
+    feature = "mysql-native"
+))]
+impl TransactionOptions {
+    pub fn new(isolation_level: Option<IsolationLevel>, isolation_first: bool) -> Self {
+        Self {
+            isolation_level,
+            isolation_first,
+        }
+    }
+}
+
 /// A default representation of an SQL database transaction. If not commited, a
 /// transaction will be rolled back by default when dropped.
 ///
@@ -40,6 +61,12 @@ pub struct DefaultTransaction<'a> {
 }
 
 impl<'a> DefaultTransaction<'a> {
+    #[cfg(any(
+        feature = "sqlite-native",
+        feature = "mssql-native",
+        feature = "postgresql-native",
+        feature = "mysql-native"
+    ))]
     pub(crate) async fn new(
         inner: &'a dyn Queryable,
         begin_stmt: &str,
@@ -196,11 +223,3 @@ impl FromStr for IsolationLevel {
         }
     }
 }
-impl TransactionOptions {
-    pub fn new(isolation_level: Option<IsolationLevel>, isolation_first: bool) -> Self {
-        Self {
-            isolation_level,
-            isolation_first,
-        }
-    }
-}

quaint/src/pooled/manager.rs

@@ -13,7 +13,7 @@ use crate::connector::MysqlUrl;
 use crate::connector::{MakeTlsConnectorManager, PostgresNativeUrl};
 use crate::{
     ast,
-    connector::{self, impl_default_TransactionCapable, IsolationLevel, Queryable, Transaction, TransactionCapable},
+    connector::{self, IsolationLevel, Queryable, Transaction, TransactionCapable},
     error::Error,
 };
 
@@ -23,7 +23,15 @@ pub struct PooledConnection {
     pub(crate) inner: MobcPooled<QuaintManager>,
 }
 
-impl_default_TransactionCapable!(PooledConnection);
+#[async_trait]
+impl TransactionCapable for PooledConnection {
+    async fn start_transaction<'a>(
+        &'a self,
+        isolation: Option<IsolationLevel>,
+    ) -> crate::Result<Box<dyn Transaction + 'a>> {
+        self.inner.start_transaction(isolation).await
+    }
+}
 
 #[async_trait]
 impl Queryable for PooledConnection {
@@ -104,7 +112,7 @@ pub enum QuaintManager {
 
 #[async_trait]
 impl Manager for QuaintManager {
-    type Connection = Box<dyn Queryable>;
+    type Connection = Box<dyn TransactionCapable>;
     type Error = Error;
 
     async fn connect(&self) -> crate::Result<Self::Connection> {

quaint/src/single.rs

@@ -2,7 +2,7 @@
 
 use crate::{
     ast,
-    connector::{self, impl_default_TransactionCapable, ConnectionInfo, IsolationLevel, Queryable, TransactionCapable},
+    connector::{self, ConnectionInfo, IsolationLevel, Queryable, TransactionCapable},
 };
 use async_trait::async_trait;
 use std::{fmt, sync::Arc};
@@ -16,7 +16,7 @@ use crate::connector::NativeConnectionInfo;
 /// The main entry point and an abstraction over a database connection.
 #[derive(Clone)]
 pub struct Quaint {
-    inner: Arc<dyn Queryable>,
+    inner: Arc<dyn TransactionCapable>,
     connection_info: Arc<ConnectionInfo>,
 }
 
@@ -26,7 +26,15 @@ impl fmt::Debug for Quaint {
     }
 }
 
-impl_default_TransactionCapable!(Quaint);
+#[async_trait]
+impl TransactionCapable for Quaint {
+    async fn start_transaction<'a>(
+        &'a self,
+        isolation: Option<IsolationLevel>,
+    ) -> crate::Result<Box<dyn connector::Transaction + 'a>> {
+        self.inner.start_transaction(isolation).await
+    }
+}
 
 impl Quaint {
     /// Create a new connection to the database. The connection string
@@ -137,28 +145,28 @@ impl Quaint {
                 let params = connector::SqliteParams::try_from(s)?;
                 let sqlite = connector::Sqlite::new(&params.file_path)?;
 
-                Arc::new(sqlite) as Arc<dyn Queryable>
+                Arc::new(sqlite) as Arc<dyn TransactionCapable>
             }
             #[cfg(feature = "mysql-native")]
             s if s.starts_with("mysql") => {
                 let url = connector::MysqlUrl::new(url::Url::parse(s)?)?;
                 let mysql = connector::Mysql::new(url).await?;
 
-                Arc::new(mysql) as Arc<dyn Queryable>
+                Arc::new(mysql) as Arc<dyn TransactionCapable>
             }
             #[cfg(feature = "postgresql-native")]
             s if s.starts_with("postgres") || s.starts_with("postgresql") => {
                 let url = connector::PostgresNativeUrl::new(url::Url::parse(s)?)?;
                 let tls_manager = connector::MakeTlsConnectorManager::new(url.clone());
                 let psql = connector::PostgreSql::new(url, &tls_manager).await?;
-                Arc::new(psql) as Arc<dyn Queryable>
+                Arc::new(psql) as Arc<dyn TransactionCapable>
             }
             #[cfg(feature = "mssql-native")]
             s if s.starts_with("jdbc:sqlserver") | s.starts_with("sqlserver") => {
                 let url = connector::MssqlUrl::new(s)?;
                 let psql = connector::Mssql::new(url).await?;
 
-                Arc::new(psql) as Arc<dyn Queryable>
+                Arc::new(psql) as Arc<dyn TransactionCapable>
             }
             _ => unimplemented!("Supported url schemes: file or sqlite, mysql, postgresql or jdbc:sqlserver."),
         };

quaint/src/tests/query.rs

@@ -115,6 +115,40 @@ async fn transactions_with_isolation_works(api: &mut dyn TestApi) -> crate::Resu
     Ok(())
 }
 
+#[test_each_connector(tags("mssql"))]
+async fn mssql_transaction_isolation_level(api: &mut dyn TestApi) -> crate::Result<()> {
+    let table = api.create_temp_table("id int, value int").await?;
+
+    let conn_a = api.conn();
+    // Start a transaction with the default isolation level, which in tests is
+    // set to READ UNCOMMITED via the DB url and insert a row, but do not commit the transaction.
+    let tx_a = conn_a.start_transaction(None).await?;
+    let insert = Insert::single_into(&table).value("value", 3).value("id", 4);
+    let rows_affected = tx_a.execute(insert.into()).await?;
+    assert_eq!(1, rows_affected);
+
+    // We want to verify that pooled connection behaves the same way, so we test both cases.
+    let pool = api.create_pool()?;
+    for conn_b in [
+        Box::new(pool.check_out().await?) as Box<dyn TransactionCapable>,
+        Box::new(api.create_additional_connection().await?),
+    ] {
+        // Start a transaction that explicitly sets the isolation level to SNAPSHOT and query the table
+        // expecting to see the old state.
+        let tx_b = conn_b.start_transaction(Some(IsolationLevel::Snapshot)).await?;
+        let res = tx_b.query(Select::from_table(&table).into()).await?;
+        assert_eq!(0, res.len());
+
+        // Start a transaction without an explicit isolation level, it should be run with the default
+        // again, which is set to READ UNCOMMITED here.
+        let tx_c = conn_b.start_transaction(None).await?;
+        let res = tx_c.query(Select::from_table(&table).into()).await?;
+        assert_eq!(1, res.len());
+    }
+
+    Ok(())
+}
+
 // SQLite only supports serializable.
 #[test_each_connector(tags("sqlite"))]
 async fn sqlite_serializable_tx(api: &mut dyn TestApi) -> crate::Result<()> {

quaint/src/tests/test_api.rs

@@ -35,5 +35,6 @@ pub trait TestApi {
     fn autogen_id(&self, name: &str) -> String;
     fn conn(&self) -> &crate::single::Quaint;
     async fn create_additional_connection(&self) -> crate::Result<crate::single::Quaint>;
+    fn create_pool(&self) -> crate::Result<crate::pooled::Quaint>;
     fn get_name(&mut self) -> String;
 }

quaint/src/tests/test_api/mssql.rs

@@ -21,6 +21,10 @@ impl<'a> MsSql<'a> {
         let names = Generator::default();
         let conn = Quaint::new(&CONN_STR).await?;
 
+        // snapshot isolation enables us to test isolation levels easily
+        conn.raw_cmd("ALTER DATABASE tempdb SET ALLOW_SNAPSHOT_ISOLATION ON")
+            .await?;
+
         Ok(Self { names, conn })
     }
 }
@@ -76,6 +80,10 @@ impl<'a> TestApi for MsSql<'a> {
         Quaint::new(&CONN_STR).await
     }
 
+    fn create_pool(&self) -> crate::Result<crate::pooled::Quaint> {
+        Ok(crate::pooled::Quaint::builder(&CONN_STR)?.build())
+    }
+
     fn render_create_table(&mut self, table_name: &str, columns: &str) -> (String, String) {
         let table_name = format!("##{table_name}");
         let create = format!(

quaint/src/tests/test_api/mysql.rs

@@ -129,6 +129,10 @@ impl<'a> TestApi for MySql<'a> {
         Quaint::new(&self.conn_str).await
     }
 
+    fn create_pool(&self) -> crate::Result<crate::pooled::Quaint> {
+        Ok(crate::pooled::Quaint::builder(&CONN_STR)?.build())
+    }
+
     fn unique_constraint(&mut self, column: &str) -> String {
         format!("UNIQUE({column})")
     }

quaint/src/tests/test_api/postgres.rs

@@ -87,6 +87,10 @@ impl<'a> TestApi for PostgreSql<'a> {
         Quaint::new(&CONN_STR).await
     }
 
+    fn create_pool(&self) -> crate::Result<crate::pooled::Quaint> {
+        Ok(crate::pooled::Quaint::builder(&CONN_STR)?.build())
+    }
+
     fn unique_constraint(&mut self, column: &str) -> String {
         format!("UNIQUE({column})")
     }

quaint/src/tests/test_api/sqlite.rs

@@ -83,6 +83,10 @@ impl<'a> TestApi for Sqlite<'a> {
         Quaint::new(CONN_STR).await
     }
 
+    fn create_pool(&self) -> crate::Result<crate::pooled::Quaint> {
+        Ok(crate::pooled::Quaint::builder(CONN_STR)?.build())
+    }
+
     fn unique_constraint(&mut self, column: &str) -> String {
         format!("UNIQUE({column})")
     }

query-engine/connector-test-kit-rs/query-engine-tests/tests/new/metrics.rs

@@ -32,7 +32,7 @@ mod metrics {
 
         match runner.connector_version() {
             Sqlite(_) => assert_eq!(total_queries, 2),
-            SqlServer(_) => assert_eq!(total_queries, 10),
+            SqlServer(_) => assert_eq!(total_queries, 12),
             MongoDb(_) => assert_eq!(total_queries, 5),
             CockroachDb(_) => assert_eq!(total_queries, 2),
             MySql(_) => assert_eq!(total_queries, 9),