Repository: ariga/atlas
Pull request #1033 (2022-08-09 13:49:18)

doc/md: Integrity hash file


Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 07367ce6..5779ba7a 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -137,7 +137,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for mysql56
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="MySQL.*/mysql56|MySQL.*/.*/mysql56|MySQL.*/.*/.*/mysql56" -dialect=mysql56 ./...
+        run: go test -race -count=2 -v -run="MySQL" -version="mysql56" ./...
   
   integration-mysql57:
     runs-on: ubuntu-latest
@@ -168,7 +168,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for mysql57
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="MySQL.*/mysql57|MySQL.*/.*/mysql57|MySQL.*/.*/.*/mysql57" -dialect=mysql57 ./...
+        run: go test -race -count=2 -v -run="MySQL" -version="mysql57" ./...
   
   integration-mysql8:
     runs-on: ubuntu-latest
@@ -199,7 +199,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for mysql8
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="MySQL.*/mysql8|MySQL.*/.*/mysql8|MySQL.*/.*/.*/mysql8" -dialect=mysql8 ./...
+        run: go test -race -count=2 -v -run="MySQL" -version="mysql8" ./...
   
   integration-maria107:
     runs-on: ubuntu-latest
@@ -230,7 +230,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for maria107
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="MySQL.*/maria107|MySQL.*/.*/maria107|MySQL.*/.*/.*/maria107" -dialect=maria107 ./...
+        run: go test -race -count=2 -v -run="MySQL" -version="maria107" ./...
   
   integration-maria102:
     runs-on: ubuntu-latest
@@ -261,7 +261,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for maria102
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="MySQL.*/maria102|MySQL.*/.*/maria102|MySQL.*/.*/.*/maria102" -dialect=maria102 ./...
+        run: go test -race -count=2 -v -run="MySQL" -version="maria102" ./...
   
   integration-maria103:
     runs-on: ubuntu-latest
@@ -292,7 +292,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for maria103
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="MySQL.*/maria103|MySQL.*/.*/maria103|MySQL.*/.*/.*/maria103" -dialect=maria103 ./...
+        run: go test -race -count=2 -v -run="MySQL" -version="maria103" ./...
   
   integration-postgres10:
     runs-on: ubuntu-latest
@@ -322,7 +322,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for postgres10
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="Postgres.*/postgres10|Postgres.*/.*/postgres10|Postgres.*/.*/.*/postgres10" -dialect=postgres10 ./...
+        run: go test -race -count=2 -v -run="Postgres" -version="postgres10" ./...
   
   integration-postgres11:
     runs-on: ubuntu-latest
@@ -352,7 +352,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for postgres11
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="Postgres.*/postgres11|Postgres.*/.*/postgres11|Postgres.*/.*/.*/postgres11" -dialect=postgres11 ./...
+        run: go test -race -count=2 -v -run="Postgres" -version="postgres11" ./...
   
   integration-postgres12:
     runs-on: ubuntu-latest
@@ -382,7 +382,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for postgres12
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="Postgres.*/postgres12|Postgres.*/.*/postgres12|Postgres.*/.*/.*/postgres12" -dialect=postgres12 ./...
+        run: go test -race -count=2 -v -run="Postgres" -version="postgres12" ./...
   
   integration-postgres13:
     runs-on: ubuntu-latest
@@ -412,7 +412,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for postgres13
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="Postgres.*/postgres13|Postgres.*/.*/postgres13|Postgres.*/.*/.*/postgres13" -dialect=postgres13 ./...
+        run: go test -race -count=2 -v -run="Postgres" -version="postgres13" ./...
   
   integration-postgres14:
     runs-on: ubuntu-latest
@@ -442,7 +442,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for postgres14
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="Postgres.*/postgres14|Postgres.*/.*/postgres14|Postgres.*/.*/.*/postgres14" -dialect=postgres14 ./...
+        run: go test -race -count=2 -v -run="Postgres" -version="postgres14" ./...
   
   integration-tidb5:
     runs-on: ubuntu-latest
@@ -466,7 +466,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for tidb5
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="TiDB.*/tidb5|TiDB.*/.*/tidb5|TiDB.*/.*/.*/tidb5" -dialect=tidb5 ./...
+        run: go test -race -count=2 -v -run="TiDB" -version="tidb5" ./...
   
   integration-tidb6:
     runs-on: ubuntu-latest
@@ -490,7 +490,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for tidb6
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="TiDB.*/tidb6|TiDB.*/.*/tidb6|TiDB.*/.*/.*/tidb6" -dialect=tidb6 ./...
+        run: go test -race -count=2 -v -run="TiDB" -version="tidb6" ./...
   
   integration-sqlite:
     runs-on: ubuntu-latest
@@ -508,7 +508,7 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for sqlite
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="SQLite.*" -dialect=sqlite ./...
+        run: go test -race -count=2 -v -run="SQLite.*" -version="sqlite" ./...
   
   integration-cockroach:
     runs-on: ubuntu-latest
@@ -532,5 +532,5 @@ jobs:
             ${{ runner.os }}-go-
       - name: Run integration tests for cockroach
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="Cockroach.*/Cockroach|Cockroach.*/.*/Cockroach|Cockroach.*/.*/.*/Cockroach" -dialect=cockroach ./...
+        run: go test -race -count=2 -v -run="Cockroach" -version="cockroach" ./...
   
\ No newline at end of file
diff --git a/cmd/atlas/go.mod b/cmd/atlas/go.mod
index 09975535..0b27bf37 100644
--- a/cmd/atlas/go.mod
+++ b/cmd/atlas/go.mod
@@ -3,7 +3,7 @@ module ariga.io/atlas/cmd/atlas
 go 1.18
 
 require (
-	ariga.io/atlas v0.5.1-0.20220808114055-54febc21ee29
+	ariga.io/atlas v0.5.1-0.20220811124629-4cc73f4dbf32
 	entgo.io/ent v0.11.1
 	github.com/fatih/color v1.13.0
 	github.com/go-sql-driver/mysql v1.6.0
diff --git a/cmd/atlas/go.sum b/cmd/atlas/go.sum
index f83a71af..2ca3b5fd 100644
--- a/cmd/atlas/go.sum
+++ b/cmd/atlas/go.sum
@@ -4,6 +4,8 @@ ariga.io/atlas v0.5.1-0.20220801112446-175f0b2f060f h1:Gdw9juu/OoFCdzup+UTT/oiz7
 ariga.io/atlas v0.5.1-0.20220801112446-175f0b2f060f/go.mod h1:ofVetkJqlaWle3mvYmaS2uyFGFcc7dSq436tmxa/Mzk=
 ariga.io/atlas v0.5.1-0.20220808114055-54febc21ee29 h1:0BzoJr9DFDrvv9TAzQIYYvn2nwXnWdJlbE9zV2dSiqo=
 ariga.io/atlas v0.5.1-0.20220808114055-54febc21ee29/go.mod h1:ft47uSh5hWGDCmQC9DsztZg6Xk+KagM5Ts/mZYKb9JE=
+ariga.io/atlas v0.5.1-0.20220811124629-4cc73f4dbf32 h1:TstmfJ+ZVxvRpOHMuzPOJkP60sU/ghQGW9WQFs/D5sU=
+ariga.io/atlas v0.5.1-0.20220811124629-4cc73f4dbf32/go.mod h1:ft47uSh5hWGDCmQC9DsztZg6Xk+KagM5Ts/mZYKb9JE=
 entgo.io/ent v0.11.1 h1:im67R+2W3Nee2bNS2YnoYz8oAF0Qz4AOlIvKRIAEISY=
 entgo.io/ent v0.11.1/go.mod h1:X5b1YfMayrRTgKGO//8IqpL7XJx0uqdeReEkxNpXROA=
 github.com/DATA-DOG/go-sqlmock v1.5.0 h1:Shsta01QNfFxHCfpW6YH2STWB0MudeXXEWMr20OEh60=
diff --git a/cmd/atlas/internal/cmdapi/migrate.go b/cmd/atlas/internal/cmdapi/migrate.go
index 2bb3a127..9df1303e 100644
--- a/cmd/atlas/internal/cmdapi/migrate.go
+++ b/cmd/atlas/internal/cmdapi/migrate.go
@@ -45,10 +45,10 @@ const (
 	migrateFlagDryRun           = "dry-run"
 	migrateFlagTo               = "to"
 	migrateFlagSchema           = "schema"
-	migrateDiffFlagVerbose      = "verbose"
 	migrateLintLatest           = "latest"
 	migrateLintGitDir           = "git-dir"
 	migrateLintGitBase          = "git-base"
+	migrateDiffQualifier        = "qualifier"
 	migrateApplyAllowDirty      = "allow-dirty"
 	migrateApplyFromVersion     = "from"
 	migrateApplyBaselineVersion = "baseline"
@@ -65,7 +65,6 @@ var (
 		DirFormat      string
 		RevisionSchema string
 		Force          bool
-		Verbose        bool
 		Apply          struct {
 			DryRun          bool
 			LogFormat       string
@@ -73,6 +72,9 @@ var (
 			FromVersion     string
 			BaselineVersion string
 		}
+		Diff struct {
+			Qualifier string // optional table qualifier
+		}
 		Lint struct {
 			Format  string // log formatting
 			Latest  uint   // latest N migration files
@@ -247,7 +249,7 @@ func init() {
 	// 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().BoolVarP(&MigrateFlags.Verbose, migrateDiffFlagVerbose, "", false, "enable verbose logging")
+	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))
@@ -289,6 +291,15 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 		return err
 	}
 	defer c.Close()
+	// Acquire a lock.
+	if l, ok := c.Driver.(schema.Locker); ok {
+		unlock, err := l.Lock(cmd.Context(), "atlas_migrate_execute", 0)
+		if err != nil {
+			return fmt.Errorf("acquiring database lock: %w", err)
+		}
+		// If unlocking fails notify the user about it.
+		defer cobra.CheckErr(unlock())
+	}
 	// Get the correct log format and destination. Currently, only os.Stdout is supported.
 	l, err := logFormat(cmd.OutOrStdout())
 	if err != nil {
@@ -315,11 +326,6 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 	if err != nil {
 		return err
 	}
-	unlock, err := ex.Lock(cmd.Context())
-	if err != nil {
-		return err
-	}
-	defer unlock()
 	pending, err := ex.Pending(cmd.Context())
 	if err != nil && !errors.Is(err, migrate.ErrNoPendingFiles) {
 		return err
@@ -407,7 +413,7 @@ func CmdMigrateDiffRun(cmd *cobra.Command, args []string) error {
 	if l, ok := dev.Driver.(schema.Locker); ok {
 		unlock, err := l.Lock(cmd.Context(), "atlas_migrate_diff", 0)
 		if err != nil {
-			return err
+			return fmt.Errorf("acquiring database lock: %w", err)
 		}
 		// If unlocking fails notify the user about it.
 		defer cobra.CheckErr(unlock())
@@ -429,7 +435,8 @@ func CmdMigrateDiffRun(cmd *cobra.Command, args []string) error {
 	}
 	opts := []migrate.PlannerOption{migrate.PlanFormat(f)}
 	if dev.URL.Schema != "" {
-		opts = append(opts, migrate.PlanSchema(dev.URL.Schema))
+		// Disable tables qualifier in schema-mode.
+		opts = append(opts, migrate.PlanWithSchemaQualifier(MigrateFlags.Diff.Qualifier))
 	}
 	// Plan the changes and create a new migration file.
 	pl := migrate.NewPlanner(dev.Driver, dir, opts...)
@@ -437,8 +444,14 @@ func CmdMigrateDiffRun(cmd *cobra.Command, args []string) error {
 	if len(args) > 0 {
 		name = args[0]
 	}
+	plan, err := func() (*migrate.Plan, error) {
+		if dev.URL.Schema != "" {
+			return pl.PlanSchema(cmd.Context(), name, desired.StateReader)
+		}
+		return pl.Plan(cmd.Context(), name, desired.StateReader)
+	}()
 	var cerr migrate.NotCleanError
-	switch plan, err := pl.Plan(cmd.Context(), name, desired.StateReader); {
+	switch {
 	case errors.Is(err, migrate.ErrNoPlan):
 		cmd.Println("The migration directory is synced with the desired state, no changes to be made")
 		return nil
@@ -539,10 +552,8 @@ func statusPrint(out io.Writer, avail, pending []migrate.File, revs migrate.Revi
 	switch len(pending) {
 	case len(avail):
 		cur = "No version applied yet"
-	case 0:
-		cur = cyan(avail[len(avail)-1].Version())
 	default:
-		cur = cyan(avail[len(avail)-len(pending)].Version())
+		cur = cyan(applied[len(applied)-1].Version())
 		// If the last pending version is partially applied, tell so.
 		if partial {
 			cur += fmt.Sprintf(" (%d statements applied)", revs[len(revs)-1].Applied)
@@ -562,16 +573,15 @@ func statusPrint(out io.Writer, avail, pending []migrate.File, revs migrate.Revi
 	if partial {
 		exec += " + 1 partially"
 	}
-	fmt.Fprintf(out, "Migration Status: %s\n", state)
-	fmt.Fprintf(out, "%s%s Current Version:\t%s\n", indent2, dash, cur)
-	fmt.Fprintf(out, "%s%s Next Version:\t%s\n", indent2, dash, next)
-	// fmt.Fprintf(out, "%s%s Available Files:\t%s\n", indent2, dash, cyan(strconv.Itoa(len(avail))))
-	fmt.Fprintf(out, "%s%s Executed Files:\t%s\n", indent2, dash, exec)
 	c := cyan
 	if len(pending) == 0 {
 		c = green
 	}
-	fmt.Fprintf(out, "%s%s Pending Files:\t%s", indent2, dash, c(strconv.Itoa(len(pending))))
+	fmt.Fprintf(out, "Migration Status: %s\n", state)
+	fmt.Fprintf(out, "%s%s Current Version:  %s\n", indent2, dash, cur)
+	fmt.Fprintf(out, "%s%s Next Version:     %s\n", indent2, dash, next)
+	fmt.Fprintf(out, "%s%s Executed Files:   %s\n", indent2, dash, exec)
+	fmt.Fprintf(out, "%s%s Pending Files:    %s", indent2, dash, c(strconv.Itoa(len(pending))))
 	if partial {
 		fmt.Fprintf(out, " (partially)")
 	}
@@ -601,7 +611,12 @@ func CmdMigrateValidateRun(cmd *cobra.Command, _ []string) error {
 	if err != nil {
 		return err
 	}
-	if _, err := ex.ReadState(cmd.Context()); err != nil && !errors.Is(err, migrate.ErrNoPendingFiles) {
+	if _, err := ex.Replay(cmd.Context(), func() migrate.StateReader {
+		if dev.URL.Schema != "" {
+			return migrate.SchemaConn(dev, "", nil)
+		}
+		return migrate.RealmConn(dev, nil)
+	}()); err != nil && !errors.Is(err, migrate.ErrNoPendingFiles) {
 		return fmt.Errorf("replaying the migration directory: %w", err)
 	}
 	return nil
diff --git a/cmd/atlas/internal/cmdapi/schema.go b/cmd/atlas/internal/cmdapi/schema.go
index 49da6d84..517a9a05 100644
--- a/cmd/atlas/internal/cmdapi/schema.go
+++ b/cmd/atlas/internal/cmdapi/schema.go
@@ -55,7 +55,6 @@ var (
 		Paths       []string
 		DryRun      bool
 		AutoApprove bool
-		Verbose     bool
 	}
 	// SchemaApply represents the 'atlas schema apply' subcommand command.
 	SchemaApply = &cobra.Command{
@@ -143,7 +142,6 @@ func init() {
 	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, "auto-approve", "", false, "Auto approve. Apply the schema changes without prompting for approval.")
-	SchemaApply.Flags().BoolVarP(&ApplyFlags.Verbose, migrateDiffFlagVerbose, "", false, "enable verbose logging")
 	SchemaApply.Flags().StringVarP(&SchemaFlags.DSN, dsnFlag, "d", "", "")
 	cobra.CheckErr(SchemaApply.Flags().MarkHidden(dsnFlag))
 	cobra.CheckErr(SchemaApply.MarkFlagRequired(urlFlag))
diff --git a/doc/md/concepts/integrity_hash_file.md b/doc/md/concepts/integrity_hash_file.md
new file mode 100644
index 00000000..aeba8218
--- /dev/null
+++ b/doc/md/concepts/integrity_hash_file.md
@@ -0,0 +1,21 @@
+---
+id: sum_file
+slug: /concepts/integrity_hash_file
+title: Integrity Sum File
+---
+
+## Integrity Hash File
+When multiple engineers work on the same project in parallel, there is a chance that their migration scripts will conflict with each other and may cause unexpected behavior. For example:
+1. One can retroactively change the content of migrations that have already run.
+2. One can accidentally change the order in which migrations are organized.
+3. One can commit semantically incorrect SQL scripts.
+
+Atlas creates the Integrity Hash File, internally named `atlas.sum` - to tackle these issues. 
+
+The atlas.sum file contains a sum of the whole directory as its first entry, and a checksum for each of the migration files. Its format is inspired by the go.sum file of a Go module, and it would look similar to this:
+```
+h1:KRFsSi68ZOarsQAJZ1mfSiMSkIOZlMq4RzyF//Pwf8A=
+20220318104614_team_A.sql h1:EGknG5Y6GQYrc4W8e/r3S61Aqx2p+NmQyVz/2m8ZNwA=
+```
+
+On each run of `atlas migrate` Atlas verifies that your migration directory matches your `atlas.sum file`. You’ll see an alert in case there’s a discrepancy between the files` signatures and the atlas.sum file.
diff --git a/doc/md/reference.md b/doc/md/reference.md
index c0f6aff7..7d38bb32 100644
--- a/doc/md/reference.md
+++ b/doc/md/reference.md
@@ -142,9 +142,9 @@ 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
-      --verbose          enable verbose logging
+      --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
 
 ```
 
@@ -333,7 +333,6 @@ migration.
                          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.
-      --verbose          enable verbose logging
 
 ```
 
diff --git a/doc/md/versioned/diff.md b/doc/md/versioned/diff.md
deleted file mode 100644
index a6a82e2c..00000000
--- a/doc/md/versioned/diff.md
+++ /dev/null
@@ -1,36 +0,0 @@
----
-id: diff
-slug: /versioned/diff
-title: Planning new migrations
----
-With the `atlas migrate` commands users can implement a kind fo workflow that we
-call _versioned migration authoring_. This kind of workflow is a synthesis between _declarative_ workflows,
-where developers specify the desired state of their database, and _versioned migrations_
-where each change is explicitly defined as a migration script with a specific version.
-
-Practically speaking, this means that the developer maintains the [schema definition](/atlas-schema/sql-resources),
-e.g the _desired state_, and Atlas maintains the `migrations/` directory, which contains the
-explicit SQL scripts to move from one version to the next.
-
-In addition, Atlas maintains a file name `atlas.sum` which is used to ensure the integrity of
-the migration directory and force developers to deal with situations where migration order or
-contents was modified after the fact.
-
-[Learn more about versioned migration authoring](/concepts/declarative-vs-versioned#migration-authoring)
-
-### Flags
-When using `migrate diff` to plan a migration users must supply multiple parameters:
-* `--dev-url` a [URL](/concepts/url) to a [Dev-database](/concepts/dev-database) that will be used
- to compute the diff.
-* `--dir` the URL of the migration directory, by default it is `file://migrations`, e.g a
- directory named `migrations` in the current working directory.
-* `--to` the URL of the desired state, can be an HCL file or another database.
-
-### Example
-```
-atlas migrate diff --dir file://my/project/migrations --to schema.hcl --dev-url mysql://root:pass@localhost:3306/dev
-```
-
-### Reference
-
-[CLI Command Reference](/cli-reference#atlas-migrate-diff)
\ No newline at end of file
diff --git a/doc/md/versioned/diff.mdx b/doc/md/versioned/diff.mdx
new file mode 100644
index 00000000..402b95ed
--- /dev/null
+++ b/doc/md/versioned/diff.mdx
@@ -0,0 +1,372 @@
+---
+id: diff
+slug: /versioned/diff
+title: Automatic Migration Authoring
+---
+
+import Tabs from '@theme/Tabs';
+import TabItem from '@theme/TabItem';
+
+With the `atlas migrate diff` command users can implement a kind of workflow that we
+call _versioned migration authoring_. This workflow is a synthesis between _declarative_ workflows,
+where developers specify the desired state of their database, and _versioned migrations_
+where each change is explicitly defined as a migration script with a specific version.
+
+Practically speaking, this means developers define the desired state with either an HCL
+[schema definition](/atlas-schema/sql-resources) or a database connection, and Atlas maintains
+the `migrations` directory, which contains the explicit SQL scripts to move from one version to
+the next.
+
+### Generate migrations from Atlas schemas
+
+Suppose we have an Atlas schema with one table and an empty migration directory:
+
+```hcl title="schema.hcl"
+schema "test" {}
+
+table "users" {
+  schema = schema.test
+  column "id" {
+    type = int
+  }
+}
+```
+
+Let's run `atlas migrate diff` with the necessary parameters to generate a migration script for
+creating our `users` table:
+
+* `--dir` the URL to the migration directory, by default it is `file://migrations`.
+* `--to` the URL of the desired state, an HCL file or a database connection.
+* `--dev-url` a [URL](/concepts/url) to a [Dev Database](/concepts/dev-database) that will be used to compute the diff.
+
+```bash
+atlas migrate diff create_users \
+  --dir="file://migrations" \
+  --to="file://schema.hcl" \
+  --dev-url="mysql://root:pass@:3306/test"
+```
+
+Run `ls migrations`, and you will notice Atlas created 2 files:
+
+<Tabs
+defaultValue="migration_file"
+values={[
+{label: '20220811074144_create_users.sql', value: 'migration_file'},
+{label: 'atlas.sum', value: 'sum_file'},
+]}>
+<TabItem value="migration_file">
+
+By default, migration files are named with the following format `{{ now }}_{{ name }}.sql`.
+If you wish to use a different file format, use the `--dir-format` option.
+
+```sql
+-- create "users" table
+CREATE TABLE `users` (`id` int NOT NULL) CHARSET utf8mb4 COLLATE utf8mb4_0900_ai_ci;
+```
+
+</TabItem>
+<TabItem value="sum_file">
+
+In addition to the migration directory, Atlas maintains a file name `atlas.sum` which is used
+to ensure the integrity of the migration directory and force developers to deal with situations
+where migration order or contents was modified after the fact.
+
+```text
+h1:t1fEP1rSsGf1gYrYCjsGyEyuM0cnhATlq93B7h8uXxY=
+20220811074144_create_users.sql h1:liZcCBbAn/HyBTqBAEVar9fJNKPTb2Eq+rEKZeCFC9M=
+```
+
+</TabItem>
+</Tabs>
+
+Let's repeat the process above by changing our HCL schema file and running Atlas migration authoring again.
+We add a new column `name` to our HCL schema:
+
+```hcl title="schema.hcl" {8-10}
+schema "test" {}
+
+table "users" {
+  schema = schema.test
+  column "id" {
+    type = int
+  }
+  column "name" {
+    type = varchar(255)
+  }
+}
+```
+
+Then, run `atlas migrate diff`:
+
+```bash
+atlas migrate diff add_users_name \
+  --dir="file://migrations" \
+  --to="file://schema.hcl" \
+  --dev-url="mysql://root:pass@:3306/test"
+```
+
+You will notice Atlas added a new file to the migration directory:
+
+<Tabs
+defaultValue="migration_file2"
+values={[
+{label: '20220811074144_create_users.sql', value: 'migration_file1'},
+{label: '20220811074314_add_users_name.sql', value: 'migration_file2'},
+{label: 'atlas.sum', value: 'sum_file'},
+]}>
+<TabItem value="migration_file1">
+
+```sql
+-- create "users" table
+CREATE TABLE `users` (`id` int NOT NULL) CHARSET utf8mb4 COLLATE utf8mb4_0900_ai_ci;
+```
+
+</TabItem>
+<TabItem value="migration_file2">
+
+```sql
+-- modify "users" table
+ALTER TABLE `users` ADD COLUMN `name` varchar(255) NOT NULL;
+```
+
+</TabItem>
+<TabItem value="sum_file">
+
+```text
+h1:w2ODzVxhTKdBVBdzqntHw7rHV8lKQF98TmNevOEZfIo=
+20220811074144_create_users_table.sql h1:KnMSZM/E4TBGidYCZ+UHxkHEWaRWeyuPIUjSHRybQqA=
+20220811074314_add_users_name.sql h1:jUpaANgD0SjI5DjaHuJxtHZ6Wq98act0MmE5oZ+NRU0=
+```
+
+</TabItem>
+</Tabs>
+
+The following diagram explains how it works. Atlas loads the **current state**
+by replaying the migration directory onto the provided [dev database](/concepts/dev-database),
+compares it against the **desired state** and writes a new migration script for moving from the current
+to the desired state.
+
+![Diff From Schema](https://atlasgo.io/uploads/images/versioned-migration/diff-from-schema.png)
+
+### Generate migrations from database schemas
+
+Suppose we have a database with a `users` table that was created manually or by an ORM like [Ent](https://entgo.io),
+we can tell Atlas that this is our **desired** state, and we want to generate a migration script to create this table.
+
+```text
+mysql> describe users;
+
++-------+------+------+-----+---------+-------+
+| Field | Type | Null | Key | Default | Extra |
++-------+------+------+-----+---------+-------+
+| id    | int  | YES  |     | NULL    |       |
++-------+------+------+-----+---------+-------+
+```
+
+Let's run `atlas migrate diff` with the necessary parameters to generate a migration script for
+creating our `users` table:
+
+```bash
+atlas migrate diff create_users \
+  --dir="file://migrations" \
+  --to="mysql://root:pass@:3306/public" \
+  --dev-url="mysql://root:pass@:3306/test"
+```
+
+Run `ls migrations`, and you will notice Atlas created 2 files:
+
+<Tabs
+defaultValue="migration_file"
+values={[
+{label: '20220811102532_create_users.sql', value: 'migration_file'},
+{label: 'atlas.sum', value: 'sum_file'},
+]}>
+<TabItem value="migration_file">
+
+By default, migration files are named with the following format `{{ now }}_{{ name }}.sql`.
+If you wish to use a different file format, use the `--dir-format` option.
+
+```sql
+-- create "users" table
+CREATE TABLE `users` (`id` int NOT NULL) CHARSET utf8mb4 COLLATE utf8mb4_0900_ai_ci;
+```
+
+</TabItem>
+<TabItem value="sum_file">
+
+In addition to the migration directory, Atlas maintains a file name `atlas.sum` which is used
+to ensure the integrity of the migration directory and force developers to deal with situations
+where migration order or contents was modified after the fact.
+
+```text
+h1:/6UW2WaPV1laJpEB7oVesKB9RuL59dgDhlTp5atDbNg=
+20220811102532_create_users.sql h1:IJiVpbh3FLMeDJSzxxKPuFU3m4AHBgThBfs3VFXAXVo=
+```
+
+</TabItem>
+</Tabs>
+
+### Generate migrations with custom qualifiers
+
+When working on a specific database schema, Atlas generates migration scripts without schema qualifiers
+to allow executing them multiple times on different schemas. However, in some cases, it is necessary
+to have those qualifiers. To address this, Atlas allows passing another flag to `migrate diff` named `--qualifier`.
+
+Let's run the [example above](#generate-migration-from-database-schema), with the `--qualifier` flag and compare
+the output:
+
+```bash
+atlas migrate diff create_users \
+  --dir="file://migrations" \
+  --to="file://schema.hcl" \
+  --dev-url="mysql://root:pass@:3306/test" \
+  --qualifier="market"
+```
+
+Running `cat migrations/*.sql` will print the same migration script but the `users` table will be qualified with
+the `market` schema:
+
+```sql
+-- create "users" table
+CREATE TABLE `market`.`users` (`id` int NOT NULL, `name` varchar(255) NOT NULL) CHARSET utf8mb4 COLLATE utf8mb4_0900_ai_ci;
+```
+
+### Generate migrations with custom formats
+
+Some migration tools use a different file format than the one used by Atlas. Therefore, Atlas provides
+an option named `--dir-format` that allows controlling the format of the migration directory.
+
+```bash
+atlas migrate diff create_users \
+  --dir="file://migrations" \
+  --to="file://schema.hcl" \
+  --dev-url="mysql://root:pass@:3306/test" \
+  --dir-format="golang-migrate"
+```
+
+Run `ls migrations`, and you will notice Atlas created 3 files:
+
+<Tabs
+defaultValue="migration_file_up"
+values={[
+{label: '20220811114629_create_users.up.sql', value: 'migration_file_up'},
+{label: '20220811114629_create_users.down.sql', value: 'migration_file_down'},
+{label: 'atlas.sum', value: 'sum_file'},
+]}>
+<TabItem value="migration_file_up">
+
+```sql
+-- create "users" table
+CREATE TABLE `users` (`name` int NOT NULL) CHARSET utf8mb4 COLLATE utf8mb4_0900_ai_ci;
+```
+
+</TabItem>
+<TabItem value="migration_file_down">
+
+```sql
+-- reverse: create "users" table
+DROP TABLE `users`;
+```
+
+</TabItem>
+<TabItem value="sum_file">
+
+In addition to the migration directory, Atlas maintains a file name `atlas.sum` which is used
+to ensure the integrity of the migration directory and force developers to deal with situations
+where migration order or contents was modified after the fact.
+
+```text
+h1:NRHsNIjvSSzprr/EzMdtszQg3t3pVLk4G4N1tX4rMfk=
+20220811114629_create_users.up.sql h1:Ng3GHrdk2davokjOctgVdxC+6QsK4JzaLX6RT3QstJc=
+```
+
+</TabItem>
+</Tabs>
+
+### Generate migrations for the entire database
+
+Atlas supports generating migrations for databases or multiple schemas. In PostgreSQL, a database can
+be created with the `CREATE DATABASE` command and can hold multiple schemas. In MySQL however, a database
+is an instance with one or more schemas.
+
+Suppose we have an Atlas schema that defines two database schemas where each one contains a single table.
+
+```hcl title="schema.hcl"
+schema "auth" {}
+schema "market" {}
+
+table "users" {
+  schema = schema.market
+  column "name" {
+    type = int
+  }
+}
+
+table "tokens" {
+  schema = schema.auth
+  column "value" {
+    type = int
+  }
+}
+```
+
+Let's run `atlas migrate diff` to generate migration scripts for creating the entire schema.
+However, unlike the previous examples where the `--dev-url` flag was set to a [URL](concepts/url.mdx)
+of a specific schema, in this case we omit the schema name from the connection string.
+
+<Tabs
+defaultValue="mysql"
+values={[
+{label: 'MySQL', value: 'mysql'},
+{label: 'PostgreSQL', value: 'postgres'},
+]}>
+<TabItem value="mysql">
+
+```bash
+atlas migrate diff --dir="file://migrations" --to="file://schema.hcl" \
+ --dev-url="mysql://root:pass@:3306/" create_all
+```
+
+Running `cat migrations/*.sql` will print the followings:
+
+```sql
+-- add new schema named "auth"
+CREATE DATABASE `auth`;
+-- add new schema named "market"
+CREATE DATABASE `market`;
+-- create "tokens" table
+CREATE TABLE `auth`.`tokens` (`value` int NOT NULL) CHARSET utf8mb4 COLLATE utf8mb4_0900_ai_ci;
+-- create "users" table
+CREATE TABLE `market`.`users` (`name` int NOT NULL) CHARSET utf8mb4 COLLATE utf8mb4_0900_ai_ci;
+```
+
+</TabItem>
+<TabItem value="postgres">
+
+```bash
+atlas migrate diff --dir="file://migrations" --to="file://schema.hcl" \
+ --dev-url="postgres://root:pass@:5434/database?sslmode=disable" create_all
+```
+
+Running `cat migrations/*.sql` will print the followings:
+
+```sql
+-- Add new schema named "auth"
+CREATE SCHEMA "auth";
+-- Add new schema named "market"
+CREATE SCHEMA "market";
+-- create "tokens" table
+CREATE TABLE "auth"."tokens" ("value" integer NOT NULL);
+-- create "users" table
+CREATE TABLE "market"."users" ("name" integer NOT NULL);
+```
+
+</TabItem>
+</Tabs>
+
+As you can see, Atlas generates statements for creating the `auth` and `market` schemas,
+and added them as qualifiers in the created tables.
+
+### Reference
+
+[CLI Command Reference](/cli-reference#atlas-migrate-diff)
\ No newline at end of file
diff --git a/doc/website/sidebars.js b/doc/website/sidebars.js
index 313bf330..9aaacc8f 100644
--- a/doc/website/sidebars.js
+++ b/doc/website/sidebars.js
@@ -36,7 +36,7 @@ module.exports = {
             label: "Versioned Workflows",
             collapsed: false,
             items: [
-                {type: 'doc', id: 'versioned/diff', label: 'Migration Planning'},
+                {type: 'doc', id: 'versioned/diff', label: 'Migration Authoring'},
                 {type: 'doc', id: 'versioned/lint', label: 'Migration Linting'},
                 {type: 'doc', id: 'versioned/new', label: 'Manual Migrations'},
             ]
diff --git a/internal/ci/ci.tmpl b/internal/ci/ci.tmpl
index 5055ab72..b91a123b 100644
--- a/internal/ci/ci.tmpl
+++ b/internal/ci/ci.tmpl
@@ -135,5 +135,5 @@ jobs:
             {{ "${{ runner.os }}-go-" }}
       - name: Run integration tests for {{ .Version }}
         working-directory: internal/integration
-        run: go test -race -count=2 -v -run="{{ .Regex }}" -dialect={{ .Version }} ./...
+        run: go test -race -count=2 -v -run="{{ .Regex }}" -version="{{ .Version }}" ./...
   {{ end }}
\ No newline at end of file
diff --git a/internal/ci/main.go b/internal/ci/main.go
index 93f2fec0..394b707a 100644
--- a/internal/ci/main.go
+++ b/internal/ci/main.go
@@ -7,10 +7,8 @@ package main
 import (
 	"bytes"
 	_ "embed"
-	"fmt"
 	"log"
 	"os"
-	"strings"
 	"text/template"
 )
 
@@ -55,7 +53,7 @@ var (
 		{
 			Version: "mysql56",
 			Image:   "mysql:5.6.35",
-			Regex:   re("MySQL.*/%smysql56"),
+			Regex:   "MySQL",
 			Env:     mysqlEnv,
 			Ports:   []string{"3306:3306"},
 			Options: mysqlOptions,
@@ -63,7 +61,7 @@ var (
 		{
 			Version: "mysql57",
 			Image:   "mysql:5.7.26",
-			Regex:   re("MySQL.*/%smysql57"),
+			Regex:   "MySQL",
 			Env:     mysqlEnv,
 			Ports:   []string{"3307:3306"},
 			Options: mysqlOptions,
@@ -71,7 +69,7 @@ var (
 		{
 			Version: "mysql8",
 			Image:   "mysql:8",
-			Regex:   re("MySQL.*/%smysql8"),
+			Regex:   "MySQL",
 			Env:     mysqlEnv,
 			Ports:   []string{"3308:3306"},
 			Options: mysqlOptions,
@@ -79,7 +77,7 @@ var (
 		{
 			Version: "maria107",
 			Image:   "mariadb:10.7",
-			Regex:   re("MySQL.*/%smaria107"),
+			Regex:   "MySQL",
 			Env:     mysqlEnv,
 			Ports:   []string{"4306:3306"},
 			Options: mysqlOptions,
@@ -87,7 +85,7 @@ var (
 		{
 			Version: "maria102",
 			Image:   "mariadb:10.2.32",
-			Regex:   re("MySQL.*/%smaria102"),
+			Regex:   "MySQL",
 			Env:     mysqlEnv,
 			Ports:   []string{"4307:3306"},
 			Options: mysqlOptions,
@@ -95,7 +93,7 @@ var (
 		{
 			Version: "maria103",
 			Image:   "mariadb:10.3.13",
-			Regex:   re("MySQL.*/%smaria103"),
+			Regex:   "MySQL",
 			Env:     mysqlEnv,
 			Ports:   []string{"4308:3306"},
 			Options: mysqlOptions,
@@ -103,7 +101,7 @@ var (
 		{
 			Version: "postgres10",
 			Image:   "postgres:10",
-			Regex:   re("Postgres.*/%spostgres10"),
+			Regex:   "Postgres",
 			Env:     pgEnv,
 			Ports:   []string{"5430:5432"},
 			Options: pgOptions,
@@ -111,7 +109,7 @@ var (
 		{
 			Version: "postgres11",
 			Image:   "postgres:11",
-			Regex:   re("Postgres.*/%spostgres11"),
+			Regex:   "Postgres",
 			Env:     pgEnv,
 			Ports:   []string{"5431:5432"},
 			Options: pgOptions,
@@ -119,7 +117,7 @@ var (
 		{
 			Version: "postgres12",
 			Image:   "postgres:12.3",
-			Regex:   re("Postgres.*/%spostgres12"),
+			Regex:   "Postgres",
 			Env:     pgEnv,
 			Ports:   []string{"5432:5432"},
 			Options: pgOptions,
@@ -127,7 +125,7 @@ var (
 		{
 			Version: "postgres13",
 			Image:   "postgres:13.1",
-			Regex:   re("Postgres.*/%spostgres13"),
+			Regex:   "Postgres",
 			Env:     pgEnv,
 			Ports:   []string{"5433:5432"},
 			Options: pgOptions,
@@ -135,7 +133,7 @@ var (
 		{
 			Version: "postgres14",
 			Image:   "postgres:14",
-			Regex:   re("Postgres.*/%spostgres14"),
+			Regex:   "Postgres",
 			Env:     pgEnv,
 			Ports:   []string{"5434:5432"},
 			Options: pgOptions,
@@ -143,13 +141,13 @@ var (
 		{
 			Version: "tidb5",
 			Image:   "pingcap/tidb:v5.4.0",
-			Regex:   re("TiDB.*/%stidb5"),
+			Regex:   "TiDB",
 			Ports:   []string{"4309:4000"},
 		},
 		{
 			Version: "tidb6",
 			Image:   "pingcap/tidb:v6.0.0",
-			Regex:   re("TiDB.*/%stidb6"),
+			Regex:   "TiDB",
 			Ports:   []string{"4310:4000"},
 		},
 		{
@@ -159,20 +157,12 @@ var (
 		{
 			Version: "cockroach",
 			Image:   "ghcr.io/ariga/cockroachdb-single-node:v21.2.11",
-			Regex:   re("Cockroach.*/%sCockroach"),
+			Regex:   "Cockroach",
 			Ports:   []string{"26257:26257"},
 		},
 	}
 )
 
-func re(p string) string {
-	var s []string
-	for i := 0; i < 3; i++ {
-		s = append(s, fmt.Sprintf(p, strings.Repeat(".*/", i)))
-	}
-	return strings.Join(s, "|")
-}
-
 func main() {
 	var buf bytes.Buffer
 	if err := template.Must(template.New("").Parse(t)).Execute(&buf, jobs); err != nil {
diff --git a/internal/integration/README.md b/internal/integration/README.md
new file mode 100644
index 00000000..84be8d6b
--- /dev/null
+++ b/internal/integration/README.md
@@ -0,0 +1,58 @@
+### This directory contains all integration tests for Atlas.
+
+The provided `docker-compose.yaml` file contains images for each database the integration tests are run on. You can
+start them by calling:
+
+```shell
+docker-compose --project-name atlas-integration up -d
+```
+
+The whole integration suite is then run by executing within this directory: 
+
+```shell
+go test ./...
+```
+
+#### Selectively running tests
+
+Running all integration tests (and keeping all database containers up all the time) consumes time and resources (and
+power). You can execute only some of the tests by using the `-run` and `-dialect` flags:
+
+The below examples don't require for you to have all docker containers running, instead only the ones used in the tests
+have to be up.
+
+Consider the following test in `mysql_test.go`:
+
+```go
+func TestMySQL_Executor(t *testing.T) {
+	myRun(t, func(t *myTest) {
+		testExecutor(t)
+	})
+}
+```
+
+If you'd wanted to run that test only for mysql56, simply pass its full name into the `-run` flag:
+
+```shell
+# Run TestMySQL_Executor for all mysql versions
+go test -run='MySQL_Executor' ./... 
+
+# Run TestMySQL_Executor for mysql 5.6 only
+go test -run='MySQL_Executor/mysql56' ./...
+```
+
+If you'd like to run the above for Postgres 10, change the name respectively:
+
+```shell
+# Run TestMySQL_Executor for all mysql versions
+go test -run='Postgres_Executor' ./... 
+
+# Run TestMySQL_Executor for postgres 10 only
+go test -run='Postgres_Executor/postgres10' ./...
+```
+
+If you want to run all tests for one specific dialect, like only TiDB 5, you can use the `-dialect` flag:
+
+```shell
+go test -run='TiDB' -dialect='tidb5' ./...
+```
\ No newline at end of file
diff --git a/internal/integration/go.mod b/internal/integration/go.mod
index e02b702e..1010d63a 100644
--- a/internal/integration/go.mod
+++ b/internal/integration/go.mod
@@ -1,13 +1,13 @@
 module ariga.io/atlas/internal/integration
 
-go 1.17
+go 1.19
 
 replace ariga.io/atlas => ../../
 
 replace ariga.io/atlas/cmd/atlas => ../../cmd/atlas
 
 require (
-	ariga.io/atlas v0.5.1-0.20220808114055-54febc21ee29
+	ariga.io/atlas v0.5.1-0.20220811124629-4cc73f4dbf32
 	entgo.io/ent v0.11.1
 	github.com/go-sql-driver/mysql v1.6.0
 	github.com/hashicorp/hcl/v2 v2.13.0
@@ -20,7 +20,7 @@ require (
 )
 
 require (
-	ariga.io/atlas/cmd/atlas v0.5.1-0.20220808114055-54febc21ee29 // indirect
+	ariga.io/atlas/cmd/atlas v0.5.1-0.20220810063018-93d5f3ca9ef3 // indirect
 	github.com/agext/levenshtein v1.2.3 // indirect
 	github.com/apparentlymart/go-textseg/v13 v13.0.0 // indirect
 	github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e // indirect
diff --git a/sql/internal/specutil/convert.go b/sql/internal/specutil/convert.go
index 9543802a..46d19206 100644
--- a/sql/internal/specutil/convert.go
+++ b/sql/internal/specutil/convert.go
@@ -506,7 +506,11 @@ func FromForeignKey(s *schema.ForeignKey) (*sqlspec.ForeignKey, error) {
 	for _, v := range s.RefColumns {
 		ref := ColumnRef(v.Name)
 		if s.Table != s.RefTable {
-			ref = externalColRef(v.Name, s.RefTable.Name)
+			if s.Table.Schema == s.RefTable.Schema {
+				ref = externalColRef(v.Name, s.RefTable.Name)
+			} else {
+				ref = crossSchemaColRef(v.Name, s.RefTable.Name, s.RefTable.Schema.Name)
+			}
 		}
 		r = append(r, ref)
 	}
@@ -631,6 +635,10 @@ func externalColRef(cName string, tName string) *schemahcl.Ref {
 	return &schemahcl.Ref{V: "$table." + tName + ".$column." + cName}
 }
 
+func crossSchemaColRef(cName, tName, sName string) *schemahcl.Ref {
+	return &schemahcl.Ref{V: "$table." + sName + "." + tName + ".$column." + cName}
+}
+
 // SchemaRef returns the schemahcl.Ref to the schema with the given name.
 func SchemaRef(n string) *schemahcl.Ref {
 	return &schemahcl.Ref{V: "$schema." + n}
diff --git a/sql/internal/specutil/spec.go b/sql/internal/specutil/spec.go
index 35734225..86ef797e 100644
--- a/sql/internal/specutil/spec.go
+++ b/sql/internal/specutil/spec.go
@@ -113,6 +113,9 @@ func Marshal(v interface{}, marshaler schemahcl.Marshaler, schemaSpec func(schem
 			d.Tables = append(d.Tables, tables...)
 			d.Schemas = append(d.Schemas, spec)
 		}
+		if err := QualifyReferencedTables(d.Tables, s); err != nil {
+			return nil, err
+		}
 	default:
 		return nil, fmt.Errorf("specutil: failed marshaling spec. %T is not supported", v)
 	}
@@ -144,6 +147,31 @@ func QualifyDuplicates(tableSpecs []*sqlspec.Table) error {
 	return nil
 }
 
+// QualifyReferencedTables sets the Qualified field equal to the schema name in any tables
+// that are references across schema boundaries from any foreign key.
+func QualifyReferencedTables(tableSpecs []*sqlspec.Table, realm *schema.Realm) error {
+	// Iterate over all tables in the realm. If any of the tables has a foreign key referencing a table in another
+	// schema, the targeted table needs to have a qualifier set.
+	for _, s := range realm.Schemas {
+		for _, t := range s.Tables {
+			for _, f := range t.ForeignKeys {
+				if f.Table.Schema != f.RefTable.Schema {
+					for _, tbl := range tableSpecs {
+						n, err := SchemaName(tbl.Schema)
+						if err != nil {
+							return err
+						}
+						if f.RefTable.Name == tbl.Name && f.RefTable.Schema.Name == n && tbl.Qualifier == "" {
+							tbl.Qualifier = n
+						}
+					}
+				}
+			}
+		}
+	}
+	return nil
+}
+
 // HCLBytesFunc returns a helper that evaluates an HCL document from a byte slice instead
 // of from an hclparse.Parser instance.
 func HCLBytesFunc(ev schemahcl.Evaluator) func(b []byte, v interface{}, inp map[string]string) error {
diff --git a/sql/internal/sqlx/plan.go b/sql/internal/sqlx/plan.go
index 44cd4030..a6a96b4b 100644
--- a/sql/internal/sqlx/plan.go
+++ b/sql/internal/sqlx/plan.go
@@ -17,12 +17,12 @@ import (
 
 type execPlanner interface {
 	ExecContext(context.Context, string, ...interface{}) (sql.Result, error)
-	PlanChanges(context.Context, string, []schema.Change) (*migrate.Plan, error)
+	PlanChanges(context.Context, string, []schema.Change, ...migrate.PlanOption) (*migrate.Plan, error)
 }
 
 // ApplyChanges is a helper used by the different drivers to apply changes.
-func ApplyChanges(ctx context.Context, changes []schema.Change, p execPlanner) error {
-	plan, err := p.PlanChanges(ctx, "apply", changes)
+func ApplyChanges(ctx context.Context, changes []schema.Change, p execPlanner, opts ...migrate.PlanOption) error {
+	plan, err := p.PlanChanges(ctx, "apply", changes, opts...)
 	if err != nil {
 		return err
 	}
@@ -248,3 +248,37 @@ func isDropped(changes []schema.Change, t *schema.Table) bool {
 	}
 	return false
 }
+
+// CheckChangesScope checks that changes can be applied
+// on a schema scope (connection).
+func CheckChangesScope(changes []schema.Change) error {
+	names := make(map[string]struct{})
+	for _, c := range changes {
+		var t *schema.Table
+		switch c := c.(type) {
+		case *schema.AddSchema, *schema.ModifySchema, *schema.DropSchema:
+			return fmt.Errorf("%T is not allowed when migration plan is scoped to one schema", c)
+		case *schema.AddTable:
+			t = c.T
+		case *schema.ModifyTable:
+			t = c.T
+		case *schema.DropTable:
+			t = c.T
+		default:
+			continue
+		}
+		if t.Schema != nil && t.Schema.Name != "" {
+			names[t.Schema.Name] = struct{}{}
+		}
+		for _, c := range t.Columns {
+			e, ok := c.Type.Type.(*schema.EnumType)
+			if ok && e.Schema != nil && e.Schema.Name != "" {
+				names[t.Schema.Name] = struct{}{}
+			}
+		}
+	}
+	if len(names) > 1 {
+		return fmt.Errorf("found %d schemas when migration plan is scoped to one", len(names))
+	}
+	return nil
+}
diff --git a/sql/internal/sqlx/sqlx.go b/sql/internal/sqlx/sqlx.go
index 804b1129..45e9fad3 100644
--- a/sql/internal/sqlx/sqlx.go
+++ b/sql/internal/sqlx/sqlx.go
@@ -214,7 +214,8 @@ func ModeInspectRealm(o *schema.InspectRealmOption) schema.InspectMode {
 // A Builder provides a syntactic sugar API for writing SQL statements.
 type Builder struct {
 	bytes.Buffer
-	QuoteChar byte
+	QuoteChar byte    // quoting identifiers
+	Schema    *string // schema qualifier
 }
 
 // P writes a list of phrases to the builder separated and
@@ -249,7 +250,16 @@ func (b *Builder) Ident(s string) *Builder {
 // Table writes the table identifier to the builder, prefixed
 // with the schema name if exists.
 func (b *Builder) Table(t *schema.Table) *Builder {
-	if t.Schema != nil {
+	switch {
+	// Custom qualifier.
+	case b.Schema != nil:
+		// Empty means skip prefix.
+		if *b.Schema != "" {
+			b.Ident(*b.Schema)
+			b.rewriteLastByte('.')
+		}
+	// Default schema qualifier.
+	case t.Schema != nil && t.Schema.Name != "":
 		b.Ident(t.Schema.Name)
 		b.rewriteLastByte('.')
 	}
@@ -339,7 +349,7 @@ func (b *Builder) rewriteLastByte(c byte) {
 	buf[len(buf)-1] = c
 }
 
-// IsQuoted reports if the given string is quoted with one of the given quotes (e.g. '\'', '"', '`').
+// IsQuoted reports if the given string is quoted with one of the given quotes (e.g. ', ", `).
 func IsQuoted(s string, q ...byte) bool {
 	for i := range q {
 		if l, r := strings.IndexByte(s, q[i]), strings.LastIndexByte(s, q[i]); l < r && l == 0 && r == len(s)-1 {
diff --git a/sql/migrate/migrate.go b/sql/migrate/migrate.go
index e508f804..241e7318 100644
--- a/sql/migrate/migrate.go
+++ b/sql/migrate/migrate.go
@@ -73,14 +73,24 @@ type (
 	// on the database.
 	PlanApplier interface {
 		// PlanChanges returns a migration plan for applying the given changeset.
-		PlanChanges(context.Context, string, []schema.Change) (*Plan, error)
+		PlanChanges(context.Context, string, []schema.Change, ...PlanOption) (*Plan, error)
 
 		// ApplyChanges is responsible for applying the given changeset.
 		// An error may return from ApplyChanges if the driver is unable
 		// to execute a change.
-		ApplyChanges(context.Context, []schema.Change) error
+		ApplyChanges(context.Context, []schema.Change, ...PlanOption) error
 	}
 
+	// PlanOptions holds the migration plan options to be used by PlanApplier.
+	PlanOptions struct {
+		// PlanWithSchemaQualifier allows setting a custom schema to prefix
+		// tables and other resources. An empty string indicates no qualifier.
+		SchemaQualifier *string
+	}
+
+	// PlanOption allows configuring a drivers' plan using functional arguments.
+	PlanOption func(*PlanOptions)
+
 	// StateReader wraps the method for reading a database/schema state.
 	// The types below provides a few builtin options for reading a state
 	// from a migration directory, a static object (e.g. a parsed file).
@@ -142,11 +152,11 @@ type (
 	// Planner can plan the steps to take to migrate from one state to another. It uses the enclosed Dir to write
 	// those changes to versioned migration files.
 	Planner struct {
-		drv    Driver    // driver to use
-		dir    Dir       // where migration files are stored and read from
-		fmt    Formatter // how to format a plan to migration files
-		schema string    // optional schema to use for inspection
-		sum    bool      // whether to create a sum file for the migration directory
+		drv  Driver       // driver to use
+		dir  Dir          // where migration files are stored and read from
+		fmt  Formatter    // how to format a plan to migration files
+		sum  bool         // whether to create a sum file for the migration directory
+		opts []PlanOption // driver options
 	}
 
 	// PlannerOption allows managing a Planner using functional arguments.
@@ -248,11 +258,16 @@ func NewPlanner(drv Driver, dir Dir, opts ...PlannerOption) *Planner {
 	return p
 }
 
-// PlanSchema allows setting a schema name to inspect
-// after replaying the migration directory.
-func PlanSchema(name string) PlannerOption {
+// PlanWithSchemaQualifier allows setting a custom schema to prefix tables and
+// other resources. An empty string indicates no prefix.
+//
+// Note, this options require the changes to be scoped to one
+// schema and returns an error otherwise.
+func PlanWithSchemaQualifier(q string) PlannerOption {
 	return func(p *Planner) {
-		p.schema = name
+		p.opts = append(p.opts, func(o *PlanOptions) {
+			o.SchemaQualifier = &q
+		})
 	}
 }
 
@@ -263,10 +278,11 @@ func PlanFormat(fmt Formatter) PlannerOption {
 	}
 }
 
-// PlanWithoutChecksum disables the hash-sum functionality for the migration directory.
-func PlanWithoutChecksum() PlannerOption {
+// PlanWithChecksum allows setting if the hash-sum functionality
+// for the migration directory is enabled or not.
+func PlanWithChecksum(b bool) PlannerOption {
 	return func(p *Planner) {
-		p.sum = false
+		p.sum = b
 	}
 }
 
@@ -274,26 +290,36 @@ var (
 	// WithFormatter calls PlanFormat.
 	// Deprecated: use PlanFormat instead.
 	WithFormatter = PlanFormat
-	// DisableChecksum calls PlanWithoutChecksum.
+	// DisableChecksum calls PlanWithChecksum(false).
 	// Deprecated: use PlanWithoutChecksum instead.
-	DisableChecksum = PlanWithoutChecksum
+	DisableChecksum = func() PlannerOption { return PlanWithChecksum(false) }
 )
 
 // Plan calculates the migration Plan required for moving the current state (from) state to
 // the next state (to). A StateReader can be a directory, static schema elements or a Driver connection.
 func (p *Planner) Plan(ctx context.Context, name string, to StateReader) (*Plan, error) {
-	var from StateReader
-	if sr, ok := p.dir.(StateReader); ok {
-		from = sr
+	return p.plan(ctx, name, to, true)
+}
+
+// PlanSchema is like Plan but limits its scope to the schema connection.
+// Note, the operation fails in case the connection was not set to a schema.
+func (p *Planner) PlanSchema(ctx context.Context, name string, to StateReader) (*Plan, error) {
+	return p.plan(ctx, name, to, false)
+}
+
+func (p *Planner) plan(ctx context.Context, name string, to StateReader, realmScope bool) (*Plan, error) {
+	from, err := NewExecutor(p.drv, p.dir, NopRevisionReadWriter{})
+	if err != nil {
+		return nil, err
 	}
-	if from == nil {
-		ex, err := NewExecutor(p.drv, p.dir, NopRevisionReadWriter{})
-		if err != nil {
-			return nil, err
+	current, err := from.Replay(ctx, func() StateReader {
+		if realmScope {
+			return RealmConn(p.drv, nil)
 		}
-		from = ex
-	}
-	current, err := from.ReadState(ctx)
+		// In case the scope is the schema connection,
+		// inspect it and return its connected realm.
+		return SchemaConn(p.drv, "", nil)
+	}())
 	if err != nil {
 		return nil, err
 	}
@@ -303,19 +329,27 @@ func (p *Planner) Plan(ctx context.Context, name string, to StateReader) (*Plan,
 	}
 	var changes []schema.Change
 	switch {
-	case p.schema != "":
-		switch s, ok := current.Schema(p.schema); {
-		case !ok:
-			return nil, fmt.Errorf("missing schema %q in realm after replaying migration directory", p.schema)
-		case len(desired.Schemas) == 0:
-			return nil, errors.New("missing schema definition in desired state")
-		case len(desired.Schemas) > 1:
-			return nil, fmt.Errorf("found %d schema definitions in desired state on schema mode", len(desired.Schemas))
+	case realmScope:
+		changes, err = p.drv.RealmDiff(current, desired)
+	default:
+		switch n, m := len(current.Schemas), len(desired.Schemas); {
+		case n == 0:
+			return nil, errors.New("no schema was found in current state after replaying migration directory")
+		case n > 1:
+			return nil, fmt.Errorf("%d schemas were found in current state after replaying migration directory", len(current.Schemas))
+		case m == 0:
+			return nil, errors.New("no schema was found in desired state")
+		case m > 1:
+			return nil, fmt.Errorf("%d schemas were found in desired state; expect 1", len(desired.Schemas))
 		default:
-			changes, err = p.drv.SchemaDiff(s, desired.Schemas[0])
+			s1, s2 := *current.Schemas[0], *desired.Schemas[0]
+			// Avoid comparing schema names when scope is limited to one schema,
+			// and the schema qualifier is controlled by the caller.
+			if s1.Name != s2.Name {
+				s1.Name = s2.Name
+			}
+			changes, err = p.drv.SchemaDiff(&s1, &s2)
 		}
-	default:
-		changes, err = p.drv.RealmDiff(current, desired)
 	}
 	if err != nil {
 		return nil, err
@@ -323,7 +357,7 @@ func (p *Planner) Plan(ctx context.Context, name string, to StateReader) (*Plan,
 	if len(changes) == 0 {
 		return nil, ErrNoPlan
 	}
-	return p.drv.PlanChanges(ctx, name, changes)
+	return p.drv.PlanChanges(ctx, name, changes, p.opts...)
 }
 
 // WritePlan writes the given Plan to the Dir based on the configured Formatter.
@@ -353,8 +387,6 @@ func (p *Planner) WritePlan(plan *Plan) error {
 var (
 	// ErrNoPendingFiles is returned if there are no pending migration files to execute on the managed database.
 	ErrNoPendingFiles = errors.New("sql/migrate: execute: nothing to do")
-	// ErrLockUnsupported is returned if there is no schema.Locker given.
-	ErrLockUnsupported = errors.New("sql/migrate: driver does not support locking")
 	// ErrSnapshotUnsupported is returned if there is no Snapshoter given.
 	ErrSnapshotUnsupported = errors.New("sql/migrate: driver does not support taking a database snapshot")
 	// ErrCleanCheckerUnsupported is returned if there is no CleanChecker given.
@@ -383,9 +415,6 @@ func NewExecutor(drv Driver, dir Dir, rrw RevisionReadWriter, opts ...ExecutorOp
 	if ex.log == nil {
 		ex.log = NopLogger{}
 	}
-	if _, ok := drv.(schema.Locker); !ok {
-		return nil, ErrLockUnsupported
-	}
 	if _, ok := drv.(Snapshoter); !ok {
 		return nil, ErrSnapshotUnsupported
 	}
@@ -433,16 +462,6 @@ func WithAllowDirty(b bool) ExecutorOption {
 	}
 }
 
-// Lock acquires a lock for the executor.
-// It is considered a user error to not call Lock before the Pending and Execute methods.
-func (e *Executor) Lock(ctx context.Context) (schema.UnlockFunc, error) {
-	unlock, err := e.drv.(schema.Locker).Lock(ctx, "atlas_migration_execute", 0)
-	if err != nil {
-		return nil, fmt.Errorf("sql/migrate: acquiring database lock: %w", err)
-	}
-	return unlock, nil
-}
-
 // Pending returns all pending (not fully applied) migration files in the migration directory.
 func (e *Executor) Pending(ctx context.Context) ([]File, error) {
 	// Don't operate with a broken migration directory.
@@ -628,15 +647,6 @@ func (e HistoryChangedError) Error() string {
 // ExecuteN executes n pending migration files. If n<=0 all pending migration files are executed. It will not attempt
 // an execution if the database is not "clean" (has only successfully applied migrations).
 func (e *Executor) ExecuteN(ctx context.Context, n int) (err error) {
-	unlock, err := e.Lock(ctx)
-	if err != nil {
-		return err
-	}
-	defer func() {
-		if err2 := unlock(); err2 != nil {
-			err = wrap(err2, err)
-		}
-	}()
 	pending, err := e.Pending(ctx)
 	if err != nil {
 		return err
@@ -663,19 +673,8 @@ func (e *Executor) ExecuteN(ctx context.Context, n int) (err error) {
 	return err
 }
 
-// ReadState implements the StateReader interface.
-//
-// It does so by calling Execute and inspecting the database thereafter.
-func (e *Executor) ReadState(ctx context.Context) (realm *schema.Realm, err error) {
-	unlock, err := e.drv.(schema.Locker).Lock(ctx, "atlas_migration_directory_state", 0)
-	if err != nil {
-		return nil, fmt.Errorf("sql/migrate: acquiring database lock: %w", err)
-	}
-	defer func() {
-		if err2 := unlock(); err2 != nil {
-			err = wrap(err2, err)
-		}
-	}()
+// Replay the migration directory and invoke the state to get back the inspection result.
+func (e *Executor) Replay(ctx context.Context, r StateReader) (_ *schema.Realm, err error) {
 	// Clean up after ourselves.
 	restore, err := e.drv.(Snapshoter).Snapshot(ctx)
 	if err != nil {
@@ -690,9 +689,7 @@ func (e *Executor) ReadState(ctx context.Context) (realm *schema.Realm, err erro
 	if err := e.ExecuteN(ctx, 0); err != nil && !errors.Is(err, ErrNoPendingFiles) {
 		return nil, fmt.Errorf("sql/migrate: read migration directory state: %w", err)
 	}
-	// Inspect the database back and return the result.
-	realm, err = e.drv.InspectRealm(ctx, nil)
-	return
+	return r.ReadState(ctx)
 }
 
 type (
diff --git a/sql/mysql/inspect.go b/sql/mysql/inspect.go
index 8c2f4436..b1df854e 100644
--- a/sql/mysql/inspect.go
+++ b/sql/mysql/inspect.go
@@ -553,7 +553,8 @@ func (i *inspect) setAutoInc(s *showTable, t *schema.Table) error {
 // createStmt loads the CREATE TABLE statement for the table.
 func (i *inspect) createStmt(ctx context.Context, t *schema.Table) error {
 	c := &CreateStmt{}
-	rows, err := i.QueryContext(ctx, Build("SHOW CREATE TABLE").Table(t).String())
+	b := &sqlx.Builder{QuoteChar: '`'}
+	rows, err := i.QueryContext(ctx, b.P("SHOW CREATE TABLE").Table(t).String())
 	if err != nil {
 		return fmt.Errorf("query CREATE TABLE %q: %w", t.Name, err)
 	}
diff --git a/sql/mysql/migrate.go b/sql/mysql/migrate.go
index 3a94cfe0..80ad48ca 100644
--- a/sql/mysql/migrate.go
+++ b/sql/mysql/migrate.go
@@ -20,7 +20,7 @@ import (
 type planApply struct{ conn }
 
 // PlanChanges returns a migration plan for the given schema changes.
-func (p *planApply) PlanChanges(_ context.Context, name string, changes []schema.Change) (*migrate.Plan, error) {
+func (p *planApply) PlanChanges(_ context.Context, name string, changes []schema.Change, opts ...migrate.PlanOption) (*migrate.Plan, error) {
 	s := &state{
 		conn: p.conn,
 		Plan: migrate.Plan{
@@ -33,6 +33,9 @@ func (p *planApply) PlanChanges(_ context.Context, name string, changes []schema
 			Transactional: false,
 		},
 	}
+	for _, o := range opts {
+		o(&s.PlanOptions)
+	}
 	if err := s.plan(changes); err != nil {
 		return nil, err
 	}
@@ -48,8 +51,8 @@ func (p *planApply) PlanChanges(_ context.Context, name string, changes []schema
 // ApplyChanges applies the changes on the database. An error is returned
 // if the driver is unable to produce a plan to it, or one of the statements
 // is failed or unsupported.
-func (p *planApply) ApplyChanges(ctx context.Context, changes []schema.Change) error {
-	return sqlx.ApplyChanges(ctx, changes, p)
+func (p *planApply) ApplyChanges(ctx context.Context, changes []schema.Change, opts ...migrate.PlanOption) error {
+	return sqlx.ApplyChanges(ctx, changes, p, opts...)
 }
 
 // state represents the state of a planning. It is not part of
@@ -58,11 +61,17 @@ func (p *planApply) ApplyChanges(ctx context.Context, changes []schema.Change) e
 type state struct {
 	conn
 	migrate.Plan
+	migrate.PlanOptions
 }
 
 // plan builds the migration plan for applying the
 // given changes on the attached connection.
 func (s *state) plan(changes []schema.Change) error {
+	if s.SchemaQualifier != nil {
+		if err := sqlx.CheckChangesScope(changes); err != nil {
+			return err
+		}
+	}
 	planned, err := s.topLevel(changes)
 	if err != nil {
 		return err
@@ -97,7 +106,7 @@ func (s *state) topLevel(changes []schema.Change) ([]schema.Change, error) {
 	for _, c := range changes {
 		switch c := c.(type) {
 		case *schema.AddSchema:
-			b := Build("CREATE DATABASE")
+			b := s.Build("CREATE DATABASE")
 			if sqlx.Has(c.Extra, &schema.IfNotExists{}) {
 				b.P("IF NOT EXISTS")
 			}
@@ -113,11 +122,11 @@ func (s *state) topLevel(changes []schema.Change) ([]schema.Change, error) {
 			s.append(&migrate.Change{
 				Cmd:     b.String(),
 				Source:  c,
-				Reverse: Build("DROP DATABASE").Ident(c.S.Name).String(),
+				Reverse: s.Build("DROP DATABASE").Ident(c.S.Name).String(),
 				Comment: fmt.Sprintf("add new schema named %q", c.S.Name),
 			})
 		case *schema.DropSchema:
-			b := Build("DROP DATABASE")
+			b := s.Build("DROP DATABASE")
 			if sqlx.Has(c.Extra, &schema.IfExists{}) {
 				b.P("IF EXISTS")
 			}
@@ -141,7 +150,7 @@ func (s *state) topLevel(changes []schema.Change) ([]schema.Change, error) {
 // modifySchema builds and appends the migrate.Changes for bringing
 // the schema into its modified state.
 func (s *state) modifySchema(modify *schema.ModifySchema) error {
-	b, r := Build(""), Build("")
+	b, r := s.Build(), s.Build()
 	for _, change := range modify.Changes {
 		switch change := change.(type) {
 		// Add schema attributes to an existing schema only if
@@ -185,7 +194,7 @@ func (s *state) modifySchema(modify *schema.ModifySchema) error {
 		}
 	}
 	if b.Len() > 0 {
-		bs := Build("ALTER DATABASE").Ident(modify.S.Name)
+		bs := s.Build("ALTER DATABASE").Ident(modify.S.Name)
 		rs := bs.Clone()
 		bs.WriteString(b.String())
 		rs.WriteString(r.String())
@@ -204,7 +213,7 @@ func (s *state) modifySchema(modify *schema.ModifySchema) error {
 func (s *state) addTable(add *schema.AddTable) error {
 	var (
 		errs []string
-		b    = Build("CREATE TABLE")
+		b    = s.Build("CREATE TABLE")
 	)
 	if sqlx.Has(add.Extra, &schema.IfNotExists{}) {
 		b.P("IF NOT EXISTS")
@@ -250,7 +259,7 @@ func (s *state) addTable(add *schema.AddTable) error {
 	s.append(&migrate.Change{
 		Cmd:     b.String(),
 		Source:  add,
-		Reverse: Build("DROP TABLE").Table(add.T).String(),
+		Reverse: s.Build("DROP TABLE").Table(add.T).String(),
 		Comment: fmt.Sprintf("create %q table", add.T.Name),
 	})
 	return nil
@@ -259,7 +268,7 @@ func (s *state) addTable(add *schema.AddTable) error {
 // dropTable builds and appends the migrate.Change
 // for dropping a table from a schema.
 func (s *state) dropTable(drop *schema.DropTable) {
-	b := Build("DROP TABLE")
+	b := s.Build("DROP TABLE")
 	if sqlx.Has(drop.Extra, &schema.IfExists{}) {
 		b.P("IF EXISTS")
 	}
@@ -332,7 +341,7 @@ func (s *state) alterTable(t *schema.Table, changes []schema.Change) error {
 		reversible = true
 	)
 	build := func(changes []schema.Change) (string, error) {
-		b := Build("ALTER TABLE").Table(t)
+		b := s.Build("ALTER TABLE").Table(t)
 		err := b.MapCommaErr(changes, func(i int, b *sqlx.Builder) error {
 			switch change := changes[i].(type) {
 			case *schema.AddColumn:
@@ -465,8 +474,8 @@ func (s *state) renameTable(c *schema.RenameTable) {
 	s.append(&migrate.Change{
 		Source:  c,
 		Comment: fmt.Sprintf("rename a table from %q to %q", c.From.Name, c.To.Name),
-		Cmd:     Build("RENAME TABLE").Table(c.From).P("TO").Table(c.To).String(),
-		Reverse: Build("RENAME TABLE").Table(c.To).P("TO").Table(c.From).String(),
+		Cmd:     s.Build("RENAME TABLE").Table(c.From).P("TO").Table(c.To).String(),
+		Reverse: s.Build("RENAME TABLE").Table(c.To).P("TO").Table(c.From).String(),
 	})
 }
 
@@ -699,9 +708,9 @@ func (s *state) columnDefault(b *sqlx.Builder, c *schema.Column) {
 }
 
 // Build instantiates a new builder and writes the given phrase to it.
-func Build(phrase string) *sqlx.Builder {
-	b := &sqlx.Builder{QuoteChar: '`'}
-	return b.P(phrase)
+func (s *state) Build(phrases ...string) *sqlx.Builder {
+	b := &sqlx.Builder{QuoteChar: '`', Schema: s.SchemaQualifier}
+	return b.P(phrases...)
 }
 
 // skipAutoChanges filters unnecessary changes that are automatically
diff --git a/sql/mysql/tidb.go b/sql/mysql/tidb.go
index b261e35f..6a3e328f 100644
--- a/sql/mysql/tidb.go
+++ b/sql/mysql/tidb.go
@@ -81,7 +81,7 @@ func flat(changes []schema.Change) []schema.Change {
 }
 
 // PlanChanges returns a migration plan for the given schema changes.
-func (p *tplanApply) PlanChanges(ctx context.Context, name string, changes []schema.Change) (*migrate.Plan, error) {
+func (p *tplanApply) PlanChanges(ctx context.Context, name string, changes []schema.Change, opts ...migrate.PlanOption) (*migrate.Plan, error) {
 	fc := flat(changes)
 	sort.SliceStable(fc, func(i, j int) bool {
 		return priority(fc[i]) < priority(fc[j])
@@ -98,7 +98,7 @@ func (p *tplanApply) PlanChanges(ctx context.Context, name string, changes []sch
 	}
 	for _, c := range fc {
 		// Use the planner of MySQL with each "atomic" change.
-		plan, err := p.planApply.PlanChanges(ctx, name, []schema.Change{c})
+		plan, err := p.planApply.PlanChanges(ctx, name, []schema.Change{c}, opts...)
 		if err != nil {
 			return nil, err
 		}
@@ -110,20 +110,8 @@ func (p *tplanApply) PlanChanges(ctx context.Context, name string, changes []sch
 	return &s.Plan, nil
 }
 
-func (p *tplanApply) ApplyChanges(ctx context.Context, changes []schema.Change) error {
-	plan, err := p.PlanChanges(ctx, "apply", changes)
-	if err != nil {
-		return err
-	}
-	for _, c := range plan.Changes {
-		if _, err := p.ExecContext(ctx, c.Cmd, c.Args...); err != nil {
-			if c.Comment != "" {
-				err = fmt.Errorf("%s: %w", c.Comment, err)
-			}
-			return err
-		}
-	}
-	return nil
+func (p *tplanApply) ApplyChanges(ctx context.Context, changes []schema.Change, opts ...migrate.PlanOption) error {
+	return sqlx.ApplyChanges(ctx, changes, p, opts...)
 }
 
 func (i *tinspect) InspectSchema(ctx context.Context, name string, opts *schema.InspectOptions) (*schema.Schema, error) {
diff --git a/sql/postgres/diff.go b/sql/postgres/diff.go
index 2d8418fd..5b3e541d 100644
--- a/sql/postgres/diff.go
+++ b/sql/postgres/diff.go
@@ -311,7 +311,8 @@ func identity(attrs []schema.Attr) (*Identity, bool) {
 // formatPartition returns the string representation of the
 // partition key according to the PostgreSQL format/grammar.
 func formatPartition(p Partition) (string, error) {
-	b := Build("PARTITION BY")
+	b := &sqlx.Builder{QuoteChar: '"'}
+	b.P("PARTITION BY")
 	switch t := strings.ToUpper(p.T); t {
 	case PartitionTypeRange, PartitionTypeList, PartitionTypeHash:
 		b.P(t)
diff --git a/sql/postgres/migrate.go b/sql/postgres/migrate.go
index 8af779a8..4056fe1c 100644
--- a/sql/postgres/migrate.go
+++ b/sql/postgres/migrate.go
@@ -20,7 +20,7 @@ import (
 type planApply struct{ conn }
 
 // PlanChanges returns a migration plan for the given schema changes.
-func (p *planApply) PlanChanges(ctx context.Context, name string, changes []schema.Change) (*migrate.Plan, error) {
+func (p *planApply) PlanChanges(ctx context.Context, name string, changes []schema.Change, opts ...migrate.PlanOption) (*migrate.Plan, error) {
 	s := &state{
 		conn: p.conn,
 		Plan: migrate.Plan{
@@ -32,6 +32,9 @@ func (p *planApply) PlanChanges(ctx context.Context, name string, changes []sche
 		altered: make(map[string]*schema.EnumType),
 		dropped: make(map[string]*schema.EnumType),
 	}
+	for _, o := range opts {
+		o(&s.PlanOptions)
+	}
 	if err := s.plan(ctx, changes); err != nil {
 		return nil, err
 	}
@@ -46,8 +49,8 @@ func (p *planApply) PlanChanges(ctx context.Context, name string, changes []sche
 // ApplyChanges applies the changes on the database. An error is returned
 // if the driver is unable to produce a plan to do so, or one of the statements
 // is failed or unsupported.
-func (p *planApply) ApplyChanges(ctx context.Context, changes []schema.Change) error {
-	return sqlx.ApplyChanges(ctx, changes, p)
+func (p *planApply) ApplyChanges(ctx context.Context, changes []schema.Change, opts ...migrate.PlanOption) error {
+	return sqlx.ApplyChanges(ctx, changes, p, opts...)
 }
 
 // state represents the state of a planning. It is not part of
@@ -56,6 +59,7 @@ func (p *planApply) ApplyChanges(ctx context.Context, changes []schema.Change) e
 type state struct {
 	conn
 	migrate.Plan
+	migrate.PlanOptions
 	// Track the enums that were created, altered and
 	// dropped, in this phase to avoid duplicate updates.
 	created, altered, dropped map[string]*schema.EnumType
@@ -64,6 +68,11 @@ type state struct {
 // Exec executes the changes on the database. An error is returned
 // if one of the operations fail, or a change is not supported.
 func (s *state) plan(ctx context.Context, changes []schema.Change) error {
+	if s.SchemaQualifier != nil {
+		if err := sqlx.CheckChangesScope(changes); err != nil {
+			return err
+		}
+	}
 	planned := s.topLevel(changes)
 	planned, err := sqlx.DetachCycles(planned)
 	if err != nil {
@@ -95,7 +104,7 @@ func (s *state) topLevel(changes []schema.Change) []schema.Change {
 	for _, c := range changes {
 		switch c := c.(type) {
 		case *schema.AddSchema:
-			b := Build("CREATE SCHEMA")
+			b := s.Build("CREATE SCHEMA")
 			if sqlx.Has(c.Extra, &schema.IfNotExists{}) {
 				b.P("IF NOT EXISTS")
 			}
@@ -103,11 +112,11 @@ func (s *state) topLevel(changes []schema.Change) []schema.Change {
 			s.append(&migrate.Change{
 				Cmd:     b.String(),
 				Source:  c,
-				Reverse: Build("DROP SCHEMA").Ident(c.S.Name).P("CASCADE").String(),
+				Reverse: s.Build("DROP SCHEMA").Ident(c.S.Name).P("CASCADE").String(),
 				Comment: fmt.Sprintf("Add new schema named %q", c.S.Name),
 			})
 		case *schema.DropSchema:
-			b := Build("DROP SCHEMA")
+			b := s.Build("DROP SCHEMA")
 			if sqlx.Has(c.Extra, &schema.IfExists{}) {
 				b.P("IF EXISTS")
 			}
@@ -132,7 +141,7 @@ func (s *state) addTable(ctx context.Context, add *schema.AddTable) error {
 	}
 	var (
 		errs []string
-		b    = Build("CREATE TABLE")
+		b    = s.Build("CREATE TABLE")
 	)
 	if sqlx.Has(add.Extra, &schema.IfNotExists{}) {
 		b.P("IF NOT EXISTS")
@@ -173,7 +182,7 @@ func (s *state) addTable(ctx context.Context, add *schema.AddTable) error {
 		Cmd:     b.String(),
 		Source:  add,
 		Comment: fmt.Sprintf("create %q table", add.T.Name),
-		Reverse: Build("DROP TABLE").Table(add.T).String(),
+		Reverse: s.Build("DROP TABLE").Table(add.T).String(),
 	})
 	s.addIndexes(add.T, add.T.Indexes...)
 	s.addComments(add.T)
@@ -182,7 +191,7 @@ func (s *state) addTable(ctx context.Context, add *schema.AddTable) error {
 
 // dropTable builds and executes the query for dropping a table from a schema.
 func (s *state) dropTable(drop *schema.DropTable) {
-	b := Build("DROP TABLE")
+	b := s.Build("DROP TABLE")
 	if sqlx.Has(drop.Extra, &schema.IfExists{}) {
 		b.P("IF EXISTS")
 	}
@@ -244,8 +253,8 @@ func (s *state) modifyTable(ctx context.Context, modify *schema.ModifyTable) err
 			changes = append(changes, &migrate.Change{
 				Source:  change,
 				Comment: fmt.Sprintf("rename an index from %q to %q", change.From.Name, change.To.Name),
-				Cmd:     Build("ALTER INDEX").Ident(change.From.Name).P("RENAME TO").Ident(change.To.Name).String(),
-				Reverse: Build("ALTER INDEX").Ident(change.To.Name).P("RENAME TO").Ident(change.From.Name).String(),
+				Cmd:     s.Build("ALTER INDEX").Ident(change.From.Name).P("RENAME TO").Ident(change.To.Name).String(),
+				Reverse: s.Build("ALTER INDEX").Ident(change.To.Name).P("RENAME TO").Ident(change.From.Name).String(),
 			})
 		case *schema.ModifyForeignKey:
 			// Foreign-key modification is translated into 2 steps.
@@ -280,7 +289,7 @@ func (s *state) modifyTable(ctx context.Context, modify *schema.ModifyTable) err
 			to, ok2 := change.To.Type.Type.(*schema.EnumType)
 			switch {
 			// Enum was changed (underlying values).
-			case ok1 && ok2 && enumIdent(modify.T.Schema, from) == enumIdent(modify.T.Schema, to):
+			case ok1 && ok2 && s.enumIdent(modify.T.Schema, from) == s.enumIdent(modify.T.Schema, to):
 				if err := s.alterEnum(modify.T, from, to); err != nil {
 					return err
 				}
@@ -291,7 +300,7 @@ func (s *state) modifyTable(ctx context.Context, modify *schema.ModifyTable) err
 				}
 			// Enum was added or changed.
 			case !ok1 && ok2 ||
-				ok1 && ok2 && enumIdent(modify.T.Schema, from) != enumIdent(modify.T.Schema, to):
+				ok1 && ok2 && s.enumIdent(modify.T.Schema, from) != s.enumIdent(modify.T.Schema, to):
 				if err := s.mayAddEnums(ctx, modify.T, change.To); err != nil {
 					return err
 				}
@@ -299,7 +308,7 @@ func (s *state) modifyTable(ctx context.Context, modify *schema.ModifyTable) err
 			alter = append(alter, &schema.ModifyColumn{To: change.To, From: change.From, Change: k})
 		case *schema.RenameColumn:
 			// "RENAME COLUMN" cannot be combined with other alterations.
-			b := Build("ALTER TABLE").Table(modify.T).P("RENAME COLUMN")
+			b := s.Build("ALTER TABLE").Table(modify.T).P("RENAME COLUMN")
 			r := b.Clone()
 			changes = append(changes, &migrate.Change{
 				Source:  change,
@@ -329,7 +338,7 @@ func (s *state) alterTable(t *schema.Table, changes []schema.Change) error {
 		reversible = true
 	)
 	build := func(alter *alterChange, changes []schema.Change) (string, error) {
-		b := Build("ALTER TABLE").Table(t)
+		b := s.Build("ALTER TABLE").Table(t)
 		err := b.MapCommaErr(changes, func(i int, b *sqlx.Builder) error {
 			switch change := changes[i].(type) {
 			case *schema.AddColumn:
@@ -353,7 +362,7 @@ func (s *state) alterTable(t *schema.Table, changes []schema.Change) error {
 				toE, toHas := change.From.Type.Type.(*schema.EnumType)
 				fromE, fromHas := change.From.Type.Type.(*schema.EnumType)
 				// In case the enum was dropped or replaced with a different one.
-				if fromHas && !toHas || fromHas && toHas && enumIdent(t.Schema, fromE) != enumIdent(t.Schema, toE) {
+				if fromHas && !toHas || fromHas && toHas && s.enumIdent(t.Schema, fromE) != s.enumIdent(t.Schema, toE) {
 					if err := s.mayDropEnum(alter, t.Schema, fromE); err != nil {
 						return err
 					}
@@ -460,7 +469,7 @@ func (s *state) alterColumn(b *sqlx.Builder, alter *alterChange, t *schema.Table
 		b.P("ALTER COLUMN").Ident(c.To.Name)
 		switch {
 		case k.Is(schema.ChangeType):
-			if err := alterType(b, alter, t, c); err != nil {
+			if err := s.alterType(b, alter, t, c); err != nil {
 				return err
 			}
 			k &= ^schema.ChangeType
@@ -511,13 +520,13 @@ func (s *state) alterColumn(b *sqlx.Builder, alter *alterChange, t *schema.Table
 
 // alterType appends the clause(s) to alter the column type and assuming the
 // "ALTER COLUMN <Name>" was called before by the alterColumn function.
-func alterType(b *sqlx.Builder, alter *alterChange, t *schema.Table, c *schema.ModifyColumn) error {
+func (s *state) alterType(b *sqlx.Builder, alter *alterChange, t *schema.Table, c *schema.ModifyColumn) error {
 	// Commands for creating and dropping serial sequences.
-	createDropSeq := func(s *SerialType) (string, string, string) {
-		seq := fmt.Sprintf(`%q.%q`, t.Schema.Name, s.sequence(t, c.To))
-		drop := Build("DROP SEQUENCE IF EXISTS").P(seq).String()
-		create := Build("CREATE SEQUENCE IF NOT EXISTS").P(seq, "OWNED BY").
-			P(fmt.Sprintf(`%q.%q.%q`, t.Schema.Name, t.Name, c.To.Name)).
+	createDropSeq := func(st *SerialType) (string, string, string) {
+		seq := fmt.Sprintf(`%s%q`, s.schemaPrefix(t.Schema), st.sequence(t, c.To))
+		drop := s.Build("DROP SEQUENCE IF EXISTS").P(seq).String()
+		create := s.Build("CREATE SEQUENCE IF NOT EXISTS").P(seq, "OWNED BY").
+			P(fmt.Sprintf(`%s%q.%q`, s.schemaPrefix(t.Schema), t.Name, c.To.Name)).
 			String()
 		return create, drop, seq
 	}
@@ -585,7 +594,7 @@ func alterType(b *sqlx.Builder, alter *alterChange, t *schema.Table, c *schema.M
 			err error
 		)
 		if e, ok := c.To.Type.Type.(*schema.EnumType); ok {
-			f = enumIdent(t.Schema, e)
+			f = s.enumIdent(t.Schema, e)
 		} else if f, err = FormatType(c.To.Type.Type); err != nil {
 			return err
 		}
@@ -601,8 +610,8 @@ func (s *state) renameTable(c *schema.RenameTable) {
 	s.append(&migrate.Change{
 		Source:  c,
 		Comment: fmt.Sprintf("rename a table from %q to %q", c.From.Name, c.To.Name),
-		Cmd:     Build("ALTER TABLE").Table(c.From).P("RENAME TO").Table(c.To).String(),
-		Reverse: Build("ALTER TABLE").Table(c.To).P("RENAME TO").Table(c.From).String(),
+		Cmd:     s.Build("ALTER TABLE").Table(c.From).P("RENAME TO").Table(c.To).String(),
+		Reverse: s.Build("ALTER TABLE").Table(c.To).P("RENAME TO").Table(c.From).String(),
 	})
 }
 
@@ -623,8 +632,8 @@ func (s *state) addComments(t *schema.Table) {
 	}
 }
 
-func (*state) tableComment(t *schema.Table, to, from string) *migrate.Change {
-	b := Build("COMMENT ON TABLE").Table(t).P("IS")
+func (s *state) tableComment(t *schema.Table, to, from string) *migrate.Change {
+	b := s.Build("COMMENT ON TABLE").Table(t).P("IS")
 	return &migrate.Change{
 		Cmd:     b.Clone().P(quote(to)).String(),
 		Comment: fmt.Sprintf("set comment to table: %q", t.Name),
@@ -632,8 +641,8 @@ func (*state) tableComment(t *schema.Table, to, from string) *migrate.Change {
 	}
 }
 
-func (*state) columnComment(t *schema.Table, c *schema.Column, to, from string) *migrate.Change {
-	b := Build("COMMENT ON COLUMN").Table(t)
+func (s *state) columnComment(t *schema.Table, c *schema.Column, to, from string) *migrate.Change {
+	b := s.Build("COMMENT ON COLUMN").Table(t)
 	b.WriteByte('.')
 	b.Ident(c.Name).P("IS")
 	return &migrate.Change{
@@ -643,8 +652,8 @@ func (*state) columnComment(t *schema.Table, c *schema.Column, to, from string)
 	}
 }
 
-func (*state) indexComment(t *schema.Table, idx *schema.Index, to, from string) *migrate.Change {
-	b := Build("COMMENT ON INDEX").Ident(idx.Name).P("IS")
+func (s *state) indexComment(t *schema.Table, idx *schema.Index, to, from string) *migrate.Change {
+	b := s.Build("COMMENT ON INDEX").Ident(idx.Name).P("IS")
 	return &migrate.Change{
 		Cmd:     b.Clone().P(quote(to)).String(),
 		Comment: fmt.Sprintf("set comment to index: %q on table: %q", idx.Name, t.Name),
@@ -680,7 +689,7 @@ func (s *state) mayAddEnums(ctx context.Context, t *schema.Table, columns ...*sc
 			// on this migration phase.
 			continue
 		}
-		name := enumIdent(t.Schema, e)
+		name := s.enumIdent(t.Schema, e)
 		if prev, ok := s.created[name]; ok {
 			if !sqlx.ValuesEqual(prev.Values, e.Values) {
 				return fmt.Errorf("enum type %s has inconsistent desired state: %q != %q", name, prev.Values, e.Values)
@@ -688,7 +697,7 @@ func (s *state) mayAddEnums(ctx context.Context, t *schema.Table, columns ...*sc
 			continue
 		}
 		s.created[name] = e
-		create, drop := createDropEnum(t.Schema, e)
+		create, drop := s.createDropEnum(t.Schema, e)
 		s.append(&migrate.Change{
 			Cmd:     create,
 			Reverse: drop,
@@ -707,7 +716,7 @@ func (s *state) alterEnum(t *schema.Table, from, to *schema.EnumType) error {
 			return fmt.Errorf("replacing or reordering enum (%q) value is not supported: %q != %q", to.T, to.Values, from.Values)
 		}
 	}
-	name := enumIdent(t.Schema, from)
+	name := s.enumIdent(t.Schema, from)
 	if prev, ok := s.altered[name]; ok {
 		if !sqlx.ValuesEqual(prev.Values, to.Values) {
 			return fmt.Errorf("enum type %s has inconsistent desired state: %q != %q", name, prev.Values, to.Values)
@@ -717,7 +726,7 @@ func (s *state) alterEnum(t *schema.Table, from, to *schema.EnumType) error {
 	s.altered[name] = to
 	for _, v := range to.Values[len(from.Values):] {
 		s.append(&migrate.Change{
-			Cmd:     Build("ALTER TYPE").P(name, "ADD VALUE", quote(v)).String(),
+			Cmd:     s.Build("ALTER TYPE").P(name, "ADD VALUE", quote(v)).String(),
 			Comment: fmt.Sprintf("add value to enum type: %q", from.T),
 		})
 	}
@@ -725,10 +734,12 @@ func (s *state) alterEnum(t *schema.Table, from, to *schema.EnumType) error {
 }
 
 func (s *state) enumExists(ctx context.Context, ns *schema.Schema, e *schema.EnumType) (bool, error) {
-	if e.Schema != nil {
-		ns = e.Schema
+	query, args := `SELECT * FROM pg_type t JOIN pg_namespace n on t.typnamespace = n.oid WHERE t.typname = $1 AND t.typtype = 'e'`, []any{e.T}
+	if es := s.enumSchema(ns, e); es != "" {
+		query += " AND n.nspname = $2"
+		args = append(args, es)
 	}
-	rows, err := s.QueryContext(ctx, `SELECT * FROM pg_type t JOIN pg_namespace n on t.typnamespace = n.oid WHERE t.typname = $1 AND n.nspname = $2 AND t.typtype = 'e'`, e.T, ns.Name)
+	rows, err := s.QueryContext(ctx, query, args...)
 	if err != nil {
 		return false, fmt.Errorf("check enum existence: %w", err)
 	}
@@ -738,7 +749,7 @@ func (s *state) enumExists(ctx context.Context, ns *schema.Schema, e *schema.Enu
 
 // mayDropEnum drops dangling enum types form the schema.
 func (s *state) mayDropEnum(alter *alterChange, ns *schema.Schema, e *schema.EnumType) error {
-	name := enumIdent(ns, e)
+	name := s.enumIdent(ns, e)
 	if _, ok := s.dropped[name]; ok {
 		return nil
 	}
@@ -753,14 +764,14 @@ func (s *state) mayDropEnum(alter *alterChange, ns *schema.Schema, e *schema.Enu
 				e1, ok := c.Type.Type.(*schema.EnumType)
 				// Although we search in siblings schemas, use the
 				// table's one for building the enum identifier.
-				if ok && enumIdent(ns, e1) == name {
+				if ok && s.enumIdent(ns, e1) == name {
 					return nil
 				}
 			}
 		}
 	}
 	s.dropped[name] = e
-	create, drop := createDropEnum(ns, e)
+	create, drop := s.createDropEnum(ns, e)
 	alter.after = append(alter.after, &migrate.Change{
 		Cmd:     drop,
 		Reverse: create,
@@ -771,7 +782,7 @@ func (s *state) mayDropEnum(alter *alterChange, ns *schema.Schema, e *schema.Enu
 
 func (s *state) addIndexes(t *schema.Table, indexes ...*schema.Index) {
 	for _, idx := range indexes {
-		b := Build("CREATE")
+		b := s.Build("CREATE")
 		if idx.Unique {
 			b.P("UNIQUE")
 		}
@@ -785,16 +796,13 @@ func (s *state) addIndexes(t *schema.Table, indexes ...*schema.Index) {
 			Cmd:     b.String(),
 			Comment: fmt.Sprintf("create index %q to table: %q", idx.Name, t.Name),
 			Reverse: func() string {
-				b := Build("DROP INDEX")
+				b := s.Build("DROP INDEX")
 				// Unlike MySQL, the DROP command is not attached to ALTER TABLE.
 				// Therefore, we print indexes with their qualified name, because
 				// the connection that executes the statements may not be attached
 				// to this schema.
 				if t.Schema != nil {
-					b.WriteByte(b.QuoteChar)
-					b.WriteString(t.Schema.Name)
-					b.WriteByte(b.QuoteChar)
-					b.WriteByte('.')
+					b.WriteString(s.schemaPrefix(t.Schema))
 				}
 				b.Ident(idx.Name)
 				return b.String()
@@ -809,7 +817,7 @@ func (s *state) column(b *sqlx.Builder, t *schema.Table, c *schema.Column) error
 		err error
 	)
 	if e, ok := c.Type.Type.(*schema.EnumType); ok {
-		f = enumIdent(t.Schema, e)
+		f = s.enumIdent(t.Schema, e)
 	} else if f, err = FormatType(c.Type.Type); err != nil {
 		return err
 	}
@@ -977,9 +985,9 @@ func (s *state) append(c ...*migrate.Change) {
 }
 
 // Build instantiates a new builder and writes the given phrase to it.
-func Build(phrase string) *sqlx.Builder {
-	b := &sqlx.Builder{QuoteChar: '"'}
-	return b.P(phrase)
+func (s *state) Build(phrases ...string) *sqlx.Builder {
+	b := &sqlx.Builder{QuoteChar: '"', Schema: s.SchemaQualifier}
+	return b.P(phrases...)
 }
 
 // skipAutoChanges filters unnecessary changes that are automatically
@@ -1089,22 +1097,9 @@ func quote(s string) string {
 	return "'" + strings.ReplaceAll(s, "'", "''") + "'"
 }
 
-func enumIdent(s *schema.Schema, e *schema.EnumType) string {
-	switch {
-	// Enum schema has higher precedence.
-	case e.Schema != nil:
-		return fmt.Sprintf("%q.%q", e.Schema.Name, e.T)
-	// Fallback to table schema if exists.
-	case s != nil:
-		return fmt.Sprintf("%q.%q", s.Name, e.T)
-	default:
-		return strconv.Quote(e.T)
-	}
-}
-
-func createDropEnum(s *schema.Schema, e *schema.EnumType) (string, string) {
-	name := enumIdent(s, e)
-	return Build("CREATE TYPE").
+func (s *state) createDropEnum(sc *schema.Schema, e *schema.EnumType) (string, string) {
+	name := s.enumIdent(sc, e)
+	return s.Build("CREATE TYPE").
 			P(name, "AS ENUM").
 			Wrap(func(b *sqlx.Builder) {
 				b.MapComma(e.Values, func(i int, b *sqlx.Builder) {
@@ -1112,5 +1107,42 @@ func createDropEnum(s *schema.Schema, e *schema.EnumType) (string, string) {
 				})
 			}).
 			String(),
-		Build("DROP TYPE").P(name).String()
+		s.Build("DROP TYPE").P(name).String()
+}
+
+func (s *state) enumIdent(ns *schema.Schema, e *schema.EnumType) string {
+	es := s.enumSchema(ns, e)
+	if es != "" {
+		return fmt.Sprintf("%q.%q", es, e.T)
+	}
+	return strconv.Quote(e.T)
+}
+
+func (s *state) enumSchema(ns *schema.Schema, e *schema.EnumType) (es string) {
+	switch {
+	// In case the plan uses a specific schema qualifier.
+	case s.SchemaQualifier != nil:
+		es = *s.SchemaQualifier
+	// Enum schema has higher precedence.
+	case e.Schema != nil:
+		es = e.Schema.Name
+	// Fallback to table schema if exists.
+	case ns != nil:
+		es = ns.Name
+	}
+	return
+}
+
+// schemaPrefix returns the schema prefix based on the planner config.
+func (s *state) schemaPrefix(ns *schema.Schema) string {
+	switch {
+	case s.SchemaQualifier != nil:
+		// In case the qualifier is empty, ignore.
+		if *s.SchemaQualifier != "" {
+			return fmt.Sprintf("%q.", *s.SchemaQualifier)
+		}
+	case ns != nil && ns.Name != "":
+		return fmt.Sprintf("%q.", ns.Name)
+	}
+	return ""
 }
diff --git a/sql/postgres/sqlspec.go b/sql/postgres/sqlspec.go
index 5305145b..71487918 100644
--- a/sql/postgres/sqlspec.go
+++ b/sql/postgres/sqlspec.go
@@ -94,6 +94,9 @@ func MarshalSpec(v interface{}, marshaler schemahcl.Marshaler) ([]byte, error) {
 			d.Schemas = append(d.Schemas, doc.Schemas...)
 			d.Enums = append(d.Enums, doc.Enums...)
 		}
+		if err := specutil.QualifyReferencedTables(d.Tables, s); err != nil {
+			return nil, err
+		}
 	default:
 		return nil, fmt.Errorf("specutil: failed marshaling spec. %T is not supported", v)
 	}
diff --git a/sql/sqlite/migrate.go b/sql/sqlite/migrate.go
index 71e4c072..e6fbea93 100644
--- a/sql/sqlite/migrate.go
+++ b/sql/sqlite/migrate.go
@@ -18,7 +18,7 @@ import (
 type planApply struct{ conn }
 
 // PlanChanges returns a migration plan for the given schema changes.
-func (p *planApply) PlanChanges(ctx context.Context, name string, changes []schema.Change) (*migrate.Plan, error) {
+func (p *planApply) PlanChanges(ctx context.Context, name string, changes []schema.Change, opts ...migrate.PlanOption) (*migrate.Plan, error) {
 	s := &state{
 		conn: p.conn,
 		Plan: migrate.Plan{
@@ -26,6 +26,14 @@ func (p *planApply) PlanChanges(ctx context.Context, name string, changes []sche
 			Reversible:    true,
 			Transactional: true,
 		},
+		PlanOptions: migrate.PlanOptions{
+			// Currently, the driver does not support attached
+			// schemas and assumed the connected schema is "main".
+			SchemaQualifier: new(string),
+		},
+	}
+	for _, o := range opts {
+		o(&s.PlanOptions)
 	}
 	if err := s.plan(ctx, changes); err != nil {
 		return nil, err
@@ -41,8 +49,8 @@ func (p *planApply) PlanChanges(ctx context.Context, name string, changes []sche
 // ApplyChanges applies the changes on the database. An error is returned
 // if the driver is unable to produce a plan to it, or one of the statements
 // is failed or unsupported.
-func (p *planApply) ApplyChanges(ctx context.Context, changes []schema.Change) error {
-	return sqlx.ApplyChanges(ctx, changes, p)
+func (p *planApply) ApplyChanges(ctx context.Context, changes []schema.Change, opts ...migrate.PlanOption) error {
+	return sqlx.ApplyChanges(ctx, changes, p, opts...)
 }
 
 // state represents the state of a planning. It's not part of
@@ -51,6 +59,7 @@ func (p *planApply) ApplyChanges(ctx context.Context, changes []schema.Change) e
 type state struct {
 	conn
 	migrate.Plan
+	migrate.PlanOptions
 	skipFKs bool
 }
 
@@ -89,7 +98,7 @@ func (s *state) plan(ctx context.Context, changes []schema.Change) (err error) {
 func (s *state) addTable(ctx context.Context, add *schema.AddTable) error {
 	var (
 		errs []string
-		b    = Build("CREATE TABLE").Ident(add.T.Name)
+		b    = s.Build("CREATE TABLE").Table(add.T)
 	)
 	if sqlx.Has(add.Extra, &schema.IfNotExists{}) {
 		b.P("IF NOT EXISTS")
@@ -125,7 +134,7 @@ func (s *state) addTable(ctx context.Context, add *schema.AddTable) error {
 	s.append(&migrate.Change{
 		Cmd:     b.String(),
 		Source:  add,
-		Reverse: Build("DROP TABLE").Table(add.T).String(),
+		Reverse: s.Build("DROP TABLE").Table(add.T).String(),
 		Comment: fmt.Sprintf("create %q table", add.T.Name),
 	})
 	if err := s.tableSeq(ctx, add); err != nil {
@@ -137,7 +146,7 @@ func (s *state) addTable(ctx context.Context, add *schema.AddTable) error {
 // dropTable builds and executes the query for dropping a table from a schema.
 func (s *state) dropTable(drop *schema.DropTable) error {
 	s.skipFKs = true
-	b := Build("DROP TABLE").Ident(drop.T.Name)
+	b := s.Build("DROP TABLE").Ident(drop.T.Name)
 	if sqlx.Has(drop.Extra, &schema.IfExists{}) {
 		b.P("IF EXISTS")
 	}
@@ -172,7 +181,7 @@ func (s *state) modifyTable(ctx context.Context, modify *schema.ModifyTable) err
 	}
 	// Drop the current table, and rename the new one to its real name.
 	s.append(&migrate.Change{
-		Cmd:    Build("DROP TABLE").Ident(modify.T.Name).String(),
+		Cmd:    s.Build("DROP TABLE").Ident(modify.T.Name).String(),
 		Source: modify,
 		Comment: fmt.Sprintf("drop %q table %s", modify.T.Name, func() string {
 			if copied {
@@ -182,7 +191,7 @@ func (s *state) modifyTable(ctx context.Context, modify *schema.ModifyTable) err
 		}()),
 	})
 	s.append(&migrate.Change{
-		Cmd:     Build("ALTER TABLE").Ident(newT.Name).P("RENAME TO").Ident(modify.T.Name).String(),
+		Cmd:     s.Build("ALTER TABLE").Ident(newT.Name).P("RENAME TO").Ident(modify.T.Name).String(),
 		Source:  modify,
 		Comment: fmt.Sprintf("rename temporary table %q to %q", newT.Name, modify.T.Name),
 	})
@@ -193,8 +202,8 @@ func (s *state) renameTable(c *schema.RenameTable) {
 	s.append(&migrate.Change{
 		Source:  c,
 		Comment: fmt.Sprintf("rename a table from %q to %q", c.From.Name, c.To.Name),
-		Cmd:     Build("ALTER TABLE").Table(c.From).P("RENAME TO").Table(c.To).String(),
-		Reverse: Build("ALTER TABLE").Table(c.To).P("RENAME TO").Table(c.From).String(),
+		Cmd:     s.Build("ALTER TABLE").Table(c.From).P("RENAME TO").Table(c.To).String(),
+		Reverse: s.Build("ALTER TABLE").Table(c.To).P("RENAME TO").Table(c.From).String(),
 	})
 }
 
@@ -265,7 +274,7 @@ func (s *state) addIndexes(t *schema.Table, indexes ...*schema.Index) error {
 			}
 			idx.Name = strings.Join(names, "_")
 		}
-		b := Build("CREATE")
+		b := s.Build("CREATE")
 		if idx.Unique {
 			b.P("UNIQUE")
 		}
@@ -281,7 +290,7 @@ func (s *state) addIndexes(t *schema.Table, indexes ...*schema.Index) error {
 		s.append(&migrate.Change{
 			Cmd:     b.String(),
 			Source:  &schema.AddIndex{I: idx},
-			Reverse: Build("DROP INDEX").Ident(idx.Name).String(),
+			Reverse: s.Build("DROP INDEX").Ident(idx.Name).String(),
 			Comment: fmt.Sprintf("create index %q to table: %q", idx.Name, t.Name),
 		})
 	}
@@ -420,7 +429,7 @@ func (s *state) alterTable(modify *schema.ModifyTable) error {
 				return err
 			}
 		case *schema.AddColumn:
-			b := Build("ALTER TABLE").Ident(modify.T.Name)
+			b := s.Build("ALTER TABLE").Ident(modify.T.Name)
 			r := b.Clone()
 			if err := s.column(b.P("ADD COLUMN"), change.C); err != nil {
 				return err
@@ -432,7 +441,7 @@ func (s *state) alterTable(modify *schema.ModifyTable) error {
 				Comment: fmt.Sprintf("add column %q to table: %q", change.C.Name, modify.T.Name),
 			})
 		case *schema.RenameColumn:
-			b := Build("ALTER TABLE").Ident(modify.T.Name).P("RENAME COLUMN")
+			b := s.Build("ALTER TABLE").Ident(modify.T.Name).P("RENAME COLUMN")
 			r := b.Clone()
 			s.append(&migrate.Change{
 				Source:  change,
@@ -520,9 +529,9 @@ func autoincPK(pk *schema.Index) bool {
 }
 
 // Build instantiates a new builder and writes the given phrase to it.
-func Build(phrase string) *sqlx.Builder {
-	b := &sqlx.Builder{QuoteChar: '`'}
-	return b.P(phrase)
+func (s *state) Build(phrases ...string) *sqlx.Builder {
+	b := &sqlx.Builder{QuoteChar: '`', Schema: s.SchemaQualifier}
+	return b.P(phrases...)
 }
 
 func defaultValue(c *schema.Column) (string, error) {


Test patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate_test.go b/cmd/atlas/internal/cmdapi/migrate_test.go
index ff5198ed..34a9af8d 100644
--- a/cmd/atlas/internal/cmdapi/migrate_test.go
+++ b/cmd/atlas/internal/cmdapi/migrate_test.go
@@ -101,7 +101,7 @@ func TestMigrate_Apply(t *testing.T) {
 		"--url", fmt.Sprintf("sqlitelockapply://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
 	)
 	require.ErrorIs(t, err, errLock)
-	require.True(t, strings.HasPrefix(s, "Error: sql/migrate: acquiring database lock: "+errLock.Error()))
+	require.True(t, strings.HasPrefix(s, "Error: acquiring database lock: "+errLock.Error()))
 
 	// Will work and print stuff to the console.
 	s, err = runCmd(
@@ -328,7 +328,7 @@ func TestMigrate_Diff(t *testing.T) {
 		"--dev-url", fmt.Sprintf("sqlitelockdiff://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
 		"--to", to,
 	)
-	require.True(t, strings.HasPrefix(s, "Error: "+errLock.Error()))
+	require.True(t, strings.HasPrefix(s, "Error: acquiring database lock: "+errLock.Error()))
 	require.ErrorIs(t, err, errLock)
 }
 
diff --git a/internal/integration/cockroach_test.go b/internal/integration/cockroach_test.go
index aec5d11e..c6140089 100644
--- a/internal/integration/cockroach_test.go
+++ b/internal/integration/cockroach_test.go
@@ -8,9 +8,9 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-	"io"
 	"log"
 	"strings"
+	"sync"
 	"testing"
 
 	"ariga.io/atlas/sql/migrate"
@@ -31,50 +31,35 @@ type crdbTest struct {
 	rrw     migrate.RevisionReadWriter
 	version string
 	port    int
+	once    sync.Once
 }
 
-var crdbTests = struct {
-	drivers map[string]*crdbTest
-	ports   map[string]int
-}{
-	drivers: make(map[string]*crdbTest),
-	ports: map[string]int{
-		"cockroach": 26257,
-	},
-}
-
-func crdbInit(d string) []io.Closer {
-	var cs []io.Closer
-	if d != "" {
-		p, ok := crdbTests.ports[d]
-		if ok {
-			crdbTests.ports = map[string]int{d: p}
-		} else {
-			crdbTests.ports = make(map[string]int)
-		}
-	}
-	for version, port := range crdbTests.ports {
-		db, err := sql.Open("postgres", fmt.Sprintf("host=localhost port=%d user=root dbname=defaultdb password=pass sslmode=disable", port))
-		if err != nil {
-			log.Fatalln(err)
-		}
-		cs = append(cs, db)
-
-		drv, err := postgres.Open(db)
-		if err != nil {
-			log.Fatalln(err)
-		}
-		crdbTests.drivers[version] = &crdbTest{db: db, drv: drv, version: version, port: port, rrw: &rrw{}}
-	}
-	return cs
+var crdbTests = map[string]*crdbTest{
+	"cockroach": {port: 26257},
 }
 
 func crdbRun(t *testing.T, fn func(*crdbTest)) {
-	for version, tt := range crdbTests.drivers {
-		t.Run(version, func(t *testing.T) {
-			tt := &crdbTest{T: t, db: tt.db, drv: tt.drv, version: version, port: tt.port, rrw: tt.rrw}
-			fn(tt)
-		})
+	for version, tt := range crdbTests {
+		if flagVersion == "" || flagVersion == version {
+			t.Run(version, func(t *testing.T) {
+				tt.once.Do(func() {
+					var err error
+					tt.version = version
+					tt.rrw = &rrw{}
+					tt.db, err = sql.Open("postgres", fmt.Sprintf("host=localhost port=%d user=root dbname=defaultdb password=pass sslmode=disable", tt.port))
+					if err != nil {
+						log.Fatalln(err)
+					}
+					dbs = append(dbs, tt.db) // close connection after all tests have been run
+					tt.drv, err = postgres.Open(tt.db)
+					if err != nil {
+						log.Fatalln(err)
+					}
+				})
+				tt := &crdbTest{T: t, db: tt.db, drv: tt.drv, version: version, port: tt.port, rrw: tt.rrw}
+				fn(tt)
+			})
+		}
 	}
 }
 
diff --git a/internal/integration/integration_test.go b/internal/integration/integration_test.go
index 2434d8f1..6e32d73a 100644
--- a/internal/integration/integration_test.go
+++ b/internal/integration/integration_test.go
@@ -31,21 +31,19 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+var (
+	dbs         []io.Closer
+	flagVersion string
+)
+
 func TestMain(m *testing.M) {
-	var dialect string
-	flag.StringVar(&dialect, "dialect", "", "[mysql56, postgres10, tidb5, ...] what dialect (version) to test")
+	flag.StringVar(&flagVersion, "version", "", "[mysql56, postgres10, tidb5, ...] what version to test")
 	flag.Parse()
-	var dbs []io.Closer
-	dbs = append(dbs, myInit(dialect)...)
-	dbs = append(dbs, pgInit(dialect)...)
-	dbs = append(dbs, tidbInit(dialect)...)
-	dbs = append(dbs, crdbInit(dialect)...)
-	defer func() {
-		for _, db := range dbs {
-			db.Close()
-		}
-	}()
-	os.Exit(m.Run())
+	code := m.Run()
+	for _, db := range dbs {
+		db.Close()
+	}
+	os.Exit(code)
 }
 
 // T holds the elements common between dialect tests.
diff --git a/internal/integration/mysql_test.go b/internal/integration/mysql_test.go
index bfb72dab..f73d8630 100644
--- a/internal/integration/mysql_test.go
+++ b/internal/integration/mysql_test.go
@@ -8,9 +8,9 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-	"io"
 	"log"
 	"strings"
+	"sync"
 	"testing"
 
 	"ariga.io/atlas/sql/migrate"
@@ -28,54 +28,40 @@ type myTest struct {
 	rrw     migrate.RevisionReadWriter
 	version string
 	port    int
+	once    sync.Once
 }
 
-var myTests = struct {
-	drivers map[string]*myTest
-	ports   map[string]int
-}{
-	drivers: make(map[string]*myTest),
-	ports: map[string]int{
-		"mysql56":  3306,
-		"mysql57":  3307,
-		"mysql8":   3308,
-		"maria107": 4306,
-		"maria102": 4307,
-		"maria103": 4308,
-	},
-}
-
-func myInit(d string) []io.Closer {
-	var cs []io.Closer
-	if d != "" {
-		p, ok := myTests.ports[d]
-		if ok {
-			myTests.ports = map[string]int{d: p}
-		} else {
-			myTests.ports = make(map[string]int)
-		}
-	}
-	for version, port := range myTests.ports {
-		db, err := sql.Open("mysql", fmt.Sprintf("root:pass@tcp(localhost:%d)/test?parseTime=True", port))
-		if err != nil {
-			log.Fatalln(err)
-		}
-		cs = append(cs, db)
-		drv, err := mysql.Open(db)
-		if err != nil {
-			log.Fatalln(err)
-		}
-		myTests.drivers[version] = &myTest{db: db, drv: drv, version: version, port: port, rrw: &rrw{}}
-	}
-	return cs
+var myTests = map[string]*myTest{
+	"mysql56":  {port: 3306},
+	"mysql57":  {port: 3307},
+	"mysql8":   {port: 3308},
+	"maria107": {port: 4306},
+	"maria102": {port: 4307},
+	"maria103": {port: 4308},
 }
 
 func myRun(t *testing.T, fn func(*myTest)) {
-	for version, tt := range myTests.drivers {
-		t.Run(version, func(t *testing.T) {
-			tt := &myTest{T: t, db: tt.db, drv: tt.drv, version: version, port: tt.port, rrw: tt.rrw}
-			fn(tt)
-		})
+	for version, tt := range myTests {
+		if flagVersion == "" || flagVersion == version {
+			t.Run(version, func(t *testing.T) {
+				tt.once.Do(func() {
+					var err error
+					tt.version = version
+					tt.rrw = &rrw{}
+					tt.db, err = sql.Open("mysql", fmt.Sprintf("root:pass@tcp(localhost:%d)/test?parseTime=True", tt.port))
+					if err != nil {
+						log.Fatalln(err)
+					}
+					dbs = append(dbs, tt.db) // close connection after all tests have been run
+					tt.drv, err = mysql.Open(tt.db)
+					if err != nil {
+						log.Fatalln(err)
+					}
+				})
+				tt := &myTest{T: t, db: tt.db, drv: tt.drv, version: version, port: tt.port, rrw: tt.rrw}
+				fn(tt)
+			})
+		}
 	}
 }
 
@@ -732,6 +718,67 @@ schema "second" {
 	})
 }
 
+func TestMySQL_HCL_ForeignKeyCrossSchema(t *testing.T) {
+	const expected = `table "credit_cards" {
+  schema = schema.financial
+  column "id" {
+    null = false
+    type = int
+  }
+  column "user_id" {
+    null = false
+    type = int
+  }
+  primary_key {
+    columns = [column.id]
+  }
+  foreign_key "user_id_fkey" {
+    columns     = [column.user_id]
+    ref_columns = [table.users.users.column.id]
+    on_update   = NO_ACTION
+    on_delete   = NO_ACTION
+  }
+  index "user_id_fkey" {
+    columns = [column.user_id]
+  }
+}
+table "users" "users" {
+  schema = schema.users
+  column "id" {
+    null = false
+    type = int
+  }
+  column "email" {
+    null = false
+    type = varchar(255)
+  }
+  primary_key {
+    columns = [column.id]
+  }
+}
+schema "financial" {
+  charset = "utf8mb4"
+  collate = "utf8mb4_general_ci"
+}
+schema "users" {
+  charset = "utf8mb4"
+  collate = "utf8mb4_general_ci"
+}
+`
+	myRun(t, func(t *myTest) {
+		t.dropSchemas("financial", "users")
+		realm := t.loadRealm()
+		hcl, err := mysql.MarshalHCL(realm)
+		require.NoError(t, err)
+		t.applyRealmHcl(string(hcl) + "\n" + expected)
+		realm, err = t.drv.InspectRealm(context.Background(), &schema.InspectRealmOption{Schemas: []string{"users", "financial"}})
+		require.NoError(t, err)
+		actual, err := mysql.MarshalHCL(realm)
+		require.NoError(t, err)
+		require.Equal(t, expected, string(actual))
+	})
+}
+
 func TestMySQL_DefaultsHCL(t *testing.T) {
 	n := "atlas_defaults"
 	myRun(t, func(t *myTest) {
diff --git a/internal/integration/postgres_test.go b/internal/integration/postgres_test.go
index 5b29f548..34c4a0ac 100644
--- a/internal/integration/postgres_test.go
+++ b/internal/integration/postgres_test.go
@@ -8,16 +8,18 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-	"io"
 	"log"
+	"os"
+	"os/exec"
+	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 
 	"ariga.io/atlas/sql/migrate"
 	"ariga.io/atlas/sql/postgres"
 	"ariga.io/atlas/sql/schema"
 	"ariga.io/atlas/sql/sqlclient"
-
 	"entgo.io/ent/dialect"
 	_ "github.com/lib/pq"
 	"github.com/stretchr/testify/require"
@@ -30,53 +32,39 @@ type pgTest struct {
 	rrw     migrate.RevisionReadWriter
 	version string
 	port    int
+	once    sync.Once
 }
 
-var pgTests = struct {
-	drivers map[string]*pgTest
-	ports   map[string]int
-}{
-	drivers: make(map[string]*pgTest),
-	ports: map[string]int{
-		"postgres10": 5430,
-		"postgres11": 5431,
-		"postgres12": 5432,
-		"postgres13": 5433,
-		"postgres14": 5434,
-	},
-}
-
-func pgInit(d string) []io.Closer {
-	var cs []io.Closer
-	if d != "" {
-		p, ok := pgTests.ports[d]
-		if ok {
-			pgTests.ports = map[string]int{d: p}
-		} else {
-			pgTests.ports = make(map[string]int)
-		}
-	}
-	for version, port := range pgTests.ports {
-		db, err := sql.Open("postgres", fmt.Sprintf("host=localhost port=%d user=postgres dbname=test password=pass sslmode=disable", port))
-		if err != nil {
-			log.Fatalln(err)
-		}
-		cs = append(cs, db)
-		drv, err := postgres.Open(db)
-		if err != nil {
-			log.Fatalln(err)
-		}
-		pgTests.drivers[version] = &pgTest{db: db, drv: drv, version: version, port: port, rrw: &rrw{}}
-	}
-	return cs
+var pgTests = map[string]*pgTest{
+	"postgres10": {port: 5430},
+	"postgres11": {port: 5431},
+	"postgres12": {port: 5432},
+	"postgres13": {port: 5433},
+	"postgres14": {port: 5434},
 }
 
 func pgRun(t *testing.T, fn func(*pgTest)) {
-	for version, tt := range pgTests.drivers {
-		t.Run(version, func(t *testing.T) {
-			tt := &pgTest{T: t, db: tt.db, drv: tt.drv, version: version, port: tt.port, rrw: tt.rrw}
-			fn(tt)
-		})
+	for version, tt := range pgTests {
+		if flagVersion == "" || flagVersion == version {
+			t.Run(version, func(t *testing.T) {
+				tt.once.Do(func() {
+					var err error
+					tt.version = version
+					tt.rrw = &rrw{}
+					tt.db, err = sql.Open("postgres", fmt.Sprintf("host=localhost port=%d user=postgres dbname=test password=pass sslmode=disable", tt.port))
+					if err != nil {
+						log.Fatalln(err)
+					}
+					dbs = append(dbs, tt.db) // close connection after all tests have been run
+					tt.drv, err = postgres.Open(tt.db)
+					if err != nil {
+						log.Fatalln(err)
+					}
+				})
+				tt := &pgTest{T: t, db: tt.db, drv: tt.drv, version: version, port: tt.port, rrw: tt.rrw}
+				fn(tt)
+			})
+		}
 	}
 }
 
@@ -548,6 +536,60 @@ schema "second" {
 	})
 }
 
+func TestPostgres_HCL_ForeignKeyCrossSchema(t *testing.T) {
+	const expected = `table "credit_cards" {
+  schema = schema.financial
+  column "id" {
+    null = false
+    type = serial
+  }
+  column "user_id" {
+    null = false
+    type = integer
+  }
+  primary_key {
+    columns = [column.id]
+  }
+  foreign_key "user_id_fkey" {
+    columns     = [column.user_id]
+    ref_columns = [table.users.users.column.id]
+    on_update   = NO_ACTION
+    on_delete   = NO_ACTION
+  }
+}
+table "users" "users" {
+  schema = schema.users
+  column "id" {
+    null = false
+    type = bigserial
+  }
+  column "email" {
+    null = false
+    type = character_varying
+  }
+  primary_key {
+    columns = [column.id]
+  }
+}
+schema "financial" {
+}
+schema "users" {
+}
+`
+	pgRun(t, func(t *pgTest) {
+		t.dropSchemas("financial", "users")
+		realm := t.loadRealm()
+		hcl, err := postgres.MarshalHCL(realm)
+		require.NoError(t, err)
+		t.applyRealmHcl(string(hcl) + "\n" + expected)
+		realm, err = t.drv.InspectRealm(context.Background(), &schema.InspectRealmOption{Schemas: []string{"users", "financial"}})
+		require.NoError(t, err)
+		actual, err := postgres.MarshalHCL(realm)
+		require.NoError(t, err)
+		require.Equal(t, expected, string(actual))
+	})
+}
+
 func (t *pgTest) applyRealmHcl(spec string) {
 	realm := t.loadRealm()
 	var desired schema.Realm
@@ -698,6 +740,78 @@ func TestPostgres_CLI_MultiSchema(t *testing.T) {
 	})
 }
 
+func TestPostgres_MigrateDiffRealm(t *testing.T) {
+	bin, err := buildCmd(t)
+	require.NoError(t, err)
+	pgRun(t, func(t *pgTest) {
+		dir := t.TempDir()
+		_, err := t.db.Exec("CREATE DATABASE migrate_diff")
+		require.NoError(t, err)
+		defer t.db.Exec("DROP DATABASE IF EXISTS migrate_diff")
+
+		hcl := `
+schema "public" {}
+table "users" {
+	schema = schema.public
+	column "id" { type = integer }
+}
+schema "other" {}
+table "posts" {
+	schema = schema.other
+	column "id" { type = integer }
+}
+`
+		err = os.WriteFile(filepath.Join(dir, "schema.hcl"), []byte(hcl), 0600)
+		diff := func(name string) string {
+			out, err := exec.Command(
+				bin, "migrate", "diff", name,
+				"--dir", fmt.Sprintf("file://%s", filepath.Join(dir, "migrations")),
+				"--to", fmt.Sprintf("file://%s", filepath.Join(dir, "schema.hcl")),
+				"--dev-url", fmt.Sprintf("postgres://postgres:pass@localhost:%d/migrate_diff?sslmode=disable", t.port),
+			).CombinedOutput()
+			require.NoError(t, err, string(out))
+			return strings.TrimSpace(string(out))
+		}
+		require.Empty(t, diff("initial"))
+
+		// Expect one file and read its contents.
+		files, err := os.ReadDir(filepath.Join(dir, "migrations"))
+		require.NoError(t, err)
+		require.Equal(t, 2, len(files))
+		require.Equal(t, "atlas.sum", files[1].Name())
+		b, err := os.ReadFile(filepath.Join(dir, "migrations", files[0].Name()))
+		require.NoError(t, err)
+		require.Equal(t,
+			`-- Add new schema named "other"
+CREATE SCHEMA "other";
+-- create "users" table
+CREATE TABLE "public"."users" ("id" integer NOT NULL);
+-- create "posts" table
+CREATE TABLE "other"."posts" ("id" integer NOT NULL);
+`, string(b))
+		require.Equal(t, "The migration directory is synced with the desired state, no changes to be made", diff("no_change"))
+
+		// Append a change to the schema and expect a migration to be created.
+		hcl += `
+table "other" "users" {
+	schema = schema.other
+	column "id" { type = integer }
+}`
+		err = os.WriteFile(filepath.Join(dir, "schema.hcl"), []byte(hcl), 0600)
+		require.Empty(t, diff("second"))
+		require.Equal(t, "The migration directory is synced with the desired state, no changes to be made", diff("no_change"))
+		files, err = os.ReadDir(filepath.Join(dir, "migrations"))
+		require.NoError(t, err)
+		require.Equal(t, 3, len(files), dir)
+		b, err = os.ReadFile(filepath.Join(dir, "migrations", files[1].Name()))
+		require.NoError(t, err)
+		require.Equal(t,
+			`-- create "users" table
+CREATE TABLE "other"."users" ("id" integer NOT NULL);
+`, string(b))
+	})
+}
+
 func TestPostgres_DefaultsHCL(t *testing.T) {
 	n := "atlas_defaults"
 	pgRun(t, func(t *pgTest) {
diff --git a/internal/integration/script_test.go b/internal/integration/script_test.go
index 32b602f5..fb7e8383 100644
--- a/internal/integration/script_test.go
+++ b/internal/integration/script_test.go
@@ -18,6 +18,7 @@ import (
 	"testing"
 	"unicode"
 
+	"ariga.io/atlas/sql/migrate"
 	"ariga.io/atlas/sql/mysql"
 	"ariga.io/atlas/sql/postgres"
 	"ariga.io/atlas/sql/schema"
@@ -34,15 +35,16 @@ func TestMySQL_Script(t *testing.T) {
 			Dir:   "testdata/mysql",
 			Setup: t.setupScript,
 			Cmds: map[string]func(ts *testscript.TestScript, neg bool, args []string){
-				"only":    cmdOnly,
-				"apply":   t.cmdApply,
-				"exist":   t.cmdExist,
-				"synced":  t.cmdSynced,
-				"cmphcl":  t.cmdCmpHCL,
-				"cmpshow": t.cmdCmpShow,
-				"cmpmig":  t.cmdCmpMig,
-				"execsql": t.cmdExec,
-				"atlas":   t.cmdCLI,
+				"only":        cmdOnly,
+				"apply":       t.cmdApply,
+				"exist":       t.cmdExist,
+				"synced":      t.cmdSynced,
+				"cmphcl":      t.cmdCmpHCL,
+				"cmpshow":     t.cmdCmpShow,
+				"cmpmig":      t.cmdCmpMig,
+				"execsql":     t.cmdExec,
+				"atlas":       t.cmdCLI,
+				"clearSchema": t.clearSchema,
 			},
 		})
 	})
@@ -54,15 +56,16 @@ func TestPostgres_Script(t *testing.T) {
 			Dir:   "testdata/postgres",
 			Setup: t.setupScript,
 			Cmds: map[string]func(ts *testscript.TestScript, neg bool, args []string){
-				"only":    cmdOnly,
-				"apply":   t.cmdApply,
-				"exist":   t.cmdExist,
-				"synced":  t.cmdSynced,
-				"cmphcl":  t.cmdCmpHCL,
-				"cmpshow": t.cmdCmpShow,
-				"cmpmig":  t.cmdCmpMig,
-				"execsql": t.cmdExec,
-				"atlas":   t.cmdCLI,
+				"only":        cmdOnly,
+				"apply":       t.cmdApply,
+				"exist":       t.cmdExist,
+				"synced":      t.cmdSynced,
+				"cmphcl":      t.cmdCmpHCL,
+				"cmpshow":     t.cmdCmpShow,
+				"cmpmig":      t.cmdCmpMig,
+				"execsql":     t.cmdExec,
+				"atlas":       t.cmdCLI,
+				"clearSchema": t.clearSchema,
 			},
 		})
 	})
@@ -381,9 +384,12 @@ func (t *liteTest) cmdCLI(ts *testscript.TestScript, neg bool, args []string) {
 }
 
 func cmdCLI(ts *testscript.TestScript, neg bool, args []string, dbURL, cliPath string) {
-	workDir := ts.Getenv("WORK")
+	var (
+		workDir = ts.Getenv("WORK")
+		r       = strings.NewReplacer("URL", dbURL, "$db", ts.Getenv("db"))
+	)
 	for i, arg := range args {
-		args[i] = strings.ReplaceAll(arg, "URL", dbURL)
+		args[i] = r.Replace(arg)
 	}
 	switch l := len(args); {
 	// If command was run with a unix redirect-like suffix.
@@ -606,6 +612,26 @@ func (t *liteTest) hclDiff(ts *testscript.TestScript, name string) ([]schema.Cha
 	return changes, nil
 }
 
+func (t *myTest) clearSchema(ts *testscript.TestScript, _ bool, args []string) {
+	if len(args) == 0 {
+		args = append(args, ts.Getenv("db"))
+	}
+	_, err := t.db.Exec("DROP DATABASE IF EXISTS " + args[0])
+	ts.Check(err)
+	_, err = t.db.Exec("CREATE DATABASE IF NOT EXISTS " + args[0])
+	ts.Check(err)
+}
+
+func (t *pgTest) clearSchema(ts *testscript.TestScript, _ bool, args []string) {
+	if len(args) == 0 {
+		args = append(args, ts.Getenv("db"))
+	}
+	_, err := t.db.Exec(fmt.Sprintf("DROP SCHEMA IF EXISTS %s CASCADE", args[0]))
+	ts.Check(err)
+	_, err = t.db.Exec("CREATE SCHEMA IF NOT EXISTS " + args[0])
+	ts.Check(err)
+}
+
 func cmdSynced(ts *testscript.TestScript, neg bool, args []string, diff func(*testscript.TestScript, string) ([]schema.Change, error)) {
 	if len(args) != 1 {
 		ts.Fatalf("unexpected number of args to synced command: %d", len(args))
@@ -620,7 +646,7 @@ func cmdSynced(ts *testscript.TestScript, neg bool, args []string, diff func(*te
 	}
 }
 
-func cmdApply(ts *testscript.TestScript, neg bool, args []string, apply func(context.Context, []schema.Change) error, diff func(*testscript.TestScript, string) ([]schema.Change, error)) {
+func cmdApply(ts *testscript.TestScript, neg bool, args []string, apply func(context.Context, []schema.Change, ...migrate.PlanOption) error, diff func(*testscript.TestScript, string) ([]schema.Change, error)) {
 	changes, err := diff(ts, args[0])
 	switch {
 	case err != nil && !neg:
diff --git a/internal/integration/testdata/mysql/cli-migrate-apply.txt b/internal/integration/testdata/mysql/cli-migrate-apply.txt
new file mode 100644
index 00000000..0837473a
--- /dev/null
+++ b/internal/integration/testdata/mysql/cli-migrate-apply.txt
@@ -0,0 +1,101 @@
+only mysql
+
+! atlas migrate apply
+stderr 'Error: checksum file not found'
+stdout 'You have a checksum error in your migration directory.'
+stdout 'atlas migrate hash --force'
+
+atlas migrate hash --force
+
+# Apply all of them
+atlas migrate apply --url URL --revisions-schema $db
+stdout 'Migrating to version 3 \(3 migrations in total\):'
+stdout '-- migrating version 1'
+stdout '-> CREATE TABLE `users` \(`id` bigint NOT NULL AUTO_INCREMENT, `age` bigint NOT NULL, `name` varchar\(255\) NOT NULL, PRIMARY KEY \(`id`\)\) CHARSET utf8mb4 COLLATE utf8mb4_bin;'
+stdout '-- migrating version 2'
+stdout '-> ALTER TABLE `users` ADD UNIQUE INDEX `age` \(`age`\);'
+stdout '-- migrating version 3'
+stdout '-> CREATE TABLE `pets` \(`id` bigint NOT NULL AUTO_INCREMENT, `name` varchar\(255\) NOT NULL, PRIMARY KEY \(`id`\)\) CHARSET utf8mb4 COLLATE utf8mb4_bin;'
+stdout '-- 3 migrations'
+stdout '-- 3 sql statements'
+cmpshow users users.sql
+cmpshow pets pets.sql
+
+atlas migrate apply --url URL --revisions-schema $db
+stdout 'The migration directory is synced with the database, no migration files to execute'
+
+clearSchema
+
+# Apply one by one
+atlas migrate apply --url URL --revisions-schema $db 1
+stdout 'Migrating to version 1 \(1 migrations in total\):'
+cmpshow users users_1.sql
+
+atlas migrate apply --url URL --revisions-schema $db 1
+stdout 'Migrating to version 2 from 1 \(1 migrations in total\):'
+cmpshow users users.sql
+
+atlas migrate apply --url URL --revisions-schema $db 1
+stdout 'Migrating to version 3 from 2 \(1 migrations in total\):'
+cmpshow users users.sql
+cmpshow pets pets.sql
+
+atlas migrate apply --url URL --revisions-schema $db 1
+stdout 'The migration directory is synced with the database, no migration files to execute'
+
+-- migrations/1_first.sql --
+CREATE TABLE `users` (`id` bigint NOT NULL AUTO_INCREMENT, `age` bigint NOT NULL, `name` varchar(255) NOT NULL, PRIMARY KEY (`id`)) CHARSET utf8mb4 COLLATE utf8mb4_bin;
+
+-- migrations/2_second.sql --
+ALTER TABLE `users` ADD UNIQUE INDEX `age` (`age`);
+
+-- migrations/3_third.sql --
+CREATE TABLE `pets` (`id` bigint NOT NULL AUTO_INCREMENT, `name` varchar(255) NOT NULL, PRIMARY KEY (`id`)) CHARSET utf8mb4 COLLATE utf8mb4_bin;
+
+-- users.sql --
+CREATE TABLE `users` (
+  `id` bigint(20) NOT NULL AUTO_INCREMENT,
+  `age` bigint(20) NOT NULL,
+  `name` varchar(255) COLLATE utf8mb4_bin NOT NULL,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `age` (`age`)
+)
+
+-- mysql8/users.sql --
+CREATE TABLE `users` (
+  `id` bigint NOT NULL AUTO_INCREMENT,
+  `age` bigint NOT NULL,
+  `name` varchar(255) COLLATE utf8mb4_bin NOT NULL,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `age` (`age`)
+)
+
+-- users_1.sql --
+CREATE TABLE `users` (
+  `id` bigint(20) NOT NULL AUTO_INCREMENT,
+  `age` bigint(20) NOT NULL,
+  `name` varchar(255) COLLATE utf8mb4_bin NOT NULL,
+  PRIMARY KEY (`id`)
+)
+
+-- mysql8/users_1.sql --
+CREATE TABLE `users` (
+  `id` bigint NOT NULL AUTO_INCREMENT,
+  `age` bigint NOT NULL,
+  `name` varchar(255) COLLATE utf8mb4_bin NOT NULL,
+  PRIMARY KEY (`id`)
+)
+
+-- pets.sql --
+CREATE TABLE `pets` (
+  `id` bigint(20) NOT NULL AUTO_INCREMENT,
+  `name` varchar(255) COLLATE utf8mb4_bin NOT NULL,
+  PRIMARY KEY (`id`)
+)
+
+-- mysql8/pets.sql --
+CREATE TABLE `pets` (
+  `id` bigint NOT NULL AUTO_INCREMENT,
+  `name` varchar(255) COLLATE utf8mb4_bin NOT NULL,
+  PRIMARY KEY (`id`)
+)
diff --git a/internal/integration/testdata/mysql/cli-migrate-diff.txt b/internal/integration/testdata/mysql/cli-migrate-diff.txt
index 2de4cd66..b63ffa2c 100644
--- a/internal/integration/testdata/mysql/cli-migrate-diff.txt
+++ b/internal/integration/testdata/mysql/cli-migrate-diff.txt
@@ -1,3 +1,5 @@
+only maria
+
 exec mkdir migrations
 
 ! atlas migrate diff --to file://1.hcl --dir file://migrations
@@ -12,6 +14,11 @@ cmpmig 0 1.sql
 atlas migrate diff --dev-url URL --to file://./2.hcl second
 cmpmig 1 2.sql
 
+# Clean migration directory and run diff with a custom qualifier.
+exec rm -rf migrations
+atlas migrate diff --dev-url URL --to file://./1.hcl --qualifier test first
+cmpmig 0 1-qualifier.sql
+
 -- 1.hcl --
 schema "script_cli_migrate_diff" {}
 
@@ -31,7 +38,11 @@ table "users" {
 
 -- 1.sql --
 -- create "users" table
-CREATE TABLE `script_cli_migrate_diff`.`users` (`id` bigint NOT NULL AUTO_INCREMENT, PRIMARY KEY (`id`)) CHARSET utf8mb4 COLLATE utf8mb4_general_ci;
+CREATE TABLE `users` (`id` bigint NOT NULL AUTO_INCREMENT, PRIMARY KEY (`id`)) CHARSET utf8mb4 COLLATE utf8mb4_general_ci;
+
+-- 1-qualifier.sql --
+-- create "users" table
+CREATE TABLE `test`.`users` (`id` bigint NOT NULL AUTO_INCREMENT, PRIMARY KEY (`id`)) CHARSET utf8mb4 COLLATE utf8mb4_general_ci;
 
 -- 2.hcl --
 schema "script_cli_migrate_diff" {}
@@ -57,16 +68,16 @@ table "users" {
 
 -- 2.sql --
 -- modify "users" table
-ALTER TABLE `script_cli_migrate_diff`.`users` ADD COLUMN `create_time` timestamp(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6);
+ALTER TABLE `users` ADD COLUMN `create_time` timestamp(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6);
 
 -- maria102/2.sql --
 -- modify "users" table
-ALTER TABLE `script_cli_migrate_diff`.`users` ADD COLUMN `create_time` timestamp(6) NOT NULL DEFAULT (current_timestamp(6));
+ALTER TABLE `users` ADD COLUMN `create_time` timestamp(6) NOT NULL DEFAULT (current_timestamp(6));
 
 -- maria103/2.sql --
 -- modify "users" table
-ALTER TABLE `script_cli_migrate_diff`.`users` ADD COLUMN `create_time` timestamp(6) NOT NULL DEFAULT (current_timestamp(6));
+ALTER TABLE `users` ADD COLUMN `create_time` timestamp(6) NOT NULL DEFAULT (current_timestamp(6));
 
 -- maria107/2.sql --
 -- modify "users" table
-ALTER TABLE `script_cli_migrate_diff`.`users` ADD COLUMN `create_time` timestamp(6) NOT NULL DEFAULT (current_timestamp(6));
+ALTER TABLE `users` ADD COLUMN `create_time` timestamp(6) NOT NULL DEFAULT (current_timestamp(6));
diff --git a/internal/integration/testdata/postgres/cli-migrate-apply.txt b/internal/integration/testdata/postgres/cli-migrate-apply.txt
new file mode 100644
index 00000000..66452cd6
--- /dev/null
+++ b/internal/integration/testdata/postgres/cli-migrate-apply.txt
@@ -0,0 +1,83 @@
+only postgres14
+
+! atlas migrate apply
+stderr 'Error: checksum file not found'
+stdout 'You have a checksum error in your migration directory.'
+stdout 'atlas migrate hash --force'
+
+atlas migrate hash --force
+
+# Apply all of them
+atlas migrate apply --url URL --revisions-schema $db
+stdout 'Migrating to version 3 \(3 migrations in total\):'
+stdout '-- migrating version 1'
+stdout '-> CREATE TABLE "users" \("id" bigint NOT NULL GENERATED BY DEFAULT AS IDENTITY, "age" bigint NOT NULL, "name" character varying NOT NULL, PRIMARY KEY \("id"\)\);'
+stdout '-- migrating version 2'
+stdout '-> CREATE UNIQUE INDEX "users_age_key" ON "users" \("age"\);'
+stdout '-- migrating version 3'
+stdout '-> CREATE TABLE "pets" \("id" bigint NOT NULL GENERATED BY DEFAULT AS IDENTITY, "name" character varying NOT NULL, PRIMARY KEY \("id"\)\);'
+stdout '-- 3 migrations'
+stdout '-- 3 sql statements'
+cmpshow users users.sql
+cmpshow pets pets.sql
+
+atlas migrate apply --url URL --revisions-schema $db 1
+stdout 'The migration directory is synced with the database, no migration files to execute'
+
+clearSchema
+
+# Apply one by one
+atlas migrate apply --url URL --revisions-schema $db 1
+stdout 'Migrating to version 1 \(1 migrations in total\):'
+cmpshow users users_1.sql
+
+atlas migrate apply --url URL --revisions-schema $db 1
+stdout 'Migrating to version 2 from 1 \(1 migrations in total\):'
+cmpshow users users.sql
+
+atlas migrate apply --url URL --revisions-schema $db 1
+stdout 'Migrating to version 3 from 2 \(1 migrations in total\):'
+cmpshow users users.sql
+cmpshow pets pets.sql
+
+atlas migrate apply --url URL --revisions-schema $db 1
+stdout 'The migration directory is synced with the database, no migration files to execute'
+
+-- migrations/1_first.sql --
+CREATE TABLE "users" ("id" bigint NOT NULL GENERATED BY DEFAULT AS IDENTITY, "age" bigint NOT NULL, "name" character varying NOT NULL, PRIMARY KEY ("id"));
+
+-- migrations/2_second.sql --
+CREATE UNIQUE INDEX "users_age_key" ON "users" ("age");
+
+-- migrations/3_third.sql --
+CREATE TABLE "pets" ("id" bigint NOT NULL GENERATED BY DEFAULT AS IDENTITY, "name" character varying NOT NULL, PRIMARY KEY ("id"));
+
+-- users.sql --
+                        Table "script_cli_migrate_apply.users"
+ Column |       Type        | Collation | Nullable |             Default
+--------+-------------------+-----------+----------+----------------------------------
+ id     | bigint            |           | not null | generated by default as identity
+ age    | bigint            |           | not null |
+ name   | character varying |           | not null |
+Indexes:
+    "users_pkey" PRIMARY KEY, btree (id)
+    "users_age_key" UNIQUE, btree (age)
+
+-- users_1.sql --
+                        Table "script_cli_migrate_apply.users"
+ Column |       Type        | Collation | Nullable |             Default
+--------+-------------------+-----------+----------+----------------------------------
+ id     | bigint            |           | not null | generated by default as identity
+ age    | bigint            |           | not null |
+ name   | character varying |           | not null |
+Indexes:
+    "users_pkey" PRIMARY KEY, btree (id)
+
+-- pets.sql --
+                        Table "script_cli_migrate_apply.pets"
+ Column |       Type        | Collation | Nullable |             Default
+--------+-------------------+-----------+----------+----------------------------------
+ id     | bigint            |           | not null | generated by default as identity
+ name   | character varying |           | not null |
+Indexes:
+    "pets_pkey" PRIMARY KEY, btree (id)
diff --git a/internal/integration/testdata/postgres/cli-migrate-diff.txt b/internal/integration/testdata/postgres/cli-migrate-diff.txt
index 1811c7ca..342da9ec 100644
--- a/internal/integration/testdata/postgres/cli-migrate-diff.txt
+++ b/internal/integration/testdata/postgres/cli-migrate-diff.txt
@@ -1,3 +1,5 @@
+only postgres10 postgres11 postgres13
+
 exec mkdir migrations
 
 ! atlas migrate diff --to file://1.hcl --dir file://migrations
@@ -29,7 +31,7 @@ table "users" {
 
 -- 1.sql --
 -- create "users" table
-CREATE TABLE "script_cli_migrate_diff"."users" ("id" bigint NOT NULL, PRIMARY KEY ("id"));
+CREATE TABLE "users" ("id" bigint NOT NULL, PRIMARY KEY ("id"));
 
 -- 2.hcl --
 schema "script_cli_migrate_diff" {}
@@ -53,4 +55,4 @@ table "users" {
 
 -- 2.sql --
 -- modify "users" table
-ALTER TABLE "script_cli_migrate_diff"."users" ADD COLUMN "create_time" timestamp(4) NOT NULL DEFAULT CURRENT_TIMESTAMP(4);
\ No newline at end of file
+ALTER TABLE "users" ADD COLUMN "create_time" timestamp(4) NOT NULL DEFAULT CURRENT_TIMESTAMP(4);
\ No newline at end of file
diff --git a/internal/integration/testdata/postgres/cli-migrate-status.txt b/internal/integration/testdata/postgres/cli-migrate-status.txt
new file mode 100644
index 00000000..9a94d434
--- /dev/null
+++ b/internal/integration/testdata/postgres/cli-migrate-status.txt
@@ -0,0 +1,51 @@
+only postgres12
+
+# make sure sum file is correct
+atlas migrate hash --force
+
+# for clean database
+atlas migrate status --url URL --revisions-schema $db
+cmp stdout status_clean.txt
+
+# apply one
+atlas migrate apply --url URL --revisions-schema $db 1
+atlas migrate status --url URL --revisions-schema $db
+cmp stdout status_one_applied.txt
+
+# apply next (and last)
+atlas migrate apply --url URL --revisions-schema $db 1
+atlas migrate status --url URL --revisions-schema $db
+cmp stdout status_ok.txt
+
+-- migrations/1.sql --
+CREATE TABLE "users" ("id" bigint NOT NULL GENERATED ALWAYS AS IDENTITY, PRIMARY KEY ("id"));
+
+-- migrations/2.sql --
+ALTER TABLE "users" ADD COLUMN "happy" boolean NOT NULL DEFAULT true;
+
+-- broken_migration.sql --
+CREATE TABLE "users" ("id" bigint NOT NULL GENERATED ALWAYS AS IDENTITY, PRIMARY KEY ("id"));
+THIS LINE ADDS A SYNTAX ERROR;
+
+-- fixed_migration.sql --
+CREATE TABLE "users" ("id" bigint NOT NULL GENERATED ALWAYS AS IDENTITY, PRIMARY KEY ("id"));
+ALTER TABLE "users" ADD COLUMN "happy" boolean NOT NULL DEFAULT true;
+
+-- status_clean.txt --
+Migration Status: PENDING
+  -- Current Version:  No version applied yet
+  -- Next Version:     1
+  -- Executed Files:   0
+  -- Pending Files:    2
+-- status_one_applied.txt --
+Migration Status: PENDING
+  -- Current Version:  1
+  -- Next Version:     2
+  -- Executed Files:   1
+  -- Pending Files:    1
+-- status_ok.txt --
+Migration Status: OK
+  -- Current Version:  2
+  -- Next Version:     Already at latest version
+  -- Executed Files:   2
+  -- Pending Files:    0
\ No newline at end of file
diff --git a/internal/integration/tidb_test.go b/internal/integration/tidb_test.go
index cd3e2466..c0b79666 100644
--- a/internal/integration/tidb_test.go
+++ b/internal/integration/tidb_test.go
@@ -8,7 +8,6 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-	"io"
 	"log"
 	"testing"
 
@@ -25,45 +24,33 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-var tidbTests = struct {
-	drivers map[string]*myTest
-	ports   map[string]int
-}{
-	drivers: make(map[string]*myTest),
-	ports:   map[string]int{"tidb5": 4309, "tidb6": 4310},
-}
-
-func tidbInit(d string) []io.Closer {
-	var cs []io.Closer
-	if d != "" {
-		p, ok := tidbTests.ports[d]
-		if ok {
-			tidbTests.ports = map[string]int{d: p}
-		} else {
-			tidbTests.ports = make(map[string]int)
-		}
-	}
-	for version, port := range tidbTests.ports {
-		db, err := sql.Open("mysql", fmt.Sprintf("root@tcp(localhost:%d)/test?parseTime=True", port))
-		if err != nil {
-			log.Fatalln(err)
-		}
-		cs = append(cs, db)
-		drv, err := mysql.Open(db)
-		if err != nil {
-			log.Fatalln(err)
-		}
-		tidbTests.drivers[version] = &myTest{db: db, drv: drv, version: version, port: port, rrw: &rrw{}}
-	}
-	return cs
+var tidbTests = map[string]*myTest{
+	"tidb5": {port: 4309},
+	"tidb6": {port: 4310},
 }
 
 func tidbRun(t *testing.T, fn func(*myTest)) {
-	for version, tt := range tidbTests.drivers {
-		t.Run(version, func(t *testing.T) {
-			tt := &myTest{T: t, db: tt.db, drv: tt.drv, version: version, port: tt.port, rrw: tt.rrw}
-			fn(tt)
-		})
+	for version, tt := range tidbTests {
+		if flagVersion == "" || flagVersion == version {
+			t.Run(version, func(t *testing.T) {
+				tt.once.Do(func() {
+					var err error
+					tt.version = version
+					tt.rrw = &rrw{}
+					tt.db, err = sql.Open("mysql", fmt.Sprintf("root@tcp(localhost:%d)/test?parseTime=True", tt.port))
+					if err != nil {
+						log.Fatalln(err)
+					}
+					dbs = append(dbs, tt.db) // close connection after all tests have been run
+					tt.drv, err = mysql.Open(tt.db)
+					if err != nil {
+						log.Fatalln(err)
+					}
+				})
+				tt := &myTest{T: t, db: tt.db, drv: tt.drv, version: version, port: tt.port, rrw: tt.rrw}
+				fn(tt)
+			})
+		}
 	}
 }
 
diff --git a/sql/internal/sqlx/dev_test.go b/sql/internal/sqlx/dev_test.go
index c5f41d2e..c3706def 100644
--- a/sql/internal/sqlx/dev_test.go
+++ b/sql/internal/sqlx/dev_test.go
@@ -53,7 +53,7 @@ func (m *mockDriver) InspectRealm(_ context.Context, opts *schema.InspectRealmOp
 	return m.realm, nil
 }
 
-func (m *mockDriver) ApplyChanges(_ context.Context, changes []schema.Change) error {
+func (m *mockDriver) ApplyChanges(_ context.Context, changes []schema.Change, _ ...migrate.PlanOption) error {
 	m.changes = append(m.changes, changes)
 	return nil
 }
diff --git a/sql/internal/sqlx/plan_test.go b/sql/internal/sqlx/plan_test.go
index 64fb5085..7976b97f 100644
--- a/sql/internal/sqlx/plan_test.go
+++ b/sql/internal/sqlx/plan_test.go
@@ -92,3 +92,23 @@ func TestDetachCycles(t *testing.T) {
 	workplaces.ForeignKeys = nil
 	require.Equal(t, deletion, planned[2:])
 }
+
+func TestCheckChangesScope(t *testing.T) {
+	err := CheckChangesScope([]schema.Change{
+		&schema.AddSchema{},
+	})
+	require.EqualError(t, err, "*schema.AddSchema is not allowed when migration plan is scoped to one schema")
+	err = CheckChangesScope([]schema.Change{
+		&schema.ModifySchema{},
+	})
+	require.EqualError(t, err, "*schema.ModifySchema is not allowed when migration plan is scoped to one schema")
+	err = CheckChangesScope([]schema.Change{
+		&schema.DropSchema{},
+	})
+	require.EqualError(t, err, "*schema.DropSchema is not allowed when migration plan is scoped to one schema")
+	err = CheckChangesScope([]schema.Change{
+		&schema.AddTable{T: schema.NewTable("users").SetSchema(schema.New("s1"))},
+		&schema.AddTable{T: schema.NewTable("users").SetSchema(schema.New("s2"))},
+	})
+	require.EqualError(t, err, "found 2 schemas when migration plan is scoped to one")
+}
diff --git a/sql/internal/sqlx/sqlx_test.go b/sql/internal/sqlx/sqlx_test.go
index 8612e94b..952bb751 100644
--- a/sql/internal/sqlx/sqlx_test.go
+++ b/sql/internal/sqlx/sqlx_test.go
@@ -61,6 +61,27 @@ func TestBuilder(t *testing.T) {
 	require.Equal(t, `CREATE TABLE "users" ("a" int NOT NULL, "b" int NOT NULL, "c" int NOT NULL, PRIMARY KEY ("a", "b", "c"))`, b.String())
 }
 
+func TestBuilder_Qualifier(t *testing.T) {
+	var (
+		s = "other"
+		b = &Builder{QuoteChar: '"', Schema: &s}
+	)
+	b.P("CREATE TABLE").Table(schema.NewTable("users"))
+	require.Equal(t, `CREATE TABLE "other"."users"`, b.String())
+
+	// Bypass table schema.
+	b.Reset()
+	b.P("CREATE TABLE").Table(schema.NewTable("users").SetSchema(schema.New("test")))
+	require.Equal(t, `CREATE TABLE "other"."users"`, b.String())
+
+	// Empty qualifier, means skip.
+	s = ""
+	b.Reset()
+	b.P("CREATE TABLE").Table(schema.NewTable("users").SetSchema(schema.New("test")))
+	require.Equal(t, `CREATE TABLE "users"`, b.String())
+
+}
+
 func TestMayWrap(t *testing.T) {
 	tests := []struct {
 		input   string
diff --git a/sql/migrate/migrate_test.go b/sql/migrate/migrate_test.go
index d43be935..06495d74 100644
--- a/sql/migrate/migrate_test.go
+++ b/sql/migrate/migrate_test.go
@@ -59,7 +59,7 @@ func TestPlanner_WritePlan(t *testing.T) {
 
 func TestPlanner_Plan(t *testing.T) {
 	var (
-		drv = &lockMockDriver{&mockDriver{}}
+		drv = &mockDriver{}
 		ctx = context.Background()
 	)
 	d, err := migrate.NewLocalDir(t.TempDir())
@@ -87,81 +87,62 @@ func TestPlanner_Plan(t *testing.T) {
 	require.Equal(t, drv.plan, plan)
 }
 
-func TestPlanner_PlanSchemaMode(t *testing.T) {
+func TestPlanner_PlanSchema(t *testing.T) {
 	var (
-		drv = &lockMockDriver{&mockDriver{}}
+		drv = &mockDriver{}
 		ctx = context.Background()
 	)
 	d, err := migrate.NewLocalDir(t.TempDir())
 	require.NoError(t, err)
 
 	// Schema is missing in dev connection.
-	pl := migrate.NewPlanner(drv, d, migrate.PlanSchema("test"))
-	plan, err := pl.Plan(ctx, "empty", migrate.Realm(nil))
-	require.EqualError(t, err, `missing schema "test" in realm after replaying migration directory`)
+	pl := migrate.NewPlanner(drv, d)
+	plan, err := pl.PlanSchema(ctx, "empty", migrate.Realm(nil))
+	require.EqualError(t, err, `not found`)
 	require.Nil(t, plan)
 
 	drv.realm = *schema.NewRealm(schema.New("test"))
-	pl = migrate.NewPlanner(drv, d, migrate.PlanSchema("test"))
-	plan, err = pl.Plan(ctx, "empty", migrate.Realm(schema.NewRealm()))
-	require.EqualError(t, err, `missing schema definition in desired state`)
+	pl = migrate.NewPlanner(drv, d)
+	plan, err = pl.PlanSchema(ctx, "empty", migrate.Realm(schema.NewRealm()))
+	require.EqualError(t, err, `no schema was found in desired state`)
 	require.Nil(t, plan)
 
 	drv.realm = *schema.NewRealm(schema.New("test"))
-	pl = migrate.NewPlanner(drv, d, migrate.PlanSchema("test"))
-	plan, err = pl.Plan(ctx, "multi", migrate.Realm(schema.NewRealm(schema.New("test"), schema.New("dev"))))
-	require.EqualError(t, err, `found 2 schema definitions in desired state on schema mode`)
+	pl = migrate.NewPlanner(drv, d)
+	plan, err = pl.PlanSchema(ctx, "empty", migrate.Realm(schema.NewRealm(schema.New("test"), schema.New("dev"))))
+	require.EqualError(t, err, `2 schemas were found in desired state; expect 1`)
 	require.Nil(t, plan)
 
 	drv.realm = *schema.NewRealm(schema.New("test"))
-	pl = migrate.NewPlanner(drv, d, migrate.PlanSchema("test"))
-	plan, err = pl.Plan(ctx, "multi", migrate.Realm(schema.NewRealm(schema.New("test"))))
+	pl = migrate.NewPlanner(drv, d)
+	plan, err = pl.PlanSchema(ctx, "multi", migrate.Realm(schema.NewRealm(schema.New("test"))))
 	require.ErrorIs(t, err, migrate.ErrNoPlan)
 	require.Nil(t, plan)
 }
 
-func TestExecutor_ReadState(t *testing.T) {
+func TestExecutor_Replay(t *testing.T) {
 	ctx := context.Background()
 	d, err := migrate.NewLocalDir("testdata/migrate")
 	require.NoError(t, err)
 
-	// Locking not supported.
-	_, err = migrate.NewExecutor(&mockDriver{}, d, migrate.NopRevisionReadWriter{})
-	require.ErrorIs(t, err, migrate.ErrLockUnsupported)
-
-	drv := &lockMockDriver{&mockDriver{}}
+	drv := &mockDriver{}
 	ex, err := migrate.NewExecutor(drv, d, migrate.NopRevisionReadWriter{})
 	require.NoError(t, err)
 
-	_, err = ex.ReadState(ctx)
+	_, err = ex.Replay(ctx, migrate.RealmConn(drv, nil))
 	require.NoError(t, err)
 	require.Equal(t, []string{"DROP TABLE IF EXISTS t;", "CREATE TABLE t(c int);"}, drv.executed)
-	require.Equal(t, 2, drv.lockCounter)
-	require.Equal(t, 2, drv.unlockCounter)
-	require.True(t, drv.released())
-
-	// Does not work if locked.
-	drv.locks = map[string]struct{}{"atlas_migration_directory_state": {}}
-	_, err = ex.ReadState(ctx)
-	require.EqualError(t, err, "sql/migrate: acquiring database lock: lockErr")
-	require.Equal(t, 2, drv.lockCounter)
-	require.Equal(t, 2, drv.unlockCounter)
-	require.False(t, drv.released())
-	drv.locks = make(map[string]struct{})
 
 	// Does not work if database is not clean.
 	drv.dirty = true
 	drv.realm = schema.Realm{Schemas: []*schema.Schema{{Name: "schema"}}}
-	_, err = ex.ReadState(ctx)
+	_, err = ex.Replay(ctx, migrate.RealmConn(drv, nil))
 	require.ErrorAs(t, err, &migrate.NotCleanError{})
-	require.Equal(t, 3, drv.lockCounter)
-	require.Equal(t, 3, drv.unlockCounter)
-	require.True(t, drv.released())
 }
 
 func TestExecutor_Pending(t *testing.T) {
 	var (
-		drv  = &lockMockDriver{&mockDriver{}}
+		drv  = &mockDriver{}
 		rrw  = &mockRevisionReadWriter{}
 		log  = &mockLogger{}
 		rev1 = &migrate.Revision{
@@ -232,23 +213,18 @@ func TestExecutor(t *testing.T) {
 	require.EqualError(t, err, "sql/migrate: execute: no revision storage given")
 	require.Nil(t, ex)
 
-	// Does not work if no locking mechanism is provided.
-	ex, err = migrate.NewExecutor(&mockDriver{}, dir, &mockRevisionReadWriter{})
-	require.ErrorIs(t, err, migrate.ErrLockUnsupported)
-	require.Nil(t, ex)
-
 	// Does not operate on invalid migration dir.
 	dir, err = migrate.NewLocalDir(t.TempDir())
 	require.NoError(t, err)
 	require.NoError(t, dir.WriteFile("atlas.sum", hash))
-	ex, err = migrate.NewExecutor(&lockMockDriver{&mockDriver{}}, dir, &mockRevisionReadWriter{})
+	ex, err = migrate.NewExecutor(&mockDriver{}, dir, &mockRevisionReadWriter{})
 	require.NoError(t, err)
 	require.NotNil(t, ex)
 	require.ErrorIs(t, ex.ExecuteN(context.Background(), 0), migrate.ErrChecksumMismatch)
 
 	// Prerequisites.
 	var (
-		drv  = &lockMockDriver{&mockDriver{}}
+		drv  = &mockDriver{}
 		rrw  = &mockRevisionReadWriter{}
 		log  = &mockLogger{}
 		rev1 = &migrate.Revision{
@@ -288,9 +264,6 @@ func TestExecutor(t *testing.T) {
 		migrate.LogStmt{SQL: "ALTER TABLE t_sub ADD c2 int;"},
 		migrate.LogDone{},
 	}, []migrate.LogEntry(*log))
-	require.Equal(t, drv.lockCounter, 1)
-	require.Equal(t, drv.unlockCounter, 1)
-	require.True(t, drv.released())
 
 	// Partly is pending.
 	p, err := ex.Pending(context.Background())
@@ -300,20 +273,17 @@ func TestExecutor(t *testing.T) {
 
 	// Apply one by one.
 	*rrw = mockRevisionReadWriter{}
-	*drv = lockMockDriver{&mockDriver{}}
+	*drv = mockDriver{}
 
 	require.NoError(t, ex.ExecuteN(context.Background(), 1))
-	require.Equal(t, drv.executed, []string{"CREATE TABLE t_sub(c int);", "ALTER TABLE t_sub ADD c1 int;"})
+	require.Equal(t, []string{"CREATE TABLE t_sub(c int);", "ALTER TABLE t_sub ADD c1 int;"}, drv.executed)
 	requireEqualRevisions(t, migrate.Revisions{rev1}, migrate.Revisions(*rrw))
 
 	require.NoError(t, ex.ExecuteN(context.Background(), 1))
-	require.Equal(t, drv.executed, []string{
+	require.Equal(t, []string{
 		"CREATE TABLE t_sub(c int);", "ALTER TABLE t_sub ADD c1 int;", "ALTER TABLE t_sub ADD c2 int;",
-	})
+	}, drv.executed)
 	requireEqualRevisions(t, migrate.Revisions{rev1, rev2}, migrate.Revisions(*rrw))
-	require.Equal(t, 2, drv.lockCounter)
-	require.Equal(t, 2, drv.unlockCounter)
-	require.True(t, drv.released())
 
 	// Partly is pending.
 	p, err = ex.Pending(context.Background())
@@ -323,7 +293,7 @@ func TestExecutor(t *testing.T) {
 
 	// Suppose first revision is already executed, only execute second migration file.
 	*rrw = mockRevisionReadWriter(migrate.Revisions{rev1})
-	*drv = lockMockDriver{&mockDriver{}}
+	*drv = mockDriver{}
 
 	require.NoError(t, ex.ExecuteN(context.Background(), 1))
 	require.Equal(t, []string{"ALTER TABLE t_sub ADD c2 int;"}, drv.executed)
@@ -335,13 +305,9 @@ func TestExecutor(t *testing.T) {
 	require.Len(t, p, 1)
 	require.Equal(t, "3_partly.sql", p[0].Name())
 
-	require.Equal(t, 1, drv.lockCounter)
-	require.Equal(t, 1, drv.unlockCounter)
-	require.True(t, drv.released())
-
 	// Failing, counter will be correct.
 	*rrw = mockRevisionReadWriter(migrate.Revisions{rev1, rev2})
-	*drv = lockMockDriver{&mockDriver{}}
+	*drv = mockDriver{}
 	drv.failOn(2, errors.New("this is an error"))
 	require.ErrorContains(t, ex.ExecuteN(context.Background(), 1), "this is an error")
 	revs, err := rrw.ReadRevisions(context.Background())
@@ -362,7 +328,7 @@ func TestExecutor(t *testing.T) {
 
 	// Re-attempting to migrate will pick up where the execution was left off.
 	revs[len(revs)-1].PartialHashes[0] = h
-	*drv = lockMockDriver{&mockDriver{}}
+	*drv = mockDriver{}
 	require.NoError(t, ex.ExecuteN(context.Background(), 1))
 	require.Equal(t, []string{"ALTER TABLE t_sub ADD c4 int;"}, drv.executed)
 
@@ -373,7 +339,7 @@ func TestExecutor(t *testing.T) {
 func TestExecutor_Baseline(t *testing.T) {
 	var (
 		rrw mockRevisionReadWriter
-		drv = &lockMockDriver{&mockDriver{dirty: true}}
+		drv = &mockDriver{dirty: true}
 		log = &mockLogger{}
 	)
 	dir, err := migrate.NewLocalDir(filepath.Join("testdata/migrate", "sub"))
@@ -417,7 +383,7 @@ func TestExecutor_Baseline(t *testing.T) {
 
 func TestExecutor_FromVersion(t *testing.T) {
 	var (
-		drv = &lockMockDriver{&mockDriver{}}
+		drv = &mockDriver{}
 		log = &mockLogger{}
 		rrw = &mockRevisionReadWriter{
 			{
@@ -455,19 +421,15 @@ func TestExecutor_FromVersion(t *testing.T) {
 type (
 	mockDriver struct {
 		migrate.Driver
-		plan          *migrate.Plan
-		changes       []schema.Change
-		applied       []schema.Change
-		realm         schema.Realm
-		executed      []string
-		locks         map[string]struct{}
-		lockCounter   int
-		unlockCounter int
-		failCounter   int
-		failWith      error
-		dirty         bool
+		plan        *migrate.Plan
+		changes     []schema.Change
+		applied     []schema.Change
+		realm       schema.Realm
+		executed    []string
+		failCounter int
+		failWith    error
+		dirty       bool
 	}
-	lockMockDriver struct{ *mockDriver }
 )
 
 // the nth call to ExecContext will fail with the given error.
@@ -487,6 +449,13 @@ func (m *mockDriver) ExecContext(_ context.Context, query string, _ ...interface
 	return nil, nil
 }
 
+func (m *mockDriver) InspectSchema(context.Context, string, *schema.InspectOptions) (*schema.Schema, error) {
+	if len(m.realm.Schemas) == 0 {
+		return nil, schema.NotExistError{Err: errors.New("not found")}
+	}
+	return m.realm.Schemas[0], nil
+}
+
 func (m *mockDriver) InspectRealm(context.Context, *schema.InspectRealmOption) (*schema.Realm, error) {
 	return &m.realm, nil
 }
@@ -499,11 +468,11 @@ func (m *mockDriver) RealmDiff(_, _ *schema.Realm) ([]schema.Change, error) {
 	return m.changes, nil
 }
 
-func (m *mockDriver) PlanChanges(context.Context, string, []schema.Change) (*migrate.Plan, error) {
+func (m *mockDriver) PlanChanges(context.Context, string, []schema.Change, ...migrate.PlanOption) (*migrate.Plan, error) {
 	return m.plan, nil
 }
 
-func (m *mockDriver) ApplyChanges(_ context.Context, changes []schema.Change) error {
+func (m *mockDriver) ApplyChanges(_ context.Context, changes []schema.Change, _ ...migrate.PlanOption) error {
 	m.applied = changes
 	return nil
 }
@@ -526,26 +495,6 @@ func (m *mockDriver) CheckClean(context.Context, *migrate.TableIdent) error {
 	return nil
 }
 
-func (m *lockMockDriver) Lock(_ context.Context, name string, _ time.Duration) (schema.UnlockFunc, error) {
-	if _, ok := m.locks[name]; ok {
-		return nil, errors.New("lockErr")
-	}
-	m.lockCounter++
-	if m.locks == nil {
-		m.locks = make(map[string]struct{})
-	}
-	m.locks[name] = struct{}{}
-	return func() error {
-		m.unlockCounter++
-		delete(m.locks, name)
-		return nil
-	}, nil
-}
-
-func (m *lockMockDriver) released() bool {
-	return len(m.locks) == 0
-}
-
 type mockRevisionReadWriter migrate.Revisions
 
 func (mockRevisionReadWriter) Ident() *migrate.TableIdent {
diff --git a/sql/mysql/migrate_test.go b/sql/mysql/migrate_test.go
index de0a81cc..55b834c3 100644
--- a/sql/mysql/migrate_test.go
+++ b/sql/mysql/migrate_test.go
@@ -175,6 +175,7 @@ func TestPlanChanges(t *testing.T) {
 	tests := []struct {
 		version  string
 		changes  []schema.Change
+		options  []migrate.PlanOption
 		wantPlan *migrate.Plan
 		wantErr  bool
 	}{
@@ -882,6 +883,56 @@ func TestPlanChanges(t *testing.T) {
 				},
 			},
 		},
+		// Empty qualifier.
+		{
+			changes: []schema.Change{
+				&schema.AddTable{T: schema.NewTable("t").SetSchema(schema.New("d")).AddColumns(schema.NewIntColumn("a", "int"))},
+			},
+			options: []migrate.PlanOption{
+				func(o *migrate.PlanOptions) { o.SchemaQualifier = new(string) },
+			},
+			wantPlan: &migrate.Plan{
+				Reversible: true,
+				Changes: []*migrate.Change{
+					{
+						Cmd:     "CREATE TABLE `t` (`a` int NOT NULL)",
+						Reverse: "DROP TABLE `t`",
+					},
+				},
+			},
+		},
+		// Custom qualifier.
+		{
+			changes: []schema.Change{
+				&schema.AddTable{T: schema.NewTable("t").SetSchema(schema.New("d")).AddColumns(schema.NewIntColumn("a", "int"))},
+			},
+			options: []migrate.PlanOption{
+				func(o *migrate.PlanOptions) {
+					s := "other"
+					o.SchemaQualifier = &s
+				},
+			},
+			wantPlan: &migrate.Plan{
+				Reversible: true,
+				Changes: []*migrate.Change{
+					{
+						Cmd:     "CREATE TABLE `other`.`t` (`a` int NOT NULL)",
+						Reverse: "DROP TABLE `other`.`t`",
+					},
+				},
+			},
+		},
+		// Empty qualifier in multi-schema mode should fail.
+		{
+			changes: []schema.Change{
+				&schema.AddTable{T: schema.NewTable("t1").SetSchema(schema.New("s1")).AddColumns(schema.NewIntColumn("a", "int"))},
+				&schema.AddTable{T: schema.NewTable("t2").SetSchema(schema.New("s2")).AddColumns(schema.NewIntColumn("a", "int"))},
+			},
+			options: []migrate.PlanOption{
+				func(o *migrate.PlanOptions) { o.SchemaQualifier = new(string) },
+			},
+			wantErr: true,
+		},
 	}
 	for i, tt := range tests {
 		t.Run(strconv.Itoa(i), func(t *testing.T) {
@@ -890,7 +941,7 @@ func TestPlanChanges(t *testing.T) {
 			}
 			db, _, err := newMigrate(tt.version)
 			require.NoError(t, err)
-			plan, err := db.PlanChanges(context.Background(), "wantPlan", tt.changes)
+			plan, err := db.PlanChanges(context.Background(), "wantPlan", tt.changes, tt.options...)
 			if tt.wantErr {
 				require.Error(t, err, "expect plan to fail")
 				return
diff --git a/sql/mysql/sqlspec_test.go b/sql/mysql/sqlspec_test.go
index 1420b13d..e9901f20 100644
--- a/sql/mysql/sqlspec_test.go
+++ b/sql/mysql/sqlspec_test.go
@@ -8,8 +8,11 @@ import (
 	"fmt"
 	"testing"
 
+	"ariga.io/atlas/schemahcl"
 	"ariga.io/atlas/sql/internal/spectest"
+	"ariga.io/atlas/sql/internal/specutil"
 	"ariga.io/atlas/sql/schema"
+	"ariga.io/atlas/sql/sqlspec"
 	"github.com/stretchr/testify/require"
 )
 
@@ -1186,6 +1189,41 @@ func TestInputVars(t *testing.T) {
 	spectest.TestInputVars(t, EvalHCL)
 }
 
+func TestQualifyReferencedTables(t *testing.T) {
+	var (
+		col       = schema.NewIntColumn("col", "integer")
+		targetCol = schema.NewIntColumn("target_column", "integer")
+		targetTbl = schema.NewTable("target_table").AddColumns(targetCol)
+		realm     = schema.NewRealm(
+			schema.New("target_schema").AddTables(targetTbl),
+			schema.New("sch").
+				AddTables(
+					schema.NewTable("tbl").
+						AddColumns(col).
+						AddForeignKeys(
+							schema.NewForeignKey("col_fk").
+								SetRefTable(targetTbl).
+								AddColumns(col).
+								AddRefColumns(targetCol),
+						),
+				),
+		)
+		tables = []*sqlspec.Table{
+			{
+				Name:   "tbl",
+				Schema: &schemahcl.Ref{V: "$schema.sch"},
+			},
+			{
+				Name:   "target_table",
+				Schema: &schemahcl.Ref{V: "$schema.target_schema"},
+			},
+		}
+	)
+	require.NoError(t, specutil.QualifyReferencedTables(tables, realm))
+	require.Zero(t, tables[0].Qualifier)
+	require.Equal(t, "target_schema", tables[1].Qualifier)
+}
+
 func TestParseType_Decimal(t *testing.T) {
 	for _, tt := range []struct {
 		input   string
diff --git a/sql/postgres/migrate_test.go b/sql/postgres/migrate_test.go
index 5ed43ca6..5c385f76 100644
--- a/sql/postgres/migrate_test.go
+++ b/sql/postgres/migrate_test.go
@@ -20,6 +20,7 @@ import (
 func TestPlanChanges(t *testing.T) {
 	tests := []struct {
 		changes  []schema.Change
+		options  []migrate.PlanOption
 		mock     func(mock)
 		wantPlan *migrate.Plan
 		wantErr  bool
@@ -551,10 +552,10 @@ func TestPlanChanges(t *testing.T) {
 				}(),
 			},
 			mock: func(m mock) {
-				m.ExpectQuery(sqltest.Escape("SELECT * FROM pg_type t JOIN pg_namespace n on t.typnamespace = n.oid WHERE t.typname = $1 AND n.nspname = $2 AND t.typtype = 'e'")).
+				m.ExpectQuery(sqltest.Escape("SELECT * FROM pg_type t JOIN pg_namespace n on t.typnamespace = n.oid WHERE t.typname = $1 AND t.typtype = 'e' AND n.nspname = $2 ")).
 					WithArgs("state", "public").
 					WillReturnRows(sqlmock.NewRows([]string{"name"}))
-				m.ExpectQuery(sqltest.Escape("SELECT * FROM pg_type t JOIN pg_namespace n on t.typnamespace = n.oid WHERE t.typname = $1 AND n.nspname = $2 AND t.typtype = 'e'")).
+				m.ExpectQuery(sqltest.Escape("SELECT * FROM pg_type t JOIN pg_namespace n on t.typnamespace = n.oid WHERE t.typname = $1 AND t.typtype = 'e' AND n.nspname = $2 ")).
 					WithArgs("status", "test").
 					WillReturnRows(sqlmock.NewRows([]string{"name"}))
 			},
@@ -996,6 +997,117 @@ func TestPlanChanges(t *testing.T) {
 				},
 			},
 		},
+		// Empty qualifier.
+		{
+			changes: []schema.Change{
+				&schema.AddTable{
+					T: schema.NewTable("posts").
+						SetSchema(schema.New("test1")).
+						AddColumns(
+							schema.NewEnumColumn("c1", schema.EnumName("enum"), schema.EnumValues("a"), schema.EnumSchema(schema.New("test2"))),
+						),
+				},
+			},
+			options: []migrate.PlanOption{
+				func(o *migrate.PlanOptions) { o.SchemaQualifier = new(string) },
+			},
+			mock: func(m mock) {
+				m.ExpectQuery(sqltest.Escape("SELECT * FROM pg_type t JOIN pg_namespace n on t.typnamespace = n.oid WHERE t.typname = $1 AND t.typtype = 'e'")).
+					WithArgs("enum").
+					WillReturnRows(sqlmock.NewRows([]string{"name"}))
+			},
+			wantPlan: &migrate.Plan{
+				Reversible:    true,
+				Transactional: true,
+				Changes: []*migrate.Change{
+					{
+						Cmd:     `CREATE TYPE "enum" AS ENUM ('a')`,
+						Reverse: `DROP TYPE "enum"`,
+					},
+					{
+						Cmd:     `CREATE TABLE "posts" ("c1" "enum" NOT NULL)`,
+						Reverse: `DROP TABLE "posts"`,
+					},
+				},
+			},
+		},
+		// Empty sequence qualifier.
+		{
+			changes: []schema.Change{
+				&schema.ModifyTable{
+					T: schema.NewTable("posts").
+						SetSchema(schema.New("public")).
+						AddColumns(
+							schema.NewColumn("c1").SetType(&SerialType{T: "serial"}),
+						),
+					Changes: schema.Changes{
+						&schema.ModifyColumn{
+							From:   schema.NewIntColumn("c1", "integer"),
+							To:     schema.NewColumn("c1").SetType(&SerialType{T: "serial"}),
+							Change: schema.ChangeType,
+						},
+					},
+				},
+			},
+			options: []migrate.PlanOption{
+				func(o *migrate.PlanOptions) { o.SchemaQualifier = new(string) },
+			},
+			wantPlan: &migrate.Plan{
+				Reversible:    true,
+				Transactional: true,
+				Changes: []*migrate.Change{
+					{
+						Cmd:     `CREATE SEQUENCE IF NOT EXISTS "posts_c1_seq" OWNED BY "posts"."c1"`,
+						Reverse: `DROP SEQUENCE IF EXISTS "posts_c1_seq"`,
+					},
+					{
+						Cmd:     `ALTER TABLE "posts" ALTER COLUMN "c1" SET DEFAULT nextval('"posts_c1_seq"')`,
+						Reverse: `ALTER TABLE "posts" ALTER COLUMN "c1" DROP DEFAULT`,
+					},
+				},
+			},
+		},
+		// Empty index qualifier.
+		{
+			changes: []schema.Change{
+				&schema.ModifyTable{
+					T: schema.NewTable("posts").
+						SetSchema(schema.New("public")).
+						AddColumns(
+							schema.NewIntColumn("c", "int"),
+						),
+					Changes: schema.Changes{
+						&schema.AddIndex{
+							I: schema.NewIndex("i").AddColumns(schema.NewIntColumn("c", "int")),
+						},
+					},
+				},
+			},
+			options: []migrate.PlanOption{
+				func(o *migrate.PlanOptions) { o.SchemaQualifier = new(string) },
+			},
+			wantPlan: &migrate.Plan{
+				Reversible:    true,
+				Transactional: true,
+				Changes: []*migrate.Change{
+					{
+						Cmd:     `CREATE INDEX "i" ON "posts" ("c")`,
+						Reverse: `DROP INDEX "i"`,
+					},
+				},
+			},
+		},
+		// Empty qualifier in multi-schema mode should fail.
+		{
+			changes: []schema.Change{
+				&schema.AddTable{T: schema.NewTable("t1").SetSchema(schema.New("s1")).AddColumns(schema.NewIntColumn("a", "int"))},
+				&schema.AddTable{T: schema.NewTable("t2").SetSchema(schema.New("s2")).AddColumns(schema.NewIntColumn("a", "int"))},
+			},
+			options: []migrate.PlanOption{
+				func(o *migrate.PlanOptions) { o.SchemaQualifier = new(string) },
+			},
+			wantErr: true,
+		},
 	}
 	for i, tt := range tests {
 		t.Run(strconv.Itoa(i), func(t *testing.T) {
@@ -1008,7 +1120,7 @@ func TestPlanChanges(t *testing.T) {
 			}
 			drv, err := Open(db)
 			require.NoError(t, err)
-			plan, err := drv.PlanChanges(context.Background(), "wantPlan", tt.changes)
+			plan, err := drv.PlanChanges(context.Background(), "wantPlan", tt.changes, tt.options...)
 			if tt.wantErr {
 				require.Error(t, err, "expect plan to fail")
 				return
diff --git a/sql/postgres/sqlspec_test.go b/sql/postgres/sqlspec_test.go
index 406d7a31..f04de979 100644
--- a/sql/postgres/sqlspec_test.go
+++ b/sql/postgres/sqlspec_test.go
@@ -9,8 +9,11 @@ import (
 	"strconv"
 	"testing"
 
+	"ariga.io/atlas/schemahcl"
 	"ariga.io/atlas/sql/internal/spectest"
+	"ariga.io/atlas/sql/internal/specutil"
 	"ariga.io/atlas/sql/schema"
+	"ariga.io/atlas/sql/sqlspec"
 	"github.com/stretchr/testify/require"
 )
 
@@ -1264,3 +1267,38 @@ func TestRegistrySanity(t *testing.T) {
 func TestInputVars(t *testing.T) {
 	spectest.TestInputVars(t, EvalHCL)
 }
+
+func TestQualifyReferencedTables(t *testing.T) {
+	var (
+		col       = schema.NewIntColumn("col", "integer")
+		targetCol = schema.NewIntColumn("target_column", "integer")
+		targetTbl = schema.NewTable("target_table").AddColumns(targetCol)
+		realm     = schema.NewRealm(
+			schema.New("target_schema").AddTables(targetTbl),
+			schema.New("sch").
+				AddTables(
+					schema.NewTable("tbl").
+						AddColumns(col).
+						AddForeignKeys(
+							schema.NewForeignKey("col_fk").
+								SetRefTable(targetTbl).
+								AddColumns(col).
+								AddRefColumns(targetCol),
+						),
+				),
+		)
+		tables = []*sqlspec.Table{
+			{
+				Name:   "tbl",
+				Schema: &schemahcl.Ref{V: "$schema.sch"},
+			},
+			{
+				Name:   "target_table",
+				Schema: &schemahcl.Ref{V: "$schema.target_schema"},
+			},
+		}
+	)
+	require.NoError(t, specutil.QualifyReferencedTables(tables, realm))
+	require.Zero(t, tables[0].Qualifier)
+	require.Equal(t, "target_schema", tables[1].Qualifier)
+}
diff --git a/sql/sqlite/migrate_test.go b/sql/sqlite/migrate_test.go
index 78b6516f..d98df146 100644
--- a/sql/sqlite/migrate_test.go
+++ b/sql/sqlite/migrate_test.go
@@ -21,6 +21,7 @@ import (
 func TestPlanChanges(t *testing.T) {
 	tests := []struct {
 		changes []schema.Change
+		options []migrate.PlanOption
 		mock    func(mock)
 		plan    *migrate.Plan
 	}{
@@ -337,6 +338,44 @@ func TestPlanChanges(t *testing.T) {
 				},
 			},
 		},
+		// The default is no qualifier.
+		{
+			changes: []schema.Change{
+				&schema.AddTable{T: schema.NewTable("t").SetSchema(schema.New("main")).AddColumns(schema.NewIntColumn("a", "int"))},
+			},
+			plan: &migrate.Plan{
+				Reversible:    true,
+				Transactional: true,
+				Changes: []*migrate.Change{
+					{
+						Cmd:     "CREATE TABLE `t` (`a` int NOT NULL)",
+						Reverse: "DROP TABLE `t`",
+					},
+				},
+			},
+		},
+		// Custom qualifier.
+		{
+			changes: []schema.Change{
+				&schema.AddTable{T: schema.NewTable("t").SetSchema(schema.New("d")).AddColumns(schema.NewIntColumn("a", "int"))},
+			},
+			options: []migrate.PlanOption{
+				func(o *migrate.PlanOptions) {
+					s := "other"
+					o.SchemaQualifier = &s
+				},
+			},
+			plan: &migrate.Plan{
+				Reversible:    true,
+				Transactional: true,
+				Changes: []*migrate.Change{
+					{
+						Cmd:     "CREATE TABLE `other`.`t` (`a` int NOT NULL)",
+						Reverse: "DROP TABLE `other`.`t`",
+					},
+				},
+			},
+		},
 	}
 	for i, tt := range tests {
 		t.Run(strconv.Itoa(i), func(t *testing.T) {
@@ -349,7 +388,7 @@ func TestPlanChanges(t *testing.T) {
 			}
 			drv, err := Open(db)
 			require.NoError(t, err)
-			plan, err := drv.PlanChanges(context.Background(), "plan", tt.changes)
+			plan, err := drv.PlanChanges(context.Background(), "plan", tt.changes, tt.options...)
 			require.NoError(t, err)
 			require.Equal(t, tt.plan.Reversible, plan.Reversible)
 			require.Equal(t, tt.plan.Transactional, plan.Transactional)