Repository: ariga/atlas
Pull request #1193 (2022-10-12 09:55:45)

doc/md: troubleshooting migration failures


Patch:
diff --git a/cmd/atlas/internal/cmdapi/cmdapi.go b/cmd/atlas/internal/cmdapi/cmdapi.go
index de6164bb..ec59d739 100644
--- a/cmd/atlas/internal/cmdapi/cmdapi.go
+++ b/cmd/atlas/internal/cmdapi/cmdapi.go
@@ -187,15 +187,18 @@ const (
 	flagExclude        = "exclude"
 	flagFile           = "file"
 	flagFrom           = "from"
+	flagFromShort      = "f"
 	flagGitBase        = "git-base"
 	flagGitDir         = "git-dir"
 	flagLatest         = "latest"
 	flagLog            = "log"
 	flagRevisionSchema = "revisions-schema"
 	flagSchema         = "schema"
+	flagSchemaShort    = "s"
 	flagTo             = "to"
 	flagTxMode         = "tx-mode"
 	flagURL            = "url"
+	flagURLShort       = "u"
 	flagVar            = "var"
 	flagQualifier      = "qualifier"
 )
@@ -268,7 +271,7 @@ func addFlagRevisionSchema(set *pflag.FlagSet, target *string) {
 func addFlagSchemas(set *pflag.FlagSet, target *[]string) {
 	set.StringSliceVarP(
 		target,
-		flagSchema, "s",
+		flagSchema, flagSchemaShort,
 		nil,
 		"set schema names",
 	)
@@ -276,7 +279,7 @@ func addFlagSchemas(set *pflag.FlagSet, target *[]string) {
 
 // addFlagURL adds a URL flag. If given, args[0] override the name, args[1] the shorthand.
 func addFlagURL(set *pflag.FlagSet, target *string, args ...string) {
-	name, short := flagURL, "u"
+	name, short := flagURL, flagURLShort
 	switch len(args) {
 	case 2:
 		short = args[1]
@@ -293,7 +296,7 @@ func addFlagURL(set *pflag.FlagSet, target *string, args ...string) {
 }
 
 func addFlagURLs(set *pflag.FlagSet, target *[]string, args ...string) {
-	name, short := flagURL, "u"
+	name, short := flagURL, flagURLShort
 	switch len(args) {
 	case 2:
 		short = args[1]
@@ -357,13 +360,16 @@ type (
 	stateReadCloser struct {
 		migrate.StateReader
 		io.Closer        // optional close function
-		Schema    string // in case we work on a single schema
+		schema    string // in case we work on a single schema
+		hcl       bool   // true if state was read from HCL files since in that case we always compare realms
 	}
 	// stateReaderConfig is given to stateReader.
 	stateReaderConfig struct {
-		urls    []string          // urls to create a migrate.StateReader from
-		dev     *sqlclient.Client // dev database connection
-		schemas []string          // schemas to work on
+		urls      []string          // urls to create a migrate.StateReader from
+		norm, dev *sqlclient.Client // database connections, while dev is considered a dev database, norm is not
+		schemas   []string          // schemas to work on
+		exclude   []string          // exclude flag values
+		vars      Vars
 	}
 )
 
@@ -383,10 +389,6 @@ func stateReader(ctx context.Context, config *stateReaderConfig) (*stateReadClos
 	switch scheme {
 	// "file" scheme is valid for both migration directory and HCL paths.
 	case "file":
-		// Replaying a migration directory or evaluating an HCL file requires a dev connection.
-		if config.dev == nil {
-			return nil, errors.New("--dev-url cannot be empty")
-		}
 		if len(config.urls) > 1 {
 			// Consider urls being HCL paths.
 			return hclStateReader(ctx, config, parsed)
@@ -423,6 +425,10 @@ func stateReader(ctx context.Context, config *stateReaderConfig) (*stateReadClos
 		case hcl:
 			return hclStateReader(ctx, config, parsed)
 		case sql:
+			// Replaying a migration directory requires a dev connection.
+			if config.dev == nil {
+				return nil, errors.New("--dev-url cannot be empty")
+			}
 			dir, err := dirURL(parsed[0], false)
 			if err != nil {
 				return nil, err
@@ -431,18 +437,25 @@ func stateReader(ctx context.Context, config *stateReaderConfig) (*stateReadClos
 			if err != nil {
 				return nil, err
 			}
+			var opts []migrate.ReplayOption
+			if v := parsed[0].Query().Get("version"); v != "" {
+				opts = append(opts, migrate.ReplayToVersion(v))
+			}
 			sr, err := ex.Replay(ctx, func() migrate.StateReader {
 				if config.dev.URL.Schema != "" {
 					return migrate.SchemaConn(config.dev, "", nil)
 				}
-				return migrate.RealmConn(config.dev, nil)
-			}())
+				return migrate.RealmConn(config.dev, &schema.InspectRealmOption{
+					Schemas: config.schemas,
+					Exclude: config.exclude,
+				})
+			}(), opts...)
 			if err != nil && !errors.Is(err, migrate.ErrNoPendingFiles) {
 				return nil, fmt.Errorf("replaying the migration directory: %w", err)
 			}
 			return &stateReadCloser{
 				StateReader: migrate.Realm(sr),
-				Schema:      config.dev.URL.Schema,
+				schema:      config.dev.URL.Schema,
 			}, nil
 		default:
 			return nil, fmt.Errorf("%q contains neither SQL nor HCL files", path)
@@ -456,30 +469,42 @@ func stateReader(ctx context.Context, config *stateReaderConfig) (*stateReadClos
 		var sr migrate.StateReader
 		switch c.URL.Schema {
 		case "":
-			sr = migrate.RealmConn(c.Driver, nil)
+			sr = migrate.RealmConn(c.Driver, &schema.InspectRealmOption{
+				Schemas: config.schemas,
+				Exclude: config.exclude,
+			})
 		default:
-			sr = migrate.SchemaConn(c.Driver, c.URL.Schema, nil)
+			sr = migrate.SchemaConn(c.Driver, c.URL.Schema, &schema.InspectOptions{Exclude: config.exclude})
 		}
 		return &stateReadCloser{
 			StateReader: sr,
 			Closer:      c,
-			Schema:      c.URL.Schema,
+			schema:      c.URL.Schema,
 		}, nil
 	}
 }
 
 // hclStateReadr returns a migrate.StateReader that reads the state from the given HCL paths urls.
 func hclStateReader(ctx context.Context, config *stateReaderConfig, urls []*url.URL) (*stateReadCloser, error) {
+	var client *sqlclient.Client
+	switch {
+	case config.dev != nil:
+		client = config.dev
+	case config.norm != nil:
+		client = config.norm
+	default:
+		return nil, errors.New("no database connection available")
+	}
 	paths := make([]string, len(urls))
 	for i, u := range urls {
-		paths[i] = u.Path
+		paths[i] = filepath.Join(u.Host, u.Path)
 	}
 	parser, err := parseHCLPaths(paths...)
 	if err != nil {
 		return nil, err
 	}
 	realm := &schema.Realm{}
-	if err := config.dev.Eval(parser, realm, nil); err != nil {
+	if err := client.Eval(parser, realm, config.vars); err != nil {
 		return nil, err
 	}
 	if len(config.schemas) > 0 {
@@ -497,22 +522,19 @@ func hclStateReader(ctx context.Context, config *stateReaderConfig, urls []*url.
 	// In case the dev connection is bound to a specific schema, we require the
 	// desired schema to contain only one schema. Thus, executing diff will be
 	// done on the content of these two schema and not the whole realm.
-	if config.dev.URL.Schema != "" && len(realm.Schemas) > 1 {
+	if client.URL.Schema != "" && len(realm.Schemas) > 1 {
 		return nil, fmt.Errorf(
 			"cannot use HCL with more than 1 schema when dev-url is limited to schema %q",
 			config.dev.URL.Schema,
 		)
 	}
-	if norm, ok := config.dev.Driver.(schema.Normalizer); ok && len(realm.Schemas) > 0 {
+	if norm, ok := client.Driver.(schema.Normalizer); ok && len(realm.Schemas) > 0 {
 		realm, err = norm.NormalizeRealm(ctx, realm)
 		if err != nil {
 			return nil, err
 		}
 	}
-	t := &stateReadCloser{StateReader: migrate.Realm(realm)}
-	if len(realm.Schemas) == 1 {
-		t.Schema = realm.Schemas[0].Name
-	}
+	t := &stateReadCloser{StateReader: migrate.Realm(realm), hcl: true}
 	return t, nil
 }
 
diff --git a/cmd/atlas/internal/cmdapi/schema.go b/cmd/atlas/internal/cmdapi/schema.go
index 619dc4d4..6aa59b88 100644
--- a/cmd/atlas/internal/cmdapi/schema.go
+++ b/cmd/atlas/internal/cmdapi/schema.go
@@ -6,6 +6,8 @@ package cmdapi
 
 import (
 	"bytes"
+	"context"
+	"errors"
 	"fmt"
 	"io/fs"
 	"os"
@@ -45,14 +47,15 @@ func schemaCmd() *cobra.Command {
 }
 
 type schemaApplyFlags struct {
-	URL         string   // URL of database to apply the changes on.
-	DevURL      string   // URL of the dev database.
-	Paths       []string // Paths to HCL files.
-	Schemas     []string // Schemas to take into account when diffing.
-	Exclude     []string // List of glob patterns used to filter resources from applying (see schema.InspectOptions).
-	DryRun      bool     // Only show SQL on screen instead of applying it.
-	AutoApprove bool     // Don't prompt for approval before applying SQL.
-	DSN         string   // Deprecated: DSN is an alias for URL.
+	url         string   // URL of database to apply the changes on.
+	devURL      string   // URL of the dev database.
+	paths       []string // Paths to HCL files.
+	toURLs      []string // URLs of the desired state.
+	schemas     []string // Schemas to take into account when diffing.
+	exclude     []string // List of glob patterns used to filter resources from applying (see schema.InspectOptions).
+	dryRun      bool     // Only show SQL on screen instead of applying it.
+	autoApprove bool     // Don't prompt for approval before applying SQL.
+	dsn         string   // Deprecated: DSN is an alias for URL.
 }
 
 // schemaApplyCmd represents the 'atlas schema apply' subcommand.
@@ -67,90 +70,95 @@ func schemaApplyCmd() *cobra.Command {
 database to the state described in the provided Atlas schema. Before running the
 migration, Atlas will print the migration plan and prompt the user for approval.
 
-The schema is provided by one or more paths (to a file or directory) using the "-f" flag:
-  atlas schema apply -u URL -f file1.hcl -f file2.hcl
-  atlas schema apply -u URL -f schema/ -f override.hcl
+The schema is provided by one or more URLs (to a HCL file or 
+directory, database or migration directory) using the "--to, -t" flag:
+  atlas schema apply -u URL --to file://file1.hcl --to file://file2.hcl
+  atlas schema apply -u URL --to file://schema/ --to file://override.hcl
 
-As a convenience, schemas may also be provided via an environment definition in
+As a convenience, schema URLs may also be provided via an environment definition in
 the project file (see: https://atlasgo.io/cli/projects).
 
 If run with the "--dry-run" flag, atlas will exit after printing out the planned
 migration.`,
-			Example: `  atlas schema apply -u "mysql://user:pass@localhost/dbname" -f atlas.hcl
-  atlas schema apply -u "mysql://localhost" -f schema.hcl --schema prod --schema staging
-  atlas schema apply -u "mysql://user:pass@localhost:3306/dbname" -f schema.hcl --dry-run
-  atlas schema apply -u "mariadb://user:pass@localhost:3306/dbname" -f schema.hcl
-  atlas schema apply --url "postgres://user:pass@host:port/dbname?sslmode=disable" -f schema.hcl
-  atlas schema apply -u "sqlite://file:ex1.db?_fk=1" -f schema.hcl`,
-			PreRunE: schemaFlagsFromEnv,
+			Example: `  atlas schema apply -u "mysql://user:pass@localhost/dbname" --to file://atlas.hcl
+  atlas schema apply -u "mysql://localhost" --to file://schema.hcl --schema prod --schema staging
+  atlas schema apply -u "mysql://user:pass@localhost:3306/dbname" --to file://schema.hcl --dry-run
+  atlas schema apply -u "mariadb://user:pass@localhost:3306/dbname" --to file://schema.hcl
+  atlas schema apply --url "postgres://user:pass@host:port/dbname?sslmode=disable" --to file://schema.hcl
+  atlas schema apply -u "sqlite://file:ex1.db?_fk=1" --to file://schema.hcl`,
+			PreRunE: func(cmd *cobra.Command, args []string) error {
+				if err := schemaFlagsFromEnv(cmd, args); err != nil {
+					return err
+				}
+				if len(flags.paths) == 0 && len(flags.toURLs) == 0 {
+					return errors.New(`one of flag(s) "file" or "to" is required`)
+				}
+				// If the old -f flag is given convert them to the URL format. If both are given, cobra would throw an
+				// error since they are marked as mutually exclusive.
+				for _, p := range flags.paths {
+					flags.toURLs = append(flags.toURLs, "file://"+p)
+				}
+				return nil
+			},
 			RunE: func(cmd *cobra.Command, args []string) error {
 				return schemaApplyRun(cmd, args, flags)
 			},
 		}
 	)
 	cmd.Flags().SortFlags = false
-	cmd.Flags().StringSliceVarP(&flags.Paths, flagFile, "f", nil, "[paths...] file or directory containing the HCL files")
-	addFlagURL(cmd.Flags(), &flags.URL)
-	addFlagExclude(cmd.Flags(), &flags.Exclude)
-	addFlagSchemas(cmd.Flags(), &flags.Schemas)
-	addFlagDevURL(cmd.Flags(), &flags.DevURL)
-	addFlagDryRun(cmd.Flags(), &flags.DryRun)
-	addFlagAutoApprove(cmd.Flags(), &flags.AutoApprove)
-	addFlagDSN(cmd.Flags(), &flags.DSN)
+	cmd.Flags().StringSliceVarP(&flags.paths, flagFile, "f", nil, "[paths...] file or directory containing the HCL files")
+	addFlagURL(cmd.Flags(), &flags.url)
+	addFlagURLs(cmd.Flags(), &flags.toURLs, flagTo, "")
+	addFlagExclude(cmd.Flags(), &flags.exclude)
+	addFlagSchemas(cmd.Flags(), &flags.schemas)
+	addFlagDevURL(cmd.Flags(), &flags.devURL)
+	addFlagDryRun(cmd.Flags(), &flags.dryRun)
+	addFlagAutoApprove(cmd.Flags(), &flags.autoApprove)
+	addFlagDSN(cmd.Flags(), &flags.dsn)
 	cobra.CheckErr(cmd.MarkFlagRequired(flagURL))
-	cobra.CheckErr(cmd.MarkFlagRequired(flagFile))
+	cmd.MarkFlagsMutuallyExclusive(flagFile, flagTo)
 	return cmd
 }
 
 func schemaApplyRun(cmd *cobra.Command, _ []string, flags schemaApplyFlags) error {
-	client, err := sqlclient.Open(cmd.Context(), flags.URL)
-	if err != nil {
-		return err
-	}
-	defer client.Close()
-	schemas, ctx := flags.Schemas, cmd.Context()
-	if client.URL.Schema != "" {
-		schemas = append(schemas, client.URL.Schema)
+	var (
+		ctx = cmd.Context()
+		err error
+	)
+	var dev *sqlclient.Client
+	if flags.devURL != "" {
+		dev, err = sqlclient.Open(ctx, flags.devURL)
+		if err != nil {
+			return err
+		}
+		defer dev.Close()
 	}
-	realm, err := client.InspectRealm(ctx, &schema.InspectRealmOption{
-		Schemas: schemas,
-		Exclude: flags.Exclude,
+	from, err := stateReader(ctx, &stateReaderConfig{
+		urls:    []string{flags.url},
+		schemas: flags.schemas,
+		exclude: flags.exclude,
 	})
 	if err != nil {
 		return err
 	}
-	desired := &schema.Realm{}
-	parsed, err := parseHCLPaths(flags.Paths...)
+	defer from.Close()
+	client, ok := from.Closer.(*sqlclient.Client)
+	if !ok {
+		return errors.New("--url must be a database connection")
+	}
+	to, err := stateReader(ctx, &stateReaderConfig{
+		urls:    flags.toURLs,
+		dev:     dev,
+		norm:    client,
+		schemas: flags.schemas,
+		exclude: flags.exclude,
+		vars:    GlobalFlags.Vars,
+	})
 	if err != nil {
 		return err
 	}
-	if err := client.Eval(parsed, desired, GlobalFlags.Vars); err != nil {
-		return err
-	}
-	if len(schemas) > 0 {
-		// Validate all schemas in file were selected by user.
-		sm := make(map[string]bool, len(schemas))
-		for _, s := range schemas {
-			sm[s] = true
-		}
-		for _, s := range desired.Schemas {
-			if !sm[s.Name] {
-				return fmt.Errorf("schema %q was not selected %q, all schemas defined in file must be selected", s.Name, schemas)
-			}
-		}
-	}
-	if _, ok := client.Driver.(schema.Normalizer); ok && flags.DevURL != "" {
-		dev, err := sqlclient.Open(ctx, flags.DevURL)
-		if err != nil {
-			return err
-		}
-		defer dev.Close()
-		desired, err = dev.Driver.(schema.Normalizer).NormalizeRealm(ctx, desired)
-		if err != nil {
-			return err
-		}
-	}
-	changes, err := client.RealmDiff(realm, desired)
+	defer to.Close()
+	changes, err := computeDiff(ctx, client, from, to)
 	if err != nil {
 		return err
 	}
@@ -161,7 +169,7 @@ func schemaApplyRun(cmd *cobra.Command, _ []string, flags schemaApplyFlags) erro
 	if err := summary(cmd, client, changes); err != nil {
 		return err
 	}
-	if !flags.DryRun && (flags.AutoApprove || promptUser()) {
+	if !flags.dryRun && (flags.autoApprove || promptUser()) {
 		if err := client.ApplyChanges(ctx, changes); err != nil {
 			return err
 		}
@@ -243,9 +251,11 @@ func schemaCleanRun(cmd *cobra.Command, _ []string, flags schemeCleanFlags) erro
 }
 
 type schemaDiffFlags struct {
-	fromURL string // URL of the current state.
-	toURL   string // URL of the desired state.
-	devURL  string // URL of a dev database.
+	fromURL []string
+	toURL   []string
+	devURL  string
+	schemas []string
+	exclude []string
 }
 
 // schemaDiffCmd represents the 'atlas schema diff' subcommand.
@@ -268,9 +278,11 @@ The database states can be read from a connected database, an HCL project or a m
 		}
 	)
 	cmd.Flags().SortFlags = false
-	addFlagURL(cmd.Flags(), &flags.fromURL, flagFrom, "")
-	addFlagURL(cmd.Flags(), &flags.toURL, flagTo, "")
+	addFlagURLs(cmd.Flags(), &flags.fromURL, flagFrom, flagFromShort)
+	addFlagURLs(cmd.Flags(), &flags.toURL, flagTo, "")
 	addFlagDevURL(cmd.Flags(), &flags.devURL)
+	addFlagSchemas(cmd.Flags(), &flags.schemas)
+	addFlagExclude(cmd.Flags(), &flags.exclude)
 	cobra.CheckErr(cmd.MarkFlagRequired(flagFrom))
 	cobra.CheckErr(cmd.MarkFlagRequired(flagTo))
 	return cmd
@@ -290,12 +302,24 @@ func schemaDiffRun(cmd *cobra.Command, _ []string, flags schemaDiffFlags) error
 		}
 		defer c.Close()
 	}
-	from, err := stateReader(ctx, &stateReaderConfig{urls: []string{flags.fromURL}, dev: c})
+	from, err := stateReader(ctx, &stateReaderConfig{
+		urls:    flags.fromURL,
+		dev:     c,
+		vars:    GlobalFlags.Vars,
+		schemas: flags.schemas,
+		exclude: flags.exclude,
+	})
 	if err != nil {
 		return err
 	}
 	defer from.Close()
-	to, err := stateReader(ctx, &stateReaderConfig{urls: []string{flags.toURL}, dev: c})
+	to, err := stateReader(ctx, &stateReaderConfig{
+		urls:    flags.toURL,
+		dev:     c,
+		vars:    GlobalFlags.Vars,
+		schemas: flags.schemas,
+		exclude: flags.exclude,
+	})
 	if err != nil {
 		return err
 	}
@@ -305,35 +329,10 @@ func schemaDiffRun(cmd *cobra.Command, _ []string, flags schemaDiffFlags) error
 		// an error already. If we land in this case, we can assume both states are database connections.
 		c = to.Closer.(*sqlclient.Client)
 	}
-	current, err := from.ReadState(ctx)
+	diff, err := computeDiff(ctx, c, from, to)
 	if err != nil {
 		return err
 	}
-	desired, err := to.ReadState(ctx)
-	if err != nil {
-		return err
-	}
-	var diff []schema.Change
-	switch {
-	// compare realm
-	case from.Schema == "" && to.Schema == "":
-		diff, err = c.RealmDiff(current, desired)
-		if err != nil {
-			return err
-		}
-	case from.Schema == "":
-		return fmt.Errorf("cannot diff schema %q with a database connection", from.Schema)
-	case to.Schema == "":
-		return fmt.Errorf("cannot diff database connection with a schema %q", to.Schema)
-	default:
-		// SchemaDiff checks for name equality which is irrelevant in the case
-		// the user wants to compare their contents, reset them to allow the comparison.
-		current.Schemas[0].Name, desired.Schemas[0].Name = "", ""
-		diff, err = c.SchemaDiff(current.Schemas[0], desired.Schemas[0])
-		if err != nil {
-			return err
-		}
-	}
 	p, err := c.PlanChanges(ctx, "plan", diff)
 	if err != nil {
 		return err
@@ -351,10 +350,10 @@ func schemaDiffRun(cmd *cobra.Command, _ []string, flags schemaDiffFlags) error
 }
 
 type schemaInspectFlags struct {
-	URL     string   // URL of database to apply the changes on.
-	Schemas []string // Schemas to take into account when diffing.
-	Exclude []string // List of glob patterns used to filter resources from applying (see schema.InspectOptions).
-	DSN     string   // Deprecated: DSN is an alias for URL.
+	url     string   // URL of database to apply the changes on.
+	schemas []string // Schemas to take into account when diffing.
+	exclude []string // List of glob patterns used to filter resources from applying (see schema.InspectOptions).
+	dsn     string   // Deprecated: DSN is an alias for URL.
 }
 
 // schemaInspectCmd represents the 'atlas schema inspect' subcommand.
@@ -388,27 +387,27 @@ flag.
 		}
 	)
 	cmd.Flags().SortFlags = false
-	addFlagURL(cmd.Flags(), &flags.URL)
-	addFlagSchemas(cmd.Flags(), &flags.Schemas)
-	addFlagExclude(cmd.Flags(), &flags.Exclude)
-	addFlagDSN(cmd.Flags(), &flags.DSN)
+	addFlagURL(cmd.Flags(), &flags.url)
+	addFlagSchemas(cmd.Flags(), &flags.schemas)
+	addFlagExclude(cmd.Flags(), &flags.exclude)
+	addFlagDSN(cmd.Flags(), &flags.dsn)
 	cobra.CheckErr(cmd.MarkFlagRequired(flagURL))
 	return cmd
 }
 
 func schemaInspectRun(cmd *cobra.Command, _ []string, flags schemaInspectFlags) error {
-	client, err := sqlclient.Open(cmd.Context(), flags.URL)
+	client, err := sqlclient.Open(cmd.Context(), flags.url)
 	if err != nil {
 		return err
 	}
 	defer client.Close()
-	schemas := flags.Schemas
+	schemas := flags.schemas
 	if client.URL.Schema != "" {
 		schemas = append(schemas, client.URL.Schema)
 	}
 	s, err := client.InspectRealm(cmd.Context(), &schema.InspectRealmOption{
 		Schemas: schemas,
-		Exclude: flags.Exclude,
+		Exclude: flags.exclude,
 	})
 	if err != nil {
 		return err
@@ -520,6 +519,39 @@ func schemaFlagsFromEnv(cmd *cobra.Command, _ []string) error {
 	return nil
 }
 
+func computeDiff(ctx context.Context, differ *sqlclient.Client, from, to *stateReadCloser) ([]schema.Change, error) {
+	current, err := from.ReadState(ctx)
+	if err != nil {
+		return nil, err
+	}
+	desired, err := to.ReadState(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var diff []schema.Change
+	switch {
+	// Compare realm if the desired state is an HCL file or both connections are not bound to a schema.
+	case to.hcl, from.schema == "" && to.schema == "":
+		diff, err = differ.RealmDiff(current, desired)
+		if err != nil {
+			return nil, err
+		}
+	case from.schema == "":
+		return nil, fmt.Errorf("cannot diff schema %q with a database connection", from.schema)
+	case to.schema == "":
+		return nil, fmt.Errorf("cannot diff database connection with a schema %q", to.schema)
+	default:
+		// SchemaDiff checks for name equality which is irrelevant in the case
+		// the user wants to compare their contents, reset them to allow the comparison.
+		current.Schemas[0].Name, desired.Schemas[0].Name = "", ""
+		diff, err = differ.SchemaDiff(current.Schemas[0], desired.Schemas[0])
+		if err != nil {
+			return nil, err
+		}
+	}
+	return diff, nil
+}
+
 func summary(cmd *cobra.Command, drv migrate.Driver, changes []schema.Change) error {
 	p, err := drv.PlanChanges(cmd.Context(), "", changes)
 	if err != nil {
diff --git a/doc/md/reference.md b/doc/md/reference.md
index f24ba8b4..f83a8316 100644
--- a/doc/md/reference.md
+++ b/doc/md/reference.md
@@ -360,11 +360,12 @@ atlas schema apply [flags]
 database to the state described in the provided Atlas schema. Before running the
 migration, Atlas will print the migration plan and prompt the user for approval.
 
-The schema is provided by one or more paths (to a file or directory) using the "-f" flag:
-  atlas schema apply -u URL -f file1.hcl -f file2.hcl
-  atlas schema apply -u URL -f schema/ -f override.hcl
+The schema is provided by one or more URLs (to a HCL file or 
+directory, database or migration directory) using the "--to, -t" flag:
+  atlas schema apply -u URL --to file://file1.hcl --to file://file2.hcl
+  atlas schema apply -u URL --to file://schema/ --to file://override.hcl
 
-As a convenience, schemas may also be provided via an environment definition in
+As a convenience, schema URLs may also be provided via an environment definition in
 the project file (see: https://atlasgo.io/cli/projects).
 
 If run with the "--dry-run" flag, atlas will exit after printing out the planned
@@ -373,17 +374,18 @@ migration.
 #### Example
 
 ```
-  atlas schema apply -u "mysql://user:pass@localhost/dbname" -f atlas.hcl
-  atlas schema apply -u "mysql://localhost" -f schema.hcl --schema prod --schema staging
-  atlas schema apply -u "mysql://user:pass@localhost:3306/dbname" -f schema.hcl --dry-run
-  atlas schema apply -u "mariadb://user:pass@localhost:3306/dbname" -f schema.hcl
-  atlas schema apply --url "postgres://user:pass@host:port/dbname?sslmode=disable" -f schema.hcl
-  atlas schema apply -u "sqlite://file:ex1.db?_fk=1" -f schema.hcl
+  atlas schema apply -u "mysql://user:pass@localhost/dbname" --to file://atlas.hcl
+  atlas schema apply -u "mysql://localhost" --to file://schema.hcl --schema prod --schema staging
+  atlas schema apply -u "mysql://user:pass@localhost:3306/dbname" --to file://schema.hcl --dry-run
+  atlas schema apply -u "mariadb://user:pass@localhost:3306/dbname" --to file://schema.hcl
+  atlas schema apply --url "postgres://user:pass@host:port/dbname?sslmode=disable" --to file://schema.hcl
+  atlas schema apply -u "sqlite://file:ex1.db?_fk=1" --to file://schema.hcl
 ```
 #### Flags
 ```
   -f, --file strings      [paths...] file or directory containing the HCL files
   -u, --url string        [driver://username:password@address/dbname?param=value] select a resource using the URL format
+      --to strings        [driver://username:password@address/dbname?param=value] select a resource using the URL format
       --exclude strings   list of glob patterns used to filter resources from applying
   -s, --schema strings    set schema names
       --dev-url string    [driver://username:password@address/dbname?param=value] select a dev database using the URL format
@@ -444,9 +446,11 @@ The database states can be read from a connected database, an HCL project or a m
 ```
 #### Flags
 ```
-      --from string      [driver://username:password@address/dbname?param=value] select a resource using the URL format
-      --to string        [driver://username:password@address/dbname?param=value] select a resource using the URL format
-      --dev-url string   [driver://username:password@address/dbname?param=value] select a dev database using the URL format
+  -f, --from strings      [driver://username:password@address/dbname?param=value] select a resource using the URL format
+      --to strings        [driver://username:password@address/dbname?param=value] select a resource using the URL format
+      --dev-url string    [driver://username:password@address/dbname?param=value] select a dev database using the URL format
+  -s, --schema strings    set schema names
+      --exclude strings   list of glob patterns used to filter resources from applying
 
 ```
 
diff --git a/doc/md/versioned/apply.mdx b/doc/md/versioned/apply.mdx
index a65cba76..92de0272 100644
--- a/doc/md/versioned/apply.mdx
+++ b/doc/md/versioned/apply.mdx
@@ -110,6 +110,11 @@ If you want to check what exactly Atlas would do when attempting a migration exe
 flag:
 * `--dry-run` to not execute any SQL but print it on the screen.
 
+### Migration status
+
+In addition to the `--dry-run` flag Atlas also provides the `atlas migrate status` command, that provides in-depth
+information about the migration status of the connected database.
+
 ### Examples
 
 First time apply with baseline on production environment:
@@ -148,3 +153,15 @@ atlas migrate apply \
   --url "mysql://root:pass@remote:3306/my_database" \
   --tx-mode "none"
 ```
+
+Show information about the migration status of a deployment:
+```shell
+atlas migrate status \
+  --url "mysql://root:pass@remote:3306/my_database" \
+  --dir "file://custom/path/to/dir" \
+  --revisions-schema "atlas_migration_history"
+```
+
+### Reference
+
+[CLI Command Reference](/cli-reference#atlas-migrate-apply)
diff --git a/doc/md/versioned/troubleshoot.mdx b/doc/md/versioned/troubleshoot.mdx
new file mode 100644
index 00000000..b75b85c9
--- /dev/null
+++ b/doc/md/versioned/troubleshoot.mdx
@@ -0,0 +1,295 @@
+---
+id: troubleshoot
+slug: /versioned/troubleshoot
+title: Migration Troubleshooting
+---
+
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+In some cases, applying a migration may fail. This document aims to provide
+general support for troubleshooting migration failures.
+
+## Retrieve Migration Status
+
+Atlas provides the `atlas migrate status` command to retrieve information about the migration status of a database. You
+can either use this command or have a look at the `atlas migrate apply` output to gather knowledge about the issue that
+caused the migration failure. An example output can be seen below.
+
+:::info
+
+Note, that `atlas migrate status` does not show information about migration attempts that were rolled back. As long as a migration failure
+is wrapped with a transaction by a database with transactional DDL support, Atlas will rollback entries to the revision
+table within the transaction.
+
+:::
+
+<Tabs
+defaultValue="apply_stdout"
+values={[
+{label: 'atlas migrate apply (STDOUT)', value: 'apply_stdout'},
+{label: 'atlas migrate apply (STDERR)', value: 'apply_stderr'},
+{label: 'atlas migrate status', value: 'status'},
+]}>
+<TabItem value="apply_stdout">
+
+```text
+$ atlas --env local migrate apply 1 --tx-mode none
+Migrating to version 2 from 1 (1 migrations in total):
+
+  -- migrating version 2
+    -> INSERT INTO `users` (`id`, `name`) VALUES (1, 'masseelch'), (2, 'rotemtam'), (3, 'a8m');
+    -> INSERT INTO `groups` (`id`, `name`) VALUES (1, 'Founders'), (2, 'Senior Engineers'), (3, 'Junior Engineers');
+    -> INSERT INTO `user_groups` (`user_id`, `group_id`) VALUES
+    -- Founders
+    (2, 1), (3, 1),
+    -- Seniors
+    (1, 2), (2, 2), (3, 2),
+    -- Constraint error (adding masseelch twice to seniors)
+    (1, 2);
+// highlight-next-line-error-message
+     Error 1062: Duplicate entry '1-2' for key 'user_groups.PRIMARY'
+
+  -------------------------
+  -- 2.521007ms
+  -- 0 migrations ok (1 with errors)
+  -- 2 sql statements ok (1 with errors)
+
+// highlight-next-line-error-message
+Error: Execution had errors:
+// highlight-next-line-error-message
+Error 1062: Duplicate entry '1-2' for key 'user_groups.PRIMARY'
+```
+
+</TabItem>
+<TabItem value="apply_stderr">
+
+```text
+$ atlas --env local migrate apply 1 --tx-mode none
+Error: sql/migrate: execute: executing statement "INSERT INTO `user_groups` (`user_id`, `group_id`) VALUES
+    -- Founders\n    (2, 1), (3, 1),
+    -- Seniors\n    (1, 2), (2, 2), (3, 2),
+    -- Constraint error (adding masseelch twice to seniors)
+    (1, 2);" from version "2": Error 1062: Duplicate entry '1-2' for key 'user_groups.PRIMARY'
+```
+
+</TabItem>
+<TabItem value="status">
+
+```text
+$ atlas --env local migrate status
+Migration Status: PENDING
+  -- Current Version: 2 (2 statements applied)
+  -- Next Version:    2 (1 statements left)
+  -- Executed Files:  2 (last one partially)
+  -- Pending Files:   2
+
+Last migration attempt had errors:
+  -- SQL:   INSERT INTO `user_groups` (`user_id`, `group_id`) VALUES     -- Founders     (2, 1), (3, 1),     -- Seniors     (1, 2), (2, 2), (3, 2),     -- Constraint error (adding masseelch twice to seniors)     (1, 2);
+// highlight-next-line-error-message
+  -- ERROR: Error 1062: Duplicate entry '1-2' for key 'user_groups.PRIMARY'
+```
+
+</TabItem>
+</Tabs>
+
+## Fixing Migration Failures
+
+In our experience, failures are commonly caused by one of the following issues:
+
+- **Syntax Error**: A SQL statement contained a syntax error. Since Atlas provides a way to [lint](/versioned/lint) a
+migration directory, this type of failure can be automatically prevented by running the linter before applying the
+migration against the correct type of database (e.g. lint against a MySQL database if your migrations were written for MySQL)
+- **Schema Dependent Change**: A SQL statement was incorrect in schema context, e.g. adding a table that already
+exists. This might happen, for example, in cases where someone manipulated the database schema manually.
+- **Data-dependent Change**: A SQL statement contained a data-dependent change. For example, in cases where your migration adds a constraint
+like `NOT NULL` and the target column contains `NULL` values, the migration will throw an error. Although Atlas' linter
+will warn the user if such a change is detected, this is still a possible scenario that could break migration execution.
+- **Connection Loss**: The connection to the database could be interrupted for a number of reasons, e.g. due to a
+network failure or database crash.
+## Handling the different types of failures
+### Connection Loss Failures
+
+Retry migration execution. Atlas stores information about applied statements alongside the revision information, and
+therefore knows  where to proceed execution, even for partially applied / not rolled back migration files. For very rare
+cases, where when working without a transaction (or after an implicit commit) a connection loss occurs between a SQL
+statement and the entry in Atlas schema history table, you might need to manually revert the last applied statement
+because in those cases Atlas will continue applying one statement early.
+
+### Syntax Errors
+
+Fix the migration file and retry migration execution. Atlas will continue the execution starting with the statement
+following the last successfully applied one.
+
+### DDL Semantics
+
+Creating or editing schema resources can result in migration failure, e.g. if you are trying to create a table that
+already exists or to modify a table that does not exist. The cause for this is either an incorrect migration file or a
+schema drift, e.g. if someone dropped, created or modified a resource.
+
+#### Option 1: Incorrect Migration File
+
+In case the migration script is incorrect, simply fix the migration file and re-apply the migration. Do not forget to
+update the migration directory checksum afterwards by running:
+
+```shell
+atlas migrate hash
+```
+For cases with multiple deployments or no (fast) editorial access to the migration file you have to fix the database
+state to match whatever the migration file is expecting. You can either do this by hand or use one of the `atlas schema`
+commands. See [Option 2](#option-2-schema-drift) for more information.
+
+Once the migration/database is fixed, you can retry migration execution.
+
+#### Option 2: Schema Drift
+
+In case of a schema drift, we still consider the migration files the source of truth. Editing their contents to match the
+managed database will most likely break other deployments or the local dev. The solution is to fix the database schema.
+This can be done manually or with the help of the `atlas schema apply` / `atlas schema diff` commands. Both commands can
+read a database state from a migration directory and compare it with the current state of the target database. The only
+difference is that `atlas schema apply` can automatically apply the changes to the database, while
+`atlas schema diff` will print the SQL for you to check and run manually. You can utilize the `atlas schema status`
+command to get information about the currently applied migration version.
+
+##### atlas schema diff
+
+<Tabs
+defaultValue="dir"
+values={[
+{label: 'Migration Directory', value: 'dir'},
+{label: 'HCL File', value: 'hcl'},
+{label: 'Another Database', value: 'db'},
+]}>
+<TabItem value="dir">
+
+Compare the state of the `from` database (the one with the schema drift), to the desired state given in the migration
+directory. Note the query parameter `version` in the connection string to the migration directory, specifying the
+version to consider the current state of the schema.
+
+```shell
+atlas schema diff \
+  --from "driver://user:pass@host:port/db" \
+  --to "file:///path/to/migrations/dir?version=myTargetVersion" \
+  --dev-url "driver://user:pass@host:port/dev_db"
+```
+
+</TabItem>
+<TabItem value="hcl">
+
+Compare the state of the `from` database (the one with the schema drift), to the desired state given in one or more HCL
+files.
+
+```shell
+atlas schema diff \
+  --from "driver://user:pass@host:port/db" \
+  --to "file:///path/to/schema_1.hcl" \
+  --to "file:///path/to/schema_2.hcl" \
+  --dev-url "driver://user:pass@host:port/dev_db"
+```
+
+</TabItem>
+<TabItem value="db">
+
+Compare the state of the `from` database (the one with the schema drift), to the desired state given by another database
+(e.g. a local dev copy of the schema).
+
+```shell
+atlas schema diff \
+  --from "driver://user:pass@host:port/current" \
+  --to "driver://user:pass@host:port/desired"
+```
+
+</TabItem>
+</Tabs>
+
+Once you apply the SQL shown by `atlas migrate diff` onto your database, the schema drift should have vanished, and you
+can retry migration applying.
+
+#### atlas schema apply
+
+Instead of only showing the SQL on the screen, you can immediately let Atlas apply it (Atlas will prompt for
+confirmation first).
+
+<Tabs
+defaultValue="dir"
+values={[
+{label: 'Migration Directory', value: 'dir'},
+{label: 'HCL File', value: 'hcl'},
+{label: 'Another Database', value: 'db'},
+]}>
+<TabItem value="dir">
+
+Migrate the state of the `url` database (the one with the schema drift), to the desired state given in the migration
+directory. Note the query parameter `version` in the connection string to the migration directory, specifying the
+version to consider the current state of the schema.
+
+```shell
+atlas schema apply \
+  --url "driver://user:pass@host:port/db" \
+  --to "file:///path/to/migrations/dir?format=atlas&version=myTargetVersion" \
+  --dev-url "driver://user:pass@host:port/dev_db"
+```
+
+</TabItem>
+<TabItem value="hcl">
+
+Compare the state of the `url` database (the one with the schema drift), to the desired state given in one or more HCL
+files.
+
+```shell
+atlas schema apply \
+  --url "driver://user:pass@host:port/db" \
+  --to "file:///path/to/schema_1.hcl" \
+  --to "file:///path/to/schema_2.hcl" \
+  --dev-url "driver://user:pass@host:port/dev_db"
+```
+
+</TabItem>
+<TabItem value="db">
+
+Compare the state of the `url` database (the one with the schema drift), to the desired state given by another database
+(e.g. a local dev copy of the schema).
+
+```shell
+atlas schema diff \
+  --from "driver://user:pass@host:port/current" \
+  --to "driver://user:pass@host:port/desired"
+```
+
+</TabItem>
+</Tabs>
+
+### Data Dependent Failures
+
+If the issue is caused by adding or changing a constraint, it may result in a migration failure if the existing data
+is not valid for the new constraint. For example, changing a column from formerly `int NULL` to `int NOT NULL` on a column that
+contains `NULL` values will cause the migration to fail. This can be fixed by "fixing" the data and retrying the
+migration applying.
+
+In any case, we advise to add the executed statements to fix the data to the migration file to not run into this issue
+again for other deployments.
+
+## A Word on Transactions
+
+At Ariga, we advise to favor correctness over performance when it comes to database changes. Wrapping migration
+execution in a transaction ensures the changes made to the database are rolled back if an error occurs along the way and
+even in cases where the connection is lost, the changes are not committed. However, some databases (most notably
+MySQL and MariaDB) do not support
+[transactional DDL](https://wiki.postgresql.org/wiki/Transactional_DDL_in_PostgreSQL:_A_Competitive_Analysis). If
+you are working with a database that does not support transactional DDL, a rollback will not restore the state previous to the
+migration execution, and you end up with a "broken" database state. However, since Atlas stores its own information about
+applied statements within the same transaction, it will start execution with the next statement in the migration file
+on a following migration attempt.
+
+## Manipulating the Schema Revisions Table
+
+Atlas stores information about applied migrations in the managed database. Sometimes, you want to notify Atlas that you
+manually applied or rolled back a migration file. Atlas provides the
+[`atlas migrate set`](/cli-reference#atlas-migrate-set) command for such cases.
+
+:::warning
+
+Please refrain from manually making changes to the Atlas revision table. Atlas' behavior after such a change is
+undefined, and it can possibly break your database to a point you cannot recover from.
+
+:::
diff --git a/doc/website/sidebars.js b/doc/website/sidebars.js
index 6edbe9f3..bc274f60 100644
--- a/doc/website/sidebars.js
+++ b/doc/website/sidebars.js
@@ -38,6 +38,7 @@ module.exports = {
                 {type: 'doc', id: 'versioned/lint', label: 'Migration Linting'},
                 {type: 'doc', id: 'versioned/new', label: 'Manual Migrations'},
                 {type: 'doc', id: 'versioned/apply', label: 'Migration Applying'},
+                {type: 'doc', id: 'versioned/troubleshoot', label: 'Migration Troubleshooting'},
                 {type: 'doc', id: 'versioned/import', label: 'Migration Import'},
             ]
         },
diff --git a/sql/migrate/migrate.go b/sql/migrate/migrate.go
index 93850629..4956bb8c 100644
--- a/sql/migrate/migrate.go
+++ b/sql/migrate/migrate.go
@@ -703,14 +703,36 @@ func (e *Executor) ExecuteN(ctx context.Context, n int) (err error) {
 		}
 		pending = pending[:n]
 	}
+	return e.exec(ctx, pending)
+}
+
+// ExecuteTo executes all pending migration files up to and including version.
+func (e *Executor) ExecuteTo(ctx context.Context, version string) (err error) {
+	pending, err := e.Pending(ctx)
+	if err != nil {
+		return err
+	}
+	// Strip pending files greater given version.
+	switch idx := FilesLastIndex(pending, func(file File) bool {
+		return file.Version() == version
+	}); idx {
+	case -1:
+		return fmt.Errorf("sql/migrate: execute: migration with version %q not found", version)
+	default:
+		pending = pending[:idx+1]
+	}
+	return e.exec(ctx, pending)
+}
+
+func (e *Executor) exec(ctx context.Context, files []File) error {
 	revs, err := e.rrw.ReadRevisions(ctx)
 	if err != nil {
 		return fmt.Errorf("sql/migrate: execute: read revisions: %w", err)
 	}
-	if err := LogIntro(e.log, revs, pending); err != nil {
+	if err := LogIntro(e.log, revs, files); err != nil {
 		return err
 	}
-	for _, m := range pending {
+	for _, m := range files {
 		if err := e.Execute(ctx, m); err != nil {
 			return err
 		}
@@ -719,8 +741,27 @@ func (e *Executor) ExecuteN(ctx context.Context, n int) (err error) {
 	return err
 }
 
+type (
+	replayConfig struct {
+		version string // to which version to replay (inclusive)
+	}
+	// ReplayOption configures a migration directory replay behavior.
+	ReplayOption func(*replayConfig)
+)
+
+// ReplayToVersion configures the last version to apply when replaying the migration directory.
+func ReplayToVersion(v string) ReplayOption {
+	return func(c *replayConfig) {
+		c.version = v
+	}
+}
+
 // Replay the migration directory and invoke the state to get back the inspection result.
-func (e *Executor) Replay(ctx context.Context, r StateReader) (_ *schema.Realm, err error) {
+func (e *Executor) Replay(ctx context.Context, r StateReader, opts ...ReplayOption) (_ *schema.Realm, err error) {
+	c := &replayConfig{}
+	for _, opt := range opts {
+		opt(c)
+	}
 	// Clean up after ourselves.
 	restore, err := e.drv.(Snapshoter).Snapshot(ctx)
 	if err != nil {
@@ -732,7 +773,13 @@ func (e *Executor) Replay(ctx context.Context, r StateReader) (_ *schema.Realm,
 		}
 	}()
 	// Replay the migration directory on the database.
-	if err := e.ExecuteN(ctx, 0); err != nil && !errors.Is(err, ErrNoPendingFiles) {
+	switch {
+	case c.version != "":
+		err = e.ExecuteTo(ctx, c.version)
+	default:
+		err = e.ExecuteN(ctx, 0)
+	}
+	if err != nil && !errors.Is(err, ErrNoPendingFiles) {
 		return nil, fmt.Errorf("sql/migrate: read migration directory state: %w", err)
 	}
 	return r.ReadState(ctx)


Test patch:
diff --git a/cmd/atlas/internal/cmdapi/schema_test.go b/cmd/atlas/internal/cmdapi/schema_test.go
index 8da71cef..5e096ef8 100644
--- a/cmd/atlas/internal/cmdapi/schema_test.go
+++ b/cmd/atlas/internal/cmdapi/schema_test.go
@@ -127,6 +127,69 @@ table "tbl" {
 		"-- Add column \"col_3\" to table: \"tbl\"\nALTER TABLE `tbl` ADD COLUMN `col_3` text NOT NULL\n",
 		s,
 	)
+
+	// Current state from migration directory with version, desired state from HCL - two missing columns.
+	s, err = runCmd(
+		schemaDiffCmd(),
+		"--from", "file://testdata/sqlite?version=20220318104614",
+		"--to", "file://"+p,
+		"--dev-url", openSQLite(t, ""),
+	)
+	require.NoError(t, err)
+	require.EqualValues(
+		t,
+		"-- Add column \"col_2\" to table: \"tbl\"\n"+
+			"ALTER TABLE `tbl` ADD COLUMN `col_2` bigint NULL\n"+
+			"-- Add column \"col_3\" to table: \"tbl\"\n"+
+			"ALTER TABLE `tbl` ADD COLUMN `col_3` text NOT NULL\n",
+		s,
+	)
+
+	// Current state from migration directory, desired state from multi file HCL - missing column.
+	p = t.TempDir()
+	var (
+		one = filepath.Join(p, "one.hcl")
+		two = filepath.Join(p, "two.hcl")
+	)
+	require.NoError(t, os.WriteFile(one, []byte(`table "tbl" {
+  schema = schema.main
+  column "col" {
+    type = int
+  }
+  column "col_2" {
+    type = bigint
+    null = true
+  }
+  column "col_3" {
+    type = text
+  }
+}`), 0644))
+	require.NoError(t, os.WriteFile(two, []byte(`schema "main" {}`), 0644))
+	s, err = runCmd(
+		schemaDiffCmd(),
+		"--from", "file://testdata/sqlite",
+		"--to", "file://"+p,
+		"--dev-url", openSQLite(t, ""),
+	)
+	require.NoError(t, err)
+	require.EqualValues(
+		t,
+		"-- Add column \"col_3\" to table: \"tbl\"\nALTER TABLE `tbl` ADD COLUMN `col_3` text NOT NULL\n",
+		s,
+	)
+	s, err = runCmd(
+		schemaDiffCmd(),
+		"--from", "file://testdata/sqlite",
+		"--to", "file://"+one,
+		"--to", "file://"+two,
+		"--dev-url", openSQLite(t, ""),
+	)
+	require.NoError(t, err)
+	require.EqualValues(
+		t,
+		"-- Add column \"col_3\" to table: \"tbl\"\nALTER TABLE `tbl` ADD COLUMN `col_3` text NOT NULL\n",
+		s,
+	)
 }
 
 func TestFmt(t *testing.T) {
diff --git a/internal/integration/integration_test.go b/internal/integration/integration_test.go
index c7405bc9..f96c1dbe 100644
--- a/internal/integration/integration_test.go
+++ b/internal/integration/integration_test.go
@@ -493,6 +493,78 @@ func testCLISchemaApplyAutoApprove(t T, h string, dsn string, args ...string) {
 	require.NotNil(t, u)
 }
 
+func testCLISchemaApplyFromMigrationDir(t T) {
+	const (
+		dbname  = "apply_migration_dir"
+		devname = dbname + "_dev"
+	)
+	t.dropSchemas(dbname, devname)
+	t.migrate(&schema.AddSchema{S: schema.New(dbname)}, &schema.AddSchema{S: schema.New(devname)})
+	defer t.migrate(&schema.DropSchema{S: schema.New(dbname)}, &schema.DropSchema{S: schema.New(devname)})
+
+	users, err := t.driver().SchemaDiff(schema.New(""), schema.New("").AddTables(t.users()))
+	require.NoError(t, err)
+	usersT := t.users()
+	usersT.Name = "users_2"
+	users2, err := t.driver().SchemaDiff(schema.New(""), schema.New("").AddTables(usersT))
+	require.NoError(t, err)
+
+	addUsers, err := t.driver().PlanChanges(context.Background(), "", users)
+	require.NoError(t, err)
+	addUsers2, err := t.driver().PlanChanges(context.Background(), "", users2)
+	require.NoError(t, err)
+
+	var (
+		fn = func(c []*migrate.Change) string {
+			var buf strings.Builder
+			for _, c := range c {
+				buf.WriteString(c.Cmd)
+				buf.WriteString("\n")
+			}
+			return buf.String()
+		}
+		one = fn(addUsers.Changes)
+		two = fn(addUsers2.Changes)
+	)
+
+	p := t.TempDir()
+	require.NoError(t, os.WriteFile(filepath.Join(p, "1.sql"), []byte(one), 0644))
+	require.NoError(t, os.WriteFile(filepath.Join(p, "2.sql"), []byte(two), 0644))
+
+	require.NoError(t, initCLI())
+	require.NoError(t, exec.Command(
+		"go", "run", "ariga.io/atlas/cmd/atlas",
+		"migrate", "hash",
+		"--dir", "file://"+p,
+	).Run())
+
+	// All versions - must contain all migration files.
+	out, err := exec.Command(
+		"go", "run", "ariga.io/atlas/cmd/atlas",
+		"schema", "apply",
+		"-u", t.url(dbname),
+		"--to", "file://"+p,
+		"--dev-url", t.url(devname),
+		"--dry-run",
+	).CombinedOutput()
+	require.NoError(t, err)
+	require.Contains(t, string(out), one)
+	require.Contains(t, string(out), two)
+
+	// One version - must contain only file one.
+	out, err = exec.Command(
+		"go", "run", "ariga.io/atlas/cmd/atlas",
+		"schema", "apply",
+		"-u", t.url(dbname),
+		"--to", "file://"+p+"?version=1",
+		"--dev-url", t.url(devname),
+		"--dry-run",
+	).CombinedOutput()
+	require.NoError(t, err)
+	require.Contains(t, string(out), one)
+	require.NotContains(t, string(out), two)
+}
+
 func testCLISchemaDiff(t T, dsn string) {
 	err := initCLI()
 
diff --git a/internal/integration/mysql_test.go b/internal/integration/mysql_test.go
index 4fb36cb7..123b2b1a 100644
--- a/internal/integration/mysql_test.go
+++ b/internal/integration/mysql_test.go
@@ -654,6 +654,11 @@ table "users" {
 			testCLISchemaApplyAutoApprove(t, fmt.Sprintf(h, charset.V, collate.V), t.url("test"))
 		})
 	})
+	t.Run("SchemaApplyFromMigrationDir", func(t *testing.T) {
+		myRun(t, func(t *myTest) {
+			testCLISchemaApplyFromMigrationDir(t)
+		})
+	})
 }
 
 func TestMySQL_CLI_MultiSchema(t *testing.T) {
diff --git a/internal/integration/postgres_test.go b/internal/integration/postgres_test.go
index 3e955586..bde04968 100644
--- a/internal/integration/postgres_test.go
+++ b/internal/integration/postgres_test.go
@@ -711,6 +711,11 @@ table "users" {
 			testCLISchemaApplyAutoApprove(t, h, t.url(""))
 		})
 	})
+	t.Run("SchemaApplyFromMigrationDir", func(t *testing.T) {
+		pgRun(t, func(t *pgTest) {
+			testCLISchemaApplyFromMigrationDir(t)
+		})
+	})
 }
 
 func TestPostgres_CLI_MultiSchema(t *testing.T) {
diff --git a/internal/integration/testdata/sqlite/cli-apply-directory.txt b/internal/integration/testdata/sqlite/cli-apply-directory.txt
index 46afbac8..c3c7c8cf 100644
--- a/internal/integration/testdata/sqlite/cli-apply-directory.txt
+++ b/internal/integration/testdata/sqlite/cli-apply-directory.txt
@@ -1,13 +1,13 @@
 ! atlas schema apply -f empty/ -u URL --auto-approve
-stderr 'no schema files found'
+stderr '"empty" contains neither SQL nor HCL files'
 
 atlas schema apply -f schema/ -u URL --auto-approve
 cmpshow users expected.sql
 
-atlas schema apply -f schema/ -u URL --auto-approve --dry-run --exclude "*.*.id" > inspected.txt
+atlas schema apply -f schema/ -u URL --auto-approve --dry-run --exclude "*.id" > inspected.txt
 cmp inspected.txt alter.txt
 
-atlas schema apply -f schema/ -u URL --auto-approve --dry-run --exclude "*.users" > inspected.txt
+atlas schema apply -f schema/ -u URL --auto-approve --dry-run --exclude "users" > inspected.txt
 cmp inspected.txt create.txt
 
 -- schema/users.hcl --
diff --git a/internal/integration/testdata/sqlite/cli-schema-project-file.txt b/internal/integration/testdata/sqlite/cli-schema-project-file.txt
index ff9a0029..2eb345d3 100644
--- a/internal/integration/testdata/sqlite/cli-schema-project-file.txt
+++ b/internal/integration/testdata/sqlite/cli-schema-project-file.txt
@@ -5,7 +5,7 @@ stderr '"url" not set'
 stderr '"url" not set'
 
 ! atlas schema apply --url URL
-stderr '"file" not set'
+stderr 'one of flag\(s\) "file" or "to" is required'
 
 ! atlas schema apply -f atlas.hcl -u URL
 stderr 'cannot parse project file'
diff --git a/sql/migrate/migrate_test.go b/sql/migrate/migrate_test.go
index 75aabb53..80e0fcf1 100644
--- a/sql/migrate/migrate_test.go
+++ b/sql/migrate/migrate_test.go
@@ -159,6 +159,31 @@ func TestExecutor_Replay(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, []string{"DROP TABLE IF EXISTS t;", "CREATE TABLE t(c int);"}, drv.executed)
 
+	// ReplayVersion option.
+	*drv = mockDriver{}
+	dir, err := migrate.NewLocalDir(filepath.FromSlash("testdata/migrate/sub"))
+	require.NoError(t, err)
+	*d = *dir
+	_, err = ex.Replay(ctx, migrate.RealmConn(drv, nil), migrate.ReplayToVersion("0"))
+	require.ErrorContains(t, err, "migration with version \"0\" not found")
+	require.Zero(t, drv.executed)
+
+	*drv = mockDriver{}
+	_, err = ex.Replay(ctx, migrate.RealmConn(drv, nil), migrate.ReplayToVersion("1.a"))
+	require.NoError(t, err)
+	require.Equal(t, []string{"CREATE TABLE t_sub(c int);", "ALTER TABLE t_sub ADD c1 int;"}, drv.executed)
+
+	*drv = mockDriver{}
+	_, err = ex.Replay(ctx, migrate.RealmConn(drv, nil), migrate.ReplayToVersion("3"))
+	require.NoError(t, err)
+	require.Equal(t, []string{
+		"CREATE TABLE t_sub(c int);",
+		"ALTER TABLE t_sub ADD c1 int;",
+		"ALTER TABLE t_sub ADD c2 int;",
+		"ALTER TABLE t_sub ADD c3 int;",
+		"ALTER TABLE t_sub ADD c4 int;",
+	}, drv.executed)
+
 	// Does not work if database is not clean.
 	drv.dirty = true
 	drv.realm = schema.Realm{Schemas: []*schema.Schema{{Name: "schema"}}}
@@ -294,6 +319,7 @@ func TestExecutor(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, ex)
 	require.ErrorIs(t, ex.ExecuteN(context.Background(), 0), migrate.ErrChecksumMismatch)
+	require.ErrorIs(t, ex.ExecuteTo(context.Background(), ""), migrate.ErrChecksumMismatch)
 
 	// Prerequisites.
 	var (
@@ -411,6 +437,13 @@ func TestExecutor(t *testing.T) {
 
 	// Everything is applied.
 	require.ErrorIs(t, ex.ExecuteN(context.Background(), 0), migrate.ErrNoPendingFiles)
+
+	// Test ExecuteTo.
+	*rrw = []*migrate.Revision{}
+	*drv = mockDriver{}
+	require.EqualError(t, ex.ExecuteTo(context.Background(), ""), "sql/migrate: execute: migration with version \"\" not found")
+	require.NoError(t, ex.ExecuteTo(context.Background(), "2.10.x-20"))
+	requireEqualRevisions(t, []*migrate.Revision{rev1, rev2}, *rrw)
 }
 
 func TestExecutor_Baseline(t *testing.T) {