Repository: ariga/atlas
Pull request #596 (2022-02-23 09:48:27)

sql/TiDB: break down alter statements
TiDB does not support multischema changes (see [issue](https://github.com/pingcap/tidb/issues/14766)) - therefore we break down each change to its own `alter` statement.  
It is therefore important that index/foreign key drop occur before their associated columns.  
Also added support for the syntax of FK reference options (e.g "ON DELETE CASCADE")

hedwigz: For benchmark - running the `integration/mysql_test` I get 99 passes and 33 fail

Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 49005a11..27d527b6 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -209,6 +209,10 @@ jobs:
           --health-interval 10s
           --health-timeout 5s
           --health-retries 5
+      tidb5:
+        image: pingcap/tidb:v5.4.0
+        ports:
+          - 4309:4000
     steps:
       - uses: actions/checkout@v2.3.4
       - uses: actions/setup-go@v2
diff --git a/internal/integration/docker-compose.yaml b/internal/integration/docker-compose.yaml
index 220f89aa..1c24aad7 100644
--- a/internal/integration/docker-compose.yaml
+++ b/internal/integration/docker-compose.yaml
@@ -128,9 +128,3 @@ services:
     image: pingcap/tidb:v5.4.0
     ports:
       - 4309:4000
-
-  tidb_latest:
-    platform: linux/amd64
-    image: pingcap/tidb:latest
-    ports:
-      - 4310:4000
diff --git a/sql/mysql/driver.go b/sql/mysql/driver.go
index 6f3a2ba3..4665fbc0 100644
--- a/sql/mysql/driver.go
+++ b/sql/mysql/driver.go
@@ -49,9 +49,9 @@ func Open(db schema.ExecQuerier) (*Driver, error) {
 	if c.tidb() {
 		return &Driver{
 			conn:        c,
-			Differ:      &sqlx.Diff{DiffDriver: &diff{c}},
+			Differ:      &sqlx.Diff{DiffDriver: &tdiff{diff{c}}},
 			Inspector:   &tinspect{inspect{c}},
-			PlanApplier: &planApply{c},
+			PlanApplier: &tplanApply{planApply{c}},
 		}, nil
 	}
 	return &Driver{
diff --git a/sql/mysql/migrate.go b/sql/mysql/migrate.go
index 4377edff..589b38c6 100644
--- a/sql/mysql/migrate.go
+++ b/sql/mysql/migrate.go
@@ -95,10 +95,11 @@ 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").Ident(c.S.Name)
+			b := Build("CREATE DATABASE")
 			if sqlx.Has(c.Extra, &schema.IfNotExists{}) {
 				b.P("IF NOT EXISTS")
 			}
+			b.Ident(c.S.Name)
 			// Schema was created with CHARSET and it is not the default database character set.
 			if a := (schema.Charset{}); sqlx.Has(c.S.Attrs, &a) && a.V != "" && a.V != s.charset {
 				b.P("CHARSET", a.V)
@@ -114,10 +115,11 @@ func (s *state) topLevel(changes []schema.Change) ([]schema.Change, error) {
 				Comment: fmt.Sprintf("add new schema named %q", c.S.Name),
 			})
 		case *schema.DropSchema:
-			b := Build("DROP DATABASE").Ident(c.S.Name)
+			b := Build("DROP DATABASE")
 			if sqlx.Has(c.Extra, &schema.IfExists{}) {
 				b.P("IF EXISTS")
 			}
+			b.Ident(c.S.Name)
 			s.append(&migrate.Change{
 				Cmd:     b.String(),
 				Source:  c,
@@ -200,11 +202,12 @@ func (s *state) modifySchema(modify *schema.ModifySchema) error {
 func (s *state) addTable(add *schema.AddTable) error {
 	var (
 		errors []string
-		b      = Build("CREATE TABLE").Table(add.T)
+		b      = Build("CREATE TABLE")
 	)
 	if sqlx.Has(add.Extra, &schema.IfNotExists{}) {
 		b.P("IF NOT EXISTS")
 	}
+	b.Table(add.T)
 	b.Wrap(func(b *sqlx.Builder) {
 		b.MapComma(add.T.Columns, func(i int, b *sqlx.Builder) {
 			if err := s.column(b, add.T, add.T.Columns[i]); err != nil {
@@ -251,10 +254,11 @@ 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").Table(drop.T)
+	b := Build("DROP TABLE")
 	if sqlx.Has(drop.Extra, &schema.IfExists{}) {
 		b.P("IF EXISTS")
 	}
+	b.Table(drop.T)
 	s.append(&migrate.Change{
 		Cmd:     b.String(),
 		Source:  drop,
diff --git a/sql/mysql/normalize.go b/sql/mysql/normalize.go
new file mode 100644
index 00000000..9b70e748
--- /dev/null
+++ b/sql/mysql/normalize.go
@@ -0,0 +1,101 @@
+// Copyright 2021-present The Atlas Authors. All rights reserved.
+// This source code is licensed under the Apache 2.0 license found
+// in the LICENSE file in the root directory of this source tree.
+
+package mysql
+
+import (
+	"context"
+	"crypto/md5"
+	"fmt"
+	"time"
+
+	"ariga.io/atlas/sql/schema"
+)
+
+// NormalizeRealm returns the normal representation of the given database.
+func (d *Driver) NormalizeRealm(ctx context.Context, r *schema.Realm) (nr *schema.Realm, err error) {
+	for _, s := range r.Schemas {
+		switch s.Name {
+		case "mysql", "information_schema", "performance_schema", "sys":
+			return nil, fmt.Errorf("sql/mysql: normalizing internal schema %q is not supported", s.Name)
+		}
+	}
+	var (
+		twins   = make(map[string]string)
+		changes = make([]schema.Change, 0, len(r.Schemas))
+		reverse = make([]schema.Change, 0, len(r.Schemas))
+		opts    = &schema.InspectRealmOption{
+			Schemas: make([]string, 0, len(r.Schemas)),
+		}
+	)
+	for _, s := range r.Schemas {
+		twin := twinName(s.Name)
+		twins[twin] = s.Name
+		s.Name = twin
+		opts.Schemas = append(opts.Schemas, s.Name)
+		// Skip adding the schema.IfNotExists clause
+		// to fail if the schema exists.
+		st := schema.New(twin).AddAttrs(s.Attrs...)
+		changes = append(changes, &schema.AddSchema{S: st})
+		reverse = append(reverse, &schema.DropSchema{S: st, Extra: []schema.Clause{&schema.IfExists{}}})
+		for _, t := range s.Tables {
+			// If objects are not strongly connected.
+			if t.Schema != s {
+				t.Schema = s
+			}
+			changes = append(changes, &schema.AddTable{T: t})
+		}
+	}
+	patch := func(r *schema.Realm) {
+		for _, s := range r.Schemas {
+			s.Name = twins[s.Name]
+		}
+	}
+	// Delete the twin resources, and return
+	// the source realm to its initial state.
+	defer func() {
+		patch(r)
+		uerr := d.ApplyChanges(ctx, reverse)
+		if err != nil {
+			err = fmt.Errorf("%w: %v", err, uerr)
+		}
+		err = uerr
+	}()
+	if err := d.ApplyChanges(ctx, changes); err != nil {
+		return nil, err
+	}
+	if nr, err = d.InspectRealm(ctx, opts); err != nil {
+		return nil, err
+	}
+	patch(nr)
+	return nr, nil
+}
+
+// NormalizeSchema returns the normal representation of the given database.
+func (d *Driver) NormalizeSchema(ctx context.Context, s *schema.Schema) (*schema.Schema, error) {
+	r := &schema.Realm{}
+	if s.Realm != nil {
+		r.Attrs = s.Realm.Attrs
+	}
+	r.Schemas = append(r.Schemas, s)
+	nr, err := d.NormalizeRealm(ctx, r)
+	if err != nil {
+		return nil, err
+	}
+	ns, ok := nr.Schema(s.Name)
+	if !ok {
+		return nil, fmt.Errorf("sql/mysql: missing normalized schema %q", s.Name)
+	}
+	return ns, nil
+}
+
+const maxLen = 64
+
+func twinName(name string) string {
+	twin := fmt.Sprintf("atlas_twin_%s_%d", name, time.Now().Unix())
+	if len(twin) <= maxLen {
+		return twin
+	}
+	return fmt.Sprintf("%s_%x", twin[:maxLen-33], md5.Sum([]byte(twin)))
+}
diff --git a/sql/mysql/tidb.go b/sql/mysql/tidb.go
index bac3e845..660e5b38 100644
--- a/sql/mysql/tidb.go
+++ b/sql/mysql/tidb.go
@@ -8,14 +8,118 @@ import (
 	"context"
 	"fmt"
 	"regexp"
+	"sort"
 	"strings"
 
 	"ariga.io/atlas/sql/internal/sqlx"
+	"ariga.io/atlas/sql/migrate"
 	"ariga.io/atlas/sql/schema"
 )
 
-type tinspect struct {
-	inspect
+type (
+	// tplanApply decorates MySQL planApply.
+	tplanApply struct{ planApply }
+	// tdiff decorates MySQL diff.
+	tdiff struct{ diff }
+	// tinspect decorates MySQL inspect.
+	tinspect struct{ inspect }
+)
+
+// priority computes the priority of each change.
+//
+// TiDB does not support multischema ALTERs (i.e. multiple changes in a single ALTER statement).
+// Therefore, we have to break down each alter. This function helps order the ALTERs so they work.
+// e.g. priority gives precedence to DropForeignKey over DropColumn, because a column cannot be
+// dropped if its foreign key was not dropped before.
+func priority(change schema.Change) int {
+	switch c := change.(type) {
+	case *schema.ModifyTable:
+		// each modifyTable should have a single change since we apply `flat` before we sort.
+		return priority(c.Changes[0])
+	case *schema.ModifySchema:
+		// each modifyTable should have a single change since we apply `flat` before we sort.
+		return priority(c.Changes[0])
+	case *schema.DropIndex, *schema.DropForeignKey, *schema.DropAttr, *schema.DropCheck:
+		return 1
+	case *schema.ModifyIndex, *schema.ModifyForeignKey:
+		return 2
+	default:
+		return 3
+	}
+}
+
+// flat takes a list of changes and breaks them down to single atomic changes (e.g: no ModifyTable
+// with multiple AddColumn inside it). Note that, the only "changes" that include sub-changes are
+// `ModifyTable` and `ModifySchema`.
+func flat(changes []schema.Change) []schema.Change {
+	var flat []schema.Change
+	for _, change := range changes {
+		switch m := change.(type) {
+		case *schema.ModifyTable:
+			for _, c := range m.Changes {
+				flat = append(flat, &schema.ModifyTable{
+					T:       m.T,
+					Changes: []schema.Change{c},
+				})
+			}
+		case *schema.ModifySchema:
+			for _, c := range m.Changes {
+				flat = append(flat, &schema.ModifySchema{
+					S:       m.S,
+					Changes: []schema.Change{c},
+				})
+			}
+		default:
+			flat = append(flat, change)
+		}
+	}
+	return flat
+}
+
+// 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) {
+	fc := flat(changes)
+	sort.SliceStable(fc, func(i, j int) bool {
+		return priority(fc[i]) < priority(fc[j])
+	})
+	s := &state{
+		conn: p.conn,
+		Plan: migrate.Plan{
+			Name: name,
+			// A plan is reversible, if all
+			// its changes are reversible.
+			Reversible:    true,
+			Transactional: false,
+		},
+	}
+	for _, c := range fc {
+		// Use the planner of MySQL with each "atomic" chanage.
+		plan, err := p.planApply.PlanChanges(ctx, name, []schema.Change{c})
+		if err != nil {
+			return nil, err
+		}
+		if !plan.Reversible {
+			s.Plan.Reversible = false
+		}
+		s.Plan.Changes = append(s.Plan.Changes, plan.Changes...)
+	}
+	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 (i *tinspect) InspectSchema(ctx context.Context, name string, opts *schema.InspectOptions) (*schema.Schema, error) {
@@ -58,7 +162,8 @@ func (i *tinspect) patchSchema(ctx context.Context, s *schema.Schema) (*schema.S
 }
 
 // e.g CONSTRAINT "" FOREIGN KEY ("foo_id") REFERENCES "foo" ("id")
-var reFK = regexp.MustCompile("(?i)CONSTRAINT\\s+[\"`]*(\\w+)[\"`]*\\s+FOREIGN\\s+KEY\\s*\\(([,\"` \\w]+)\\)\\s+REFERENCES\\s+[\"`]*(\\w+)[\"`]*\\s*\\(([,\"` \\w]+)\\)")
+var reFK = regexp.MustCompile("(?i)CONSTRAINT\\s+[\"`]*(\\w+)[\"`]*\\s+FOREIGN\\s+KEY\\s*\\(([,\"` \\w]+)\\)\\s+REFERENCES\\s+[\"`]*(\\w+)[\"`]*\\s*\\(([,\"` \\w]+)\\).*")
+var reActions = regexp.MustCompile(fmt.Sprintf("(?i)(ON)\\s+(UPDATE|DELETE)\\s+(%s|%s|%s|%s|%s)", schema.NoAction, schema.Restrict, schema.SetNull, schema.SetDefault, schema.Cascade))
 
 func (i *tinspect) setFKs(s *schema.Schema, t *schema.Table) error {
 	var c CreateStmt
@@ -69,13 +174,22 @@ func (i *tinspect) setFKs(s *schema.Schema, t *schema.Table) error {
 		if len(m) != 5 {
 			return fmt.Errorf("unexpected number of matches for a table constraint: %q", m)
 		}
-		ctName, clmns, refTableName, refClmns := m[1], m[2], m[3], m[4]
+		stmt, ctName, clmns, refTableName, refClmns := m[0], m[1], m[2], m[3], m[4]
 		fk := &schema.ForeignKey{
 			Symbol: ctName,
 			Table:  t,
-			// There is no support in TiDB for FKs so inherently there are no actions on update/delete.
-			OnUpdate: schema.NoAction,
-			OnDelete: schema.NoAction,
+		}
+		actions := reActions.FindAllStringSubmatch(stmt, 2)
+		for _, actionMatches := range actions {
+			actionType, actionOp := actionMatches[2], actionMatches[3]
+			switch actionType {
+			case "UPDATE":
+				fk.OnUpdate = schema.ReferenceOption(actionOp)
+			case "DELETE":
+				fk.OnDelete = schema.ReferenceOption(actionOp)
+			default:
+				return fmt.Errorf("action type %s is none of 'UPDATE'/'DELETE'", actionType)
+			}
 		}
 		refTable, ok := s.Table(refTableName)
 		if !ok {
diff --git a/sql/postgres/inspect.go b/sql/postgres/inspect.go
index 489ece46..7297c8ae 100644
--- a/sql/postgres/inspect.go
+++ b/sql/postgres/inspect.go
@@ -681,6 +681,16 @@ type (
 		schema.Attr
 		Columns []string
 	}
+
+	// Cascade clause for appending to the different DROP changes.
+	// For example, A schema.DropSchema with this clause will be
+	// applied as follows:
+	//
+	//	DROP SCHEMA <name> CASCADE
+	//
+	Cascade struct {
+		schema.Clause
+	}
 )
 
 const (
@@ -701,7 +711,7 @@ SELECT
 	pg_catalog.obj_description(t2.oid, 'pg_class') AS COMMENT
 FROM
 	INFORMATION_SCHEMA.TABLES AS t1
-	JOIN pg_catalog.pg_class AS t2 ON t2.oid = to_regclass(t1.table_schema || '.' || t1.table_name)::oid
+	JOIN pg_catalog.pg_class AS t2 ON t2.oid = to_regclass(t1.table_schema || '.' || quote_ident(t1.table_name))::oid
 WHERE
 	t1.table_type = 'BASE TABLE'
 	AND t1.table_schema IN (%s)
@@ -715,7 +725,7 @@ SELECT
 	pg_catalog.obj_description(t2.oid, 'pg_class') AS COMMENT
 FROM
 	INFORMATION_SCHEMA.TABLES AS t1
-	JOIN pg_catalog.pg_class AS t2 ON t2.oid = to_regclass(t1.table_schema || '.' || t1.table_name)::oid
+	JOIN pg_catalog.pg_class AS t2 ON t2.oid = to_regclass(t1.table_schema || '.' || quote_ident(t1.table_name))::oid
 WHERE
 	t1.table_type = 'BASE TABLE'
 	AND t1.table_schema IN (%s)
diff --git a/sql/postgres/migrate.go b/sql/postgres/migrate.go
index ee9119ca..7a7f258b 100644
--- a/sql/postgres/migrate.go
+++ b/sql/postgres/migrate.go
@@ -86,10 +86,11 @@ func (s *state) topLevel(changes []schema.Change) []schema.Change {
 	for _, c := range changes {
 		switch c := c.(type) {
 		case *schema.AddSchema:
-			b := Build("CREATE SCHEMA").Ident(c.S.Name)
+			b := Build("CREATE SCHEMA")
 			if sqlx.Has(c.Extra, &schema.IfNotExists{}) {
 				b.P("IF NOT EXISTS")
 			}
+			b.Ident(c.S.Name)
 			s.append(&migrate.Change{
 				Cmd:     b.String(),
 				Source:  c,
@@ -97,9 +98,13 @@ func (s *state) topLevel(changes []schema.Change) []schema.Change {
 				Comment: fmt.Sprintf("Add new schema named %q", c.S.Name),
 			})
 		case *schema.DropSchema:
-			b := Build("DROP SCHEMA").Ident(c.S.Name)
+			b := Build("DROP SCHEMA")
 			if sqlx.Has(c.Extra, &schema.IfExists{}) {
-				b.P("IF NOT EXISTS")
+				b.P("IF EXISTS")
+			}
+			b.Ident(c.S.Name)
+			if sqlx.Has(c.Extra, &Cascade{}) {
+				b.P("CASCADE")
 			}
 			s.append(&migrate.Change{
 				Cmd:     b.String(),
@@ -119,10 +124,11 @@ func (s *state) addTable(ctx context.Context, add *schema.AddTable) error {
 	if err := s.addTypes(ctx, add.T.Columns...); err != nil {
 		return err
 	}
-	b := Build("CREATE TABLE").Table(add.T)
+	b := Build("CREATE TABLE")
 	if sqlx.Has(add.Extra, &schema.IfNotExists{}) {
 		b.P("IF NOT EXISTS")
 	}
+	b.Table(add.T)
 	b.Wrap(func(b *sqlx.Builder) {
 		b.MapComma(add.T.Columns, func(i int, b *sqlx.Builder) {
 			s.column(b, add.T.Columns[i])
@@ -155,10 +161,11 @@ 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").Table(drop.T)
+	b := Build("DROP TABLE")
 	if sqlx.Has(drop.Extra, &schema.IfExists{}) {
 		b.P("IF EXISTS")
 	}
+	b.Table(drop.T)
 	s.append(&migrate.Change{
 		Cmd:     b.String(),
 		Source:  drop,
diff --git a/sql/schema/inspect.go b/sql/schema/inspect.go
index 2ad0deef..930cb6b7 100644
--- a/sql/schema/inspect.go
+++ b/sql/schema/inspect.go
@@ -64,3 +64,15 @@ type (
 		InspectRealm(ctx context.Context, opts *InspectRealmOption) (*Realm, error)
 	}
 )
+
+// Normalizer is the interface implemented by the different database drivers for
+// "normalizing" schema objects. i.e. converting schema objects defined in natural
+// form to their representation in the database. Thus, two schema objects are equal
+// if their normal forms are equal.
+type Normalizer interface {
+	// NormalizeSchema returns the normal representation of a schema.
+	NormalizeSchema(context.Context, *Schema) (*Schema, error)
+
+	// NormalizeRealm returns the normal representation of a database.
+	NormalizeRealm(context.Context, *Realm) (*Realm, error)
+}


Test patch:
diff --git a/internal/integration/integration_test.go b/internal/integration/integration_test.go
index 90a37d92..9ffc0b7b 100644
--- a/internal/integration/integration_test.go
+++ b/internal/integration/integration_test.go
@@ -404,7 +404,7 @@ func ensureNoChange(t T, tables ...*schema.Table) {
 		tt, ok := realm.Schemas[0].Table(tables[i].Name)
 		require.True(t, ok)
 		changes := t.diff(tt, tables[i])
-		require.Empty(t, changes)
+		require.Emptyf(t, changes, "changes should be empty for table %s, but instead was %#v", tt.Name , changes)
 	}
 }
 
diff --git a/internal/integration/mysql_test.go b/internal/integration/mysql_test.go
index bca2796d..f1f43e1c 100644
--- a/internal/integration/mysql_test.go
+++ b/internal/integration/mysql_test.go
@@ -37,11 +37,7 @@ func myRun(t *testing.T, fn func(*myTest)) {
 	myTests.Do(func() {
 		myTests.drivers = make(map[string]*myTest)
 		for version, port := range map[string]int{"56": 3306, "57": 3307, "8": 3308, "Maria107": 4306, "Maria102": 4307, "Maria103": 4308} {
-			password := ":pass"
-			if version == "TiDB" {
-				password = ""
-			}
-			db, err := sql.Open("mysql", fmt.Sprintf("root%s@tcp(localhost:%d)/test?parseTime=True", password, port))
+			db, err := sql.Open("mysql", fmt.Sprintf("root:pass@tcp(localhost:%d)/test?parseTime=True", port))
 			require.NoError(t, err)
 			drv, err := mysql.Open(db)
 			require.NoError(t, err)
@@ -1279,7 +1275,7 @@ func (t *myTest) defaultAttrs() []schema.Attr {
 		collation = "latin1_swedish_ci"
 	)
 	switch {
-	case t.version == "TiDB":
+	case strings.Contains(t.version, "TiDB"):
 		charset = "utf8mb4"
 		collation = "utf8mb4_bin"
 	case t.version == "8":
diff --git a/internal/integration/tidb_test.go b/internal/integration/tidb_test.go
new file mode 100644
index 00000000..e3fc7d85
--- /dev/null
+++ b/internal/integration/tidb_test.go
@@ -0,0 +1,431 @@
+// Copyright 2021-present The Atlas Authors. All rights reserved.
+// This source code is licensed under the Apache 2.0 license found
+// in the LICENSE file in the root directory of this source tree.
+
+package integration
+
+import (
+	"database/sql"
+	"fmt"
+	"testing"
+
+	"ariga.io/atlas/sql/mysql"
+	"ariga.io/atlas/sql/schema"
+
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/stretchr/testify/require"
+)
+
+func tidbRun(t *testing.T, fn func(*myTest)) {
+	port := 4309
+	db, err := sql.Open("mysql", fmt.Sprintf("root@tcp(localhost:%d)/test?parseTime=True", port))
+	require.NoError(t, err)
+	drv, err := mysql.Open(db)
+	require.NoError(t, err)
+	t.Run("TiDB5", func(t *testing.T) {
+		tt := &myTest{T: t, db: db, drv: drv, version: "TiDB5", port: port}
+		fn(tt)
+	})
+}
+
+func TestTiDB_AddDropTable(t *testing.T) {
+	tidbRun(t, func(t *myTest) {
+		testAddDrop(t)
+	})
+}
+
+func TestTiDB_Relation(t *testing.T) {
+	tidbRun(t, func(t *myTest) {
+		testRelation(t)
+	})
+}
+
+func TestTiDB_AddIndexedColumns(t *testing.T) {
+	tidbRun(t, func(t *myTest) {
+		usersT := &schema.Table{
+			Name:    "users",
+			Columns: []*schema.Column{{Name: "id", Type: &schema.ColumnType{Type: &schema.IntegerType{T: "int"}}}},
+		}
+		t.migrate(&schema.AddTable{T: usersT})
+		t.dropTables(usersT.Name)
+		usersT.Columns = append(usersT.Columns, &schema.Column{
+			Name:    "a",
+			Type:    &schema.ColumnType{Raw: "bigint", Type: &schema.IntegerType{T: "bigint"}, Null: true},
+			Default: &schema.RawExpr{X: "10"},
+		}, &schema.Column{
+			Name:    "b",
+			Type:    &schema.ColumnType{Raw: "bigint", Type: &schema.IntegerType{T: "bigint"}, Null: true},
+			Default: &schema.RawExpr{X: "10"},
+		}, &schema.Column{
+			Name:    "c",
+			Type:    &schema.ColumnType{Raw: "bigint", Type: &schema.IntegerType{T: "bigint"}, Null: true},
+			Default: &schema.RawExpr{X: "10"},
+		})
+		parts := usersT.Columns[len(usersT.Columns)-3:]
+		usersT.Indexes = append(usersT.Indexes, &schema.Index{
+			Unique: true,
+			Name:   "a_b_c_unique",
+			Parts:  []*schema.IndexPart{{C: parts[0]}, {C: parts[1]}, {C: parts[2]}},
+		})
+		changes := t.diff(t.loadUsers(), usersT)
+		require.NotEmpty(t, changes, "usersT contains 2 new columns and 1 new index")
+		t.migrate(&schema.ModifyTable{T: usersT, Changes: changes})
+		ensureNoChange(t, usersT)
+
+		// In MySQL, dropping a column should remove it from the key.
+		// However, on TiDB an explicit DROP/ADD INDEX is required.
+		idx, ok := usersT.Index("a_b_c_unique")
+		require.True(t, ok)
+		idx.Parts = idx.Parts[:len(idx.Parts)-1]
+		usersT.Columns = usersT.Columns[:len(usersT.Columns)-1]
+		changes = t.diff(t.loadUsers(), usersT)
+		t.migrate(&schema.ModifyTable{T: usersT, Changes: changes})
+		ensureNoChange(t, t.loadUsers())
+
+		// Dropping a column from both table and index.
+		usersT = t.loadUsers()
+		idx, ok = usersT.Index("a_b_c_unique")
+		require.True(t, ok)
+		require.Len(t, idx.Parts, 2)
+		usersT.Columns = usersT.Columns[:len(usersT.Columns)-1]
+		idx.Parts = idx.Parts[:len(idx.Parts)-1]
+		changes = t.diff(t.loadUsers(), usersT)
+		require.Len(t, changes, 2)
+		t.migrate(&schema.ModifyTable{T: usersT, Changes: changes})
+		ensureNoChange(t, t.loadUsers())
+
+		// Dropping a column should remove
+		// single-column keys as well.
+		usersT = t.loadUsers()
+		idx, ok = usersT.Index("a_b_c_unique")
+		require.True(t, ok)
+		require.Len(t, idx.Parts, 1)
+		usersT.Columns = usersT.Columns[:len(usersT.Columns)-1]
+		// In MySQL, dropping a column should remove its index.
+		// However, on TiDB an explicit DROP INDEX is required.
+		usersT.Indexes = nil
+		changes = t.diff(t.loadUsers(), usersT)
+		require.Len(t, changes, 2)
+		t.migrate(&schema.ModifyTable{T: usersT, Changes: changes})
+		ensureNoChange(t, t.loadUsers())
+		_, ok = t.loadUsers().Index("a_b_c_unique")
+		require.False(t, ok)
+	})
+}
+
+func TestTiDB_AddColumns(t *testing.T) {
+	tidbRun(t, func(t *myTest) {
+		usersT := t.users()
+		t.dropTables(usersT.Name)
+		t.migrate(&schema.AddTable{T: usersT})
+		usersT.Columns = append(
+			usersT.Columns,
+			&schema.Column{Name: "a", Type: &schema.ColumnType{Raw: "tinyblob", Type: &schema.BinaryType{T: "tinyblob"}}},
+			&schema.Column{Name: "b", Type: &schema.ColumnType{Raw: "mediumblob", Type: &schema.BinaryType{T: "mediumblob"}}},
+			&schema.Column{Name: "c", Type: &schema.ColumnType{Raw: "blob", Type: &schema.BinaryType{T: "blob"}}},
+			&schema.Column{Name: "d", Type: &schema.ColumnType{Raw: "longblob", Type: &schema.BinaryType{T: "longblob"}}},
+			&schema.Column{Name: "e", Type: &schema.ColumnType{Raw: "binary", Type: &schema.BinaryType{T: "binary"}}},
+			&schema.Column{Name: "f", Type: &schema.ColumnType{Raw: "varbinary(255)", Type: &schema.BinaryType{T: "varbinary(255)"}}, Default: &schema.Literal{V: "foo"}},
+			&schema.Column{Name: "g", Type: &schema.ColumnType{Type: &schema.StringType{T: "varchar", Size: 255}}},
+			&schema.Column{Name: "h", Type: &schema.ColumnType{Raw: "varchar(255)", Type: &schema.StringType{T: "varchar(255)"}}},
+			&schema.Column{Name: "i", Type: &schema.ColumnType{Raw: "tinytext", Type: &schema.StringType{T: "tinytext"}}},
+			&schema.Column{Name: "j", Type: &schema.ColumnType{Raw: "mediumtext", Type: &schema.StringType{T: "mediumtext"}}},
+			&schema.Column{Name: "k", Type: &schema.ColumnType{Raw: "text", Type: &schema.StringType{T: "text"}}},
+			&schema.Column{Name: "l", Type: &schema.ColumnType{Raw: "longtext", Type: &schema.StringType{T: "longtext"}}},
+			&schema.Column{Name: "m", Type: &schema.ColumnType{Type: &schema.DecimalType{T: "decimal", Precision: 10, Scale: 6}}},
+			&schema.Column{Name: "m1", Type: &schema.ColumnType{Type: &schema.DecimalType{T: "decimal"}}},
+			&schema.Column{Name: "m2", Type: &schema.ColumnType{Type: &schema.DecimalType{T: "decimal", Precision: 2}}},
+			&schema.Column{Name: "n", Type: &schema.ColumnType{Type: &schema.DecimalType{T: "numeric", Precision: 10, Scale: 2}}},
+			&schema.Column{Name: "n1", Type: &schema.ColumnType{Type: &schema.DecimalType{T: "numeric"}}},
+			&schema.Column{Name: "n2", Type: &schema.ColumnType{Type: &schema.DecimalType{T: "numeric", Precision: 2}}},
+			&schema.Column{Name: "o", Type: &schema.ColumnType{Type: &schema.FloatType{T: "float", Precision: 2}}},
+			&schema.Column{Name: "p", Type: &schema.ColumnType{Type: &schema.FloatType{T: "double", Precision: 14}}},
+			&schema.Column{Name: "q", Type: &schema.ColumnType{Type: &schema.FloatType{T: "real", Precision: 14}}},
+			&schema.Column{Name: "r", Type: &schema.ColumnType{Type: &schema.IntegerType{T: "int"}}},
+			&schema.Column{Name: "s", Type: &schema.ColumnType{Type: &schema.IntegerType{T: "bigint"}}},
+			&schema.Column{Name: "t", Type: &schema.ColumnType{Type: &schema.IntegerType{T: "smallint"}}},
+			&schema.Column{Name: "u", Type: &schema.ColumnType{Type: &schema.EnumType{T: "enum", Values: []string{"a", "b", "c"}}}},
+			&schema.Column{Name: "v", Type: &schema.ColumnType{Type: &schema.StringType{T: "char(36)"}}},
+			&schema.Column{Name: "x", Type: &schema.ColumnType{Type: &schema.SpatialType{T: "line"}}},
+			&schema.Column{Name: "z", Type: &schema.ColumnType{Type: &schema.TimeType{T: "timestamp"}}, Default: &schema.RawExpr{X: "CURRENT_TIMESTAMP"}},
+		)
+		changes := t.diff(t.loadUsers(), usersT)
+		require.Len(t, changes, 27)
+		t.migrate(&schema.ModifyTable{T: usersT, Changes: changes})
+		ensureNoChange(t, usersT)
+	})
+}
+
+func TestTiDB_ColumnInt(t *testing.T) {
+	t.Run("ChangeType", func(t *testing.T) {
+		tidbRun(t, func(t *myTest) {
+			usersT := &schema.Table{
+				Name:    "users",
+				Columns: []*schema.Column{{Name: "a", Type: &schema.ColumnType{Type: &schema.IntegerType{T: "int"}}}},
+			}
+			t.migrate(&schema.AddTable{T: usersT})
+			t.dropTables(usersT.Name)
+			for _, typ := range []string{"tinyint", "smallint", "mediumint", "bigint"} {
+				usersT.Columns[0].Type.Type = &schema.IntegerType{T: typ}
+				changes := t.diff(t.loadUsers(), usersT)
+				require.Len(t, changes, 1)
+				t.migrate(&schema.ModifyTable{T: usersT, Changes: changes})
+				ensureNoChange(t, usersT)
+			}
+		})
+	})
+
+	t.Run("ChangeDefault", func(t *testing.T) {
+		tidbRun(t, func(t *myTest) {
+			usersT := &schema.Table{
+				Name:    "users",
+				Columns: []*schema.Column{{Name: "a", Type: &schema.ColumnType{Type: &schema.IntegerType{T: "int"}}, Default: &schema.RawExpr{X: "1"}}},
+			}
+			t.migrate(&schema.AddTable{T: usersT})
+			t.dropTables(usersT.Name)
+			ensureNoChange(t, usersT)
+			for _, x := range []string{"2", "'3'", "10.1"} {
+				usersT.Columns[0].Default.(*schema.RawExpr).X = x
+				changes := t.diff(t.loadUsers(), usersT)
+				require.Len(t, changes, 1)
+				t.migrate(&schema.ModifyTable{T: usersT, Changes: changes})
+				ensureNoChange(t, usersT)
+			}
+		})
+	})
+}
+
+func TestTiDB_ColumnString(t *testing.T) {
+	t.Run("ChangeType", func(t *testing.T) {
+		tidbRun(t, func(t *myTest) {
+			usersT := &schema.Table{
+				Name:    "users",
+				Columns: []*schema.Column{{Name: "a", Type: &schema.ColumnType{Type: &schema.StringType{T: "varchar(20)"}}}},
+			}
+			t.migrate(&schema.AddTable{T: usersT})
+			t.dropTables(usersT.Name)
+			for _, typ := range []string{"varchar(255)", "char(120)", "tinytext", "mediumtext", "longtext"} {
+				usersT.Columns[0].Type.Type = &schema.StringType{T: typ}
+				changes := t.diff(t.loadUsers(), usersT)
+				require.Len(t, changes, 1)
+				t.migrate(&schema.ModifyTable{T: usersT, Changes: changes})
+				ensureNoChange(t, usersT)
+			}
+		})
+	})
+
+	t.Run("AddWithDefault", func(t *testing.T) {
+		tidbRun(t, func(t *myTest) {
+			usersT := &schema.Table{
+				Name: "users",
+				Columns: []*schema.Column{
+					{Name: "a", Type: &schema.ColumnType{Type: &schema.StringType{T: "varchar(255)"}}, Default: &schema.RawExpr{X: "hello"}},
+					{Name: "b", Type: &schema.ColumnType{Type: &schema.StringType{T: "char(255)"}}, Default: &schema.RawExpr{X: "'world'"}},
+				},
+			}
+			t.migrate(&schema.AddTable{T: usersT})
+			t.dropTables(usersT.Name)
+			ensureNoChange(t, usersT)
+		})
+	})
+
+	t.Run("ChangeDefault", func(t *testing.T) {
+		tidbRun(t, func(t *myTest) {
+			usersT := &schema.Table{
+				Name:    "users",
+				Columns: []*schema.Column{{Name: "a", Type: &schema.ColumnType{Type: &schema.StringType{T: "varchar(255)"}}, Default: &schema.RawExpr{X: "hello"}}},
+			}
+			t.migrate(&schema.AddTable{T: usersT})
+			t.dropTables(usersT.Name)
+			ensureNoChange(t, usersT)
+			for _, x := range []string{"2", "'3'", "'world'"} {
+				usersT.Columns[0].Default.(*schema.RawExpr).X = x
+				changes := t.diff(t.loadUsers(), usersT)
+				require.Len(t, changes, 1)
+				t.migrate(&schema.ModifyTable{T: usersT, Changes: changes})
+				ensureNoChange(t, usersT)
+			}
+		})
+	})
+}
+
+func TestTiDB_ColumnBool(t *testing.T) {
+	t.Run("Add", func(t *testing.T) {
+		tidbRun(t, func(t *myTest) {
+			usersT := &schema.Table{
+				Name: "users",
+				Columns: []*schema.Column{
+					{Name: "a", Type: &schema.ColumnType{Type: &schema.BoolType{T: "bool"}}},
+					{Name: "b", Type: &schema.ColumnType{Type: &schema.BoolType{T: "boolean"}}},
+					{Name: "c", Type: &schema.ColumnType{Type: &schema.BoolType{T: "tinyint"}}},
+					{Name: "d", Type: &schema.ColumnType{Type: &schema.BoolType{T: "tinyint(1)"}}},
+				},
+			}
+			t.migrate(&schema.AddTable{T: usersT})
+			t.dropTables(usersT.Name)
+			ensureNoChange(t, usersT)
+		})
+	})
+
+	t.Run("AddWithDefault", func(t *testing.T) {
+		tidbRun(t, func(t *myTest) {
+			usersT := &schema.Table{
+				Name: "users",
+				Columns: []*schema.Column{
+					{Name: "a", Type: &schema.ColumnType{Type: &schema.BoolType{T: "bool"}}, Default: &schema.RawExpr{X: "1"}},
+					{Name: "b", Type: &schema.ColumnType{Type: &schema.BoolType{T: "bool"}}, Default: &schema.RawExpr{X: "0"}},
+					{Name: "c", Type: &schema.ColumnType{Type: &schema.BoolType{T: "bool"}}, Default: &schema.RawExpr{X: "'1'"}},
+					{Name: "d", Type: &schema.ColumnType{Type: &schema.BoolType{T: "bool"}}, Default: &schema.RawExpr{X: "'0'"}},
+					{Name: "e", Type: &schema.ColumnType{Type: &schema.BoolType{T: "bool"}}, Default: &schema.RawExpr{X: "true"}},
+					{Name: "f", Type: &schema.ColumnType{Type: &schema.BoolType{T: "bool"}}, Default: &schema.RawExpr{X: "false"}},
+					{Name: "g", Type: &schema.ColumnType{Type: &schema.BoolType{T: "bool"}}, Default: &schema.RawExpr{X: "TRUE"}},
+					{Name: "h", Type: &schema.ColumnType{Type: &schema.BoolType{T: "bool"}}, Default: &schema.RawExpr{X: "FALSE"}},
+				},
+			}
+			t.migrate(&schema.AddTable{T: usersT})
+			t.dropTables(usersT.Name)
+			ensureNoChange(t, usersT)
+		})
+	})
+
+	t.Run("ChangeDefault", func(t *testing.T) {
+		tidbRun(t, func(t *myTest) {
+			usersT := &schema.Table{
+				Name: "users",
+				Columns: []*schema.Column{
+					{Name: "a", Type: &schema.ColumnType{Type: &schema.BoolType{T: "bool"}}, Default: &schema.RawExpr{X: "1"}},
+				},
+			}
+			t.migrate(&schema.AddTable{T: usersT})
+			t.dropTables(usersT.Name)
+			ensureNoChange(t, usersT)
+			// Change default from "true" to "false" to "true".
+			for _, x := range []string{"false", "true"} {
+				usersT.Columns[0].Default.(*schema.RawExpr).X = x
+				changes := t.diff(t.loadUsers(), usersT)
+				require.Len(t, changes, 1)
+				t.migrate(&schema.ModifyTable{T: usersT, Changes: changes})
+				ensureNoChange(t, usersT)
+			}
+		})
+	})
+
+	t.Run("ChangeNull", func(t *testing.T) {
+		tidbRun(t, func(t *myTest) {
+			usersT := &schema.Table{
+				Name: "users",
+				Columns: []*schema.Column{
+					{Name: "a", Type: &schema.ColumnType{Type: &schema.BoolType{T: "bool"}, Null: true}},
+				},
+			}
+			t.migrate(&schema.AddTable{T: usersT})
+			t.dropTables(usersT.Name)
+			ensureNoChange(t, usersT)
+			usersT.Columns[0].Type.Null = false
+			changes := t.diff(t.loadUsers(), usersT)
+			require.Len(t, changes, 1)
+			t.migrate(&schema.ModifyTable{T: usersT, Changes: changes})
+			ensureNoChange(t, usersT)
+		})
+	})
+}
+
+func TestTiDB_ForeignKey(t *testing.T) {
+	t.Run("ChangeAction", func(t *testing.T) {
+		tidbRun(t, func(t *myTest) {
+			usersT, postsT := t.users(), t.posts()
+			t.dropTables(postsT.Name, usersT.Name)
+			t.migrate(&schema.AddTable{T: usersT}, &schema.AddTable{T: postsT})
+			ensureNoChange(t, postsT, usersT)
+
+			postsT = t.loadPosts()
+			fk, ok := postsT.ForeignKey("author_id")
+			require.True(t, ok)
+			fk.OnUpdate = schema.SetNull
+			fk.OnDelete = schema.Cascade
+			changes := t.diff(t.loadPosts(), postsT)
+			require.Len(t, changes, 1)
+			modifyF, ok := changes[0].(*schema.ModifyForeignKey)
+			require.True(t, ok)
+			require.True(t, modifyF.Change == schema.ChangeUpdateAction|schema.ChangeDeleteAction)
+
+			t.migrate(&schema.ModifyTable{T: postsT, Changes: changes})
+			ensureNoChange(t, postsT, usersT)
+		})
+	})
+
+	t.Run("UnsetNull", func(t *testing.T) {
+		tidbRun(t, func(t *myTest) {
+			usersT, postsT := t.users(), t.posts()
+			t.dropTables(postsT.Name, usersT.Name)
+			fk, ok := postsT.ForeignKey("author_id")
+			require.True(t, ok)
+			fk.OnDelete = schema.SetNull
+			fk.OnUpdate = schema.SetNull
+			t.migrate(&schema.AddTable{T: usersT}, &schema.AddTable{T: postsT})
+			ensureNoChange(t, postsT, usersT)
+
+			postsT = t.loadPosts()
+			c, ok := postsT.Column("author_id")
+			require.True(t, ok)
+			c.Type.Null = false
+			fk, ok = postsT.ForeignKey("author_id")
+			require.True(t, ok)
+			fk.OnUpdate = schema.NoAction
+			fk.OnDelete = schema.NoAction
+			changes := t.diff(t.loadPosts(), postsT)
+			require.Len(t, changes, 2)
+			modifyC, ok := changes[0].(*schema.ModifyColumn)
+			require.True(t, ok)
+			require.True(t, modifyC.Change == schema.ChangeNull)
+			modifyF, ok := changes[1].(*schema.ModifyForeignKey)
+			require.True(t, ok)
+			require.True(t, modifyF.Change == schema.ChangeUpdateAction|schema.ChangeDeleteAction)
+
+			t.migrate(&schema.ModifyTable{T: postsT, Changes: changes})
+			ensureNoChange(t, postsT, usersT)
+		})
+	})
+
+	t.Run("AddDrop", func(t *testing.T) {
+		tidbRun(t, func(t *myTest) {
+			usersT := t.users()
+			t.dropTables(usersT.Name)
+			t.migrate(&schema.AddTable{T: usersT})
+			ensureNoChange(t, usersT)
+
+			// Add foreign key.
+			usersT.Columns = append(usersT.Columns, &schema.Column{
+				Name: "spouse_id",
+				Type: &schema.ColumnType{Raw: "bigint", Type: &schema.IntegerType{T: "bigint"}, Null: true},
+			})
+			usersT.ForeignKeys = append(usersT.ForeignKeys, &schema.ForeignKey{
+				Symbol:     "spouse_id",
+				Table:      usersT,
+				Columns:    usersT.Columns[len(usersT.Columns)-1:],
+				RefTable:   usersT,
+				RefColumns: usersT.Columns[:1],
+				OnDelete:   schema.NoAction,
+			})
+
+			changes := t.diff(t.loadUsers(), usersT)
+			require.Len(t, changes, 2)
+			addC, ok := changes[0].(*schema.AddColumn)
+			require.True(t, ok)
+			require.Equal(t, "spouse_id", addC.C.Name)
+			addF, ok := changes[1].(*schema.AddForeignKey)
+			require.True(t, ok)
+			require.Equal(t, "spouse_id", addF.F.Symbol)
+			t.migrate(&schema.ModifyTable{T: usersT, Changes: changes})
+			ensureNoChange(t, usersT)
+
+			// Drop foreign keys.
+			usersT.Columns = usersT.Columns[:len(usersT.Columns)-1]
+			usersT.ForeignKeys = usersT.ForeignKeys[:len(usersT.ForeignKeys)-1]
+			changes = t.diff(t.loadUsers(), usersT)
+			require.Len(t, changes, 2)
+			t.migrate(&schema.ModifyTable{T: usersT, Changes: changes})
+			ensureNoChange(t, usersT)
+		})
+	})
+}
diff --git a/sql/mysql/migrate_test.go b/sql/mysql/migrate_test.go
index b24db365..75b844db 100644
--- a/sql/mysql/migrate_test.go
+++ b/sql/mysql/migrate_test.go
@@ -25,9 +25,9 @@ func TestMigrate_ApplyChanges(t *testing.T) {
 		WillReturnResult(sqlmock.NewResult(0, 1))
 	mk.ExpectExec(sqltest.Escape("DROP TABLE `users`")).
 		WillReturnResult(sqlmock.NewResult(0, 0))
-	mk.ExpectExec(sqltest.Escape("DROP TABLE `public`.`pets` IF EXISTS")).
+	mk.ExpectExec(sqltest.Escape("DROP TABLE IF EXISTS `public`.`pets`")).
 		WillReturnResult(sqlmock.NewResult(0, 0))
-	mk.ExpectExec(sqltest.Escape("CREATE TABLE `pets` IF NOT EXISTS (`a` int NOT NULL DEFAULT (int(rand())), `b` bigint NOT NULL DEFAULT 1, `c` bigint NULL, PRIMARY KEY (`a`, `b`), UNIQUE INDEX `b_c_unique` (`b`, `c`) COMMENT \"comment\")")).
+	mk.ExpectExec(sqltest.Escape("CREATE TABLE IF NOT EXISTS `pets` (`a` int NOT NULL DEFAULT (int(rand())), `b` bigint NOT NULL DEFAULT 1, `c` bigint NULL, PRIMARY KEY (`a`, `b`), UNIQUE INDEX `b_c_unique` (`b`, `c`) COMMENT \"comment\")")).
 		WillReturnResult(sqlmock.NewResult(0, 0))
 	mk.ExpectExec(sqltest.Escape("ALTER TABLE `users` DROP INDEX `id_spouse_id`")).
 		WillReturnResult(sqlmock.NewResult(0, 0))
@@ -176,6 +176,24 @@ func TestPlanChanges(t *testing.T) {
 		wantPlan *migrate.Plan
 		wantErr  bool
 	}{
+		{
+			input: []schema.Change{
+				&schema.AddSchema{S: schema.New("test").SetCharset("utf8mb4"), Extra: []schema.Clause{&schema.IfNotExists{}}},
+				&schema.DropSchema{S: schema.New("test").SetCharset("utf8mb4"), Extra: []schema.Clause{&schema.IfExists{}}},
+			},
+			wantPlan: &migrate.Plan{
+				Reversible: false,
+				Changes: []*migrate.Change{
+					{
+						Cmd:     "CREATE DATABASE IF NOT EXISTS `test` CHARSET utf8mb4",
+						Reverse: "DROP DATABASE `test`",
+					},
+					{
+						Cmd: "DROP DATABASE IF EXISTS `test`",
+					},
+				},
+			},
+		},
 		{
 			input: []schema.Change{
 				func() schema.Change {
diff --git a/sql/mysql/normalize_test.go b/sql/mysql/normalize_test.go
new file mode 100644
index 00000000..24aae6a4
--- /dev/null
+++ b/sql/mysql/normalize_test.go
@@ -0,0 +1,62 @@
+// Copyright 2021-present The Atlas Authors. All rights reserved.
+// This source code is licensed under the Apache 2.0 license found
+// in the LICENSE file in the root directory of this source tree.
+
+package mysql
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"ariga.io/atlas/sql/migrate"
+	"ariga.io/atlas/sql/schema"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestDriver_NormalizeRealm(t *testing.T) {
+	var (
+		apply   = &mockApply{}
+		inspect = &mockInspect{
+			realm: schema.NewRealm(schema.New("test").SetCharset("utf8mb4")),
+		}
+		drv = &Driver{
+			Inspector:   inspect,
+			PlanApplier: apply,
+		}
+	)
+	normal, err := drv.NormalizeRealm(context.Background(), schema.NewRealm(schema.New("test")))
+	require.NoError(t, err)
+	require.Equal(t, normal, inspect.realm)
+
+	require.Len(t, inspect.schemas, 1)
+	require.True(t, strings.HasPrefix(inspect.schemas[0], "atlas_twin_test_"))
+
+	require.Len(t, apply.changes, 2, "expect 2 calls (create and drop)")
+	require.Len(t, apply.changes[0], 1)
+	require.Equal(t, &schema.AddSchema{S: schema.New(inspect.schemas[0])}, apply.changes[0][0])
+	require.Len(t, apply.changes[1], 1)
+	require.Equal(t, &schema.DropSchema{S: schema.New(inspect.schemas[0]), Extra: []schema.Clause{&schema.IfExists{}}}, apply.changes[1][0])
+}
+
+type mockInspect struct {
+	schema.Inspector
+	schemas []string
+	realm   *schema.Realm
+}
+
+func (m *mockInspect) InspectRealm(_ context.Context, opts *schema.InspectRealmOption) (*schema.Realm, error) {
+	m.schemas = append(m.schemas, opts.Schemas...)
+	return m.realm, nil
+}
+
+type mockApply struct {
+	migrate.PlanApplier
+	changes [][]schema.Change
+}
+
+func (m *mockApply) ApplyChanges(_ context.Context, changes []schema.Change) error {
+	m.changes = append(m.changes, changes)
+	return nil
+}
diff --git a/sql/postgres/migrate_test.go b/sql/postgres/migrate_test.go
index 9a7a4d8f..346ef00e 100644
--- a/sql/postgres/migrate_test.go
+++ b/sql/postgres/migrate_test.go
@@ -22,6 +22,29 @@ func TestPlanChanges(t *testing.T) {
 		mock    func(mock)
 		plan    *migrate.Plan
 	}{
+		{
+			changes: []schema.Change{
+				&schema.AddSchema{S: schema.New("test"), Extra: []schema.Clause{&schema.IfNotExists{}}},
+				&schema.DropSchema{S: schema.New("test"), Extra: []schema.Clause{&schema.IfExists{}}},
+				&schema.DropSchema{S: schema.New("test"), Extra: []schema.Clause{&Cascade{}}},
+			},
+			plan: &migrate.Plan{
+				Reversible:    false,
+				Transactional: true,
+				Changes: []*migrate.Change{
+					{
+						Cmd:     `CREATE SCHEMA IF NOT EXISTS "test"`,
+						Reverse: `DROP SCHEMA "test"`,
+					},
+					{
+						Cmd: `DROP SCHEMA IF EXISTS "test"`,
+					},
+					{
+						Cmd: `DROP SCHEMA "test" CASCADE`,
+					},
+				},
+			},
+		},
 		{
 			changes: []schema.Change{
 				func() schema.Change {