Repository: ariga/atlas
Pull request #1158 (2022-09-21 12:40:32)

sql/migrate: gracefully handle missing migration files


Patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate.go b/cmd/atlas/internal/cmdapi/migrate.go
index 6f7fd654..f58472fb 100644
--- a/cmd/atlas/internal/cmdapi/migrate.go
+++ b/cmd/atlas/internal/cmdapi/migrate.go
@@ -368,7 +368,7 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 		return err
 	}
 	if errors.Is(err, migrate.ErrNoPendingFiles) {
-		cmd.Println("The migration directory is synced with the database, no migration files to execute")
+		cmd.Println("No migration files to execute")
 		return nil
 	}
 	if n > 0 {
diff --git a/doc/website/blog/2022-09-05-announcing-migration-execution.mdx b/doc/website/blog/2022-09-05-announcing-migration-execution.mdx
index 89448326..3fe0b154 100644
--- a/doc/website/blog/2022-09-05-announcing-migration-execution.mdx
+++ b/doc/website/blog/2022-09-05-announcing-migration-execution.mdx
@@ -245,7 +245,7 @@ atlas migrate apply \
 ```
 
 ```shell
-The migration directory is synced with the database, no migration files to execute
+No migration files to execute
 ```
 
 </TabItem>
diff --git a/sql/migrate/migrate.go b/sql/migrate/migrate.go
index d2198c61..ec146fee 100644
--- a/sql/migrate/migrate.go
+++ b/sql/migrate/migrate.go
@@ -391,6 +391,18 @@ var (
 	ErrRevisionNotExist = errors.New("sql/migrate: revision not found")
 )
 
+// MissingMigrationError is returned if a revision is partially applied but
+// the matching migration file is not found in the migration directory.
+type MissingMigrationError struct{ Version, Description string }
+
+// Error implements error.
+func (e MissingMigrationError) Error() string {
+	return fmt.Sprintf(
+		"sql/migrate: missing migration: revision %q is partially applied but migration file was not found",
+		fmt.Sprintf("%s_%s.sql", e.Version, e.Description),
+	)
+}
+
 // NewExecutor creates a new Executor with default values.
 func NewExecutor(drv Driver, dir Dir, rrw RevisionReadWriter, opts ...ExecutorOption) (*Executor, error) {
 	if drv == nil {
@@ -523,12 +535,27 @@ func (e *Executor) Pending(ctx context.Context) ([]File, error) {
 		}
 		pending = migrations[idx:]
 	default:
-		last := revs[len(revs)-1]
-		idx := FilesLastIndex(migrations, func(f File) bool {
-			return f.Version() == last.Version
-		})
+		var (
+			last      = revs[len(revs)-1]
+			partially = last.Applied != last.Total
+			fn        = func(f File) bool { return f.Version() <= last.Version }
+		)
+		if partially {
+			// If the last file is partially applied, we need to find the matching migration file in order to
+			// continue execution at the correct statement.
+			fn = func(f File) bool { return f.Version() == last.Version }
+		}
+		// Consider all migration files having a version < the latest revision version as pending. If the
+		// last revision is partially applied, it is considered pending as well.
+		idx := FilesLastIndex(migrations, fn)
 		if idx == -1 {
-			return nil, fmt.Errorf("version %q exists in revisions table was not found in the migration directory", last.Version)
+			// If we cannot find the matching migration version for a partially applied migration,
+			// error out since we cannot determine how to proceed from here.
+			if partially {
+				return nil, &MissingMigrationError{last.Version, last.Description}
+			}
+			// All migrations have a higher version than the latest revision. Take every migration file as pending.
+			return migrations, nil
 		}
 		// If this file was not partially applied, take the next one.
 		if last.Applied == last.Total {


Test patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate_test.go b/cmd/atlas/internal/cmdapi/migrate_test.go
index 5dfccd55..94f58ae9 100644
--- a/cmd/atlas/internal/cmdapi/migrate_test.go
+++ b/cmd/atlas/internal/cmdapi/migrate_test.go
@@ -84,7 +84,7 @@ func TestMigrate_Apply(t *testing.T) {
 		"-u", openSQLite(t, ""),
 	)
 	require.NoError(t, err)
-	require.Equal(t, "The migration directory is synced with the database, no migration files to execute\n", s)
+	require.Equal(t, "No migration files to execute\n", s)
 
 	// Fails on directory without sum file.
 	require.NoError(t, os.Rename(
@@ -222,10 +222,10 @@ func TestMigrate_Apply(t *testing.T) {
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test2.db")),
 		"--tx-mode", "none",
 	)
-	require.NoError(t, exec.Command("cp", "-r", "testdata/sqlite2", "testdata/sqlite3").Run())
 	t.Cleanup(func() {
-		require.NoError(t, os.RemoveAll("testdata/sqlite3"))
+		_ = os.RemoveAll("testdata/sqlite3")
 	})
+	require.NoError(t, exec.Command("cp", "-r", "testdata/sqlite2", "testdata/sqlite3").Run())
 	sed(t, "s/col_2/col_5/g", "testdata/sqlite3/20220318104615_second.sql")
 	_, err = runCmd(Root, "migrate", "hash", "--dir", "file://testdata/sqlite3")
 	require.NoError(t, err)
@@ -262,10 +262,10 @@ func TestMigrate_Apply(t *testing.T) {
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test2.db")),
 	)
 	require.NoError(t, err)
-	require.Equal(t, "The migration directory is synced with the database, no migration files to execute\n", s)
+	require.Equal(t, "No migration files to execute\n", s)
 
-	// Dry run will print the statements in second migration file without executing them. No changes to the revisions
-	// will be done.
+	// Dry run will print the statements in second migration file without executing them.
+	// No changes to the revisions will be done.
 	s, err = runCmd(
 		Root, "migrate", "apply",
 		"--dir", "file://testdata/sqlite",
@@ -291,6 +291,80 @@ func TestMigrate_Apply(t *testing.T) {
 	revs, err = rrw.ReadRevisions(ctx)
 	require.NoError(t, err)
 	require.Len(t, revs, 1)
+	MigrateFlags.Apply.DryRun = false // global flag, undo for rest of tests
+
+	// Prerequisites for testing missing migration behavior.
+	c1, err = sqlclient.Open(ctx, fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test3.db")))
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, c1.Close())
+	})
+	require.NoError(t, os.Rename(
+		"testdata/sqlite3/20220318104615_second.sql",
+		"testdata/sqlite3/20220318104616_second.sql",
+	))
+	_, err = runCmd(Root, "migrate", "hash", "--dir", "file://testdata/sqlite3")
+	require.NoError(t, err)
+	rrw, err = migrate2.NewEntRevisions(ctx, c1)
+	require.NoError(t, err)
+	require.NoError(t, rrw.Migrate(ctx))
+
+	// No changes if the last revision has a greater version than the last migration.
+	require.NoError(t, rrw.WriteRevision(ctx, &migrate.Revision{Version: "zzz"}))
+	s, err = runCmd(
+		Root, "migrate", "apply",
+		"--dir", "file://testdata/sqlite3",
+		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test3.db")),
+	)
+	require.NoError(t, err)
+	require.Equal(t, "No migration files to execute\n", s)
+
+	// If the revision is before the last but after the first migration, only the last one is pending.
+	_, err = c1.ExecContext(ctx, "DROP table `atlas_schema_revisions`")
+	require.NoError(t, err)
+	s, err = runCmd(
+		Root, "migrate", "apply", "1",
+		"--dir", "file://testdata/sqlite3",
+		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test3.db")),
+	)
+	require.NoError(t, rrw.WriteRevision(ctx, &migrate.Revision{Version: "20220318104615"}))
+	require.NoError(t, err)
+	s, err = runCmd(
+		Root, "migrate", "apply",
+		"--dir", "file://testdata/sqlite3",
+		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test3.db")),
+	)
+	require.NoError(t, err)
+	require.NotContains(t, s, "20220318104614")                     // log to version
+	require.Contains(t, s, "20220318104616")                        // log to version
+	require.Contains(t, s, "ALTER TABLE `tbl` ADD `col_2` bigint;") // logs statement
+
+	// If the revision is before every migration file, every file is pending.
+	_, err = c1.ExecContext(ctx, "DROP table `atlas_schema_revisions`; DROP table `tbl`;")
+	require.NoError(t, err)
+	require.NoError(t, rrw.Migrate(ctx))
+	require.NoError(t, rrw.WriteRevision(ctx, &migrate.Revision{Version: "1"}))
+	require.NoError(t, err)
+	s, err = runCmd(
+		Root, "migrate", "apply",
+		"--dir", "file://testdata/sqlite3",
+		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test3.db")),
+	)
+	require.NoError(t, err)
+	require.Contains(t, s, "20220318104614")                         // log to version
+	require.Contains(t, s, "20220318104616")                         // log to version
+	require.Contains(t, s, "CREATE TABLE tbl (`col` int NOT NULL);") // logs statement
+	require.Contains(t, s, "ALTER TABLE `tbl` ADD `col_2` bigint;")  // logs statement
+
+	// If the revision is partially applied, error out.
+	require.NoError(t, rrw.WriteRevision(ctx, &migrate.Revision{Version: "z", Description: "z", Total: 1}))
+	require.NoError(t, err)
+	_, err = runCmd(
+		Root, "migrate", "apply",
+		"--dir", "file://testdata/sqlite3",
+		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test3.db")),
+	)
+	require.EqualError(t, err, migrate.MissingMigrationError{Version: "z", Description: "z"}.Error())
 }
 
 func TestMigrate_ApplyBaseline(t *testing.T) {
@@ -303,7 +377,7 @@ func TestMigrate_ApplyBaseline(t *testing.T) {
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
 	)
 	require.NoError(t, err)
-	require.Contains(t, s, "The migration directory is synced with the database, no migration files to execute")
+	require.Contains(t, s, "No migration files to execute")
 
 	// Next run without baseline should run the migration from the baseline.
 	s, err = runCmd(
diff --git a/internal/integration/integration_test.go b/internal/integration/integration_test.go
index 419d66a1..fd01d312 100644
--- a/internal/integration/integration_test.go
+++ b/internal/integration/integration_test.go
@@ -205,7 +205,7 @@ func testCLIMigrateApplyBC(t T, dialect string) {
 		"--url", t.url(""),
 	).CombinedOutput()
 	require.NoError(t, err, string(out))
-	require.Contains(t, string(out), "The migration directory is synced with the database")
+	require.Equal(t, "No migration files to execute\n", string(out))
 
 	// Connection to schema without flag will error.
 	out, err = exec.Command(
@@ -226,7 +226,7 @@ func testCLIMigrateApplyBC(t T, dialect string) {
 		"--revisions-schema", "atlas_schema_revisions",
 	).CombinedOutput()
 	require.NoError(t, err)
-	require.Contains(t, string(out), "The migration directory is synced with the database")
+	require.Equal(t, "No migration files to execute\n", string(out))
 
 	// Providing the flag to the schema instead will work as well.
 	t.migrate(
@@ -241,7 +241,7 @@ func testCLIMigrateApplyBC(t T, dialect string) {
 		"--revisions-schema", "bc_test",
 	).CombinedOutput()
 	require.NoError(t, err, string(out))
-	require.NotContains(t, string(out), "The migration directory is synced with the database")
+	require.NotContains(t, string(out), "No migration files to execute\n")
 
 	// Consecutive attempts do not need the flag anymore.
 	out, err = exec.Command(
@@ -251,7 +251,7 @@ func testCLIMigrateApplyBC(t T, dialect string) {
 		"--url", t.url("bc_test"),
 	).CombinedOutput()
 	require.NoError(t, err)
-	require.Contains(t, string(out), "The migration directory is synced with the database")
+	require.Equal(t, "No migration files to execute\n", string(out))
 
 	// Last, if bound to schema and no "old" behavior extra schema does
 	// exist, the revision table will be saved in the connected one.
diff --git a/internal/integration/testdata/mysql/cli-migrate-apply.txt b/internal/integration/testdata/mysql/cli-migrate-apply.txt
index ca0965a6..451d1896 100644
--- a/internal/integration/testdata/mysql/cli-migrate-apply.txt
+++ b/internal/integration/testdata/mysql/cli-migrate-apply.txt
@@ -22,7 +22,7 @@ cmpshow users users.sql
 cmpshow pets pets.sql
 
 atlas migrate apply --url URL --revisions-schema $db
-stdout 'The migration directory is synced with the database, no migration files to execute'
+stdout 'No migration files to execute'
 
 clearSchema
 
@@ -41,7 +41,7 @@ cmpshow users users.sql
 cmpshow pets pets.sql
 
 atlas migrate apply --url URL --revisions-schema $db 1
-stdout 'The migration directory is synced with the database, no migration files to execute'
+stdout 'No migration files to execute'
 
 -- migrations/1_first.sql --
 CREATE TABLE `users` (`id` bigint NOT NULL AUTO_INCREMENT, `age` bigint NOT NULL, `name` varchar(255) NOT NULL, PRIMARY KEY (`id`)) CHARSET utf8mb4 COLLATE utf8mb4_bin;
diff --git a/internal/integration/testdata/postgres/cli-migrate-apply.txt b/internal/integration/testdata/postgres/cli-migrate-apply.txt
index 220700af..715f8c86 100644
--- a/internal/integration/testdata/postgres/cli-migrate-apply.txt
+++ b/internal/integration/testdata/postgres/cli-migrate-apply.txt
@@ -20,7 +20,7 @@ cmpshow users users.sql
 cmpshow pets pets.sql
 
 atlas migrate apply --url URL --revisions-schema $db 1
-stdout 'The migration directory is synced with the database, no migration files to execute'
+stdout 'No migration files to execute'
 
 clearSchema
 
@@ -39,7 +39,7 @@ cmpshow users users.sql
 cmpshow pets pets.sql
 
 atlas migrate apply --url URL --revisions-schema $db 1
-stdout 'The migration directory is synced with the database, no migration files to execute'
+stdout 'No migration files to execute'
 
 clearSchema
 
diff --git a/internal/integration/testdata/sqlite/cli-migrate-apply.txt b/internal/integration/testdata/sqlite/cli-migrate-apply.txt
index 946a53d6..f334ff23 100644
--- a/internal/integration/testdata/sqlite/cli-migrate-apply.txt
+++ b/internal/integration/testdata/sqlite/cli-migrate-apply.txt
@@ -18,7 +18,7 @@ cmpshow users users.sql
 cmpshow pets pets.sql
 
 atlas migrate apply --url URL 1
-stdout 'The migration directory is synced with the database, no migration files to execute'
+stdout 'No migration files to execute'
 
 clearSchema
 
@@ -33,7 +33,7 @@ cmpshow users users.sql
 cmpshow pets pets.sql
 
 atlas migrate apply --url URL 1
-stdout 'The migration directory is synced with the database, no migration files to execute'
+stdout 'No migration files to execute'
 
 clearSchema
 
diff --git a/sql/migrate/migrate_test.go b/sql/migrate/migrate_test.go
index 83ada4bd..e8a3afbf 100644
--- a/sql/migrate/migrate_test.go
+++ b/sql/migrate/migrate_test.go
@@ -195,6 +195,53 @@ func TestExecutor_Pending(t *testing.T) {
 	p, err = ex.Pending(context.Background())
 	require.NoError(t, err)
 	require.Len(t, p, 1)
+
+	// Nothing to do if all migrations are applied.
+	rev3.Applied = rev3.Total
+	*rrw = []*migrate.Revision{rev1, rev2, rev3}
+	p, err = ex.Pending(context.Background())
+	require.ErrorIs(t, err, migrate.ErrNoPendingFiles)
+	require.Len(t, p, 0)
+
+	// If there is a revision in the past with no existing migration, we don't care.
+	rev3.Applied = rev3.Total
+	*rrw = []*migrate.Revision{{Version: "2.11"}, rev3}
+	p, err = ex.Pending(context.Background())
+	require.ErrorIs(t, err, migrate.ErrNoPendingFiles)
+	require.Len(t, p, 0)
+
+	// If only the last migration file is applied, we expect there are no pending files.
+	*rrw = []*migrate.Revision{rev3}
+	p, err = ex.Pending(context.Background())
+	require.ErrorIs(t, err, migrate.ErrNoPendingFiles)
+	require.Len(t, p, 0)
+
+	// If the last applied revision has no matching migration file, and the last
+	// migration version precedes that revision, there is nothing to do.
+	*rrw = []*migrate.Revision{{Version: "5"}}
+	p, err = ex.Pending(context.Background())
+	require.ErrorIs(t, err, migrate.ErrNoPendingFiles)
+	require.Len(t, p, 0)
+
+	// The applied revision precedes every migration file. Expect all files pending.
+	*rrw = []*migrate.Revision{{Version: "1.1"}}
+	p, err = ex.Pending(context.Background())
+	require.NoError(t, err)
+	require.Len(t, p, 3)
+
+	// All except one file are applied. The latest revision does not exist and its version is smaller than the last
+	// migration file. Expect the last file pending.
+	*rrw = []*migrate.Revision{rev1, rev2, {Version: "2.11"}}
+	p, err = ex.Pending(context.Background())
+	require.NoError(t, err)
+	require.Len(t, p, 1)
+	require.Equal(t, rev3.Version, p[0].Version())
+
+	// If the last revision is partially applied and the matching migration file does not exist, we have a problem.
+	*rrw = []*migrate.Revision{{Version: "deleted", Description: "desc", Total: 1}}
+	p, err = ex.Pending(context.Background())
+	require.EqualError(t, err, migrate.MissingMigrationError{Version: "deleted", Description: "desc"}.Error())
+	require.Len(t, p, 0)
 }
 
 func TestExecutor(t *testing.T) {