Repository: ariga/atlas
Pull request #1035 (2022-08-09 19:37:06)

sql/migrate: skip adding table qualifiers when connection/work is scoped to a schema
This is almost done.

Changes to be made in this PR:
- Add tests for `migrate diff` in MySQL/PostgreSQL 

Changes to be added in the next PRs:
- Add additional validation on the driver side; disallow using this feature programmatically, when the Atlas schema is not scoped to 1 database schema.
- Accept additional flag to allow changing the schema.
- Add docs and examples for Ent and golang-migrate.


Patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate.go b/cmd/atlas/internal/cmdapi/migrate.go
index 2bb3a127..c742691b 100644
--- a/cmd/atlas/internal/cmdapi/migrate.go
+++ b/cmd/atlas/internal/cmdapi/migrate.go
@@ -429,7 +429,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(""))
 	}
 	// Plan the changes and create a new migration file.
 	pl := migrate.NewPlanner(dev.Driver, dir, opts...)
@@ -437,8 +438,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
@@ -601,7 +608,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/internal/integration/go.mod b/internal/integration/go.mod
index e02b702e..cf9339cd 100644
--- a/internal/integration/go.mod
+++ b/internal/integration/go.mod
@@ -1,6 +1,6 @@
 module ariga.io/atlas/internal/integration
 
-go 1.17
+go 1.19
 
 replace ariga.io/atlas => ../../
 
@@ -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/sqlx/plan.go b/sql/internal/sqlx/plan.go
index 44cd4030..6be2ec43 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
 	}
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..48e03f8a 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.
@@ -663,10 +697,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) {
+// 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) {
 	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)
@@ -690,9 +722,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..c2d4bae1 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,6 +61,7 @@ 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
@@ -97,7 +101,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 +117,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 +145,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 +189,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 +208,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 +254,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 +263,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 +336,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 +469,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 +703,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..561ba016 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
@@ -95,7 +99,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 +107,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 +136,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 +177,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 +186,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 +248,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 +284,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 +295,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 +303,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 +333,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 +357,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 +464,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 +515,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 +589,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 +605,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 +627,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 +636,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 +647,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 +684,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 +692,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 +711,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 +721,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 +729,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 +744,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 +759,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 +777,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 +791,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 +812,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 +980,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 +1092,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 +1102,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/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/internal/integration/postgres_test.go b/internal/integration/postgres_test.go
index 6d028fb0..f4b858ff 100644
--- a/internal/integration/postgres_test.go
+++ b/internal/integration/postgres_test.go
@@ -9,6 +9,9 @@ import (
 	"database/sql"
 	"fmt"
 	"log"
+	"os"
+	"os/exec"
+	"path/filepath"
 	"strings"
 	"sync"
 	"testing"
@@ -684,6 +687,59 @@ 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() string {
+			out, err := exec.Command(
+				bin, "migrate", "diff",
+				"--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())
+
+		// 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())
+	})
+}
+
 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 5e409468..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"
@@ -645,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-diff.txt b/internal/integration/testdata/mysql/cli-migrate-diff.txt
index b485335e..d59a9b77 100644
--- a/internal/integration/testdata/mysql/cli-migrate-diff.txt
+++ b/internal/integration/testdata/mysql/cli-migrate-diff.txt
@@ -33,7 +33,7 @@ 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;
 
 -- 2.hcl --
 schema "script_cli_migrate_diff" {}
@@ -59,16 +59,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-diff.txt b/internal/integration/testdata/postgres/cli-migrate-diff.txt
index 92568087..342da9ec 100644
--- a/internal/integration/testdata/postgres/cli-migrate-diff.txt
+++ b/internal/integration/testdata/postgres/cli-migrate-diff.txt
@@ -31,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" {}
@@ -55,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/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/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..c15837d7 100644
--- a/sql/migrate/migrate_test.go
+++ b/sql/migrate/migrate_test.go
@@ -87,7 +87,7 @@ 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{}}
 		ctx = context.Background()
@@ -96,31 +96,31 @@ func TestPlanner_PlanSchemaMode(t *testing.T) {
 	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)
@@ -133,7 +133,7 @@ func TestExecutor_ReadState(t *testing.T) {
 	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)
@@ -142,7 +142,7 @@ func TestExecutor_ReadState(t *testing.T) {
 
 	// Does not work if locked.
 	drv.locks = map[string]struct{}{"atlas_migration_directory_state": {}}
-	_, err = ex.ReadState(ctx)
+	_, err = ex.Replay(ctx, migrate.RealmConn(drv, nil))
 	require.EqualError(t, err, "sql/migrate: acquiring database lock: lockErr")
 	require.Equal(t, 2, drv.lockCounter)
 	require.Equal(t, 2, drv.unlockCounter)
@@ -152,7 +152,7 @@ func TestExecutor_ReadState(t *testing.T) {
 	// 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)
@@ -487,6 +487,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 +506,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
 }
diff --git a/sql/mysql/migrate_test.go b/sql/mysql/migrate_test.go
index de0a81cc..73dba6b0 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,45 @@ 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`",
+					},
+				},
+			},
+		},
 	}
 	for i, tt := range tests {
 		t.Run(strconv.Itoa(i), func(t *testing.T) {
@@ -890,7 +930,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/postgres/migrate_test.go b/sql/postgres/migrate_test.go
index 5ed43ca6..f85bb68e 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,106 @@ 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"`,
+					},
+				},
+			},
+		},
 	}
 	for i, tt := range tests {
 		t.Run(strconv.Itoa(i), func(t *testing.T) {
@@ -1008,7 +1109,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/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)