From 13783e3f7fb8096c86baaab2b284d61520718d20 Mon Sep 17 00:00:00 2001 From: Fernando Takagi Date: Tue, 14 Jul 2026 12:54:03 -0300 Subject: [PATCH] notransaction + wrap txError --- README.md | 6 +++ migrate.go | 80 +++++++++++++++++++++++++-------------- migrate_test.go | 59 ++++++++++++++++++++++++++--- sqlparse/sqlparse_test.go | 49 ++++++++++++++++++++++++ 4 files changed, 160 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index 27e89a45..d2793d7a 100644 --- a/README.md +++ b/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). diff --git a/migrate.go b/migrate.go index a1a21b90..be6c6b63 100644 --- a/migrate.go +++ b/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) diff --git a/migrate_test.go b/migrate_test.go index c86f5057..a5a826cc 100644 --- a/migrate_test.go +++ b/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{ diff --git a/sqlparse/sqlparse_test.go b/sqlparse/sqlparse_test.go index 0f3f8a7c..380dfb65 100644 --- a/sqlparse/sqlparse_test.go +++ b/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