Repository: ariga/atlas
Pull request #988 (2022-07-31 19:45:01)

sql/migrate: allow setting --from and --baseline versions
`--from` and `--baseline` may look the same but they are not; this will be explained soon in the docs.


Patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate.go b/cmd/atlas/internal/cmdapi/migrate.go
index 313bb755..90d63932 100644
--- a/cmd/atlas/internal/cmdapi/migrate.go
+++ b/cmd/atlas/internal/cmdapi/migrate.go
@@ -35,20 +35,23 @@ import (
 )
 
 const (
-	migrateFlagURL             = "url"
-	migrateFlagDevURL          = "dev-url"
-	migrateFlagDir             = "dir"
-	migrateFlagForce           = "force"
-	migrateFlagFormat          = "format"
-	migrateFlagLog             = "log"
-	migrateFlagRevisionsSchema = "revisions-schema"
-	migrateFlagDryRun          = "dry-run"
-	migrateFlagTo              = "to"
-	migrateFlagSchema          = "schema"
-	migrateDiffFlagVerbose     = "verbose"
-	migrateLintLatest          = "latest"
-	migrateLintGitDir          = "git-dir"
-	migrateLintGitBase         = "git-base"
+	migrateFlagURL              = "url"
+	migrateFlagDevURL           = "dev-url"
+	migrateFlagDir              = "dir"
+	migrateFlagForce            = "force"
+	migrateFlagFormat           = "format"
+	migrateFlagLog              = "log"
+	migrateFlagRevisionsSchema  = "revisions-schema"
+	migrateFlagDryRun           = "dry-run"
+	migrateFlagTo               = "to"
+	migrateFlagSchema           = "schema"
+	migrateDiffFlagVerbose      = "verbose"
+	migrateLintLatest           = "latest"
+	migrateLintGitDir           = "git-dir"
+	migrateLintGitBase          = "git-base"
+	migrateApplyAllowDirty      = "allow-dirty"
+	migrateApplyFromVersion     = "from"
+	migrateApplyBaselineVersion = "baseline"
 )
 
 var (
@@ -60,12 +63,17 @@ var (
 		ToURLs         []string
 		Schemas        []string
 		Format         string
-		LogFormat      string
 		RevisionSchema string
-		DryRun         bool
 		Force          bool
 		Verbose        bool
-		Lint           struct {
+		Apply          struct {
+			DryRun          bool
+			LogFormat       string
+			AllowDirty      bool
+			FromVersion     string
+			BaselineVersion string
+		}
+		Lint struct {
 			Format  string // log formatting
 			Latest  uint   // latest N migration files
 			GitDir  string // repository working dir
@@ -228,9 +236,12 @@ func init() {
 	MigrateCmd.PersistentFlags().BoolVarP(&MigrateFlags.Force, migrateFlagForce, "", false, "force a command to run on a broken migration directory state")
 	MigrateCmd.PersistentFlags().SortFlags = false
 	// Apply flags.
-	MigrateApplyCmd.Flags().StringVarP(&MigrateFlags.LogFormat, migrateFlagLog, "", logFormatTTY, "log format to use")
+	MigrateApplyCmd.Flags().StringVarP(&MigrateFlags.Apply.LogFormat, migrateFlagLog, "", logFormatTTY, "log format to use")
 	revisionsFlag(MigrateApplyCmd.Flags())
-	MigrateApplyCmd.Flags().BoolVarP(&MigrateFlags.DryRun, migrateFlagDryRun, "", false, "do not actually execute any SQL but show it on screen")
+	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.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())
 	cobra.CheckErr(MigrateApplyCmd.MarkFlagRequired(migrateFlagURL))
 	// Diff flags.
@@ -283,7 +294,6 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 	if err != nil {
 		return err
 	}
-	// Currently, only in DB revisions are supported.
 	var rrw migrate.RevisionReadWriter
 	rrw, err = entmigrate.NewEntRevisions(c, []entmigrate.Option{entmigrate.WithSchema(MigrateFlags.RevisionSchema)}...)
 	if err != nil {
@@ -292,17 +302,16 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 	if err := rrw.(*entmigrate.EntRevisions).Init(cmd.Context()); err != nil {
 		return err
 	}
-	defer func(rrw *entmigrate.EntRevisions, ctx context.Context) {
-		if err2 := rrw.Flush(ctx); err2 != nil {
+	defer func(rrw *entmigrate.EntRevisions) {
+		if err2 := rrw.Flush(cmd.Context()); err2 != nil {
 			if err != nil {
-				err = fmt.Errorf("%v: %w", err2, err)
-			} else {
-				err = err2
+				err2 = fmt.Errorf("%w: %v", err, err2)
 			}
+			err = err2
 		}
-	}(rrw.(*entmigrate.EntRevisions), cmd.Context())
+	}(rrw.(*entmigrate.EntRevisions))
 	// Determine pending files and lock the database while working.
-	ex, err := migrate.NewExecutor(c.Driver, dir, rrw, migrate.WithLogger(l))
+	ex, err := migrate.NewExecutor(c.Driver, dir, rrw, executorOptions(l)...)
 	if err != nil {
 		return err
 	}
@@ -337,12 +346,12 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 		drv migrate.Driver
 		tx  *sqlclient.TxClient
 	)
-	if MigrateFlags.DryRun {
+	if MigrateFlags.Apply.DryRun {
 		drv = &dryRunDriver{c.Driver}
 		rrw = &dryRunRevisions{rrw}
 	}
 	for _, f := range pending {
-		if !MigrateFlags.DryRun {
+		if !MigrateFlags.Apply.DryRun {
 			// Wrap the file execution in a transaction.
 			tx, err = c.Tx(cmd.Context(), nil)
 			if err != nil {
@@ -350,19 +359,19 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 			}
 			drv = tx.Driver
 		}
-		ex, err := migrate.NewExecutor(drv, dir, rrw, migrate.WithLogger(l))
+		ex, err := migrate.NewExecutor(drv, dir, rrw, executorOptions(l)...)
 		if err != nil {
 			return err
 		}
 		if err := ex.Execute(cmd.Context(), f); err != nil {
-			if !MigrateFlags.DryRun {
+			if !MigrateFlags.Apply.DryRun {
 				if err2 := tx.Rollback(); err2 != nil {
 					err = fmt.Errorf("%v: %w", err2, err)
 				}
 			}
 			return err
 		}
-		if !MigrateFlags.DryRun {
+		if !MigrateFlags.Apply.DryRun {
 			if err := tx.Commit(); err != nil {
 				return err
 			}
@@ -372,6 +381,20 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
+func executorOptions(l migrate.Logger) []migrate.ExecutorOption {
+	opts := []migrate.ExecutorOption{migrate.WithLogger(l)}
+	if MigrateFlags.Apply.AllowDirty {
+		opts = append(opts, migrate.WithAllowDirty(true))
+	}
+	if v := MigrateFlags.Apply.BaselineVersion; v != "" {
+		opts = append(opts, migrate.WithBaselineVersion(v))
+	}
+	if v := MigrateFlags.Apply.FromVersion; v != "" {
+		opts = append(opts, migrate.WithFromVersion(v))
+	}
+	return opts
+}
+
 // CmdMigrateDiffRun is the command executed when running the CLI with 'migrate diff' args.
 func CmdMigrateDiffRun(cmd *cobra.Command, args []string) error {
 	// Open a dev driver.
@@ -865,9 +888,7 @@ func formatter() (migrate.Formatter, error) {
 	}
 }
 
-const (
-	logFormatTTY = "tty"
-)
+const logFormatTTY = "tty"
 
 // LogTTY is a migrate.Logger that pretty prints execution progress.
 // If the connected out is not a tty, it will fall back to a non-colorful output.
@@ -945,11 +966,11 @@ func zero(v int) int {
 }
 
 func logFormat(out io.Writer) (migrate.Logger, error) {
-	switch MigrateFlags.LogFormat {
+	switch l := MigrateFlags.Apply.LogFormat; l {
 	case logFormatTTY:
 		return &LogTTY{out: out}, nil
 	default:
-		return nil, fmt.Errorf("unknown log-format %q", MigrateFlags.LogFormat)
+		return nil, fmt.Errorf("unknown log-format %q", l)
 	}
 }
 
@@ -1023,6 +1044,11 @@ func (dryRunDriver) Lock(context.Context, string, time.Duration) (schema.UnlockF
 	return func() error { return nil }, nil
 }
 
+// CheckClean implements the migrate.CleanChecker interface.
+func (dryRunDriver) CheckClean(context.Context, *migrate.TableIdent) error {
+	return nil
+}
+
 // Snapshot implements the migrate.Snapshoter interface.
 func (dryRunDriver) Snapshot(context.Context) (migrate.RestoreFunc, error) {
 	// We dry-run, we don't execute anything. Snapshotting not required.
diff --git a/cmd/atlas/internal/migrate/migrate.go b/cmd/atlas/internal/migrate/migrate.go
index ffeccad8..e566c534 100644
--- a/cmd/atlas/internal/migrate/migrate.go
+++ b/cmd/atlas/internal/migrate/migrate.go
@@ -17,13 +17,13 @@ import (
 	"entgo.io/ent/dialect/sql"
 )
 
+// DefaultRevisionSchema is the default schema for storing revisions table.
 const DefaultRevisionSchema = "atlas_schema_revisions"
 
 type (
 	// A EntRevisions provides implementation for the migrate.RevisionReadWriter interface.
 	EntRevisions struct {
 		ac     *sqlclient.Client  // underlying Atlas client
-		sc     *sqlclient.Client  // underlying Atlas client connected to the named schema
 		ec     *ent.Client        // underlying Ent client
 		schema string             // name of the schema the revision table resides in
 		cache  []migrate.Revision // cache stores writes to Ent for blocked connections (like in SQLite).
@@ -56,11 +56,15 @@ func WithSchema(s string) Option {
 	}
 }
 
-// Init makes sure the revision table does exist in the connected database.
+// Ident returns the table identifier.
+func (r *EntRevisions) Ident() *migrate.TableIdent {
+	return &migrate.TableIdent{Name: revision.Table, Schema: r.schema}
+}
+
+// Init runs migration for the revisions' table in the connected database.
 func (r *EntRevisions) Init(ctx context.Context) error {
 	// Try to open a connection to the schema we are storing the revision table in.
-	var err error
-	r.sc, err = sqlclient.OpenURL(ctx, r.ac.URL.URL, sqlclient.OpenSchema(r.schema))
+	sc, err := sqlclient.OpenURL(ctx, r.ac.URL.URL, sqlclient.OpenSchema(r.schema))
 	// If the driver does not support changing the schema (most likely SQLite) use the existing connection.
 	if err != nil && errors.Is(err, sqlclient.ErrUnsupported) {
 		r.ec = ent.NewClient(ent.Driver(sql.OpenDB(r.ac.Name, r.ac.DB)))
@@ -79,14 +83,14 @@ func (r *EntRevisions) Init(ctx context.Context) error {
 		}
 	}
 	// If the previous connection attempt was unsuccessful, re-try with the schema present.
-	if r.sc == nil {
-		r.sc, err = sqlclient.OpenURL(ctx, r.ac.URL.URL, sqlclient.OpenSchema(r.schema))
+	if sc == nil {
+		sc, err = sqlclient.OpenURL(ctx, r.ac.URL.URL, sqlclient.OpenSchema(r.schema))
 		if err != nil {
 			return err
 		}
 	}
-	r.ac.AddClosers(r.sc)
-	r.ec = ent.NewClient(ent.Driver(sql.OpenDB(r.sc.Name, r.sc.DB)))
+	r.ac.AddClosers(sc)
+	r.ec = ent.NewClient(ent.Driver(sql.OpenDB(sc.Name, sc.DB)))
 	return r.ec.Schema.Create(ctx)
 }
 
@@ -99,7 +103,7 @@ func (r *EntRevisions) ReadRevision(ctx context.Context, v string) (*migrate.Rev
 		return nil, err
 	}
 	if ent.IsNotFound(err) {
-		return nil, migrate.ErrNotExist
+		return nil, migrate.ErrRevisionNotExist
 	}
 	return fromEnt(rev), nil
 }
diff --git a/doc/md/reference.md b/doc/md/reference.md
index 2f4bd034..a56ce407 100644
--- a/doc/md/reference.md
+++ b/doc/md/reference.md
@@ -107,7 +107,10 @@ 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")
   -u, --url string                [driver://username:password@address/dbname?param=value] select a database using the URL format
diff --git a/sql/migrate/dir.go b/sql/migrate/dir.go
index 64059d96..198e8aa9 100644
--- a/sql/migrate/dir.go
+++ b/sql/migrate/dir.go
@@ -48,8 +48,6 @@ type (
 		Desc() string
 		// Version returns the version of the migration File.
 		Version() string
-		// Baseline indicates if this file is marked as a baseline migration.
-		Baseline() bool
 		// Bytes returns the read content of the file.
 		Bytes() []byte
 		// Stmts returns the set of SQL statements this file holds.
@@ -131,12 +129,6 @@ func (f LocalFile) Name() string {
 	return f.n
 }
 
-// Baseline reports if this file marked as a baseline migration.
-func (f LocalFile) Baseline() bool {
-	_, ok := directive(string(f.b), directiveBaseline, directivePrefixSQL)
-	return ok
-}
-
 // Desc implements File.Desc.
 func (f LocalFile) Desc() string {
 	parts := strings.SplitN(f.n, "_", 2)
@@ -148,7 +140,7 @@ func (f LocalFile) Desc() string {
 
 // Version implements File.Version.
 func (f LocalFile) Version() string {
-	return strings.SplitN(f.n, "_", 2)[0]
+	return strings.SplitN(strings.TrimSuffix(f.n, ".sql"), "_", 2)[0]
 }
 
 // Stmts implements Scanner.Stmts. It reads migration file line-by-line and expects a statement to be one line only.
@@ -374,10 +366,9 @@ func FilesLastIndex(files []File, f func(File) bool) int {
 }
 
 const (
+	// atlas:sum directive.
 	directiveSum  = "sum"
 	sumModeIgnore = "ignore"
-	// atlas:baseline directive.
-	directiveBaseline = "baseline"
 	// atlas:delimiter directive.
 	directiveDelimiter = "delimiter"
 	directivePrefixSQL = "-- "
diff --git a/sql/migrate/migrate.go b/sql/migrate/migrate.go
index 1763c1be..c1433fbd 100644
--- a/sql/migrate/migrate.go
+++ b/sql/migrate/migrate.go
@@ -10,7 +10,7 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
-	"os"
+	"io/fs"
 	"strings"
 	"time"
 
@@ -131,20 +131,6 @@ func Conn(drv Driver, opts *schema.InspectRealmOption) StateReader {
 }
 
 type (
-	// A RevisionReadWriter reads and writes information about a Revisions to a persistent storage.
-	// If the implementation happens provide an "Init() error" method,
-	// it will be called once before attempting to read or write.
-	//
-	// Atlas drivers provide a sql based implementation of this interface by default.
-	RevisionReadWriter interface {
-		// ReadRevisions returns all revisions.
-		ReadRevisions(context.Context) (Revisions, error)
-		// ReadRevision returns a revision by version. Returns ErrNotExist if the version does not exist.
-		ReadRevision(context.Context, string) (*Revision, error)
-		// WriteRevision saves the revision to the storage.
-		WriteRevision(context.Context, *Revision) error
-	}
-
 	// Planner can plan the steps to take to migrate from one state to another. It uses the enclosed Dir to write
 	// those changes to versioned migration files.
 	Planner struct {
@@ -157,6 +143,20 @@ type (
 	// PlannerOption allows managing a Planner using functional arguments.
 	PlannerOption func(*Planner)
 
+	// A RevisionReadWriter wraps the functionality for reading
+	// and writing migration revisions in a database table.
+	RevisionReadWriter interface {
+		// Ident returns an object identifies this history table.
+		Ident() *TableIdent
+		// ReadRevisions returns all revisions.
+		ReadRevisions(context.Context) (Revisions, error)
+		// ReadRevision returns a revision by version.
+		// Returns ErrRevisionNotExist if the version does not exist.
+		ReadRevision(context.Context, string) (*Revision, error)
+		// WriteRevision saves the revision to the storage.
+		WriteRevision(context.Context, *Revision) error
+	}
+
 	// A Revision denotes an applied migration in a deployment. Used to track migration executions state of a database.
 	Revision struct {
 		// Version of the migration.
@@ -191,10 +191,13 @@ type (
 
 	// Executor is responsible to manage and execute a set of migration files against a database.
 	Executor struct {
-		drv Driver             // The Driver to access and manage the database.
-		dir Dir                // The Dir with migration files to use.
-		rrw RevisionReadWriter // The RevisionReadWriter to read and write database revisions to.
-		log Logger             // The Logger to use.
+		drv         Driver             // The Driver to access and manage the database.
+		dir         Dir                // The Dir with migration files to use.
+		rrw         RevisionReadWriter // The RevisionReadWriter to read and write database revisions to.
+		log         Logger             // The Logger to use.
+		fromVer     string             // Calculate pending files from the given version (including it).
+		baselineVer string             // Start the first migration after the given baseline version.
+		allowDirty  bool               // Allow start working on a non-clean database.
 	}
 
 	// ExecutorOption allows configuring an Executor using functional arguments.
@@ -290,8 +293,10 @@ var (
 	ErrLockUnsupported = errors.New("sql/migrate: driver does not support locking")
 	// ErrSnapshotUnsupported is returned if there is no Snapshoter given.
 	ErrSnapshotUnsupported = errors.New("sql/migrate: driver does not support taking a database snapshot")
-	// ErrNotExist is returned if the requested revision is not found in the storage.
-	ErrNotExist = errors.New("sql/migrate: revision not found")
+	// ErrCleanCheckerUnsupported is returned if there is no CleanChecker given.
+	ErrCleanCheckerUnsupported = errors.New("sql/migrate: driver does not support checking if database is clean")
+	// ErrRevisionNotExist is returned if the requested revision is not found in the storage.
+	ErrRevisionNotExist = errors.New("sql/migrate: revision not found")
 )
 
 // NewExecutor creates a new Executor with default values. // TODO(masseelch): Operator Version and other Meta
@@ -320,6 +325,12 @@ func NewExecutor(drv Driver, dir Dir, rrw RevisionReadWriter, opts ...ExecutorOp
 	if _, ok := drv.(Snapshoter); !ok {
 		return nil, ErrSnapshotUnsupported
 	}
+	if _, ok := drv.(CleanChecker); !ok {
+		return nil, ErrCleanCheckerUnsupported
+	}
+	if ex.baselineVer != "" && ex.allowDirty {
+		return nil, errors.New("sql/migrate: execute: baseline and allow-dirty are mutually exclusive")
+	}
 	return ex, nil
 }
 
@@ -331,6 +342,33 @@ func WithLogger(log Logger) ExecutorOption {
 	}
 }
 
+// WithFromVersion allows passing a file version as a starting point for calculating
+// pending migration scripts. It can be useful for skipping specific files.
+func WithFromVersion(v string) ExecutorOption {
+	return func(ex *Executor) error {
+		ex.fromVer = v
+		return nil
+	}
+}
+
+// WithBaselineVersion allows setting the baseline version of the database on the
+// first migration. Hence, all versions up to and including this version are skipped.
+func WithBaselineVersion(v string) ExecutorOption {
+	return func(ex *Executor) error {
+		ex.baselineVer = v
+		return nil
+	}
+}
+
+// WithAllowDirty defines if we can start working on a non-clean database
+// in the first migration execution.
+func WithAllowDirty(b bool) ExecutorOption {
+	return func(ex *Executor) error {
+		ex.allowDirty = b
+		return nil
+	}
+}
+
 // Lock acquires a lock for the executor.
 // It is considered a user error to not call Lock before the Pending and Execute methods.
 func (e *Executor) Lock(ctx context.Context) (schema.UnlockFunc, error) {
@@ -358,37 +396,59 @@ func (e *Executor) Pending(ctx context.Context) ([]File, error) {
 		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.
-	hf, err := readHashFile(e.dir)
-	if err != nil && !errors.Is(err, os.ErrNotExist) {
+	_, 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, os.ErrNotExist) {
+	if errors.Is(err, fs.ErrNotExist) {
 		return nil, ErrNoPendingFiles
 	}
-	// For up to len(revisions) revisions the migration files must both match in order and content.
-	if len(revs) > len(migrations) { // TODO(masseelch): keep compaction in mind.
-		return nil, errors.New("sql/migrate: execute: revisions and migrations mismatch: more revisions than migrations")
-	}
-	for i := range revs {
-		m := migrations[i]
-		r := revs[i]
-		// All fully applied revisions must match both version and description.
-		// In case of a not fully applied migration, the hash might change since the file could have been edited to fix
-		// a previous failure. Thus, hashes don't have to match in that case.
-		hash := (i < len(revs)-1 && hf[i].H != r.Hash) || (i == len(revs)-1 && r.Complete() == nil && hf[i].H != r.Hash)
-		if m.Version() != r.Version || m.Desc() != r.Description || hash {
-			return nil, fmt.Errorf("sql/migrate: execute: revisions and migrations mismatch: rev %q <> file %q", r.Version, m.Version())
+	var pending []File
+	switch {
+	// If it is the first time we run.
+	case len(revs) == 0:
+		var cerr *NotCleanError
+		if err = e.drv.(CleanChecker).CheckClean(ctx, e.rrw.Ident()); err != nil && !errors.As(err, &cerr) {
+			return nil, err
 		}
-	}
-	// If there are no revisions yet, return all migration files as pending.
-	if len(revs) == 0 {
-		return migrations, nil
-	}
-	pending := migrations[len(revs):]
-	// If the last revision is not applied in full, it is considered pending as well.
-	if last := revs[len(revs)-1]; last.Applied < last.Total {
-		pending = append(migrations[len(revs)-1:len(revs)], pending...)
+		// In case the workspace is not clean one of the flags are required.
+		if cerr != nil && !e.allowDirty && e.baselineVer == "" {
+			return nil, fmt.Errorf("%w. baseline version or allow-dirty are required", cerr)
+		}
+		pending = migrations
+		if e.baselineVer != "" {
+			baseline := FilesLastIndex(migrations, func(f File) bool {
+				return f.Version() == e.baselineVer
+			})
+			if baseline == -1 {
+				return nil, fmt.Errorf("baseline version %q was not found", e.baselineVer)
+			}
+			pending = migrations[baseline+1:]
+		}
+	// 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)
+		}
+		pending = migrations[idx:]
+	default:
+		last := revs[len(revs)-1]
+		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:]
 	}
 	if len(pending) == 0 {
 		return nil, ErrNoPendingFiles
@@ -426,10 +486,10 @@ func (e *Executor) Execute(ctx context.Context, m File) (err error) {
 	// If there already is a revision with this version in the database,
 	// and it is partially applied, continue where the last attempt was left off.
 	r, err := e.rrw.ReadRevision(ctx, version)
-	if err != nil && !errors.Is(err, ErrNotExist) {
+	if err != nil && !errors.Is(err, ErrRevisionNotExist) {
 		return fmt.Errorf("sql/migrate: execute: read revision: %w", err)
 	}
-	if errors.Is(err, ErrNotExist) {
+	if errors.Is(err, ErrRevisionNotExist) {
 		// Haven't seen this file before, create a new revision.
 		r = &Revision{
 			Version:     version,
@@ -577,6 +637,19 @@ type (
 	// RestoreFunc is returned by the Snapshoter to explicitly restore the database state.
 	RestoreFunc func(context.Context) error
 
+	// TableIdent describes a table identifier returned by the revisions table.
+	TableIdent struct {
+		Name   string // name of the table.
+		Schema string // optional schema.
+	}
+
+	// CleanChecker wraps the single CheckClean method.
+	CleanChecker interface {
+		// CheckClean checks if the connected realm or schema does not contain any resources besides the
+		// revision history table. A NotCleanError is returned in case the connection is not-empty.
+		CheckClean(context.Context, *TableIdent) error
+	}
+
 	// NotCleanError is returned when the connected dev-db is not in a clean state (aka it has schemas and tables).
 	// This check is done to ensure no data is lost by overriding it when working on the dev-db.
 	NotCleanError struct {
@@ -588,21 +661,26 @@ func (e NotCleanError) Error() string {
 	return "sql/migrate: connected database is not clean: " + e.Reason
 }
 
-// NopRevisionReadWriter is a RevisionsReadWriter that does nothing.
+// NopRevisionReadWriter is a RevisionReadWriter that does nothing.
 // It is useful for one-time replay of the migration directory.
 type NopRevisionReadWriter struct{}
 
-// ReadRevisions implements RevisionsReadWriter.ReadRevisions,
+// Ident implements RevisionsReadWriter.TableIdent.
+func (NopRevisionReadWriter) Ident() *TableIdent {
+	return nil
+}
+
+// ReadRevisions implements RevisionsReadWriter.ReadRevisions.
 func (NopRevisionReadWriter) ReadRevisions(context.Context) (Revisions, error) {
 	return nil, nil
 }
 
-// ReadRevision implements RevisionsReadWriter.ReadRevision,
+// ReadRevision implements RevisionsReadWriter.ReadRevision.
 func (NopRevisionReadWriter) ReadRevision(context.Context, string) (*Revision, error) {
-	return nil, ErrNotExist
+	return nil, ErrRevisionNotExist
 }
 
-// WriteRevision implements RevisionsReadWriter.WriteRevision,
+// WriteRevision implements RevisionsReadWriter.WriteRevision.
 func (NopRevisionReadWriter) WriteRevision(context.Context, *Revision) error {
 	return nil
 }
diff --git a/sql/mysql/driver.go b/sql/mysql/driver.go
index 762caa9b..f6645254 100644
--- a/sql/mysql/driver.go
+++ b/sql/mysql/driver.go
@@ -162,6 +162,37 @@ func (d *Driver) Snapshot(ctx context.Context) (migrate.RestoreFunc, error) {
 	}, nil
 }
 
+// CheckClean implements migrate.CleanChecker.
+func (d *Driver) CheckClean(ctx context.Context, revT *migrate.TableIdent) error {
+	s, err := d.InspectSchema(ctx, "", nil)
+	if err != nil && !schema.IsNotExistError(err) {
+		return err
+	}
+	if s != nil {
+		switch {
+		case len(s.Tables) == 0, (revT.Schema == "" || s.Name == revT.Schema) && len(s.Tables) == 1 && s.Tables[0].Name == revT.Name:
+			return nil
+		default:
+			return &migrate.NotCleanError{Reason: fmt.Sprintf("found table %q in schema %q", s.Tables[0].Name, s.Name)}
+		}
+	}
+	r, err := d.InspectRealm(ctx, nil)
+	if err != nil {
+		return err
+	}
+	switch n := len(r.Schemas); {
+	case n > 1:
+		return migrate.NotCleanError{Reason: fmt.Sprintf("found multiple schemas: %d", len(r.Schemas))}
+	case n == 1 && r.Schemas[0].Name != revT.Schema:
+		return migrate.NotCleanError{Reason: fmt.Sprintf("found schema %q", r.Schemas[0].Name)}
+	case n == 1 && len(r.Schemas[0].Tables) > 1:
+		return migrate.NotCleanError{Reason: fmt.Sprintf("found multiple tables: %d", len(r.Schemas[0].Tables))}
+	case n == 1 && len(r.Schemas[0].Tables) == 1 && r.Schemas[0].Tables[0].Name != revT.Name:
+		return migrate.NotCleanError{Reason: fmt.Sprintf("found table %q", r.Schemas[0].Tables[0].Name)}
+	}
+	return nil
+}
+
 func acquire(ctx context.Context, conn schema.ExecQuerier, name string, timeout time.Duration) error {
 	rows, err := conn.QueryContext(ctx, "SELECT GET_LOCK(?, ?)", name, int(timeout.Seconds()))
 	if err != nil {
diff --git a/sql/postgres/driver.go b/sql/postgres/driver.go
index 7a6bd739..6592bfe3 100644
--- a/sql/postgres/driver.go
+++ b/sql/postgres/driver.go
@@ -213,6 +213,36 @@ func (d *Driver) Snapshot(ctx context.Context) (migrate.RestoreFunc, error) {
 	return nil, migrate.NotCleanError{Reason: fmt.Sprintf("found schema %q", realm.Schemas[0].Name)}
 }
 
+// CheckClean implements migrate.CleanChecker.
+func (d *Driver) CheckClean(ctx context.Context, revT *migrate.TableIdent) error {
+	if d.schema != "" {
+		switch s, err := d.InspectSchema(ctx, d.schema, nil); {
+		case err != nil:
+			return err
+		case len(s.Tables) == 0, (revT.Schema == "" || s.Name == revT.Schema) && len(s.Tables) == 1 && s.Tables[0].Name == revT.Name:
+			return nil
+		default:
+			return &migrate.NotCleanError{Reason: fmt.Sprintf("found table %q in schema %q", s.Tables[0].Name, s.Name)}
+		}
+	}
+	r, err := d.InspectRealm(ctx, nil)
+	if err != nil {
+		return err
+	}
+	for _, s := range r.Schemas {
+		switch {
+		case len(s.Tables) == 0 && s.Name == "public":
+		case len(s.Tables) == 0 || s.Name != revT.Schema:
+			return &migrate.NotCleanError{Reason: fmt.Sprintf("found schema %q", s.Name)}
+		case len(s.Tables) > 1:
+			return &migrate.NotCleanError{Reason: fmt.Sprintf("found %d tables in schema %q", len(s.Tables), s.Name)}
+		case len(s.Tables) == 1 && s.Tables[0].Name != revT.Name:
+			return &migrate.NotCleanError{Reason: fmt.Sprintf("found table %q in schema %q", s.Tables[0].Name, s.Name)}
+		}
+	}
+	return nil
+}
+
 func acquire(ctx context.Context, conn schema.ExecQuerier, id uint32, timeout time.Duration) error {
 	switch {
 	// With timeout (context-based).
diff --git a/sql/schema/dsl.go b/sql/schema/dsl.go
index e568fae4..c534ac33 100644
--- a/sql/schema/dsl.go
+++ b/sql/schema/dsl.go
@@ -79,6 +79,15 @@ func NewRealm(schemas ...*Schema) *Realm {
 	return r
 }
 
+// AddSchemas adds and links the given schemas to the realm.
+func (r *Realm) AddSchemas(schemas ...*Schema) *Realm {
+	for _, s := range schemas {
+		s.SetRealm(r)
+	}
+	r.Schemas = append(r.Schemas, schemas...)
+	return r
+}
+
 // SetCharset sets or appends the Charset attribute
 // to the realm with the given value.
 func (r *Realm) SetCharset(v string) *Realm {
diff --git a/sql/sqlite/driver.go b/sql/sqlite/driver.go
index 292663fd..cc8b4089 100644
--- a/sql/sqlite/driver.go
+++ b/sql/sqlite/driver.go
@@ -113,6 +113,25 @@ func (d *Driver) Snapshot(ctx context.Context) (migrate.RestoreFunc, error) {
 	}, nil
 }
 
+// CheckClean implements migrate.CleanChecker.
+func (d *Driver) CheckClean(ctx context.Context, revT *migrate.TableIdent) error {
+	r, err := d.InspectRealm(ctx, nil)
+	if err != nil {
+		return err
+	}
+	switch n := len(r.Schemas); {
+	case n > 1:
+		return migrate.NotCleanError{Reason: fmt.Sprintf("found multiple schemas: %d", len(r.Schemas))}
+	case n == 1 && r.Schemas[0].Name != mainFile:
+		return migrate.NotCleanError{Reason: fmt.Sprintf("found schema %q", r.Schemas[0].Name)}
+	case n == 1 && len(r.Schemas[0].Tables) > 1:
+		return migrate.NotCleanError{Reason: fmt.Sprintf("found multiple tables: %d", len(r.Schemas[0].Tables))}
+	case n == 1 && len(r.Schemas[0].Tables) == 1 && r.Schemas[0].Tables[0].Name != revT.Name:
+		return migrate.NotCleanError{Reason: fmt.Sprintf("found table %q", r.Schemas[0].Tables[0].Name)}
+	}
+	return nil
+}
+
 // Lock implements the schema.Locker interface.
 func (d *Driver) Lock(_ context.Context, name string, timeout time.Duration) (schema.UnlockFunc, error) {
 	path := filepath.Join(os.TempDir(), name+".lock")


Test patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate_test.go b/cmd/atlas/internal/cmdapi/migrate_test.go
index c539477a..d5e15929 100644
--- a/cmd/atlas/internal/cmdapi/migrate_test.go
+++ b/cmd/atlas/internal/cmdapi/migrate_test.go
@@ -582,6 +582,10 @@ func (d *sqliteLockerDriver) Snapshot(ctx context.Context) (migrate.RestoreFunc,
 	return d.Driver.(migrate.Snapshoter).Snapshot(ctx)
 }
 
+func (d *sqliteLockerDriver) CheckClean(ctx context.Context, revT *migrate.TableIdent) error {
+	return d.Driver.(migrate.CleanChecker).CheckClean(ctx, revT)
+}
+
 func countFiles(t *testing.T, p string) int {
 	files, err := os.ReadDir(p)
 	require.NoError(t, err)
diff --git a/internal/integration/integration_test.go b/internal/integration/integration_test.go
index 741a27d9..2434d8f1 100644
--- a/internal/integration/integration_test.go
+++ b/internal/integration/integration_test.go
@@ -527,6 +527,10 @@ func plan(t T, name string, changes ...schema.Change) *migrate.Plan {
 
 type rrw migrate.Revisions
 
+func (r *rrw) Ident() *migrate.TableIdent {
+	return &migrate.TableIdent{}
+}
+
 func (r *rrw) WriteRevision(_ context.Context, rev *migrate.Revision) error {
 	for i, rev2 := range *r {
 		if rev2.Version == rev.Version {
@@ -544,7 +548,7 @@ func (r *rrw) ReadRevision(_ context.Context, v string) (*migrate.Revision, erro
 			return rev, nil
 		}
 	}
-	return nil, migrate.ErrNotExist
+	return nil, migrate.ErrRevisionNotExist
 }
 
 func (r *rrw) ReadRevisions(context.Context) (migrate.Revisions, error) {
diff --git a/sql/migrate/dir_test.go b/sql/migrate/dir_test.go
index 1a29908d..405203eb 100644
--- a/sql/migrate/dir_test.go
+++ b/sql/migrate/dir_test.go
@@ -182,16 +182,3 @@ func TestLocalDir(t *testing.T) {
 	require.Equal(t, "2.10.x-20", files[1].Version())
 	require.Equal(t, "description", files[1].Desc())
 }
-
-func TestLocalFile_Baseline(t *testing.T) {
-	dir, err := migrate.NewLocalDir("testdata/baseline")
-	require.NoError(t, err)
-	files, err := dir.Files()
-	require.NoError(t, err)
-	idx := migrate.FilesLastIndex(files, func(f migrate.File) bool {
-		return f.Baseline()
-	})
-	require.Equal(t, 2, idx)
-	require.True(t, files[idx].Baseline())
-	require.False(t, files[idx+1].Baseline())
-}
diff --git a/sql/migrate/migrate_test.go b/sql/migrate/migrate_test.go
index c11b30f0..32d8f037 100644
--- a/sql/migrate/migrate_test.go
+++ b/sql/migrate/migrate_test.go
@@ -303,28 +303,6 @@ func TestExecutor(t *testing.T) {
 	require.Equal(t, 1, drv.unlockCounter)
 	require.True(t, drv.released())
 
-	// Unknown revision present.
-	*rrw = mockRevisionReadWriter(migrate.Revisions{&migrate.Revision{Version: "unknown"}})
-	*drv = lockMockDriver{&mockDriver{}}
-	require.EqualError(t, ex.ExecuteN(context.Background(), 0),
-		`sql/migrate: execute: revisions and migrations mismatch: rev "unknown" <> file "1.a"`,
-	)
-
-	require.Equal(t, drv.lockCounter, 1)
-	require.Equal(t, drv.unlockCounter, 1)
-	require.True(t, drv.released())
-
-	// More revisions than migration files.
-	*rrw = mockRevisionReadWriter(migrate.Revisions{{Version: "unknown"}, rev1, rev2, rev2})
-	*drv = lockMockDriver{&mockDriver{}}
-	require.EqualError(t, ex.ExecuteN(context.Background(), 0),
-		`sql/migrate: execute: revisions and migrations mismatch: more revisions than migrations`,
-	)
-
-	require.Equal(t, drv.lockCounter, 1)
-	require.Equal(t, drv.unlockCounter, 1)
-	require.True(t, drv.released())
-
 	// Failing, counter will be correct.
 	*rrw = mockRevisionReadWriter(migrate.Revisions{rev1, rev2})
 	*drv = lockMockDriver{&mockDriver{}}
@@ -355,6 +333,77 @@ func TestExecutor(t *testing.T) {
 	require.ErrorIs(t, ex.ExecuteN(context.Background(), 0), migrate.ErrNoPendingFiles)
 }
 
+func TestExecutor_Baseline(t *testing.T) {
+	var (
+		drv = &lockMockDriver{&mockDriver{dirty: true}}
+		rrw = &mockRevisionReadWriter{}
+		log = &mockLogger{}
+	)
+	dir, err := migrate.NewLocalDir(filepath.Join("testdata/migrate", "sub"))
+	require.NoError(t, err)
+	ex, err := migrate.NewExecutor(drv, dir, rrw, migrate.WithLogger(log))
+	require.NoError(t, err)
+
+	// Require baseline-version or explicit flag to work on a dirty workspace.
+	files, err := ex.Pending(context.Background())
+	require.EqualError(t, err, "sql/migrate: connected database is not clean: found table. baseline version or allow-dirty are required")
+	require.Nil(t, files)
+
+	ex, err = migrate.NewExecutor(drv, dir, rrw, migrate.WithLogger(log), migrate.WithAllowDirty(true))
+	require.NoError(t, err)
+	files, err = ex.Pending(context.Background())
+	require.NoError(t, err)
+	require.Len(t, files, 3)
+
+	ex, err = migrate.NewExecutor(drv, dir, rrw, migrate.WithLogger(log), migrate.WithBaselineVersion("2.10.x-20"))
+	require.NoError(t, err)
+	files, err = ex.Pending(context.Background())
+	require.NoError(t, err)
+	require.Len(t, files, 1)
+
+	ex, err = migrate.NewExecutor(drv, dir, rrw, migrate.WithLogger(log), migrate.WithBaselineVersion("3"))
+	require.NoError(t, err)
+	files, err = ex.Pending(context.Background())
+	require.ErrorIs(t, err, migrate.ErrNoPendingFiles)
+}
+
+func TestExecutor_FromVersion(t *testing.T) {
+	var (
+		drv = &lockMockDriver{&mockDriver{}}
+		log = &mockLogger{}
+		rrw = &mockRevisionReadWriter{
+			{
+				Version:     "1.a",
+				Description: "sub.up",
+				Applied:     2,
+				Total:       2,
+				Hash:        "nXyZR020M/mH7LxkoTkJr7BcQkipVg90imQ9I4595dw=",
+			},
+		}
+	)
+	dir, err := migrate.NewLocalDir(filepath.Join("testdata/migrate", "sub"))
+	require.NoError(t, err)
+	ex, err := migrate.NewExecutor(drv, dir, rrw, migrate.WithLogger(log))
+	require.NoError(t, err)
+	files, err := ex.Pending(context.Background())
+	require.NoError(t, err)
+	require.Len(t, files, 2)
+
+	// Control the starting point.
+	ex, err = migrate.NewExecutor(drv, dir, rrw, migrate.WithLogger(log), migrate.WithFromVersion("3"))
+	require.NoError(t, err)
+	files, err = ex.Pending(context.Background())
+	require.NoError(t, err)
+	require.Len(t, files, 1)
+
+	// Starting point was not found.
+	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.Nil(t, files)
+}
+
 type (
 	mockDriver struct {
 		migrate.Driver
@@ -368,6 +417,7 @@ type (
 		unlockCounter int
 		failCounter   int
 		failWith      error
+		dirty         bool
 	}
 	lockMockDriver struct{ *mockDriver }
 )
@@ -392,16 +442,20 @@ func (m *mockDriver) ExecContext(_ context.Context, query string, _ ...interface
 func (m *mockDriver) InspectRealm(context.Context, *schema.InspectRealmOption) (*schema.Realm, error) {
 	return &m.realm, nil
 }
+
 func (m *mockDriver) RealmDiff(_, _ *schema.Realm) ([]schema.Change, error) {
 	return m.changes, nil
 }
+
 func (m *mockDriver) PlanChanges(context.Context, string, []schema.Change) (*migrate.Plan, error) {
 	return m.plan, nil
 }
+
 func (m *mockDriver) ApplyChanges(_ context.Context, changes []schema.Change) error {
 	m.applied = changes
 	return nil
 }
+
 func (m *mockDriver) Snapshot(context.Context) (migrate.RestoreFunc, error) {
 	if len(m.realm.Schemas) > 0 {
 		return nil, migrate.NotCleanError{}
@@ -412,6 +466,14 @@ func (m *mockDriver) Snapshot(context.Context) (migrate.RestoreFunc, error) {
 		return nil
 	}, nil
 }
+
+func (m *mockDriver) CheckClean(context.Context, *migrate.TableIdent) error {
+	if m.dirty {
+		return &migrate.NotCleanError{Reason: "found table"}
+	}
+	return nil
+}
+
 func (m *lockMockDriver) Lock(_ context.Context, name string, _ time.Duration) (schema.UnlockFunc, error) {
 	if _, ok := m.locks[name]; ok {
 		return nil, errors.New("lockErr")
@@ -427,12 +489,25 @@ func (m *lockMockDriver) Lock(_ context.Context, name string, _ time.Duration) (
 		return nil
 	}, nil
 }
+
 func (m *lockMockDriver) released() bool {
 	return len(m.locks) == 0
 }
 
 type mockRevisionReadWriter migrate.Revisions
 
+func (mockRevisionReadWriter) Ident() *migrate.TableIdent {
+	return nil
+}
+
+func (mockRevisionReadWriter) Exists(_ context.Context) (bool, error) {
+	return true, nil
+}
+
+func (mockRevisionReadWriter) Init(_ context.Context) error {
+	return nil
+}
+
 func (rrw *mockRevisionReadWriter) WriteRevision(_ context.Context, r *migrate.Revision) error {
 	for i, rev := range *rrw {
 		if rev.Version == r.Version {
@@ -450,7 +525,7 @@ func (rrw *mockRevisionReadWriter) ReadRevision(_ context.Context, v string) (*m
 			return r, nil
 		}
 	}
-	return nil, migrate.ErrNotExist
+	return nil, migrate.ErrRevisionNotExist
 }
 
 func (rrw *mockRevisionReadWriter) ReadRevisions(context.Context) (migrate.Revisions, error) {
diff --git a/sql/migrate/testdata/baseline/1_v1.sql b/sql/migrate/testdata/baseline/1_v1.sql
deleted file mode 100644
index 4dc4d431..00000000
--- a/sql/migrate/testdata/baseline/1_v1.sql
+++ /dev/null
@@ -1,2 +0,0 @@
-CREATE DATABASE market;
-CREATE TABLE orders (name varchar(255));
diff --git a/sql/migrate/testdata/baseline/2_v2.sql b/sql/migrate/testdata/baseline/2_v2.sql
deleted file mode 100644
index 3303abe4..00000000
--- a/sql/migrate/testdata/baseline/2_v2.sql
+++ /dev/null
@@ -1,1 +0,0 @@
-CREATE TABLE products (name varchar(255), price decimal);
diff --git a/sql/migrate/testdata/baseline/3_baseline.sql b/sql/migrate/testdata/baseline/3_baseline.sql
deleted file mode 100644
index 5f6eefc9..00000000
--- a/sql/migrate/testdata/baseline/3_baseline.sql
+++ /dev/null
@@ -1,6 +0,0 @@
--- atlas:baseline
-
-CREATE DATABASE market;
-CREATE TABLE orders (name varchar(255));
-CREATE TABLE products (name varchar(255), price decimal);
-
diff --git a/sql/migrate/testdata/baseline/4_initial.sql b/sql/migrate/testdata/baseline/4_initial.sql
deleted file mode 100644
index 4e2c3634..00000000
--- a/sql/migrate/testdata/baseline/4_initial.sql
+++ /dev/null
@@ -1,1 +0,0 @@
-ALTER TABLE orders ADD COLUMN product_id bigint;
\ No newline at end of file
diff --git a/sql/mysql/driver_test.go b/sql/mysql/driver_test.go
index be248a23..dd212ab1 100644
--- a/sql/mysql/driver_test.go
+++ b/sql/mysql/driver_test.go
@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"ariga.io/atlas/sql/internal/sqltest"
+	"ariga.io/atlas/sql/migrate"
 	"ariga.io/atlas/sql/schema"
 
 	"github.com/DATA-DOG/go-sqlmock"
@@ -127,6 +128,58 @@ func TestDriver_UnlockError(t *testing.T) {
 	})
 }
 
+func TestDriver_CheckClean(t *testing.T) {
+	s := schema.New("test")
+	drv := &Driver{Inspector: &mockInspector{schema: s}}
+	// Empty schema.
+	err := drv.CheckClean(context.Background(), nil)
+	require.NoError(t, err)
+	// Revisions table found.
+	s.AddTables(schema.NewTable("revisions"))
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Name: "revisions", Schema: "test"})
+	require.NoError(t, err)
+	// Multiple tables.
+	s.Tables = []*schema.Table{schema.NewTable("a"), schema.NewTable("revisions")}
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Name: "revisions", Schema: "test"})
+	require.EqualError(t, err, `sql/migrate: connected database is not clean: found table "a" in schema "test"`)
+
+	r := schema.NewRealm()
+	drv.Inspector = &mockInspector{realm: r}
+	// Empty realm.
+	err = drv.CheckClean(context.Background(), nil)
+	require.NoError(t, err)
+	// Revisions table found.
+	s.Tables = []*schema.Table{schema.NewTable("revisions")}
+	r.AddSchemas(s)
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Name: "revisions", Schema: "test"})
+	require.NoError(t, err)
+	// Unknown table.
+	s.Tables[0].Name = "unknown"
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Schema: "test", Name: "revisions"})
+	require.EqualError(t, err, `sql/migrate: connected database is not clean: found table "unknown"`)
+	// Multiple tables.
+	s.Tables = []*schema.Table{schema.NewTable("a"), schema.NewTable("revisions")}
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Schema: "test", Name: "revisions"})
+	require.EqualError(t, err, `sql/migrate: connected database is not clean: found multiple tables: 2`)
+}
+
+type mockInspector struct {
+	schema.Inspector
+	realm  *schema.Realm
+	schema *schema.Schema
+}
+
+func (m *mockInspector) InspectSchema(context.Context, string, *schema.InspectOptions) (*schema.Schema, error) {
+	if m.schema == nil {
+		return nil, &schema.NotExistError{}
+	}
+	return m.schema, nil
+}
+
+func (m *mockInspector) InspectRealm(context.Context, *schema.InspectRealmOption) (*schema.Realm, error) {
+	return m.realm, nil
+}
+
 type mockOpener struct {
 	*sql.DB
 	opened uint
diff --git a/sql/postgres/driver_test.go b/sql/postgres/driver_test.go
index 0c3a34bf..997c254a 100644
--- a/sql/postgres/driver_test.go
+++ b/sql/postgres/driver_test.go
@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"ariga.io/atlas/sql/internal/sqltest"
+	"ariga.io/atlas/sql/migrate"
 	"ariga.io/atlas/sql/schema"
 
 	"github.com/DATA-DOG/go-sqlmock"
@@ -101,3 +102,61 @@ func TestDriver_UnlockError(t *testing.T) {
 		require.Error(t, unlock())
 	})
 }
+
+func TestDriver_CheckClean(t *testing.T) {
+	s := schema.New("test")
+	drv := &Driver{Inspector: &mockInspector{schema: s}, schema: "test"}
+	// Empty schema.
+	err := drv.CheckClean(context.Background(), nil)
+	require.NoError(t, err)
+	// Revisions table found.
+	s.AddTables(schema.NewTable("revisions"))
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Name: "revisions", Schema: "test"})
+	require.NoError(t, err)
+	// Multiple tables.
+	s.Tables = []*schema.Table{schema.NewTable("a"), schema.NewTable("revisions")}
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Name: "revisions", Schema: "test"})
+	require.EqualError(t, err, `sql/migrate: connected database is not clean: found table "a" in schema "test"`)
+
+	r := schema.NewRealm()
+	drv.schema = ""
+	drv.Inspector = &mockInspector{realm: r}
+	// Empty realm.
+	err = drv.CheckClean(context.Background(), nil)
+	require.NoError(t, err)
+	// Revisions table found.
+	s.Tables = []*schema.Table{schema.NewTable("revisions").SetSchema(s)}
+	r.AddSchemas(s)
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Name: "revisions", Schema: "test"})
+	require.NoError(t, err)
+	// Unknown table.
+	s.Tables[0].Name = "unknown"
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Schema: "test", Name: "revisions"})
+	require.EqualError(t, err, `sql/migrate: connected database is not clean: found table "unknown" in schema "test"`)
+	// Multiple tables.
+	s.Tables = []*schema.Table{schema.NewTable("a"), schema.NewTable("revisions")}
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Schema: "test", Name: "revisions"})
+	require.EqualError(t, err, `sql/migrate: connected database is not clean: found 2 tables in schema "test"`)
+	// With auto created public schema.
+	s.Tables = []*schema.Table{schema.NewTable("revisions")}
+	r.AddSchemas(schema.New("public"))
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Schema: "test", Name: "revisions"})
+	require.NoError(t, err)
+}
+
+type mockInspector struct {
+	schema.Inspector
+	realm  *schema.Realm
+	schema *schema.Schema
+}
+
+func (m *mockInspector) InspectSchema(context.Context, string, *schema.InspectOptions) (*schema.Schema, error) {
+	if m.schema == nil {
+		return nil, &schema.NotExistError{}
+	}
+	return m.schema, nil
+}
+
+func (m *mockInspector) InspectRealm(context.Context, *schema.InspectRealmOption) (*schema.Realm, error) {
+	return m.realm, nil
+}
diff --git a/sql/sqlite/driver_test.go b/sql/sqlite/driver_test.go
index bc123cdf..3cb0235c 100644
--- a/sql/sqlite/driver_test.go
+++ b/sql/sqlite/driver_test.go
@@ -12,6 +12,9 @@ import (
 	"testing"
 	"time"
 
+	"ariga.io/atlas/sql/migrate"
+
+	"ariga.io/atlas/sql/schema"
 	"github.com/stretchr/testify/require"
 )
 
@@ -46,3 +49,38 @@ func TestDriver_LockAcquired(t *testing.T) {
 	// Acquiring a lock on another value works as well.
 	_, err = drv.Lock(context.Background(), "another", time.Second)
 }
+
+func TestDriver_CheckClean(t *testing.T) {
+	var (
+		r   = schema.NewRealm()
+		drv = &Driver{Inspector: &mockInspector{realm: r}}
+	)
+	// Empty realm.
+	err := drv.CheckClean(context.Background(), nil)
+	require.NoError(t, err)
+	// Empty schema.
+	r.AddSchemas(schema.New("main"))
+	err = drv.CheckClean(context.Background(), nil)
+	require.NoError(t, err)
+	// Schema with revisions table only.
+	r.Schemas[0].AddTables(schema.NewTable("revisions"))
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Name: "revisions"})
+	require.NoError(t, err)
+	// Unknown table.
+	r.Schemas[0].Tables[0].Name = "unknown"
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Name: "revisions"})
+	require.EqualError(t, err, `sql/migrate: connected database is not clean: found table "unknown"`)
+	// Multiple tables.
+	r.Schemas[0].Tables = []*schema.Table{schema.NewTable("a"), schema.NewTable("revisions")}
+	err = drv.CheckClean(context.Background(), &migrate.TableIdent{Schema: "test", Name: "revisions"})
+	require.EqualError(t, err, `sql/migrate: connected database is not clean: found multiple tables: 2`)
+}
+
+type mockInspector struct {
+	schema.Inspector
+	realm *schema.Realm
+}
+
+func (m *mockInspector) InspectRealm(context.Context, *schema.InspectRealmOption) (*schema.Realm, error) {
+	return m.realm, nil
+}
diff --git a/sql/sqlite/migrate_test.go b/sql/sqlite/migrate_test.go
index 86e60095..13918678 100644
--- a/sql/sqlite/migrate_test.go
+++ b/sql/sqlite/migrate_test.go
@@ -11,7 +11,6 @@ import (
 	"testing"
 
 	"ariga.io/atlas/sql/internal/sqltest"
-
 	"ariga.io/atlas/sql/migrate"
 	"ariga.io/atlas/sql/schema"