Repository: ariga/atlas
Pull request #1179 (2022-10-07 13:17:24)

cmd/atlas/internal/cmdapi: no longer use global variables for command…
…s and flags to ease testing and workaround cobras weird flag handling when subcommands are involved


Patch:
diff --git a/cmd/atlas/internal/cmdapi/cmdapi.go b/cmd/atlas/internal/cmdapi/cmdapi.go
index 61771c7d..385e79a0 100644
--- a/cmd/atlas/internal/cmdapi/cmdapi.go
+++ b/cmd/atlas/internal/cmdapi/cmdapi.go
@@ -6,11 +6,21 @@
 package cmdapi
 
 import (
+	"context"
 	"encoding/csv"
+	"errors"
 	"fmt"
+	"io"
+	"net/url"
+	"os"
+	"path/filepath"
 	"strings"
 
+	"ariga.io/atlas/sql/migrate"
+	"ariga.io/atlas/sql/schema"
+	"ariga.io/atlas/sql/sqlclient"
 	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
 	"github.com/zclconf/go-cty/cty"
 	"golang.org/x/mod/semver"
 )
@@ -59,17 +69,10 @@ Atlas is licensed under Apache 2.0 as found in https://github.com/ariga/atlas/bl
 )
 
 func init() {
-	Root.AddCommand(schemaCmd)
 	Root.AddCommand(versionCmd)
 	Root.AddCommand(licenseCmd)
 }
 
-// receivesEnv configures cmd to receive the common '--env' flag.
-func receivesEnv(cmd *cobra.Command) {
-	cmd.PersistentFlags().StringVarP(&GlobalFlags.SelectedEnv, "env", "", "", "set which env from the project file to use")
-	cmd.PersistentFlags().VarP(&GlobalFlags.Vars, varFlag, "", "input variables")
-}
-
 // inputValsFromEnv populates GlobalFlags.Vars from the active environment. If we are working
 // inside a project, the "var" flag is not propagated to the schema definition. Instead, it
 // is used to evaluate the project file which can pass input values via the "values" block
@@ -79,7 +82,7 @@ func inputValsFromEnv(cmd *cobra.Command) error {
 	if err != nil {
 		return err
 	}
-	if fl := cmd.Flag(varFlag); fl == nil {
+	if fl := cmd.Flag(flagVar); fl == nil {
 		return nil
 	}
 	values, err := activeEnv.asMap()
@@ -94,7 +97,7 @@ func inputValsFromEnv(cmd *cobra.Command) error {
 		pairs = append(pairs, fmt.Sprintf("%s=%s", k, v))
 	}
 	vars := strings.Join(pairs, ",")
-	if err := cmd.Flags().Set(varFlag, vars); err != nil {
+	if err := cmd.Flags().Set(flagVar, vars); err != nil {
 		return err
 	}
 	return nil
@@ -164,3 +167,330 @@ func (v *Vars) Set(s string) error {
 func (v *Vars) Type() string {
 	return "<name>=<value>"
 }
+
+const (
+	flagAllowDirty     = "allow-dirty"
+	flagAutoApprove    = "auto-approve"
+	flagBaseline       = "baseline"
+	flagDevURL         = "dev-url"
+	flagDirURL         = "dir"
+	flagDirFormat      = "dir-format"
+	flagDryRun         = "dry-run"
+	flagDSN            = "dsn" // deprecated in favor of flagURL
+	flagEnv            = "env"
+	flagExclude        = "exclude"
+	flagFile           = "file"
+	flagFrom           = "from"
+	flagGitBase        = "git-base"
+	flagGitDir         = "git-dir"
+	flagLatest         = "latest"
+	flagLog            = "log"
+	flagRevisionSchema = "revisions-schema"
+	flagSchema         = "schema"
+	flagTo             = "to"
+	flagTxMode         = "tx-mode"
+	flagURL            = "url"
+	flagVar            = "var"
+	flagQualifier      = "qualifier"
+)
+
+func addFlagEnvVar(set *pflag.FlagSet) {
+	set.StringVar(&GlobalFlags.SelectedEnv, flagEnv, "", "set which env from the project file to use")
+	set.Var(&GlobalFlags.Vars, flagVar, "input variables")
+}
+
+func addFlagAutoApprove(set *pflag.FlagSet, target *bool) {
+	set.BoolVar(target, flagAutoApprove, false, "apply changes without prompting for approval")
+}
+
+func addFlagDirFormat(set *pflag.FlagSet, target *string) {
+	set.StringVar(target, flagDirFormat, "atlas", "select migration file format")
+}
+
+// addFlagURL adds a URL flag. If given, args[0] override the name, args[1] the shorthand, args[2] the default value.
+func addFlagDirURL(set *pflag.FlagSet, target *string, args ...string) {
+	name, short, val := flagDirURL, "", "file://migrations"
+	switch len(args) {
+	case 3:
+		val = args[2]
+		fallthrough
+	case 2:
+		short = args[1]
+		fallthrough
+	case 1:
+		name = args[0]
+	}
+	set.StringVarP(target, name, short, val, "select migration directory using URL format")
+}
+
+func addFlagDevURL(set *pflag.FlagSet, target *string) {
+	set.StringVar(
+		target,
+		flagDevURL,
+		"",
+		"[driver://username:password@address/dbname?param=value] select a dev database using the URL format",
+	)
+}
+
+func addFlagDSN(set *pflag.FlagSet, target *string) {
+	set.StringVarP(target, flagDSN, "d", "", "")
+	cobra.CheckErr(set.MarkHidden(flagDSN))
+}
+
+func addFlagDryRun(set *pflag.FlagSet, target *bool) {
+	set.BoolVar(target, flagDryRun, false, "print SQL without executing it")
+}
+
+func addFlagExclude(set *pflag.FlagSet, target *[]string) {
+	set.StringSliceVar(
+		target,
+		flagExclude,
+		nil,
+		"list of glob patterns used to filter resources from applying",
+	)
+}
+
+func addFlagLog(set *pflag.FlagSet, target *string) {
+	set.StringVar(target, flagLog, "", "go template to use to format logs")
+}
+
+func addFlagRevisionSchema(set *pflag.FlagSet, target *string) {
+	set.StringVar(target, flagRevisionSchema, "", "name of the schema the revisions table resides in")
+}
+
+func addFlagSchemas(set *pflag.FlagSet, target *[]string) {
+	set.StringSliceVarP(
+		target,
+		flagSchema, "s",
+		nil,
+		"set schema names",
+	)
+}
+
+// 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"
+	switch len(args) {
+	case 2:
+		short = args[1]
+		fallthrough
+	case 1:
+		name = args[0]
+	}
+	set.StringVarP(
+		target,
+		name, short,
+		"",
+		"[driver://username:password@address/dbname?param=value] select a resource using the URL format",
+	)
+}
+
+func addFlagURLs(set *pflag.FlagSet, target *[]string, args ...string) {
+	name, short := flagURL, "u"
+	switch len(args) {
+	case 2:
+		short = args[1]
+		fallthrough
+	case 1:
+		name = args[0]
+	}
+	set.StringSliceVarP(
+		target,
+		name, short,
+		nil,
+		"[driver://username:password@address/dbname?param=value] select a resource using the URL format",
+	)
+}
+
+func dsn2url(cmd *cobra.Command) error {
+	dsnF, urlF := cmd.Flag(flagDSN), cmd.Flag(flagURL)
+	switch {
+	case dsnF == nil:
+	case dsnF.Changed && urlF.Changed:
+		return errors.New(`both flags "url" and "dsn" were set`)
+	case dsnF.Changed && !urlF.Changed:
+		return cmd.Flags().Set(flagURL, dsnF.Value.String())
+	}
+	return nil
+}
+
+// maySetFlag sets the flag with the provided name to envVal if such a flag exists
+// on the cmd, it was not set by the user via the command line and if envVal is not
+// an empty string.
+func maySetFlag(cmd *cobra.Command, name, envVal string) error {
+	if fl := cmd.Flag(name); fl == nil || fl.Changed || envVal == "" {
+		return nil
+	}
+	return cmd.Flags().Set(name, envVal)
+}
+
+type (
+	// stateReadCloser is a migrate.StateReader with an optional io.Closer.
+	stateReadCloser struct {
+		migrate.StateReader
+		io.Closer        // optional close function
+		Schema    string // in case we work on a single schema
+	}
+	// 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
+	}
+)
+
+// stateReader returns a migrate.StateReader that reads the state from the given urls.
+func stateReader(ctx context.Context, config *stateReaderConfig) (*stateReadCloser, error) {
+	scheme, err := selectScheme(config.urls)
+	if err != nil {
+		return nil, err
+	}
+	parsed := make([]*url.URL, len(config.urls))
+	for i, u := range config.urls {
+		parsed[i], err = url.Parse(u)
+		if err != nil {
+			return nil, err
+		}
+	}
+	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)
+		}
+		path := filepath.Join(parsed[0].Host, parsed[0].Path)
+		fi, err := os.Stat(path)
+		if err != nil {
+			return nil, err
+		}
+		if !fi.IsDir() {
+			// If there is only one url given, and it is a file, consider it an HCL path.
+			return hclStateReader(ctx, config, parsed)
+		}
+		files, err := os.ReadDir(path)
+		if err != nil {
+			return nil, err
+		}
+		// Check all files, if we find both HCL and SQL files, abort. Otherwise, proceed accordingly.
+		var hcl, sql bool
+		for _, f := range files {
+			ext := filepath.Ext(f.Name())
+			switch {
+			case hcl && ext == ".sql", sql && ext == ".hcl":
+				return nil, fmt.Errorf("ambiguos files: %q contains both SQL and HCL files", path)
+			case ext == ".hcl":
+				hcl = true
+			case ext == ".sql":
+				sql = true
+			default:
+				// unknown extension, we don't care
+			}
+		}
+		switch {
+		case hcl:
+			return hclStateReader(ctx, config, parsed)
+		case sql:
+			dir, err := dirURL(parsed[0], false)
+			if err != nil {
+				return nil, err
+			}
+			ex, err := migrate.NewExecutor(config.dev.Driver, dir, migrate.NopRevisionReadWriter{})
+			if err != nil {
+				return nil, err
+			}
+			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)
+			}())
+			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,
+			}, nil
+		default:
+			return nil, fmt.Errorf("%q contains neither SQL nor HCL files", path)
+		}
+	default:
+		// All other schemes are database (or docker) connections.
+		c, err := sqlclient.Open(ctx, config.urls[0]) // call to selectScheme already checks for len > 0
+		if err != nil {
+			return nil, err
+		}
+		var sr migrate.StateReader
+		switch c.URL.Schema {
+		case "":
+			sr = migrate.RealmConn(c.Driver, nil)
+		default:
+			sr = migrate.SchemaConn(c.Driver, c.URL.Schema, nil)
+		}
+		return &stateReadCloser{
+			StateReader: sr,
+			Closer:      c,
+			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) {
+	paths := make([]string, len(urls))
+	for i, u := range urls {
+		paths[i] = 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 {
+		return nil, err
+	}
+	if len(config.schemas) > 0 {
+		// Validate all schemas in file were selected by user.
+		sm := make(map[string]bool, len(config.schemas))
+		for _, s := range config.schemas {
+			sm[s] = true
+		}
+		for _, s := range realm.Schemas {
+			if !sm[s.Name] {
+				return nil, fmt.Errorf("schema %q from paths %q is not requested (all schemas in HCL must be requested)", s.Name, paths)
+			}
+		}
+	}
+	// 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 {
+		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 {
+		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
+	}
+	return t, nil
+}
+
+// Close redirects calls to Close to the enclosed io.Closer.
+func (sr *stateReadCloser) Close() {
+	if sr.Closer != nil {
+		sr.Closer.Close()
+	}
+}
diff --git a/cmd/atlas/internal/cmdapi/diff.go b/cmd/atlas/internal/cmdapi/diff.go
deleted file mode 100644
index 88b0196f..00000000
--- a/cmd/atlas/internal/cmdapi/diff.go
+++ /dev/null
@@ -1,305 +0,0 @@
-// Copyright 2021-present The Atlas Authors. All rights reserved.
-// This source code is licensed under the Apache 2.0 license found
-// in the LICENSE file in the root directory of this source tree.
-
-package cmdapi
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"io"
-	"net/url"
-	"os"
-	"path/filepath"
-	"strings"
-
-	"ariga.io/atlas/sql/migrate"
-	"ariga.io/atlas/sql/schema"
-	"ariga.io/atlas/sql/sqlclient"
-
-	"github.com/spf13/cobra"
-)
-
-type diffCmdOpts struct {
-	fromURL string
-	toURL   string
-	devURL  string
-}
-
-// newDiffCmd returns a new *cobra.Command that runs cmdDiffRun with the given flags.
-func newDiffCmd() *cobra.Command {
-	var opts diffCmdOpts
-	cmd := &cobra.Command{
-		Use:   "diff",
-		Short: "Calculate and print the diff between two schemas.",
-		Long: `'atlas schema diff' reads the state of two given schema definitions, 
-calculates the difference in their schemas, and prints a plan of
-SQL statements to migrate the "from" database to the schema of the "to" database.
-The database states can be read from a connected database, an HCL project or a migration directory.`,
-		Example: `  atlas schema diff --from mysql://user:pass@localhost:3306/test --to file://schema.hcl
-  atlas schema diff --from mysql://user:pass@localhost:3306 --to file://schema_1.hcl --to file://schema_2.hcl
-  atlas schema diff --from mysql://user:pass@localhost:3306 --to file://migrations`,
-		RunE: func(cmd *cobra.Command, args []string) error {
-			return cmdDiffRun(cmd, &opts)
-		},
-	}
-	cmd.Flags().StringVarP(&opts.fromURL, "from", "", "", "[driver://username:password@protocol(address)/dbname?param=value] select a database using the URL format")
-	cmd.Flags().StringVarP(&opts.toURL, "to", "", "", "[driver://username:password@protocol(address)/dbname?param=value] select a database using the URL format")
-	cmd.Flags().StringVarP(&opts.devURL, "dev-url", "", "", "[driver://username:password@protocol(address)/dbname?param=value] select a database using the URL format")
-	cmd.Flags().SortFlags = false
-	cobra.CheckErr(cmd.MarkFlagRequired("from"))
-	cobra.CheckErr(cmd.MarkFlagRequired("to"))
-	return cmd
-}
-
-func init() {
-	schemaCmd.AddCommand(newDiffCmd())
-}
-
-// cmdDiffRun connects to the given databases, and prints an SQL plan to get from
-// the "from" schema to the "to" schema.
-func cmdDiffRun(cmd *cobra.Command, flags *diffCmdOpts) error {
-	var (
-		ctx = cmd.Context()
-		c   *sqlclient.Client
-	)
-	// We need a driver for diffing and planning. If given, dev database has precedence.
-	if flags.devURL != "" {
-		var err error
-		c, err = sqlclient.Open(ctx, flags.devURL)
-		if err != nil {
-			return err
-		}
-		defer c.Close()
-	}
-	from, err := stateReader(ctx, &stateReaderConfig{urls: []string{flags.fromURL}, dev: c})
-	if err != nil {
-		return err
-	}
-	defer from.Close()
-	to, err := stateReader(ctx, &stateReaderConfig{urls: []string{flags.toURL}, dev: c})
-	if err != nil {
-		return err
-	}
-	defer to.Close()
-	if c == nil {
-		// If not both states are provided by a database connection, the call to state-reader would have returned
-		// 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)
-	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
-	}
-	if len(p.Changes) == 0 {
-		cmd.Println("Schemas are synced, no changes to be made.")
-	}
-	for _, c := range p.Changes {
-		if c.Comment != "" {
-			cmd.Println("--", strings.ToUpper(c.Comment[:1])+c.Comment[1:])
-		}
-		cmd.Println(c.Cmd)
-	}
-	return nil
-}
-
-type (
-	// stateReadCloser is a migrate.StateReader with an optional io.Closer.
-	stateReadCloser struct {
-		migrate.StateReader
-		io.Closer        // optional close function
-		Schema    string // in case we work on a single schema
-	}
-	// 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
-	}
-)
-
-// stateReader returns a migrate.StateReader that reads the state from the given urls.
-func stateReader(ctx context.Context, config *stateReaderConfig) (*stateReadCloser, error) {
-	scheme, err := selectScheme(config.urls)
-	if err != nil {
-		return nil, err
-	}
-	parsed := make([]*url.URL, len(config.urls))
-	for i, u := range config.urls {
-		parsed[i], err = url.Parse(u)
-		if err != nil {
-			return nil, err
-		}
-	}
-	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)
-		}
-		path := filepath.Join(parsed[0].Host, parsed[0].Path)
-		fi, err := os.Stat(path)
-		if err != nil {
-			return nil, err
-		}
-		if !fi.IsDir() {
-			// If there is only one url given, and it is a file, consider it an HCL path.
-			return hclStateReader(ctx, config, parsed)
-		}
-		files, err := os.ReadDir(path)
-		if err != nil {
-			return nil, err
-		}
-		// Check all files, if we find both HCL and SQL files, abort. Otherwise, proceed accordingly.
-		var hcl, sql bool
-		for _, f := range files {
-			ext := filepath.Ext(f.Name())
-			switch {
-			case hcl && ext == ".sql", sql && ext == ".hcl":
-				return nil, fmt.Errorf("ambiguos files: %q contains both SQL and HCL files", path)
-			case ext == ".hcl":
-				hcl = true
-			case ext == ".sql":
-				sql = true
-			default:
-				// unknown extension, we don't care
-			}
-		}
-		switch {
-		case hcl:
-			return hclStateReader(ctx, config, parsed)
-		case sql:
-			dir, err := dirURL(parsed[0], false)
-			if err != nil {
-				return nil, err
-			}
-			ex, err := migrate.NewExecutor(config.dev.Driver, dir, migrate.NopRevisionReadWriter{})
-			if err != nil {
-				return nil, err
-			}
-			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)
-			}())
-			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,
-			}, nil
-		default:
-			return nil, fmt.Errorf("%q contains neither SQL nor HCL files", path)
-		}
-	default:
-		// All other schemes are database (or docker) connections.
-		c, err := sqlclient.Open(ctx, config.urls[0]) // call to selectScheme already checks for len > 0
-		if err != nil {
-			return nil, err
-		}
-		var sr migrate.StateReader
-		switch c.URL.Schema {
-		case "":
-			sr = migrate.RealmConn(c.Driver, nil)
-		default:
-			sr = migrate.SchemaConn(c.Driver, c.URL.Schema, nil)
-		}
-		return &stateReadCloser{
-			StateReader: sr,
-			Closer:      c,
-			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) {
-	paths := make([]string, len(urls))
-	for i, u := range urls {
-		paths[i] = 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 {
-		return nil, err
-	}
-	if len(config.schemas) > 0 {
-		// Validate all schemas in file were selected by user.
-		sm := make(map[string]bool, len(config.schemas))
-		for _, s := range config.schemas {
-			sm[s] = true
-		}
-		for _, s := range realm.Schemas {
-			if !sm[s.Name] {
-				return nil, fmt.Errorf("schema %q from paths %q is not requested (all schemas in HCL must be requested)", s.Name, paths)
-			}
-		}
-	}
-	// 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 {
-		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 {
-		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
-	}
-	return t, nil
-}
-
-// Close redirects calls to Close to the enclosed io.Closer.
-func (sr *stateReadCloser) Close() {
-	if sr.Closer != nil {
-		sr.Closer.Close()
-	}
-}
diff --git a/cmd/atlas/internal/cmdapi/migrate.go b/cmd/atlas/internal/cmdapi/migrate.go
index bd4b9ba3..7ba9f0b6 100644
--- a/cmd/atlas/internal/cmdapi/migrate.go
+++ b/cmd/atlas/internal/cmdapi/migrate.go
@@ -32,93 +32,54 @@ import (
 	"github.com/hashicorp/hcl/v2/hclparse"
 	"github.com/hashicorp/hcl/v2/hclsyntax"
 	"github.com/spf13/cobra"
-	"github.com/spf13/pflag"
 )
 
-const (
-	migrateFlagURL              = "url"
-	migrateFlagDevURL           = "dev-url"
-	migrateFlagDir              = "dir"
-	migrateFlagDirFormat        = "dir-format"
-	migrateFlagLog              = "log"
-	migrateFlagRevisionsSchema  = "revisions-schema"
-	migrateFlagDryRun           = "dry-run"
-	migrateFlagTo               = "to"
-	migrateFlagFrom             = "from"
-	migrateFlagSchema           = "schema"
-	migrateLintLatest           = "latest"
-	migrateLintGitDir           = "git-dir"
-	migrateLintGitBase          = "git-base"
-	migrateDiffQualifier        = "qualifier"
-	migrateApplyAllowDirty      = "allow-dirty"
-	migrateApplyBaselineVersion = "baseline"
-	migrateApplyTxMode          = "tx-mode"
-)
+func init() {
+	migrateCmd := migrateCmd()
+	migrateCmd.AddCommand(
+		migrateApplyCmd(),
+		migrateDiffCmd(),
+		migrateHashCmd(),
+		migrateImportCmd(),
+		migrateLintCmd(),
+		migrateNewCmd(),
+		migrateSetCmd(),
+		migrateStatusCmd(),
+		migrateValidateCmd(),
+	)
+	Root.AddCommand(migrateCmd)
+}
 
-var (
-	// MigrateFlags are the flags used in MigrateCmd (and sub-commands).
-	MigrateFlags struct {
-		URL            string
-		DevURL         string
-		ToURLs         []string
-		Schemas        []string
-		DirURL         string
-		DirFormat      string // Deprecated: format is enclosed in the dir URL now.
-		RevisionSchema string
-		Apply          struct {
-			DryRun          bool
-			LogFormat       string
-			AllowDirty      bool
-			FromVersion     string
-			BaselineVersion string
-			TxMode          string
-		}
-		Diff struct {
-			Qualifier string // optional table qualifier
-		}
-		Import struct {
-			FromURL string
-			ToURL   string
-		}
-		Lint struct {
-			Format  string // log formatting
-			Latest  uint   // latest N migration files
-			GitDir  string // repository working dir
-			GitBase string // branch name to compare with
-		}
-		Status struct {
-			Format string // log formatting
-		}
-	}
-	// MigrateCmd represents the migrate command. It wraps several other sub-commands.
-	MigrateCmd = &cobra.Command{
+// migrateCmd represents the subcommand 'atlas migrate'.
+func migrateCmd() *cobra.Command {
+	cmd := &cobra.Command{
 		Use:   "migrate",
 		Short: "Manage versioned migration files",
 		Long:  "'atlas migrate' wraps several sub-commands for migration management.",
-		PersistentPreRunE: func(cmd *cobra.Command, _ []string) error {
-			if err := migrateFlagsFromEnv(cmd, nil); err != nil {
-				return err
-			}
-			if err := dirFormatBC(); err != nil {
-				return err
-			}
-			dir, err := dir(MigrateFlags.DirURL, false)
-			if err != nil {
-				return err
-			}
-			if err := migrate.Validate(dir); err != nil {
-				printChecksumErr(cmd.OutOrStderr())
-				cmd.SilenceUsage = true
-				return err
-			}
-			return nil
-		},
 	}
-	// MigrateApplyCmd represents the 'atlas migrate apply' subcommand.
-	MigrateApplyCmd = &cobra.Command{
-		Use:   "apply [flags] [count]",
-		Short: "Applies pending migration files on the connected database.",
-		Long: `'atlas migrate apply' reads the migration state of the connected database and computes what migrations are pending.
+	addFlagEnvVar(cmd.PersistentFlags())
+	return cmd
+}
+
+type migrateApplyFlags struct {
+	url             string
+	dirURL          string
+	revisionSchema  string
+	dryRun          bool
+	logFormat       string
+	allowDirty      bool   // allow working on a database that already has resources
+	fromVersion     string // compute pending files based on this version
+	baselineVersion string // apply with this version as baseline
+	txMode          string // (none, file, all)
+}
+
+func migrateApplyCmd() *cobra.Command {
+	var (
+		flags migrateApplyFlags
+		cmd   = &cobra.Command{
+			Use:   "apply [flags] [amount]",
+			Short: "Applies pending migration files on the connected database.",
+			Long: `'atlas migrate apply' reads the migration state of the connected database and computes what migrations are pending.
 It then attempts to apply the pending migration files in the correct order onto the database. 
 The first argument denotes the maximum number of migration files to apply.
 As a safety measure 'atlas migrate apply' will abort with an error, if:
@@ -126,234 +87,39 @@ As a safety measure 'atlas migrate apply' will abort with an error, if:
   - the migration and database history do not match each other
 
 If run with the "--dry-run" flag, atlas will not execute any SQL.`,
-		Example: `  atlas migrate apply -u mysql://user:pass@localhost:3306/dbname
+			Example: `  atlas migrate apply -u mysql://user:pass@localhost:3306/dbname
   atlas migrate apply --dir file:///path/to/migration/directory --url mysql://user:pass@localhost:3306/dbname 1
   atlas migrate apply --env dev 1
   atlas migrate apply --dry-run --env dev 1`,
-		Args: cobra.MaximumNArgs(1),
-		RunE: CmdMigrateApplyRun,
-	}
-	// MigrateDiffCmd represents the 'atlas migrate diff' subcommand.
-	MigrateDiffCmd = &cobra.Command{
-		Use:   "diff [flags] [name]",
-		Short: "Compute the diff between the migration directory and a desired state and create a new migration file.",
-		Long: `'atlas migrate diff' uses the dev-database to re-run all migration files in the migration directory, compares
-it to a given desired state and create a new migration file containing SQL statements to migrate the migration
-directory state to the desired schema. The desired state can be another connected database or an HCL file.`,
-		Example: `  atlas migrate diff --dev-url mysql://user:pass@localhost:3306/dev --to file://schema.hcl
-  atlas migrate diff --dev-url mysql://user:pass@localhost:3306/dev --to file://atlas.hcl add_users_table
-  atlas migrate diff --dev-url mysql://user:pass@localhost:3306/dev --to mysql://user:pass@localhost:3306/dbname
-  atlas migrate diff --env dev`,
-		Args: cobra.MaximumNArgs(1),
-		// If the migration directory does not exist on the validation attempt, this command will create it and
-		// consider the new migration directory "valid".
-		PersistentPreRunE: func(cmd *cobra.Command, _ []string) error {
-			if err := migrateFlagsFromEnv(cmd, nil); err != nil {
-				return err
-			}
-			if err := dirFormatBC(); err != nil {
-				return err
-			}
-			dir, err := dir(MigrateFlags.DirURL, true)
-			if err != nil {
-				return err
-			}
-			if err := migrate.Validate(dir); err != nil {
-				printChecksumErr(cmd.OutOrStderr())
-				cmd.SilenceUsage = true
-				return err
-			}
-			return nil
-		},
-		RunE: CmdMigrateDiffRun,
-	}
-	// MigrateHashCmd represents the 'atlas migrate hash' command.
-	MigrateHashCmd = &cobra.Command{
-		Use:   "hash",
-		Short: "Hash (re-)creates an integrity hash file for the migration directory.",
-		Long: `'atlas migrate hash' computes the integrity hash sum of the migration directory and stores it in the atlas.sum file.
-This command should be used whenever a manual change in the migration directory was made.`,
-		Example: `  atlas migrate hash`,
-		PersistentPreRunE: func(cmd *cobra.Command, _ []string) error {
-			if err := migrateFlagsFromEnv(cmd, nil); err != nil {
-				return err
-			}
-			if err := dirFormatBC(); err != nil {
-				return err
-			}
-			return nil
-		},
-		RunE: CmdMigrateHashRun,
-	}
-	// MigrateImportCmd represents the 'atlas migrate import' command.
-	MigrateImportCmd = &cobra.Command{
-		Use:     "import",
-		Short:   "Import a migration directory from another migration management tool to the Atlas format.",
-		Example: `  atlas migrate import --from file:///path/to/source/directory?format=liquibase --to file:///path/to/migration/directory`,
-		// Validate the source directory. Consider a directory with no sum file valid, since it might be an import
-		// from an existing project.
-		PersistentPreRunE: func(cmd *cobra.Command, _ []string) error {
-			if err := migrateFlagsFromEnv(cmd, nil); err != nil {
-				return err
-			}
-			if err := dirFormatBC(); err != nil {
-				return err
-			}
-			dir, err := dir(MigrateFlags.Import.FromURL, false)
-			if err != nil {
-				return err
-			}
-			if err := migrate.Validate(dir); err != nil && !errors.Is(err, migrate.ErrChecksumNotFound) {
-				printChecksumErr(cmd.OutOrStderr())
-				cmd.SilenceUsage = true
-				return err
-			}
-			return nil
-		},
-		RunE: CmdMigrateImportRun,
-	}
-	// MigrateNewCmd represents the 'atlas migrate new' command.
-	MigrateNewCmd = &cobra.Command{
-		Use:     "new [name]",
-		Short:   "Creates a new empty migration file in the migration directory.",
-		Long:    `'atlas migrate new' creates a new migration according to the configured formatter without any statements in it.`,
-		Example: `  atlas migrate new my-new-migration`,
-		Args:    cobra.MaximumNArgs(1),
-		RunE:    CmdMigrateNewRun,
-	}
-	// MigrateSetCmd represents the 'atlas migrate set' command.
-	MigrateSetCmd = &cobra.Command{
-		Use:   "set <version> [flags]",
-		Short: "Set the current version of the migration history table.",
-		Long: `'atlas migrate set' edits the revision table to consider all migrations up to and including the given version
-to be applied. This command is usually used after manually making changes to the managed database.`,
-		Example: `  atlas migrate set-revision 3 --url mysql://user:pass@localhost:3306/
-  atlas migrate set-revision 4 --env local
-  atlas migrate set-revision 1.2.4 --url mysql://user:pass@localhost:3306/my_db --revision-schema my_revisions`,
-		Args: cobra.ExactArgs(1),
-		RunE: CmdMigrateSetRun,
-	}
-	// MigrateStatusCmd represents the 'atlas migrate status' command.
-	MigrateStatusCmd = &cobra.Command{
-		Use:   "status [flags]",
-		Short: "Get information about the current migration status.",
-		Long:  `'atlas migrate status' reports information about the current status of a connected database compared to the migration directory.`,
-		Example: `  atlas migrate status --url mysql://user:pass@localhost:3306/
-  atlas migrate status --url mysql://user:pass@localhost:3306/ --dir file:///path/to/migration/directory`,
-		RunE: CmdMigrateStatusRun,
-	}
-	// MigrateValidateCmd represents the 'atlas migrate validate' command.
-	MigrateValidateCmd = &cobra.Command{
-		Use:   "validate [flags]",
-		Short: "Validates the migration directories checksum and SQL statements.",
-		Long: `'atlas migrate validate' computes the integrity hash sum of the migration directory and compares it to the
-atlas.sum file. If there is a mismatch it will be reported. If the --dev-url flag is given, the migration
-files are executed on the connected database in order to validate SQL semantics.`,
-		Example: `  atlas migrate validate
-  atlas migrate validate --dir file:///path/to/migration/directory
-  atlas migrate validate --dir file:///path/to/migration/directory --dev-url mysql://user:pass@localhost:3306/dev
-  atlas migrate validate --env dev`,
-		RunE: CmdMigrateValidateRun,
-	}
-	// MigrateLintCmd represents the 'atlas migrate lint' command.
-	MigrateLintCmd = &cobra.Command{
-		Use:   "lint",
-		Short: "Run analysis on the migration directory",
-		Example: `  atlas migrate lint --env dev
-  atlas migrate lint --dir file:///path/to/migration/directory --dev-url mysql://root:pass@localhost:3306 --latest 1
-  atlas migrate lint --dir file:///path/to/migration/directory --dev-url mysql://root:pass@localhost:3306 --git-base master
-  atlas migrate lint --dir file:///path/to/migration/directory --dev-url mysql://root:pass@localhost:3306 --log '{{ json .Files }}'`,
-		// Override the parent 'migrate' pre-run function to allow executing
-		// 'migrate lint' on directories that are not maintained by Atlas.
-		PersistentPreRunE: func(cmd *cobra.Command, _ []string) error {
-			if err := migrateFlagsFromEnv(cmd, nil); err != nil {
-				return err
-			}
-			if err := dirFormatBC(); err != nil {
-				return err
-			}
-			return nil
-		},
-		RunE: CmdMigrateLintRun,
-	}
-)
-
-func init() {
-	// Add sub-commands.
-	Root.AddCommand(MigrateCmd)
-	MigrateCmd.AddCommand(MigrateApplyCmd)
-	MigrateCmd.AddCommand(MigrateDiffCmd)
-	MigrateCmd.AddCommand(MigrateHashCmd)
-	MigrateCmd.AddCommand(MigrateNewCmd)
-	MigrateCmd.AddCommand(MigrateValidateCmd)
-	MigrateCmd.AddCommand(MigrateStatusCmd)
-	MigrateCmd.AddCommand(MigrateLintCmd)
-	MigrateCmd.AddCommand(MigrateImportCmd)
-	MigrateCmd.AddCommand(MigrateSetCmd)
-	// Reusable flags.
-	urlFlag := func(f *string, name, short string, set *pflag.FlagSet) {
-		set.StringVarP(f, name, short, "", "[driver://username:password@address/dbname?param=value] select a database using the URL format")
-	}
-	revisionsFlag := func(set *pflag.FlagSet) {
-		set.StringVarP(&MigrateFlags.RevisionSchema, migrateFlagRevisionsSchema, "", "", "schema name where the revisions table resides")
-	}
-	dirURLFlag := func(f *string, name, short string, set *pflag.FlagSet) {
-		set.StringVarP(f, name, short, "file://migrations", "select migration directory using URL format")
-	}
-	// Global flags.
-	dirURLFlag(&MigrateFlags.DirURL, migrateFlagDir, "", MigrateCmd.PersistentFlags())
-	MigrateCmd.PersistentFlags().StringSliceVarP(&MigrateFlags.Schemas, migrateFlagSchema, "", nil, "set schema names")
-	MigrateCmd.PersistentFlags().StringVarP(&MigrateFlags.DirFormat, migrateFlagDirFormat, "", formatAtlas, "set migration file format")
-	MigrateCmd.PersistentFlags().SortFlags = false
-	// Apply flags.
-	MigrateApplyCmd.Flags().StringVarP(&MigrateFlags.Apply.LogFormat, migrateFlagLog, "", logFormatTTY, "log format to use")
-	revisionsFlag(MigrateApplyCmd.Flags())
-	MigrateApplyCmd.Flags().BoolVarP(&MigrateFlags.Apply.DryRun, migrateFlagDryRun, "", false, "do not actually execute any SQL but show it on screen")
-	MigrateApplyCmd.Flags().StringVarP(&MigrateFlags.Apply.FromVersion, migrateFlagFrom, "", "", "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().StringVarP(&MigrateFlags.Apply.TxMode, migrateApplyTxMode, "", txModeFile, "set transaction mode [none, file, all]")
-	MigrateApplyCmd.Flags().BoolVarP(&MigrateFlags.Apply.AllowDirty, migrateApplyAllowDirty, "", false, "allow start working on a non-clean database")
-	urlFlag(&MigrateFlags.URL, migrateFlagURL, "u", MigrateApplyCmd.Flags())
-	MigrateApplyCmd.Flags().SortFlags = false
-	cobra.CheckErr(MigrateApplyCmd.MarkFlagRequired(migrateFlagURL))
-	MigrateApplyCmd.MarkFlagsMutuallyExclusive(migrateFlagFrom, migrateApplyBaselineVersion)
-	cobra.CheckErr(MigrateApplyCmd.Flags().MarkHidden(migrateFlagSchema))
-	// Diff flags.
-	urlFlag(&MigrateFlags.DevURL, migrateFlagDevURL, "", MigrateDiffCmd.Flags())
-	MigrateDiffCmd.Flags().StringSliceVarP(&MigrateFlags.ToURLs, migrateFlagTo, "", nil, "[driver://username:password@address/dbname?param=value ...] select a desired state using the URL format")
-	MigrateDiffCmd.Flags().StringVarP(&MigrateFlags.Diff.Qualifier, migrateDiffQualifier, "", "", "qualify tables with custom qualifier when working on a single schema")
-	MigrateDiffCmd.Flags().SortFlags = false
-	cobra.CheckErr(MigrateDiffCmd.MarkFlagRequired(migrateFlagDevURL))
-	cobra.CheckErr(MigrateDiffCmd.MarkFlagRequired(migrateFlagTo))
-	// Import flags.
-	dirURLFlag(&MigrateFlags.Import.FromURL, migrateFlagFrom, "", MigrateImportCmd.Flags())
-	dirURLFlag(&MigrateFlags.Import.ToURL, migrateFlagTo, "", MigrateImportCmd.Flags())
-	MigrateImportCmd.Flags().SortFlags = false
-	// Validate flags.
-	urlFlag(&MigrateFlags.DevURL, migrateFlagDevURL, "", MigrateValidateCmd.Flags())
-	// Status flags.
-	urlFlag(&MigrateFlags.URL, migrateFlagURL, "u", MigrateStatusCmd.Flags())
-	revisionsFlag(MigrateStatusCmd.Flags())
-	MigrateStatusCmd.Flags().StringVarP(&MigrateFlags.Status.Format, migrateFlagLog, "", "", "custom logging using a Go template")
-	// Set flags.
-	urlFlag(&MigrateFlags.URL, migrateFlagURL, "u", MigrateSetCmd.Flags())
-	// Hash flags.
-	MigrateHashCmd.Flags().Bool("force", false, "")
-	cobra.CheckErr(MigrateHashCmd.Flags().MarkDeprecated("force", "you can safely omit it."))
-	// Lint flags.
-	urlFlag(&MigrateFlags.DevURL, migrateFlagDevURL, "", MigrateLintCmd.Flags())
-	MigrateLintCmd.PersistentFlags().StringVarP(&MigrateFlags.Lint.Format, migrateFlagLog, "", "", "custom logging using a Go template")
-	MigrateLintCmd.PersistentFlags().UintVarP(&MigrateFlags.Lint.Latest, migrateLintLatest, "", 0, "run analysis on the latest N migration files")
-	MigrateLintCmd.PersistentFlags().StringVarP(&MigrateFlags.Lint.GitBase, migrateLintGitBase, "", "", "run analysis against the base Git branch")
-	MigrateLintCmd.PersistentFlags().StringVarP(&MigrateFlags.Lint.GitDir, migrateLintGitDir, "", ".", "path to the repository working directory")
-	cobra.CheckErr(MigrateLintCmd.MarkFlagRequired(migrateFlagDevURL))
-	receivesEnv(MigrateCmd)
+			Args: cobra.MaximumNArgs(1),
+			PreRunE: func(cmd *cobra.Command, args []string) error {
+				if err := migrateFlagsFromEnv(cmd); err != nil {
+					return err
+				}
+				return checkDir(cmd, flags.dirURL, false)
+			},
+			RunE: func(cmd *cobra.Command, args []string) error {
+				return migrateApplyRun(cmd, args, flags)
+			},
+		}
+	)
+	cmd.Flags().SortFlags = false
+	addFlagURL(cmd.Flags(), &flags.url)
+	addFlagDirURL(cmd.Flags(), &flags.dirURL)
+	addFlagLog(cmd.Flags(), &flags.logFormat)
+	addFlagRevisionSchema(cmd.Flags(), &flags.revisionSchema)
+	addFlagDryRun(cmd.Flags(), &flags.dryRun)
+	cmd.Flags().StringVarP(&flags.fromVersion, flagFrom, "", "", "calculate pending files from the given version (including it)")
+	cmd.Flags().StringVarP(&flags.baselineVersion, flagBaseline, "", "", "start the first migration after the given baseline version")
+	cmd.Flags().StringVarP(&flags.txMode, flagTxMode, "", txModeFile, "set transaction mode [none, file, all]")
+	cmd.Flags().BoolVarP(&flags.allowDirty, flagAllowDirty, "", false, "allow start working on a non-clean database")
+	cmd.MarkFlagsMutuallyExclusive(flagFrom, flagBaseline)
+	cobra.CheckErr(cmd.MarkFlagRequired(flagURL))
+	return cmd
 }
 
-const applyLockValue = "atlas_migrate_execute"
-
-// CmdMigrateApplyRun is the command executed when running the CLI with 'migrate apply' args.
-func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
+// migrateApplyCmd represents the 'atlas migrate apply' subcommand.
+func migrateApplyRun(cmd *cobra.Command, args []string, flags migrateApplyFlags) error {
 	var (
 		n   int
 		err error
@@ -368,12 +134,12 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 		}
 	}
 	// Open the migration directory.
-	dir, err := dir(MigrateFlags.DirURL, false)
+	dir, err := dir(flags.dirURL, false)
 	if err != nil {
 		return err
 	}
 	// Open a client to the database.
-	c, err := sqlclient.Open(cmd.Context(), MigrateFlags.URL)
+	c, err := sqlclient.Open(cmd.Context(), flags.url)
 	if err != nil {
 		return err
 	}
@@ -387,16 +153,16 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 		// If unlocking fails notify the user about it.
 		defer cobra.CheckErr(unlock())
 	}
-	if err := checkRevisionSchemaClarity(cmd, c); err != nil {
+	if err := checkRevisionSchemaClarity(cmd, c, flags.revisionSchema); err != nil {
 		return err
 	}
 	// Get the correct log format and destination. Currently, only os.Stdout is supported.
-	l, err := logFormat(cmd.OutOrStdout())
+	l, err := logFormat(cmd.OutOrStdout(), flags.logFormat)
 	if err != nil {
 		return err
 	}
 	var rrw migrate.RevisionReadWriter
-	rrw, err = entRevisions(cmd.Context(), c)
+	rrw, err = entRevisions(cmd.Context(), c, flags.revisionSchema)
 	if err != nil {
 		return err
 	}
@@ -404,7 +170,20 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 		return err
 	}
 	// Determine pending files and lock the database while working.
-	ex, err := migrate.NewExecutor(c.Driver, dir, rrw, executorOptions(l)...)
+	opts := []migrate.ExecutorOption{
+		migrate.WithLogger(l),
+		migrate.WithOperatorVersion(operatorVersion()),
+	}
+	if flags.allowDirty {
+		opts = append(opts, migrate.WithAllowDirty(true))
+	}
+	if v := flags.baselineVersion; v != "" {
+		opts = append(opts, migrate.WithBaselineVersion(v))
+	}
+	if v := flags.fromVersion; v != "" {
+		opts = append(opts, migrate.WithFromVersion(v))
+	}
+	ex, err := migrate.NewExecutor(c.Driver, dir, rrw, opts...)
 	if err != nil {
 		return err
 	}
@@ -431,7 +210,13 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 		return err
 	}
 	var (
-		mux = tx{c: c, rrw: rrw}
+		mux = tx{
+			dryRun: flags.dryRun,
+			mode:   flags.txMode,
+			schema: flags.revisionSchema,
+			c:      c,
+			rrw:    rrw,
+		}
 		drv migrate.Driver
 	)
 	for _, f := range pending {
@@ -439,7 +224,7 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 		if err != nil {
 			return err
 		}
-		ex, err := migrate.NewExecutor(drv, dir, rrw, executorOptions(l)...)
+		ex, err := migrate.NewExecutor(drv, dir, rrw, opts...)
 		if err != nil {
 			return err
 		}
@@ -457,179 +242,60 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 	return mux.commit()
 }
 
-func checkRevisionSchemaClarity(cmd *cobra.Command, c *sqlclient.Client) error {
-	// The "old" default  behavior for the revision schema location was to store the revision table in its own schema.
-	// Now, the table is saved in the connected schema, if any. To keep the backwards compatability, we now require
-	// for schema bound connections to have the schema-revision flag present if there is no revision table in the schema
-	// but the old default schema does have one.
-	if c.URL.Schema != "" && MigrateFlags.RevisionSchema == "" {
-		// If the schema does not contain a revision table, but we can find a table in the previous default schema,
-		// abort and tell the user to specify the intention.
-		opts := &schema.InspectOptions{Tables: []string{revision.Table}}
-		s, err := c.InspectSchema(cmd.Context(), "", opts)
-		var ok bool
-		switch {
-		case schema.IsNotExistError(err):
-			// If the schema does not exist, the table does not as well.
-		case err != nil:
-			return err
-		default:
-			// Connected schema does exist, check if the table does.
-			_, ok = s.Table(revision.Table)
-		}
-		if !ok { // Either schema or table does not exist.
-			// Check for the old default schema. If it does not exist, we have no problem.
-			s, err := c.InspectSchema(cmd.Context(), defaultRevisionSchema, opts)
-			switch {
-			case schema.IsNotExistError(err):
-				// Schema does not exist, we can proceed.
-			case err != nil:
-				return err
-			default:
-				if _, ok := s.Table(revision.Table); ok {
-					fmt.Fprintf(cmd.OutOrStderr(),
-						`We couldn't find a revision table in the connected schema but found one in 
-the schema 'atlas_schema_revisions' and cannot determine the desired behavior.
-
-As a safety guard, we require you to specify whether to use the existing
-table in 'atlas_schema_revisions' or create a new one in the connected schema
-by providing the '--revisions-schema' flag or deleting the 'atlas_schema_revisions'
-schema if it is unused.
-
-`)
-					cmd.SilenceUsage = true
-					cmd.SilenceErrors = true
-					return errors.New("ambiguous revision table")
-				}
-			}
-		}
-	}
-	return nil
-}
-
-func entRevisions(ctx context.Context, c *sqlclient.Client) (*cmdmigrate.EntRevisions, error) {
-	return cmdmigrate.NewEntRevisions(ctx, c, cmdmigrate.WithSchema(revisionSchemaName(c)))
-}
-
-// defaultRevisionSchema is the default schema for storing revisions table.
-const defaultRevisionSchema = "atlas_schema_revisions"
-
-func revisionSchemaName(c *sqlclient.Client) string {
-	switch {
-	case MigrateFlags.RevisionSchema != "":
-		return MigrateFlags.RevisionSchema
-	case c.URL.Schema != "":
-		return c.URL.Schema
-	default:
-		return defaultRevisionSchema
-	}
-}
-
-// tx handles wrapping migration execution in transactions.
-type tx struct {
-	c   *sqlclient.Client
-	tx  *sqlclient.TxClient
-	rrw migrate.RevisionReadWriter
-}
-
-// driver returns the migrate.Driver to use to execute migration statements.
-func (tx *tx) driver(ctx context.Context) (migrate.Driver, migrate.RevisionReadWriter, error) {
-	if MigrateFlags.Apply.DryRun {
-		// If the --dry-run flag is given we don't want to execute any statements on the database.
-		return &dryRunDriver{tx.c.Driver}, &dryRunRevisions{tx.rrw}, nil
-	}
-	switch MigrateFlags.Apply.TxMode {
-	case txModeNone:
-		return tx.c.Driver, tx.rrw, nil
-	case txModeFile:
-		// In file-mode, this function is called each time a new file is executed. Open a transaction.
-		if tx.tx != nil {
-			return nil, nil, errors.New("unexpected active transaction")
-		}
-		var err error
-		tx.tx, err = tx.c.Tx(ctx, nil)
-		if err != nil {
-			return nil, nil, err
-		}
-		tx.rrw, err = entRevisions(ctx, tx.tx.Client)
-		if err != nil {
-			return nil, nil, err
-		}
-		return tx.tx.Driver, tx.rrw, nil
-	case txModeAll:
-		// In file-mode, this function is called each time a new file is executed. Since we wrap all files into one
-		// huge transaction, if there already is an opened one, use that.
-		if tx.tx == nil {
-			var err error
-			tx.tx, err = tx.c.Tx(ctx, nil)
-			if err != nil {
-				return nil, nil, err
-			}
-			tx.rrw, err = entRevisions(ctx, tx.tx.Client)
-			if err != nil {
-				return nil, nil, err
-			}
-		}
-		return tx.tx.Driver, tx.rrw, nil
-	default:
-		return nil, nil, fmt.Errorf("unknown tx-mode %q", MigrateFlags.Apply.TxMode)
-	}
+type migrateDiffFlags struct {
+	desiredURLs       []string
+	dirURL, dirFormat string
+	devURL            string
+	schemas           []string
+	revisionSchema    string // revision schema name
+	qualifier         string // optional table qualifier
 }
 
-// mayRollback may roll back a transaction depending on the given transaction mode.
-func (tx *tx) mayRollback(err error) error {
-	if tx.tx != nil && err != nil {
-		if err2 := tx.tx.Rollback(); err2 != nil {
-			err = fmt.Errorf("%v: %w", err2, err)
+// migrateDiffCmd represents the 'atlas migrate diff' subcommand.
+func migrateDiffCmd() *cobra.Command {
+	var (
+		flags migrateDiffFlags
+		cmd   = &cobra.Command{
+			Use:   "diff [flags] [name]",
+			Short: "Compute the diff between the migration directory and a desired state and create a new migration file.",
+			Long: `'atlas migrate diff' uses the dev-database to re-run all migration files in the migration directory, compares
+it to a given desired state and create a new migration file containing SQL statements to migrate the migration
+directory state to the desired schema. The desired state can be another connected database or an HCL file.`,
+			Example: `  atlas migrate diff --dev-url mysql://user:pass@localhost:3306/dev --to file://schema.hcl
+  atlas migrate diff --dev-url mysql://user:pass@localhost:3306/dev --to file://atlas.hcl add_users_table
+  atlas migrate diff --dev-url mysql://user:pass@localhost:3306/dev --to mysql://user:pass@localhost:3306/dbname
+  atlas migrate diff --env dev`,
+			Args: cobra.MaximumNArgs(1),
+			PreRunE: func(cmd *cobra.Command, args []string) error {
+				if err := migrateFlagsFromEnv(cmd); err != nil {
+					return err
+				}
+				if err := dirFormatBC(flags.dirFormat, &flags.dirURL); err != nil {
+					return err
+				}
+				return checkDir(cmd, flags.dirURL, true)
+			},
+			RunE: func(cmd *cobra.Command, args []string) error {
+				return migrateDiffRun(cmd, args, flags)
+			},
 		}
-	}
-	return err
-}
-
-// mayCommit may commit a transaction depending on the given transaction mode.
-func (tx *tx) mayCommit() error {
-	// Only commit if each file is wrapped in a transaction.
-	if !MigrateFlags.Apply.DryRun && MigrateFlags.Apply.TxMode == txModeFile {
-		return tx.commit()
-	}
-	return nil
-}
-
-// commit the transaction, if one is active.
-func (tx *tx) commit() error {
-	if tx.tx == nil {
-		return nil
-	}
-	defer func() { tx.tx = nil }()
-	return tx.tx.Commit()
-}
-
-func executorOptions(l migrate.Logger) []migrate.ExecutorOption {
-	opts := []migrate.ExecutorOption{
-		migrate.WithLogger(l),
-		migrate.WithOperatorVersion(operatorVersion()),
-	}
-	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
-}
-
-func operatorVersion() string {
-	v, _ := parse(version)
-	return "Atlas CLI " + v
+	)
+	cmd.Flags().SortFlags = false
+	addFlagURLs(cmd.Flags(), &flags.desiredURLs, flagTo)
+	addFlagDevURL(cmd.Flags(), &flags.devURL)
+	addFlagDirURL(cmd.Flags(), &flags.dirURL)
+	addFlagDirFormat(cmd.Flags(), &flags.dirFormat)
+	addFlagRevisionSchema(cmd.Flags(), &flags.revisionSchema)
+	addFlagSchemas(cmd.Flags(), &flags.schemas)
+	cmd.Flags().StringVar(&flags.qualifier, flagQualifier, "", "qualify tables with custom qualifier when working on a single schema")
+	cobra.CheckErr(cmd.MarkFlagRequired(flagTo))
+	cobra.CheckErr(cmd.MarkFlagRequired(flagDevURL))
+	return cmd
 }
 
-// CmdMigrateDiffRun is the command executed when running the CLI with 'migrate diff' args.
-func CmdMigrateDiffRun(cmd *cobra.Command, args []string) error {
+func migrateDiffRun(cmd *cobra.Command, args []string, flags migrateDiffFlags) error {
 	// Open a dev driver.
-	dev, err := sqlclient.Open(cmd.Context(), MigrateFlags.DevURL)
+	dev, err := sqlclient.Open(cmd.Context(), flags.devURL)
 	if err != nil {
 		return err
 	}
@@ -644,7 +310,7 @@ func CmdMigrateDiffRun(cmd *cobra.Command, args []string) error {
 		defer cobra.CheckErr(unlock())
 	}
 	// Open the migration directory.
-	u, err := url.Parse(MigrateFlags.DirURL)
+	u, err := url.Parse(flags.dirURL)
 	if err != nil {
 		return err
 	}
@@ -657,7 +323,7 @@ func CmdMigrateDiffRun(cmd *cobra.Command, args []string) error {
 		return err
 	}
 	// Get a state reader for the desired state.
-	desired, err := to(cmd.Context(), dev)
+	desired, err := toTarget(cmd.Context(), dev, flags.desiredURLs, flags.schemas)
 	if err != nil {
 		return err
 	}
@@ -665,7 +331,7 @@ func CmdMigrateDiffRun(cmd *cobra.Command, args []string) error {
 	opts := []migrate.PlannerOption{migrate.PlanFormat(f)}
 	if dev.URL.Schema != "" {
 		// Disable tables qualifier in schema-mode.
-		opts = append(opts, migrate.PlanWithSchemaQualifier(MigrateFlags.Diff.Qualifier))
+		opts = append(opts, migrate.PlanWithSchemaQualifier(flags.qualifier))
 	}
 	// Plan the changes and create a new migration file.
 	pl := migrate.NewPlanner(dev.Driver, dir, opts...)
@@ -694,34 +360,98 @@ func CmdMigrateDiffRun(cmd *cobra.Command, args []string) error {
 	}
 }
 
-// CmdMigrateHashRun is the command executed when running the CLI with 'migrate hash' args.
-func CmdMigrateHashRun(*cobra.Command, []string) error {
-	dir, err := dir(MigrateFlags.DirURL, false)
-	if err != nil {
-		return err
-	}
-	sum, err := dir.Checksum()
-	if err != nil {
-		return err
-	}
-	return migrate.WriteSumFile(dir, sum)
+type migrateHashFlags struct{ dirURL, dirFormat string }
+
+// migrateHashCmd represents the 'atlas migrate hash' subcommand.
+func migrateHashCmd() *cobra.Command {
+	var (
+		flags migrateHashFlags
+		cmd   = &cobra.Command{
+			Use:   "hash [flags]",
+			Short: "Hash (re-)creates an integrity hash file for the migration directory.",
+			Long: `'atlas migrate hash' computes the integrity hash sum of the migration directory and stores it in the atlas.sum file.
+This command should be used whenever a manual change in the migration directory was made.`,
+			Example: `  atlas migrate hash`,
+			PreRunE: func(cmd *cobra.Command, args []string) error {
+				if err := migrateFlagsFromEnv(cmd); err != nil {
+					return err
+				}
+				return dirFormatBC(flags.dirFormat, &flags.dirURL)
+			},
+			RunE: func(cmd *cobra.Command, args []string) error {
+				dir, err := dir(flags.dirURL, false)
+				if err != nil {
+					return err
+				}
+				sum, err := dir.Checksum()
+				if err != nil {
+					return err
+				}
+				return migrate.WriteSumFile(dir, sum)
+			},
+		}
+	)
+	addFlagDirURL(cmd.Flags(), &flags.dirURL)
+	addFlagDirFormat(cmd.Flags(), &flags.dirFormat)
+	cmd.Flags().Bool("force", false, "")
+	cobra.CheckErr(cmd.Flags().MarkDeprecated("force", "you can safely omit it."))
+	return cmd
 }
 
-// CmdMigrateImportRun is the command executed when running the CLI with 'migrate import' args.
-func CmdMigrateImportRun(cmd *cobra.Command, _ []string) error {
-	p, err := url.Parse(MigrateFlags.Import.FromURL)
+type migrateImportFlags struct{ fromURL, toURL, dirFormat string }
+
+// migrateImportCmd represents the 'atlas migrate import' subcommand.
+func migrateImportCmd() *cobra.Command {
+	var (
+		flags migrateImportFlags
+		cmd   = &cobra.Command{
+			Use:     "import [flags]",
+			Short:   "Import a migration directory from another migration management tool to the Atlas format.",
+			Example: `  atlas migrate import --from file:///path/to/source/directory?format=liquibase --to file:///path/to/migration/directory`,
+			// Validate the source directory. Consider a directory with no sum file
+			// valid, since it might be an import from an existing project.
+			PreRunE: func(cmd *cobra.Command, _ []string) error {
+				if err := migrateFlagsFromEnv(cmd); err != nil {
+					return err
+				}
+				if err := dirFormatBC(flags.dirFormat, &flags.fromURL); err != nil {
+					return err
+				}
+				d, err := dir(flags.fromURL, false)
+				if err != nil {
+					return err
+				}
+				if err = migrate.Validate(d); err != nil && !errors.Is(err, migrate.ErrChecksumNotFound) {
+					printChecksumError(cmd)
+					return err
+				}
+				return nil
+			},
+			RunE: func(cmd *cobra.Command, args []string) error {
+				return migrateImportRun(cmd, args, flags)
+			},
+		}
+	)
+	cmd.Flags().SortFlags = false
+	addFlagDirURL(cmd.Flags(), &flags.fromURL, flagFrom)
+	addFlagDirURL(cmd.Flags(), &flags.toURL, flagTo)
+	addFlagDirFormat(cmd.Flags(), &flags.dirFormat)
+	return cmd
+}
+
+func migrateImportRun(cmd *cobra.Command, _ []string, flags migrateImportFlags) error {
+	p, err := url.Parse(flags.fromURL)
 	if err != nil {
 		return err
 	}
 	if f := p.Query().Get("format"); f == "" || f == formatAtlas {
 		return fmt.Errorf("cannot import a migration directory already in %q format", formatAtlas)
 	}
-	src, err := dir(MigrateFlags.Import.FromURL, false)
+	src, err := dir(flags.fromURL, false)
 	if err != nil {
 		return err
 	}
-	MigrateFlags.DirFormat = formatAtlas // remove assignment once the --dir-format flag is removed
-	trgt, err := dir(MigrateFlags.Import.ToURL, true)
+	trgt, err := dir(flags.toURL, true)
 	if err != nil {
 		return err
 	}
@@ -784,12 +514,145 @@ func CmdMigrateImportRun(cmd *cobra.Command, _ []string) error {
 	if err != nil {
 		return err
 	}
-	return migrate.WriteSumFile(trgt, sum)
+	return migrate.WriteSumFile(trgt, sum)
+}
+
+type migrateLintFlags struct {
+	dirURL, dirFormat string
+	devURL            string
+	logFormat         string
+	latest            uint
+	gitBase, gitDir   string
+}
+
+// migrateLintCmd represents the 'atlas migrate lint' subcommand.
+func migrateLintCmd() *cobra.Command {
+	var (
+		flags migrateLintFlags
+		cmd   = &cobra.Command{
+			Use:   "lint [flags]",
+			Short: "Run analysis on the migration directory",
+			Example: `  atlas migrate lint --env dev
+  atlas migrate lint --dir file:///path/to/migration/directory --dev-url mysql://root:pass@localhost:3306 --latest 1
+  atlas migrate lint --dir file:///path/to/migration/directory --dev-url mysql://root:pass@localhost:3306 --git-base master
+  atlas migrate lint --dir file:///path/to/migration/directory --dev-url mysql://root:pass@localhost:3306 --log '{{ json .Files }}'`,
+			PreRunE: func(cmd *cobra.Command, args []string) error {
+				if err := migrateFlagsFromEnv(cmd); err != nil {
+					return err
+				}
+				return dirFormatBC(flags.dirFormat, &flags.dirURL)
+			},
+			RunE: func(cmd *cobra.Command, args []string) error {
+				return migrateLintRun(cmd, args, flags)
+			},
+		}
+	)
+	cmd.Flags().SortFlags = false
+	addFlagDevURL(cmd.Flags(), &flags.devURL)
+	addFlagDirURL(cmd.Flags(), &flags.dirURL)
+	cmd.Flags().StringVarP(&flags.logFormat, flagLog, "", "", "custom logging using a Go template")
+	cmd.Flags().UintVarP(&flags.latest, flagLatest, "", 0, "run analysis on the latest N migration files")
+	cmd.Flags().StringVarP(&flags.gitBase, flagGitBase, "", "", "run analysis against the base Git branch")
+	cmd.Flags().StringVarP(&flags.gitDir, flagGitDir, "", ".", "path to the repository working directory")
+	cobra.CheckErr(cmd.MarkFlagRequired(flagDevURL))
+	return cmd
+}
+
+func migrateLintRun(cmd *cobra.Command, _ []string, flags migrateLintFlags) error {
+	dev, err := sqlclient.Open(cmd.Context(), flags.devURL)
+	if err != nil {
+		return err
+	}
+	defer dev.Close()
+	dir, err := dir(flags.dirURL, false)
+	if err != nil {
+		return err
+	}
+	var detect lint.ChangeDetector
+	switch {
+	case flags.latest == 0 && flags.gitBase == "":
+		return fmt.Errorf("--%s or --%s is required", flagLatest, flagGitBase)
+	case flags.latest > 0 && flags.gitBase != "":
+		return fmt.Errorf("--%s and --%s are mutually exclusive", flagLatest, flagGitBase)
+	case flags.latest > 0:
+		detect = lint.LatestChanges(dir, int(flags.latest))
+	case flags.gitBase != "":
+		detect, err = lint.NewGitChangeDetector(
+			dir,
+			lint.WithWorkDir(flags.gitDir),
+			lint.WithBase(flags.gitBase),
+			lint.WithMigrationsPath(dir.(interface{ Path() string }).Path()),
+		)
+		if err != nil {
+			return err
+		}
+	}
+	format := lint.DefaultTemplate
+	if f := flags.logFormat; f != "" {
+		format, err = template.New("format").Funcs(lint.TemplateFuncs).Parse(f)
+		if err != nil {
+			return fmt.Errorf("parse log format: %w", err)
+		}
+	}
+	env, err := selectEnv(GlobalFlags.SelectedEnv)
+	if err != nil {
+		return err
+	}
+	az, err := sqlcheck.AnalyzerFor(dev.Name, env.Lint.Remain())
+	if err != nil {
+		return err
+	}
+	r := &lint.Runner{
+		Dev:            dev,
+		Dir:            dir,
+		ChangeDetector: detect,
+		ReportWriter: &lint.TemplateWriter{
+			T: format,
+			W: cmd.OutOrStdout(),
+		},
+		Analyzers: az,
+	}
+	err = r.Run(cmd.Context())
+	// Print the error in case it was not printed before.
+	cmd.SilenceErrors = errors.As(err, &lint.SilentError{})
+	cmd.SilenceUsage = cmd.SilenceErrors
+	return err
+}
+
+type migrateNewFlags struct{ dirURL, dirFormat string }
+
+// migrateNewCmd represents the 'atlas migrate new' subcommand.
+func migrateNewCmd() *cobra.Command {
+	var (
+		flags migrateNewFlags
+		cmd   = &cobra.Command{
+			Use:     "new [flags] [name]",
+			Short:   "Creates a new empty migration file in the migration directory.",
+			Long:    `'atlas migrate new' creates a new migration according to the configured formatter without any statements in it.`,
+			Example: `  atlas migrate new my-new-migration`,
+			Args:    cobra.MaximumNArgs(1),
+			PreRunE: func(cmd *cobra.Command, _ []string) error {
+				if err := migrateFlagsFromEnv(cmd); err != nil {
+					return err
+				}
+				if err := dirFormatBC(flags.dirFormat, &flags.dirURL); err != nil {
+					return err
+				}
+				return checkDir(cmd, flags.dirURL, false)
+			},
+			RunE: func(cmd *cobra.Command, args []string) error {
+				return migrateNewRun(cmd, args, flags)
+			},
+		}
+	)
+	cmd.Flags().SortFlags = false
+	addFlagDirURL(cmd.Flags(), &flags.dirURL)
+	addFlagDirFormat(cmd.Flags(), &flags.dirFormat)
+	return cmd
 }
 
-// CmdMigrateNewRun is the command executed when running the CLI with 'migrate new' args.
-func CmdMigrateNewRun(_ *cobra.Command, args []string) error {
-	u, err := url.Parse(MigrateFlags.DirURL)
+func migrateNewRun(_ *cobra.Command, args []string, flags migrateNewFlags) error {
+	u, err := url.Parse(flags.dirURL)
 	if err != nil {
 		return err
 	}
@@ -808,9 +671,49 @@ func CmdMigrateNewRun(_ *cobra.Command, args []string) error {
 	return migrate.NewPlanner(nil, dir, migrate.PlanFormat(f)).WritePlan(&migrate.Plan{Name: name})
 }
 
-// CmdMigrateSetRun is the command executed when running the CLI with 'migrate set' args.
-func CmdMigrateSetRun(cmd *cobra.Command, args []string) error {
-	dir, err := dir(MigrateFlags.DirURL, false)
+type migrateSetFlags struct {
+	url               string
+	dirURL, dirFormat string
+	revisionSchema    string
+}
+
+// migrateSetCmd represents the 'atlas migrate set' subcommand.
+func migrateSetCmd() *cobra.Command {
+	var (
+		flags migrateSetFlags
+		cmd   = &cobra.Command{
+			Use:   "set [flags] <version>",
+			Short: "Set the current version of the migration history table.",
+			Long: `'atlas migrate set' edits the revision table to consider all migrations up to and including the given version
+to be applied. This command is usually used after manually making changes to the managed database.`,
+			Example: `  atlas migrate set-revision 3 --url mysql://user:pass@localhost:3306/
+  atlas migrate set-revision 4 --env local
+  atlas migrate set-revision 1.2.4 --url mysql://user:pass@localhost:3306/my_db --revision-schema my_revisions`,
+			Args: cobra.ExactArgs(1),
+			PreRunE: func(cmd *cobra.Command, _ []string) error {
+				if err := migrateFlagsFromEnv(cmd); err != nil {
+					return err
+				}
+				if err := dirFormatBC(flags.dirFormat, &flags.dirURL); err != nil {
+					return err
+				}
+				return checkDir(cmd, flags.dirURL, false)
+			},
+			RunE: func(cmd *cobra.Command, args []string) error {
+				return migrateSetRun(cmd, args, flags)
+			},
+		}
+	)
+	cmd.Flags().SortFlags = false
+	addFlagURL(cmd.Flags(), &flags.url)
+	addFlagDirURL(cmd.Flags(), &flags.dirURL)
+	addFlagDirFormat(cmd.Flags(), &flags.dirFormat)
+	addFlagRevisionSchema(cmd.Flags(), &flags.revisionSchema)
+	return cmd
+}
+
+func migrateSetRun(cmd *cobra.Command, args []string, flags migrateSetFlags) error {
+	dir, err := dir(flags.dirURL, false)
 	if err != nil {
 		return err
 	}
@@ -824,7 +727,7 @@ func CmdMigrateSetRun(cmd *cobra.Command, args []string) error {
 	}); idx == -1 {
 		return fmt.Errorf("migration with version %q not found", args[0])
 	}
-	client, err := sqlclient.Open(cmd.Context(), MigrateFlags.URL)
+	client, err := sqlclient.Open(cmd.Context(), flags.url)
 	if err != nil {
 		return err
 	}
@@ -838,11 +741,11 @@ func CmdMigrateSetRun(cmd *cobra.Command, args []string) error {
 		// If unlocking fails notify the user about it.
 		defer cobra.CheckErr(unlock())
 	}
-	if err := checkRevisionSchemaClarity(cmd, client); err != nil {
+	if err := checkRevisionSchemaClarity(cmd, client, flags.revisionSchema); err != nil {
 		return err
 	}
 	// Ensure revision table exists.
-	rrw, err := entRevisions(cmd.Context(), client)
+	rrw, err := entRevisions(cmd.Context(), client, flags.revisionSchema)
 	if err != nil {
 		return err
 	}
@@ -854,7 +757,7 @@ func CmdMigrateSetRun(cmd *cobra.Command, args []string) error {
 	if err != nil {
 		return err
 	}
-	rrw, err = entRevisions(cmd.Context(), tx.Client)
+	rrw, err = entRevisions(cmd.Context(), tx.Client, flags.revisionSchema)
 	if err != nil {
 		return err
 	}
@@ -942,22 +845,60 @@ func CmdMigrateSetRun(cmd *cobra.Command, args []string) error {
 	return tx.Commit()
 }
 
-// CmdMigrateStatusRun is the command executed when running the CLI with 'migrate status' args.
-func CmdMigrateStatusRun(cmd *cobra.Command, _ []string) error {
-	dir, err := dir(MigrateFlags.DirURL, false)
+type migrateStatusFlags struct {
+	url               string
+	dirURL, dirFormat string
+	revisionSchema    string
+	logFormat         string
+}
+
+// migrateStatusCmd represents the 'atlas migrate status' subcommand.
+func migrateStatusCmd() *cobra.Command {
+	var (
+		flags migrateStatusFlags
+		cmd   = &cobra.Command{
+			Use:   "status [flags]",
+			Short: "Get information about the current migration status.",
+			Long:  `'atlas migrate status' reports information about the current status of a connected database compared to the migration directory.`,
+			Example: `  atlas migrate status --url mysql://user:pass@localhost:3306/
+  atlas migrate status --url mysql://user:pass@localhost:3306/ --dir file:///path/to/migration/directory`,
+			PreRunE: func(cmd *cobra.Command, _ []string) error {
+				if err := migrateFlagsFromEnv(cmd); err != nil {
+					return err
+				}
+				if err := dirFormatBC(flags.dirFormat, &flags.dirURL); err != nil {
+					return err
+				}
+				return checkDir(cmd, flags.dirURL, false)
+			},
+			RunE: func(cmd *cobra.Command, args []string) error {
+				return migrateStatusRun(cmd, args, flags)
+			},
+		}
+	)
+	cmd.Flags().SortFlags = false
+	addFlagURL(cmd.Flags(), &flags.url)
+	addFlagDirURL(cmd.Flags(), &flags.dirURL)
+	addFlagDirFormat(cmd.Flags(), &flags.dirFormat)
+	addFlagRevisionSchema(cmd.Flags(), &flags.revisionSchema)
+	return cmd
+}
+
+func migrateStatusRun(cmd *cobra.Command, _ []string, flags migrateStatusFlags) error {
+	dir, err := dir(flags.dirURL, false)
 	if err != nil {
 		return err
 	}
-	client, err := sqlclient.Open(cmd.Context(), MigrateFlags.URL)
+	client, err := sqlclient.Open(cmd.Context(), flags.url)
 	if err != nil {
 		return err
 	}
 	defer client.Close()
-	if err := checkRevisionSchemaClarity(cmd, client); err != nil {
+	if err := checkRevisionSchemaClarity(cmd, client, flags.revisionSchema); err != nil {
 		return err
 	}
 	var format = cmdmigrate.DefaultTemplate
-	if f := MigrateFlags.Status.Format; f != "" {
+	if f := flags.logFormat; f != "" {
 		format, err = template.New("format").Funcs(cmdmigrate.TemplateFuncs).Parse(f)
 		if err != nil {
 			return fmt.Errorf("parse log format: %w", err)
@@ -967,25 +908,64 @@ func CmdMigrateStatusRun(cmd *cobra.Command, _ []string) error {
 		Client:       client,
 		Dir:          dir,
 		ReportWriter: &cmdmigrate.TemplateWriter{T: format, W: cmd.OutOrStdout()},
-		Schema:       revisionSchemaName(client),
+		Schema:       revisionSchemaName(client, flags.revisionSchema),
 	}).Status(cmd.Context())
 }
 
-// CmdMigrateValidateRun is the command executed when running the CLI with 'migrate validate' args.
-func CmdMigrateValidateRun(cmd *cobra.Command, _ []string) error {
+type migrateValidateFlags struct {
+	devURL            string
+	dirURL, dirFormat string
+}
+
+// migrateValidateCmd represents the 'atlas migrate validate' subcommand.
+func migrateValidateCmd() *cobra.Command {
+	var (
+		flags migrateValidateFlags
+		cmd   = &cobra.Command{
+			Use:   "validate [flags]",
+			Short: "Validates the migration directories checksum and SQL statements.",
+			Long: `'atlas migrate validate' computes the integrity hash sum of the migration directory and compares it to the
+atlas.sum file. If there is a mismatch it will be reported. If the --dev-url flag is given, the migration
+files are executed on the connected database in order to validate SQL semantics.`,
+			Example: `  atlas migrate validate
+  atlas migrate validate --dir file:///path/to/migration/directory
+  atlas migrate validate --dir file:///path/to/migration/directory --dev-url mysql://user:pass@localhost:3306/dev
+  atlas migrate validate --env dev`,
+			PreRunE: func(cmd *cobra.Command, _ []string) error {
+				if err := migrateFlagsFromEnv(cmd); err != nil {
+					return err
+				}
+				if err := dirFormatBC(flags.dirFormat, &flags.dirURL); err != nil {
+					return err
+				}
+				return checkDir(cmd, flags.dirURL, false)
+			},
+			RunE: func(cmd *cobra.Command, args []string) error {
+				return migrateValidateRun(cmd, args, flags)
+			},
+		}
+	)
+	cmd.Flags().SortFlags = false
+	addFlagDevURL(cmd.Flags(), &flags.devURL)
+	addFlagDirURL(cmd.Flags(), &flags.dirURL)
+	addFlagDirFormat(cmd.Flags(), &flags.dirFormat)
+	return cmd
+}
+
+func migrateValidateRun(cmd *cobra.Command, _ []string, flags migrateValidateFlags) error {
 	// Validating the integrity is done by the PersistentPreRun already.
-	if MigrateFlags.DevURL == "" {
+	if flags.devURL == "" {
 		// If there is no --dev-url given do not attempt to replay the migration directory.
 		return nil
 	}
 	// Open a client for the dev-db.
-	dev, err := sqlclient.Open(cmd.Context(), MigrateFlags.DevURL)
+	dev, err := sqlclient.Open(cmd.Context(), flags.devURL)
 	if err != nil {
 		return err
 	}
 	defer dev.Close()
 	// Currently, only our own migration file format is supported.
-	dir, err := dir(MigrateFlags.DirURL, false)
+	dir, err := dir(flags.dirURL, false)
 	if err != nil {
 		return err
 	}
@@ -1004,65 +984,74 @@ func CmdMigrateValidateRun(cmd *cobra.Command, _ []string) error {
 	return nil
 }
 
-// CmdMigrateLintRun is the command executed when running the CLI with 'migrate lint' args.
-func CmdMigrateLintRun(cmd *cobra.Command, _ []string) error {
-	dev, err := sqlclient.Open(cmd.Context(), MigrateFlags.DevURL)
-	if err != nil {
-		return err
-	}
-	defer dev.Close()
-	dir, err := dir(MigrateFlags.DirURL, false)
-	if err != nil {
-		return err
-	}
-	var detect lint.ChangeDetector
-	switch {
-	case MigrateFlags.Lint.Latest == 0 && MigrateFlags.Lint.GitBase == "":
-		return fmt.Errorf("--%s or --%s is required", migrateLintLatest, migrateLintGitBase)
-	case MigrateFlags.Lint.Latest > 0 && MigrateFlags.Lint.GitBase != "":
-		return fmt.Errorf("--%s and --%s are mutually exclusive", migrateLintLatest, migrateLintGitBase)
-	case MigrateFlags.Lint.Latest > 0:
-		detect = lint.LatestChanges(dir, int(MigrateFlags.Lint.Latest))
-	case MigrateFlags.Lint.GitBase != "":
-		detect, err = lint.NewGitChangeDetector(
-			dir,
-			lint.WithWorkDir(MigrateFlags.Lint.GitDir),
-			lint.WithBase(MigrateFlags.Lint.GitBase),
-			lint.WithMigrationsPath(dir.(interface{ Path() string }).Path()),
-		)
-		if err != nil {
+const applyLockValue = "atlas_migrate_execute"
+
+func checkRevisionSchemaClarity(cmd *cobra.Command, c *sqlclient.Client, revisionSchemaFlag string) error {
+	// The "old" default  behavior for the revision schema location was to store the revision table in its own schema.
+	// Now, the table is saved in the connected schema, if any. To keep the backwards compatability, we now require
+	// for schema bound connections to have the schema-revision flag present if there is no revision table in the schema
+	// but the old default schema does have one.
+	if c.URL.Schema != "" && revisionSchemaFlag == "" {
+		// If the schema does not contain a revision table, but we can find a table in the previous default schema,
+		// abort and tell the user to specify the intention.
+		opts := &schema.InspectOptions{Tables: []string{revision.Table}}
+		s, err := c.InspectSchema(cmd.Context(), "", opts)
+		var ok bool
+		switch {
+		case schema.IsNotExistError(err):
+			// If the schema does not exist, the table does not as well.
+		case err != nil:
 			return err
+		default:
+			// Connected schema does exist, check if the table does.
+			_, ok = s.Table(revision.Table)
 		}
-	}
-	format := lint.DefaultTemplate
-	if f := MigrateFlags.Lint.Format; f != "" {
-		format, err = template.New("format").Funcs(lint.TemplateFuncs).Parse(f)
-		if err != nil {
-			return fmt.Errorf("parse log format: %w", err)
+		if !ok { // Either schema or table does not exist.
+			// Check for the old default schema. If it does not exist, we have no problem.
+			s, err := c.InspectSchema(cmd.Context(), defaultRevisionSchema, opts)
+			switch {
+			case schema.IsNotExistError(err):
+				// Schema does not exist, we can proceed.
+			case err != nil:
+				return err
+			default:
+				if _, ok := s.Table(revision.Table); ok {
+					fmt.Fprintf(cmd.OutOrStderr(),
+						`We couldn't find a revision table in the connected schema but found one in 
+the schema 'atlas_schema_revisions' and cannot determine the desired behavior.
+
+As a safety guard, we require you to specify whether to use the existing
+table in 'atlas_schema_revisions' or create a new one in the connected schema
+by providing the '--revisions-schema' flag or deleting the 'atlas_schema_revisions'
+schema if it is unused.
+
+`)
+					cmd.SilenceUsage = true
+					cmd.SilenceErrors = true
+					return errors.New("ambiguous revision table")
+				}
+			}
 		}
 	}
-	env, err := selectEnv(GlobalFlags.SelectedEnv)
-	if err != nil {
-		return err
-	}
-	az, err := sqlcheck.AnalyzerFor(dev.Name, env.Lint.Remain())
-	if err != nil {
-		return err
-	}
-	r := &lint.Runner{
-		Dev:            dev,
-		Dir:            dir,
-		ChangeDetector: detect,
-		ReportWriter: &lint.TemplateWriter{
-			T: format,
-			W: cmd.OutOrStdout(),
-		},
-		Analyzers: az,
+	return nil
+}
+
+func entRevisions(ctx context.Context, c *sqlclient.Client, flag string) (*cmdmigrate.EntRevisions, error) {
+	return cmdmigrate.NewEntRevisions(ctx, c, cmdmigrate.WithSchema(revisionSchemaName(c, flag)))
+}
+
+// defaultRevisionSchema is the default schema for storing revisions table.
+const defaultRevisionSchema = "atlas_schema_revisions"
+
+func revisionSchemaName(c *sqlclient.Client, flag string) string {
+	switch {
+	case flag != "":
+		return flag
+	case c.URL.Schema != "":
+		return c.URL.Schema
+	default:
+		return defaultRevisionSchema
 	}
-	err = r.Run(cmd.Context())
-	// Print the error in case it was not printed before.
-	cmd.SilenceErrors = errors.As(err, &lint.SilentError{})
-	return err
 }
 
 const (
@@ -1071,16 +1060,90 @@ const (
 	txModeFile = "file"
 )
 
-func printChecksumErr(out io.Writer) {
-	fmt.Fprintf(out, `You have a checksum error in your migration directory.
-This happens if you manually create or edit a migration file.
-Please check your migration files and run
+// tx handles wrapping migration execution in transactions.
+type tx struct {
+	dryRun       bool
+	mode, schema string
+	c            *sqlclient.Client
+	tx           *sqlclient.TxClient
+	rrw          migrate.RevisionReadWriter
+}
 
-'atlas migrate hash'
+// driver returns the migrate.Driver to use to execute migration statements.
+func (tx *tx) driver(ctx context.Context) (migrate.Driver, migrate.RevisionReadWriter, error) {
+	if tx.dryRun {
+		// If the --dry-run flag is given we don't want to execute any statements on the database.
+		return &dryRunDriver{tx.c.Driver}, &dryRunRevisions{tx.rrw}, nil
+	}
+	switch tx.mode {
+	case txModeNone:
+		return tx.c.Driver, tx.rrw, nil
+	case txModeFile:
+		// In file-mode, this function is called each time a new file is executed. Open a transaction.
+		if tx.tx != nil {
+			return nil, nil, errors.New("unexpected active transaction")
+		}
+		var err error
+		tx.tx, err = tx.c.Tx(ctx, nil)
+		if err != nil {
+			return nil, nil, err
+		}
+		tx.rrw, err = entRevisions(ctx, tx.tx.Client, tx.schema)
+		if err != nil {
+			return nil, nil, err
+		}
+		return tx.tx.Driver, tx.rrw, nil
+	case txModeAll:
+		// In file-mode, this function is called each time a new file is executed. Since we wrap all files into one
+		// huge transaction, if there already is an opened one, use that.
+		if tx.tx == nil {
+			var err error
+			tx.tx, err = tx.c.Tx(ctx, nil)
+			if err != nil {
+				return nil, nil, err
+			}
+			tx.rrw, err = entRevisions(ctx, tx.tx.Client, tx.schema)
+			if err != nil {
+				return nil, nil, err
+			}
+		}
+		return tx.tx.Driver, tx.rrw, nil
+	default:
+		return nil, nil, fmt.Errorf("unknown tx-mode %q", tx.mode)
+	}
+}
 
-to re-hash the contents and resolve the error
+// mayRollback may roll back a transaction depending on the given transaction mode.
+func (tx *tx) mayRollback(err error) error {
+	if tx.tx != nil && err != nil {
+		if err2 := tx.tx.Rollback(); err2 != nil {
+			err = fmt.Errorf("%v: %w", err2, err)
+		}
+	}
+	return err
+}
 
-`)
+// mayCommit may commit a transaction depending on the given transaction mode.
+func (tx *tx) mayCommit() error {
+	// Only commit if each file is wrapped in a transaction.
+	if !tx.dryRun && tx.mode == txModeFile {
+		return tx.commit()
+	}
+	return nil
+}
+
+// commit the transaction, if one is active.
+func (tx *tx) commit() error {
+	if tx.tx == nil {
+		return nil
+	}
+	defer func() { tx.tx = nil }()
+	return tx.tx.Commit()
+}
+
+func operatorVersion() string {
+	v, _ := parse(version)
+	return "Atlas CLI " + v
 }
 
 // dir parses u and calls dirURL.
@@ -1132,19 +1195,15 @@ func dirURL(u *url.URL, create bool) (migrate.Dir, error) {
 }
 
 // dirFormatBC ensures the soon-to-be deprecated --dir-format flag gets set on all migration directory URLs.
-func dirFormatBC() error {
-	for _, s := range []*string{
-		&MigrateFlags.DirURL,
-		&MigrateFlags.Import.FromURL,
-		&MigrateFlags.Import.ToURL,
-	} {
+func dirFormatBC(flag string, urls ...*string) error {
+	for _, s := range urls {
 		u, err := url.Parse(*s)
 		if err != nil {
 			return err
 		}
-		if !u.Query().Has("format") && MigrateFlags.DirFormat != "" {
+		if !u.Query().Has("format") && flag != "" {
 			q := u.Query()
-			q.Set("format", MigrateFlags.DirFormat)
+			q.Set("format", flag)
 			u.RawQuery = q.Encode()
 			*s = u.String()
 		}
@@ -1152,6 +1211,31 @@ func dirFormatBC() error {
 	return nil
 }
 
+func checkDir(cmd *cobra.Command, url string, create bool) error {
+	d, err := dir(url, create)
+	if err != nil {
+		return err
+	}
+	if err = migrate.Validate(d); err != nil {
+		printChecksumError(cmd)
+		return err
+	}
+	return nil
+}
+
+func printChecksumError(cmd *cobra.Command) {
+	fmt.Fprintf(cmd.OutOrStderr(), `You have a checksum error in your migration directory.
+This happens if you manually create or edit a migration file.
+Please check your migration files and run
+
+'atlas migrate hash'
+
+to re-hash the contents and resolve the error
+
+`)
+	cmd.SilenceUsage = true
+}
+
 type target struct {
 	migrate.StateReader        // desired state.
 	io.Closer                  // optional close function.
@@ -1159,17 +1243,16 @@ type target struct {
 }
 
 // to returns a migrate.StateReader for the given to flag.
-func to(ctx context.Context, dev *sqlclient.Client) (*target, error) {
-	scheme, err := selectScheme(MigrateFlags.ToURLs)
+func toTarget(ctx context.Context, dev *sqlclient.Client, urls, schemas []string) (*target, error) {
+	scheme, err := selectScheme(urls)
 	if err != nil {
 		return nil, err
 	}
-	schemas := MigrateFlags.Schemas
 	switch scheme {
 	case "file": // hcl file
 		realm := &schema.Realm{}
-		paths := make([]string, 0, len(MigrateFlags.ToURLs))
-		for _, u := range MigrateFlags.ToURLs {
+		paths := make([]string, 0, len(urls))
+		for _, u := range urls {
 			paths = append(paths, strings.TrimPrefix(u, "file://"))
 		}
 		parsed, err := parseHCLPaths(paths...)
@@ -1209,7 +1292,7 @@ func to(ctx context.Context, dev *sqlclient.Client) (*target, error) {
 		}
 		return t, nil
 	default: // database connection
-		client, err := sqlclient.Open(ctx, MigrateFlags.ToURLs[0])
+		client, err := sqlclient.Open(ctx, urls[0])
 		if err != nil {
 			return nil, err
 		}
@@ -1426,16 +1509,16 @@ func zero(v int) int {
 	return v
 }
 
-func logFormat(out io.Writer) (migrate.Logger, error) {
-	switch l := MigrateFlags.Apply.LogFormat; l {
-	case logFormatTTY:
+func logFormat(out io.Writer, format string) (migrate.Logger, error) {
+	switch l := format; l {
+	case "", logFormatTTY:
 		return &LogTTY{out: out}, nil
 	default:
 		return nil, fmt.Errorf("unknown log-format %q", l)
 	}
 }
 
-func migrateFlagsFromEnv(cmd *cobra.Command, _ []string) error {
+func migrateFlagsFromEnv(cmd *cobra.Command) error {
 	activeEnv, err := selectEnv(GlobalFlags.SelectedEnv)
 	if err != nil {
 		return err
@@ -1443,34 +1526,34 @@ func migrateFlagsFromEnv(cmd *cobra.Command, _ []string) error {
 	if err := inputValsFromEnv(cmd); err != nil {
 		return err
 	}
-	if err := maySetFlag(cmd, migrateFlagDevURL, activeEnv.DevURL); err != nil {
+	if err := maySetFlag(cmd, flagDevURL, activeEnv.DevURL); err != nil {
 		return err
 	}
-	if err := maySetFlag(cmd, migrateFlagDir, activeEnv.Migration.Dir); err != nil {
+	if err := maySetFlag(cmd, flagDirURL, activeEnv.Migration.Dir); err != nil {
 		return err
 	}
-	if err := maySetFlag(cmd, migrateFlagDirFormat, activeEnv.Migration.Format); err != nil {
+	if err := maySetFlag(cmd, flagDirFormat, activeEnv.Migration.Format); err != nil {
 		return err
 	}
 	switch cmd.Name() {
 	case "lint":
-		if err := maySetFlag(cmd, migrateFlagLog, activeEnv.Lint.Log); err != nil {
+		if err := maySetFlag(cmd, flagLog, activeEnv.Lint.Log); err != nil {
 			return err
 		}
-		if err := maySetFlag(cmd, migrateLintLatest, strconv.Itoa(activeEnv.Lint.Latest)); err != nil {
+		if err := maySetFlag(cmd, flagLatest, strconv.Itoa(activeEnv.Lint.Latest)); err != nil {
 			return err
 		}
-		if err := maySetFlag(cmd, migrateLintGitDir, activeEnv.Lint.Git.Dir); err != nil {
+		if err := maySetFlag(cmd, flagGitDir, activeEnv.Lint.Git.Dir); err != nil {
 			return err
 		}
-		if err := maySetFlag(cmd, migrateLintGitBase, activeEnv.Lint.Git.Base); err != nil {
+		if err := maySetFlag(cmd, flagGitBase, activeEnv.Lint.Git.Base); err != nil {
 			return err
 		}
 	default:
-		if err := maySetFlag(cmd, migrateFlagURL, activeEnv.URL); err != nil {
+		if err := maySetFlag(cmd, flagURL, activeEnv.URL); err != nil {
 			return err
 		}
-		if err := maySetFlag(cmd, migrateFlagRevisionsSchema, activeEnv.Migration.RevisionsSchema); err != nil {
+		if err := maySetFlag(cmd, flagRevisionSchema, activeEnv.Migration.RevisionsSchema); err != nil {
 			return err
 		}
 	}
@@ -1485,11 +1568,11 @@ func migrateFlagsFromEnv(cmd *cobra.Command, _ []string) error {
 		}
 		srcs[i] = "file://" + s
 	}
-	if err := maySetFlag(cmd, migrateFlagTo, strings.Join(srcs, ",")); err != nil {
+	if err := maySetFlag(cmd, flagTo, strings.Join(srcs, ",")); err != nil {
 		return err
 	}
 	if s := "[" + strings.Join(activeEnv.Schemas, "") + "]"; len(activeEnv.Schemas) > 0 {
-		if err := maySetFlag(cmd, migrateFlagSchema, s); err != nil {
+		if err := maySetFlag(cmd, flagSchema, s); err != nil {
 			return err
 		}
 	}
diff --git a/cmd/atlas/internal/cmdapi/schema.go b/cmd/atlas/internal/cmdapi/schema.go
index 64c2d6c0..d095a570 100644
--- a/cmd/atlas/internal/cmdapi/schema.go
+++ b/cmd/atlas/internal/cmdapi/schema.go
@@ -6,7 +6,6 @@ package cmdapi
 
 import (
 	"bytes"
-	"errors"
 	"fmt"
 	"io/fs"
 	"os"
@@ -20,58 +19,51 @@ import (
 	"github.com/hashicorp/hcl/v2/hclwrite"
 	"github.com/manifoldco/promptui"
 	"github.com/spf13/cobra"
-	"github.com/zclconf/go-cty/cty"
 )
 
-const (
-	urlFlag         = "url"
-	schemaFlag      = "schema"
-	excludeFlag     = "exclude"
-	devURLFlag      = "dev-url"
-	fileFlag        = "file"
-	dsnFlag         = "dsn"
-	varFlag         = "var"
-	autoApproveFlag = "auto-approve"
-)
+func init() {
+	schemaCmd := schemaCmd()
+	schemaCmd.AddCommand(
+		schemaApplyCmd(),
+		schemaCleanCmd(),
+		schemaDiffCmd(),
+		schemaFmtCmd(),
+		schemaInspectCmd(),
+	)
+	Root.AddCommand(schemaCmd)
+}
 
-var (
-	// schemaCmd represents the subcommand 'atlas schema'.
-	schemaCmd = &cobra.Command{
+// schemaCmd represents the subcommand 'atlas schema'.
+func schemaCmd() *cobra.Command {
+	cmd := &cobra.Command{
 		Use:   "schema",
 		Short: "Work with atlas schemas.",
-		Long:  "The `atlas schema` command groups subcommands for working with Atlas schemas.",
-	}
-
-	// SchemaFlags are common flags used by schema commands.
-	SchemaFlags struct {
-		URL     string
-		Schemas []string
-		Exclude []string
-
-		// Deprecated: DSN is an alias for URL.
-		DSN string
-	}
-
-	// ApplyFlags are the flags used in SchemaApply command.
-	ApplyFlags struct {
-		DevURL      string
-		Paths       []string
-		DryRun      bool
-		AutoApprove bool
+		Long:  "The `atlas schema` command groups subcommands working with declarative Atlas schemas.",
 	}
+	addFlagEnvVar(cmd.PersistentFlags())
+	return cmd
+}
 
-	// CleanFlags are the flags used in SchemaClean command.
-	CleanFlags struct {
-		URL         string
-		AutoApprove bool
-	}
+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.
+}
 
-	// SchemaApply represents the 'atlas schema apply' subcommand command.
-	SchemaApply = &cobra.Command{
-		Use:   "apply",
-		Short: "Apply an atlas schema to a target database.",
-		// Use 80-columns as max width.
-		Long: `'atlas schema apply' plans and executes a database migration to bring a given
+// schemaApplyCmd represents the 'atlas schema apply' subcommand.
+func schemaApplyCmd() *cobra.Command {
+	var (
+		flags schemaApplyFlags
+		cmd   = &cobra.Command{
+			Use:   "apply",
+			Short: "Apply an atlas schema to a target database.",
+			// Use 80-columns as max width.
+			Long: `'atlas schema apply' plans and executes a database migration to bring a given
 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.
 
@@ -84,235 +76,131 @@ 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.`,
-		PreRunE: schemaFlagsFromEnv,
-		RunE:    CmdApplyRun,
-		Example: `  atlas schema apply -u "mysql://user:pass@localhost/dbname" -f atlas.hcl
+			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`,
-	}
-
-	// SchemaClean represents the 'atlas schema clean' subcommand.
-	SchemaClean = &cobra.Command{
-		Use:   "clean [flags]",
-		Short: "Removes all objects from the connected database.",
-		Long: `'atlas schema clean' drops all objects in the connected database and leaves it in an empty state.
-As a safety feature, 'atlas schema clean' will ask for confirmation before attempting to execute any SQL.`,
-		Example: `  atlas schema clean -u mysql://user:pass@localhost:3306/dbname
-  atlas schema clean -u mysql://user:pass@localhost:3306/`,
-		PreRunE: schemaFlagsFromEnv,
-		RunE:    CmdCleanRun,
-	}
-
-	// SchemaInspect represents the 'atlas schema inspect' subcommand.
-	SchemaInspect = &cobra.Command{
-		Use:   "inspect",
-		Short: "Inspect a database and print its schema in Atlas DDL syntax.",
-		Long: `'atlas schema inspect' connects to the given database and inspects its schema.
-It then prints to the screen the schema of that database in Atlas DDL syntax. This output can be
-saved to a file, commonly by redirecting the output to a file named with a ".hcl" suffix:
-
-  atlas schema inspect -u "mysql://user:pass@localhost:3306/dbname" > schema.hcl
-
-This file can then be edited and used with the` + " `atlas schema apply` " + `command to plan
-and execute schema migrations against the given database. In cases where users wish to inspect
-all multiple schemas in a given database (for instance a MySQL server may contain multiple named
-databases), omit the relevant part from the url, e.g. "mysql://user:pass@localhost:3306/".
-To select specific schemas from the databases, users may use the "--schema" (or "-s" shorthand)
-flag.
-	`,
-		PreRunE: schemaFlagsFromEnv,
-		RunE:    CmdInspectRun,
-		Example: `  atlas schema inspect -u "mysql://user:pass@localhost:3306/dbname"
-  atlas schema inspect -u "mariadb://user:pass@localhost:3306/" --schema=schemaA,schemaB -s schemaC
-  atlas schema inspect --url "postgres://user:pass@host:port/dbname?sslmode=disable"
-  atlas schema inspect -u "sqlite://file:ex1.db?_fk=1"`,
-	}
-
-	// SchemaFmt represents the 'atlas schema fmt' subcommand.
-	SchemaFmt = &cobra.Command{
-		Use:   "fmt [path ...]",
-		Short: "Formats Atlas HCL files",
-		Long: `'atlas schema fmt' formats all ".hcl" files under the given path using
-canonical HCL layout style as defined by the github.com/hashicorp/hcl/v2/hclwrite package.
-Unless stated otherwise, the fmt command will use the current directory.
-
-After running, the command will print the names of the files it has formatted. If all
-files in the directory are formatted, no input will be printed out.
-`,
-		Run: CmdFmtRun,
-	}
-)
-
-const (
-	answerApply = "Apply"
-	answerAbort = "Abort"
-)
-
-func init() {
-	// Common flags.
-	receivesEnv(schemaCmd)
-
-	// Schema apply flags.
-	schemaCmd.AddCommand(SchemaApply)
-	SchemaApply.Flags().SortFlags = false
-	SchemaApply.Flags().StringSliceVarP(&ApplyFlags.Paths, fileFlag, "f", nil, "[paths...] file or directory containing the HCL files")
-	SchemaApply.Flags().StringVarP(&SchemaFlags.URL, urlFlag, "u", "", "URL to the database using the format:\n[driver://username:password@address/dbname?param=value]")
-	SchemaApply.Flags().StringSliceVarP(&SchemaFlags.Exclude, excludeFlag, "", nil, "List of glob patterns used to filter resources from applying.")
-	SchemaApply.Flags().StringSliceVarP(&SchemaFlags.Schemas, schemaFlag, "s", nil, "Set schema names.")
-	SchemaApply.Flags().StringVarP(&ApplyFlags.DevURL, devURLFlag, "", "", "URL for the dev database. Used to validate schemas and calculate diffs\nbefore running migration.")
-	SchemaApply.Flags().BoolVarP(&ApplyFlags.DryRun, "dry-run", "", false, "Dry-run. Print SQL plan without prompting for execution.")
-	SchemaApply.Flags().BoolVarP(&ApplyFlags.AutoApprove, autoApproveFlag, "", false, "Auto approve. Apply the schema changes without prompting for approval.")
-	SchemaApply.Flags().StringVarP(&SchemaFlags.DSN, dsnFlag, "d", "", "")
-	cobra.CheckErr(SchemaApply.Flags().MarkHidden(dsnFlag))
-	cobra.CheckErr(SchemaApply.MarkFlagRequired(urlFlag))
-	cobra.CheckErr(SchemaApply.MarkFlagRequired(fileFlag))
-
-	// Schema clean flags.
-	SchemaClean.Flags().StringVarP(&CleanFlags.URL, urlFlag, "u", "", "URL to the database using the format:\n[driver://username:password@address/dbname?param=value]")
-	SchemaClean.Flags().BoolVarP(&CleanFlags.AutoApprove, autoApproveFlag, "", false, "Auto approve. Apply the schema changes without prompting for approval.")
-	cobra.CheckErr(SchemaClean.MarkFlagRequired(urlFlag))
-
-	// Schema inspect flags.
-	schemaCmd.AddCommand(SchemaInspect)
-	SchemaInspect.Flags().StringVarP(&SchemaFlags.URL, urlFlag, "u", "", "[driver://username:password@protocol(address)/dbname?param=value] select a database using the URL format")
-	SchemaInspect.Flags().StringSliceVarP(&SchemaFlags.Schemas, schemaFlag, "s", nil, "Set schema name")
-	SchemaInspect.Flags().StringSliceVarP(&SchemaFlags.Exclude, excludeFlag, "", nil, "List of glob patterns used to filter resources from inspection")
-	SchemaInspect.Flags().StringVarP(&SchemaFlags.DSN, dsnFlag, "d", "", "")
-	cobra.CheckErr(SchemaInspect.Flags().MarkHidden(dsnFlag))
-	cobra.CheckErr(SchemaInspect.MarkFlagRequired(urlFlag))
-
-	// Schema fmt.
-	schemaCmd.AddCommand(SchemaFmt)
-	schemaCmd.AddCommand(SchemaClean)
-}
-
-// selectEnv returns the Env from the current project file based on the selected
-// argument. If selected is "", or no project file exists in the current directory
-// a zero-value Env is returned.
-func selectEnv(selected string) (*Env, error) {
-	env := &Env{
-		Lint:      &Lint{},
-		Migration: &Migration{},
-	}
-	if selected == "" {
-		return env, nil
-	}
-	if _, err := os.Stat(projectFileName); os.IsNotExist(err) {
-		return nil, fmt.Errorf("project file %q was not found", projectFileName)
-	}
-	return LoadEnv(projectFileName, selected, WithInput(GlobalFlags.Vars))
+			PreRunE: schemaFlagsFromEnv,
+			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)
+	cobra.CheckErr(cmd.MarkFlagRequired(flagURL))
+	cobra.CheckErr(cmd.MarkFlagRequired(flagFile))
+	return cmd
 }
 
-func schemaFlagsFromEnv(cmd *cobra.Command, _ []string) error {
-	activeEnv, err := selectEnv(GlobalFlags.SelectedEnv)
+func schemaApplyRun(cmd *cobra.Command, _ []string, flags schemaApplyFlags) error {
+	client, err := sqlclient.Open(cmd.Context(), flags.URL)
 	if err != nil {
 		return err
 	}
-	if err := inputValsFromEnv(cmd); err != nil {
-		return err
-	}
-	if err := dsn2url(cmd); err != nil {
-		return err
-	}
-	if err := maySetFlag(cmd, urlFlag, activeEnv.URL); err != nil {
-		return err
+	defer client.Close()
+	schemas, ctx := flags.Schemas, cmd.Context()
+	if client.URL.Schema != "" {
+		schemas = append(schemas, client.URL.Schema)
 	}
-	if err := maySetFlag(cmd, devURLFlag, activeEnv.DevURL); err != nil {
+	realm, err := client.InspectRealm(ctx, &schema.InspectRealmOption{
+		Schemas: schemas,
+		Exclude: flags.Exclude,
+	})
+	if err != nil {
 		return err
 	}
-	srcs, err := activeEnv.Sources()
+	desired := &schema.Realm{}
+	parsed, err := parseHCLPaths(flags.Paths...)
 	if err != nil {
 		return err
 	}
-	if err := maySetFlag(cmd, fileFlag, strings.Join(srcs, "")); err != nil {
+	if err := client.Eval(parsed, desired, GlobalFlags.Vars); err != nil {
 		return err
 	}
-	if s := strings.Join(activeEnv.Schemas, ","); s != "" {
-		if err := maySetFlag(cmd, schemaFlag, s); 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 s := strings.Join(activeEnv.Exclude, ","); s != "" {
-		if err := maySetFlag(cmd, excludeFlag, s); err != nil {
+	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
 		}
 	}
-	return nil
-}
-
-// maySetFlag sets the flag with the provided name to envVal if such a flag exists
-// on the cmd, it was not set by the user via the command line and if envVal is not
-// an empty string.
-func maySetFlag(cmd *cobra.Command, name, envVal string) error {
-	fl := cmd.Flag(name)
-	if fl == nil {
-		return nil
-	}
-	if fl.Changed {
-		return nil
-	}
-	if envVal == "" {
-		return nil
-	}
-	return cmd.Flags().Set(name, envVal)
-}
-
-func dsn2url(cmd *cobra.Command) error {
-	dsnF, urlF := cmd.Flag(dsnFlag), cmd.Flag(urlFlag)
-	switch {
-	case dsnF == nil:
-	case dsnF.Changed && urlF.Changed:
-		return errors.New(`both flags "url" and "dsn" were set`)
-	case dsnF.Changed && !urlF.Changed:
-		return cmd.Flags().Set(urlFlag, dsnF.Value.String())
-	}
-	return nil
-}
-
-// CmdInspectRun is the command used when running CLI.
-func CmdInspectRun(cmd *cobra.Command, _ []string) error {
-	// Create the client.
-	client, err := sqlclient.Open(cmd.Context(), SchemaFlags.URL)
+	changes, err := client.RealmDiff(realm, desired)
 	if err != nil {
 		return err
 	}
-	defer client.Close()
-	schemas := SchemaFlags.Schemas
-	if client.URL.Schema != "" {
-		schemas = append(schemas, client.URL.Schema)
+	if len(changes) == 0 {
+		cmd.Println("Schema is synced, no changes to be made")
+		return nil
 	}
-	s, err := client.InspectRealm(cmd.Context(), &schema.InspectRealmOption{
-		Schemas: schemas,
-		Exclude: SchemaFlags.Exclude,
-	})
-	if err != nil {
+	if err := summary(cmd, client, changes); err != nil {
 		return err
 	}
-	ddl, err := client.MarshalSpec(s)
-	if err != nil {
-		return err
+	if !flags.DryRun && (flags.AutoApprove || promptUser()) {
+		if err := client.ApplyChanges(ctx, changes); err != nil {
+			return err
+		}
 	}
-	cmd.Print(string(ddl))
 	return nil
 }
 
-// CmdApplyRun is the command used when running CLI.
-func CmdApplyRun(cmd *cobra.Command, _ []string) error {
-	c, err := sqlclient.Open(cmd.Context(), SchemaFlags.URL)
-	if err != nil {
-		return err
-	}
-	defer c.Close()
-	return applyRun(cmd, c, ApplyFlags.DevURL, ApplyFlags.Paths, ApplyFlags.DryRun, ApplyFlags.AutoApprove, GlobalFlags.Vars)
+type schemeCleanFlags struct {
+	URL         string // URL of database to apply the changes on.
+	AutoApprove bool   // Don't prompt for approval before applying SQL.
+}
+
+// schemaCleanCmd represents the 'atlas schema clean' subcommand.
+func schemaCleanCmd() *cobra.Command {
+	var (
+		flags schemeCleanFlags
+		cmd   = &cobra.Command{
+			Use:   "clean [flags]",
+			Short: "Removes all objects from the connected database.",
+			Long: `'atlas schema clean' drops all objects in the connected database and leaves it in an empty state.
+As a safety feature, 'atlas schema clean' will ask for confirmation before attempting to execute any SQL.`,
+			Example: `  atlas schema clean -u mysql://user:pass@localhost:3306/dbname
+  atlas schema clean -u mysql://user:pass@localhost:3306/`,
+			PreRunE: schemaFlagsFromEnv,
+			RunE: func(cmd *cobra.Command, args []string) error {
+				return schemaCleanRun(cmd, args, flags)
+			},
+		}
+	)
+	cmd.Flags().SortFlags = false
+	addFlagURL(cmd.Flags(), &flags.URL)
+	addFlagAutoApprove(cmd.Flags(), &flags.AutoApprove)
+	cobra.CheckErr(cmd.MarkFlagRequired(flagURL))
+	return cmd
 }
 
-// CmdCleanRun is the command executed when running the CLI with 'schema clean' args.
-func CmdCleanRun(cmd *cobra.Command, _ []string) error {
+func schemaCleanRun(cmd *cobra.Command, _ []string, flags schemeCleanFlags) error {
 	// Open a client to the database.
-	c, err := sqlclient.Open(cmd.Context(), CleanFlags.URL)
+	c, err := sqlclient.Open(cmd.Context(), flags.URL)
 	if err != nil {
 		return err
 	}
@@ -346,7 +234,7 @@ func CmdCleanRun(cmd *cobra.Command, _ []string) error {
 	if err := summary(cmd, c, drop); err != nil {
 		return err
 	}
-	if CleanFlags.AutoApprove || promptUser() {
+	if flags.AutoApprove || promptUser() {
 		if err := c.ApplyChanges(cmd.Context(), drop); err != nil {
 			return err
 		}
@@ -354,12 +242,105 @@ func CmdCleanRun(cmd *cobra.Command, _ []string) error {
 	return nil
 }
 
-func summary(cmd *cobra.Command, drv migrate.Driver, changes []schema.Change) error {
-	p, err := drv.PlanChanges(cmd.Context(), "", changes)
+type schemaDiffFlags struct {
+	fromURL string // URL of the current state.
+	toURL   string // URL of the desired state.
+	devURL  string // URL of a dev database.
+}
+
+// schemaDiffCmd represents the 'atlas schema diff' subcommand.
+func schemaDiffCmd() *cobra.Command {
+	var (
+		flags schemaDiffFlags
+		cmd   = &cobra.Command{
+			Use:   "diff",
+			Short: "Calculate and print the diff between two schemas.",
+			Long: `'atlas schema diff' reads the state of two given schema definitions, 
+calculates the difference in their schemas, and prints a plan of
+SQL statements to migrate the "from" database to the schema of the "to" database.
+The database states can be read from a connected database, an HCL project or a migration directory.`,
+			Example: `  atlas schema diff --from mysql://user:pass@localhost:3306/test --to file://schema.hcl
+  atlas schema diff --from mysql://user:pass@localhost:3306 --to file://schema_1.hcl --to file://schema_2.hcl
+  atlas schema diff --from mysql://user:pass@localhost:3306 --to file://migrations`,
+			RunE: func(cmd *cobra.Command, args []string) error {
+				return schemaDiffRun(cmd, args, flags)
+			},
+		}
+	)
+	cmd.Flags().SortFlags = false
+	addFlagURL(cmd.Flags(), &flags.fromURL, flagFrom, "")
+	addFlagURL(cmd.Flags(), &flags.toURL, flagTo, "")
+	addFlagDevURL(cmd.Flags(), &flags.devURL)
+	cobra.CheckErr(cmd.MarkFlagRequired(flagFrom))
+	cobra.CheckErr(cmd.MarkFlagRequired(flagTo))
+	return cmd
+}
+
+func schemaDiffRun(cmd *cobra.Command, _ []string, flags schemaDiffFlags) error {
+	var (
+		ctx = cmd.Context()
+		c   *sqlclient.Client
+	)
+	// We need a driver for diffing and planning. If given, dev database has precedence.
+	if flags.devURL != "" {
+		var err error
+		c, err = sqlclient.Open(ctx, flags.devURL)
+		if err != nil {
+			return err
+		}
+		defer c.Close()
+	}
+	from, err := stateReader(ctx, &stateReaderConfig{urls: []string{flags.fromURL}, dev: c})
 	if err != nil {
 		return err
 	}
-	cmd.Println("-- Planned Changes:")
+	defer from.Close()
+	to, err := stateReader(ctx, &stateReaderConfig{urls: []string{flags.toURL}, dev: c})
+	if err != nil {
+		return err
+	}
+	defer to.Close()
+	if c == nil {
+		// If not both states are provided by a database connection, the call to state-reader would have returned
+		// 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)
+	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
+	}
+	if len(p.Changes) == 0 {
+		cmd.Println("Schemas are synced, no changes to be made.")
+	}
 	for _, c := range p.Changes {
 		if c.Comment != "" {
 			cmd.Println("--", strings.ToUpper(c.Comment[:1])+c.Comment[1:])
@@ -369,79 +350,192 @@ func summary(cmd *cobra.Command, drv migrate.Driver, changes []schema.Change) er
 	return nil
 }
 
-// CmdFmtRun formats all HCL files in a given directory using canonical HCL formatting
-// rules.
-func CmdFmtRun(cmd *cobra.Command, args []string) {
-	if len(args) == 0 {
-		args = append(args, "./")
-	}
-	for _, path := range args {
-		handlePath(cmd, path)
-	}
+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.
 }
 
-func applyRun(cmd *cobra.Command, client *sqlclient.Client, devURL string, paths []string, dryRun, autoApprove bool, input map[string]cty.Value) error {
-	schemas, ctx := SchemaFlags.Schemas, cmd.Context()
+// schemaInspectCmd represents the 'atlas schema inspect' subcommand.
+func schemaInspectCmd() *cobra.Command {
+	var (
+		flags schemaInspectFlags
+		cmd   = &cobra.Command{
+			Use:   "inspect",
+			Short: "Inspect a database and print its schema in Atlas DDL syntax.",
+			Long: `'atlas schema inspect' connects to the given database and inspects its schema.
+It then prints to the screen the schema of that database in Atlas DDL syntax. This output can be
+saved to a file, commonly by redirecting the output to a file named with a ".hcl" suffix:
+
+  atlas schema inspect -u "mysql://user:pass@localhost:3306/dbname" > schema.hcl
+
+This file can then be edited and used with the` + " `atlas schema apply` " + `command to plan
+and execute schema migrations against the given database. In cases where users wish to inspect
+all multiple schemas in a given database (for instance a MySQL server may contain multiple named
+databases), omit the relevant part from the url, e.g. "mysql://user:pass@localhost:3306/".
+To select specific schemas from the databases, users may use the "--schema" (or "-s" shorthand)
+flag.
+	`,
+			Example: `  atlas schema inspect -u "mysql://user:pass@localhost:3306/dbname"
+  atlas schema inspect -u "mariadb://user:pass@localhost:3306/" --schema=schemaA,schemaB -s schemaC
+  atlas schema inspect --url "postgres://user:pass@host:port/dbname?sslmode=disable"
+  atlas schema inspect -u "sqlite://file:ex1.db?_fk=1"`,
+			PreRunE: schemaFlagsFromEnv,
+			RunE: func(cmd *cobra.Command, args []string) error {
+				return schemaInspectRun(cmd, args, flags)
+			},
+		}
+	)
+	cmd.Flags().SortFlags = false
+	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)
+	if err != nil {
+		return err
+	}
+	defer client.Close()
+	schemas := flags.Schemas
 	if client.URL.Schema != "" {
 		schemas = append(schemas, client.URL.Schema)
 	}
-	realm, err := client.InspectRealm(ctx, &schema.InspectRealmOption{
+	s, err := client.InspectRealm(cmd.Context(), &schema.InspectRealmOption{
 		Schemas: schemas,
-		Exclude: SchemaFlags.Exclude,
+		Exclude: flags.Exclude,
 	})
 	if err != nil {
 		return err
 	}
-	desired := &schema.Realm{}
-	parsed, err := parseHCLPaths(paths...)
+	ddl, err := client.MarshalSpec(s)
 	if err != nil {
 		return err
 	}
-	if err := client.Eval(parsed, desired, input); err != nil {
-		return err
+	cmd.Print(string(ddl))
+	return nil
+}
+
+// schemaFmtCmd represents the 'atlas schema fmt' subcommand.
+func schemaFmtCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "fmt [path ...]",
+		Short: "Formats Atlas HCL files",
+		Long: `'atlas schema fmt' formats all ".hcl" files under the given paths using
+canonical HCL layout style as defined by the github.com/hashicorp/hcl/v2/hclwrite package.
+Unless stated otherwise, the fmt command will use the current directory.
+
+After running, the command will print the names of the files it has formatted. If all
+files in the directory are formatted, no input will be printed out.
+`,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			return schemaFmtRun(cmd, args)
+		},
 	}
-	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)
-			}
-		}
+	return cmd
+}
+
+func schemaFmtRun(cmd *cobra.Command, args []string) error {
+	if len(args) == 0 {
+		args = append(args, "./")
 	}
-	if _, ok := client.Driver.(schema.Normalizer); ok && devURL != "" {
-		dev, err := sqlclient.Open(ctx, ApplyFlags.DevURL)
+	for _, path := range args {
+		tasks, err := tasks(path)
 		if err != nil {
 			return err
 		}
-		defer dev.Close()
-		desired, err = dev.Driver.(schema.Normalizer).NormalizeRealm(ctx, desired)
-		if err != nil {
-			return err
+		for _, task := range tasks {
+			changed, err := fmtFile(task)
+			if err != nil {
+				return err
+			}
+			if changed {
+				cmd.Println(task.path)
+			}
 		}
 	}
-	changes, err := client.RealmDiff(realm, desired)
+	return nil
+}
+
+// selectEnv returns the Env from the current project file based on the selected
+// argument. If selected is "", or no project file exists in the current directory
+// a zero-value Env is returned.
+func selectEnv(selected string) (*Env, error) {
+	env := &Env{
+		Lint:      &Lint{},
+		Migration: &Migration{},
+	}
+	if selected == "" {
+		return env, nil
+	}
+	if _, err := os.Stat(projectFileName); os.IsNotExist(err) {
+		return nil, fmt.Errorf("project file %q was not found", projectFileName)
+	}
+	return LoadEnv(projectFileName, selected, WithInput(GlobalFlags.Vars))
+}
+
+func schemaFlagsFromEnv(cmd *cobra.Command, _ []string) error {
+	activeEnv, err := selectEnv(GlobalFlags.SelectedEnv)
 	if err != nil {
 		return err
 	}
-	if len(changes) == 0 {
-		cmd.Println("Schema is synced, no changes to be made")
-		return nil
+	if err := inputValsFromEnv(cmd); err != nil {
+		return err
 	}
-	if err := summary(cmd, client, changes); err != nil {
+	if err := dsn2url(cmd); err != nil {
 		return err
 	}
-	if !dryRun && (autoApprove || promptUser()) {
-		if err := client.ApplyChanges(ctx, changes); err != nil {
+	if err := maySetFlag(cmd, flagURL, activeEnv.URL); err != nil {
+		return err
+	}
+	if err := maySetFlag(cmd, flagDevURL, activeEnv.DevURL); err != nil {
+		return err
+	}
+	srcs, err := activeEnv.Sources()
+	if err != nil {
+		return err
+	}
+	if err := maySetFlag(cmd, flagFile, strings.Join(srcs, "")); err != nil {
+		return err
+	}
+	if s := strings.Join(activeEnv.Schemas, ","); s != "" {
+		if err := maySetFlag(cmd, flagSchema, s); err != nil {
+			return err
+		}
+	}
+	if s := strings.Join(activeEnv.Exclude, ","); s != "" {
+		if err := maySetFlag(cmd, flagExclude, s); err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
+func summary(cmd *cobra.Command, drv migrate.Driver, changes []schema.Change) error {
+	p, err := drv.PlanChanges(cmd.Context(), "", changes)
+	if err != nil {
+		return err
+	}
+	cmd.Println("-- Planned Changes:")
+	for _, c := range p.Changes {
+		if c.Comment != "" {
+			cmd.Println("--", strings.ToUpper(c.Comment[:1])+c.Comment[1:])
+		}
+		cmd.Println(c.Cmd)
+	}
+	return nil
+}
+
+const (
+	answerApply = "Apply"
+	answerAbort = "Abort"
+)
+
 func promptUser() bool {
 	prompt := promptui.Select{
 		Label: "Are you sure?",
@@ -452,18 +546,6 @@ func promptUser() bool {
 	return result == answerApply
 }
 
-func handlePath(cmd *cobra.Command, path string) {
-	tasks, err := tasks(path)
-	cobra.CheckErr(err)
-	for _, task := range tasks {
-		changed, err := fmtFile(task)
-		cobra.CheckErr(err)
-		if changed {
-			cmd.Println(task.path)
-		}
-	}
-}
-
 func tasks(path string) ([]fmttask, error) {
 	var tasks []fmttask
 	stat, err := os.Stat(path)
diff --git a/doc/md/reference.md b/doc/md/reference.md
index c6feef53..17454a4e 100644
--- a/doc/md/reference.md
+++ b/doc/md/reference.md
@@ -48,8 +48,6 @@ atlas migrate
 
 #### Flags
 ```
-      --dir string           select migration directory using URL format (default "file://migrations")
-      --dir-format string    set migration file format (default "atlas")
       --env string           set which env from the project file to use
       --var <name>=<value>   input variables (default [])
 
@@ -62,7 +60,7 @@ Applies pending migration files on the connected database.
 
 #### Usage
 ```
-atlas migrate apply [flags] [count]
+atlas migrate apply [amount] [flags]
 ```
 
 #### Details
@@ -85,14 +83,15 @@ If run with the "--dry-run" flag, atlas will not execute any SQL.
 ```
 #### Flags
 ```
-      --log string                log format to use (default "tty")
-      --revisions-schema string   schema name where the revisions table resides
-      --dry-run                   do not actually execute any SQL but show it on screen
+  -u, --url string                [driver://username:password@address/dbname?param=value] select a resource using the URL format
+      --dir string                select migration directory using URL format (default "file://migrations")
+      --log string                go template to use to format logs
+      --revisions-schema string   name of the schema the revisions table resides in
+      --dry-run                   print SQL without executing it
       --from string               calculate pending files from the given version (including it)
       --baseline string           start the first migration after the given baseline version
       --tx-mode string            set transaction mode [none, file, all] (default "file")
       --allow-dirty               allow start working on a non-clean database
-  -u, --url string                [driver://username:password@address/dbname?param=value] select a database using the URL format
 
 ```
 
@@ -103,7 +102,7 @@ Compute the diff between the migration directory and a desired state and create
 
 #### Usage
 ```
-atlas migrate diff [flags] [name]
+atlas migrate diff [name] [flags]
 ```
 
 #### Details
@@ -121,9 +120,13 @@ directory state to the desired schema. The desired state can be another connecte
 ```
 #### Flags
 ```
-      --dev-url string     [driver://username:password@address/dbname?param=value] select a database using the URL format
-      --to strings         [driver://username:password@address/dbname?param=value ...] select a desired state using the URL format
-      --qualifier string   qualify tables with custom qualifier when working on a single schema
+  -u, --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
+      --dir string                select migration directory using URL format (default "file://migrations")
+      --dir-format string         select migration file format (default "atlas")
+      --revisions-schema string   name of the schema the revisions table resides in
+  -s, --schema strings            set schema names
+      --qualifier string          qualify tables with custom qualifier when working on a single schema
 
 ```
 
@@ -134,7 +137,7 @@ Hash (re-)creates an integrity hash file for the migration directory.
 
 #### Usage
 ```
-atlas migrate hash
+atlas migrate hash [flags]
 ```
 
 #### Details
@@ -146,6 +149,13 @@ This command should be used whenever a manual change in the migration directory
 ```
   atlas migrate hash
 ```
+#### Flags
+```
+      --dir string          select migration directory using URL format (default "file://migrations")
+      --dir-format string   select migration file format (default "atlas")
+
+```
+
 
 ### atlas migrate import
 
@@ -163,8 +173,9 @@ atlas migrate import [flags]
 ```
 #### Flags
 ```
-      --from string   select migration directory using URL format (default "file://migrations")
-      --to string     select migration directory using URL format (default "file://migrations")
+      --from string         select migration directory using URL format (default "file://migrations")
+      --to string           select migration directory using URL format (default "file://migrations")
+      --dir-format string   select migration file format (default "atlas")
 
 ```
 
@@ -188,11 +199,12 @@ atlas migrate lint [flags]
 ```
 #### Flags
 ```
-      --dev-url string    [driver://username:password@address/dbname?param=value] select a database using the URL format
+      --dev-url string    [driver://username:password@address/dbname?param=value] select a dev database using the URL format
+      --dir string        select migration directory using URL format (default "file://migrations")
+      --log string        custom logging using a Go template
+      --latest uint       run analysis on the latest N migration files
       --git-base string   run analysis against the base Git branch
       --git-dir string    path to the repository working directory (default ".")
-      --latest uint       run analysis on the latest N migration files
-      --log string        custom logging using a Go template
 
 ```
 
@@ -203,7 +215,7 @@ Creates a new empty migration file in the migration directory.
 
 #### Usage
 ```
-atlas migrate new [name]
+atlas migrate new [name] [flags]
 ```
 
 #### Details
@@ -214,6 +226,13 @@ atlas migrate new [name]
 ```
   atlas migrate new my-new-migration
 ```
+#### Flags
+```
+      --dir string          select migration directory using URL format (default "file://migrations")
+      --dir-format string   select migration file format (default "atlas")
+
+```
+
 
 ### atlas migrate set
 
@@ -237,7 +256,10 @@ to be applied. This command is usually used after manually making changes to the
 ```
 #### Flags
 ```
-  -u, --url string   [driver://username:password@address/dbname?param=value] select a database using the URL format
+  -u, --url string                [driver://username:password@address/dbname?param=value] select a resource using the URL format
+      --dir string                select migration directory using URL format (default "file://migrations")
+      --dir-format string         select migration file format (default "atlas")
+      --revisions-schema string   name of the schema the revisions table resides in
 
 ```
 
@@ -262,9 +284,10 @@ atlas migrate status [flags]
 ```
 #### Flags
 ```
-      --log string                custom logging using a Go template
-      --revisions-schema string   schema name where the revisions table resides
-  -u, --url string                [driver://username:password@address/dbname?param=value] select a database using the URL format
+  -u, --url string                [driver://username:password@address/dbname?param=value] select a resource using the URL format
+      --dir string                select migration directory using URL format (default "file://migrations")
+      --dir-format string         select migration file format (default "atlas")
+      --revisions-schema string   name of the schema the revisions table resides in
 
 ```
 
@@ -293,7 +316,9 @@ files are executed on the connected database in order to validate SQL semantics.
 ```
 #### Flags
 ```
-      --dev-url string   [driver://username:password@address/dbname?param=value] select a database using the URL format
+      --dev-url string      [driver://username:password@address/dbname?param=value] select a dev database using the URL format
+      --dir string          select migration directory using URL format (default "file://migrations")
+      --dir-format string   select migration file format (default "atlas")
 
 ```
 
@@ -308,7 +333,7 @@ atlas schema
 ```
 
 #### Details
-The `atlas schema` command groups subcommands for working with Atlas schemas.
+The `atlas schema` command groups subcommands working with declarative Atlas schemas.
 
 #### Flags
 ```
@@ -355,14 +380,12 @@ migration.
 #### Flags
 ```
   -f, --file strings      [paths...] file or directory containing the HCL files
-  -u, --url string        URL to the database using the format:
-                          [driver://username:password@address/dbname?param=value]
-      --exclude strings   List of glob patterns used to filter resources from applying.
-  -s, --schema strings    Set schema names.
-      --dev-url string    URL for the dev database. Used to validate schemas and calculate diffs
-                          before running migration.
-      --dry-run           Dry-run. Print SQL plan without prompting for execution.
-      --auto-approve      Auto approve. Apply the schema changes without prompting for approval.
+  -u, --url string        [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
+      --dry-run           print SQL without executing it
+      --auto-approve      apply changes without prompting for approval
 
 ```
 
@@ -388,9 +411,8 @@ As a safety feature, 'atlas schema clean' will ask for confirmation before attem
 ```
 #### Flags
 ```
-      --auto-approve   Auto approve. Apply the schema changes without prompting for approval.
-  -u, --url string     URL to the database using the format:
-                       [driver://username:password@address/dbname?param=value]
+  -u, --url string     [driver://username:password@address/dbname?param=value] select a resource using the URL format
+      --auto-approve   apply changes without prompting for approval
 
 ```
 
@@ -419,9 +441,9 @@ The database states can be read from a connected database, an HCL project or a m
 ```
 #### Flags
 ```
-      --from string      [driver://username:password@protocol(address)/dbname?param=value] select a database using the URL format
-      --to string        [driver://username:password@protocol(address)/dbname?param=value] select a database using the URL format
-      --dev-url string   [driver://username:password@protocol(address)/dbname?param=value] select a database using the URL format
+      --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
 
 ```
 
@@ -436,7 +458,7 @@ atlas schema fmt [path ...]
 ```
 
 #### Details
-'atlas schema fmt' formats all ".hcl" files under the given path using
+'atlas schema fmt' formats all ".hcl" files under the given paths using
 canonical HCL layout style as defined by the github.com/hashicorp/hcl/v2/hclwrite package.
 Unless stated otherwise, the fmt command will use the current directory.
 
@@ -479,9 +501,9 @@ flag.
 ```
 #### Flags
 ```
-      --exclude strings   List of glob patterns used to filter resources from inspection
-  -s, --schema strings    Set schema name
-  -u, --url string        [driver://username:password@protocol(address)/dbname?param=value] select a database using the URL format
+  -u, --url string        [driver://username:password@address/dbname?param=value] select a resource 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/sql/sqlclient/client.go b/sql/sqlclient/client.go
index 21a293d3..1e448c8c 100644
--- a/sql/sqlclient/client.go
+++ b/sql/sqlclient/client.go
@@ -201,7 +201,7 @@ func OpenURL(ctx context.Context, u *url.URL, opts ...OpenOption) (*Client, erro
 	}
 	v, ok := drivers.Load(u.Scheme)
 	if !ok {
-		return nil, fmt.Errorf("sql/sqlclient: no opener was register with name %q", u.Scheme)
+		return nil, fmt.Errorf("sql/sqlclient: no opener was registered with name %q", u.Scheme)
 	}
 	drv := v.(*driver)
 	// If there is a schema given and the driver allows to change the schema for the url, do it.


Test patch:
diff --git a/cmd/atlas/internal/cmdapi/cmdapi_test.go b/cmd/atlas/internal/cmdapi/cmdapi_test.go
index d3a92fc2..8679da7e 100644
--- a/cmd/atlas/internal/cmdapi/cmdapi_test.go
+++ b/cmd/atlas/internal/cmdapi/cmdapi_test.go
@@ -6,10 +6,15 @@ package cmdapi
 
 import (
 	"bytes"
+	"context"
+	"database/sql"
+	"fmt"
 	"os"
 	"os/exec"
 	"testing"
 
+	"ariga.io/atlas/sql/sqlite"
+	"github.com/spf13/cobra"
 	"github.com/stretchr/testify/require"
 )
 
@@ -59,3 +64,39 @@ func TestCLI_Version(t *testing.T) {
 		})
 	}
 }
+
+func runCmd(cmd *cobra.Command, args ...string) (string, error) {
+	var out bytes.Buffer
+	cmd.SetOut(&out)
+	cmd.SetErr(&out)
+	// Cobra checks for the args to equal nil and if so uses os.Args[1:].
+	// In tests, this leads to go tooling arguments being part of the command arguments.
+	if args == nil {
+		args = []string{}
+	}
+	cmd.SetArgs(args)
+	err := cmd.Execute()
+	return out.String(), err
+}
+
+// openSQLite creates a sqlite db, seeds it with the seed query and returns the url to it.
+func openSQLite(t *testing.T, seed string) string {
+	f, err := os.CreateTemp("", "sqlite.db")
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		os.Remove(f.Name())
+	})
+	dsn := fmt.Sprintf("file:%s?cache=shared&_fk=1", f.Name())
+	db, err := sql.Open("sqlite3", dsn)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		db.Close()
+	})
+	drv, err := sqlite.Open(db)
+	require.NoError(t, err)
+	if len(seed) > 0 {
+		_, err := drv.ExecContext(context.Background(), seed)
+		require.NoError(t, err)
+	}
+	return fmt.Sprintf("sqlite://%s", dsn)
+}
diff --git a/cmd/atlas/internal/cmdapi/diff_test.go b/cmd/atlas/internal/cmdapi/diff_test.go
deleted file mode 100644
index 5eca9ec0..00000000
--- a/cmd/atlas/internal/cmdapi/diff_test.go
+++ /dev/null
@@ -1,151 +0,0 @@
-// Copyright 2021-present The Atlas Authors. All rights reserved.
-// This source code is licensed under the Apache 2.0 license found
-// in the LICENSE file in the root directory of this source tree.
-
-package cmdapi
-
-import (
-	"bytes"
-	"context"
-	"database/sql"
-	"fmt"
-	"os"
-	"path/filepath"
-	"testing"
-
-	"ariga.io/atlas/sql/sqlite"
-
-	_ "github.com/mattn/go-sqlite3"
-	"github.com/spf13/cobra"
-	"github.com/stretchr/testify/require"
-)
-
-func TestCmdSchemaDiff(t *testing.T) {
-	// Creates the missing table.
-	s, err := runCmd(
-		newDiffCmd(),
-		"--from", openSQLite(t, ""),
-		"--to", openSQLite(t, "create table t1 (id int);"),
-	)
-	require.NoError(t, err)
-	require.EqualValues(t, "-- Create \"t1\" table\nCREATE TABLE `t1` (`id` int NULL)\n", s)
-
-	// No changes.
-	s, err = runCmd(
-		newDiffCmd(),
-		"--from", openSQLite(t, ""),
-		"--to", openSQLite(t, ""),
-	)
-	require.NoError(t, err)
-	require.EqualValues(t, "Schemas are synced, no changes to be made.\n", s)
-
-	// Desired state from migration directory requires dev database.
-	_, err = runCmd(
-		newDiffCmd(),
-		"--from", "file://testdata/sqlite",
-		"--to", openSQLite(t, ""),
-	)
-	require.EqualError(t, err, "--dev-url cannot be empty")
-
-	// Desired state from migration directory.
-	s, err = runCmd(
-		newDiffCmd(),
-		"--from", openSQLite(t, ""),
-		"--to", "file://testdata/sqlite",
-		"--dev-url", openSQLite(t, ""),
-	)
-	require.NoError(t, err)
-	require.EqualValues(t, "-- Create \"tbl\" table\nCREATE TABLE `tbl` (`col` int NOT NULL, `col_2` bigint NULL)\n", s)
-
-	// Desired state from migration directory.
-	s, err = runCmd(
-		newDiffCmd(),
-		"--from", openSQLite(t, ""),
-		"--to", "file://testdata/sqlite",
-		"--dev-url", openSQLite(t, ""),
-	)
-	require.NoError(t, err)
-	require.EqualValues(t, "-- Create \"tbl\" table\nCREATE TABLE `tbl` (`col` int NOT NULL, `col_2` bigint NULL)\n", s)
-
-	// Current state from migration directory, desired state from HCL - synced.
-	p := filepath.Join(t.TempDir(), "schema.hcl")
-	require.NoError(t, os.WriteFile(p, []byte(`schema "main" {}
-table "tbl" {
-  schema = schema.main
-  column "col" {
-    type = int
-  }
-  column "col_2" {
-    type = bigint
-    null = true
-  }
-}`), 0644))
-	s, err = runCmd(
-		newDiffCmd(),
-		"--from", "file://testdata/sqlite",
-		"--to", "file://"+p,
-		"--dev-url", openSQLite(t, ""),
-	)
-	require.NoError(t, err)
-	require.EqualValues(t, "Schemas are synced, no changes to be made.\n", s)
-
-	// Current state from migration directory, desired state from HCL - missing column.
-	p = filepath.Join(t.TempDir(), "schema.hcl")
-	require.NoError(t, os.WriteFile(p, []byte(`schema "main" {}
-table "tbl" {
-  schema = schema.main
-  column "col" {
-    type = int
-  }
-  column "col_2" {
-    type = bigint
-    null = true
-  }
-  column "col_3" {
-    type = text
-  }
-}`), 0644))
-	s, err = runCmd(
-		newDiffCmd(),
-		"--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,
-	)
-}
-
-// openSQLite creates a sqlite db, seeds it with the seed query and returns the url to it.
-func openSQLite(t *testing.T, seed string) string {
-	f, err := os.CreateTemp("", "sqlite.db")
-	require.NoError(t, err)
-	t.Cleanup(func() {
-		os.Remove(f.Name())
-	})
-	dsn := fmt.Sprintf("file:%s?cache=shared&_fk=1", f.Name())
-	db, err := sql.Open("sqlite3", dsn)
-	require.NoError(t, err)
-	t.Cleanup(func() {
-		db.Close()
-	})
-	drv, err := sqlite.Open(db)
-	require.NoError(t, err)
-	if len(seed) > 0 {
-		_, err := drv.ExecContext(context.Background(), seed)
-		require.NoError(t, err)
-	}
-	return fmt.Sprintf("sqlite://%s", dsn)
-}
-
-func runCmd(cmd *cobra.Command, args ...string) (string, error) {
-	var out bytes.Buffer
-	cmd.SetOut(&out)
-	cmd.SetErr(&out)
-	cmd.SetArgs(args)
-	err := cmd.Execute()
-	return out.String(), err
-}
diff --git a/cmd/atlas/internal/cmdapi/migrate_test.go b/cmd/atlas/internal/cmdapi/migrate_test.go
index b6da6f99..d01bd5c5 100644
--- a/cmd/atlas/internal/cmdapi/migrate_test.go
+++ b/cmd/atlas/internal/cmdapi/migrate_test.go
@@ -32,7 +32,7 @@ import (
 )
 
 func TestMigrate(t *testing.T) {
-	_, err := runCmd(Root, "migrate")
+	_, err := runCmd(migrateCmd())
 	require.NoError(t, err)
 }
 
@@ -42,7 +42,7 @@ func TestMigrate_Import(t *testing.T) {
 		t.Run(tool, func(t *testing.T) { // remove this once --dir-format is removed. Test is kept to ensure BC.
 			path := filepath.FromSlash("testdata/import/" + tool)
 			out, err := runCmd(
-				Root, "migrate", "import",
+				migrateImportCmd(),
 				"--from", "file://"+path,
 				"--to", "file://"+p,
 				"--dir-format", tool,
@@ -70,7 +70,7 @@ func TestMigrate_Import(t *testing.T) {
 		t.Run(tool, func(t *testing.T) {
 			path := filepath.FromSlash("testdata/import/" + tool)
 			out, err := runCmd(
-				Root, "migrate", "import",
+				migrateImportCmd(),
 				"--from", fmt.Sprintf("file://%s?format=%s", path, tool),
 				"--to", "file://"+p,
 			)
@@ -107,7 +107,7 @@ func TestMigrate_Apply(t *testing.T) {
 
 	// Fails on empty directory.
 	s, err := runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://"+p,
 		"-u", openSQLite(t, ""),
 	)
@@ -124,7 +124,7 @@ func TestMigrate_Apply(t *testing.T) {
 	})
 
 	_, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite",
 		"--url", openSQLite(t, ""),
 	)
@@ -158,7 +158,7 @@ func TestMigrate_Apply(t *testing.T) {
 	require.NoError(t, f.Close())
 
 	s, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite",
 		"--url", fmt.Sprintf("sqlitelockapply://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
 	)
@@ -168,7 +168,7 @@ func TestMigrate_Apply(t *testing.T) {
 	// Apply zero throws error.
 	for _, n := range []string{"-1", "0"} {
 		_, err = runCmd(
-			Root, "migrate", "apply",
+			migrateApplyCmd(),
 			"--dir", "file://testdata/sqlite",
 			"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
 			"--", n,
@@ -178,7 +178,7 @@ func TestMigrate_Apply(t *testing.T) {
 
 	// Will work and print stuff to the console.
 	s, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
 		"1",
@@ -192,7 +192,7 @@ func TestMigrate_Apply(t *testing.T) {
 
 	// Transactions will be wrapped per file. If the second file has an error, first still is applied.
 	s, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite2",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test2.db")),
 	)
@@ -227,7 +227,7 @@ func TestMigrate_Apply(t *testing.T) {
 
 	// Running again will pick up the failed statement and try it again.
 	s, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite2",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test2.db")),
 	)
@@ -245,7 +245,7 @@ func TestMigrate_Apply(t *testing.T) {
 
 	// Editing an applied line will raise error.
 	s, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite2",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test2.db")),
 		"--tx-mode", "none",
@@ -255,10 +255,10 @@ func TestMigrate_Apply(t *testing.T) {
 	})
 	require.NoError(t, exec.Command("cp", "-r", "testdata/sqlite2", "testdata/sqlite3").Run())
 	sed(t, "s/col_2/col_5/g", "testdata/sqlite3/20220318104615_second.sql")
-	_, err = runCmd(Root, "migrate", "hash", "--dir", "file://testdata/sqlite3")
+	_, err = runCmd(migrateHashCmd(), "--dir", "file://testdata/sqlite3")
 	require.NoError(t, err)
 	s, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite3",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test2.db")),
 	)
@@ -267,10 +267,10 @@ func TestMigrate_Apply(t *testing.T) {
 	// Fixing the migration file will finish without errors.
 	sed(t, "s/col_5/col_2/g", "testdata/sqlite3/20220318104615_second.sql")
 	sed(t, "s/asdasd //g", "testdata/sqlite3/20220318104615_second.sql")
-	_, err = runCmd(Root, "migrate", "hash", "--dir", "file://testdata/sqlite3")
+	_, err = runCmd(migrateHashCmd(), "--dir", "file://testdata/sqlite3")
 	require.NoError(t, err)
 	s, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite3",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test2.db")),
 	)
@@ -285,7 +285,7 @@ func TestMigrate_Apply(t *testing.T) {
 
 	// Running again will report database being in clean state.
 	s, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite3",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test2.db")),
 	)
@@ -295,7 +295,7 @@ func TestMigrate_Apply(t *testing.T) {
 	// Dry run will print the statements in second migration file without executing them.
 	// No changes to the revisions will be done.
 	s, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
 		"--dry-run",
@@ -319,7 +319,6 @@ func TestMigrate_Apply(t *testing.T) {
 	revs, err = rrw.ReadRevisions(ctx)
 	require.NoError(t, err)
 	require.Len(t, revs, 1)
-	MigrateFlags.Apply.DryRun = false // global flag, undo for rest of tests
 
 	// Prerequisites for testing missing migration behavior.
 	c1, err = sqlclient.Open(ctx, fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test3.db")))
@@ -331,7 +330,7 @@ func TestMigrate_Apply(t *testing.T) {
 		"testdata/sqlite3/20220318104615_second.sql",
 		"testdata/sqlite3/20220318104616_second.sql",
 	))
-	_, err = runCmd(Root, "migrate", "hash", "--dir", "file://testdata/sqlite3")
+	_, err = runCmd(migrateHashCmd(), "--dir", "file://testdata/sqlite3")
 	require.NoError(t, err)
 	rrw, err = migrate2.NewEntRevisions(ctx, c1)
 	require.NoError(t, err)
@@ -340,7 +339,7 @@ func TestMigrate_Apply(t *testing.T) {
 	// No changes if the last revision has a greater version than the last migration.
 	require.NoError(t, rrw.WriteRevision(ctx, &migrate.Revision{Version: "zzz"}))
 	s, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite3",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test3.db")),
 	)
@@ -351,14 +350,14 @@ func TestMigrate_Apply(t *testing.T) {
 	_, err = c1.ExecContext(ctx, "DROP table `atlas_schema_revisions`")
 	require.NoError(t, err)
 	s, err = runCmd(
-		Root, "migrate", "apply", "1",
+		migrateApplyCmd(), "1",
 		"--dir", "file://testdata/sqlite3",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test3.db")),
 	)
 	require.NoError(t, rrw.WriteRevision(ctx, &migrate.Revision{Version: "20220318104615"}))
 	require.NoError(t, err)
 	s, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite3",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test3.db")),
 	)
@@ -374,7 +373,7 @@ func TestMigrate_Apply(t *testing.T) {
 	require.NoError(t, rrw.WriteRevision(ctx, &migrate.Revision{Version: "1"}))
 	require.NoError(t, err)
 	s, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite3",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test3.db")),
 	)
@@ -388,7 +387,7 @@ func TestMigrate_Apply(t *testing.T) {
 	require.NoError(t, rrw.WriteRevision(ctx, &migrate.Revision{Version: "z", Description: "z", Total: 1}))
 	require.NoError(t, err)
 	_, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/sqlite3",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test3.db")),
 	)
@@ -401,7 +400,7 @@ func TestMigrate_ApplyTxMode(t *testing.T) {
 			p := t.TempDir()
 			// Apply the first 2 migrations.
 			s, err := runCmd(
-				Root, "migrate", "apply",
+				migrateApplyCmd(),
 				"--dir", "file://testdata/sqlitetx",
 				"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
 				"--tx-mode", mode,
@@ -417,7 +416,7 @@ func TestMigrate_ApplyTxMode(t *testing.T) {
 
 			// Apply the rest.
 			s, err = runCmd(
-				Root, "migrate", "apply",
+				migrateApplyCmd(),
 				"--dir", "file://testdata/sqlitetx",
 				"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
 				"--tx-mode", mode,
@@ -430,7 +429,7 @@ func TestMigrate_ApplyTxMode(t *testing.T) {
 			if mode != "none" {
 				// Apply the first 2 migrations for the faulty one.
 				s, err = runCmd(
-					Root, "migrate", "apply",
+					migrateApplyCmd(),
 					"--dir", "file://testdata/sqlitetx_2",
 					"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test_2.db")),
 					"--tx-mode", mode,
@@ -453,7 +452,7 @@ func TestMigrate_ApplyTxMode(t *testing.T) {
 
 				// Apply the rest, expect it to fail due to constraint error, but only the new one is reported.
 				s, err = runCmd(
-					Root, "migrate", "apply",
+					migrateApplyCmd(),
 					"--dir", "file://testdata/sqlitetx_2",
 					"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test_2.db")),
 					"--tx-mode", mode,
@@ -470,7 +469,7 @@ func TestMigrate_ApplyBaseline(t *testing.T) {
 	p := t.TempDir()
 	// Run migration with baseline should store this revision in the database.
 	s, err := runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/baseline1",
 		"--baseline", "1",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
@@ -480,7 +479,7 @@ func TestMigrate_ApplyBaseline(t *testing.T) {
 
 	// Next run without baseline should run the migration from the baseline.
 	s, err = runCmd(
-		Root, "migrate", "apply",
+		migrateApplyCmd(),
 		"--dir", "file://testdata/baseline2",
 		"--url", fmt.Sprintf("sqlite://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
 	)
@@ -494,7 +493,7 @@ func TestMigrate_Diff(t *testing.T) {
 
 	// Will create migration directory if not existing.
 	_, err := runCmd(
-		Root, "migrate", "diff",
+		migrateDiffCmd(),
 		"name",
 		"--dir", "file://"+filepath.Join(p, "migrations"),
 		"--dev-url", openSQLite(t, ""),
@@ -506,7 +505,7 @@ func TestMigrate_Diff(t *testing.T) {
 	// Expect no clean dev error.
 	p = t.TempDir()
 	s, err := runCmd(
-		Root, "migrate", "diff",
+		migrateDiffCmd(),
 		"name",
 		"--dir", "file://"+p,
 		"--dev-url", openSQLite(t, "create table t (c int);"),
@@ -517,7 +516,7 @@ func TestMigrate_Diff(t *testing.T) {
 
 	// Works (on empty directory).
 	s, err = runCmd(
-		Root, "migrate", "diff",
+		migrateDiffCmd(),
 		"name",
 		"--dir", "file://"+p,
 		"--dev-url", openSQLite(t, ""),
@@ -542,7 +541,7 @@ func TestMigrate_Diff(t *testing.T) {
 	require.NoError(t, err)
 	require.NoError(t, f.Close())
 	s, err = runCmd(
-		Root, "migrate", "diff",
+		migrateDiffCmd(),
 		"name",
 		"--dir", "file://"+t.TempDir(),
 		"--dev-url", fmt.Sprintf("sqlitelockdiff://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
@@ -558,14 +557,14 @@ func TestMigrate_New(t *testing.T) {
 		v = time.Now().UTC().Format("20060102150405")
 	)
 
-	s, err := runCmd(Root, "migrate", "new", "--dir", "file://"+p)
+	s, err := runCmd(migrateNewCmd(), "--dir", "file://"+p)
 	require.Zero(t, s)
 	require.NoError(t, err)
 	require.FileExists(t, filepath.Join(p, v+".sql"))
 	require.FileExists(t, filepath.Join(p, "atlas.sum"))
 	require.Equal(t, 2, countFiles(t, p))
 
-	s, err = runCmd(Root, "migrate", "new", "my-migration-file", "--dir", "file://"+p)
+	s, err = runCmd(migrateNewCmd(), "my-migration-file", "--dir", "file://"+p)
 	require.Zero(t, s)
 	require.NoError(t, err)
 	require.FileExists(t, filepath.Join(p, v+"_my-migration-file.sql"))
@@ -573,7 +572,7 @@ func TestMigrate_New(t *testing.T) {
 	require.Equal(t, 3, countFiles(t, p))
 
 	p = t.TempDir()
-	s, err = runCmd(Root, "migrate", "new", "golang-migrate", "--dir", "file://"+p, "--dir-format", formatGolangMigrate)
+	s, err = runCmd(migrateNewCmd(), "golang-migrate", "--dir", "file://"+p, "--dir-format", formatGolangMigrate)
 	require.Zero(t, s)
 	require.NoError(t, err)
 	require.FileExists(t, filepath.Join(p, v+"_golang-migrate.up.sql"))
@@ -581,14 +580,14 @@ func TestMigrate_New(t *testing.T) {
 	require.Equal(t, 3, countFiles(t, p))
 
 	p = t.TempDir()
-	s, err = runCmd(Root, "migrate", "new", "goose", "--dir", "file://"+p+"?format="+formatGoose)
+	s, err = runCmd(migrateNewCmd(), "goose", "--dir", "file://"+p+"?format="+formatGoose)
 	require.Zero(t, s)
 	require.NoError(t, err)
 	require.FileExists(t, filepath.Join(p, v+"_goose.sql"))
 	require.Equal(t, 2, countFiles(t, p))
 
 	p = t.TempDir()
-	s, err = runCmd(Root, "migrate", "new", "flyway", "--dir", "file://"+p+"?format="+formatFlyway)
+	s, err = runCmd(migrateNewCmd(), "flyway", "--dir", "file://"+p+"?format="+formatFlyway)
 	require.Zero(t, s)
 	require.NoError(t, err)
 	require.FileExists(t, filepath.Join(p, fmt.Sprintf("V%s__%s.sql", v, formatFlyway)))
@@ -596,14 +595,14 @@ func TestMigrate_New(t *testing.T) {
 	require.Equal(t, 3, countFiles(t, p))
 
 	p = t.TempDir()
-	s, err = runCmd(Root, "migrate", "new", "liquibase", "--dir", "file://"+p+"?format="+formatLiquibase)
+	s, err = runCmd(migrateNewCmd(), "liquibase", "--dir", "file://"+p+"?format="+formatLiquibase)
 	require.Zero(t, s)
 	require.NoError(t, err)
 	require.FileExists(t, filepath.Join(p, v+"_liquibase.sql"))
 	require.Equal(t, 2, countFiles(t, p))
 
 	p = t.TempDir()
-	s, err = runCmd(Root, "migrate", "new", "dbmate", "--dir", "file://"+p+"?format="+formatDBMate)
+	s, err = runCmd(migrateNewCmd(), "dbmate", "--dir", "file://"+p+"?format="+formatDBMate)
 	require.Zero(t, s)
 	require.NoError(t, err)
 	require.FileExists(t, filepath.Join(p, v+"_dbmate.sql"))
@@ -612,23 +611,21 @@ func TestMigrate_New(t *testing.T) {
 	f := filepath.Join("testdata", "mysql", "new.sql")
 	require.NoError(t, os.WriteFile(f, []byte("contents"), 0600))
 	t.Cleanup(func() { os.Remove(f) })
-	s, err = runCmd(Root, "migrate", "new", "--dir", "file://testdata/mysql")
+	s, err = runCmd(migrateNewCmd(), "--dir", "file://testdata/mysql")
 	require.NotZero(t, s)
 	require.Error(t, err)
 }
 
 func TestMigrate_Validate(t *testing.T) {
 	// Without re-playing.
-	MigrateFlags.DevURL = ""         // global flags are set from other tests ...
-	MigrateFlags.DirFormat = "atlas" // global flags are set from other tests ...
-	s, err := runCmd(Root, "migrate", "validate", "--dir", "file://testdata/mysql")
+	s, err := runCmd(migrateValidateCmd(), "--dir", "file://testdata/mysql")
 	require.Zero(t, s)
 	require.NoError(t, err)
 
 	f := filepath.Join("testdata", "mysql", "new.sql")
 	require.NoError(t, os.WriteFile(f, []byte("contents"), 0600))
 	t.Cleanup(func() { os.Remove(f) })
-	s, err = runCmd(Root, "migrate", "validate", "--dir", "file://testdata/mysql")
+	s, err = runCmd(migrateValidateCmd(), "--dir", "file://testdata/mysql")
 	require.NotZero(t, s)
 	require.Error(t, err)
 	require.NoError(t, os.Remove(f))
@@ -637,10 +634,10 @@ func TestMigrate_Validate(t *testing.T) {
 	p := t.TempDir()
 	require.NoError(t, os.WriteFile(filepath.Join(p, "1_initial.sql"), []byte("create table t1 (c1 int)"), 0644))
 	require.NoError(t, os.WriteFile(filepath.Join(p, "2_second.sql"), []byte("create table t2 (c2 int)"), 0644))
-	_, err = runCmd(Root, "migrate", "hash", "--dir", "file://"+p)
+	_, err = runCmd(migrateHashCmd(), "--dir", "file://"+p)
 	require.NoError(t, err)
 	s, err = runCmd(
-		Root, "migrate", "validate",
+		migrateValidateCmd(),
 		"--dir", "file://"+p,
 		"--dev-url", openSQLite(t, ""),
 	)
@@ -648,17 +645,17 @@ func TestMigrate_Validate(t *testing.T) {
 	require.NoError(t, err)
 
 	// Should fail since the files are not compatible with SQLite.
-	_, err = runCmd(Root, "migrate", "validate", "--dir", "file://testdata/mysql", "--dev-url", openSQLite(t, ""))
+	_, err = runCmd(migrateValidateCmd(), "--dir", "file://testdata/mysql", "--dev-url", openSQLite(t, ""))
 	require.Error(t, err)
 }
 
 func TestMigrate_Hash(t *testing.T) {
-	s, err := runCmd(Root, "migrate", "hash", "--dir", "file://testdata/mysql")
+	s, err := runCmd(migrateHashCmd(), "--dir", "file://testdata/mysql")
 	require.Zero(t, s)
 	require.NoError(t, err)
 
 	// Prints a warning if --force flag is still used.
-	s, err = runCmd(Root, "migrate", "hash", "--dir", "file://testdata/mysql", "--force")
+	s, err = runCmd(migrateHashCmd(), "--dir", "file://testdata/mysql", "--force")
 	require.NoError(t, err)
 	require.Equal(t, "Flag --force has been deprecated, you can safely omit it.\n", s)
 
@@ -666,7 +663,7 @@ func TestMigrate_Hash(t *testing.T) {
 	err = copyFile(filepath.Join("testdata", "mysql", "20220318104614_initial.sql"), filepath.Join(p, "20220318104614_initial.sql"))
 	require.NoError(t, err)
 
-	s, err = runCmd(Root, "migrate", "hash", "--dir", "file://"+p)
+	s, err = runCmd(migrateHashCmd(), "--dir", "file://"+p)
 	require.Zero(t, s)
 	require.NoError(t, err)
 	require.FileExists(t, filepath.Join(p, "atlas.sum"))
@@ -685,7 +682,7 @@ func TestMigrate_Hash(t *testing.T) {
 		filepath.Join("testdata", "mysql", "20220318104614_initial.sql"),
 		filepath.Join(p, "20220318104614_initial.sql"),
 	))
-	s, err = runCmd(Root, "migrate", "hash", "--dir", "file://"+os.Getenv("MIGRATION_DIR"))
+	s, err = runCmd(migrateHashCmd(), "--dir", "file://"+os.Getenv("MIGRATION_DIR"))
 	require.NotZero(t, s)
 	require.Error(t, err)
 }
@@ -693,7 +690,7 @@ func TestMigrate_Hash(t *testing.T) {
 func TestMigrate_Lint(t *testing.T) {
 	p := t.TempDir()
 	s, err := runCmd(
-		Root, "migrate", "lint",
+		migrateLintCmd(),
 		"--dir", "file://"+p,
 		"--dev-url", openSQLite(t, ""),
 		"--latest", "1",
@@ -706,7 +703,7 @@ func TestMigrate_Lint(t *testing.T) {
 	err = os.WriteFile(filepath.Join(p, "2.sql"), []byte("DROP TABLE t;"), 0600)
 	require.NoError(t, err)
 	s, err = runCmd(
-		Root, "migrate", "lint",
+		migrateLintCmd(),
 		"--dir", "file://"+p,
 		"--dev-url", openSQLite(t, ""),
 		"--latest", "1",
@@ -714,7 +711,7 @@ func TestMigrate_Lint(t *testing.T) {
 	require.Error(t, err)
 	require.Equal(t, "2.sql: destructive changes detected:\n\n\tL1: Dropping table \"t\"\n\n", s)
 	s, err = runCmd(
-		Root, "migrate", "lint",
+		migrateLintCmd(),
 		"--dir", "file://"+p,
 		"--dev-url", openSQLite(t, ""),
 		"--latest", "1",
@@ -727,7 +724,7 @@ func TestMigrate_Lint(t *testing.T) {
 	require.NoError(t, os.Rename(filepath.Join(p, "1.sql"), filepath.Join(p, "1.up.sql")))
 	require.NoError(t, os.Rename(filepath.Join(p, "2.sql"), filepath.Join(p, "1.down.sql")))
 	s, err = runCmd(
-		Root, "migrate", "lint",
+		migrateLintCmd(),
 		"--dir", "file://"+p+"?format="+formatGolangMigrate,
 		"--dev-url", openSQLite(t, ""),
 		"--latest", "2",
@@ -737,11 +734,10 @@ func TestMigrate_Lint(t *testing.T) {
 	require.Equal(t, "1.up.sql:0", s)
 
 	// Invalid files.
-	MigrateFlags.Lint.Format = ""
 	err = os.WriteFile(filepath.Join(p, "2.up.sql"), []byte("BORING"), 0600)
 	require.NoError(t, err)
 	s, err = runCmd(
-		Root, "migrate", "lint",
+		migrateLintCmd(),
 		"--dir", "file://"+p+"?format="+formatGolangMigrate,
 		"--dev-url", openSQLite(t, ""),
 		"--latest", "1",
diff --git a/cmd/atlas/internal/cmdapi/schema_test.go b/cmd/atlas/internal/cmdapi/schema_test.go
index 91b3fcd2..8da71cef 100644
--- a/cmd/atlas/internal/cmdapi/schema_test.go
+++ b/cmd/atlas/internal/cmdapi/schema_test.go
@@ -5,10 +5,8 @@
 package cmdapi
 
 import (
-	"bytes"
 	"context"
 	"fmt"
-	"io/ioutil"
 	"os"
 	"path"
 	"path/filepath"
@@ -32,6 +30,105 @@ const (
 `
 )
 
+func TestCmdSchemaDiff(t *testing.T) {
+	// Creates the missing table.
+	s, err := runCmd(
+		schemaDiffCmd(),
+		"--from", openSQLite(t, ""),
+		"--to", openSQLite(t, "create table t1 (id int);"),
+	)
+	require.NoError(t, err)
+	require.EqualValues(t, "-- Create \"t1\" table\nCREATE TABLE `t1` (`id` int NULL)\n", s)
+
+	// No changes.
+	s, err = runCmd(
+		schemaDiffCmd(),
+		"--from", openSQLite(t, ""),
+		"--to", openSQLite(t, ""),
+	)
+	require.NoError(t, err)
+	require.EqualValues(t, "Schemas are synced, no changes to be made.\n", s)
+
+	// Desired state from migration directory requires dev database.
+	_, err = runCmd(
+		schemaDiffCmd(),
+		"--from", "file://testdata/sqlite",
+		"--to", openSQLite(t, ""),
+	)
+	require.EqualError(t, err, "--dev-url cannot be empty")
+
+	// Desired state from migration directory.
+	s, err = runCmd(
+		schemaDiffCmd(),
+		"--from", openSQLite(t, ""),
+		"--to", "file://testdata/sqlite",
+		"--dev-url", openSQLite(t, ""),
+	)
+	require.NoError(t, err)
+	require.EqualValues(t, "-- Create \"tbl\" table\nCREATE TABLE `tbl` (`col` int NOT NULL, `col_2` bigint NULL)\n", s)
+
+	// Desired state from migration directory.
+	s, err = runCmd(
+		schemaDiffCmd(),
+		"--from", openSQLite(t, ""),
+		"--to", "file://testdata/sqlite",
+		"--dev-url", openSQLite(t, ""),
+	)
+	require.NoError(t, err)
+	require.EqualValues(t, "-- Create \"tbl\" table\nCREATE TABLE `tbl` (`col` int NOT NULL, `col_2` bigint NULL)\n", s)
+
+	// Current state from migration directory, desired state from HCL - synced.
+	p := filepath.Join(t.TempDir(), "schema.hcl")
+	require.NoError(t, os.WriteFile(p, []byte(`schema "main" {}
+table "tbl" {
+  schema = schema.main
+  column "col" {
+    type = int
+  }
+  column "col_2" {
+    type = bigint
+    null = true
+  }
+}`), 0644))
+	s, err = runCmd(
+		schemaDiffCmd(),
+		"--from", "file://testdata/sqlite",
+		"--to", "file://"+p,
+		"--dev-url", openSQLite(t, ""),
+	)
+	require.NoError(t, err)
+	require.EqualValues(t, "Schemas are synced, no changes to be made.\n", s)
+
+	// Current state from migration directory, desired state from HCL - missing column.
+	p = filepath.Join(t.TempDir(), "schema.hcl")
+	require.NoError(t, os.WriteFile(p, []byte(`schema "main" {}
+table "tbl" {
+  schema = schema.main
+  column "col" {
+    type = int
+  }
+  column "col_2" {
+    type = bigint
+    null = true
+  }
+  column "col_3" {
+    type = text
+  }
+}`), 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,
+	)
+}
+
 func TestFmt(t *testing.T) {
 	for _, tt := range []struct {
 		name          string
@@ -100,7 +197,8 @@ func TestFmt(t *testing.T) {
 	} {
 		t.Run(tt.name, func(t *testing.T) {
 			dir := setupFmtTest(t, tt.inputDir)
-			out := runFmt(t, tt.args)
+			out, err := runCmd(schemaFmtCmd(), tt.args...)
+			require.NoError(t, err)
 			assertDir(t, dir, tt.expectedDir)
 			require.EqualValues(t, tt.expectedOut, out)
 		})
@@ -115,32 +213,21 @@ func TestSchema_Clean(t *testing.T) {
 	require.NoError(t, err)
 
 	// Apply migrations onto database.
-	MigrateFlags.Apply.BaselineVersion = ""
-	_, err = runCmd(Root, "migrate", "apply", "--dir", "file://testdata/sqlite", "--url", u)
+	_, err = runCmd(migrateApplyCmd(), "--dir", "file://testdata/sqlite", "--url", u)
 	require.NoError(t, err)
 
 	// Run clean and expect to be clean.
-	_, err = runCmd(Root, "migrate", "apply", "--dir", "file://testdata/sqlite", "--url", u)
+	_, err = runCmd(migrateApplyCmd(), "--dir", "file://testdata/sqlite", "--url", u)
 	require.NoError(t, err)
-	s, err := runCmd(Root, "schema", "clean", "--url", u, "--auto-approve")
+	s, err := runCmd(schemaCleanCmd(), "--url", u, "--auto-approve")
 	require.NoError(t, err)
 	require.NotZero(t, s)
 	require.NoError(t, c.Driver.(migrate.CleanChecker).CheckClean(context.Background(), nil))
 }
 
-func runFmt(t *testing.T, args []string) string {
-	var out bytes.Buffer
-	SchemaFmt.ResetCommands() // Detach from sub-commands and parents, needed to skip input validation done by them.
-	SchemaFmt.SetOut(&out)
-	SchemaFmt.SetArgs(args)
-	err := SchemaFmt.Execute()
-	require.NoError(t, err)
-	return out.String()
-}
-
 func assertDir(t *testing.T, dir string, expected map[string]string) {
 	act := make(map[string]string)
-	files, err := ioutil.ReadDir(dir)
+	files, err := os.ReadDir(dir)
 	require.NoError(t, err)
 	for _, f := range files {
 		if f.IsDir() {
diff --git a/sql/sqlclient/client_test.go b/sql/sqlclient/client_test.go
index 297d0a71..d9c45192 100644
--- a/sql/sqlclient/client_test.go
+++ b/sql/sqlclient/client_test.go
@@ -57,7 +57,7 @@ func TestRegisterOpen(t *testing.T) {
 	require.Equal(t, "schema", c.URL.Schema)
 
 	c1, err = sqlclient.Open(context.Background(), "postgres://:3306")
-	require.EqualError(t, err, `sql/sqlclient: no opener was register with name "postgres"`)
+	require.EqualError(t, err, `sql/sqlclient: no opener was registered with name "postgres"`)
 }
 
 func TestClient_AddClosers(t *testing.T) {