Repository: ariga/atlas
Pull request #1104 (2022-08-29 07:22:01)

sql/migrate: pending files are all files after the last fully applied one
Next PR will add the possibility to rollback more than one file in a transaction and therefore we need to re-apply more than one file in that case.

masseelch: > Can you please elaborate on the use case here? Like what is the difference from the previous logic.
> 
> Also, let’s add test cases for it.

The previous logic would look at the revision, take the last one (regardless of state), find the version matching in the migration directory and pick up the execution there. 

Now, we want to add a transaction mode to group all pending migrations. In that case, we have to go back to the last fully applied revision and then start over from there (since everything would have been rolled back).

Patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate.go b/cmd/atlas/internal/cmdapi/migrate.go
index 5811c6a7..e7c142ec 100644
--- a/cmd/atlas/internal/cmdapi/migrate.go
+++ b/cmd/atlas/internal/cmdapi/migrate.go
@@ -241,10 +241,11 @@ func init() {
 	MigrateApplyCmd.Flags().StringVarP(&MigrateFlags.Apply.LogFormat, migrateFlagLog, "", logFormatTTY, "log format to use")
 	revisionsFlag(MigrateApplyCmd.Flags())
 	MigrateApplyCmd.Flags().BoolVarP(&MigrateFlags.Apply.DryRun, migrateFlagDryRun, "", false, "do not actually execute any SQL but show it on screen")
-	MigrateApplyCmd.Flags().StringVarP(&MigrateFlags.Apply.BaselineVersion, migrateApplyFromVersion, "", "", "calculate pending files from the given version (including it)")
+	MigrateApplyCmd.Flags().StringVarP(&MigrateFlags.Apply.FromVersion, migrateApplyFromVersion, "", "", "calculate pending files from the given version (including it)")
 	MigrateApplyCmd.Flags().StringVarP(&MigrateFlags.Apply.BaselineVersion, migrateApplyBaselineVersion, "", "", "start the first migration after the given baseline version")
 	MigrateApplyCmd.Flags().BoolVarP(&MigrateFlags.Apply.AllowDirty, migrateApplyAllowDirty, "", false, "allow start working on a non-clean database")
 	urlFlag(&MigrateFlags.URL, migrateFlagURL, "u", MigrateApplyCmd.Flags())
+	MigrateApplyCmd.Flags().SortFlags = false
 	cobra.CheckErr(MigrateApplyCmd.MarkFlagRequired(migrateFlagURL))
 	// Diff flags.
 	urlFlag(&MigrateFlags.DevURL, migrateFlagDevURL, "", MigrateDiffCmd.Flags())
@@ -741,7 +742,7 @@ func dir(create bool) (migrate.Dir, error) {
 		f = func() (migrate.Dir, error) { return sqltool.NewFlywayDir(parts[1]) }
 	case formatLiquibase:
 		f = func() (migrate.Dir, error) { return sqltool.NewLiquibaseDir(parts[1]) }
-	case formatDbmate:
+	case formatDBMate:
 		f = func() (migrate.Dir, error) { return sqltool.NewDBMateDir(parts[1]) }
 	default:
 		return nil, fmt.Errorf("unknown dir format %q", MigrateFlags.DirFormat)
@@ -931,7 +932,7 @@ const (
 	formatGoose         = "goose"
 	formatFlyway        = "flyway"
 	formatLiquibase     = "liquibase"
-	formatDbmate        = "dbmate"
+	formatDBMate        = "dbmate"
 )
 
 func formatter() (migrate.Formatter, error) {
@@ -946,7 +947,7 @@ func formatter() (migrate.Formatter, error) {
 		return sqltool.FlywayFormatter, nil
 	case formatLiquibase:
 		return sqltool.LiquibaseFormatter, nil
-	case formatDbmate:
+	case formatDBMate:
 		return sqltool.DBMateFormatter, nil
 	default:
 		return nil, fmt.Errorf("unknown format %q", MigrateFlags.DirFormat)
@@ -1083,14 +1084,10 @@ func migrateFlagsFromEnv(cmd *cobra.Command, _ []string) error {
 
 type (
 	// dryRunDriver wraps a migrate.Driver without executing any SQL statements.
-	dryRunDriver struct {
-		migrate.Driver
-	}
+	dryRunDriver struct{ migrate.Driver }
 
 	// dryRunRevisions wraps a migrate.RevisionReadWriter without executing any SQL statements.
-	dryRunRevisions struct {
-		migrate.RevisionReadWriter
-	}
+	dryRunRevisions struct{ migrate.RevisionReadWriter }
 )
 
 // QueryContext overrides the wrapped schema.ExecQuerier to not execute any SQL.
diff --git a/cmd/atlas/internal/migrate/ent/convert.tmpl b/cmd/atlas/internal/migrate/ent/convert.tmpl
new file mode 100644
index 00000000..1d8bcf35
--- /dev/null
+++ b/cmd/atlas/internal/migrate/ent/convert.tmpl
@@ -0,0 +1,34 @@
+{{/* gotype: entgo.io/ent/entc/gen.Graph */}}
+
+{{ define "convert" }}
+
+{{ $pkg := base $.Config.Package }}
+{{ template "header" $ }}
+
+import "ariga.io/atlas/sql/migrate"
+
+{{ range $n := $.Nodes }}
+    {{ if eq $n.Name "Revision" }}
+        {{ $builder := $n.CreateName }}
+        {{ $receiver := receiver $builder }}
+
+        // SetRevision takes the values for each field from the given migrate.Revision.
+        func ({{ $receiver }} *{{ $builder }}) SetRevision(rev *migrate.Revision) *{{ $builder }} {
+            {{- range $f := $n.Fields }}
+                {{ $receiver }}.Set{{ $f.StructField }}(rev.{{ $f.StructField }})
+            {{- end }}
+            return {{ $receiver }}
+        }
+
+        // AtlasRevision returns an migrate.Revision from the current Revision.
+        func({{ $n.Receiver}} *Revision) AtlasRevision() *migrate.Revision {
+            return &migrate.Revision{
+                {{- range $f := $n.Fields }}
+                    {{ $f.StructField }}: r.{{ $f.StructField }},
+                {{- end }}
+            }
+        }
+    {{ end }}
+{{ end }}
+
+{{ end }}
diff --git a/doc/md/reference.md b/doc/md/reference.md
index 1f2658d9..cb3c84e4 100644
--- a/doc/md/reference.md
+++ b/doc/md/reference.md
@@ -107,12 +107,12 @@ If run with the "--dry-run" flag, atlas will not execute any SQL.
 ```
 #### Flags
 ```
-      --allow-dirty               allow start working on a non-clean database
-      --baseline string           start the first migration after the given baseline version
-      --dry-run                   do not actually execute any SQL but show it on screen
-      --from string               calculate pending files from the given version (including it)
       --log string                log format to use (default "tty")
       --revisions-schema string   schema name where the revisions table resides (default "atlas_schema_revisions")
+      --dry-run                   do not actually execute any SQL but show it on screen
+      --from string               calculate pending files from the given version (including it)
+      --baseline string           start the first migration after the given baseline version
+      --allow-dirty               allow start working on a non-clean database
   -u, --url string                [driver://username:password@address/dbname?param=value] select a database using the URL format
 
 ```
diff --git a/sql/migrate/migrate.go b/sql/migrate/migrate.go
index c4865998..04249064 100644
--- a/sql/migrate/migrate.go
+++ b/sql/migrate/migrate.go
@@ -10,7 +10,6 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
-	"io/fs"
 	"strings"
 	"time"
 
@@ -228,7 +227,7 @@ type (
 
 const (
 	// RevisionTypeUnknown represents an unknown revision type.
-	// This is type is unexpected and exists here to only ensure
+	// This type is unexpected and exists here to only ensure
 	// the type is not set to the zero value.
 	RevisionTypeUnknown RevisionType = 0
 
@@ -478,13 +477,7 @@ func (e *Executor) Pending(ctx context.Context) ([]File, error) {
 	if err != nil {
 		return nil, fmt.Errorf("sql/migrate: execute: select migration files: %w", err)
 	}
-	// Check if the existing revisions did come from the migration directory and not from somewhere else.
-	_, err = readHashFile(e.dir)
-	if err != nil && !errors.Is(err, fs.ErrNotExist) {
-		return nil, fmt.Errorf("sql/migrate: execute: read %s file: %w", HashFileName, err)
-	}
-	// If there is no atlas.sum file there are no migration files.
-	if errors.Is(err, fs.ErrNotExist) {
+	if len(migrations) == 0 {
 		return nil, ErrNoPendingFiles
 	}
 	var pending []File
@@ -505,7 +498,7 @@ func (e *Executor) Pending(ctx context.Context) ([]File, error) {
 				return f.Version() == e.baselineVer
 			})
 			if baseline == -1 {
-				return nil, fmt.Errorf("baseline version %q was not found", e.baselineVer)
+				return nil, fmt.Errorf("baseline version %q not found", e.baselineVer)
 			}
 			f := migrations[baseline]
 			// Mark the revision in the database as baseline revision.
@@ -514,29 +507,32 @@ func (e *Executor) Pending(ctx context.Context) ([]File, error) {
 			}
 			pending = migrations[baseline+1:]
 		}
-	// Not the first time we execute and a
-	// custom starting point was provided.
+	// Not the first time we execute and a custom starting point was provided.
 	case e.fromVer != "":
 		idx := FilesLastIndex(migrations, func(f File) bool {
 			return f.Version() == e.fromVer
 		})
 		if idx == -1 {
-			return nil, fmt.Errorf("starting point version %q was not found in the migration directory", e.fromVer)
+			return nil, fmt.Errorf("starting point version %q not found in the migration directory", e.fromVer)
 		}
 		pending = migrations[idx:]
 	default:
-		last := revs[len(revs)-1]
+		// Find the last fully applied revision.
+		lastIdx := RevisionsLastIndex(revs, func(r *Revision) bool {
+			return r.Applied == r.Total
+		})
+		if lastIdx == -1 {
+			pending = migrations
+			break
+		}
+		last := revs[lastIdx]
 		idx := FilesLastIndex(migrations, func(f File) bool {
 			return f.Version() == last.Version
 		})
 		if idx == -1 {
 			return nil, fmt.Errorf("version %q exists in revisions table was not found in the migration directory", last.Version)
 		}
-		// If this file was not partially applied, take the next one.
-		if last.Applied == last.Total {
-			idx++
-		}
-		pending = migrations[idx:]
+		pending = migrations[idx+1:]
 	}
 	if len(pending) == 0 {
 		return nil, ErrNoPendingFiles
@@ -644,8 +640,7 @@ func (e HistoryChangedError) Error() string {
 	return fmt.Sprintf("sql/migrate: execute: history changed: statement %d from file %q changed", e.Stmt, e.File)
 }
 
-// ExecuteN executes n pending migration files. If n<=0 all pending migration files are executed. It will not attempt
-// an execution if the database is not "clean" (has only successfully applied migrations).
+// ExecuteN executes n pending migration files. If n<=0 all pending migration files are executed.
 func (e *Executor) ExecuteN(ctx context.Context, n int) (err error) {
 	pending, err := e.Pending(ctx)
 	if err != nil {
@@ -860,3 +855,14 @@ func wrap(err1, err2 error) error {
 	}
 	return err1
 }
+
+// RevisionsLastIndex returns the index of the last revision
+// satisfying f(i), or -1 if none do.
+func RevisionsLastIndex(revs Revisions, f func(*Revision) bool) int {
+	for i := len(revs) - 1; i >= 0; i-- {
+		if f(revs[i]) {
+			return i
+		}
+	}
+	return -1
+}


Test patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate_test.go b/cmd/atlas/internal/cmdapi/migrate_test.go
index 610118af..8f2498fa 100644
--- a/cmd/atlas/internal/cmdapi/migrate_test.go
+++ b/cmd/atlas/internal/cmdapi/migrate_test.go
@@ -383,7 +383,7 @@ func TestMigrate_New(t *testing.T) {
 	require.Equal(t, 2, countFiles(t, p))
 
 	p = t.TempDir()
-	s, err = runCmd(Root, "migrate", "new", "dbmate", "--dir", "file://"+p, "--dir-format", formatDbmate)
+	s, err = runCmd(Root, "migrate", "new", "dbmate", "--dir", "file://"+p, "--dir-format", formatDBMate)
 	require.Zero(t, s)
 	require.NoError(t, err)
 	require.FileExists(t, filepath.Join(p, v+"_dbmate.sql"))
diff --git a/sql/migrate/migrate_test.go b/sql/migrate/migrate_test.go
index e084f334..4725e00a 100644
--- a/sql/migrate/migrate_test.go
+++ b/sql/migrate/migrate_test.go
@@ -195,6 +195,21 @@ func TestExecutor_Pending(t *testing.T) {
 	p, err = ex.Pending(context.Background())
 	require.NoError(t, err)
 	require.Len(t, p, 1)
+
+	// All versions have revisions, but are last two are rolled back (0 applied).
+	rev2.Applied = 0
+	rev3.Applied = 0
+	*rrw = mockRevisionReadWriter(migrate.Revisions{rev1, rev2, rev3})
+	p, err = ex.Pending(context.Background())
+	require.NoError(t, err)
+	require.Len(t, p, 2)
+
+	// All versions have revisions, but are all rolled back (0 applied).
+	rev1.Applied = 0
+	*rrw = mockRevisionReadWriter(migrate.Revisions{rev1, rev2, rev3})
+	p, err = ex.Pending(context.Background())
+	require.NoError(t, err)
+	require.Len(t, p, 3)
 }
 
 func TestExecutor(t *testing.T) {
@@ -414,7 +429,7 @@ func TestExecutor_FromVersion(t *testing.T) {
 	ex, err = migrate.NewExecutor(drv, dir, rrw, migrate.WithLogger(log), migrate.WithFromVersion("4"))
 	require.NoError(t, err)
 	files, err = ex.Pending(context.Background())
-	require.EqualError(t, err, `starting point version "4" was not found in the migration directory`)
+	require.EqualError(t, err, `starting point version "4" not found in the migration directory`)
 	require.Nil(t, files)
 }