Repository: ariga/atlas
Pull request #316 (2021-12-19 12:03:11)

sql: support migrate directory and add implement mysql planner
The 2 commits don't have 2 much in common, but I added the mysql plan implementation to give some idea how the usage looks like.

The rest of the drivers will be added in the next diffs.

a8m: > Nice work, maybe add it to atlas next week?

Why next week? ;) 

Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index b51fb778..e0dab797 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -182,6 +182,8 @@ jobs:
     steps:
       - uses: actions/checkout@v2.3.4
       - uses: actions/setup-go@v2
+        with:
+          go-version: 1.17
       - uses: actions/cache@v2.1.6
         with:
           path: ~/go/pkg/mod
diff --git a/sql/migrate/migrate.go b/sql/migrate/migrate.go
new file mode 100644
index 00000000..ade2b537
--- /dev/null
+++ b/sql/migrate/migrate.go
@@ -0,0 +1,421 @@
+// 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 migrate
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"text/template"
+	"time"
+
+	"ariga.io/atlas/sql/schema"
+)
+
+type (
+	// A Plan defines a planned changeset that its execution brings the database to
+	// the new desired state. Additional information is calculated by the different
+	// drivers to indicate if the changeset is transactional (can be rolled-back) and
+	// reversible (a down file can be generated to it).
+	Plan struct {
+		// Name of the plan. Provided by the user or auto-generated.
+		Name string
+
+		// Reversible describes if the changeset is reversible.
+		Reversible bool
+
+		// Transactional describes if the changeset is transactional.
+		Transactional bool
+
+		// Changes defines the list of changeset in the plan.
+		Changes []*Change
+	}
+
+	// A Change of migration.
+	Change struct {
+		// Cmd or statement to execute.
+		Cmd string
+
+		// A Comment describes the change.
+		Comment string
+
+		// Reverse contains the "reversed statement" if
+		// command is reversible.
+		Reverse string
+
+		// The Source that caused this change.
+		Source schema.Change
+	}
+)
+
+type (
+	// The Driver interface must be implemented by the different dialects to support database
+	// migration authoring/planning and applying. ExecQuerier, Inspector and Differ, provide
+	// basic schema primitives for inspecting database schemas, calculate the difference between
+	// schema elements, and executing raw SQL statements. The PlanApplier interface wraps the
+	// methods for generating migration plan for applying the actual changes on the database.
+	Driver interface {
+		schema.Differ
+		schema.ExecQuerier
+		schema.Inspector
+		PlanApplier
+	}
+
+	// PlanApplier wraps the methods for planning and applying changes
+	// on the database.
+	PlanApplier interface {
+		PlanChanges(context.Context, string, []schema.Change) (*Plan, error)
+		ApplyChanges(context.Context, []schema.Change) error
+	}
+
+	// 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).
+	//
+	// In next Go version, the State will be replaced with the following
+	// union type `interface { Realm | Schema }`.
+	StateReader interface {
+		ReadState(ctx context.Context) (*schema.Realm, error)
+	}
+
+	// The StateReaderFunc type is an adapter to allow the use of
+	// ordinary functions as state readers.
+	StateReaderFunc func(ctx context.Context) (*schema.Realm, error)
+)
+
+// ReadState calls f(ctx).
+func (f StateReaderFunc) ReadState(ctx context.Context) (*schema.Realm, error) {
+	return f(ctx)
+}
+
+// ErrNoPlan is returned by Plan when there is no change between the two states.
+var ErrNoPlan = errors.New("sql/migrate: no plan for matched states")
+
+// Realm returns a state reader for the static Realm object.
+func Realm(r *schema.Realm) StateReader {
+	return StateReaderFunc(func(context.Context) (*schema.Realm, error) {
+		return r, nil
+	})
+}
+
+// Schema returns a state reader for the static Schema object.
+func Schema(s *schema.Schema) StateReader {
+	return StateReaderFunc(func(context.Context) (*schema.Realm, error) {
+		r := &schema.Realm{Schemas: []*schema.Schema{s}}
+		if s.Realm != nil {
+			r.Attrs = s.Realm.Attrs
+		}
+		s.Realm = r
+		return r, nil
+	})
+}
+
+type (
+	// Dir represents a versioned migration directory.
+	Dir struct {
+		fs        fs.FS
+		conn      Driver
+		pattern   string
+		templates []struct{ N, T *template.Template }
+	}
+
+	// DirOption allows configuring the Dir
+	// using functional options.
+	DirOption func(*Dir) error
+)
+
+// NewDir creates a new workspace directory based
+// on the given configuration options.
+func NewDir(opts ...DirOption) (*Dir, error) {
+	d := &Dir{
+		fs: dirFS{
+			dir: "migrations",
+			FS:  os.DirFS("migrations"),
+		},
+		pattern: "*.sql",
+	}
+	for _, opt := range opts {
+		if err := opt(d); err != nil {
+			return nil, err
+		}
+	}
+	if len(d.templates) == 0 {
+		d.templates = []struct{ N, T *template.Template }{defaultTemplate}
+	}
+	return d, nil
+}
+
+type (
+	// dirFS wraps the os.DirFS with additional writing capabilities.
+	dirFS struct {
+		fs.FS
+		dir string
+	}
+	// FileRemoveWriter wraps the WriteFile and RemoveFile methods
+	// to allow editing the migration directory on development.
+	FileRemoveWriter interface {
+		RemoveFile(name string) error
+		WriteFile(name string, data []byte, perm fs.FileMode) error
+	}
+)
+
+// WriteFile implements the FileWriter interface.
+func (d *dirFS) WriteFile(name string, data []byte, perm fs.FileMode) error {
+	return os.WriteFile(filepath.Join(d.dir, name), data, perm)
+}
+
+// RemoveFile implements the FileWriter interface.
+func (d *dirFS) RemoveFile(name string) error {
+	return os.Remove(filepath.Join(d.dir, name))
+}
+
+// DirFS configures the FS used by the migration directory.
+func DirFS(fs fs.FS) DirOption {
+	return func(d *Dir) error {
+		d.fs = fs
+		return nil
+	}
+}
+
+// DirPath configures the FS used by the migration directory
+// to point the given OS directory.
+func DirPath(path string) DirOption {
+	return func(d *Dir) error {
+		fi, err := os.Stat(path)
+		if err != nil {
+			return err
+		}
+		if !fi.IsDir() {
+			return fmt.Errorf("sql/migrate: %q not a dir", path)
+		}
+		d.fs = &dirFS{FS: os.DirFS(path), dir: path}
+		return nil
+	}
+}
+
+// DirGlob configures the glob/pattern for reading
+// migration files from the directory. For example:
+//
+//	migrate.NewDir(
+//		migrate.DirPath("migrations"),
+//		migrate.DirGlob("*.up.sql"),
+//	)
+//
+func DirGlob(pattern string) DirOption {
+	return func(d *Dir) error {
+		d.pattern = pattern
+		return nil
+	}
+}
+
+// DirConn provides a Driver connection to a database. It is usually connected to
+// an ephemeral database for emulating migration changes on it and calculating the
+// "current state" to be compared with the "desired state".
+func DirConn(conn Driver) DirOption {
+	return func(d *Dir) error {
+		d.conn = conn
+		return nil
+	}
+}
+
+var (
+	// TemplateFuncs defines the global functions available for the templates.
+	TemplateFuncs = template.FuncMap{
+		"hasSuffix": strings.HasSuffix,
+		"timestamp": func() int64 {
+			return time.Now().Unix()
+		},
+	}
+	defaultTemplate = struct {
+		N, T *template.Template
+	}{
+		N: template.Must(template.New("name").
+			Funcs(TemplateFuncs).
+			Parse("{{timestamp}}_{{.Name}}.sql")),
+		T: template.Must(template.New("name").
+			Funcs(TemplateFuncs).
+			Parse(`
+{{- range $c := .Changes }}
+	{{- $cmd := $c.Cmd }}
+	{{- if not (hasSuffix $c.Cmd ";") }}
+		{{- $cmd = print $cmd ";" }}
+	{{- end }}
+	{{- println $cmd }}
+{{- end }}`)),
+	}
+)
+
+// DirTemplates configures template files for writing
+// the Plan object to the migration directory.
+//
+//	migrate.NewDir(
+//		migrate.DirPath("migrations"),
+//		migrate.DirTemplates("{{timestamp}}{{.Name}}.up.sql", "{{range c := .Changes}}{{println c.Cmd}}{{end}}"),
+//	)
+//
+//	migrate.NewDir(
+//		migrate.DirPath("migrations"),
+//		migrate.DirTemplates(
+//			"{{timestamp}}{{.Name}}.up.sql", "{{range $c := .Changes}}{{println $c.Cmd}}{{end}}",
+//			"{{timestamp}}{{.Name}}.down.sql", "{{range $c := .Changes}}{{println $c.Reverse}}{{end}}",
+//		),
+//	)
+//
+func DirTemplates(nameFileTmpl ...string) DirOption {
+	return func(d *Dir) error {
+		if n := len(nameFileTmpl); n == 0 || n%2 == 1 {
+			return fmt.Errorf("odd or zero argument count")
+		}
+		for i := 0; i < len(nameFileTmpl); i += 2 {
+			if err := d.addTemplate(nameFileTmpl[i], nameFileTmpl[i+1]); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+}
+
+// ReadState reads the current database/realm state that is stored in the migration directory.
+// The given emulator driver is used for playing all migration files against to it.
+func (d *Dir) ReadState(ctx context.Context) (*schema.Realm, error) {
+	realm, _, err := d.readStateOf(ctx, -1)
+	return realm, err
+}
+
+// Plan calculates the migration Plan required for moving from the directory state to
+// the next state (to). A StateReader, can be another directory, static schema elements
+// or a Driver connection.
+func (d *Dir) Plan(ctx context.Context, name string, to StateReader) (*Plan, error) {
+	current, err := d.ReadState(ctx)
+	if err != nil {
+		return nil, err
+	}
+	desired, err := to.ReadState(ctx)
+	if err != nil {
+		return nil, err
+	}
+	changes, err := d.conn.RealmDiff(current, desired)
+	if err != nil {
+		return nil, err
+	}
+	if len(changes) == 0 {
+		return nil, ErrNoPlan
+	}
+	return d.conn.PlanChanges(ctx, name, changes)
+}
+
+// WritePlan writes the given plan to the directory
+// based on the given Write configuration.
+func (d *Dir) WritePlan(p *Plan) error {
+	rw, ok := d.fs.(FileRemoveWriter)
+	if !ok {
+		return fmt.Errorf("fs.FS does not support editing: %T", d.fs)
+	}
+	for _, t := range d.templates {
+		var b bytes.Buffer
+		if err := t.N.Execute(&b, p); err != nil {
+			return err
+		}
+		if b.String() == "" {
+			return errors.New("file name cannot be empty")
+		}
+		name := b.String()
+		b.Reset()
+		if err := t.T.Execute(&b, p); err != nil {
+			return err
+		}
+		if err := rw.WriteFile(name, b.Bytes(), 0644); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Compact compacts the first n migration files into one. If n < 0, all files are selected.
+func (d *Dir) Compact(ctx context.Context, name string, n int) error {
+	rw, ok := d.fs.(FileRemoveWriter)
+	if !ok {
+		return fmt.Errorf("fs.FS does not support editing: %T", d.fs)
+	}
+	desired, files, err := d.readStateOf(ctx, n)
+	if err != nil {
+		return err
+	}
+	changes, err := d.conn.RealmDiff(&schema.Realm{}, desired)
+	if err != nil {
+		return err
+	}
+	if len(changes) == 0 {
+		return nil
+	}
+	for _, c := range changes {
+		// Add the "IF NOT EXISTS" clause to the schema
+		// creation as it is usually created manually.
+		if c, ok := c.(*schema.AddSchema); ok {
+			c.Extra = append(c.Extra, &schema.IfNotExists{})
+		}
+	}
+	plan, err := d.conn.PlanChanges(ctx, name, changes)
+	if err != nil {
+		return err
+	}
+	for _, f := range files {
+		if err := rw.RemoveFile(f); err != nil {
+			return fmt.Errorf("remove file: %q: %w", f, err)
+		}
+	}
+	return d.WritePlan(plan)
+}
+
+// readStateOf of first n files. If n < 0, all files are selected.
+func (d *Dir) readStateOf(ctx context.Context, n int) (*schema.Realm, []string, error) {
+	files, err := fs.Glob(d.fs, d.pattern)
+	if err != nil {
+		return nil, nil, err
+	}
+	switch {
+	case n == 0 || n >= len(files):
+		return nil, nil, fmt.Errorf("sql/migrate: invalid number for selected files: %d", n)
+	case n > 0:
+		files = files[:n]
+	}
+	// Files are expected to be sorted lexicographically.
+	sort.Slice(files, func(i, j int) bool {
+		return files[i] < files[j]
+	})
+	for _, f := range files {
+		buf, err := fs.ReadFile(d.fs, f)
+		if err != nil {
+			return nil, nil, fmt.Errorf("sql/migrate: scan migration script %q: %w", f, err)
+		}
+		if _, err := d.conn.ExecContext(ctx, string(buf)); err != nil {
+			return nil, nil, fmt.Errorf("sql/migrate: execute migration script %q: %w", f, err)
+		}
+	}
+	realm, err := d.conn.InspectRealm(ctx, nil)
+	if err != nil {
+		return nil, nil, err
+	}
+	return realm, files, nil
+}
+
+func (d *Dir) addTemplate(nameTmpl, fileTmpl string) error {
+	nameT, err := template.New("name").Funcs(TemplateFuncs).Parse(nameTmpl)
+	if err != nil {
+		return err
+	}
+	fileT, err := template.New("file").Funcs(TemplateFuncs).Parse(fileTmpl)
+	if err != nil {
+		return err
+	}
+	d.templates = append(d.templates, struct{ N, T *template.Template }{N: nameT, T: fileT})
+	return nil
+}
diff --git a/sql/mysql/driver.go b/sql/mysql/driver.go
index 736e9eaa..fdd933c6 100644
--- a/sql/mysql/driver.go
+++ b/sql/mysql/driver.go
@@ -43,7 +43,7 @@ func Open(db schema.ExecQuerier) (*Driver, error) {
 	return &Driver{
 		conn:      c,
 		Differ:    &sqlx.Diff{DiffDriver: &diff{c}},
-		Execer:    &migrate{c},
+		Execer:    &planApply{c},
 		Inspector: &inspect{c},
 	}, nil
 }
diff --git a/sql/mysql/migrate.go b/sql/mysql/migrate.go
index b742746b..e7935e09 100644
--- a/sql/mysql/migrate.go
+++ b/sql/mysql/migrate.go
@@ -11,43 +11,94 @@ import (
 	"strings"
 
 	"ariga.io/atlas/sql/internal/sqlx"
+	"ariga.io/atlas/sql/migrate"
 	"ariga.io/atlas/sql/schema"
 )
 
-// A migrate provides migration capabilities for schema elements.
-type migrate struct{ conn }
+// A planApply provides migration capabilities for schema elements.
+type planApply struct{ conn }
 
-// Exec executes the changes on the database. An error is returned
-// if one of the operations fail, or a change is not supported.
-func (m *migrate) Exec(ctx context.Context, changes []schema.Change) error {
-	planned, err := m.topLevel(ctx, changes)
+// PlanChanges returns a migration plan for the given schema changes.
+func (p *planApply) PlanChanges(_ context.Context, name string, changes []schema.Change) (*migrate.Plan, error) {
+	s := &state{
+		conn: p.conn,
+		Plan: migrate.Plan{
+			Name: name,
+			// A plan is reversible, if all
+			// its changes are reversible.
+			Reversible: true,
+			// All statements generated by state will cause implicit commit.
+			// https://dev.mysql.com/doc/refman/8.0/en/implicit-commit.html
+			Transactional: false,
+		},
+	}
+	if err := s.plan(changes); err != nil {
+		return nil, err
+	}
+	for _, c := range s.Changes {
+		if c.Reverse == "" {
+			s.Reversible = false
+		}
+	}
+	return &s.Plan, nil
+}
+
+// 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 {
+	plan, err := p.PlanChanges(ctx, "apply", changes)
 	if err != nil {
 		return err
 	}
-	planned, err = sqlx.DetachCycles(planned)
+	for _, c := range plan.Changes {
+		if _, err := p.ExecContext(ctx, c.Cmd); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Exec calls ApplyChanges. Exists here for backwards compatibility.
+func (p *planApply) Exec(ctx context.Context, changes []schema.Change) error {
+	return p.ApplyChanges(ctx, changes)
+}
+
+// state represents the state of an planning. It is not part of
+// planApply so that multiple planning/applying can be called
+// in parallel.
+type state struct {
+	conn
+	migrate.Plan
+}
+
+// plan builds the migration plan for applying the
+// given changes on the attached connection.
+func (s *state) plan(changes []schema.Change) error {
+	planned := s.topLevel(changes)
+	planned, err := sqlx.DetachCycles(planned)
 	if err != nil {
 		return err
 	}
 	for _, c := range planned {
 		switch c := c.(type) {
 		case *schema.AddTable:
-			err = m.addTable(ctx, c)
+			s.addTable(c)
 		case *schema.DropTable:
-			err = m.dropTable(ctx, c)
+			s.dropTable(c)
 		case *schema.ModifyTable:
-			err = m.modifyTable(ctx, c)
+			if err := s.modifyTable(c); err != nil {
+				return err
+			}
 		default:
-			err = fmt.Errorf("unsupported change %T", c)
-		}
-		if err != nil {
-			return err
+			return fmt.Errorf("unsupported change %T", c)
 		}
 	}
 	return nil
 }
 
-// topLevel executes first the changes for creating or dropping schemas (top-level schema elements).
-func (m *migrate) topLevel(ctx context.Context, changes []schema.Change) ([]schema.Change, error) {
+// topLevel appends first the changes for creating or dropping schemas (top-level schema elements).
+func (s *state) topLevel(changes []schema.Change) []schema.Change {
 	planned := make([]schema.Change, 0, len(changes))
 	for _, c := range changes {
 		switch c := c.(type) {
@@ -62,34 +113,44 @@ func (m *migrate) topLevel(ctx context.Context, changes []schema.Change) ([]sche
 			if a := (schema.Collation{}); sqlx.Has(c.S.Attrs, &a) {
 				b.P("COLLATE", a.V)
 			}
-			if _, err := m.ExecContext(ctx, b.String()); err != nil {
-				return nil, fmt.Errorf("add schema: %w", err)
-			}
+			s.append(&migrate.Change{
+				Cmd:     b.String(),
+				Source:  c,
+				Reverse: Build("DROP DATABASE").Ident(c.S.Name).String(),
+				Comment: fmt.Sprintf("Add new schema named %q", c.S.Name),
+			})
 		case *schema.DropSchema:
-			if _, err := m.ExecContext(ctx, Build("DROP DATABASE").Ident(c.S.Name).String()); err != nil {
-				return nil, fmt.Errorf("drop schema: %w", err)
+			b := Build("DROP DATABASE").Ident(c.S.Name)
+			if sqlx.Has(c.Extra, &schema.IfExists{}) {
+				b.P("IF EXISTS")
 			}
+			s.append(&migrate.Change{
+				Cmd:     b.String(),
+				Source:  c,
+				Comment: fmt.Sprintf("Drop schema named %q", c.S.Name),
+			})
 		default:
 			planned = append(planned, c)
 		}
 	}
-	return planned, nil
+	return planned
 }
 
-// addTable builds and executes the query for creating a table in a schema.
-func (m *migrate) addTable(ctx context.Context, add *schema.AddTable) error {
+// addTable builds and appends the migrate.Change
+// for creating a table in a schema.
+func (s *state) addTable(add *schema.AddTable) {
 	b := Build("CREATE TABLE").Table(add.T)
 	if sqlx.Has(add.Extra, &schema.IfNotExists{}) {
 		b.P("IF NOT EXISTS")
 	}
 	b.Wrap(func(b *sqlx.Builder) {
 		b.MapComma(add.T.Columns, func(i int, b *sqlx.Builder) {
-			m.column(b, add.T, add.T.Columns[i])
+			s.column(b, add.T, add.T.Columns[i])
 		})
 		if pk := add.T.PrimaryKey; pk != nil {
 			b.Comma().P("PRIMARY KEY")
-			m.indexParts(b, pk.Parts)
-			m.attr(b, pk.Attrs...)
+			s.indexParts(b, pk.Parts)
+			s.attr(b, pk.Attrs...)
 		}
 		if len(add.T.Indexes) > 0 {
 			b.Comma()
@@ -100,35 +161,40 @@ func (m *migrate) addTable(ctx context.Context, add *schema.AddTable) error {
 				b.P("UNIQUE")
 			}
 			b.P("INDEX").Ident(idx.Name)
-			m.indexParts(b, idx.Parts)
-			m.attr(b, idx.Attrs...)
+			s.indexParts(b, idx.Parts)
+			s.attr(b, idx.Attrs...)
 		})
 		if len(add.T.ForeignKeys) > 0 {
 			b.Comma()
-			m.fks(b, add.T.ForeignKeys...)
+			s.fks(b, add.T.ForeignKeys...)
 		}
 	})
-	m.tableAttr(b, add.T.Attrs...)
-	if _, err := m.ExecContext(ctx, b.String()); err != nil {
-		return fmt.Errorf("create table: %w", err)
-	}
-	return nil
+	s.tableAttr(b, add.T.Attrs...)
+	s.append(&migrate.Change{
+		Cmd:     b.String(),
+		Source:  add,
+		Reverse: Build("DROP TABLE").Table(add.T).String(),
+		Comment: fmt.Sprintf("Create %q table", add.T.Name),
+	})
 }
 
-// dropTable builds and executes the query for dropping a table from a schema.
-func (m *migrate) dropTable(ctx context.Context, drop *schema.DropTable) 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)
 	if sqlx.Has(drop.Extra, &schema.IfExists{}) {
 		b.P("IF EXISTS")
 	}
-	if _, err := m.ExecContext(ctx, b.String()); err != nil {
-		return fmt.Errorf("drop table: %w", err)
-	}
-	return nil
+	s.append(&migrate.Change{
+		Cmd:     b.String(),
+		Source:  drop,
+		Comment: fmt.Sprintf("Drop %q table", drop.T.Name),
+	})
 }
 
-// modifyTable builds and executes the queries for bringing the table into its modified state.
-func (m *migrate) modifyTable(ctx context.Context, modify *schema.ModifyTable) error {
+// modifyTable builds and appends the migrate.Changes for bringing
+// the table into its modified state.
+func (s *state) modifyTable(modify *schema.ModifyTable) error {
 	var changes [2][]schema.Change
 	for _, change := range skipAutoChanges(modify.Changes) {
 		switch change := change.(type) {
@@ -171,55 +237,77 @@ func (m *migrate) modifyTable(ctx context.Context, modify *schema.ModifyTable) e
 	}
 	for i := range changes {
 		if len(changes[i]) > 0 {
-			if err := m.alterTable(ctx, modify.T, changes[i]); err != nil {
-				return err
-			}
+			s.alterTable(modify.T, changes[i])
 		}
 	}
 	return nil
 }
 
-// alterTable modifies the given table by executing on it a list of changes in one SQL statement.
-func (m *migrate) alterTable(ctx context.Context, t *schema.Table, changes []schema.Change) error {
-	b := Build("ALTER TABLE").Table(t)
+// alterTable modifies the given table by executing on it a list of
+// changes in one SQL statement.
+func (s *state) alterTable(t *schema.Table, changes []schema.Change) {
+	var (
+		reversible = true
+		reverse    = Build("ALTER TABLE").Table(t)
+		b          = Build("ALTER TABLE").Table(t)
+	)
 	b.MapComma(changes, func(i int, b *sqlx.Builder) {
 		switch change := changes[i].(type) {
 		case *schema.AddColumn:
 			b.P("ADD COLUMN")
-			m.column(b, t, change.C)
+			s.column(b, t, change.C)
+			reverse.P("DROP COLUMN").Ident(change.C.Name)
 		case *schema.ModifyColumn:
 			b.P("MODIFY COLUMN")
-			m.column(b, t, change.To)
+			s.column(b, t, change.To)
+			reverse.P("MODIFY COLUMN")
+			s.column(reverse, t, change.From)
 		case *schema.DropColumn:
 			b.P("DROP COLUMN").Ident(change.C.Name)
+			reversible = false
 		case *schema.AddIndex:
 			b.P("ADD")
 			if change.I.Unique {
 				b.P("UNIQUE")
 			}
 			b.P("INDEX").Ident(change.I.Name)
-			m.indexParts(b, change.I.Parts)
-			m.attr(b, change.I.Attrs...)
+			s.indexParts(b, change.I.Parts)
+			s.attr(b, change.I.Attrs...)
+			reverse.P("DROP INDEX").Ident(change.I.Name)
 		case *schema.DropIndex:
 			b.P("DROP INDEX").Ident(change.I.Name)
+			reversible = false
 		case *schema.AddForeignKey:
 			b.P("ADD")
-			m.fks(b, change.F)
+			s.fks(b, change.F)
+			reverse.P("DROP FOREIGN KEY").Ident(change.F.Symbol)
 		case *schema.DropForeignKey:
 			b.P("DROP FOREIGN KEY").Ident(change.F.Symbol)
+			reversible = false
 		case *schema.AddAttr:
-			m.tableAttr(b, change.A)
+			s.tableAttr(b, change.A)
+			// Unsupported reverse operation.
+			reversible = false
 		case *schema.ModifyAttr:
-			m.tableAttr(b, change.To)
+			s.tableAttr(b, change.To)
+			s.tableAttr(reverse, change.From)
 		}
 	})
-	if _, err := m.ExecContext(ctx, b.String()); err != nil {
-		return fmt.Errorf("alter table: %w", err)
+	change := &migrate.Change{
+		Cmd: b.String(),
+		Source: &schema.ModifyTable{
+			T:       t,
+			Changes: changes,
+		},
+		Comment: fmt.Sprintf("Modify %q table", t.Name),
 	}
-	return nil
+	if reversible {
+		change.Reverse = reverse.String()
+	}
+	s.append(change)
 }
 
-func (m *migrate) column(b *sqlx.Builder, t *schema.Table, c *schema.Column) {
+func (s *state) column(b *sqlx.Builder, t *schema.Table, c *schema.Column) {
 	b.Ident(c.Name).P(mustFormat(c.Type.Type))
 	if !c.Type.Null {
 		b.P("NOT")
@@ -236,7 +324,7 @@ func (m *migrate) column(b *sqlx.Builder, t *schema.Table, c *schema.Column) {
 	}
 	// Add manually the JSON_VALID constraint for older
 	// versions < 10.4.3. See Driver.checks for full info.
-	if _, ok := c.Type.Type.(*schema.JSONType); ok && m.mariadb() && m.compareV("10.4.3") == -1 && !sqlx.Has(c.Attrs, &Check{}) {
+	if _, ok := c.Type.Type.(*schema.JSONType); ok && s.mariadb() && s.compareV("10.4.3") == -1 && !sqlx.Has(c.Attrs, &Check{}) {
 		b.P("CHECK").Wrap(func(b *sqlx.Builder) {
 			b.WriteString(fmt.Sprintf("json_valid(`%s`)", c.Name))
 		})
@@ -246,7 +334,7 @@ func (m *migrate) column(b *sqlx.Builder, t *schema.Table, c *schema.Column) {
 		case *schema.Collation:
 			// Define the collation explicitly
 			// in case it is not the default.
-			if m.collation(t) != a.V {
+			if s.collation(t) != a.V {
 				b.P("COLLATE", a.V)
 			}
 		case *OnUpdate:
@@ -257,12 +345,12 @@ func (m *migrate) column(b *sqlx.Builder, t *schema.Table, c *schema.Column) {
 				b.P(strconv.FormatInt(a.V, 10))
 			}
 		default:
-			m.attr(b, a)
+			s.attr(b, a)
 		}
 	}
 }
 
-func (m *migrate) indexParts(b *sqlx.Builder, parts []*schema.IndexPart) {
+func (s *state) indexParts(b *sqlx.Builder, parts []*schema.IndexPart) {
 	b.Wrap(func(b *sqlx.Builder) {
 		b.MapComma(parts, func(i int, b *sqlx.Builder) {
 			switch part := parts[i]; {
@@ -280,7 +368,7 @@ func (m *migrate) indexParts(b *sqlx.Builder, parts []*schema.IndexPart) {
 	})
 }
 
-func (m *migrate) fks(b *sqlx.Builder, fks ...*schema.ForeignKey) {
+func (s *state) fks(b *sqlx.Builder, fks ...*schema.ForeignKey) {
 	b.MapComma(fks, func(i int, b *sqlx.Builder) {
 		fk := fks[i]
 		if fk.Symbol != "" {
@@ -309,7 +397,7 @@ func (m *migrate) fks(b *sqlx.Builder, fks ...*schema.ForeignKey) {
 
 // tableAttr writes the given table attribute to the SQL
 // statement builder when a table is created or altered.
-func (m *migrate) tableAttr(b *sqlx.Builder, attrs ...schema.Attr) {
+func (s *state) tableAttr(b *sqlx.Builder, attrs ...schema.Attr) {
 	for _, a := range attrs {
 		switch a := a.(type) {
 		case *AutoIncrement:
@@ -320,22 +408,26 @@ func (m *migrate) tableAttr(b *sqlx.Builder, attrs ...schema.Attr) {
 		case *schema.Charset:
 			b.P("CHARACTER SET", a.V)
 		default:
-			m.attr(b, a)
+			s.attr(b, a)
 		}
 	}
 }
 
 // collation returns the table collation from its attributes
 // or from the default defined in the schema or the database.
-func (m *migrate) collation(t *schema.Table) string {
+func (s *state) collation(t *schema.Table) string {
 	var c schema.Collation
 	if sqlx.Has(t.Attrs, &c) || t.Schema != nil && sqlx.Has(t.Schema.Attrs, &c) {
 		return c.V
 	}
-	return m.collate
+	return s.collate
+}
+
+func (s *state) append(c *migrate.Change) {
+	s.Changes = append(s.Changes, c)
 }
 
-func (*migrate) attr(b *sqlx.Builder, attrs ...schema.Attr) {
+func (*state) attr(b *sqlx.Builder, attrs ...schema.Attr) {
 	for _, a := range attrs {
 		switch a := a.(type) {
 		case *schema.Collation:
diff --git a/sql/schema/migrate.go b/sql/schema/migrate.go
index c919e574..ca53866c 100644
--- a/sql/schema/migrate.go
+++ b/sql/schema/migrate.go
@@ -231,8 +231,8 @@ type (
 	// drivers for executing schema changes.
 	Execer interface {
 		// Exec responsible for executing the given changeset.
-		// An error may return from Exec if the driver does not
-		// know how to execute a change.
+		// An error may return from Exec if the driver does not know
+		// how to exec a change.
 		Exec(ctx context.Context, changes []Change) error
 	}
 )


Test patch:
diff --git a/sql/migrate/migrate_test.go b/sql/migrate/migrate_test.go
new file mode 100644
index 00000000..0dcdfc9f
--- /dev/null
+++ b/sql/migrate/migrate_test.go
@@ -0,0 +1,184 @@
+// 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 migrate_test
+
+import (
+	"context"
+	"database/sql"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"ariga.io/atlas/sql/migrate"
+	"ariga.io/atlas/sql/schema"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestDir_Plan(t *testing.T) {
+	var (
+		m   = &mockDriver{}
+		ctx = context.Background()
+	)
+	dir, err := migrate.NewDir(
+		migrate.DirPath("testdata"),
+		migrate.DirGlob("*.up.sql"),
+		migrate.DirConn(m),
+	)
+	require.NoError(t, err)
+	plan, err := dir.Plan(ctx, "plan_name", migrate.Realm(nil))
+	require.Equal(t, migrate.ErrNoPlan, err)
+	require.Nil(t, plan)
+	require.Equal(t, []string{"CREATE TABLE t(c int);"}, m.executed)
+
+	m.executed = nil
+	dir, err = migrate.NewDir(
+		migrate.DirPath("testdata"),
+		migrate.DirGlob("*.sql"),
+		migrate.DirConn(m),
+	)
+	require.NoError(t, err)
+	plan, err = dir.Plan(ctx, "plan_name", migrate.Realm(nil))
+	require.Equal(t, migrate.ErrNoPlan, err)
+	require.Nil(t, plan)
+	require.Equal(t, []string{"DROP TABLE IF EXISTS t;", "CREATE TABLE t(c int);", "CREATE TABLE t(c int);"}, m.executed)
+
+	m.changes = append(m.changes, &schema.AddTable{T: &schema.Table{Name: "t1"}}, &schema.AddTable{T: &schema.Table{Name: "t2"}})
+	m.plan = &migrate.Plan{
+		Name: "add_t1_t2",
+		Changes: []*migrate.Change{
+			{Cmd: "CREATE TABLE t1(c int);"},
+			{Cmd: "CREATE TABLE t2(c int);"},
+		},
+	}
+	plan, err = dir.Plan(ctx, "plan_name", migrate.Realm(nil))
+	require.NoError(t, err)
+	require.NotNil(t, plan)
+}
+
+func TestDir_WritePlan_Compact(t *testing.T) {
+	var (
+		f   = &mockFS{}
+		m   = &mockDriver{}
+		ctx = context.Background()
+	)
+	dir, err := migrate.NewDir(
+		migrate.DirFS(f),
+		migrate.DirConn(m),
+	)
+	require.NoError(t, err)
+	err = dir.WritePlan(&migrate.Plan{
+		Name: "add_t1_t2",
+		Changes: []*migrate.Change{
+			{Cmd: "CREATE TABLE t1 (c int)"},
+			{Cmd: "CREATE TABLE t2 (c int);"},
+		},
+	})
+	require.NoError(t, err)
+	require.Equal(t, "CREATE TABLE t1 (c int);\nCREATE TABLE t2 (c int);\n", f.files[0].F)
+
+	f.files = nil
+	dir, err = migrate.NewDir(
+		migrate.DirFS(f),
+		migrate.DirConn(m),
+		migrate.DirTemplates("{{.Name}}.sql", `{{range $c := .Changes}}{{printf "--%s\n%s;\n" $c.Comment $c.Cmd}}{{end}}`),
+	)
+	require.NoError(t, err)
+	err = dir.WritePlan(&migrate.Plan{
+		Name: "add_t1_t2",
+		Changes: []*migrate.Change{
+			{Cmd: "CREATE TABLE t1 (c int)", Comment: "Create a new table named t1."},
+			{Cmd: "CREATE TABLE t2 (c int)", Comment: "Create a new table named t2."},
+		},
+	})
+	require.NoError(t, err)
+	require.Equal(t, "add_t1_t2.sql", f.files[0].N)
+	require.Equal(t, "--Create a new table named t1.\nCREATE TABLE t1 (c int);\n--Create a new table named t2.\nCREATE TABLE t2 (c int);\n", f.files[0].F)
+
+	err = dir.WritePlan(&migrate.Plan{
+		Name: "add_t3",
+		Changes: []*migrate.Change{
+			{Cmd: "CREATE TABLE t3 (c int)", Comment: "Create a new table named t3."},
+		},
+	})
+	require.NoError(t, err)
+	require.Equal(t, "add_t3.sql", f.files[1].N)
+	require.Equal(t, "--Create a new table named t3.\nCREATE TABLE t3 (c int);\n", f.files[1].F)
+
+	m.changes = append(m.changes, &schema.AddTable{T: &schema.Table{Name: "t1"}}, &schema.AddTable{T: &schema.Table{Name: "t2"}})
+	m.plan = &migrate.Plan{Name: "schema", Changes: []*migrate.Change{{Cmd: "CREATE TABLE t1 (c int)"}, {Cmd: "CREATE TABLE t2 (c int)"}, {Cmd: "CREATE TABLE t3 (c int)"}}}
+	err = dir.Compact(ctx, "schema", -1)
+	require.NoError(t, err)
+	require.Len(t, f.files, 1)
+	require.Equal(t, "schema.sql", f.files[0].N)
+	require.Equal(t, "--\nCREATE TABLE t1 (c int);\n--\nCREATE TABLE t2 (c int);\n--\nCREATE TABLE t3 (c int);\n", f.files[0].F)
+
+}
+
+type mockFS struct {
+	fs.GlobFS
+	files []struct{ N, F string }
+}
+
+func (f *mockFS) Glob(pattern string) ([]string, error) {
+	var matches []string
+	for i := range f.files {
+		match, err := filepath.Match(pattern, f.files[i].N)
+		if err != nil {
+			return nil, err
+		}
+		if match {
+			matches = append(matches, f.files[i].N)
+		}
+	}
+	return matches, nil
+}
+
+func (f *mockFS) ReadFile(name string) ([]byte, error) {
+	for i := range f.files {
+		if f.files[i].N == name {
+			return []byte(f.files[i].F), nil
+		}
+	}
+	return nil, os.ErrNotExist
+}
+
+func (f *mockFS) WriteFile(name string, data []byte, _ fs.FileMode) error {
+	f.files = append(f.files, struct{ N, F string }{N: name, F: string(data)})
+	return nil
+}
+
+func (f *mockFS) RemoveFile(name string) error {
+	for i := range f.files {
+		if f.files[i].N == name {
+			f.files = append(f.files[:i], f.files[i+1:]...)
+			return nil
+		}
+	}
+	return os.ErrNotExist
+}
+
+type mockDriver struct {
+	migrate.Driver
+	plan     *migrate.Plan
+	changes  []schema.Change
+	executed []string
+}
+
+func (m *mockDriver) ExecContext(_ context.Context, query string, _ ...interface{}) (sql.Result, error) {
+	m.executed = append(m.executed, query)
+	return nil, nil
+}
+
+func (m mockDriver) InspectRealm(context.Context, *schema.InspectRealmOption) (*schema.Realm, error) {
+	return &schema.Realm{}, nil
+}
+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) {
+	return m.plan, nil
+}
diff --git a/sql/migrate/testdata/1.down.sql b/sql/migrate/testdata/1.down.sql
new file mode 100644
index 00000000..69fc596f
--- /dev/null
+++ b/sql/migrate/testdata/1.down.sql
@@ -0,0 +1,1 @@
+DROP TABLE IF EXISTS t;
\ No newline at end of file
diff --git a/sql/migrate/testdata/1.sql b/sql/migrate/testdata/1.sql
new file mode 100644
index 00000000..5a3f49dc
--- /dev/null
+++ b/sql/migrate/testdata/1.sql
@@ -0,0 +1,1 @@
+CREATE TABLE t(c int);
\ No newline at end of file
diff --git a/sql/migrate/testdata/1.up.sql b/sql/migrate/testdata/1.up.sql
new file mode 100644
index 00000000..5a3f49dc
--- /dev/null
+++ b/sql/migrate/testdata/1.up.sql
@@ -0,0 +1,1 @@
+CREATE TABLE t(c int);
\ No newline at end of file
diff --git a/sql/mysql/migrate_test.go b/sql/mysql/migrate_test.go
index 8679ce97..6b8802d9 100644
--- a/sql/mysql/migrate_test.go
+++ b/sql/mysql/migrate_test.go
@@ -9,6 +9,7 @@ import (
 	"testing"
 
 	"ariga.io/atlas/sql/internal/sqltest"
+	"ariga.io/atlas/sql/migrate"
 	"ariga.io/atlas/sql/schema"
 
 	"github.com/DATA-DOG/go-sqlmock"
@@ -36,7 +37,7 @@ func TestMigrate_Exec(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))
-	err = migrate.Exec(context.Background(), []schema.Change{
+	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"}}}},
 		&schema.DropTable{T: &schema.Table{Name: "users"}},
@@ -65,7 +66,7 @@ func TestMigrate_Exec(t *testing.T) {
 		},
 	})
 	require.NoError(t, err)
-	err = migrate.Exec(context.Background(), func() []schema.Change {
+	err = migrate.ApplyChanges(context.Background(), func() []schema.Change {
 		users := &schema.Table{
 			Name: "users",
 			Columns: []*schema.Column{
@@ -140,7 +141,7 @@ func TestMigrate_DetachCycles(t *testing.T) {
 		WillReturnResult(sqlmock.NewResult(0, 0))
 	mk.ExpectExec(sqltest.Escape("ALTER TABLE `workplaces` ADD CONSTRAINT `owner` FOREIGN KEY (`owner_id`) REFERENCES `users` (`id`)")).
 		WillReturnResult(sqlmock.NewResult(0, 0))
-	err = migrate.Exec(context.Background(), func() []schema.Change {
+	err = migrate.ApplyChanges(context.Background(), func() []schema.Change {
 		users := &schema.Table{
 			Name: "users",
 			Columns: []*schema.Column{
@@ -169,7 +170,116 @@ func TestMigrate_DetachCycles(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func newMigrate(version string) (schema.Execer, *mock, error) {
+func TestPlanChanges(t *testing.T) {
+	tests := []struct {
+		changes []schema.Change
+		plan    *migrate.Plan
+	}{
+		{
+			changes: []schema.Change{
+				&schema.AddSchema{S: &schema.Schema{Name: "test", Attrs: []schema.Attr{&schema.Charset{V: "latin"}}}},
+			},
+			plan: &migrate.Plan{
+				Reversible: true,
+				Changes:    []*migrate.Change{{Cmd: "CREATE DATABASE `test` CHARACTER SET latin", Reverse: "DROP DATABASE `test`"}}},
+		},
+		{
+			changes: []schema.Change{
+				&schema.DropSchema{S: &schema.Schema{Name: "atlas", Attrs: []schema.Attr{&schema.Charset{V: "latin"}}}},
+			},
+			plan: &migrate.Plan{
+				Changes: []*migrate.Change{{Cmd: "DROP DATABASE `atlas`"}},
+			},
+		},
+		{
+			changes: []schema.Change{
+				&schema.DropSchema{S: &schema.Schema{Name: "atlas", Attrs: []schema.Attr{&schema.Charset{V: "latin"}}}},
+			},
+			plan: &migrate.Plan{
+				Changes: []*migrate.Change{{Cmd: "DROP DATABASE `atlas`"}},
+			},
+		},
+		{
+			changes: []schema.Change{
+				&schema.AddTable{
+					T: &schema.Table{
+						Name: "posts",
+						Columns: []*schema.Column{
+							{Name: "id", Type: &schema.ColumnType{Type: &schema.IntegerType{T: "bigint"}}},
+							{Name: "text", Type: &schema.ColumnType{Type: &schema.StringType{T: "text"}, Null: true}},
+						},
+					},
+				},
+			},
+			plan: &migrate.Plan{
+				Reversible: true,
+				Changes:    []*migrate.Change{{Cmd: "CREATE TABLE `posts` (`id` bigint NOT NULL, `text` text NULL)", Reverse: "DROP TABLE `posts`"}},
+			},
+		},
+		{
+			changes: []schema.Change{
+				&schema.DropTable{T: &schema.Table{Name: "posts"}},
+			},
+			plan: &migrate.Plan{
+				Changes: []*migrate.Change{{Cmd: "DROP TABLE `posts`"}},
+			},
+		},
+		{
+			changes: []schema.Change{
+				func() schema.Change {
+					users := &schema.Table{
+						Name: "users",
+						Columns: []*schema.Column{
+							{Name: "id", Type: &schema.ColumnType{Type: &schema.IntegerType{T: "bigint"}}},
+						},
+					}
+					return &schema.ModifyTable{
+						T: users,
+						Changes: []schema.Change{
+							&schema.AddColumn{
+								C: &schema.Column{Name: "name", Type: &schema.ColumnType{Type: &schema.StringType{T: "varchar(255)"}}},
+							},
+							&schema.AddIndex{
+								I: &schema.Index{
+									Name: "id_key",
+									Parts: []*schema.IndexPart{
+										{C: users.Columns[0]},
+									},
+									Attrs: []schema.Attr{
+										&schema.Comment{Text: "comment"},
+									},
+								},
+							},
+						},
+					}
+				}(),
+			},
+			plan: &migrate.Plan{
+				Reversible: true,
+				Changes: []*migrate.Change{
+					{
+						Cmd:     "ALTER TABLE `users` ADD COLUMN `name` varchar(255) NOT NULL, ADD INDEX `id_key` (`id`) COMMENT \"comment\"",
+						Reverse: "ALTER TABLE `users` DROP COLUMN `name` DROP INDEX `id_key`",
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		db, _, err := newMigrate("8.0.13")
+		require.NoError(t, err)
+		plan, err := db.PlanChanges(context.Background(), "plan", tt.changes)
+		require.NoError(t, err)
+		require.Equal(t, tt.plan.Reversible, plan.Reversible)
+		require.Equal(t, tt.plan.Transactional, plan.Transactional)
+		for i, c := range plan.Changes {
+			require.Equal(t, tt.plan.Changes[i].Cmd, c.Cmd)
+			require.Equal(t, tt.plan.Changes[i].Reverse, c.Reverse)
+		}
+	}
+}
+
+func newMigrate(version string) (migrate.PlanApplier, *mock, error) {
 	db, m, err := sqlmock.New()
 	if err != nil {
 		return nil, nil, err
@@ -180,5 +290,5 @@ func newMigrate(version string) (schema.Execer, *mock, error) {
 	if err != nil {
 		return nil, nil, err
 	}
-	return drv, mk, nil
+	return drv.Execer.(migrate.PlanApplier), mk, nil
 }