Repository: ariga/atlas
Pull request #2329 (2023-11-30 14:31:09)

sql: dependency aware plans


Patch:
diff --git a/cmd/atlas/main.go b/cmd/atlas/main.go
index bc12cc07..a321ac7b 100644
--- a/cmd/atlas/main.go
+++ b/cmd/atlas/main.go
@@ -9,6 +9,8 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"os/signal"
+	"syscall"
 	"time"
 
 	"ariga.io/atlas/cmd/atlas/internal/cmdapi"
@@ -30,7 +32,22 @@ import (
 
 func main() {
 	cmdapi.Root.SetOut(os.Stdout)
-	ctx := newContext()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		// On first signal seen, cancel the context. On the second signal, force stop immediately.
+		stop := make(chan os.Signal, 2)
+		defer close(stop)
+		signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP)
+		defer signal.Stop(stop)
+		<-stop   // wait for first interrupt
+		cancel() // cancel context to gracefully stop
+		fmt.Fprintln(cmdapi.Root.OutOrStdout(), "\ninterrupt received, wait for exit or ^C to terminate")
+		// Wait for the context to be canceled. Issuing a second interrupt will cause the process to force stop.
+		<-stop // will not block if no signal received due to main routine exiting
+		os.Exit(1)
+	}()
+	ctx = extendContext(context.Background())
 	done := initialize(ctx)
 	update := checkForUpdate(ctx)
 	err := cmdapi.Root.ExecuteContext(ctx)
diff --git a/cmd/atlas/main_oss.go b/cmd/atlas/main_oss.go
index 9d17ec34..4218a50f 100644
--- a/cmd/atlas/main_oss.go
+++ b/cmd/atlas/main_oss.go
@@ -8,14 +8,9 @@ package main
 
 import (
 	"context"
-	"os"
-	"os/signal"
 )
 
-func newContext() context.Context {
-	ctx, _ := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
-	return ctx
-}
+func extendContext(ctx context.Context) context.Context { return ctx }
 
 func vercheckEndpoint(context.Context) string {
 	return vercheckURL
diff --git a/sql/internal/sqlx/plan.go b/sql/internal/sqlx/plan.go
index 54864d53..42270e1d 100644
--- a/sql/internal/sqlx/plan.go
+++ b/sql/internal/sqlx/plan.go
@@ -9,6 +9,7 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"slices"
 	"sort"
 
 	"ariga.io/atlas/sql/migrate"
@@ -380,3 +381,232 @@ func byKeys[T any](m map[string]T) []struct {
 	})
 	return vs
 }
+
+// SortChanges is a helper function to sort to level changes based on their priority.
+func SortChanges(changes []schema.Change) []schema.Change {
+	var views, drop, other []schema.Change
+	for _, c := range changes {
+		switch c.(type) {
+		case *schema.AddView, *schema.DropView, *schema.ModifyView:
+			views = append(views, c)
+		case *schema.DropTable, *schema.DropFunc, *schema.DropProc, *schema.DropObject:
+			drop = append(drop, c)
+		default:
+			other = append(other, c)
+		}
+	}
+	if planned, err := sortViewChanges(views); err == nil { // no cycles.
+		views = planned
+	}
+	// To keep backwards compatibility with previous sorting and also in case we miss any dependency between changes
+	// (see, dependsOn function) we push views and drop changes to the end, unless there is a dependency requirement.
+	changes = append(other, append(views, drop...)...)
+	edges := make(map[schema.Change][]schema.Change)
+	for _, c := range changes {
+		for _, c2 := range changes {
+			if c != c2 && dependsOn(c, c2) {
+				edges[c] = append(edges[c], c2)
+			}
+		}
+	}
+	var (
+		add     func(schema.Change)
+		added   = make(map[schema.Change]bool)
+		planned = make([]schema.Change, 0, len(changes))
+	)
+	add = func(c schema.Change) {
+		if added[c] {
+			return
+		}
+		added[c] = true
+		for _, d := range edges[c] {
+			if !added[d] {
+				add(d)
+			}
+		}
+		planned = append(planned, c)
+	}
+	for _, c := range changes {
+		if !added[c] {
+			add(c)
+		}
+	}
+	return planned
+}
+
+// dependsOn reports if the given change depends on the other change.
+func dependsOn(c1, c2 schema.Change) bool {
+	switch c1 := c1.(type) {
+	case *schema.AddTable:
+		switch c2 := c2.(type) {
+		case *schema.DropTable:
+			return c1.T.Name == c2.T.Name && c1.T.Schema == c2.T.Schema // Table recreation.
+		case *schema.AddTable:
+			if refTo(c1.T.ForeignKeys, c2.T) {
+				return true
+			}
+		case *schema.ModifyTable:
+			if refTo(c1.T.ForeignKeys, c2.T) {
+				return true
+			}
+		}
+		return depOfAdd(c1.T.Deps, c2)
+	case *schema.DropTable:
+		// If it is a drop of a table, the change must occur
+		// after all resources that rely on it will be dropped.
+		switch c2 := c2.(type) {
+		case *schema.DropTable:
+			if refTo(c2.T.ForeignKeys, c1.T) {
+				return true
+			}
+		case *schema.ModifyTable:
+			return slices.ContainsFunc(c2.Changes, func(c schema.Change) bool {
+				fk, ok := c.(*schema.DropForeignKey)
+				return ok && fk.F.RefTable == c2.T
+			})
+		}
+		return depOfDrop(c1.T, c2)
+	case *schema.ModifyTable:
+		switch c2 := c2.(type) {
+		case *schema.AddTable:
+			// Table modification relies on its creation.
+			if c1.T.Name == c2.T.Name && c1.T.Schema == c2.T.Schema {
+				return true
+			}
+			// Tables need to be created before referencing them.
+			return slices.ContainsFunc(c1.Changes, func(c schema.Change) bool {
+				fk, ok := c.(*schema.AddForeignKey)
+				return ok && fk.F.RefTable == c2.T
+			})
+		case *schema.ModifyTable:
+			if c1.T != c2.T {
+				addC := make(map[*schema.Column]bool)
+				for _, c := range c2.Changes {
+					if add, ok := c.(*schema.AddColumn); ok {
+						addC[add.C] = true
+					}
+				}
+				return slices.ContainsFunc(c1.Changes, func(c schema.Change) bool {
+					fk, ok := c.(*schema.AddForeignKey)
+					return ok && fk.F.RefTable == c2.T && slices.ContainsFunc(fk.F.RefColumns, func(c *schema.Column) bool { return addC[c] })
+				})
+			}
+		}
+		return depOfAdd(c1.T.Deps, c2)
+	case *schema.AddView:
+		if c2, ok := c2.(*schema.DropView); ok && c1.V.Name == c2.V.Name && c1.V.Schema == c2.V.Schema {
+			return true // View recreation.
+		}
+		return depOfAdd(c1.V.Deps, c2)
+	case *schema.DropView:
+		return depOfDrop(c1.V, c2)
+	case *schema.ModifyView:
+		if c2, ok := c2.(*schema.AddView); ok {
+			// View modification relies on its creation.
+			return c1.From.Name == c2.V.Name && c1.From.Schema == c2.V.Schema
+		}
+		return depOfAdd(c1.To.Deps, c2)
+	case *schema.AddFunc:
+		if c2, ok := c2.(*schema.DropFunc); ok && c1.F.Name == c2.F.Name && c1.F.Schema == c2.F.Schema {
+			return true // Func recreation.
+		}
+		return depOfAdd(c1.F.Deps, c2)
+	case *schema.DropFunc:
+		return depOfDrop(c1.F, c2)
+	case *schema.ModifyFunc:
+		if c2, ok := c2.(*schema.AddFunc); ok {
+			// Func modification relies on its creation.
+			return c1.From.Name == c2.F.Name && c1.From.Schema == c2.F.Schema
+		}
+		return depOfAdd(c1.To.Deps, c2)
+	case *schema.AddProc:
+		if c2, ok := c2.(*schema.DropProc); ok && c1.P.Name == c2.P.Name && c1.P.Schema == c2.P.Schema {
+			return true // Proc recreation.
+		}
+		return depOfAdd(c1.P.Deps, c2)
+	case *schema.DropProc:
+		return depOfDrop(c1.P, c2)
+	case *schema.ModifyProc:
+		if c2, ok := c2.(*schema.AddProc); ok {
+			// Proc modification relies on its creation.
+			return c1.From.Name == c2.P.Name && c1.From.Schema == c2.P.Schema
+		}
+		return depOfAdd(c1.To.Deps, c2)
+	case *schema.DropObject:
+		t, ok := c1.O.(schema.Type)
+		if !ok {
+			return false
+		}
+		// Dropping a type must occur after all its usage were dropped.
+		switch c2 := c2.(type) {
+		case *schema.DropTable:
+			return slices.ContainsFunc(c2.T.Columns, func(c *schema.Column) bool {
+				return dependsOnT(c.Type.Type, t)
+			})
+		case *schema.ModifyTable:
+			return slices.ContainsFunc(c2.Changes, func(c schema.Change) bool {
+				d, ok := c.(*schema.DropColumn)
+				return ok && dependsOnT(d.C.Type.Type, t)
+			})
+		case *schema.DropFunc:
+			return c2.F.Ret == t || slices.ContainsFunc(c2.F.Args, func(f *schema.FuncArg) bool {
+				return dependsOnT(f.Type, t)
+			})
+		case *schema.DropProc:
+			return slices.ContainsFunc(c2.P.Args, func(f *schema.FuncArg) bool {
+				return dependsOnT(f.Type, t)
+			})
+		}
+	}
+	return false
+}
+
+// depOfDrops checks if the given object is a dependency of the given change.
+func depOfDrop(o schema.Object, c schema.Change) bool {
+	var deps []schema.Object
+	switch c := c.(type) {
+	case *schema.DropTable:
+		deps = c.T.Deps
+	case *schema.DropView:
+		deps = c.V.Deps
+	case *schema.DropFunc:
+		deps = c.F.Deps
+	case *schema.DropProc:
+		deps = c.P.Deps
+	}
+	return slices.Contains(deps, o)
+}
+
+// depOfAdd checks if the given change is a creation of a resource exists in the given list.
+func depOfAdd(refs []schema.Object, c schema.Change) bool {
+	var o schema.Object
+	switch c := c.(type) {
+	case *schema.AddTable:
+		o = c.T
+	case *schema.ModifyTable:
+		o = c.T
+	case *schema.AddView:
+		o = c.V
+	case *schema.AddFunc:
+		o = c.F
+	case *schema.AddProc:
+		o = c.P
+	case *schema.AddObject:
+		o = c.O
+	default:
+		return false
+	}
+	return slices.Contains(refs, o)
+}
+
+// refTo reports if the given foreign keys reference the given table.
+func refTo(fks []*schema.ForeignKey, to *schema.Table) bool {
+	return slices.ContainsFunc(fks, func(fk *schema.ForeignKey) bool { return fk.RefTable == to })
+}
+
+// dependsOnT reports if t1 depends on t2.
+func dependsOnT(t1, t2 schema.Type) bool {
+	// Comparing might panic due to mismatch types.
+	defer func() { recover() }()
+	return t1 == t2 || schema.UnderlyingType(t1) == t2
+}
diff --git a/sql/internal/sqlx/sqlx_oss.go b/sql/internal/sqlx/sqlx_oss.go
index f8820d42..11ee9ecc 100644
--- a/sql/internal/sqlx/sqlx_oss.go
+++ b/sql/internal/sqlx/sqlx_oss.go
@@ -10,8 +10,7 @@ import (
 	"ariga.io/atlas/sql/schema"
 )
 
-// PlanViewChanges (should) plan view changes in the current order they should be applied.
-// It is unimplemented for community version as views are not supported there.
-func PlanViewChanges(changes []schema.Change) ([]schema.Change, error) {
+// sortViewChanges is an optional function to sort to views by their dependencies.
+func sortViewChanges(changes []schema.Change) ([]schema.Change, error) {
 	return changes, nil // unimplemented.
 }
diff --git a/sql/mysql/migrate.go b/sql/mysql/migrate.go
index fbcb2b8c..6e9544cc 100644
--- a/sql/mysql/migrate.go
+++ b/sql/mysql/migrate.go
@@ -85,8 +85,7 @@ func (s *state) plan(changes []schema.Change) error {
 	if err != nil {
 		return err
 	}
-	var views []schema.Change
-	for _, c := range planned {
+	for _, c := range sqlx.SortChanges(planned) {
 		switch c := c.(type) {
 		case *schema.AddTable:
 			err = s.addTable(c)
@@ -108,20 +107,6 @@ func (s *state) plan(changes []schema.Change) error {
 			err = s.dropFunc(c)
 		case *schema.DropProc:
 			err = s.dropProc(c)
-		case *schema.AddView, *schema.DropView, *schema.ModifyView, *schema.RenameView:
-			views = append(views, c)
-		default:
-			err = fmt.Errorf("unsupported change %T", c)
-		}
-		if err != nil {
-			return err
-		}
-	}
-	if views, err = sqlx.PlanViewChanges(views); err != nil {
-		return err
-	}
-	for _, c := range views {
-		switch c := c.(type) {
 		case *schema.AddView:
 			err = s.addView(c)
 		case *schema.DropView:
@@ -130,6 +115,8 @@ func (s *state) plan(changes []schema.Change) error {
 			err = s.modifyView(c)
 		case *schema.RenameView:
 			s.renameView(c)
+		default:
+			err = fmt.Errorf("unsupported change %T", c)
 		}
 		if err != nil {
 			return err
diff --git a/sql/postgres/driver.go b/sql/postgres/driver.go
index dfadc619..4c5c9a79 100644
--- a/sql/postgres/driver.go
+++ b/sql/postgres/driver.go
@@ -251,17 +251,17 @@ func (d *Driver) RealmRestoreFunc(desired *schema.Realm) migrate.RestoreFunc {
 
 func withCascade(changes schema.Changes) schema.Changes {
 	for _, c := range changes {
-		if d, ok := c.(*schema.DropTable); ok {
-			d.Extra = append(d.Extra, &Cascade{})
-		}
-		if d, ok := c.(*schema.DropView); ok {
-			d.Extra = append(d.Extra, &schema.IfExists{})
-		}
-		if d, ok := c.(*schema.DropProc); ok {
-			d.Extra = append(d.Extra, &schema.IfExists{}, &Cascade{})
-		}
-		if d, ok := c.(*schema.DropFunc); ok {
-			d.Extra = append(d.Extra, &schema.IfExists{}, &Cascade{})
+		switch c := c.(type) {
+		case *schema.DropTable:
+			c.Extra = append(c.Extra, &schema.IfExists{}, &Cascade{})
+		case *schema.DropView:
+			c.Extra = append(c.Extra, &schema.IfExists{}, &Cascade{})
+		case *schema.DropProc:
+			c.Extra = append(c.Extra, &schema.IfExists{}, &Cascade{})
+		case *schema.DropFunc:
+			c.Extra = append(c.Extra, &schema.IfExists{}, &Cascade{})
+		case *schema.DropObject:
+			c.Extra = append(c.Extra, &schema.IfExists{}, &Cascade{})
 		}
 	}
 	return changes
diff --git a/sql/postgres/inspect.go b/sql/postgres/inspect.go
index e685f783..d52162bd 100644
--- a/sql/postgres/inspect.go
+++ b/sql/postgres/inspect.go
@@ -111,6 +111,9 @@ func (i *inspect) InspectSchema(ctx context.Context, name string, opts *schema.I
 			return nil, err
 		}
 	}
+	if err := i.inspectDeps(ctx, r, opts); err != nil {
+		return nil, err
+	}
 	return sqlx.ExcludeSchema(r.Schemas[0], opts.Exclude)
 }
 
@@ -1035,6 +1038,16 @@ func (d *DomainType) Ref() *schemahcl.Ref {
 	return &schemahcl.Ref{V: "$domain." + d.T}
 }
 
+// Underlying returns the underlying type of the domain.
+func (d *DomainType) Underlying() schema.Type {
+	return d.Type
+}
+
+// Underlying returns the underlying type of the array.
+func (a *ArrayType) Underlying() schema.Type {
+	return a.Type
+}
+
 // String implements fmt.Stringer interface.
 func (o ReferenceOption) String() string {
 	return string(o)
diff --git a/sql/postgres/migrate.go b/sql/postgres/migrate.go
index 10272dd3..361211f5 100644
--- a/sql/postgres/migrate.go
+++ b/sql/postgres/migrate.go
@@ -81,11 +81,7 @@ func (s *state) plan(changes []schema.Change) error {
 	if planned, err = sqlx.DetachCycles(planned); err != nil {
 		return err
 	}
-	var (
-		views []schema.Change
-		drop  struct{ T, O, F []schema.Change }
-	)
-	for _, c := range planned {
+	for _, c := range sqlx.SortChanges(planned) {
 		switch c := c.(type) {
 		case *schema.AddTable:
 			err = s.addTable(c)
@@ -105,26 +101,6 @@ func (s *state) plan(changes []schema.Change) error {
 			err = s.renameFunc(c)
 		case *schema.RenameProc:
 			err = s.renameProc(c)
-		case *schema.AddView, *schema.DropView, *schema.ModifyView, *schema.RenameView:
-			views = append(views, c)
-		case *schema.DropTable:
-			drop.T = append(drop.T, c)
-		case *schema.DropObject:
-			drop.O = append(drop.O, c)
-		case *schema.DropFunc, *schema.DropProc:
-			drop.F = append(drop.F, c)
-		default:
-			err = fmt.Errorf("unsupported change %T", c)
-		}
-		if err != nil {
-			return err
-		}
-	}
-	if views, err = sqlx.PlanViewChanges(views); err != nil {
-		return err
-	}
-	for _, c := range views {
-		switch c := c.(type) {
 		case *schema.AddView:
 			err = s.addView(c)
 		case *schema.DropView:
@@ -133,14 +109,6 @@ func (s *state) plan(changes []schema.Change) error {
 			err = s.modifyView(c)
 		case *schema.RenameView:
 			s.renameView(c)
-		}
-		if err != nil {
-			return err
-		}
-	}
-	for _, c := range append(drop.T, append(drop.O, drop.F...)...) {
-		var err error
-		switch c := c.(type) {
 		case *schema.DropTable:
 			err = s.dropTable(c)
 		case *schema.DropObject:
@@ -149,6 +117,8 @@ func (s *state) plan(changes []schema.Change) error {
 			err = s.dropFunc(c)
 		case *schema.DropProc:
 			err = s.dropProc(c)
+		default:
+			err = fmt.Errorf("unsupported change %T", c)
 		}
 		if err != nil {
 			return err
diff --git a/sql/schema/schema.go b/sql/schema/schema.go
index 0115da06..3f0a4afb 100644
--- a/sql/schema/schema.go
+++ b/sql/schema/schema.go
@@ -571,3 +571,11 @@ func UnderlyingExpr(x Expr) Expr {
 	}
 	return x
 }
+
+// UnderlyingType returns the underlying type of t.
+func UnderlyingType(t Type) Type {
+	if w, ok := t.(interface{ Underlying() Type }); ok {
+		return UnderlyingType(w.Underlying())
+	}
+	return t
+}


Test patch:
diff --git a/internal/integration/integration_test.go b/internal/integration/integration_test.go
index e2d2a1ff..df8c5e13 100644
--- a/internal/integration/integration_test.go
+++ b/internal/integration/integration_test.go
@@ -5,11 +5,14 @@
 package integration
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"database/sql"
 	"flag"
+	"fmt"
 	"io"
+	"net"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -40,6 +43,78 @@ func TestMain(m *testing.M) {
 	os.Exit(code)
 }
 
+func TestCLI_Interrupt(t *testing.T) {
+	var (
+		stdout, stderr bytes.Buffer
+		connected      = make(chan struct{})
+		signal         = make(chan struct{})
+		handler        = func(c net.Conn) {
+			connected <- struct{}{}
+			<-signal // wait for signal sent
+			c.Close()
+		}
+	)
+	t.Cleanup(func() {
+		close(connected)
+		close(signal)
+	})
+
+	// First interrupt will not cancel the process, only print a warning.
+	var (
+		port = newListener(t, handler)
+		cmd  = exec.Command(
+			execPath(t),
+			"schema", "inspect",
+			"--url", fmt.Sprintf("postgres://user:pass@localhost:%d/db?sslmode=disable", port), // mock not responding db
+		)
+	)
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+	require.NoError(t, cmd.Start())
+	<-connected // wait for connection
+	require.NoError(t, cmd.Process.Signal(os.Interrupt))
+	signal <- struct{}{}
+	require.Error(t, cmd.Wait())
+	require.Equal(t, "\ninterrupt received, wait for exit or ^C to terminate\n", stdout.String())
+	require.Contains(t, stderr.String(), "read: connection reset by peer") // server closed
+
+	// Two interrupts force stop
+	r, w := io.Pipe()
+	port = newListener(t, handler)
+	cmd = exec.Command(
+		execPath(t),
+		"schema", "inspect",
+		"--url", fmt.Sprintf("postgres://user:pass@localhost:%d/db?sslmode=disable", port), // mock not responding db
+	)
+	cmd.Stdout = w
+	require.NoError(t, cmd.Start())
+	<-connected // wait for connection
+	require.NoError(t, cmd.Process.Signal(os.Interrupt))
+	// Wait for the cmd to print the warning (it works, we checked before) and then issue the second signal.
+	br := bufio.NewReader(r)
+	_, err := br.ReadString('\n')
+	require.NoError(t, err)
+	out, err := br.ReadString('\n')
+	require.NoError(t, err)
+	require.Equal(t, "interrupt received, wait for exit or ^C to terminate\n", out)
+	require.NoError(t, cmd.Process.Signal(os.Interrupt))
+	require.Error(t, cmd.Wait())
+}
+
+func newListener(t *testing.T, handler func(c net.Conn)) int {
+	a, err := net.ResolveTCPAddr("tcp", "localhost:0")
+	require.NoError(t, err)
+	l, err := net.ListenTCP("tcp", a)
+	require.NoError(t, err)
+	t.Cleanup(func() { l.Close() })
+	go func() {
+		c, err := l.Accept()
+		require.NoError(t, err)
+		handler(c)
+	}()
+	return l.Addr().(*net.TCPAddr).Port
+}
+
 // T holds the elements common between dialect tests.
 type T interface {
 	testing.TB
diff --git a/sql/internal/sqlx/plan_test.go b/sql/internal/sqlx/plan_test.go
index 029c19d8..a4eddb5e 100644
--- a/sql/internal/sqlx/plan_test.go
+++ b/sql/internal/sqlx/plan_test.go
@@ -153,3 +153,16 @@ func TestCheckChangesScope(t *testing.T) {
 	})
 	require.EqualError(t, err, "found 2 schemas when migration plan is scoped to one: [\"s1\" \"s2\"]")
 }
+
+func TestBadTypeComparison(t *testing.T) {
+	var o1 struct {
+		schema.Type
+		schema.Object
+		_ []func()
+	}
+	require.Panics(t, func() { _ = schema.Type(o1) == schema.Type(o1) })
+	SortChanges([]schema.Change{
+		&schema.DropObject{O: o1},
+		&schema.DropTable{T: schema.NewTable("t1").AddColumns(schema.NewColumn("c1").SetType(o1))},
+	})
+}
diff --git a/sql/mysql/migrate_test.go b/sql/mysql/migrate_test.go
index 922f2f5d..1973fd05 100644
--- a/sql/mysql/migrate_test.go
+++ b/sql/mysql/migrate_test.go
@@ -27,9 +27,11 @@ 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("CREATE TABLE `users` (`id` bigint NOT NULL)")).
+		WillReturnResult(sqlmock.NewResult(0, 0))
 	mk.ExpectExec(sqltest.Escape("DROP TABLE IF EXISTS `public`.`pets`")).
 		WillReturnResult(sqlmock.NewResult(0, 0))
-	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\")")).
+	mk.ExpectExec(sqltest.Escape("CREATE TABLE IF NOT EXISTS `public`.`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))
@@ -39,6 +41,7 @@ func TestMigrate_ApplyChanges(t *testing.T) {
 		WillReturnResult(sqlmock.NewResult(0, 0))
 	mk.ExpectExec(sqltest.Escape("CREATE TABLE `comments` (`id` bigint NOT NULL, `post_id` bigint NULL, CONSTRAINT `comment` FOREIGN KEY (`post_id`) REFERENCES `posts` (`id`))")).
 		WillReturnResult(sqlmock.NewResult(0, 0))
+	ns := &schema.Schema{Name: "public"}
 	err = migrate.ApplyChanges(context.Background(), []schema.Change{
 		&schema.AddSchema{S: &schema.Schema{Name: "test", Attrs: []schema.Attr{&schema.Charset{V: "latin"}}}},
 		&schema.DropSchema{S: &schema.Schema{Name: "atlas", Attrs: []schema.Attr{&schema.Charset{V: "latin"}}}},
@@ -50,10 +53,18 @@ func TestMigrate_ApplyChanges(t *testing.T) {
 				},
 			},
 		},
+		&schema.AddTable{
+			T: &schema.Table{
+				Name: "users",
+				Columns: []*schema.Column{
+					{Name: "id", Type: &schema.ColumnType{Raw: "bigint", Type: &schema.IntegerType{T: "bigint"}}},
+				},
+			},
+		},
 		&schema.DropTable{
 			T: &schema.Table{
 				Name:   "pets",
-				Schema: &schema.Schema{Name: "public"},
+				Schema: ns,
 				Columns: []*schema.Column{
 					{Name: "id", Type: &schema.ColumnType{Raw: "bigint", Type: &schema.IntegerType{T: "bigint"}}},
 				},
@@ -63,7 +74,8 @@ func TestMigrate_ApplyChanges(t *testing.T) {
 		&schema.AddTable{
 			T: func() *schema.Table {
 				t := &schema.Table{
-					Name: "pets",
+					Name:   "pets",
+					Schema: ns,
 					Columns: []*schema.Column{
 						{Name: "a", Type: &schema.ColumnType{Raw: "int", Type: &schema.IntegerType{T: "int"}}, Default: &schema.RawExpr{X: "(int(rand()))"}},
 						{Name: "b", Type: &schema.ColumnType{Raw: "bigint", Type: &schema.IntegerType{T: "bigint"}}, Default: &schema.Literal{V: "1"}},
diff --git a/sql/postgres/migrate_test.go b/sql/postgres/migrate_test.go
index da71c206..f4befb03 100644
--- a/sql/postgres/migrate_test.go
+++ b/sql/postgres/migrate_test.go
@@ -368,12 +368,12 @@ func TestPlanChanges(t *testing.T) {
 		},
 		{
 			changes: []schema.Change{
-				&schema.DropObject{O: &schema.EnumType{T: "status", Values: []string{"on", "off"}, Schema: schema.New("public")}},
 				&schema.DropTable{T: schema.NewTable("logs").
 					AddColumns(
 						schema.NewColumn("status").SetType(&schema.EnumType{T: "status", Values: []string{"on", "off"}, Schema: schema.New("public")}),
 					),
 				},
+				&schema.DropObject{O: &schema.EnumType{T: "status", Values: []string{"on", "off"}, Schema: schema.New("public")}},
 				&schema.DropObject{O: &schema.EnumType{T: "state", Values: []string{"on", "off"}, Schema: schema.New("public")}},
 			},
 			wantPlan: &migrate.Plan{
@@ -872,13 +872,6 @@ func TestPlanChanges(t *testing.T) {
 		},
 		{
 			changes: []schema.Change{
-				&schema.DropObject{
-					O: &schema.EnumType{
-						T:      "state",
-						Values: []string{"on", "off"},
-						Schema: schema.New("public"),
-					},
-				},
 				&schema.DropTable{
 					T: schema.NewTable("users").
 						SetSchema(schema.New("public")).
@@ -887,6 +880,13 @@ func TestPlanChanges(t *testing.T) {
 							schema.NewEnumColumn("status", schema.EnumName("status"), schema.EnumValues("on", "off"), schema.EnumSchema(schema.New("public"))),
 						),
 				},
+				&schema.DropObject{
+					O: &schema.EnumType{
+						T:      "state",
+						Values: []string{"on", "off"},
+						Schema: schema.New("public"),
+					},
+				},
 				&schema.DropObject{
 					O: &schema.EnumType{
 						T:      "status",
@@ -940,6 +940,12 @@ func TestPlanChanges(t *testing.T) {
 							),
 					)
 				return []schema.Change{
+					&schema.DropTable{
+						T: s.Tables[0],
+					},
+					&schema.DropTable{
+						T: s.Tables[1],
+					},
 					&schema.DropObject{
 						O: &schema.EnumType{
 							T:      "state",
@@ -947,12 +953,6 @@ func TestPlanChanges(t *testing.T) {
 							Schema: schema.New("public"),
 						},
 					},
-					&schema.DropTable{
-						T: s.Tables[0],
-					},
-					&schema.DropTable{
-						T: s.Tables[1],
-					},
 				}
 			}(),
 			wantPlan: &migrate.Plan{