767 no tx migration

sqlx-core/src/migrate/migration.rs

@@ -11,6 +11,7 @@ pub struct Migration {
     pub migration_type: MigrationType,
     pub sql: Cow<'static, str>,
     pub checksum: Cow<'static, [u8]>,
+    pub no_tx: bool,
 }
 
 impl Migration {
@@ -19,6 +20,7 @@ impl Migration {
         description: Cow<'static, str>,
         migration_type: MigrationType,
         sql: Cow<'static, str>,
+        no_tx: bool,
     ) -> Self {
         let checksum = Cow::Owned(Vec::from(Sha384::digest(sql.as_bytes()).as_slice()));
 
@@ -28,6 +30,7 @@ impl Migration {
             migration_type,
             sql,
             checksum,
+            no_tx,
         }
     }
 }

sqlx-core/src/migrate/migrator.rs

@@ -21,6 +21,8 @@ pub struct Migrator {
     pub ignore_missing: bool,
     #[doc(hidden)]
     pub locking: bool,
+    #[doc(hidden)]
+    pub no_tx: bool
 }
 
 fn validate_applied_migrations(
@@ -47,6 +49,7 @@ impl Migrator {
     pub const DEFAULT: Migrator = Migrator {
         migrations: Cow::Borrowed(&[]),
         ignore_missing: false,
+        no_tx: false,
         locking: true,
     };
 

sqlx-core/src/migrate/source.rs

@@ -97,7 +97,7 @@ pub fn resolve_blocking(path: PathBuf) -> Result<Vec<(Migration, PathBuf)>, Reso
         let parts = file_name.splitn(2, '_').collect::<Vec<_>>();
 
         if parts.len() != 2 || !parts[1].ends_with(".sql") {
-            // not of the format: <VERSION>_<DESCRIPTION>.sql; ignore
+            // not of the format: <VERSION>_<DESCRIPTION>.<REVERSIBLE_DIRECTION>.sql; ignore
             continue;
         }
 
@@ -108,9 +108,11 @@ pub fn resolve_blocking(path: PathBuf) -> Result<Vec<(Migration, PathBuf)>, Reso
             })?;
 
         let migration_type = MigrationType::from_filename(parts[1]);
+
         // remove the `.sql` and replace `_` with ` `
         let description = parts[1]
             .trim_end_matches(migration_type.suffix())
+            .trim_end_matches(".no_tx")
             .replace('_', " ")
             .to_owned();
 
@@ -122,12 +124,16 @@ pub fn resolve_blocking(path: PathBuf) -> Result<Vec<(Migration, PathBuf)>, Reso
             source: Some(e),
         })?;
 
+        // opt-out of migration transaction
+        let no_tx = sql.starts_with("-- no-transaction");
+
         migrations.push((
             Migration::new(
                 version,
                 Cow::Owned(description),
                 migration_type,
                 Cow::Owned(sql),
+                no_tx,
             ),
             entry_path,
         ));

sqlx-macros-core/src/migrate.rs

@@ -36,6 +36,7 @@ impl ToTokens for QuoteMigration {
             description,
             migration_type,
             checksum,
+            no_tx,
             ..
         } = &self.migration;
 
@@ -69,6 +70,7 @@ impl ToTokens for QuoteMigration {
                 description: ::std::borrow::Cow::Borrowed(#description),
                 migration_type:  #migration_type,
                 sql: ::std::borrow::Cow::Borrowed(#sql),
+                no_tx: #no_tx,
                 checksum: ::std::borrow::Cow::Borrowed(&[
                     #(#checksum),*
                 ]),

sqlx-postgres/src/migrate.rs

@@ -208,30 +208,46 @@ CREATE TABLE IF NOT EXISTS _sqlx_migrations (
         migration: &'m Migration,
     ) -> BoxFuture<'m, Result<Duration, MigrateError>> {
         Box::pin(async move {
-            let mut tx = self.begin().await?;
             let start = Instant::now();
-
-            // Use a single transaction for the actual migration script and the essential bookeeping so we never
-            // execute migrations twice. See https://github.com/launchbadge/sqlx/issues/1966.
-            // The `execution_time` however can only be measured for the whole transaction. This value _only_ exists for
-            // data lineage and debugging reasons, so it is not super important if it is lost. So we initialize it to -1
-            // and update it once the actual transaction completed.
-            let _ = tx.execute(&*migration.sql).await?;
-
-            // language=SQL
-            let _ = query(
-                r#"
+            if !migration.no_tx {
+                let mut tx = self.begin().await?;
+
+                // Use a single transaction for the actual migration script and the essential bookeeping so we never
+                // execute migrations twice. See https://github.com/launchbadge/sqlx/issues/1966.
+                // The `execution_time` however can only be measured for the whole transaction. This value _only_ exists for
+                // data lineage and debugging reasons, so it is not super important if it is lost. So we initialize it to -1
+                // and update it once the actual transaction completed.
+                let _ = tx.execute(&*migration.sql).await?;
+
+                // language=SQL
+                let _ = query(
+                    r#"
     INSERT INTO _sqlx_migrations ( version, description, success, checksum, execution_time )
     VALUES ( $1, $2, TRUE, $3, -1 )
                 "#,
-            )
-            .bind(migration.version)
-            .bind(&*migration.description)
-            .bind(&*migration.checksum)
-            .execute(&mut *tx)
-            .await?;
+                )
+                .bind(migration.version)
+                .bind(&*migration.description)
+                .bind(&*migration.checksum)
+                .execute(&mut *tx)
+                .await?;
 
-            tx.commit().await?;
+                tx.commit().await?;
+            } else {
+                query(&migration.sql).execute(&mut *self).await?;
+                // language=SQL
+                let _ = query(
+                    r#"
+    INSERT INTO _sqlx_migrations ( version, description, success, checksum, execution_time )
+    VALUES ( $1, $2, TRUE, $3, -1 )
+                "#,
+                )
+                .bind(migration.version)
+                .bind(&migration.description)
+                .bind(&*migration.checksum)
+                .execute(&mut *self)
+                .await?;
+            }
 
             // Update `elapsed_time`.
             // NOTE: The process may disconnect/die at this point, so the elapsed time value might be lost. We accept

src/macros/test.md

@@ -133,6 +133,7 @@ use sqlx::{PgPool, Row};
 #       migrations: Cow::Borrowed(&[]),
 #       ignore_missing: false,
 #       locking: true,
+#       no_tx: false
 #   };
 # } 
 

tests/postgres/migrate.rs

@@ -66,8 +66,28 @@ async fn reversible(mut conn: PoolConnection<Postgres>) -> anyhow::Result<()> {
     Ok(())
 }
 
+#[sqlx::test(migrations = false)]
+async fn no_tx(mut conn: PoolConnection<Postgres>) -> anyhow::Result<()> {
+    clean_up(&mut conn).await?;
+    let migrator = Migrator::new(Path::new("tests/postgres/migrations_no_tx")).await?;
+
+    // run migration
+    migrator.run(&mut conn).await?;
+
+    // check outcome
+    let res: String = conn
+        .fetch_one("SELECT datname FROM pg_database WHERE datname = 'test_db'")
+        .await?
+        .get(0);
+
+    assert_eq!(res, "test_db");
+
+    Ok(())
+}
+
 /// Ensure that we have a clean initial state.
 async fn clean_up(conn: &mut PgConnection) -> anyhow::Result<()> {
+    conn.execute("DROP DATABASE IF EXISTS test_db").await.ok();
     conn.execute("DROP TABLE migrations_simple_test").await.ok();
     conn.execute("DROP TABLE migrations_reversible_test")
         .await

tests/postgres/migrations_no_tx/0_create_db.sql

@@ -0,0 +1,3 @@
+-- no-transaction
+
+CREATE DATABASE test_db;

tests/postgres/test-attr.rs

@@ -127,7 +127,7 @@ async fn it_gets_posts_mixed_fixtures_path(pool: PgPool) -> sqlx::Result<()> {
 // This should apply migrations and then `../fixtures/postgres/users.sql` and `../fixtures/postgres/posts.sql`
 #[sqlx::test(
     migrations = "tests/postgres/migrations",
-    fixtures(path = "../fixtures/postgres", scripts("users.sql", "posts"))
+    fixtures("../fixtures/postgres/users.sql", "../fixtures/postgres/posts.sql")
 )]
 async fn it_gets_posts_custom_relative_fixtures_path(pool: PgPool) -> sqlx::Result<()> {
     let post_contents: Vec<String> =