Implement notransaction option and wrap statement errors

README.md

@@ -135,6 +135,12 @@ CREATE UNIQUE INDEX CONCURRENTLY people_unique_id_idx ON people (id);
 DROP INDEX people_unique_id_idx;
 ```
 
+Because a `notransaction` migration runs outside of a transaction, it is **not atomic**: if it fails partway through, the statements that already ran remain applied and the migration is *not* recorded as complete.
+
+Each statement in a `notransaction` migration must be a single SQL command. Statements are sent using PostgreSQL's simple query protocol, which wraps a message containing multiple commands in an implicit transaction. Grouping several commands inside one `StatementBegin`/`StatementEnd` block therefore reintroduces a transaction.
+
+The `notransaction` option applies to the whole `Up` (or `Down`) section. A file may contain several `-- +migrate Up` sections, but if any of them sets `notransaction` then *all* of that migration's up statements run outside a transaction. You cannot mix transactional and non-transactional statements in one migration, so keep statements that must run outside a transaction in their own migration file.
+
 ## Embedding migrations with libraries that implement `http.FileSystem`
 
 You can also embed migrations with any library that implements `http.FileSystem`, like [`vfsgen`](https://github.com/shurcooL/vfsgen), [`parcello`](https://github.com/phogolabs/parcello), or [`go-resources`](https://github.com/omeid/go-resources).

migrate.go

@@ -4,8 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"github.com/jackc/pgx/v5"
-	"github.com/jackc/pgx/v5/pgtype"
 	"io"
 	"net/http"
 	"os"
@@ -16,6 +14,10 @@ import (
 	"strings"
 	"time"
 
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgtype"
+
 	"github.com/heroiclabs/sql-migrate/sqlparse"
 )
 
@@ -73,9 +75,13 @@ func (p *PlanError) Error() string {
 		p.Migration.Id, p.ErrorMessage)
 }
 
-// TxError is returned when any error is encountered during a database
-// transaction. It contains the relevant *Migration and notes it's Id in the
-// Error function output.
+// TxError is returned when an error is encountered while applying a migration
+// or recording it in the tracking table. It contains the relevant *Migration
+// and notes its Id in the Error function output.
+//
+// Note: for migrations run with the notransaction option there is no
+// surrounding transaction, so a TxError does not imply the migration's
+// statements were rolled back.
 type TxError struct {
 	Migration *Migration
 	Err       error
@@ -391,35 +397,24 @@ func (ms MigrationSet) applyMigrations(ctx context.Context, db *pgx.Conn, dir Mi
 	applied := 0
 
 	for _, migration := range migrations {
-		tx, err := db.Begin(ctx)
-		if err != nil {
-			return applied, fmt.Errorf("failed to init db transaction: %s", err.Error())
-		}
-
-		for _, stmt := range migration.Queries {
-			if _, err = tx.Exec(ctx, stmt); err != nil {
-				tx.Rollback(ctx)
-				return applied, fmt.Errorf("failed to exec migration statement %q: %s", stmt, err.Error())
+		if migration.DisableTransaction {
+			if err := ms.runMigration(ctx, db, dir, migration); err != nil {
+				return applied, err
+			}
+		} else {
+			tx, err := db.Begin(ctx)
+			if err != nil {
+				return applied, newTxError(migration, fmt.Errorf("failed to init db transaction: %s", err.Error()))
 			}
-		}
 
-		switch dir {
-		case Up:
-			if _, err = tx.Exec(ctx, fmt.Sprintf("INSERT INTO %q (id, applied_at) VALUES ($1, now())", ms.TableName), migration.Id); err != nil {
+			if err := ms.runMigration(ctx, tx, dir, migration); err != nil {
 				tx.Rollback(ctx)
-				return applied, newTxError(migration, err)
+				return applied, err
 			}
-		case Down:
-			if _, err = tx.Exec(ctx, fmt.Sprintf("DELETE FROM %q WHERE id = $1", ms.TableName), migration.Id); err != nil {
-				tx.Rollback(ctx)
+
+			if err := tx.Commit(ctx); err != nil {
 				return applied, newTxError(migration, err)
 			}
-		default:
-			panic("Invalid direction")
-		}
-
-		if err := tx.Commit(ctx); err != nil {
-			return applied, newTxError(migration, err)
 		}
 
 		applied++
@@ -428,6 +423,35 @@ func (ms MigrationSet) applyMigrations(ctx context.Context, db *pgx.Conn, dir Mi
 	return applied, nil
 }
 
+// executor is the subset of *pgx.Conn and pgx.Tx used to run a migration.
+type executor interface {
+	Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
+}
+
+// runMigration executes the migration's statements and records the result in the tracking table using the given executor.
+func (ms MigrationSet) runMigration(ctx context.Context, exec executor, dir MigrationDirection, migration *PlannedMigration) error {
+	for _, stmt := range migration.Queries {
+		if _, err := exec.Exec(ctx, stmt); err != nil {
+			return newTxError(migration, fmt.Errorf("failed to exec migration statement %q: %s", stmt, err.Error()))
+		}
+	}
+
+	switch dir {
+	case Up:
+		if _, err := exec.Exec(ctx, fmt.Sprintf("INSERT INTO %q (id, applied_at) VALUES ($1, now())", ms.getTableName()), migration.Id); err != nil {
+			return newTxError(migration, err)
+		}
+	case Down:
+		if _, err := exec.Exec(ctx, fmt.Sprintf("DELETE FROM %q WHERE id = $1", ms.getTableName()), migration.Id); err != nil {
+			return newTxError(migration, err)
+		}
+	default:
+		panic("Invalid direction")
+	}
+
+	return nil
+}
+
 // Plan a migration.
 func PlanMigration(ctx context.Context, db *pgx.Conn, m MigrationSource, dir MigrationDirection, max int) ([]*PlannedMigration, error) {
 	return migSet.PlanMigration(ctx, db, m, dir, max)

migrate_test.go

@@ -3,9 +3,10 @@ package migrate
 import (
 	"context"
 	"fmt"
+	"net/http"
+
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgtype"
-	"net/http"
 
 	. "gopkg.in/check.v1"
 )
@@ -334,18 +335,22 @@ func (s *SqliteMigrateSuite) TestMigrateDownFull(c *C) {
 	c.Assert(n, Equals, 0)
 }
 
-func (s *SqliteMigrateSuite) TestMigrateTransaction(c *C) {
-	migrations := &MemoryMigrationSource{
+func insertPeopleMigrations(disableTx bool, up ...string) *MemoryMigrationSource {
+	return &MemoryMigrationSource{
 		Migrations: []*Migration{
 			testMigrations[0],
 			testMigrations[1],
 			{
-				Id:   "125",
-				Up:   []string{"INSERT INTO people (id, first_name) VALUES (1, 'Test')", "SELECT fail"},
-				Down: []string{}, // Not important here
+				Id:                   "125",
+				Up:                   up,
+				DisableTransactionUp: disableTx,
 			},
 		},
 	}
+}
+
+func (s *SqliteMigrateSuite) TestMigrateTransaction(c *C) {
+	migrations := insertPeopleMigrations(false, "INSERT INTO people (id, first_name) VALUES (1, 'Test')", "SELECT fail")
 
 	ctx := context.Background()
 	// Should fail, transaction should roll back the INSERT.
@@ -360,6 +365,48 @@ func (s *SqliteMigrateSuite) TestMigrateTransaction(c *C) {
 	c.Assert(count, Equals, 0)
 }
 
+func (s *SqliteMigrateSuite) TestMigrateNoTransaction(c *C) {
+	migrations := insertPeopleMigrations(true, "INSERT INTO people (id, first_name) VALUES (1, 'Test')")
+
+	ctx := context.Background()
+	n, err := Exec(ctx, s.Db, migrations, Up)
+	c.Assert(err, IsNil)
+	c.Assert(n, Equals, 3)
+
+	// The row should be inserted and the migration recorded.
+	var count int
+	err = s.Db.QueryRow(ctx, "SELECT COUNT(*) FROM people").Scan(&count)
+	c.Assert(err, IsNil)
+	c.Assert(count, Equals, 1)
+
+	var recorded int
+	err = s.Db.QueryRow(ctx, fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE id = $1", DefaultMigrationTableName), "125").Scan(&recorded)
+	c.Assert(err, IsNil)
+	c.Assert(recorded, Equals, 1)
+}
+
+func (s *SqliteMigrateSuite) TestMigrateNoTransactionNoRollback(c *C) {
+	migrations := insertPeopleMigrations(true, "INSERT INTO people (id, first_name) VALUES (1, 'Test')", "SELECT fail")
+
+	ctx := context.Background()
+	// Should fail on the second statement. Without a surrounding transaction,
+	// the INSERT is NOT rolled back.
+	n, err := Exec(ctx, s.Db, migrations, Up)
+	c.Assert(err, Not(IsNil))
+	c.Assert(n, Equals, 2)
+
+	// INSERT should still be present.
+	var count int
+	err = s.Db.QueryRow(ctx, "SELECT COUNT(*) FROM people").Scan(&count)
+	c.Assert(err, IsNil)
+	c.Assert(count, Equals, 1)
+
+	var recorded int
+	err = s.Db.QueryRow(ctx, fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE id = $1", DefaultMigrationTableName), "125").Scan(&recorded)
+	c.Assert(err, IsNil)
+	c.Assert(recorded, Equals, 0)
+}
+
 func (s *SqliteMigrateSuite) TestPlanMigration(c *C) {
 	migrations := &MemoryMigrationSource{
 		Migrations: []*Migration{

sqlparse/sqlparse_test.go

@@ -88,6 +88,39 @@ func (s *SqlParseSuite) TestIntentionallyBadStatements(c *C) {
 	}
 }
 
+func (s *SqlParseSuite) TestNoTransaction(c *C) {
+	type testData struct {
+		sql         string
+		disableUp   bool
+		disableDown bool
+	}
+
+	tests := []testData{
+		{
+			sql:         functxt,
+			disableUp:   false,
+			disableDown: false,
+		},
+		{
+			sql:         notxUptxt,
+			disableUp:   true,
+			disableDown: false,
+		},
+		{
+			sql:         notxDowntxt,
+			disableUp:   false,
+			disableDown: true,
+		},
+	}
+
+	for _, test := range tests {
+		migration, err := ParseMigration(strings.NewReader(test.sql))
+		c.Assert(err, IsNil)
+		c.Assert(migration.DisableTransactionUp, Equals, test.disableUp)
+		c.Assert(migration.DisableTransactionDown, Equals, test.disableDown)
+	}
+}
+
 func (s *SqlParseSuite) TestJustComment(c *C) {
 	for _, test := range justAComment {
 		_, err := ParseMigration(strings.NewReader(test))
@@ -186,6 +219,22 @@ CREATE TABLE fancier_post (
 DROP TABLE fancier_post;
 `
 
+// notransaction on the Up direction only
+var notxUptxt = `-- +migrate Up notransaction
+CREATE INDEX CONCURRENTLY idx_post_title ON post (title);
+
+-- +migrate Down
+DROP INDEX idx_post_title;
+`
+
+// notransaction on the Down direction only
+var notxDowntxt = `-- +migrate Up
+CREATE INDEX idx_post_title ON post (title);
+
+-- +migrate Down notransaction
+DROP INDEX CONCURRENTLY idx_post_title;
+`
+
 // raise error when statements are not explicitly ended
 var intentionallyBad = []string{
 	// first statement missing terminator