Repository: ariga/atlas
Pull request #310 (2021-12-14 19:20:40)

sql: inline/expose diff, exec and inspect on the different driver


Patch:
diff --git a/cmd/action/mux.go b/cmd/action/mux.go
index 0a51f5a0..b93b1edb 100644
--- a/cmd/action/mux.go
+++ b/cmd/action/mux.go
@@ -19,14 +19,19 @@ type (
 
 	// Driver implements the Atlas interface.
 	Driver struct {
-		schema.Differ
-		schema.Execer
-		schema.Inspector
+		driver
 		MarshalSpec   func(v interface{}, marshaler schemaspec.Marshaler) ([]byte, error)
 		UnmarshalSpec func(data []byte, unmarshaler schemaspec.Unmarshaler, v interface{}) error
 		interceptor   *interceptor
 	}
 
+	// A schema driver.
+	driver interface {
+		schema.Differ
+		schema.Execer
+		schema.Inspector
+	}
+
 	schemaUnmarshal struct {
 		unmarshalSpec func(data []byte, unmarshaler schemaspec.Unmarshaler, v interface{}) error
 		unmarshaler   schemaspec.Unmarshaler
diff --git a/cmd/action/provider.go b/cmd/action/provider.go
index 0f5387a4..0917c2ad 100644
--- a/cmd/action/provider.go
+++ b/cmd/action/provider.go
@@ -23,9 +23,7 @@ func mysqlProvider(dsn string) (*Driver, error) {
 		return nil, err
 	}
 	return &Driver{
-		Differ:        drv.Diff(),
-		Execer:        drv.Migrate(),
-		Inspector:     drv,
+		driver:        drv,
 		interceptor:   i,
 		MarshalSpec:   mysql.MarshalSpec,
 		UnmarshalSpec: mysql.UnmarshalSpec,
@@ -43,9 +41,7 @@ func postgresProvider(dsn string) (*Driver, error) {
 		return nil, err
 	}
 	return &Driver{
-		Differ:        drv.Diff(),
-		Execer:        drv.Migrate(),
-		Inspector:     drv,
+		driver:        drv,
 		interceptor:   i,
 		MarshalSpec:   postgres.MarshalSpec,
 		UnmarshalSpec: postgres.UnmarshalSpec,
diff --git a/sql/mysql/diff.go b/sql/mysql/diff.go
index e9acc814..dbc69ad1 100644
--- a/sql/mysql/diff.go
+++ b/sql/mysql/diff.go
@@ -15,14 +15,7 @@ import (
 )
 
 // A diff provides a MySQL implementation for sqlx.DiffDriver.
-type diff struct{ *Driver }
-
-// Diff returns a MySQL schema differ.
-func (d *Driver) Diff() schema.Differ {
-	return &sqlx.Diff{
-		DiffDriver: &diff{Driver: d},
-	}
-}
+type diff struct{ conn }
 
 // SchemaAttrDiff returns a changeset for migrating schema attributes from one state to the other.
 func (d *diff) SchemaAttrDiff(from, to *schema.Schema) []schema.Change {
diff --git a/sql/mysql/driver.go b/sql/mysql/driver.go
index 58f512bd..736e9eaa 100644
--- a/sql/mysql/driver.go
+++ b/sql/mysql/driver.go
@@ -5,10 +5,7 @@
 package mysql
 
 import (
-	"context"
-	"database/sql"
 	"fmt"
-	"strconv"
 	"strings"
 
 	"ariga.io/atlas/sql/internal/sqlx"
@@ -17,550 +14,43 @@ import (
 	"golang.org/x/mod/semver"
 )
 
-// Driver represents a MySQL driver for introspecting database schemas
-// and apply migrations changes on them.
-type Driver struct {
-	schema.ExecQuerier
-	// System variables that are set on `Open`.
-	version string
-	collate string
-	charset string
-}
+type (
+	// Driver represents a MySQL driver for introspecting database schemas,
+	// generating diff between schema elements and apply migrations changes.
+	Driver struct {
+		conn
+		schema.Differ
+		schema.Execer
+		schema.Inspector
+	}
+
+	// database connection and its information.
+	conn struct {
+		schema.ExecQuerier
+		// System variables that are set on `Open`.
+		version string
+		collate string
+		charset string
+	}
+)
 
 // Open opens a new MySQL driver.
 func Open(db schema.ExecQuerier) (*Driver, error) {
-	drv := &Driver{ExecQuerier: db}
-	if err := db.QueryRow(variablesQuery).Scan(&drv.version, &drv.collate, &drv.charset); err != nil {
+	c := conn{ExecQuerier: db}
+	if err := db.QueryRow(variablesQuery).Scan(&c.version, &c.collate, &c.charset); err != nil {
 		return nil, fmt.Errorf("mysql: scanning system variables: %w", err)
 	}
-	return drv, nil
-}
-
-// InspectRealm returns schema descriptions of all resources in the given realm.
-func (d *Driver) InspectRealm(ctx context.Context, opts *schema.InspectRealmOption) (*schema.Realm, error) {
-	schemas, err := d.schemas(ctx, opts)
-	if err != nil {
-		return nil, err
-	}
-	realm := &schema.Realm{Schemas: schemas, Attrs: []schema.Attr{&schema.Charset{V: d.charset}, &schema.Collation{V: d.collate}}}
-	for _, s := range schemas {
-		names, err := d.tableNames(ctx, s.Name, nil)
-		if err != nil {
-			return nil, err
-		}
-		for _, name := range names {
-			t, err := d.inspectTable(ctx, name, &schema.InspectTableOptions{Schema: s.Name}, s)
-			if err != nil {
-				return nil, err
-			}
-			s.Tables = append(s.Tables, t)
-		}
-		s.Realm = realm
-	}
-	sqlx.LinkSchemaTables(schemas)
-	return realm, nil
-}
-
-// InspectSchema returns schema descriptions of all tables in the given schema.
-func (d *Driver) InspectSchema(ctx context.Context, name string, opts *schema.InspectOptions) (*schema.Schema, error) {
-	schemas, err := d.schemas(ctx, &schema.InspectRealmOption{
-		Schemas: []string{name},
-	})
-	if err != nil {
-		return nil, err
-	}
-	if len(schemas) == 0 {
-		return nil, &schema.NotExistError{
-			Err: fmt.Errorf("mysql: schema %q was not found", name),
-		}
-	}
-	names, err := d.tableNames(ctx, name, opts)
-	if err != nil {
-		return nil, err
-	}
-	s := schemas[0]
-	for _, name := range names {
-		t, err := d.inspectTable(ctx, name, &schema.InspectTableOptions{Schema: s.Name}, s)
-		if err != nil {
-			return nil, err
-		}
-		s.Tables = append(s.Tables, t)
-	}
-	sqlx.LinkSchemaTables(schemas)
-	s.Realm = &schema.Realm{Schemas: schemas, Attrs: []schema.Attr{&schema.Charset{V: d.charset}, &schema.Collation{V: d.collate}}}
-	return s, nil
-}
-
-// InspectTable returns the schema description of the given table.
-func (d *Driver) InspectTable(ctx context.Context, name string, opts *schema.InspectTableOptions) (*schema.Table, error) {
-	return d.inspectTable(ctx, name, opts, nil)
-}
-
-func (d *Driver) inspectTable(ctx context.Context, name string, opts *schema.InspectTableOptions, top *schema.Schema) (*schema.Table, error) {
-	t, err := d.table(ctx, name, opts)
-	if err != nil {
-		return nil, err
-	}
-	if top != nil {
-		// Link the table to its top element if provided.
-		t.Schema = top
-	}
-	if err := d.columns(ctx, t); err != nil {
-		return nil, err
-	}
-	if err := d.indexes(ctx, t); err != nil {
-		return nil, err
-	}
-	if err := d.fks(ctx, t); err != nil {
-		return nil, err
-	}
-	if err := d.checks(ctx, t); err != nil {
-		return nil, err
-	}
-	return t, nil
-}
-
-// schemas returns the list of the schemas in the database.
-func (d *Driver) schemas(ctx context.Context, opts *schema.InspectRealmOption) ([]*schema.Schema, error) {
-	var (
-		args  []interface{}
-		query = schemasQuery
-	)
-	if opts != nil && len(opts.Schemas) > 0 {
-		query += " WHERE `SCHEMA_NAME` IN (" + strings.Repeat("?, ", len(opts.Schemas)-1) + "?)"
-		for _, s := range opts.Schemas {
-			args = append(args, s)
-		}
-	} else {
-		query += " WHERE `SCHEMA_NAME` NOT IN ('mysql', 'information_schema', 'performance_schema', 'sys')"
-	}
-	rows, err := d.QueryContext(ctx, query, args...)
-	if err != nil {
-		return nil, fmt.Errorf("mysql: querying schemas: %w", err)
-	}
-	defer rows.Close()
-	var schemas []*schema.Schema
-	for rows.Next() {
-		var name, charset, collation string
-		if err := rows.Scan(&name, &charset, &collation); err != nil {
-			return nil, err
-		}
-		schemas = append(schemas, &schema.Schema{
-			Name: name,
-			Attrs: []schema.Attr{
-				&schema.Charset{
-					V: charset,
-				},
-				&schema.Collation{
-					V: collation,
-				},
-			},
-		})
-	}
-	return schemas, nil
-}
-
-// table returns the table from the database, or a NotExistError if the table was not found.
-func (d *Driver) table(ctx context.Context, name string, opts *schema.InspectTableOptions) (*schema.Table, error) {
-	var (
-		args  = []interface{}{name}
-		query = tableQuery
-	)
-	if opts != nil && opts.Schema != "" {
-		query = tableSchemaQuery
-		args = append(args, opts.Schema)
-	}
-	row := d.QueryRowContext(ctx, query, args...)
-	var (
-		autoinc                              sql.NullInt64
-		tSchema, charset, collation, comment sql.NullString
-	)
-	if err := row.Scan(&tSchema, &charset, &collation, &autoinc, &comment); err != nil {
-		if err == sql.ErrNoRows {
-			return nil, &schema.NotExistError{
-				Err: fmt.Errorf("mysql: table %q was not found", name),
-			}
-		}
-		return nil, err
-	}
-	t := &schema.Table{Name: name, Schema: &schema.Schema{Name: tSchema.String}}
-	if sqlx.ValidString(charset) {
-		t.Attrs = append(t.Attrs, &schema.Charset{
-			V: charset.String,
-		})
-	}
-	if sqlx.ValidString(collation) {
-		t.Attrs = append(t.Attrs, &schema.Collation{
-			V: collation.String,
-		})
-	}
-	if sqlx.ValidString(comment) {
-		t.Attrs = append(t.Attrs, &schema.Comment{
-			Text: comment.String,
-		})
-	}
-	if autoinc.Valid {
-		t.Attrs = append(t.Attrs, &AutoIncrement{
-			V: autoinc.Int64,
-		})
-	}
-	return t, nil
-}
-
-// columns queries and appends the columns of the given table.
-func (d *Driver) columns(ctx context.Context, t *schema.Table) error {
-	rows, err := d.QueryContext(ctx, columnsQuery, t.Schema.Name, t.Name)
-	if err != nil {
-		return fmt.Errorf("mysql: querying %q columns: %w", t.Name, err)
-	}
-	defer rows.Close()
-	for rows.Next() {
-		if err := d.addColumn(t, rows); err != nil {
-			return fmt.Errorf("mysql: %w", err)
-		}
-	}
-	return rows.Err()
-}
-
-// addColumn scans the current row and adds a new column from it to the table.
-func (d *Driver) addColumn(t *schema.Table, rows *sql.Rows) error {
-	var name, typ, comment, nullable, key, defaults, extra, charset, collation sql.NullString
-	if err := rows.Scan(&name, &typ, &comment, &nullable, &key, &defaults, &extra, &charset, &collation); err != nil {
-		return err
-	}
-	c := &schema.Column{
-		Name: name.String,
-		Type: &schema.ColumnType{
-			Raw:  typ.String,
-			Null: nullable.String == "YES",
-		},
-	}
-	ct, err := parseRawType(c.Type.Raw)
-	if err != nil {
-		return err
-	}
-	c.Type.Type = ct
-	if err := extraAttr(c, extra.String); err != nil {
-		return err
-	}
-	if sqlx.ValidString(defaults) {
-		x := defaults.String
-		// From MariaDB 10.2.7, literals are quoted to distinguish them from expressions.
-		if d.mariadb() && strings.HasPrefix(x, "'") && strings.HasSuffix(x, "'") {
-			x = x[1 : len(x)-1]
-		}
-		c.Default = &schema.RawExpr{X: x}
-	}
-	if sqlx.ValidString(comment) {
-		c.Attrs = append(c.Attrs, &schema.Comment{
-			Text: comment.String,
-		})
-	}
-	if sqlx.ValidString(charset) {
-		c.Attrs = append(c.Attrs, &schema.Charset{
-			V: charset.String,
-		})
-	}
-	if sqlx.ValidString(collation) {
-		c.Attrs = append(c.Attrs, &schema.Collation{
-			V: collation.String,
-		})
-	}
-	t.Columns = append(t.Columns, c)
-	if key.String == "PRI" {
-		if t.PrimaryKey == nil {
-			t.PrimaryKey = &schema.Index{Table: t, Name: key.String}
-		}
-		t.PrimaryKey.Parts = append(t.PrimaryKey.Parts, &schema.IndexPart{
-			C:     c,
-			SeqNo: len(t.PrimaryKey.Parts),
-		})
-	}
-	return nil
-}
-
-func parseRawType(raw string) (schema.Type, error) {
-	parts, size, unsigned, err := parseColumn(raw)
-	if err != nil {
-		return nil, err
-	}
-	switch t := parts[0]; t {
-	case tBit:
-		return &BitType{
-			T: t,
-		}, nil
-	case tTinyInt, tSmallInt, tMediumInt, tInt, tBigInt:
-		if size == 1 {
-			return &schema.BoolType{
-				T: t,
-			}, nil
-		}
-		// For integer types, the size represents the display width and does not
-		// constrain the range of values that can be stored in the column.
-		// The storage byte-size is inferred from the type name (i.e TINYINT takes
-		// a single byte).
-		ft := &schema.IntegerType{
-			T:        t,
-			Unsigned: unsigned,
-		}
-		if attr := parts[len(parts)-1]; attr == "zerofill" && size != 0 {
-			ft.Attrs = []schema.Attr{
-				&DisplayWidth{
-					N: int(size),
-				},
-				&ZeroFill{
-					A: attr,
-				},
-			}
-		}
-		return ft, nil
-	case tNumeric, tDecimal:
-		dt := &schema.DecimalType{
-			T: t,
-		}
-		if len(parts) > 1 {
-			p, err := strconv.ParseInt(parts[1], 10, 64)
-			if err != nil {
-				return nil, fmt.Errorf("parse precision %q", parts[1])
-			}
-			dt.Precision = int(p)
-		}
-		if len(parts) > 2 {
-			s, err := strconv.ParseInt(parts[2], 10, 64)
-			if err != nil {
-				return nil, fmt.Errorf("parse scale %q", parts[1])
-			}
-			dt.Scale = int(s)
-		}
-		return dt, nil
-	case tFloat, tDouble, tReal:
-		ft := &schema.FloatType{
-			T: t,
-		}
-		if len(parts) > 1 {
-			p, err := strconv.ParseInt(parts[1], 10, 64)
-			if err != nil {
-				return nil, fmt.Errorf("parse precision %q", parts[1])
-			}
-			ft.Precision = int(p)
-		}
-		return ft, nil
-	case tBinary, tVarBinary:
-		return &schema.BinaryType{
-			T:    t,
-			Size: int(size),
-		}, nil
-	case tTinyBlob, tMediumBlob, tBlob, tLongBlob:
-		return &schema.BinaryType{
-			T: t,
-		}, nil
-	case tChar, tVarchar:
-		return &schema.StringType{
-			T:    t,
-			Size: int(size),
-		}, nil
-	case tTinyText, tMediumText, tText, tLongText:
-		return &schema.StringType{
-			T: t,
-		}, nil
-	case tEnum, tSet:
-		// Parse the enum values according to the MySQL format.
-		// github.com/mysql/mysql-server/blob/8.0/sql/field.cc#Field_enum::sql_type
-		rv := strings.TrimSuffix(strings.TrimPrefix(raw, "enum("), ")")
-		if rv == "" {
-			return nil, fmt.Errorf("mysql: unexpected enum type: %q", raw)
-		}
-		values := strings.Split(rv, "','")
-		for i := range values {
-			values[i] = strings.Trim(values[i], "'")
-		}
-		if t == tEnum {
-			return &schema.EnumType{
-				Values: values,
-			}, nil
-		}
-		return &SetType{
-			Values: values,
-		}, nil
-	case tDate, tDateTime, tTime, tTimestamp, tYear:
-		return &schema.TimeType{
-			T: t,
-		}, nil
-	case tJSON:
-		return &schema.JSONType{
-			T: t,
-		}, nil
-	case tPoint, tMultiPoint, tLineString, tMultiLineString, tPolygon, tMultiPolygon, tGeometry, tGeoCollection, tGeometryCollection:
-		return &schema.SpatialType{
-			T: t,
-		}, nil
-	default:
-		return &schema.UnsupportedType{
-			T: t,
-		}, nil
-	}
-}
-
-// indexes queries and appends the indexes of the given table.
-func (d *Driver) indexes(ctx context.Context, t *schema.Table) error {
-	query := indexesQuery
-	if d.supportsIndexExpr() {
-		query = indexesExprQuery
-	}
-	rows, err := d.QueryContext(ctx, query, t.Schema.Name, t.Name)
-	if err != nil {
-		return fmt.Errorf("mysql: querying %q indexes: %w", t.Name, err)
-	}
-	defer rows.Close()
-	if err := d.addIndexes(t, rows); err != nil {
-		return err
-	}
-	return rows.Err()
-}
-
-// addIndexes scans the rows and adds the indexes to the table.
-func (d *Driver) addIndexes(t *schema.Table, rows *sql.Rows) error {
-	names := make(map[string]*schema.Index)
-	for rows.Next() {
-		var (
-			nonuniq                                 bool
-			seqno                                   int
-			name, indexType                         string
-			column, subPart, expr, comment, collate sql.NullString
-		)
-		if err := rows.Scan(&name, &column, &nonuniq, &seqno, &indexType, &collate, &comment, &subPart, &expr); err != nil {
-			return fmt.Errorf("mysql: scanning index: %w", err)
-		}
-		// Ignore primary keys.
-		if name == "PRIMARY" {
-			continue
-		}
-		idx, ok := names[name]
-		if !ok {
-			idx = &schema.Index{
-				Name:   name,
-				Unique: !nonuniq,
-				Table:  t,
-				Attrs: []schema.Attr{
-					&IndexType{T: indexType},
-				},
-			}
-			if sqlx.ValidString(comment) {
-				idx.Attrs = append(t.Attrs, &schema.Comment{
-					Text: comment.String,
-				})
-			}
-			names[name] = idx
-			t.Indexes = append(t.Indexes, idx)
-		}
-		// Rows are ordered by SEQ_IN_INDEX that specifies the
-		// position of the column in the index definition.
-		part := &schema.IndexPart{
-			SeqNo: seqno,
-			Attrs: []schema.Attr{&schema.Collation{V: collate.String}},
-		}
-		switch {
-		case sqlx.ValidString(expr):
-			part.X = &schema.RawExpr{
-				X: expr.String,
-			}
-		case sqlx.ValidString(column):
-			part.C, ok = t.Column(column.String)
-			if !ok {
-				return fmt.Errorf("mysql: column %q was not found for index %q", column.String, idx.Name)
-			}
-			if sqlx.ValidString(subPart) {
-				n, err := strconv.Atoi(subPart.String)
-				if err != nil {
-					return fmt.Errorf("mysql: parse index prefix size %q: %w", subPart.String, err)
-				}
-				part.Attrs = append(part.Attrs, &SubPart{
-					Len: n,
-				})
-			}
-			part.C.Indexes = append(part.C.Indexes, idx)
-		default:
-			return fmt.Errorf("mysql: invalid part for index %q", idx.Name)
-		}
-		idx.Parts = append(idx.Parts, part)
-	}
-	return nil
-}
-
-// fks queries and appends the foreign keys of the given table.
-func (d *Driver) fks(ctx context.Context, t *schema.Table) error {
-	rows, err := d.QueryContext(ctx, fksQuery, t.Schema.Name, t.Name)
-	if err != nil {
-		return fmt.Errorf("mysql: querying %q foreign keys: %w", t.Name, err)
-	}
-	defer rows.Close()
-	if err := sqlx.ScanFKs(t, rows); err != nil {
-		return fmt.Errorf("mysql: %w", err)
-	}
-	return rows.Err()
-}
-
-// checks queries and appends the check constraints of the given table.
-func (d *Driver) checks(ctx context.Context, t *schema.Table) error {
-	query, ok := d.supportsCheck()
-	if !ok {
-		return nil
-	}
-	rows, err := d.QueryContext(ctx, query, t.Schema.Name, t.Name)
-	if err != nil {
-		return fmt.Errorf("mysql: querying %q check constraints: %w", t.Name, err)
-	}
-	defer rows.Close()
-	for rows.Next() {
-		var name, clause, enforced sql.NullString
-		if err := rows.Scan(&name, &clause, &enforced); err != nil {
-			return fmt.Errorf("mysql: %w", err)
-		}
-		check := &Check{
-			Name:     name.String,
-			Clause:   unescape(clause.String),
-			Enforced: clause.String != "NO",
-		}
-		t.Attrs = append(t.Attrs, check)
-		// In MariaDB, JSON is an alias to LONGTEXT, and the JSON_VALID
-		// CHECK constraint is automatically created for the column for
-		// versions >= 10.4.3. However, we expect tools like Atlas and
-		// Ent to manually add this CHECK for older versions of MariaDB.
-		if d.mariadb() {
-			c, ok := t.Column(check.Name)
-			if ok && c.Type.Raw == tLongText && check.Clause == fmt.Sprintf("json_valid(`%s`)", c.Name) {
-				c.Type.Raw = tJSON
-				c.Type.Type = &schema.JSONType{T: tJSON}
-			}
-		}
-
-	}
-	return rows.Err()
-}
-
-// tableNames returns a list of all tables exist in the schema.
-func (d *Driver) tableNames(ctx context.Context, schema string, opts *schema.InspectOptions) ([]string, error) {
-	query, args := tablesQuery, []interface{}{schema}
-	if opts != nil && len(opts.Tables) > 0 {
-		query += " AND `TABLE_NAME` IN (" + strings.Repeat("?, ", len(opts.Tables)-1) + "?)"
-		for _, s := range opts.Tables {
-			args = append(args, s)
-		}
-	}
-	rows, err := d.QueryContext(ctx, query, args...)
-	if err != nil {
-		return nil, fmt.Errorf("mysql: querying schema tables: %w", err)
-	}
-	names, err := sqlx.ScanStrings(rows)
-	if err != nil {
-		return nil, fmt.Errorf("mysql: scanning table names: %w", err)
-	}
-	return names, nil
+	return &Driver{
+		conn:      c,
+		Differ:    &sqlx.Diff{DiffDriver: &diff{c}},
+		Execer:    &migrate{c},
+		Inspector: &inspect{c},
+	}, nil
 }
 
 // supportsCheck reports if the connected database supports
 // the CHECK clause, and return the querying for getting them.
-func (d *Driver) supportsCheck() (string, bool) {
+func (d *conn) supportsCheck() (string, bool) {
 	v, q := "8.0.16", myChecksQuery
 	if d.mariadb() {
 		v, q = "10.2.1", marChecksQuery
@@ -570,26 +60,26 @@ func (d *Driver) supportsCheck() (string, bool) {
 
 // supportsIndexExpr reports if the connected database supports
 // index expressions (functional key part).
-func (d *Driver) supportsIndexExpr() bool {
+func (d *conn) supportsIndexExpr() bool {
 	return !d.mariadb() && d.compareV("8.0.13") != -1
 }
 
 // supportsDisplayWidth reports if the connected database supports
 // getting the display width information from the information schema.
-func (d *Driver) supportsDisplayWidth() bool {
+func (d *conn) supportsDisplayWidth() bool {
 	// MySQL v8.0.19 dropped the display width
 	// information from the information schema
 	return d.mariadb() || d.compareV("8.0.19") == -1
 }
 
 // mariadb reports if the Driver is connected to a MariaDB database.
-func (d *Driver) mariadb() bool {
+func (d *conn) mariadb() bool {
 	return strings.Index(d.version, "MariaDB") > 0
 }
 
 // compareV returns an integer comparing two versions according to
 // semantic version precedence.
-func (d *Driver) compareV(w string) int {
+func (d *conn) compareV(w string) int {
 	v := d.version
 	if d.mariadb() {
 		v = v[:strings.Index(v, "MariaDB")-1]
@@ -597,202 +87,6 @@ func (d *Driver) compareV(w string) int {
 	return semver.Compare("v"+v, "v"+w)
 }
 
-// parseColumn returns column parts, size and signed-info from a MySQL type.
-func parseColumn(typ string) (parts []string, size int64, unsigned bool, err error) {
-	switch parts = strings.FieldsFunc(typ, func(r rune) bool {
-		return r == '(' || r == ')' || r == ' ' || r == ','
-	}); parts[0] {
-	case tTinyInt, tSmallInt, tMediumInt, tInt, tBigInt:
-		if attr := parts[len(parts)-1]; attr == "unsigned" || attr == "zerofill" {
-			unsigned = true
-		}
-		if len(parts) > 2 || len(parts) == 2 && !unsigned {
-			size, err = strconv.ParseInt(parts[1], 10, 64)
-		}
-	case tBinary, tVarBinary, tChar, tVarchar:
-		size, err = strconv.ParseInt(parts[1], 10, 64)
-	}
-	if err != nil {
-		return nil, 0, false, fmt.Errorf("parse %q to int: %w", parts[1], err)
-	}
-	return parts, size, unsigned, nil
-}
-
-// extraAttr parses the EXTRA column from the INFORMATION_SCHEMA.COLUMNS table
-// and appends its parsed representation to the column.
-func extraAttr(c *schema.Column, extra string) error {
-	switch extra := strings.ToLower(extra); extra {
-	case "", "null": // ignore.
-	case "default_generated":
-		// The column has an expression default value
-		// and it's handled in Driver.addColumn.
-	case "auto_increment":
-		c.Attrs = append(c.Attrs, &AutoIncrement{A: extra})
-	case "default_generated on update current_timestamp", "on update current_timestamp",
-		"on update current_timestamp()" /* MariaDB format. */ :
-		c.Attrs = append(c.Attrs, &OnUpdate{A: extra})
-	default:
-		return fmt.Errorf("unknown attribute %q", extra)
-	}
-	return nil
-}
-
-const (
-	// Query to list system variables.
-	variablesQuery = "SELECT @@version, @@collation_server, @@character_set_server"
-
-	// Query to list database schemas.
-	schemasQuery = "SELECT `SCHEMA_NAME`, `DEFAULT_CHARACTER_SET_NAME`, `DEFAULT_COLLATION_NAME` from `INFORMATION_SCHEMA`.`SCHEMATA`"
-
-	// Query to list schema tables.
-	tablesQuery = "SELECT `TABLE_NAME` FROM `INFORMATION_SCHEMA`.`TABLES` WHERE `TABLE_TYPE` = 'BASE TABLE' AND `TABLE_SCHEMA` = ? ORDER BY `TABLE_NAME`"
-
-	// Query to list table columns.
-	columnsQuery = "SELECT `COLUMN_NAME`, `COLUMN_TYPE`, `COLUMN_COMMENT`, `IS_NULLABLE`, `COLUMN_KEY`, `COLUMN_DEFAULT`, `EXTRA`, `CHARACTER_SET_NAME`, `COLLATION_NAME` FROM `INFORMATION_SCHEMA`.`COLUMNS` WHERE `TABLE_SCHEMA` = ? AND `TABLE_NAME` = ? ORDER BY `ORDINAL_POSITION`"
-
-	// Query to list table indexes.
-	indexesQuery     = "SELECT `INDEX_NAME`, `COLUMN_NAME`, `NON_UNIQUE`, `SEQ_IN_INDEX`, `INDEX_TYPE`, `COLLATION`, `INDEX_COMMENT`, `SUB_PART`, NULL AS `EXPRESSION` FROM `INFORMATION_SCHEMA`.`STATISTICS` WHERE `TABLE_SCHEMA` = ? AND `TABLE_NAME` = ? ORDER BY `index_name`, `seq_in_index`"
-	indexesExprQuery = "SELECT `INDEX_NAME`, `COLUMN_NAME`, `NON_UNIQUE`, `SEQ_IN_INDEX`, `INDEX_TYPE`, `COLLATION`, `INDEX_COMMENT`, `SUB_PART`, `EXPRESSION` FROM `INFORMATION_SCHEMA`.`STATISTICS` WHERE `TABLE_SCHEMA` = ? AND `TABLE_NAME` = ? ORDER BY `index_name`, `seq_in_index`"
-
-	// Query to list table information.
-	tableQuery = `
-SELECT
-	t1.TABLE_SCHEMA,
-	t2.CHARACTER_SET_NAME,
-	t1.TABLE_COLLATION,
-	t1.AUTO_INCREMENT,
-	t1.TABLE_COMMENT
-FROM
-	INFORMATION_SCHEMA.TABLES AS t1
-	JOIN INFORMATION_SCHEMA.COLLATIONS AS t2
-	ON t1.TABLE_COLLATION = t2.COLLATION_NAME
-WHERE
-	TABLE_NAME = ?
-	AND TABLE_SCHEMA = (SELECT DATABASE())
-`
-	tableSchemaQuery = `
-SELECT
-	t1.TABLE_SCHEMA,
-	t2.CHARACTER_SET_NAME,
-	t1.TABLE_COLLATION,
-	t1.AUTO_INCREMENT,
-	t1.TABLE_COMMENT
-FROM
-	INFORMATION_SCHEMA.TABLES AS t1
-	JOIN INFORMATION_SCHEMA.COLLATIONS AS t2
-	ON t1.TABLE_COLLATION = t2.COLLATION_NAME
-WHERE
-	TABLE_NAME = ?
-	AND TABLE_SCHEMA = ?
-`
-
-	// Query to list table check constraints.
-	myChecksQuery  = `SELECT t1.CONSTRAINT_NAME, t2.CHECK_CLAUSE, t1.ENFORCED` + checksQuery
-	marChecksQuery = `SELECT t1.CONSTRAINT_NAME, t2.CHECK_CLAUSE, "YES" AS ENFORCED` + checksQuery
-	checksQuery    = `
-FROM
-	INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t1
-	JOIN INFORMATION_SCHEMA.CHECK_CONSTRAINTS AS t2
-	ON t1.CONSTRAINT_NAME = t2.CONSTRAINT_NAME
-WHERE
-	t1.CONSTRAINT_TYPE = 'CHECK'
-	AND t1.TABLE_SCHEMA = ?
-	AND t1.TABLE_NAME = ?
-ORDER BY
-	t1.CONSTRAINT_NAME
-`
-
-	// Query to list table foreign keys.
-	fksQuery = `
-SELECT
-	t1.CONSTRAINT_NAME,
-	t1.TABLE_NAME,
-	t1.COLUMN_NAME,
-	t1.TABLE_SCHEMA,
-	t1.REFERENCED_TABLE_NAME,
-	t1.REFERENCED_COLUMN_NAME,
-	t1.REFERENCED_TABLE_SCHEMA,
-	t3.UPDATE_RULE,
-	t3.DELETE_RULE
-FROM
-	INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS t1
-	JOIN INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t2
-	JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS t3
-	ON t1.CONSTRAINT_NAME = t2.CONSTRAINT_NAME
-	AND t1.CONSTRAINT_NAME = t3.CONSTRAINT_NAME
-	AND t1.TABLE_SCHEMA = t2.TABLE_SCHEMA
-	AND t1.TABLE_SCHEMA = t3.CONSTRAINT_SCHEMA
-WHERE
-	t2.CONSTRAINT_TYPE = 'FOREIGN KEY'
-	AND t1.TABLE_SCHEMA = ?
-	AND t1.TABLE_NAME = ?
-ORDER BY
-	t1.CONSTRAINT_NAME,
-	t1.ORDINAL_POSITION`
-)
-
-var _ schema.TableInspector = (*Driver)(nil)
-
-type (
-	// AutoIncrement attribute for columns with "AUTO_INCREMENT" as a default.
-	AutoIncrement struct {
-		schema.Attr
-		A string
-		V int64
-	}
-
-	// OnUpdate attribute for columns with "ON UPDATE CURRENT_TIMESTAMP" as a default.
-	OnUpdate struct {
-		schema.Attr
-		A string
-	}
-
-	// SubPart attribute defines an option index prefix length for columns.
-	SubPart struct {
-		schema.Attr
-		Len int
-	}
-
-	// Check attributes defines a CHECK constraint.
-	Check struct {
-		schema.Attr
-		Name     string
-		Clause   string
-		Enforced bool
-	}
-
-	// The DisplayWidth represents a display width of an integer type.
-	DisplayWidth struct {
-		schema.Attr
-		N int
-	}
-
-	// The ZeroFill represents the ZEROFILL attribute which is
-	// deprecated for MySQL version >= 8.0.17.
-	ZeroFill struct {
-		schema.Attr
-		A string
-	}
-
-	// IndexType represents an index type.
-	IndexType struct {
-		schema.Attr
-		T string // BTREE, FULLTEXT, HASH, RTREE
-	}
-
-	// BitType represents a bit type.
-	BitType struct {
-		schema.Type
-		T string
-	}
-
-	// SetType represents a set type.
-	SetType struct {
-		schema.Type
-		Values []string
-	}
-)
-
 // MySQL standard unescape field function from its codebase:
 // https://github.com/mysql/mysql-server/blob/8.0/sql/dd/impl/utils.cc
 func unescape(s string) string {
diff --git a/sql/mysql/inspect.go b/sql/mysql/inspect.go
new file mode 100644
index 00000000..8f6f4eec
--- /dev/null
+++ b/sql/mysql/inspect.go
@@ -0,0 +1,737 @@
+// 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"
+	"database/sql"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"ariga.io/atlas/sql/internal/sqlx"
+	"ariga.io/atlas/sql/schema"
+)
+
+// A diff provides a MySQL implementation for schema.Inspector.
+type inspect struct{ conn }
+
+var _ schema.Inspector = (*inspect)(nil)
+
+// InspectRealm returns schema descriptions of all resources in the given realm.
+func (i *inspect) InspectRealm(ctx context.Context, opts *schema.InspectRealmOption) (*schema.Realm, error) {
+	schemas, err := i.schemas(ctx, opts)
+	if err != nil {
+		return nil, err
+	}
+	realm := &schema.Realm{Schemas: schemas, Attrs: []schema.Attr{&schema.Charset{V: i.charset}, &schema.Collation{V: i.collate}}}
+	for _, s := range schemas {
+		names, err := i.tableNames(ctx, s.Name, nil)
+		if err != nil {
+			return nil, err
+		}
+		for _, name := range names {
+			t, err := i.inspectTable(ctx, name, &schema.InspectTableOptions{Schema: s.Name}, s)
+			if err != nil {
+				return nil, err
+			}
+			s.Tables = append(s.Tables, t)
+		}
+		s.Realm = realm
+	}
+	sqlx.LinkSchemaTables(schemas)
+	return realm, nil
+}
+
+// InspectSchema returns schema descriptions of all tables in the given schema.
+func (i *inspect) InspectSchema(ctx context.Context, name string, opts *schema.InspectOptions) (*schema.Schema, error) {
+	schemas, err := i.schemas(ctx, &schema.InspectRealmOption{
+		Schemas: []string{name},
+	})
+	if err != nil {
+		return nil, err
+	}
+	if len(schemas) == 0 {
+		return nil, &schema.NotExistError{
+			Err: fmt.Errorf("mysql: schema %q was not found", name),
+		}
+	}
+	names, err := i.tableNames(ctx, name, opts)
+	if err != nil {
+		return nil, err
+	}
+	s := schemas[0]
+	for _, name := range names {
+		t, err := i.inspectTable(ctx, name, &schema.InspectTableOptions{Schema: s.Name}, s)
+		if err != nil {
+			return nil, err
+		}
+		s.Tables = append(s.Tables, t)
+	}
+	sqlx.LinkSchemaTables(schemas)
+	s.Realm = &schema.Realm{Schemas: schemas, Attrs: []schema.Attr{&schema.Charset{V: i.charset}, &schema.Collation{V: i.collate}}}
+	return s, nil
+}
+
+// InspectTable returns the schema description of the given table.
+func (i *inspect) InspectTable(ctx context.Context, name string, opts *schema.InspectTableOptions) (*schema.Table, error) {
+	return i.inspectTable(ctx, name, opts, nil)
+}
+
+func (i *inspect) inspectTable(ctx context.Context, name string, opts *schema.InspectTableOptions, top *schema.Schema) (*schema.Table, error) {
+	t, err := i.table(ctx, name, opts)
+	if err != nil {
+		return nil, err
+	}
+	if top != nil {
+		// Link the table to its top element if provided.
+		t.Schema = top
+	}
+	if err := i.columns(ctx, t); err != nil {
+		return nil, err
+	}
+	if err := i.indexes(ctx, t); err != nil {
+		return nil, err
+	}
+	if err := i.fks(ctx, t); err != nil {
+		return nil, err
+	}
+	if err := i.checks(ctx, t); err != nil {
+		return nil, err
+	}
+	return t, nil
+}
+
+// schemas returns the list of the schemas in the database.
+func (i *inspect) schemas(ctx context.Context, opts *schema.InspectRealmOption) ([]*schema.Schema, error) {
+	var (
+		args  []interface{}
+		query = schemasQuery
+	)
+	if opts != nil && len(opts.Schemas) > 0 {
+		query += " WHERE `SCHEMA_NAME` IN (" + strings.Repeat("?, ", len(opts.Schemas)-1) + "?)"
+		for _, s := range opts.Schemas {
+			args = append(args, s)
+		}
+	} else {
+		query += " WHERE `SCHEMA_NAME` NOT IN ('mysql', 'information_schema', 'performance_schema', 'sys')"
+	}
+	rows, err := i.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, fmt.Errorf("mysql: querying schemas: %w", err)
+	}
+	defer rows.Close()
+	var schemas []*schema.Schema
+	for rows.Next() {
+		var name, charset, collation string
+		if err := rows.Scan(&name, &charset, &collation); err != nil {
+			return nil, err
+		}
+		schemas = append(schemas, &schema.Schema{
+			Name: name,
+			Attrs: []schema.Attr{
+				&schema.Charset{
+					V: charset,
+				},
+				&schema.Collation{
+					V: collation,
+				},
+			},
+		})
+	}
+	return schemas, nil
+}
+
+// table returns the table from the database, or a NotExistError if the table was not found.
+func (i *inspect) table(ctx context.Context, name string, opts *schema.InspectTableOptions) (*schema.Table, error) {
+	var (
+		args  = []interface{}{name}
+		query = tableQuery
+	)
+	if opts != nil && opts.Schema != "" {
+		query = tableSchemaQuery
+		args = append(args, opts.Schema)
+	}
+	row := i.QueryRowContext(ctx, query, args...)
+	var (
+		autoinc                              sql.NullInt64
+		tSchema, charset, collation, comment sql.NullString
+	)
+	if err := row.Scan(&tSchema, &charset, &collation, &autoinc, &comment); err != nil {
+		if err == sql.ErrNoRows {
+			return nil, &schema.NotExistError{
+				Err: fmt.Errorf("mysql: table %q was not found", name),
+			}
+		}
+		return nil, err
+	}
+	t := &schema.Table{Name: name, Schema: &schema.Schema{Name: tSchema.String}}
+	if sqlx.ValidString(charset) {
+		t.Attrs = append(t.Attrs, &schema.Charset{
+			V: charset.String,
+		})
+	}
+	if sqlx.ValidString(collation) {
+		t.Attrs = append(t.Attrs, &schema.Collation{
+			V: collation.String,
+		})
+	}
+	if sqlx.ValidString(comment) {
+		t.Attrs = append(t.Attrs, &schema.Comment{
+			Text: comment.String,
+		})
+	}
+	if autoinc.Valid {
+		t.Attrs = append(t.Attrs, &AutoIncrement{
+			V: autoinc.Int64,
+		})
+	}
+	return t, nil
+}
+
+// columns queries and appends the columns of the given table.
+func (i *inspect) columns(ctx context.Context, t *schema.Table) error {
+	rows, err := i.QueryContext(ctx, columnsQuery, t.Schema.Name, t.Name)
+	if err != nil {
+		return fmt.Errorf("mysql: querying %q columns: %w", t.Name, err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		if err := i.addColumn(t, rows); err != nil {
+			return fmt.Errorf("mysql: %w", err)
+		}
+	}
+	return rows.Err()
+}
+
+// addColumn scans the current row and adds a new column from it to the table.
+func (i *inspect) addColumn(t *schema.Table, rows *sql.Rows) error {
+	var name, typ, comment, nullable, key, defaults, extra, charset, collation sql.NullString
+	if err := rows.Scan(&name, &typ, &comment, &nullable, &key, &defaults, &extra, &charset, &collation); err != nil {
+		return err
+	}
+	c := &schema.Column{
+		Name: name.String,
+		Type: &schema.ColumnType{
+			Raw:  typ.String,
+			Null: nullable.String == "YES",
+		},
+	}
+	ct, err := parseRawType(c.Type.Raw)
+	if err != nil {
+		return err
+	}
+	c.Type.Type = ct
+	if err := extraAttr(c, extra.String); err != nil {
+		return err
+	}
+	if sqlx.ValidString(defaults) {
+		x := defaults.String
+		// From MariaDB 10.2.7, literals are quoted to distinguish them from expressions.
+		if i.mariadb() && strings.HasPrefix(x, "'") && strings.HasSuffix(x, "'") {
+			x = x[1 : len(x)-1]
+		}
+		c.Default = &schema.RawExpr{X: x}
+	}
+	if sqlx.ValidString(comment) {
+		c.Attrs = append(c.Attrs, &schema.Comment{
+			Text: comment.String,
+		})
+	}
+	if sqlx.ValidString(charset) {
+		c.Attrs = append(c.Attrs, &schema.Charset{
+			V: charset.String,
+		})
+	}
+	if sqlx.ValidString(collation) {
+		c.Attrs = append(c.Attrs, &schema.Collation{
+			V: collation.String,
+		})
+	}
+	t.Columns = append(t.Columns, c)
+	if key.String == "PRI" {
+		if t.PrimaryKey == nil {
+			t.PrimaryKey = &schema.Index{Table: t, Name: key.String}
+		}
+		t.PrimaryKey.Parts = append(t.PrimaryKey.Parts, &schema.IndexPart{
+			C:     c,
+			SeqNo: len(t.PrimaryKey.Parts),
+		})
+	}
+	return nil
+}
+
+func parseRawType(raw string) (schema.Type, error) {
+	parts, size, unsigned, err := parseColumn(raw)
+	if err != nil {
+		return nil, err
+	}
+	switch t := parts[0]; t {
+	case tBit:
+		return &BitType{
+			T: t,
+		}, nil
+	case tTinyInt, tSmallInt, tMediumInt, tInt, tBigInt:
+		if size == 1 {
+			return &schema.BoolType{
+				T: t,
+			}, nil
+		}
+		// For integer types, the size represents the display width and does not
+		// constrain the range of values that can be stored in the column.
+		// The storage byte-size is inferred from the type name (i.e TINYINT takes
+		// a single byte).
+		ft := &schema.IntegerType{
+			T:        t,
+			Unsigned: unsigned,
+		}
+		if attr := parts[len(parts)-1]; attr == "zerofill" && size != 0 {
+			ft.Attrs = []schema.Attr{
+				&DisplayWidth{
+					N: int(size),
+				},
+				&ZeroFill{
+					A: attr,
+				},
+			}
+		}
+		return ft, nil
+	case tNumeric, tDecimal:
+		dt := &schema.DecimalType{
+			T: t,
+		}
+		if len(parts) > 1 {
+			p, err := strconv.ParseInt(parts[1], 10, 64)
+			if err != nil {
+				return nil, fmt.Errorf("parse precision %q", parts[1])
+			}
+			dt.Precision = int(p)
+		}
+		if len(parts) > 2 {
+			s, err := strconv.ParseInt(parts[2], 10, 64)
+			if err != nil {
+				return nil, fmt.Errorf("parse scale %q", parts[1])
+			}
+			dt.Scale = int(s)
+		}
+		return dt, nil
+	case tFloat, tDouble, tReal:
+		ft := &schema.FloatType{
+			T: t,
+		}
+		if len(parts) > 1 {
+			p, err := strconv.ParseInt(parts[1], 10, 64)
+			if err != nil {
+				return nil, fmt.Errorf("parse precision %q", parts[1])
+			}
+			ft.Precision = int(p)
+		}
+		return ft, nil
+	case tBinary, tVarBinary:
+		return &schema.BinaryType{
+			T:    t,
+			Size: int(size),
+		}, nil
+	case tTinyBlob, tMediumBlob, tBlob, tLongBlob:
+		return &schema.BinaryType{
+			T: t,
+		}, nil
+	case tChar, tVarchar:
+		return &schema.StringType{
+			T:    t,
+			Size: int(size),
+		}, nil
+	case tTinyText, tMediumText, tText, tLongText:
+		return &schema.StringType{
+			T: t,
+		}, nil
+	case tEnum, tSet:
+		// Parse the enum values according to the MySQL format.
+		// github.com/mysql/mysql-server/blob/8.0/sql/field.cc#Field_enum::sql_type
+		rv := strings.TrimSuffix(strings.TrimPrefix(raw, "enum("), ")")
+		if rv == "" {
+			return nil, fmt.Errorf("mysql: unexpected enum type: %q", raw)
+		}
+		values := strings.Split(rv, "','")
+		for i := range values {
+			values[i] = strings.Trim(values[i], "'")
+		}
+		if t == tEnum {
+			return &schema.EnumType{
+				Values: values,
+			}, nil
+		}
+		return &SetType{
+			Values: values,
+		}, nil
+	case tDate, tDateTime, tTime, tTimestamp, tYear:
+		return &schema.TimeType{
+			T: t,
+		}, nil
+	case tJSON:
+		return &schema.JSONType{
+			T: t,
+		}, nil
+	case tPoint, tMultiPoint, tLineString, tMultiLineString, tPolygon, tMultiPolygon, tGeometry, tGeoCollection, tGeometryCollection:
+		return &schema.SpatialType{
+			T: t,
+		}, nil
+	default:
+		return &schema.UnsupportedType{
+			T: t,
+		}, nil
+	}
+}
+
+// indexes queries and appends the indexes of the given table.
+func (i *inspect) indexes(ctx context.Context, t *schema.Table) error {
+	query := indexesQuery
+	if i.supportsIndexExpr() {
+		query = indexesExprQuery
+	}
+	rows, err := i.QueryContext(ctx, query, t.Schema.Name, t.Name)
+	if err != nil {
+		return fmt.Errorf("mysql: querying %q indexes: %w", t.Name, err)
+	}
+	defer rows.Close()
+	if err := i.addIndexes(t, rows); err != nil {
+		return err
+	}
+	return rows.Err()
+}
+
+// addIndexes scans the rows and adds the indexes to the table.
+func (i *inspect) addIndexes(t *schema.Table, rows *sql.Rows) error {
+	names := make(map[string]*schema.Index)
+	for rows.Next() {
+		var (
+			nonuniq                                 bool
+			seqno                                   int
+			name, indexType                         string
+			column, subPart, expr, comment, collate sql.NullString
+		)
+		if err := rows.Scan(&name, &column, &nonuniq, &seqno, &indexType, &collate, &comment, &subPart, &expr); err != nil {
+			return fmt.Errorf("mysql: scanning index: %w", err)
+		}
+		// Ignore primary keys.
+		if name == "PRIMARY" {
+			continue
+		}
+		idx, ok := names[name]
+		if !ok {
+			idx = &schema.Index{
+				Name:   name,
+				Unique: !nonuniq,
+				Table:  t,
+				Attrs: []schema.Attr{
+					&IndexType{T: indexType},
+				},
+			}
+			if sqlx.ValidString(comment) {
+				idx.Attrs = append(t.Attrs, &schema.Comment{
+					Text: comment.String,
+				})
+			}
+			names[name] = idx
+			t.Indexes = append(t.Indexes, idx)
+		}
+		// Rows are ordered by SEQ_IN_INDEX that specifies the
+		// position of the column in the index definition.
+		part := &schema.IndexPart{
+			SeqNo: seqno,
+			Attrs: []schema.Attr{&schema.Collation{V: collate.String}},
+		}
+		switch {
+		case sqlx.ValidString(expr):
+			part.X = &schema.RawExpr{
+				X: expr.String,
+			}
+		case sqlx.ValidString(column):
+			part.C, ok = t.Column(column.String)
+			if !ok {
+				return fmt.Errorf("mysql: column %q was not found for index %q", column.String, idx.Name)
+			}
+			if sqlx.ValidString(subPart) {
+				n, err := strconv.Atoi(subPart.String)
+				if err != nil {
+					return fmt.Errorf("mysql: parse index prefix size %q: %w", subPart.String, err)
+				}
+				part.Attrs = append(part.Attrs, &SubPart{
+					Len: n,
+				})
+			}
+			part.C.Indexes = append(part.C.Indexes, idx)
+		default:
+			return fmt.Errorf("mysql: invalid part for index %q", idx.Name)
+		}
+		idx.Parts = append(idx.Parts, part)
+	}
+	return nil
+}
+
+// fks queries and appends the foreign keys of the given table.
+func (i *inspect) fks(ctx context.Context, t *schema.Table) error {
+	rows, err := i.QueryContext(ctx, fksQuery, t.Schema.Name, t.Name)
+	if err != nil {
+		return fmt.Errorf("mysql: querying %q foreign keys: %w", t.Name, err)
+	}
+	defer rows.Close()
+	if err := sqlx.ScanFKs(t, rows); err != nil {
+		return fmt.Errorf("mysql: %w", err)
+	}
+	return rows.Err()
+}
+
+// checks queries and appends the check constraints of the given table.
+func (i *inspect) checks(ctx context.Context, t *schema.Table) error {
+	query, ok := i.supportsCheck()
+	if !ok {
+		return nil
+	}
+	rows, err := i.QueryContext(ctx, query, t.Schema.Name, t.Name)
+	if err != nil {
+		return fmt.Errorf("mysql: querying %q check constraints: %w", t.Name, err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var name, clause, enforced sql.NullString
+		if err := rows.Scan(&name, &clause, &enforced); err != nil {
+			return fmt.Errorf("mysql: %w", err)
+		}
+		check := &Check{
+			Name:     name.String,
+			Clause:   unescape(clause.String),
+			Enforced: clause.String != "NO",
+		}
+		t.Attrs = append(t.Attrs, check)
+		// In MariaDB, JSON is an alias to LONGTEXT, and the JSON_VALID
+		// CHECK constraint is automatically created for the column for
+		// versions >= 10.4.3. However, we expect tools like Atlas and
+		// Ent to manually add this CHECK for older versions of MariaDB.
+		if i.mariadb() {
+			c, ok := t.Column(check.Name)
+			if ok && c.Type.Raw == tLongText && check.Clause == fmt.Sprintf("json_valid(`%s`)", c.Name) {
+				c.Type.Raw = tJSON
+				c.Type.Type = &schema.JSONType{T: tJSON}
+			}
+		}
+
+	}
+	return rows.Err()
+}
+
+// tableNames returns a list of all tables exist in the schema.
+func (i *inspect) tableNames(ctx context.Context, schema string, opts *schema.InspectOptions) ([]string, error) {
+	query, args := tablesQuery, []interface{}{schema}
+	if opts != nil && len(opts.Tables) > 0 {
+		query += " AND `TABLE_NAME` IN (" + strings.Repeat("?, ", len(opts.Tables)-1) + "?)"
+		for _, s := range opts.Tables {
+			args = append(args, s)
+		}
+	}
+	rows, err := i.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, fmt.Errorf("mysql: querying schema tables: %w", err)
+	}
+	names, err := sqlx.ScanStrings(rows)
+	if err != nil {
+		return nil, fmt.Errorf("mysql: scanning table names: %w", err)
+	}
+	return names, nil
+}
+
+// parseColumn returns column parts, size and signed-info from a MySQL type.
+func parseColumn(typ string) (parts []string, size int64, unsigned bool, err error) {
+	switch parts = strings.FieldsFunc(typ, func(r rune) bool {
+		return r == '(' || r == ')' || r == ' ' || r == ','
+	}); parts[0] {
+	case tTinyInt, tSmallInt, tMediumInt, tInt, tBigInt:
+		if attr := parts[len(parts)-1]; attr == "unsigned" || attr == "zerofill" {
+			unsigned = true
+		}
+		if len(parts) > 2 || len(parts) == 2 && !unsigned {
+			size, err = strconv.ParseInt(parts[1], 10, 64)
+		}
+	case tBinary, tVarBinary, tChar, tVarchar:
+		size, err = strconv.ParseInt(parts[1], 10, 64)
+	}
+	if err != nil {
+		return nil, 0, false, fmt.Errorf("parse %q to int: %w", parts[1], err)
+	}
+	return parts, size, unsigned, nil
+}
+
+// extraAttr parses the EXTRA column from the INFORMATION_SCHEMA.COLUMNS table
+// and appends its parsed representation to the column.
+func extraAttr(c *schema.Column, extra string) error {
+	switch extra := strings.ToLower(extra); extra {
+	case "", "null": // ignore.
+	case "default_generated":
+		// The column has an expression default value
+		// and it's handled in Driver.addColumn.
+	case "auto_increment":
+		c.Attrs = append(c.Attrs, &AutoIncrement{A: extra})
+	case "default_generated on update current_timestamp", "on update current_timestamp",
+		"on update current_timestamp()" /* MariaDB format. */ :
+		c.Attrs = append(c.Attrs, &OnUpdate{A: extra})
+	default:
+		return fmt.Errorf("unknown attribute %q", extra)
+	}
+	return nil
+}
+
+const (
+	// Query to list system variables.
+	variablesQuery = "SELECT @@version, @@collation_server, @@character_set_server"
+
+	// Query to list database schemas.
+	schemasQuery = "SELECT `SCHEMA_NAME`, `DEFAULT_CHARACTER_SET_NAME`, `DEFAULT_COLLATION_NAME` from `INFORMATION_SCHEMA`.`SCHEMATA`"
+
+	// Query to list schema tables.
+	tablesQuery = "SELECT `TABLE_NAME` FROM `INFORMATION_SCHEMA`.`TABLES` WHERE `TABLE_TYPE` = 'BASE TABLE' AND `TABLE_SCHEMA` = ? ORDER BY `TABLE_NAME`"
+
+	// Query to list table columns.
+	columnsQuery = "SELECT `COLUMN_NAME`, `COLUMN_TYPE`, `COLUMN_COMMENT`, `IS_NULLABLE`, `COLUMN_KEY`, `COLUMN_DEFAULT`, `EXTRA`, `CHARACTER_SET_NAME`, `COLLATION_NAME` FROM `INFORMATION_SCHEMA`.`COLUMNS` WHERE `TABLE_SCHEMA` = ? AND `TABLE_NAME` = ? ORDER BY `ORDINAL_POSITION`"
+
+	// Query to list table indexes.
+	indexesQuery     = "SELECT `INDEX_NAME`, `COLUMN_NAME`, `NON_UNIQUE`, `SEQ_IN_INDEX`, `INDEX_TYPE`, `COLLATION`, `INDEX_COMMENT`, `SUB_PART`, NULL AS `EXPRESSION` FROM `INFORMATION_SCHEMA`.`STATISTICS` WHERE `TABLE_SCHEMA` = ? AND `TABLE_NAME` = ? ORDER BY `index_name`, `seq_in_index`"
+	indexesExprQuery = "SELECT `INDEX_NAME`, `COLUMN_NAME`, `NON_UNIQUE`, `SEQ_IN_INDEX`, `INDEX_TYPE`, `COLLATION`, `INDEX_COMMENT`, `SUB_PART`, `EXPRESSION` FROM `INFORMATION_SCHEMA`.`STATISTICS` WHERE `TABLE_SCHEMA` = ? AND `TABLE_NAME` = ? ORDER BY `index_name`, `seq_in_index`"
+
+	// Query to list table information.
+	tableQuery = `
+SELECT
+	t1.TABLE_SCHEMA,
+	t2.CHARACTER_SET_NAME,
+	t1.TABLE_COLLATION,
+	t1.AUTO_INCREMENT,
+	t1.TABLE_COMMENT
+FROM
+	INFORMATION_SCHEMA.TABLES AS t1
+	JOIN INFORMATION_SCHEMA.COLLATIONS AS t2
+	ON t1.TABLE_COLLATION = t2.COLLATION_NAME
+WHERE
+	TABLE_NAME = ?
+	AND TABLE_SCHEMA = (SELECT DATABASE())
+`
+	tableSchemaQuery = `
+SELECT
+	t1.TABLE_SCHEMA,
+	t2.CHARACTER_SET_NAME,
+	t1.TABLE_COLLATION,
+	t1.AUTO_INCREMENT,
+	t1.TABLE_COMMENT
+FROM
+	INFORMATION_SCHEMA.TABLES AS t1
+	JOIN INFORMATION_SCHEMA.COLLATIONS AS t2
+	ON t1.TABLE_COLLATION = t2.COLLATION_NAME
+WHERE
+	TABLE_NAME = ?
+	AND TABLE_SCHEMA = ?
+`
+
+	// Query to list table check constraints.
+	myChecksQuery  = `SELECT t1.CONSTRAINT_NAME, t2.CHECK_CLAUSE, t1.ENFORCED` + checksQuery
+	marChecksQuery = `SELECT t1.CONSTRAINT_NAME, t2.CHECK_CLAUSE, "YES" AS ENFORCED` + checksQuery
+	checksQuery    = `
+FROM
+	INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t1
+	JOIN INFORMATION_SCHEMA.CHECK_CONSTRAINTS AS t2
+	ON t1.CONSTRAINT_NAME = t2.CONSTRAINT_NAME
+WHERE
+	t1.CONSTRAINT_TYPE = 'CHECK'
+	AND t1.TABLE_SCHEMA = ?
+	AND t1.TABLE_NAME = ?
+ORDER BY
+	t1.CONSTRAINT_NAME
+`
+
+	// Query to list table foreign keys.
+	fksQuery = `
+SELECT
+	t1.CONSTRAINT_NAME,
+	t1.TABLE_NAME,
+	t1.COLUMN_NAME,
+	t1.TABLE_SCHEMA,
+	t1.REFERENCED_TABLE_NAME,
+	t1.REFERENCED_COLUMN_NAME,
+	t1.REFERENCED_TABLE_SCHEMA,
+	t3.UPDATE_RULE,
+	t3.DELETE_RULE
+FROM
+	INFORMATION_SCHEMA.KEY_COLUMN_USAGE AS t1
+	JOIN INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS t2
+	JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS AS t3
+	ON t1.CONSTRAINT_NAME = t2.CONSTRAINT_NAME
+	AND t1.CONSTRAINT_NAME = t3.CONSTRAINT_NAME
+	AND t1.TABLE_SCHEMA = t2.TABLE_SCHEMA
+	AND t1.TABLE_SCHEMA = t3.CONSTRAINT_SCHEMA
+WHERE
+	t2.CONSTRAINT_TYPE = 'FOREIGN KEY'
+	AND t1.TABLE_SCHEMA = ?
+	AND t1.TABLE_NAME = ?
+ORDER BY
+	t1.CONSTRAINT_NAME,
+	t1.ORDINAL_POSITION`
+)
+
+type (
+	// AutoIncrement attribute for columns with "AUTO_INCREMENT" as a default.
+	AutoIncrement struct {
+		schema.Attr
+		A string
+		V int64
+	}
+
+	// OnUpdate attribute for columns with "ON UPDATE CURRENT_TIMESTAMP" as a default.
+	OnUpdate struct {
+		schema.Attr
+		A string
+	}
+
+	// SubPart attribute defines an option index prefix length for columns.
+	SubPart struct {
+		schema.Attr
+		Len int
+	}
+
+	// Check attributes defines a CHECK constraint.
+	Check struct {
+		schema.Attr
+		Name     string
+		Clause   string
+		Enforced bool
+	}
+
+	// The DisplayWidth represents a display width of an integer type.
+	DisplayWidth struct {
+		schema.Attr
+		N int
+	}
+
+	// The ZeroFill represents the ZEROFILL attribute which is
+	// deprecated for MySQL version >= 8.0.17.
+	ZeroFill struct {
+		schema.Attr
+		A string
+	}
+
+	// IndexType represents an index type.
+	IndexType struct {
+		schema.Attr
+		T string // BTREE, FULLTEXT, HASH, RTREE
+	}
+
+	// BitType represents a bit type.
+	BitType struct {
+		schema.Type
+		T string
+	}
+
+	// SetType represents a set type.
+	SetType struct {
+		schema.Type
+		Values []string
+	}
+)
diff --git a/sql/mysql/driver_test.go b/sql/mysql/inspect_test.go
similarity index 100%
rename from sql/mysql/driver_test.go
rename to sql/mysql/inspect_test.go
diff --git a/sql/mysql/migrate.go b/sql/mysql/migrate.go
index 91c7c8ed..410164d3 100644
--- a/sql/mysql/migrate.go
+++ b/sql/mysql/migrate.go
@@ -15,12 +15,7 @@ import (
 )
 
 // A migrate provides migration capabilities for schema elements.
-type migrate struct{ *Driver }
-
-// Migrate returns a MySQL schema executor.
-func (d *Driver) Migrate() schema.Execer {
-	return &migrate{Driver: d}
-}
+type migrate 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.
diff --git a/sql/postgres/diff.go b/sql/postgres/diff.go
index 03d77d0d..b558f76c 100644
--- a/sql/postgres/diff.go
+++ b/sql/postgres/diff.go
@@ -15,14 +15,7 @@ import (
 )
 
 // A diff provides a PostgreSQL implementation for sqlx.DiffDriver.
-type diff struct{ *Driver }
-
-// Diff returns a PostgreSQL schema differ.
-func (d *Driver) Diff() schema.Differ {
-	return &sqlx.Diff{
-		DiffDriver: &diff{Driver: d},
-	}
-}
+type diff struct{ conn }
 
 // SchemaAttrDiff returns a changeset for migrating schema attributes from one state to the other.
 func (d *diff) SchemaAttrDiff(_, _ *schema.Schema) []schema.Change {
@@ -265,7 +258,7 @@ func (d *diff) valuesEqual(x, y string) (bool, error) {
 	var b bool
 	// The DEFAULT expressions are safe to be inlined in the SELECT
 	// statement same as we inline them in the CREATE TABLE statement.
-	if err := d.Driver.QueryRow(fmt.Sprintf("SELECT %s = %s", x, y)).Scan(&b); err != nil {
+	if err := d.QueryRow(fmt.Sprintf("SELECT %s = %s", x, y)).Scan(&b); err != nil {
 		return false, err
 	}
 	return b, nil
@@ -277,7 +270,7 @@ func (d *diff) typesEqual(x, y string) (bool, error) {
 	var b bool
 	// The datatype are safe to be inlined in the SELECT statement
 	// same as we inline them in the CREATE TABLE statement.
-	if err := d.Driver.QueryRow(fmt.Sprintf("SELECT '%s'::regtype = '%s'::regtype", x, y)).Scan(&b); err != nil {
+	if err := d.QueryRow(fmt.Sprintf("SELECT '%s'::regtype = '%s'::regtype", x, y)).Scan(&b); err != nil {
 		return false, err
 	}
 	return b, nil
diff --git a/sql/postgres/driver.go b/sql/postgres/driver.go
index 53f5fecc..0b03e521 100644
--- a/sql/postgres/driver.go
+++ b/sql/postgres/driver.go
@@ -5,30 +5,38 @@
 package postgres
 
 import (
-	"context"
-	"database/sql"
 	"fmt"
-	"strings"
 
 	"ariga.io/atlas/sql/internal/sqlx"
+
 	"ariga.io/atlas/sql/schema"
 
 	"golang.org/x/mod/semver"
 )
 
-// Driver represents a PostgreSQL driver for introspecting database schemas
-// and apply migrations changes on them.
-type Driver struct {
-	schema.ExecQuerier
-	// System variables that are set on `Open`.
-	collate string
-	ctype   string
-	version string
-}
+type (
+	// Driver represents a PostgreSQL driver for introspecting database schemas,
+	// generating diff between schema elements and apply migrations changes.
+	Driver struct {
+		conn
+		schema.Differ
+		schema.Execer
+		schema.Inspector
+	}
+
+	// database connection and its information.
+	conn struct {
+		schema.ExecQuerier
+		// System variables that are set on `Open`.
+		collate string
+		ctype   string
+		version string
+	}
+)
 
 // Open opens a new PostgreSQL driver.
 func Open(db schema.ExecQuerier) (*Driver, error) {
-	drv := &Driver{ExecQuerier: db}
+	c := conn{ExecQuerier: db}
 	rows, err := db.Query(paramsQuery)
 	if err != nil {
 		return nil, fmt.Errorf("postgres: scanning system variables: %w", err)
@@ -45,780 +53,20 @@ func Open(db schema.ExecQuerier) (*Driver, error) {
 	if len(params) != 3 {
 		return nil, fmt.Errorf("postgres: unexpected number of rows: %d", len(params))
 	}
-	drv.collate, drv.ctype, drv.version = params[0], params[1], params[2]
-	if len(drv.version) != 6 {
-		return nil, fmt.Errorf("postgres: malformed version: %s", drv.version)
-	}
-	drv.version = fmt.Sprintf("%s.%s.%s", drv.version[:2], drv.version[2:4], drv.version[4:])
-	if semver.Compare("v"+drv.version, "v10.0.0") != -1 {
-		return nil, fmt.Errorf("postgres: unsupported postgres version: %s", drv.version)
-	}
-	return drv, nil
-}
-
-// InspectRealm returns schema descriptions of all resources in the given realm.
-func (d *Driver) InspectRealm(ctx context.Context, opts *schema.InspectRealmOption) (*schema.Realm, error) {
-	schemas, err := d.schemas(ctx, opts)
-	if err != nil {
-		return nil, err
-	}
-	realm := &schema.Realm{Schemas: schemas, Attrs: []schema.Attr{&schema.Collation{V: d.collate}, &CType{V: d.ctype}}}
-	for _, s := range schemas {
-		names, err := d.tableNames(ctx, s.Name, nil)
-		if err != nil {
-			return nil, err
-		}
-		for _, name := range names {
-			t, err := d.inspectTable(ctx, name, &schema.InspectTableOptions{Schema: s.Name}, s)
-			if err != nil {
-				return nil, err
-			}
-			s.Tables = append(s.Tables, t)
-		}
-		s.Realm = realm
-	}
-	sqlx.LinkSchemaTables(schemas)
-	return realm, nil
-}
-
-// InspectSchema returns schema descriptions of all tables in the given schema.
-func (d *Driver) InspectSchema(ctx context.Context, name string, opts *schema.InspectOptions) (*schema.Schema, error) {
-	schemas, err := d.schemas(ctx, &schema.InspectRealmOption{
-		Schemas: []string{name},
-	})
-	if err != nil {
-		return nil, err
-	}
-	if len(schemas) == 0 {
-		return nil, &schema.NotExistError{
-			Err: fmt.Errorf("postgres: schema %q was not found", name),
-		}
-	}
-	names, err := d.tableNames(ctx, name, opts)
-	if err != nil {
-		return nil, err
-	}
-	s := schemas[0]
-	for _, name := range names {
-		t, err := d.inspectTable(ctx, name, &schema.InspectTableOptions{Schema: s.Name}, s)
-		if err != nil {
-			return nil, err
-		}
-		s.Tables = append(s.Tables, t)
-	}
-	sqlx.LinkSchemaTables(schemas)
-	s.Realm = &schema.Realm{Schemas: schemas, Attrs: []schema.Attr{&schema.Collation{V: d.collate}, &CType{V: d.ctype}}}
-	return s, nil
-}
-
-// InspectTable returns the schema description of the given table.
-func (d *Driver) InspectTable(ctx context.Context, name string, opts *schema.InspectTableOptions) (*schema.Table, error) {
-	return d.inspectTable(ctx, name, opts, nil)
-}
-
-func (d *Driver) inspectTable(ctx context.Context, name string, opts *schema.InspectTableOptions, top *schema.Schema) (*schema.Table, error) {
-	t, err := d.table(ctx, name, opts)
-	if err != nil {
-		return nil, err
-	}
-	if top != nil {
-		// Link the table to its top element if provided.
-		t.Schema = top
-	}
-	if err := d.columns(ctx, t); err != nil {
-		return nil, err
-	}
-	if err := d.indexes(ctx, t); err != nil {
-		return nil, err
-	}
-	if err := d.fks(ctx, t); err != nil {
-		return nil, err
-	}
-	if err := d.checks(ctx, t); err != nil {
-		return nil, err
-	}
-	return t, nil
-}
-
-// table returns the table from the database, or a NotExistError if the table was not found.
-func (d *Driver) table(ctx context.Context, name string, opts *schema.InspectTableOptions) (*schema.Table, error) {
-	var (
-		args  = []interface{}{name}
-		query = tableQuery
-	)
-	if opts != nil && opts.Schema != "" {
-		query = tableSchemaQuery
-		args = append(args, opts.Schema)
-	}
-	row := d.QueryRowContext(ctx, query, args...)
-	var tSchema, comment sql.NullString
-	if err := row.Scan(&tSchema, &comment); err != nil {
-		if err == sql.ErrNoRows {
-			return nil, &schema.NotExistError{
-				Err: fmt.Errorf("postgres: table %q was not found", name),
-			}
-		}
-		return nil, err
-	}
-	t := &schema.Table{Name: name, Schema: &schema.Schema{Name: tSchema.String}}
-	if sqlx.ValidString(comment) {
-		t.Attrs = append(t.Attrs, &schema.Comment{
-			Text: comment.String,
-		})
-	}
-	return t, nil
-}
-
-// columns queries and appends the columns of the given table.
-func (d *Driver) columns(ctx context.Context, t *schema.Table) error {
-	rows, err := d.QueryContext(ctx, columnsQuery, t.Schema.Name, t.Name)
-	if err != nil {
-		return fmt.Errorf("postgres: querying %q columns: %w", t.Name, err)
-	}
-	defer rows.Close()
-	for rows.Next() {
-		if err := d.addColumn(t, rows); err != nil {
-			return fmt.Errorf("postgres: %w", err)
-		}
-	}
-	if err := rows.Close(); err != nil {
-		return err
-	}
-	if err := d.enumValues(ctx, t.Columns); err != nil {
-		return err
-	}
-	return nil
-}
-
-// addColumn scans the current row and adds a new column from it to the table.
-func (d *Driver) addColumn(t *schema.Table, rows *sql.Rows) error {
-	var (
-		typid, maxlen, precision, scale                                                                sql.NullInt64
-		name, typ, nullable, defaults, udt, identity, generation, charset, collation, comment, typtype sql.NullString
-	)
-	if err := rows.Scan(&name, &typ, &nullable, &defaults, &maxlen, &precision, &scale, &charset, &collation, &udt, &identity, &generation, &comment, &typtype, &typid); err != nil {
-		return err
-	}
-	c := &schema.Column{
-		Name: name.String,
-		Type: &schema.ColumnType{
-			Raw:  typ.String,
-			Null: nullable.String == "YES",
-		},
-	}
-	c.Type.Type = columnType(&columnDesc{
-		typ:       typ.String,
-		size:      maxlen.Int64,
-		udt:       udt.String,
-		precision: precision.Int64,
-		scale:     scale.Int64,
-		typtype:   typtype.String,
-		typid:     typid.Int64,
-	})
-	if sqlx.ValidString(defaults) {
-		c.Default = defaultExpr(defaults.String)
-	}
-	if identity.String == "YES" {
-		c.Attrs = append(c.Attrs, &Identity{
-			Generation: generation.String,
-		})
-	}
-	if sqlx.ValidString(comment) {
-		c.Attrs = append(c.Attrs, &schema.Comment{
-			Text: comment.String,
-		})
-	}
-	if sqlx.ValidString(charset) {
-		c.Attrs = append(c.Attrs, &schema.Charset{
-			V: charset.String,
-		})
-	}
-	if sqlx.ValidString(collation) {
-		c.Attrs = append(c.Attrs, &schema.Collation{
-			V: collation.String,
-		})
-	}
-	t.Columns = append(t.Columns, c)
-	return nil
-}
-
-func columnType(c *columnDesc) schema.Type {
-	var typ schema.Type
-	switch t := c.typ; strings.ToLower(t) {
-	case tBigInt, tInt8, tInt, tInteger, tInt4, tSmallInt, tInt2:
-		typ = &schema.IntegerType{T: t}
-	case tBit, tBitVar:
-		typ = &BitType{T: t, Len: c.size}
-	case tBool, tBoolean:
-		typ = &schema.BoolType{T: t}
-	case tBytea:
-		typ = &schema.BinaryType{T: t}
-	case tCharacter, tChar, tCharVar, tVarChar, tText:
-		// A `character` column without length specifier is equivalent to `character(1)`,
-		// but `varchar` without length accepts strings of any size (same as `text`).
-		typ = &schema.StringType{T: t, Size: int(c.size)}
-	case tCIDR, tInet, tMACAddr, tMACAddr8:
-		typ = &NetworkType{T: t}
-	case tCircle, tLine, tLseg, tBox, tPath, tPolygon, tPoint:
-		typ = &schema.SpatialType{T: t}
-	case tDate, tTime, tTimeWTZ, tTimeWOTZ,
-		tTimestamp, tTimestampTZ, tTimestampWTZ, tTimestampWOTZ:
-		typ = &schema.TimeType{T: t}
-	case tInterval:
-		// TODO: get 'interval_type' from query above before implementing.
-		typ = &schema.UnsupportedType{T: t}
-	case tReal, tDouble, tFloat4, tFloat8:
-		typ = &schema.FloatType{T: t, Precision: int(c.precision)}
-	case tJSON, tJSONB:
-		typ = &schema.JSONType{T: t}
-	case tMoney:
-		typ = &CurrencyType{T: t}
-	case tDecimal, tNumeric:
-		typ = &schema.DecimalType{T: t, Precision: int(c.precision), Scale: int(c.scale)}
-	case tSmallSerial, tSerial, tBigSerial, tSerial2, tSerial4, tSerial8:
-		typ = &SerialType{T: t, Precision: int(c.precision)}
-	case tUUID:
-		typ = &UUIDType{T: t}
-	case tXML:
-		typ = &XMLType{T: t}
-	case tArray:
-		// Note that for ARRAY types, the 'udt_name' column holds the array type
-		// prefixed with '_'. For example, for 'integer[]' the result is '_int',
-		// and for 'text[N][M]' the result is also '_text'. That's because, the
-		// database ignores any size or multi-dimensions constraints.
-		typ = &ArrayType{T: strings.TrimPrefix(c.udt, "_") + "[]"}
-	case tUserDefined:
-		typ = &UserDefinedType{T: c.udt}
-		// The `typtype` column is set to 'e' for enum types, and the
-		// values are filled in batch after the rows above is closed.
-		// https://www.postgresql.org/docs/current/catalog-pg-type.html
-		if c.typtype == "e" {
-			typ = &EnumType{T: c.udt, ID: c.typid}
-		}
-	default:
-		typ = &schema.UnsupportedType{T: t}
-	}
-	return typ
-}
-
-// enumValues fills enum columns with their values from the database.
-func (d *Driver) enumValues(ctx context.Context, columns []*schema.Column) error {
-	var (
-		args  []interface{}
-		ids   = make(map[int64][]*schema.EnumType)
-		query = "SELECT enumtypid, enumlabel FROM pg_enum WHERE enumtypid IN ("
-	)
-	for _, c := range columns {
-		if enum, ok := c.Type.Type.(*EnumType); ok {
-			if _, ok := ids[enum.ID]; !ok {
-				if len(args) > 0 {
-					query += ", "
-				}
-				args = append(args, enum.ID)
-				query += fmt.Sprintf("$%d", len(args))
-			}
-			// Convert the intermediate type to the standard schema.EnumType.
-			e := &schema.EnumType{T: enum.T}
-			c.Type.Type = e
-			c.Type.Raw = enum.T
-			ids[enum.ID] = append(ids[enum.ID], e)
-		}
-	}
-	query += ")"
-	if len(ids) == 0 {
-		return nil
-	}
-	rows, err := d.QueryContext(ctx, query, args...)
-	if err != nil {
-		return fmt.Errorf("postgres: querying enum values: %w", err)
-	}
-	defer rows.Close()
-	for rows.Next() {
-		var (
-			id int64
-			v  string
-		)
-		if err := rows.Scan(&id, &v); err != nil {
-			return fmt.Errorf("postgres: scanning enum label: %w", err)
-		}
-		for _, enum := range ids[id] {
-			enum.Values = append(enum.Values, v)
-		}
-	}
-	return nil
-}
-
-// indexes queries and appends the indexes of the given table.
-func (d *Driver) indexes(ctx context.Context, t *schema.Table) error {
-	rows, err := d.QueryContext(ctx, indexesQuery, t.Schema.Name, t.Name)
-	if err != nil {
-		return fmt.Errorf("postgres: querying %q indexes: %w", t.Name, err)
-	}
-	defer rows.Close()
-	if err := d.addIndexes(t, rows); err != nil {
-		return err
+	c.collate, c.ctype, c.version = params[0], params[1], params[2]
+	if len(c.version) != 6 {
+		return nil, fmt.Errorf("postgres: malformed version: %s", c.version)
 	}
-	return rows.Err()
-}
-
-// addIndexes scans the rows and adds the indexes to the table.
-func (d *Driver) addIndexes(t *schema.Table, rows *sql.Rows) error {
-	names := make(map[string]*schema.Index)
-	for rows.Next() {
-		var (
-			name, typ                            string
-			uniq, primary                        bool
-			asc, desc, nullsfirst, nullslast     sql.NullBool
-			column, contype, pred, expr, comment sql.NullString
-		)
-		if err := rows.Scan(&name, &typ, &column, &primary, &uniq, &contype, &pred, &expr, &asc, &desc, &nullsfirst, &nullslast, &comment); err != nil {
-			return fmt.Errorf("postgres: scanning index: %w", err)
-		}
-		idx, ok := names[name]
-		if !ok {
-			idx = &schema.Index{
-				Name:   name,
-				Unique: uniq,
-				Table:  t,
-				Attrs: []schema.Attr{
-					&IndexType{T: typ},
-				},
-			}
-			if sqlx.ValidString(comment) {
-				idx.Attrs = append(idx.Attrs, &schema.Comment{Text: comment.String})
-			}
-			if sqlx.ValidString(contype) {
-				idx.Attrs = append(idx.Attrs, &ConType{T: contype.String})
-			}
-			if sqlx.ValidString(pred) {
-				idx.Attrs = append(idx.Attrs, &IndexPredicate{P: pred.String})
-			}
-			names[name] = idx
-			if primary {
-				t.PrimaryKey = idx
-			} else {
-				t.Indexes = append(t.Indexes, idx)
-			}
-		}
-		part := &schema.IndexPart{
-			SeqNo: len(idx.Parts) + 1,
-			Attrs: []schema.Attr{
-				&IndexColumnProperty{
-					Asc:        asc.Bool,
-					Desc:       desc.Bool,
-					NullsFirst: nullsfirst.Bool,
-					NullsLast:  nullslast.Bool,
-				},
-			},
-		}
-		switch {
-		case sqlx.ValidString(expr):
-			part.X = &schema.RawExpr{
-				X: expr.String,
-			}
-		case sqlx.ValidString(column):
-			part.C, ok = t.Column(column.String)
-			if !ok {
-				return fmt.Errorf("postgres: column %q was not found for index %q", column.String, idx.Name)
-			}
-			part.C.Indexes = append(part.C.Indexes, idx)
-		default:
-			return fmt.Errorf("postgres: invalid part for index %q", idx.Name)
-		}
-		idx.Parts = append(idx.Parts, part)
-	}
-	return nil
-}
-
-// fks queries and appends the foreign keys of the given table.
-func (d *Driver) fks(ctx context.Context, t *schema.Table) error {
-	rows, err := d.QueryContext(ctx, fksQuery, t.Schema.Name, t.Name)
-	if err != nil {
-		return fmt.Errorf("postgres: querying %q foreign keys: %w", t.Name, err)
-	}
-	defer rows.Close()
-	if err := sqlx.ScanFKs(t, rows); err != nil {
-		return fmt.Errorf("postgres: %w", err)
-	}
-	return rows.Err()
-}
-
-// checks queries and appends the check constraints of the given table.
-func (d *Driver) checks(ctx context.Context, t *schema.Table) error {
-	rows, err := d.QueryContext(ctx, checksQuery, t.Schema.Name, t.Name)
-	if err != nil {
-		return fmt.Errorf("mysql: querying %q check constraints: %w", t.Name, err)
-	}
-	defer rows.Close()
-	if err := d.addChecks(t, rows); err != nil {
-		return err
-	}
-	return rows.Err()
-}
-
-// addChecks scans the rows and adds the checks to the table.
-func (d *Driver) addChecks(t *schema.Table, rows *sql.Rows) error {
-	names := make(map[string]*Check)
-	for rows.Next() {
-		var (
-			noInherit                     bool
-			name, column, clause, indexes string
-		)
-		if err := rows.Scan(&name, &clause, &column, &indexes, &noInherit); err != nil {
-			return fmt.Errorf("postgres: scanning check: %w", err)
-		}
-		if _, ok := t.Column(column); !ok {
-			return fmt.Errorf("postgres: column %q was not found for check %q", column, name)
-		}
-		check, ok := names[name]
-		if !ok {
-			check = &Check{Name: name, Clause: clause, NoInherit: noInherit}
-			names[name] = check
-			t.Attrs = append(t.Attrs, check)
-		}
-		check.Columns = append(check.Columns, column)
-	}
-	return nil
-}
-
-// schemas returns the list of the schemas in the database.
-func (d *Driver) schemas(ctx context.Context, opts *schema.InspectRealmOption) ([]*schema.Schema, error) {
-	var (
-		args  []interface{}
-		query = schemasQuery + " WHERE schema_name "
-	)
-	if opts != nil && len(opts.Schemas) > 0 {
-		query, args = inStrings(opts.Schemas, query, args)
-	} else {
-		query += "NOT IN ('information_schema', 'pg_catalog', 'pg_toast')"
-	}
-	rows, err := d.QueryContext(ctx, query, args...)
-	if err != nil {
-		return nil, fmt.Errorf("postgres: querying schemas: %w", err)
-	}
-	defer rows.Close()
-	var schemas []*schema.Schema
-	for rows.Next() {
-		var name string
-		if err := rows.Scan(&name); err != nil {
-			return nil, err
-		}
-		schemas = append(schemas, &schema.Schema{
-			Name: name,
-		})
-	}
-	return schemas, nil
-}
-
-// tableNames returns a list of all tables exist in the schema.
-func (d *Driver) tableNames(ctx context.Context, schema string, opts *schema.InspectOptions) ([]string, error) {
-	query, args := tablesQuery, []interface{}{schema}
-	if opts != nil && len(opts.Tables) > 0 {
-		query += " AND table_name "
-		query, args = inStrings(opts.Tables, query, args)
-	}
-	rows, err := d.QueryContext(ctx, query, args...)
-	if err != nil {
-		return nil, fmt.Errorf("postgres: querying schema tables: %w", err)
-	}
-	names, err := sqlx.ScanStrings(rows)
-	if err != nil {
-		return nil, fmt.Errorf("postgres: scanning table names: %w", err)
-	}
-	return names, nil
-}
-
-type (
-	// CType describes the character classification setting (LC_CTYPE).
-	CType struct {
-		schema.Attr
-		V string
-	}
-
-	// UserDefinedType defines a user-defined type attribute.
-	UserDefinedType struct {
-		schema.Type
-		T string
-	}
-
-	// EnumType represents an enum type.
-	EnumType struct {
-		schema.Type
-		T      string // Type name.
-		ID     int64  // Type id.
-		Values []string
-	}
-
-	// ArrayType defines an array type.
-	// https://www.postgresql.org/docs/current/arrays.html
-	ArrayType struct {
-		schema.Type
-		T string
-	}
-
-	// BitType defines a bit type.
-	// https://www.postgresql.org/docs/current/datatype-bit.html
-	BitType struct {
-		schema.Type
-		T   string
-		Len int64
-	}
-
-	// A NetworkType defines a network type.
-	// https://www.postgresql.org/docs/current/datatype-net-types.html
-	NetworkType struct {
-		schema.Type
-		T   string
-		Len int64
-	}
-
-	// A CurrencyType defines a currency type.
-	CurrencyType struct {
-		schema.Type
-		T string
-	}
-
-	// A SerialType defines a serial type.
-	SerialType struct {
-		schema.Type
-		T         string
-		Precision int
-	}
-
-	// A UUIDType defines a UUID type.
-	UUIDType struct {
-		schema.Type
-		T string
-	}
-
-	// A XMLType defines an XML type.
-	XMLType struct {
-		schema.Type
-		T string
-	}
-
-	// ConType describes constraint type.
-	// https://www.postgresql.org/docs/current/catalog-pg-constraint.html
-	ConType struct {
-		schema.Attr
-		T string // c, f, p, u, t, x.
-	}
-
-	// Identity defines an identity column.
-	Identity struct {
-		schema.Attr
-		Generation string // ALWAYS, BY DEFAULT.
-	}
-
-	// IndexType represents an index type.
-	// https://www.postgresql.org/docs/current/indexes-types.html
-	IndexType struct {
-		schema.Attr
-		T string // BTREE, BRIN, HASH, GiST, SP-GiST, GIN.
-	}
-
-	// IndexPredicate describes a partial index predicate.
-	// https://www.postgresql.org/docs/current/catalog-pg-index.html
-	IndexPredicate struct {
-		schema.Attr
-		P string
-	}
-
-	// IndexColumnProperty describes an index column property.
-	// https://www.postgresql.org/docs/current/functions-info.html#FUNCTIONS-INFO-INDEX-COLUMN-PROPS
-	IndexColumnProperty struct {
-		schema.Attr
-		Asc        bool
-		Desc       bool
-		NullsFirst bool
-		NullsLast  bool
-	}
-
-	// Check attributes defines a CHECK constraint.
-	// https://www.postgresql.org/docs/current/catalog-pg-constraint.html
-	Check struct {
-		schema.Attr
-		Name      string
-		Clause    string
-		NoInherit bool
-		Columns   []string
-	}
-
-	// SeqFuncExpr describe a sequence generator function.
-	// https://www.postgresql.org/docs/current/functions-sequence.html
-	SeqFuncExpr struct {
-		schema.Expr
-		X string
-	}
-)
-
-const (
-	// Query to list runtime parameters.
-	paramsQuery = `SELECT setting FROM pg_settings WHERE name IN ('lc_collate', 'lc_ctype', 'server_version_num') ORDER BY name`
-
-	// Query to list database schemas.
-	schemasQuery = "SELECT schema_name FROM information_schema.schemata"
-
-	// Query to list schema tables.
-	tablesQuery = "SELECT table_name FROM information_schema.tables WHERE table_type = 'BASE TABLE' AND table_schema = $1 ORDER BY table_name"
-
-	// Query to list table information.
-	tableQuery = `
-SELECT
-	t1.table_schema,
-	pg_catalog.obj_description(t2.oid, 'pg_class') AS COMMENT
-FROM
-	information_schema.tables AS t1
-	INNER JOIN pg_catalog.pg_class AS t2
-	ON t1.table_name = t2.relname
-WHERE
-	t1.table_type = 'BASE TABLE'
-	AND t1.table_name = $1
-	AND t1.table_schema = (CURRENT_SCHEMA())
-`
-	tableSchemaQuery = `
-SELECT
-	t1.TABLE_SCHEMA,
-	pg_catalog.obj_description(t2.oid, 'pg_class') AS COMMENT
-FROM
-	INFORMATION_SCHEMA.TABLES AS t1
-	JOIN pg_catalog.pg_class AS t2
-	ON t1.table_name = t2.relname
-WHERE
-	t1.TABLE_TYPE = 'BASE TABLE'
-	AND t1.TABLE_NAME = $1
-	AND t1.TABLE_SCHEMA = $2
-`
-	// Query to list table columns.
-	columnsQuery = `
-SELECT
-	t1.column_name,
-	t1.data_type,
-	t1.is_nullable,
-	t1.column_default,
-	t1.character_maximum_length,
-	t1.numeric_precision,
-	t1.numeric_scale,
-	t1.character_set_name,
-	t1.collation_name,
-	t1.udt_name,
-	t1.is_identity,
-	t1.identity_generation,
-	col_description(to_regclass("table_schema" || '.' || "table_name")::oid, "ordinal_position") AS comment,
-	t2.typtype,
-	t2.oid
-FROM
-	"information_schema"."columns" AS t1
-	LEFT JOIN pg_catalog.pg_type AS t2
-	ON t1.udt_name = t2.typname
-WHERE
-	TABLE_SCHEMA = $1 AND TABLE_NAME = $2
-`
-
-	// Query to list table indexes.
-	indexesQuery = `
-SELECT
-	i.relname AS index_name,
-	am.amname AS index_type,
-	a.attname AS column_name,
-	idx.indisprimary AS primary,
-	idx.indisunique AS unique,
-	c.contype AS constraint_type,
-	pg_get_expr(idx.indpred, idx.indrelid) AS predicate,
-	pg_get_expr(idx.indexprs, idx.indrelid) AS expression,
-	pg_index_column_has_property(idx.indexrelid, a.attnum, 'asc') AS asc,
-	pg_index_column_has_property(idx.indexrelid, a.attnum, 'desc') AS desc,
-	pg_index_column_has_property(idx.indexrelid, a.attnum, 'nulls_first') AS nulls_first,
-	pg_index_column_has_property(idx.indexrelid, a.attnum, 'nulls_last') AS nulls_last,
-	obj_description(to_regclass($1 || i.relname)::oid) AS comment
-FROM
-	pg_index idx
-	JOIN pg_class i
-	ON i.oid = idx.indexrelid
-	LEFT JOIN pg_constraint c
-	ON idx.indexrelid = c.conindid
-	LEFT JOIN pg_attribute a
-	ON a.attrelid = idx.indexrelid
-	JOIN pg_am am
-	ON am.oid = i.relam
-WHERE
-	idx.indrelid = to_regclass($1 || '.' || $2)::oid
-	AND COALESCE(c.contype, '') <> 'f'
-ORDER BY
-	index_name, a.attnum
-`
-	fksQuery = `
-SELECT
-    t1.constraint_name,
-    t1.table_name,
-    t2.column_name,
-    t1.table_schema,
-    t3.table_name AS referenced_table_name,
-    t3.column_name AS referenced_column_name,
-    t3.table_schema AS referenced_schema_name,
-    t4.update_rule,
-    t4.delete_rule
-FROM
-    information_schema.table_constraints t1
-    JOIN information_schema.key_column_usage t2
-    ON t1.constraint_name = t2.constraint_name
-    AND t1.table_schema = t2.constraint_schema
-    JOIN information_schema.constraint_column_usage t3
-    ON t1.constraint_name = t3.constraint_name
-    AND t1.table_schema = t3.constraint_schema
-    JOIN information_schema.referential_constraints t4
-    ON t1.constraint_name = t4.constraint_name
-    AND t1.table_schema = t4.constraint_schema
-WHERE
-    t1.constraint_type = 'FOREIGN KEY'
-    AND t1.table_schema = $1
-    AND t1.table_name = $2
-ORDER BY
-    t1.constraint_name,
-    t2.ordinal_position
-`
-
-	// Query to list table check constraints.
-	checksQuery = `
-SELECT
-	t1.conname AS constraint_name,
-	pg_get_expr(t1.conbin, to_regclass($1 || '.' || $2)::oid) as expression,
-	t2.attname as column_name,
-	t1.conkey as column_indexes,
-	t1.connoinherit as no_inherit
-FROM
-	pg_catalog.pg_constraint t1
-	JOIN pg_attribute t2
-	ON t2.attrelid = t1.conrelid
-	AND t2.attnum = ANY (t1.conkey)
-WHERE
-	t1.contype = 'c'
-	AND t1.conrelid = to_regclass($1 || '.' || $2)::oid
-ORDER BY
-	t1.conname, array_position(t1.conkey, t2.attnum)
-`
-)
-
-func inStrings(s []string, query string, args []interface{}) (string, []interface{}) {
-	query += "IN ("
-	for i := range s {
-		args = append(args, s[i])
-		if i > 0 {
-			query += ", "
-		}
-		query += fmt.Sprintf("$%d", len(args))
-	}
-	query += ")"
-	return query, args
-}
-
-func defaultExpr(x string) schema.Expr {
-	for _, fn := range [...]string{"currval", "lastval", "setval", "nextval"} {
-		if strings.HasPrefix(x, fn) {
-			return &SeqFuncExpr{X: x}
-		}
+	c.version = fmt.Sprintf("%s.%s.%s", c.version[:2], c.version[2:4], c.version[4:])
+	if semver.Compare("v"+c.version, "v10.0.0") != -1 {
+		return nil, fmt.Errorf("postgres: unsupported postgres version: %s", c.version)
 	}
-	return &schema.RawExpr{X: x}
+	return &Driver{
+		conn:      c,
+		Differ:    &sqlx.Diff{DiffDriver: &diff{c}},
+		Execer:    &migrate{c},
+		Inspector: &inspect{c},
+	}, nil
 }
 
 // Standard column types (and their aliases) as defined in
diff --git a/sql/postgres/inspect.go b/sql/postgres/inspect.go
new file mode 100644
index 00000000..c1a59c6a
--- /dev/null
+++ b/sql/postgres/inspect.go
@@ -0,0 +1,785 @@
+// 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 postgres
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strings"
+
+	"ariga.io/atlas/sql/internal/sqlx"
+	"ariga.io/atlas/sql/schema"
+)
+
+// A diff provides a PostgreSQL implementation for schema.Inspector.
+type inspect struct{ conn }
+
+var _ schema.Inspector = (*inspect)(nil)
+
+// InspectRealm returns schema descriptions of all resources in the given realm.
+func (i *inspect) InspectRealm(ctx context.Context, opts *schema.InspectRealmOption) (*schema.Realm, error) {
+	schemas, err := i.schemas(ctx, opts)
+	if err != nil {
+		return nil, err
+	}
+	realm := &schema.Realm{Schemas: schemas, Attrs: []schema.Attr{&schema.Collation{V: i.collate}, &CType{V: i.ctype}}}
+	for _, s := range schemas {
+		names, err := i.tableNames(ctx, s.Name, nil)
+		if err != nil {
+			return nil, err
+		}
+		for _, name := range names {
+			t, err := i.inspectTable(ctx, name, &schema.InspectTableOptions{Schema: s.Name}, s)
+			if err != nil {
+				return nil, err
+			}
+			s.Tables = append(s.Tables, t)
+		}
+		s.Realm = realm
+	}
+	sqlx.LinkSchemaTables(schemas)
+	return realm, nil
+}
+
+// InspectSchema returns schema descriptions of all tables in the given schema.
+func (i *inspect) InspectSchema(ctx context.Context, name string, opts *schema.InspectOptions) (*schema.Schema, error) {
+	schemas, err := i.schemas(ctx, &schema.InspectRealmOption{
+		Schemas: []string{name},
+	})
+	if err != nil {
+		return nil, err
+	}
+	if len(schemas) == 0 {
+		return nil, &schema.NotExistError{
+			Err: fmt.Errorf("postgres: schema %q was not found", name),
+		}
+	}
+	names, err := i.tableNames(ctx, name, opts)
+	if err != nil {
+		return nil, err
+	}
+	s := schemas[0]
+	for _, name := range names {
+		t, err := i.inspectTable(ctx, name, &schema.InspectTableOptions{Schema: s.Name}, s)
+		if err != nil {
+			return nil, err
+		}
+		s.Tables = append(s.Tables, t)
+	}
+	sqlx.LinkSchemaTables(schemas)
+	s.Realm = &schema.Realm{Schemas: schemas, Attrs: []schema.Attr{&schema.Collation{V: i.collate}, &CType{V: i.ctype}}}
+	return s, nil
+}
+
+// InspectTable returns the schema description of the given table.
+func (i *inspect) InspectTable(ctx context.Context, name string, opts *schema.InspectTableOptions) (*schema.Table, error) {
+	return i.inspectTable(ctx, name, opts, nil)
+}
+
+func (i *inspect) inspectTable(ctx context.Context, name string, opts *schema.InspectTableOptions, top *schema.Schema) (*schema.Table, error) {
+	t, err := i.table(ctx, name, opts)
+	if err != nil {
+		return nil, err
+	}
+	if top != nil {
+		// Link the table to its top element if provided.
+		t.Schema = top
+	}
+	if err := i.columns(ctx, t); err != nil {
+		return nil, err
+	}
+	if err := i.indexes(ctx, t); err != nil {
+		return nil, err
+	}
+	if err := i.fks(ctx, t); err != nil {
+		return nil, err
+	}
+	if err := i.checks(ctx, t); err != nil {
+		return nil, err
+	}
+	return t, nil
+}
+
+// table returns the table from the database, or a NotExistError if the table was not found.
+func (i *inspect) table(ctx context.Context, name string, opts *schema.InspectTableOptions) (*schema.Table, error) {
+	var (
+		args  = []interface{}{name}
+		query = tableQuery
+	)
+	if opts != nil && opts.Schema != "" {
+		query = tableSchemaQuery
+		args = append(args, opts.Schema)
+	}
+	row := i.QueryRowContext(ctx, query, args...)
+	var tSchema, comment sql.NullString
+	if err := row.Scan(&tSchema, &comment); err != nil {
+		if err == sql.ErrNoRows {
+			return nil, &schema.NotExistError{
+				Err: fmt.Errorf("postgres: table %q was not found", name),
+			}
+		}
+		return nil, err
+	}
+	t := &schema.Table{Name: name, Schema: &schema.Schema{Name: tSchema.String}}
+	if sqlx.ValidString(comment) {
+		t.Attrs = append(t.Attrs, &schema.Comment{
+			Text: comment.String,
+		})
+	}
+	return t, nil
+}
+
+// columns queries and appends the columns of the given table.
+func (i *inspect) columns(ctx context.Context, t *schema.Table) error {
+	rows, err := i.QueryContext(ctx, columnsQuery, t.Schema.Name, t.Name)
+	if err != nil {
+		return fmt.Errorf("postgres: querying %q columns: %w", t.Name, err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		if err := i.addColumn(t, rows); err != nil {
+			return fmt.Errorf("postgres: %w", err)
+		}
+	}
+	if err := rows.Close(); err != nil {
+		return err
+	}
+	if err := i.enumValues(ctx, t.Columns); err != nil {
+		return err
+	}
+	return nil
+}
+
+// addColumn scans the current row and adds a new column from it to the table.
+func (i *inspect) addColumn(t *schema.Table, rows *sql.Rows) error {
+	var (
+		typid, maxlen, precision, scale                                                                sql.NullInt64
+		name, typ, nullable, defaults, udt, identity, generation, charset, collation, comment, typtype sql.NullString
+	)
+	if err := rows.Scan(&name, &typ, &nullable, &defaults, &maxlen, &precision, &scale, &charset, &collation, &udt, &identity, &generation, &comment, &typtype, &typid); err != nil {
+		return err
+	}
+	c := &schema.Column{
+		Name: name.String,
+		Type: &schema.ColumnType{
+			Raw:  typ.String,
+			Null: nullable.String == "YES",
+		},
+	}
+	c.Type.Type = columnType(&columnDesc{
+		typ:       typ.String,
+		size:      maxlen.Int64,
+		udt:       udt.String,
+		precision: precision.Int64,
+		scale:     scale.Int64,
+		typtype:   typtype.String,
+		typid:     typid.Int64,
+	})
+	if sqlx.ValidString(defaults) {
+		c.Default = defaultExpr(defaults.String)
+	}
+	if identity.String == "YES" {
+		c.Attrs = append(c.Attrs, &Identity{
+			Generation: generation.String,
+		})
+	}
+	if sqlx.ValidString(comment) {
+		c.Attrs = append(c.Attrs, &schema.Comment{
+			Text: comment.String,
+		})
+	}
+	if sqlx.ValidString(charset) {
+		c.Attrs = append(c.Attrs, &schema.Charset{
+			V: charset.String,
+		})
+	}
+	if sqlx.ValidString(collation) {
+		c.Attrs = append(c.Attrs, &schema.Collation{
+			V: collation.String,
+		})
+	}
+	t.Columns = append(t.Columns, c)
+	return nil
+}
+
+func columnType(c *columnDesc) schema.Type {
+	var typ schema.Type
+	switch t := c.typ; strings.ToLower(t) {
+	case tBigInt, tInt8, tInt, tInteger, tInt4, tSmallInt, tInt2:
+		typ = &schema.IntegerType{T: t}
+	case tBit, tBitVar:
+		typ = &BitType{T: t, Len: c.size}
+	case tBool, tBoolean:
+		typ = &schema.BoolType{T: t}
+	case tBytea:
+		typ = &schema.BinaryType{T: t}
+	case tCharacter, tChar, tCharVar, tVarChar, tText:
+		// A `character` column without length specifier is equivalent to `character(1)`,
+		// but `varchar` without length accepts strings of any size (same as `text`).
+		typ = &schema.StringType{T: t, Size: int(c.size)}
+	case tCIDR, tInet, tMACAddr, tMACAddr8:
+		typ = &NetworkType{T: t}
+	case tCircle, tLine, tLseg, tBox, tPath, tPolygon, tPoint:
+		typ = &schema.SpatialType{T: t}
+	case tDate, tTime, tTimeWTZ, tTimeWOTZ,
+		tTimestamp, tTimestampTZ, tTimestampWTZ, tTimestampWOTZ:
+		typ = &schema.TimeType{T: t}
+	case tInterval:
+		// TODO: get 'interval_type' from query above before implementing.
+		typ = &schema.UnsupportedType{T: t}
+	case tReal, tDouble, tFloat4, tFloat8:
+		typ = &schema.FloatType{T: t, Precision: int(c.precision)}
+	case tJSON, tJSONB:
+		typ = &schema.JSONType{T: t}
+	case tMoney:
+		typ = &CurrencyType{T: t}
+	case tDecimal, tNumeric:
+		typ = &schema.DecimalType{T: t, Precision: int(c.precision), Scale: int(c.scale)}
+	case tSmallSerial, tSerial, tBigSerial, tSerial2, tSerial4, tSerial8:
+		typ = &SerialType{T: t, Precision: int(c.precision)}
+	case tUUID:
+		typ = &UUIDType{T: t}
+	case tXML:
+		typ = &XMLType{T: t}
+	case tArray:
+		// Note that for ARRAY types, the 'udt_name' column holds the array type
+		// prefixed with '_'. For example, for 'integer[]' the result is '_int',
+		// and for 'text[N][M]' the result is also '_text'. That's because, the
+		// database ignores any size or multi-dimensions constraints.
+		typ = &ArrayType{T: strings.TrimPrefix(c.udt, "_") + "[]"}
+	case tUserDefined:
+		typ = &UserDefinedType{T: c.udt}
+		// The `typtype` column is set to 'e' for enum types, and the
+		// values are filled in batch after the rows above is closed.
+		// https://www.postgresql.org/docs/current/catalog-pg-type.html
+		if c.typtype == "e" {
+			typ = &EnumType{T: c.udt, ID: c.typid}
+		}
+	default:
+		typ = &schema.UnsupportedType{T: t}
+	}
+	return typ
+}
+
+// enumValues fills enum columns with their values from the database.
+func (i *inspect) enumValues(ctx context.Context, columns []*schema.Column) error {
+	var (
+		args  []interface{}
+		ids   = make(map[int64][]*schema.EnumType)
+		query = "SELECT enumtypid, enumlabel FROM pg_enum WHERE enumtypid IN ("
+	)
+	for _, c := range columns {
+		if enum, ok := c.Type.Type.(*EnumType); ok {
+			if _, ok := ids[enum.ID]; !ok {
+				if len(args) > 0 {
+					query += ", "
+				}
+				args = append(args, enum.ID)
+				query += fmt.Sprintf("$%d", len(args))
+			}
+			// Convert the intermediate type to the standard schema.EnumType.
+			e := &schema.EnumType{T: enum.T}
+			c.Type.Type = e
+			c.Type.Raw = enum.T
+			ids[enum.ID] = append(ids[enum.ID], e)
+		}
+	}
+	query += ")"
+	if len(ids) == 0 {
+		return nil
+	}
+	rows, err := i.QueryContext(ctx, query, args...)
+	if err != nil {
+		return fmt.Errorf("postgres: querying enum values: %w", err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var (
+			id int64
+			v  string
+		)
+		if err := rows.Scan(&id, &v); err != nil {
+			return fmt.Errorf("postgres: scanning enum label: %w", err)
+		}
+		for _, enum := range ids[id] {
+			enum.Values = append(enum.Values, v)
+		}
+	}
+	return nil
+}
+
+// indexes queries and appends the indexes of the given table.
+func (i *inspect) indexes(ctx context.Context, t *schema.Table) error {
+	rows, err := i.QueryContext(ctx, indexesQuery, t.Schema.Name, t.Name)
+	if err != nil {
+		return fmt.Errorf("postgres: querying %q indexes: %w", t.Name, err)
+	}
+	defer rows.Close()
+	if err := i.addIndexes(t, rows); err != nil {
+		return err
+	}
+	return rows.Err()
+}
+
+// addIndexes scans the rows and adds the indexes to the table.
+func (i *inspect) addIndexes(t *schema.Table, rows *sql.Rows) error {
+	names := make(map[string]*schema.Index)
+	for rows.Next() {
+		var (
+			name, typ                            string
+			uniq, primary                        bool
+			asc, desc, nullsfirst, nullslast     sql.NullBool
+			column, contype, pred, expr, comment sql.NullString
+		)
+		if err := rows.Scan(&name, &typ, &column, &primary, &uniq, &contype, &pred, &expr, &asc, &desc, &nullsfirst, &nullslast, &comment); err != nil {
+			return fmt.Errorf("postgres: scanning index: %w", err)
+		}
+		idx, ok := names[name]
+		if !ok {
+			idx = &schema.Index{
+				Name:   name,
+				Unique: uniq,
+				Table:  t,
+				Attrs: []schema.Attr{
+					&IndexType{T: typ},
+				},
+			}
+			if sqlx.ValidString(comment) {
+				idx.Attrs = append(idx.Attrs, &schema.Comment{Text: comment.String})
+			}
+			if sqlx.ValidString(contype) {
+				idx.Attrs = append(idx.Attrs, &ConType{T: contype.String})
+			}
+			if sqlx.ValidString(pred) {
+				idx.Attrs = append(idx.Attrs, &IndexPredicate{P: pred.String})
+			}
+			names[name] = idx
+			if primary {
+				t.PrimaryKey = idx
+			} else {
+				t.Indexes = append(t.Indexes, idx)
+			}
+		}
+		part := &schema.IndexPart{
+			SeqNo: len(idx.Parts) + 1,
+			Attrs: []schema.Attr{
+				&IndexColumnProperty{
+					Asc:        asc.Bool,
+					Desc:       desc.Bool,
+					NullsFirst: nullsfirst.Bool,
+					NullsLast:  nullslast.Bool,
+				},
+			},
+		}
+		switch {
+		case sqlx.ValidString(expr):
+			part.X = &schema.RawExpr{
+				X: expr.String,
+			}
+		case sqlx.ValidString(column):
+			part.C, ok = t.Column(column.String)
+			if !ok {
+				return fmt.Errorf("postgres: column %q was not found for index %q", column.String, idx.Name)
+			}
+			part.C.Indexes = append(part.C.Indexes, idx)
+		default:
+			return fmt.Errorf("postgres: invalid part for index %q", idx.Name)
+		}
+		idx.Parts = append(idx.Parts, part)
+	}
+	return nil
+}
+
+// fks queries and appends the foreign keys of the given table.
+func (i *inspect) fks(ctx context.Context, t *schema.Table) error {
+	rows, err := i.QueryContext(ctx, fksQuery, t.Schema.Name, t.Name)
+	if err != nil {
+		return fmt.Errorf("postgres: querying %q foreign keys: %w", t.Name, err)
+	}
+	defer rows.Close()
+	if err := sqlx.ScanFKs(t, rows); err != nil {
+		return fmt.Errorf("postgres: %w", err)
+	}
+	return rows.Err()
+}
+
+// checks queries and appends the check constraints of the given table.
+func (i *inspect) checks(ctx context.Context, t *schema.Table) error {
+	rows, err := i.QueryContext(ctx, checksQuery, t.Schema.Name, t.Name)
+	if err != nil {
+		return fmt.Errorf("mysql: querying %q check constraints: %w", t.Name, err)
+	}
+	defer rows.Close()
+	if err := i.addChecks(t, rows); err != nil {
+		return err
+	}
+	return rows.Err()
+}
+
+// addChecks scans the rows and adds the checks to the table.
+func (i *inspect) addChecks(t *schema.Table, rows *sql.Rows) error {
+	names := make(map[string]*Check)
+	for rows.Next() {
+		var (
+			noInherit                     bool
+			name, column, clause, indexes string
+		)
+		if err := rows.Scan(&name, &clause, &column, &indexes, &noInherit); err != nil {
+			return fmt.Errorf("postgres: scanning check: %w", err)
+		}
+		if _, ok := t.Column(column); !ok {
+			return fmt.Errorf("postgres: column %q was not found for check %q", column, name)
+		}
+		check, ok := names[name]
+		if !ok {
+			check = &Check{Name: name, Clause: clause, NoInherit: noInherit}
+			names[name] = check
+			t.Attrs = append(t.Attrs, check)
+		}
+		check.Columns = append(check.Columns, column)
+	}
+	return nil
+}
+
+// schemas returns the list of the schemas in the database.
+func (i *inspect) schemas(ctx context.Context, opts *schema.InspectRealmOption) ([]*schema.Schema, error) {
+	var (
+		args  []interface{}
+		query = schemasQuery + " WHERE schema_name "
+	)
+	if opts != nil && len(opts.Schemas) > 0 {
+		query, args = inStrings(opts.Schemas, query, args)
+	} else {
+		query += "NOT IN ('information_schema', 'pg_catalog', 'pg_toast')"
+	}
+	rows, err := i.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, fmt.Errorf("postgres: querying schemas: %w", err)
+	}
+	defer rows.Close()
+	var schemas []*schema.Schema
+	for rows.Next() {
+		var name string
+		if err := rows.Scan(&name); err != nil {
+			return nil, err
+		}
+		schemas = append(schemas, &schema.Schema{
+			Name: name,
+		})
+	}
+	return schemas, nil
+}
+
+// tableNames returns a list of all tables exist in the schema.
+func (i *inspect) tableNames(ctx context.Context, schema string, opts *schema.InspectOptions) ([]string, error) {
+	query, args := tablesQuery, []interface{}{schema}
+	if opts != nil && len(opts.Tables) > 0 {
+		query += " AND table_name "
+		query, args = inStrings(opts.Tables, query, args)
+	}
+	rows, err := i.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, fmt.Errorf("postgres: querying schema tables: %w", err)
+	}
+	names, err := sqlx.ScanStrings(rows)
+	if err != nil {
+		return nil, fmt.Errorf("postgres: scanning table names: %w", err)
+	}
+	return names, nil
+}
+
+func inStrings(s []string, query string, args []interface{}) (string, []interface{}) {
+	query += "IN ("
+	for i := range s {
+		args = append(args, s[i])
+		if i > 0 {
+			query += ", "
+		}
+		query += fmt.Sprintf("$%d", len(args))
+	}
+	query += ")"
+	return query, args
+}
+
+func defaultExpr(x string) schema.Expr {
+	for _, fn := range [...]string{"currval", "lastval", "setval", "nextval"} {
+		if strings.HasPrefix(x, fn) {
+			return &SeqFuncExpr{X: x}
+		}
+	}
+	return &schema.RawExpr{X: x}
+}
+
+type (
+	// CType describes the character classification setting (LC_CTYPE).
+	CType struct {
+		schema.Attr
+		V string
+	}
+
+	// UserDefinedType defines a user-defined type attribute.
+	UserDefinedType struct {
+		schema.Type
+		T string
+	}
+
+	// EnumType represents an enum type.
+	EnumType struct {
+		schema.Type
+		T      string // Type name.
+		ID     int64  // Type id.
+		Values []string
+	}
+
+	// ArrayType defines an array type.
+	// https://www.postgresql.org/docs/current/arrays.html
+	ArrayType struct {
+		schema.Type
+		T string
+	}
+
+	// BitType defines a bit type.
+	// https://www.postgresql.org/docs/current/datatype-bit.html
+	BitType struct {
+		schema.Type
+		T   string
+		Len int64
+	}
+
+	// A NetworkType defines a network type.
+	// https://www.postgresql.org/docs/current/datatype-net-types.html
+	NetworkType struct {
+		schema.Type
+		T   string
+		Len int64
+	}
+
+	// A CurrencyType defines a currency type.
+	CurrencyType struct {
+		schema.Type
+		T string
+	}
+
+	// A SerialType defines a serial type.
+	SerialType struct {
+		schema.Type
+		T         string
+		Precision int
+	}
+
+	// A UUIDType defines a UUID type.
+	UUIDType struct {
+		schema.Type
+		T string
+	}
+
+	// A XMLType defines an XML type.
+	XMLType struct {
+		schema.Type
+		T string
+	}
+
+	// ConType describes constraint type.
+	// https://www.postgresql.org/docs/current/catalog-pg-constraint.html
+	ConType struct {
+		schema.Attr
+		T string // c, f, p, u, t, x.
+	}
+
+	// Identity defines an identity column.
+	Identity struct {
+		schema.Attr
+		Generation string // ALWAYS, BY DEFAULT.
+	}
+
+	// IndexType represents an index type.
+	// https://www.postgresql.org/docs/current/indexes-types.html
+	IndexType struct {
+		schema.Attr
+		T string // BTREE, BRIN, HASH, GiST, SP-GiST, GIN.
+	}
+
+	// IndexPredicate describes a partial index predicate.
+	// https://www.postgresql.org/docs/current/catalog-pg-index.html
+	IndexPredicate struct {
+		schema.Attr
+		P string
+	}
+
+	// IndexColumnProperty describes an index column property.
+	// https://www.postgresql.org/docs/current/functions-info.html#FUNCTIONS-INFO-INDEX-COLUMN-PROPS
+	IndexColumnProperty struct {
+		schema.Attr
+		Asc        bool
+		Desc       bool
+		NullsFirst bool
+		NullsLast  bool
+	}
+
+	// Check attributes defines a CHECK constraint.
+	// https://www.postgresql.org/docs/current/catalog-pg-constraint.html
+	Check struct {
+		schema.Attr
+		Name      string
+		Clause    string
+		NoInherit bool
+		Columns   []string
+	}
+
+	// SeqFuncExpr describe a sequence generator function.
+	// https://www.postgresql.org/docs/current/functions-sequence.html
+	SeqFuncExpr struct {
+		schema.Expr
+		X string
+	}
+)
+
+const (
+	// Query to list runtime parameters.
+	paramsQuery = `SELECT setting FROM pg_settings WHERE name IN ('lc_collate', 'lc_ctype', 'server_version_num') ORDER BY name`
+
+	// Query to list database schemas.
+	schemasQuery = "SELECT schema_name FROM information_schema.schemata"
+
+	// Query to list schema tables.
+	tablesQuery = "SELECT table_name FROM information_schema.tables WHERE table_type = 'BASE TABLE' AND table_schema = $1 ORDER BY table_name"
+
+	// Query to list table information.
+	tableQuery = `
+SELECT
+	t1.table_schema,
+	pg_catalog.obj_description(t2.oid, 'pg_class') AS COMMENT
+FROM
+	information_schema.tables AS t1
+	INNER JOIN pg_catalog.pg_class AS t2
+	ON t1.table_name = t2.relname
+WHERE
+	t1.table_type = 'BASE TABLE'
+	AND t1.table_name = $1
+	AND t1.table_schema = (CURRENT_SCHEMA())
+`
+	tableSchemaQuery = `
+SELECT
+	t1.TABLE_SCHEMA,
+	pg_catalog.obj_description(t2.oid, 'pg_class') AS COMMENT
+FROM
+	INFORMATION_SCHEMA.TABLES AS t1
+	JOIN pg_catalog.pg_class AS t2
+	ON t1.table_name = t2.relname
+WHERE
+	t1.TABLE_TYPE = 'BASE TABLE'
+	AND t1.TABLE_NAME = $1
+	AND t1.TABLE_SCHEMA = $2
+`
+	// Query to list table columns.
+	columnsQuery = `
+SELECT
+	t1.column_name,
+	t1.data_type,
+	t1.is_nullable,
+	t1.column_default,
+	t1.character_maximum_length,
+	t1.numeric_precision,
+	t1.numeric_scale,
+	t1.character_set_name,
+	t1.collation_name,
+	t1.udt_name,
+	t1.is_identity,
+	t1.identity_generation,
+	col_description(to_regclass("table_schema" || '.' || "table_name")::oid, "ordinal_position") AS comment,
+	t2.typtype,
+	t2.oid
+FROM
+	"information_schema"."columns" AS t1
+	LEFT JOIN pg_catalog.pg_type AS t2
+	ON t1.udt_name = t2.typname
+WHERE
+	TABLE_SCHEMA = $1 AND TABLE_NAME = $2
+`
+
+	// Query to list table indexes.
+	indexesQuery = `
+SELECT
+	i.relname AS index_name,
+	am.amname AS index_type,
+	a.attname AS column_name,
+	idx.indisprimary AS primary,
+	idx.indisunique AS unique,
+	c.contype AS constraint_type,
+	pg_get_expr(idx.indpred, idx.indrelid) AS predicate,
+	pg_get_expr(idx.indexprs, idx.indrelid) AS expression,
+	pg_index_column_has_property(idx.indexrelid, a.attnum, 'asc') AS asc,
+	pg_index_column_has_property(idx.indexrelid, a.attnum, 'desc') AS desc,
+	pg_index_column_has_property(idx.indexrelid, a.attnum, 'nulls_first') AS nulls_first,
+	pg_index_column_has_property(idx.indexrelid, a.attnum, 'nulls_last') AS nulls_last,
+	obj_description(to_regclass($1 || i.relname)::oid) AS comment
+FROM
+	pg_index idx
+	JOIN pg_class i
+	ON i.oid = idx.indexrelid
+	LEFT JOIN pg_constraint c
+	ON idx.indexrelid = c.conindid
+	LEFT JOIN pg_attribute a
+	ON a.attrelid = idx.indexrelid
+	JOIN pg_am am
+	ON am.oid = i.relam
+WHERE
+	idx.indrelid = to_regclass($1 || '.' || $2)::oid
+	AND COALESCE(c.contype, '') <> 'f'
+ORDER BY
+	index_name, a.attnum
+`
+	fksQuery = `
+SELECT
+    t1.constraint_name,
+    t1.table_name,
+    t2.column_name,
+    t1.table_schema,
+    t3.table_name AS referenced_table_name,
+    t3.column_name AS referenced_column_name,
+    t3.table_schema AS referenced_schema_name,
+    t4.update_rule,
+    t4.delete_rule
+FROM
+    information_schema.table_constraints t1
+    JOIN information_schema.key_column_usage t2
+    ON t1.constraint_name = t2.constraint_name
+    AND t1.table_schema = t2.constraint_schema
+    JOIN information_schema.constraint_column_usage t3
+    ON t1.constraint_name = t3.constraint_name
+    AND t1.table_schema = t3.constraint_schema
+    JOIN information_schema.referential_constraints t4
+    ON t1.constraint_name = t4.constraint_name
+    AND t1.table_schema = t4.constraint_schema
+WHERE
+    t1.constraint_type = 'FOREIGN KEY'
+    AND t1.table_schema = $1
+    AND t1.table_name = $2
+ORDER BY
+    t1.constraint_name,
+    t2.ordinal_position
+`
+
+	// Query to list table check constraints.
+	checksQuery = `
+SELECT
+	t1.conname AS constraint_name,
+	pg_get_expr(t1.conbin, to_regclass($1 || '.' || $2)::oid) as expression,
+	t2.attname as column_name,
+	t1.conkey as column_indexes,
+	t1.connoinherit as no_inherit
+FROM
+	pg_catalog.pg_constraint t1
+	JOIN pg_attribute t2
+	ON t2.attrelid = t1.conrelid
+	AND t2.attnum = ANY (t1.conkey)
+WHERE
+	t1.contype = 'c'
+	AND t1.conrelid = to_regclass($1 || '.' || $2)::oid
+ORDER BY
+	t1.conname, array_position(t1.conkey, t2.attnum)
+`
+)
diff --git a/sql/postgres/driver_test.go b/sql/postgres/inspect_test.go
similarity index 100%
rename from sql/postgres/driver_test.go
rename to sql/postgres/inspect_test.go
diff --git a/sql/postgres/migrate.go b/sql/postgres/migrate.go
index 8ade837d..e8cd3557 100644
--- a/sql/postgres/migrate.go
+++ b/sql/postgres/migrate.go
@@ -14,12 +14,7 @@ import (
 )
 
 // A migrate provides migration capabilities for schema elements.
-type migrate struct{ *Driver }
-
-// Migrate returns a PostgreSQL schema executor.
-func (d *Driver) Migrate() schema.Execer {
-	return &migrate{Driver: d}
-}
+type migrate 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.
@@ -283,7 +278,7 @@ func (m *migrate) alterType(ctx context.Context, from, to *schema.EnumType) erro
 }
 
 func (m *migrate) enumExists(ctx context.Context, name string) (bool, error) {
-	rows, err := m.Driver.QueryContext(ctx, "SELECT * FROM pg_type WHERE typname = $1 AND typtype = 'e'", name)
+	rows, err := m.QueryContext(ctx, "SELECT * FROM pg_type WHERE typname = $1 AND typtype = 'e'", name)
 	if err != nil {
 		return false, fmt.Errorf("check index existance: %w", err)
 	}
diff --git a/sql/schema/inspect.go b/sql/schema/inspect.go
index 0829aa4e..f3913c67 100644
--- a/sql/schema/inspect.go
+++ b/sql/schema/inspect.go
@@ -43,37 +43,30 @@ type (
 		Tables []string
 	}
 
-	// Inspector is the interface implemented by the different database
-	// drivers for inspecting multiple tables.
-	Inspector interface {
-		// InspectSchema returns the schema description by its name. A NotExistError
-		// error is returned if the schema does not exists in the database.
-		InspectSchema(ctx context.Context, name string, opts *InspectOptions) (*Schema, error)
-	}
-
 	// InspectTableOptions describes options for TableInspector.
 	InspectTableOptions struct {
 		// Schema defines an optional schema to inspect.
 		Schema string
 	}
 
-	// TableInspector is the interface implemented by the different database
-	// drivers for inspecting their schema tables.
-	TableInspector interface {
-		// InspectTable returns the table description by its name. A NotExistError
-		// error is returned if the table does not exists in the database.
-		InspectTable(ctx context.Context, name string, opts *InspectTableOptions) (*Table, error)
-	}
-
 	// InspectRealmOption describes options for RealmInspector.
 	InspectRealmOption struct {
 		// Schemas to inspect. Empty means all tables in the schema.
 		Schemas []string
 	}
 
-	// RealmInspector is the interface implemented by the different database
-	// drivers for inspecting multiple schemas (realm).
-	RealmInspector interface {
+	// Inspector is the interface implemented by the different database
+	// drivers for inspecting multiple tables.
+	Inspector interface {
+		// InspectSchema returns the schema description by its name. A NotExistError
+		// error is returned if the schema does not exists in the database.
+		InspectSchema(ctx context.Context, name string, opts *InspectOptions) (*Schema, error)
+
+		// InspectTable returns the table description by its name. A NotExistError
+		// error is returned if the table does not exists in the database.
+		InspectTable(ctx context.Context, name string, opts *InspectTableOptions) (*Table, error)
+
+		// InspectRealm returns the description of the connected database.
 		InspectRealm(ctx context.Context, opts *InspectRealmOption) (*Realm, error)
 	}
 )
diff --git a/sql/sqlite/diff.go b/sql/sqlite/diff.go
index bf817fca..34455c17 100644
--- a/sql/sqlite/diff.go
+++ b/sql/sqlite/diff.go
@@ -13,16 +13,7 @@ import (
 )
 
 // A diff provides a SQLite implementation for sqlx.DiffDriver.
-type diff struct {
-	version string
-}
-
-// Diff returns a SQLite schema differ.
-func (d Driver) Diff() schema.Differ {
-	return &sqlx.Diff{
-		DiffDriver: &diff{version: d.version},
-	}
-}
+type diff struct{ conn }
 
 // SchemaAttrDiff returns a changeset for migrating schema attributes from one state to the other.
 func (d *diff) SchemaAttrDiff(_, _ *schema.Schema) []schema.Change {
@@ -43,7 +34,7 @@ func (d *diff) TableAttrDiff(from, to *schema.Table) []schema.Change {
 			A: &WithoutRowID{},
 		})
 	}
-	// TODO: support diffing constraints after it's supported.
+	// TODO: support diffing constraints after it's added on inspection.
 	return changes
 }
 
diff --git a/sql/sqlite/driver.go b/sql/sqlite/driver.go
index cd5a2eaa..a7680c5a 100644
--- a/sql/sqlite/driver.go
+++ b/sql/sqlite/driver.go
@@ -5,655 +5,57 @@
 package sqlite
 
 import (
-	"context"
-	"database/sql"
 	"fmt"
-	"regexp"
-	"strconv"
-	"strings"
-	"unicode"
 
 	"ariga.io/atlas/sql/internal/sqlx"
 
 	"ariga.io/atlas/sql/schema"
 )
 
-// Driver represents an SQLite driver for introspecting database schemas
-// and apply migrations changes on them.
-type Driver struct {
-	schema.ExecQuerier
-	// System variables that are set on `Open`.
-	fkEnabled  bool
-	version    string
-	collations []string
-}
+type (
+	// Driver represents a SQLite driver for introspecting database schemas,
+	// generating diff between schema elements and apply migrations changes.
+	Driver struct {
+		conn
+		schema.Differ
+		schema.Execer
+		schema.Inspector
+	}
+
+	// database connection and its information.
+	conn struct {
+		schema.ExecQuerier
+		// System variables that are set on `Open`.
+		fkEnabled  bool
+		version    string
+		collations []string
+	}
+)
 
 // Open opens a new SQLite driver.
 func Open(db schema.ExecQuerier) (*Driver, error) {
-	drv := &Driver{ExecQuerier: db}
-	if err := db.QueryRow("SELECT sqlite_version()").Scan(&drv.version); err != nil {
+	c := conn{ExecQuerier: db}
+	if err := db.QueryRow("SELECT sqlite_version()").Scan(&c.version); err != nil {
 		return nil, fmt.Errorf("sqlite: scanning database version: %w", err)
 	}
-	if err := db.QueryRow("PRAGMA foreign_keys").Scan(&drv.fkEnabled); err != nil {
+	if err := db.QueryRow("PRAGMA foreign_keys").Scan(&c.fkEnabled); err != nil {
 		return nil, fmt.Errorf("sqlite: check foreign_keys pragma: %w", err)
 	}
 	rows, err := db.Query("SELECT name FROM pragma_collation_list()")
 	if err != nil {
 		return nil, fmt.Errorf("sqlite: check collation_list pragma: %w", err)
 	}
-	if drv.collations, err = sqlx.ScanStrings(rows); err != nil {
+	if c.collations, err = sqlx.ScanStrings(rows); err != nil {
 		return nil, fmt.Errorf("sqlite: scanning database collations: %w", err)
 	}
-	return drv, nil
-}
-
-// InspectRealm returns schema descriptions of all resources in the given realm.
-func (d *Driver) InspectRealm(ctx context.Context, opts *schema.InspectRealmOption) (*schema.Realm, error) {
-	schemas, err := d.databases(ctx, opts)
-	if err != nil {
-		return nil, err
-	}
-	if len(schemas) > 1 {
-		return nil, fmt.Errorf("sqlite: multiple database files are not supported by the driver. got: %d", len(schemas))
-	}
-	realm := &schema.Realm{Schemas: schemas}
-	for _, s := range schemas {
-		tables, err := d.tables(ctx, nil)
-		if err != nil {
-			return nil, err
-		}
-		for _, t := range tables {
-			t.Schema = s
-			t, err := d.inspectTable(ctx, t)
-			if err != nil {
-				return nil, err
-			}
-			s.Tables = append(s.Tables, t)
-		}
-		s.Realm = realm
-	}
-	sqlx.LinkSchemaTables(realm.Schemas)
-	return realm, nil
-}
-
-// InspectSchema returns schema descriptions of all tables in the given schema.
-func (d *Driver) InspectSchema(ctx context.Context, name string, opts *schema.InspectOptions) (*schema.Schema, error) {
-	schemas, err := d.databases(ctx, &schema.InspectRealmOption{
-		Schemas: []string{name},
-	})
-	if err != nil {
-		return nil, err
-	}
-	if len(schemas) == 0 {
-		return nil, &schema.NotExistError{
-			Err: fmt.Errorf("sqlite: schema %q was not found", name),
-		}
-	}
-	tables, err := d.tables(ctx, opts)
-	if err != nil {
-		return nil, err
-	}
-	s := schemas[0]
-	for _, t := range tables {
-		t.Schema = s
-		t, err := d.inspectTable(ctx, t)
-		if err != nil {
-			return nil, err
-		}
-		s.Tables = append(s.Tables, t)
-	}
-	sqlx.LinkSchemaTables(schemas)
-	s.Realm = &schema.Realm{Schemas: schemas}
-	return s, nil
-}
-
-// InspectTable returns the schema description of the given table.
-func (d *Driver) InspectTable(ctx context.Context, name string, opts *schema.InspectTableOptions) (*schema.Table, error) {
-	if opts != nil && opts.Schema != "main" {
-		return nil, fmt.Errorf("sqlite: querying attached database is not supported. got: %q", opts.Schema)
-	}
-	s, err := d.InspectSchema(ctx, "main", &schema.InspectOptions{
-		Tables: []string{name},
-	})
-	if err != nil {
-		return nil, err
-	}
-	if len(s.Tables) == 0 {
-		return nil, &schema.NotExistError{
-			Err: fmt.Errorf("sqlite: table %q was not found", name),
-		}
-	}
-	return s.Tables[0], nil
-}
-
-func (d *Driver) inspectTable(ctx context.Context, t *schema.Table) (*schema.Table, error) {
-	if err := d.columns(ctx, t); err != nil {
-		return nil, err
-	}
-	if err := d.indexes(ctx, t); err != nil {
-		return nil, err
-	}
-	if err := d.fks(ctx, t); err != nil {
-		return nil, err
-	}
-	// TODO(a8m): extract CHECK constraints from 'CREATE TABLE' statement.
-	return t, nil
-}
-
-// columns queries and appends the columns of the given table.
-func (d *Driver) columns(ctx context.Context, t *schema.Table) error {
-	rows, err := d.QueryContext(ctx, fmt.Sprintf(columnsQuery, t.Name))
-	if err != nil {
-		return fmt.Errorf("sqlite: querying %q columns: %w", t.Name, err)
-	}
-	defer rows.Close()
-	for rows.Next() {
-		if err := d.addColumn(t, rows); err != nil {
-			return fmt.Errorf("sqlite: %w", err)
-		}
-	}
-	autoinc(t)
-	return rows.Err()
-}
-
-// addColumn scans the current row and adds a new column from it to the table.
-func (d *Driver) addColumn(t *schema.Table, rows *sql.Rows) error {
-	var (
-		nullable, primary   bool
-		name, typ, defaults sql.NullString
-		err                 error
-	)
-	if err = rows.Scan(&name, &typ, &nullable, &defaults, &primary); err != nil {
-		return err
-	}
-	c := &schema.Column{
-		Name: name.String,
-		Type: &schema.ColumnType{
-			Raw:  typ.String,
-			Null: nullable,
-		},
-	}
-	c.Type.Type, err = parseRawType(typ.String)
-	if err != nil {
-		return err
-	}
-	if sqlx.ValidString(defaults) {
-		c.Default = &schema.RawExpr{
-			X: defaults.String,
-		}
-	}
-	// TODO(a8m): extract collation from 'CREATE TABLE' statement.
-	t.Columns = append(t.Columns, c)
-	if primary {
-		if t.PrimaryKey == nil {
-			t.PrimaryKey = &schema.Index{
-				Name:   "PRIMARY",
-				Unique: true,
-				Table:  t,
-			}
-		}
-		// Columns are ordered by the `pk` field.
-		t.PrimaryKey.Parts = append(t.PrimaryKey.Parts, &schema.IndexPart{
-			C:     c,
-			SeqNo: len(t.PrimaryKey.Parts) + 1,
-		})
-	}
-	return nil
-}
-
-func parseRawType(c string) (schema.Type, error) {
-	// A datatype may be zero or more names.
-	// https://www.sqlite.org/datatypes.html
-	if c == "" {
-		return &schema.UnsupportedType{}, nil
-	}
-	parts := columnParts(c)
-	switch t := parts[0]; t {
-	case "bool", "boolean":
-		return &schema.BoolType{T: t}, nil
-	case "blob":
-		return &schema.BinaryType{T: t}, nil
-	case "int2", "int8", "int", "integer", "tinyint", "smallint", "mediumint", "bigint", "unsigned big int":
-		// All integer types have the same "type affinity".
-		return &schema.IntegerType{T: t}, nil
-	case "real", "double", "double precision", "float":
-		return &schema.FloatType{T: t}, nil
-	case "numeric", "decimal":
-		ct := &schema.DecimalType{T: t}
-		if len(parts) > 1 {
-			p, err := strconv.ParseInt(parts[1], 10, 64)
-			if err != nil {
-				return nil, fmt.Errorf("parse precision %q", parts[1])
-			}
-			ct.Precision = int(p)
-		}
-		if len(parts) > 2 {
-			s, err := strconv.ParseInt(parts[2], 10, 64)
-			if err != nil {
-				return nil, fmt.Errorf("parse scale %q", parts[1])
-			}
-			ct.Scale = int(s)
-		}
-		return ct, nil
-	case "char", "character", "varchar", "varying character", "nchar", "native character", "nvarchar", "text", "clob":
-		ct := &schema.StringType{T: t}
-		if len(parts) > 1 {
-			p, err := strconv.ParseInt(parts[1], 10, 64)
-			if err != nil {
-				return nil, fmt.Errorf("parse size %q", parts[1])
-			}
-			ct.Size = int(p)
-		}
-		return ct, nil
-	case "json":
-		return &schema.JSONType{T: t}, nil
-	case "date", "datetime", "time", "timestamp":
-		return &schema.TimeType{T: t}, nil
-	case "uuid":
-		return &UUIDType{T: t}, nil
-	default:
-		return nil, fmt.Errorf("unknown column type %q", t)
-	}
-}
-
-// indexes queries and appends the indexes of the given table.
-func (d *Driver) indexes(ctx context.Context, t *schema.Table) error {
-	rows, err := d.QueryContext(ctx, fmt.Sprintf(indexesQuery, t.Name))
-	if err != nil {
-		return fmt.Errorf("sqlite: querying %q indexes: %w", t.Name, err)
-	}
-	if err := d.addIndexes(t, rows); err != nil {
-		return fmt.Errorf("sqlite: scan %q indexes: %w", t.Name, err)
-	}
-	for _, idx := range t.Indexes {
-		if err := d.indexColumns(ctx, t, idx); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-// addIndexes scans the rows and adds the indexes to the table.
-func (d *Driver) addIndexes(t *schema.Table, rows *sql.Rows) error {
-	defer rows.Close()
-	for rows.Next() {
-		var (
-			uniq, partial      bool
-			name, origin, stmt sql.NullString
-		)
-		if err := rows.Scan(&name, &uniq, &origin, &partial, &stmt); err != nil {
-			return err
-		}
-		if origin.String == "pk" {
-			continue
-		}
-		idx := &schema.Index{
-			Name:   name.String,
-			Unique: uniq,
-			Table:  t,
-			Attrs: []schema.Attr{
-				&CreateStmt{S: stmt.String},
-				&IndexOrigin{O: origin.String},
-			},
-		}
-		if partial {
-			i := strings.Index(stmt.String, "WHERE")
-			if i == -1 {
-				return fmt.Errorf("missing partial WHERE clause in: %s", stmt.String)
-			}
-			idx.Attrs = append(idx.Attrs, &IndexPredicate{
-				P: strings.TrimSpace(stmt.String[i+5:]),
-			})
-		}
-		t.Indexes = append(t.Indexes, idx)
-	}
-	return nil
-}
-
-func (d *Driver) indexColumns(ctx context.Context, t *schema.Table, idx *schema.Index) error {
-	rows, err := d.QueryContext(ctx, fmt.Sprintf(indexColumnsQuery, idx.Name))
-	if err != nil {
-		return fmt.Errorf("sqlite: querying %q indexes: %w", t.Name, err)
-	}
-	defer rows.Close()
-	for rows.Next() {
-		var name sql.NullString
-		if err := rows.Scan(&name); err != nil {
-			return fmt.Errorf("sqlite: scanning index names: %w", err)
-		}
-		switch c, ok := t.Column(name.String); {
-		case ok:
-			idx.Parts = append(idx.Parts, &schema.IndexPart{
-				SeqNo: len(idx.Parts) + 1,
-				C:     c,
-			})
-		// NULL name indicates that the index-part is an expression and we
-		// should extract it from the `CREATE INDEX` statement (not supported atm).
-		case !sqlx.ValidString(name):
-			idx.Parts = append(idx.Parts, &schema.IndexPart{
-				SeqNo: len(idx.Parts) + 1,
-				X:     &schema.RawExpr{X: "<unsupported>"},
-			})
-		default:
-			return fmt.Errorf("sqlite: column %q was not found for index %q", name.String, idx.Name)
-		}
-	}
-	return nil
-}
-
-// fks queries and appends the foreign-keys of the given table.
-func (d *Driver) fks(ctx context.Context, t *schema.Table) error {
-	rows, err := d.QueryContext(ctx, fmt.Sprintf(fksQuery, t.Name))
-	if err != nil {
-		return fmt.Errorf("sqlite: querying %q foreign-keys: %w", t.Name, err)
-	}
-	if err := d.addFKs(t, rows); err != nil {
-		return fmt.Errorf("sqlite: scan %q foreign-keys: %w", t.Name, err)
-	}
-	return fillConstName(t)
-}
-
-func (d *Driver) addFKs(t *schema.Table, rows *sql.Rows) error {
-	ids := make(map[int]*schema.ForeignKey)
-	for rows.Next() {
-		var (
-			id                                                  int
-			column, refColumn, refTable, updateRule, deleteRule string
-		)
-		if err := rows.Scan(&id, &column, &refColumn, &refTable, &updateRule, &deleteRule); err != nil {
-			return err
-		}
-		fk, ok := ids[id]
-		if !ok {
-			fk = &schema.ForeignKey{
-				Symbol:   strconv.Itoa(id),
-				Table:    t,
-				RefTable: t,
-				OnDelete: schema.ReferenceOption(deleteRule),
-				OnUpdate: schema.ReferenceOption(updateRule),
-			}
-			if refTable != t.Name {
-				fk.RefTable = &schema.Table{Name: refTable, Schema: &schema.Schema{Name: t.Schema.Name}}
-			}
-			ids[id] = fk
-			t.ForeignKeys = append(t.ForeignKeys, fk)
-		}
-		c, ok := t.Column(column)
-		if !ok {
-			return fmt.Errorf("column %q was not found for fk %q", column, fk.Symbol)
-		}
-		// Rows are ordered by SEQ that specifies the
-		// position of the column in the FK definition.
-		if _, ok := fk.Column(c.Name); !ok {
-			fk.Columns = append(fk.Columns, c)
-			c.ForeignKeys = append(c.ForeignKeys, fk)
-		}
-
-		// Stub referenced columns or link if it is a self-reference.
-		var rc *schema.Column
-		if fk.Table != fk.RefTable {
-			rc = &schema.Column{Name: refColumn}
-		} else if c, ok := t.Column(refColumn); ok {
-			rc = c
-		} else {
-			return fmt.Errorf("referenced column %q was not found for fk %q", refColumn, fk.Symbol)
-		}
-		if _, ok := fk.RefColumn(rc.Name); !ok {
-			fk.RefColumns = append(fk.RefColumns, rc)
-		}
-	}
-	// TODO(a8m): extract the foreign-key name from the `CREATE TABLE` statement.
-	return nil
-}
-
-// tableNames returns a list of all tables exist in the schema.
-func (d *Driver) tables(ctx context.Context, opts *schema.InspectOptions) ([]*schema.Table, error) {
-	var (
-		args  []interface{}
-		query = tablesQuery
-	)
-	if opts != nil && len(opts.Tables) > 0 {
-		query += " AND name IN (" + strings.Repeat("?, ", len(opts.Tables)-1) + "?)"
-		for _, s := range opts.Tables {
-			args = append(args, s)
-		}
-	}
-	rows, err := d.QueryContext(ctx, query, args...)
-	if err != nil {
-		return nil, fmt.Errorf("sqlite: querying schema tables: %w", err)
-	}
-	defer rows.Close()
-	var tables []*schema.Table
-	for rows.Next() {
-		var name, stmt string
-		if err := rows.Scan(&name, &stmt); err != nil {
-			return nil, fmt.Errorf("sqlite: scanning table: %w", err)
-		}
-		stmt = strings.TrimSpace(stmt)
-		t := &schema.Table{
-			Name: name,
-			Attrs: []schema.Attr{
-				&CreateStmt{S: strings.TrimSpace(stmt)},
-			},
-		}
-		if strings.HasSuffix(stmt, "WITHOUT ROWID") || strings.HasSuffix(stmt, "without rowid") {
-			t.Attrs = append(t.Attrs, &WithoutRowID{})
-		}
-		tables = append(tables, t)
-	}
-	return tables, nil
-}
-
-// schemas returns the list of the schemas in the database.
-func (d *Driver) databases(ctx context.Context, opts *schema.InspectRealmOption) ([]*schema.Schema, error) {
-	var (
-		args  []interface{}
-		query = databasesQuery
-	)
-	if opts != nil && len(opts.Schemas) > 0 {
-		query += " WHERE name IN (" + strings.Repeat("?, ", len(opts.Schemas)-1) + "?)"
-		for _, s := range opts.Schemas {
-			args = append(args, s)
-		}
-	}
-	rows, err := d.QueryContext(ctx, query, args...)
-	if err != nil {
-		return nil, fmt.Errorf("sqlite: querying schemas: %w", err)
-	}
-	defer rows.Close()
-	var schemas []*schema.Schema
-	for rows.Next() {
-		var name, file sql.NullString
-		if err := rows.Scan(&name, &file); err != nil {
-			return nil, err
-		}
-		// File is missing if the database is not
-		// associated with a file (:memory: mode).
-		if file.String == "" {
-			file.String = ":memory:"
-		}
-		schemas = append(schemas, &schema.Schema{
-			Name:  name.String,
-			Attrs: []schema.Attr{&File{Name: file.String}},
-		})
-	}
-	return schemas, nil
+	return &Driver{
+		conn:      c,
+		Differ:    &sqlx.Diff{DiffDriver: &diff{c}},
+		Execer:    &migrate{c},
+		Inspector: &inspect{c},
+	}, nil
 }
 
-type (
-	// File describes a database file.
-	File struct {
-		schema.Attr
-		Name string
-	}
-
-	// CreateStmt describes the SQL statement used to create a resource.
-	CreateStmt struct {
-		schema.Attr
-		S string
-	}
-
-	// AutoIncrement describes the `AUTOINCREMENT` configuration.
-	// https://www.sqlite.org/autoinc.html
-	AutoIncrement struct {
-		schema.Attr
-	}
-
-	// WithoutRowID describes the `WITHOUT ROWID` configuration.
-	// See: https://sqlite.org/withoutrowid.html
-	WithoutRowID struct {
-		schema.Attr
-	}
-
-	// IndexPredicate describes a partial index predicate.
-	// See: https://www.sqlite.org/partialindex.html
-	IndexPredicate struct {
-		schema.Attr
-		P string
-	}
-
-	// IndexOrigin describes how the index was created.
-	// See: https://www.sqlite.org/pragma.html#pragma_index_list
-	IndexOrigin struct {
-		schema.Attr
-		O string
-	}
-
-	// A UUIDType defines a UUID type.
-	UUIDType struct {
-		schema.Type
-		T string
-	}
-)
-
-func columnParts(t string) []string {
-	t = strings.TrimSpace(strings.ToLower(t))
-	parts := strings.FieldsFunc(t, func(r rune) bool {
-		return r == '(' || r == ')' || r == ' ' || r == ','
-	})
-	for k := 0; k < 2; k++ {
-		// Join the type back if it was separated with space (e.g. 'varying character').
-		if len(parts) > 1 && !isNumber(parts[0]) && !isNumber(parts[1]) {
-			parts[1] = parts[0] + " " + parts[1]
-			parts = parts[1:]
-		}
-	}
-	return parts
-}
-
-// isNumber reports whether the string is a number (category N).
-func isNumber(s string) bool {
-	for _, r := range s {
-		if !unicode.IsNumber(r) {
-			return false
-		}
-	}
-	return true
-}
-
-var reAutoinc = regexp.MustCompile(`(?i)PRIMARY\s+KEY\s+AUTOINCREMENT`)
-
-// autoinc checks if the table contains a "PRIMARY KEY AUTOINCREMENT" on its
-// CREATE statement, according to https://www.sqlite.org/syntax/column-constraint.html.
-// This is a workaround until we will embed a proper SQLite parser in atlas.
-func autoinc(t *schema.Table) {
-	if t.PrimaryKey == nil || len(t.PrimaryKey.Parts) != 1 || t.PrimaryKey.Parts[0].C == nil {
-		return
-	}
-	if c := (CreateStmt{}); !sqlx.Has(t.Attrs, &c) || !reAutoinc.MatchString(c.S) {
-		return
-	}
-	// Annotate table elements with "AUTOINCREMENT".
-	t.PrimaryKey.Attrs = append(t.PrimaryKey.Attrs, AutoIncrement{})
-	t.PrimaryKey.Parts[0].C.Attrs = append(t.PrimaryKey.Parts[0].C.Attrs, AutoIncrement{})
-}
-
-// The following regexes extract named foreign-key constraints defined in the table-constraints or inlined
-// as column-constraints. Note, we assume the SQL statements are valid as they are returned by SQLite.
-var (
-	reConstC = regexp.MustCompile("(?i)(?:[(,]\\s*)[\"`]*(\\w+)[\"`]*[^,]*\\s+CONSTRAINT\\s+[\"`]*(\\w+)[\"`]*\\s+REFERENCES\\s+[\"`]*(\\w+)[\"`]*\\s*\\(([,\"` \\w]+)\\)")
-	reConstT = regexp.MustCompile("(?i)CONSTRAINT\\s+[\"`]*(\\w+)[\"`]*\\s+FOREIGN\\s+KEY\\s*\\(([,\"` \\w]+)\\)\\s+REFERENCES\\s+[\"`]*(\\w+)[\"`]*\\s*\\(([,\"` \\w]+)\\)")
-)
-
-// fillConstName fills foreign-key constrain names from CREATE TABLE statement.
-func fillConstName(t *schema.Table) error {
-	var c CreateStmt
-	if !sqlx.Has(t.Attrs, &c) {
-		return fmt.Errorf("missing CREATE statment for table: %q", t.Name)
-	}
-	// Loop over table constraints.
-	for _, m := range reConstT.FindAllStringSubmatch(c.S, -1) {
-		if len(m) != 5 {
-			return fmt.Errorf("unexpected number of matches for a table constraint: %q", m)
-		}
-		// Pattern matches "constraint_name", "columns", "ref_table" and "ref_columns".
-		for _, fk := range t.ForeignKeys {
-			// Found a foreign-key match for the constraint.
-			if matchFK(fk, columns(m[2]), m[3], columns(m[4])) {
-				fk.Symbol = m[1]
-				break
-			}
-		}
-	}
-	// Loop over inlined column constraints.
-	for _, m := range reConstC.FindAllStringSubmatch(c.S, -1) {
-		if len(m) != 5 {
-			return fmt.Errorf("unexpected number of matches for a column constraint: %q", m)
-		}
-		// Pattern matches "column", "constraint_name", "ref_table" and "ref_columns".
-		for _, fk := range t.ForeignKeys {
-			// Found a foreign-key match for the constraint.
-			if matchFK(fk, columns(m[1]), m[3], columns(m[4])) {
-				fk.Symbol = m[2]
-				break
-			}
-		}
-	}
-	return nil
-}
-
-// columns from the matched regex above.
-func columns(s string) []string {
-	names := strings.Split(s, ",")
-	for i := range names {
-		names[i] = strings.Trim(strings.TrimSpace(names[i]), "`\"")
-	}
-	return names
-}
-
-// matchFK reports if the foreign-key matches the given attributes.
-func matchFK(fk *schema.ForeignKey, columns []string, refTable string, refColumns []string) bool {
-	if len(fk.Columns) != len(columns) || fk.RefTable.Name != refTable || len(fk.RefColumns) != len(refColumns) {
-		return false
-	}
-	for i := range columns {
-		if fk.Columns[i].Name != columns[i] {
-			return false
-		}
-	}
-	for i := range refColumns {
-		if fk.RefColumns[i].Name != refColumns[i] {
-			return false
-		}
-	}
-	return true
-}
-
-const (
-	// Query to list attached database files.
-	databasesQuery = "SELECT `name`, `file` FROM pragma_database_list()"
-	// Query to list database tables.
-	tablesQuery = "SELECT `name`, `sql` FROM sqlite_master WHERE `type`='table' AND `name` NOT LIKE 'sqlite_%'"
-	// Query to list table information.
-	columnsQuery = "SELECT `name`, `type`, (not `notnull`) AS `nullable`, `dflt_value`, (`pk` <> 0) AS `pk`  FROM pragma_table_info('%s') ORDER BY `pk`, `cid`"
-	// Query to list table indexes.
-	indexesQuery = "SELECT `il`.`name`, `il`.`unique`, `il`.`origin`, `il`.`partial`, `m`.`sql` FROM pragma_index_list('%s') AS il JOIN sqlite_master AS m ON il.name = m.name"
-	// Query to list index columns.
-	indexColumnsQuery = "SELECT name FROM pragma_index_info('%s') ORDER BY seqno"
-	// Query to list table foreign-keys.
-	fksQuery = "SELECT `id`, `from`, `to`, `table`, `on_update`, `on_delete` FROM pragma_foreign_key_list('%s') ORDER BY id, seq"
-)
-
 // SQLite standard data types as defined in its codebase and documentation.
 // https://www.sqlite.org/datatype3.html
 // https://github.com/sqlite/sqlite/blob/master/src/global.c
diff --git a/sql/sqlite/inspect.go b/sql/sqlite/inspect.go
new file mode 100644
index 00000000..a8fd4bff
--- /dev/null
+++ b/sql/sqlite/inspect.go
@@ -0,0 +1,630 @@
+// 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 sqlite
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"regexp"
+	"strconv"
+	"strings"
+	"unicode"
+
+	"ariga.io/atlas/sql/internal/sqlx"
+	"ariga.io/atlas/sql/schema"
+)
+
+// A diff provides a PostgreSQL implementation for schema.Inspector.
+type inspect struct{ conn }
+
+var _ schema.Inspector = (*inspect)(nil)
+
+// InspectRealm returns schema descriptions of all resources in the given realm.
+func (i *inspect) InspectRealm(ctx context.Context, opts *schema.InspectRealmOption) (*schema.Realm, error) {
+	schemas, err := i.databases(ctx, opts)
+	if err != nil {
+		return nil, err
+	}
+	if len(schemas) > 1 {
+		return nil, fmt.Errorf("sqlite: multiple database files are not supported by the driver. got: %d", len(schemas))
+	}
+	realm := &schema.Realm{Schemas: schemas}
+	for _, s := range schemas {
+		tables, err := i.tables(ctx, nil)
+		if err != nil {
+			return nil, err
+		}
+		for _, t := range tables {
+			t.Schema = s
+			t, err := i.inspectTable(ctx, t)
+			if err != nil {
+				return nil, err
+			}
+			s.Tables = append(s.Tables, t)
+		}
+		s.Realm = realm
+	}
+	sqlx.LinkSchemaTables(realm.Schemas)
+	return realm, nil
+}
+
+// InspectSchema returns schema descriptions of all tables in the given schema.
+func (i *inspect) InspectSchema(ctx context.Context, name string, opts *schema.InspectOptions) (*schema.Schema, error) {
+	schemas, err := i.databases(ctx, &schema.InspectRealmOption{
+		Schemas: []string{name},
+	})
+	if err != nil {
+		return nil, err
+	}
+	if len(schemas) == 0 {
+		return nil, &schema.NotExistError{
+			Err: fmt.Errorf("sqlite: schema %q was not found", name),
+		}
+	}
+	tables, err := i.tables(ctx, opts)
+	if err != nil {
+		return nil, err
+	}
+	s := schemas[0]
+	for _, t := range tables {
+		t.Schema = s
+		t, err := i.inspectTable(ctx, t)
+		if err != nil {
+			return nil, err
+		}
+		s.Tables = append(s.Tables, t)
+	}
+	sqlx.LinkSchemaTables(schemas)
+	s.Realm = &schema.Realm{Schemas: schemas}
+	return s, nil
+}
+
+// InspectTable returns the schema description of the given table.
+func (i *inspect) InspectTable(ctx context.Context, name string, opts *schema.InspectTableOptions) (*schema.Table, error) {
+	if opts != nil && opts.Schema != "main" {
+		return nil, fmt.Errorf("sqlite: querying attached database is not supported. got: %q", opts.Schema)
+	}
+	s, err := i.InspectSchema(ctx, "main", &schema.InspectOptions{
+		Tables: []string{name},
+	})
+	if err != nil {
+		return nil, err
+	}
+	if len(s.Tables) == 0 {
+		return nil, &schema.NotExistError{
+			Err: fmt.Errorf("sqlite: table %q was not found", name),
+		}
+	}
+	return s.Tables[0], nil
+}
+
+func (i *inspect) inspectTable(ctx context.Context, t *schema.Table) (*schema.Table, error) {
+	if err := i.columns(ctx, t); err != nil {
+		return nil, err
+	}
+	if err := i.indexes(ctx, t); err != nil {
+		return nil, err
+	}
+	if err := i.fks(ctx, t); err != nil {
+		return nil, err
+	}
+	// TODO(a8m): extract CHECK constraints from 'CREATE TABLE' statement.
+	return t, nil
+}
+
+// columns queries and appends the columns of the given table.
+func (i *inspect) columns(ctx context.Context, t *schema.Table) error {
+	rows, err := i.QueryContext(ctx, fmt.Sprintf(columnsQuery, t.Name))
+	if err != nil {
+		return fmt.Errorf("sqlite: querying %q columns: %w", t.Name, err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		if err := i.addColumn(t, rows); err != nil {
+			return fmt.Errorf("sqlite: %w", err)
+		}
+	}
+	autoinc(t)
+	return rows.Err()
+}
+
+// addColumn scans the current row and adds a new column from it to the table.
+func (i *inspect) addColumn(t *schema.Table, rows *sql.Rows) error {
+	var (
+		nullable, primary   bool
+		name, typ, defaults sql.NullString
+		err                 error
+	)
+	if err = rows.Scan(&name, &typ, &nullable, &defaults, &primary); err != nil {
+		return err
+	}
+	c := &schema.Column{
+		Name: name.String,
+		Type: &schema.ColumnType{
+			Raw:  typ.String,
+			Null: nullable,
+		},
+	}
+	c.Type.Type, err = parseRawType(typ.String)
+	if err != nil {
+		return err
+	}
+	if sqlx.ValidString(defaults) {
+		c.Default = &schema.RawExpr{
+			X: defaults.String,
+		}
+	}
+	// TODO(a8m): extract collation from 'CREATE TABLE' statement.
+	t.Columns = append(t.Columns, c)
+	if primary {
+		if t.PrimaryKey == nil {
+			t.PrimaryKey = &schema.Index{
+				Name:   "PRIMARY",
+				Unique: true,
+				Table:  t,
+			}
+		}
+		// Columns are ordered by the `pk` field.
+		t.PrimaryKey.Parts = append(t.PrimaryKey.Parts, &schema.IndexPart{
+			C:     c,
+			SeqNo: len(t.PrimaryKey.Parts) + 1,
+		})
+	}
+	return nil
+}
+
+func parseRawType(c string) (schema.Type, error) {
+	// A datatype may be zero or more names.
+	// https://www.sqlite.org/datatypes.html
+	if c == "" {
+		return &schema.UnsupportedType{}, nil
+	}
+	parts := columnParts(c)
+	switch t := parts[0]; t {
+	case "bool", "boolean":
+		return &schema.BoolType{T: t}, nil
+	case "blob":
+		return &schema.BinaryType{T: t}, nil
+	case "int2", "int8", "int", "integer", "tinyint", "smallint", "mediumint", "bigint", "unsigned big int":
+		// All integer types have the same "type affinity".
+		return &schema.IntegerType{T: t}, nil
+	case "real", "double", "double precision", "float":
+		return &schema.FloatType{T: t}, nil
+	case "numeric", "decimal":
+		ct := &schema.DecimalType{T: t}
+		if len(parts) > 1 {
+			p, err := strconv.ParseInt(parts[1], 10, 64)
+			if err != nil {
+				return nil, fmt.Errorf("parse precision %q", parts[1])
+			}
+			ct.Precision = int(p)
+		}
+		if len(parts) > 2 {
+			s, err := strconv.ParseInt(parts[2], 10, 64)
+			if err != nil {
+				return nil, fmt.Errorf("parse scale %q", parts[1])
+			}
+			ct.Scale = int(s)
+		}
+		return ct, nil
+	case "char", "character", "varchar", "varying character", "nchar", "native character", "nvarchar", "text", "clob":
+		ct := &schema.StringType{T: t}
+		if len(parts) > 1 {
+			p, err := strconv.ParseInt(parts[1], 10, 64)
+			if err != nil {
+				return nil, fmt.Errorf("parse size %q", parts[1])
+			}
+			ct.Size = int(p)
+		}
+		return ct, nil
+	case "json":
+		return &schema.JSONType{T: t}, nil
+	case "date", "datetime", "time", "timestamp":
+		return &schema.TimeType{T: t}, nil
+	case "uuid":
+		return &UUIDType{T: t}, nil
+	default:
+		return nil, fmt.Errorf("unknown column type %q", t)
+	}
+}
+
+// indexes queries and appends the indexes of the given table.
+func (i *inspect) indexes(ctx context.Context, t *schema.Table) error {
+	rows, err := i.QueryContext(ctx, fmt.Sprintf(indexesQuery, t.Name))
+	if err != nil {
+		return fmt.Errorf("sqlite: querying %q indexes: %w", t.Name, err)
+	}
+	if err := i.addIndexes(t, rows); err != nil {
+		return fmt.Errorf("sqlite: scan %q indexes: %w", t.Name, err)
+	}
+	for _, idx := range t.Indexes {
+		if err := i.indexColumns(ctx, t, idx); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// addIndexes scans the rows and adds the indexes to the table.
+func (i *inspect) addIndexes(t *schema.Table, rows *sql.Rows) error {
+	defer rows.Close()
+	for rows.Next() {
+		var (
+			uniq, partial      bool
+			name, origin, stmt sql.NullString
+		)
+		if err := rows.Scan(&name, &uniq, &origin, &partial, &stmt); err != nil {
+			return err
+		}
+		if origin.String == "pk" {
+			continue
+		}
+		idx := &schema.Index{
+			Name:   name.String,
+			Unique: uniq,
+			Table:  t,
+			Attrs: []schema.Attr{
+				&CreateStmt{S: stmt.String},
+				&IndexOrigin{O: origin.String},
+			},
+		}
+		if partial {
+			i := strings.Index(stmt.String, "WHERE")
+			if i == -1 {
+				return fmt.Errorf("missing partial WHERE clause in: %s", stmt.String)
+			}
+			idx.Attrs = append(idx.Attrs, &IndexPredicate{
+				P: strings.TrimSpace(stmt.String[i+5:]),
+			})
+		}
+		t.Indexes = append(t.Indexes, idx)
+	}
+	return nil
+}
+
+func (i *inspect) indexColumns(ctx context.Context, t *schema.Table, idx *schema.Index) error {
+	rows, err := i.QueryContext(ctx, fmt.Sprintf(indexColumnsQuery, idx.Name))
+	if err != nil {
+		return fmt.Errorf("sqlite: querying %q indexes: %w", t.Name, err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var name sql.NullString
+		if err := rows.Scan(&name); err != nil {
+			return fmt.Errorf("sqlite: scanning index names: %w", err)
+		}
+		switch c, ok := t.Column(name.String); {
+		case ok:
+			idx.Parts = append(idx.Parts, &schema.IndexPart{
+				SeqNo: len(idx.Parts) + 1,
+				C:     c,
+			})
+		// NULL name indicates that the index-part is an expression and we
+		// should extract it from the `CREATE INDEX` statement (not supported atm).
+		case !sqlx.ValidString(name):
+			idx.Parts = append(idx.Parts, &schema.IndexPart{
+				SeqNo: len(idx.Parts) + 1,
+				X:     &schema.RawExpr{X: "<unsupported>"},
+			})
+		default:
+			return fmt.Errorf("sqlite: column %q was not found for index %q", name.String, idx.Name)
+		}
+	}
+	return nil
+}
+
+// fks queries and appends the foreign-keys of the given table.
+func (i *inspect) fks(ctx context.Context, t *schema.Table) error {
+	rows, err := i.QueryContext(ctx, fmt.Sprintf(fksQuery, t.Name))
+	if err != nil {
+		return fmt.Errorf("sqlite: querying %q foreign-keys: %w", t.Name, err)
+	}
+	if err := i.addFKs(t, rows); err != nil {
+		return fmt.Errorf("sqlite: scan %q foreign-keys: %w", t.Name, err)
+	}
+	return fillConstName(t)
+}
+
+func (i *inspect) addFKs(t *schema.Table, rows *sql.Rows) error {
+	ids := make(map[int]*schema.ForeignKey)
+	for rows.Next() {
+		var (
+			id                                                  int
+			column, refColumn, refTable, updateRule, deleteRule string
+		)
+		if err := rows.Scan(&id, &column, &refColumn, &refTable, &updateRule, &deleteRule); err != nil {
+			return err
+		}
+		fk, ok := ids[id]
+		if !ok {
+			fk = &schema.ForeignKey{
+				Symbol:   strconv.Itoa(id),
+				Table:    t,
+				RefTable: t,
+				OnDelete: schema.ReferenceOption(deleteRule),
+				OnUpdate: schema.ReferenceOption(updateRule),
+			}
+			if refTable != t.Name {
+				fk.RefTable = &schema.Table{Name: refTable, Schema: &schema.Schema{Name: t.Schema.Name}}
+			}
+			ids[id] = fk
+			t.ForeignKeys = append(t.ForeignKeys, fk)
+		}
+		c, ok := t.Column(column)
+		if !ok {
+			return fmt.Errorf("column %q was not found for fk %q", column, fk.Symbol)
+		}
+		// Rows are ordered by SEQ that specifies the
+		// position of the column in the FK definition.
+		if _, ok := fk.Column(c.Name); !ok {
+			fk.Columns = append(fk.Columns, c)
+			c.ForeignKeys = append(c.ForeignKeys, fk)
+		}
+
+		// Stub referenced columns or link if it is a self-reference.
+		var rc *schema.Column
+		if fk.Table != fk.RefTable {
+			rc = &schema.Column{Name: refColumn}
+		} else if c, ok := t.Column(refColumn); ok {
+			rc = c
+		} else {
+			return fmt.Errorf("referenced column %q was not found for fk %q", refColumn, fk.Symbol)
+		}
+		if _, ok := fk.RefColumn(rc.Name); !ok {
+			fk.RefColumns = append(fk.RefColumns, rc)
+		}
+	}
+	// TODO(a8m): extract the foreign-key name from the `CREATE TABLE` statement.
+	return nil
+}
+
+// tableNames returns a list of all tables exist in the schema.
+func (i *inspect) tables(ctx context.Context, opts *schema.InspectOptions) ([]*schema.Table, error) {
+	var (
+		args  []interface{}
+		query = tablesQuery
+	)
+	if opts != nil && len(opts.Tables) > 0 {
+		query += " AND name IN (" + strings.Repeat("?, ", len(opts.Tables)-1) + "?)"
+		for _, s := range opts.Tables {
+			args = append(args, s)
+		}
+	}
+	rows, err := i.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, fmt.Errorf("sqlite: querying schema tables: %w", err)
+	}
+	defer rows.Close()
+	var tables []*schema.Table
+	for rows.Next() {
+		var name, stmt string
+		if err := rows.Scan(&name, &stmt); err != nil {
+			return nil, fmt.Errorf("sqlite: scanning table: %w", err)
+		}
+		stmt = strings.TrimSpace(stmt)
+		t := &schema.Table{
+			Name: name,
+			Attrs: []schema.Attr{
+				&CreateStmt{S: strings.TrimSpace(stmt)},
+			},
+		}
+		if strings.HasSuffix(stmt, "WITHOUT ROWID") || strings.HasSuffix(stmt, "without rowid") {
+			t.Attrs = append(t.Attrs, &WithoutRowID{})
+		}
+		tables = append(tables, t)
+	}
+	return tables, nil
+}
+
+// schemas returns the list of the schemas in the database.
+func (i *inspect) databases(ctx context.Context, opts *schema.InspectRealmOption) ([]*schema.Schema, error) {
+	var (
+		args  []interface{}
+		query = databasesQuery
+	)
+	if opts != nil && len(opts.Schemas) > 0 {
+		query += " WHERE name IN (" + strings.Repeat("?, ", len(opts.Schemas)-1) + "?)"
+		for _, s := range opts.Schemas {
+			args = append(args, s)
+		}
+	}
+	rows, err := i.QueryContext(ctx, query, args...)
+	if err != nil {
+		return nil, fmt.Errorf("sqlite: querying schemas: %w", err)
+	}
+	defer rows.Close()
+	var schemas []*schema.Schema
+	for rows.Next() {
+		var name, file sql.NullString
+		if err := rows.Scan(&name, &file); err != nil {
+			return nil, err
+		}
+		// File is missing if the database is not
+		// associated with a file (:memory: mode).
+		if file.String == "" {
+			file.String = ":memory:"
+		}
+		schemas = append(schemas, &schema.Schema{
+			Name:  name.String,
+			Attrs: []schema.Attr{&File{Name: file.String}},
+		})
+	}
+	return schemas, nil
+}
+
+type (
+	// File describes a database file.
+	File struct {
+		schema.Attr
+		Name string
+	}
+
+	// CreateStmt describes the SQL statement used to create a resource.
+	CreateStmt struct {
+		schema.Attr
+		S string
+	}
+
+	// AutoIncrement describes the `AUTOINCREMENT` configuration.
+	// https://www.sqlite.org/autoinc.html
+	AutoIncrement struct {
+		schema.Attr
+	}
+
+	// WithoutRowID describes the `WITHOUT ROWID` configuration.
+	// See: https://sqlite.org/withoutrowid.html
+	WithoutRowID struct {
+		schema.Attr
+	}
+
+	// IndexPredicate describes a partial index predicate.
+	// See: https://www.sqlite.org/partialindex.html
+	IndexPredicate struct {
+		schema.Attr
+		P string
+	}
+
+	// IndexOrigin describes how the index was created.
+	// See: https://www.sqlite.org/pragma.html#pragma_index_list
+	IndexOrigin struct {
+		schema.Attr
+		O string
+	}
+
+	// A UUIDType defines a UUID type.
+	UUIDType struct {
+		schema.Type
+		T string
+	}
+)
+
+func columnParts(t string) []string {
+	t = strings.TrimSpace(strings.ToLower(t))
+	parts := strings.FieldsFunc(t, func(r rune) bool {
+		return r == '(' || r == ')' || r == ' ' || r == ','
+	})
+	for k := 0; k < 2; k++ {
+		// Join the type back if it was separated with space (e.g. 'varying character').
+		if len(parts) > 1 && !isNumber(parts[0]) && !isNumber(parts[1]) {
+			parts[1] = parts[0] + " " + parts[1]
+			parts = parts[1:]
+		}
+	}
+	return parts
+}
+
+// isNumber reports whether the string is a number (category N).
+func isNumber(s string) bool {
+	for _, r := range s {
+		if !unicode.IsNumber(r) {
+			return false
+		}
+	}
+	return true
+}
+
+var reAutoinc = regexp.MustCompile(`(?i)PRIMARY\s+KEY\s+AUTOINCREMENT`)
+
+// autoinc checks if the table contains a "PRIMARY KEY AUTOINCREMENT" on its
+// CREATE statement, according to https://www.sqlite.org/syntax/column-constraint.html.
+// This is a workaround until we will embed a proper SQLite parser in atlas.
+func autoinc(t *schema.Table) {
+	if t.PrimaryKey == nil || len(t.PrimaryKey.Parts) != 1 || t.PrimaryKey.Parts[0].C == nil {
+		return
+	}
+	if c := (CreateStmt{}); !sqlx.Has(t.Attrs, &c) || !reAutoinc.MatchString(c.S) {
+		return
+	}
+	// Annotate table elements with "AUTOINCREMENT".
+	t.PrimaryKey.Attrs = append(t.PrimaryKey.Attrs, AutoIncrement{})
+	t.PrimaryKey.Parts[0].C.Attrs = append(t.PrimaryKey.Parts[0].C.Attrs, AutoIncrement{})
+}
+
+// The following regexes extract named foreign-key constraints defined in the table-constraints or inlined
+// as column-constraints. Note, we assume the SQL statements are valid as they are returned by SQLite.
+var (
+	reConstC = regexp.MustCompile("(?i)(?:[(,]\\s*)[\"`]*(\\w+)[\"`]*[^,]*\\s+CONSTRAINT\\s+[\"`]*(\\w+)[\"`]*\\s+REFERENCES\\s+[\"`]*(\\w+)[\"`]*\\s*\\(([,\"` \\w]+)\\)")
+	reConstT = regexp.MustCompile("(?i)CONSTRAINT\\s+[\"`]*(\\w+)[\"`]*\\s+FOREIGN\\s+KEY\\s*\\(([,\"` \\w]+)\\)\\s+REFERENCES\\s+[\"`]*(\\w+)[\"`]*\\s*\\(([,\"` \\w]+)\\)")
+)
+
+// fillConstName fills foreign-key constrain names from CREATE TABLE statement.
+func fillConstName(t *schema.Table) error {
+	var c CreateStmt
+	if !sqlx.Has(t.Attrs, &c) {
+		return fmt.Errorf("missing CREATE statment for table: %q", t.Name)
+	}
+	// Loop over table constraints.
+	for _, m := range reConstT.FindAllStringSubmatch(c.S, -1) {
+		if len(m) != 5 {
+			return fmt.Errorf("unexpected number of matches for a table constraint: %q", m)
+		}
+		// Pattern matches "constraint_name", "columns", "ref_table" and "ref_columns".
+		for _, fk := range t.ForeignKeys {
+			// Found a foreign-key match for the constraint.
+			if matchFK(fk, columns(m[2]), m[3], columns(m[4])) {
+				fk.Symbol = m[1]
+				break
+			}
+		}
+	}
+	// Loop over inlined column constraints.
+	for _, m := range reConstC.FindAllStringSubmatch(c.S, -1) {
+		if len(m) != 5 {
+			return fmt.Errorf("unexpected number of matches for a column constraint: %q", m)
+		}
+		// Pattern matches "column", "constraint_name", "ref_table" and "ref_columns".
+		for _, fk := range t.ForeignKeys {
+			// Found a foreign-key match for the constraint.
+			if matchFK(fk, columns(m[1]), m[3], columns(m[4])) {
+				fk.Symbol = m[2]
+				break
+			}
+		}
+	}
+	return nil
+}
+
+// columns from the matched regex above.
+func columns(s string) []string {
+	names := strings.Split(s, ",")
+	for i := range names {
+		names[i] = strings.Trim(strings.TrimSpace(names[i]), "`\"")
+	}
+	return names
+}
+
+// matchFK reports if the foreign-key matches the given attributes.
+func matchFK(fk *schema.ForeignKey, columns []string, refTable string, refColumns []string) bool {
+	if len(fk.Columns) != len(columns) || fk.RefTable.Name != refTable || len(fk.RefColumns) != len(refColumns) {
+		return false
+	}
+	for i := range columns {
+		if fk.Columns[i].Name != columns[i] {
+			return false
+		}
+	}
+	for i := range refColumns {
+		if fk.RefColumns[i].Name != refColumns[i] {
+			return false
+		}
+	}
+	return true
+}
+
+const (
+	// Query to list attached database files.
+	databasesQuery = "SELECT `name`, `file` FROM pragma_database_list()"
+	// Query to list database tables.
+	tablesQuery = "SELECT `name`, `sql` FROM sqlite_master WHERE `type`='table' AND `name` NOT LIKE 'sqlite_%'"
+	// Query to list table information.
+	columnsQuery = "SELECT `name`, `type`, (not `notnull`) AS `nullable`, `dflt_value`, (`pk` <> 0) AS `pk`  FROM pragma_table_info('%s') ORDER BY `pk`, `cid`"
+	// Query to list table indexes.
+	indexesQuery = "SELECT `il`.`name`, `il`.`unique`, `il`.`origin`, `il`.`partial`, `m`.`sql` FROM pragma_index_list('%s') AS il JOIN sqlite_master AS m ON il.name = m.name"
+	// Query to list index columns.
+	indexColumnsQuery = "SELECT name FROM pragma_index_info('%s') ORDER BY seqno"
+	// Query to list table foreign-keys.
+	fksQuery = "SELECT `id`, `from`, `to`, `table`, `on_update`, `on_delete` FROM pragma_foreign_key_list('%s') ORDER BY id, seq"
+)
diff --git a/sql/sqlite/driver_test.go b/sql/sqlite/inspect_test.go
similarity index 100%
rename from sql/sqlite/driver_test.go
rename to sql/sqlite/inspect_test.go
diff --git a/sql/sqlite/migrate.go b/sql/sqlite/migrate.go
index 25765c47..24431d62 100644
--- a/sql/sqlite/migrate.go
+++ b/sql/sqlite/migrate.go
@@ -14,12 +14,7 @@ import (
 )
 
 // A migrate provides migration capabilities for schema elements.
-type migrate struct{ *Driver }
-
-// Migrate returns a SQLite schema executor.
-func (d *Driver) Migrate() schema.Execer {
-	return &migrate{Driver: d}
-}
+type migrate 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.


Test patch:
diff --git a/cmd/action/diff_test.go b/cmd/action/diff_test.go
index 19bf4fc8..3e0206ee 100644
--- a/cmd/action/diff_test.go
+++ b/cmd/action/diff_test.go
@@ -7,6 +7,8 @@ import (
 	"ariga.io/atlas/schema/schemaspec/schemahcl"
 	"ariga.io/atlas/sql/mysql"
 	"ariga.io/atlas/sql/schema"
+
+	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/stretchr/testify/require"
 )
 
@@ -43,7 +45,7 @@ table "groups" {
 }
 `
 	ctx := context.Background()
-	d := dummyDriver()
+	d := mockDriver(t)
 	changes := diff(t, d, beforeHcl, afterHcl)
 	var changeDescs []*changeDesc
 	for _, ch := range changes {
@@ -81,12 +83,16 @@ func diff(t *testing.T, d *Driver, beforeHcl, afterHcl string) []schema.Change {
 	return diff
 }
 
-func dummyDriver() *Driver {
-	i := &interceptor{}
-	drv := &mysql.Driver{ExecQuerier: i}
+func mockDriver(t *testing.T) *Driver {
+	db, m, err := sqlmock.New()
+	require.NoError(t, err)
+	m.ExpectQuery(".*").
+		WillReturnRows(sqlmock.NewRows([]string{"1", "2", "3"}).AddRow("8.0.19", "utf8_general_ci", "utf8"))
+	i := &interceptor{ExecQuerier: db}
+	drv, err := mysql.Open(i)
+	require.NoError(t, err)
 	return &Driver{
-		Execer:      drv.Migrate(),
-		Differ:      drv.Diff(),
+		driver:      drv,
 		interceptor: i,
 	}
 }
diff --git a/internal/integration/mysql_test.go b/internal/integration/mysql_test.go
index 5d084446..4582ba4f 100644
--- a/internal/integration/mysql_test.go
+++ b/internal/integration/mysql_test.go
@@ -916,20 +916,20 @@ func (t *myTest) applyHcl(spec string) {
 	err := mysql.UnmarshalSpec([]byte(spec), schemahcl.Unmarshal, &desired)
 	require.NoError(t, err)
 	existing := realm.Schemas[0]
-	diff, err := t.drv.Diff().SchemaDiff(existing, &desired)
+	diff, err := t.drv.SchemaDiff(existing, &desired)
 	require.NoError(t, err)
-	err = t.drv.Migrate().Exec(context.Background(), diff)
+	err = t.drv.Exec(context.Background(), diff)
 	require.NoError(t, err)
 }
 
 func (t *myTest) diff(t1, t2 *schema.Table) []schema.Change {
-	changes, err := t.drv.Diff().TableDiff(t1, t2)
+	changes, err := t.drv.TableDiff(t1, t2)
 	require.NoError(t, err)
 	return changes
 }
 
 func (t *myTest) migrate(changes ...schema.Change) {
-	err := t.drv.Migrate().Exec(context.Background(), changes)
+	err := t.drv.Exec(context.Background(), changes)
 	require.NoError(t, err)
 }
 
diff --git a/internal/integration/postgres_test.go b/internal/integration/postgres_test.go
index 6da3c6e0..8f1d7baa 100644
--- a/internal/integration/postgres_test.go
+++ b/internal/integration/postgres_test.go
@@ -151,7 +151,7 @@ func TestPostgres_ColumnInt(t *testing.T) {
 				Name:    "users",
 				Columns: []*schema.Column{{Name: "a", Type: &schema.ColumnType{Type: &schema.IntegerType{T: "bigint"}}}},
 			}
-			err := t.drv.Migrate().Exec(ctx, []schema.Change{&schema.AddTable{T: usersT}})
+			err := t.drv.Exec(ctx, []schema.Change{&schema.AddTable{T: usersT}})
 			require.NoError(t, err)
 			t.dropTables(usersT.Name)
 			change(usersT.Columns[0])
@@ -226,7 +226,7 @@ func TestPostgres_Enums(t *testing.T) {
 		})
 
 		// Create table with an enum column.
-		err := t.drv.Migrate().Exec(ctx, []schema.Change{&schema.AddTable{T: usersT}})
+		err := t.drv.Exec(ctx, []schema.Change{&schema.AddTable{T: usersT}})
 		require.NoError(t, err, "create a new table with an enum column")
 		t.dropTables(usersT.Name)
 		ensureNoChange(t, usersT)
@@ -238,7 +238,7 @@ func TestPostgres_Enums(t *testing.T) {
 		)
 		changes := t.diff(t.loadUsers(), usersT)
 		require.Len(t, changes, 1)
-		err = t.drv.Migrate().Exec(ctx, []schema.Change{&schema.ModifyTable{T: usersT, Changes: changes}})
+		err = t.drv.Exec(ctx, []schema.Change{&schema.ModifyTable{T: usersT, Changes: changes}})
 		require.NoError(t, err, "add a new enum column to existing table")
 		ensureNoChange(t, usersT)
 
@@ -247,7 +247,7 @@ func TestPostgres_Enums(t *testing.T) {
 		e.Values = append(e.Values, "tuesday")
 		changes = t.diff(t.loadUsers(), usersT)
 		require.Len(t, changes, 1)
-		err = t.drv.Migrate().Exec(ctx, []schema.Change{&schema.ModifyTable{T: usersT, Changes: changes}})
+		err = t.drv.Exec(ctx, []schema.Change{&schema.ModifyTable{T: usersT, Changes: changes}})
 		require.NoError(t, err, "append a value to existing enum")
 		ensureNoChange(t, usersT)
 
@@ -256,7 +256,7 @@ func TestPostgres_Enums(t *testing.T) {
 		e.Values = append(e.Values, "wednesday", "thursday", "friday", "saturday")
 		changes = t.diff(t.loadUsers(), usersT)
 		require.Len(t, changes, 1)
-		err = t.drv.Migrate().Exec(ctx, []schema.Change{&schema.ModifyTable{T: usersT, Changes: changes}})
+		err = t.drv.Exec(ctx, []schema.Change{&schema.ModifyTable{T: usersT, Changes: changes}})
 		require.NoError(t, err, "append multiple values to existing enum")
 		ensureNoChange(t, usersT)
 	})
@@ -834,9 +834,9 @@ func (t *pgTest) applyHcl(spec string) {
 	err := postgres.UnmarshalSpec([]byte(spec), schemahcl.Unmarshal, &desired)
 	require.NoError(t, err)
 	existing := realm.Schemas[0]
-	diff, err := t.drv.Diff().SchemaDiff(existing, &desired)
+	diff, err := t.drv.SchemaDiff(existing, &desired)
 	require.NoError(t, err)
-	err = t.drv.Migrate().Exec(context.Background(), diff)
+	err = t.drv.Exec(context.Background(), diff)
 	require.NoError(t, err)
 }
 
@@ -947,13 +947,13 @@ func (t *pgTest) realm() *schema.Realm {
 }
 
 func (t *pgTest) diff(t1, t2 *schema.Table) []schema.Change {
-	changes, err := t.drv.Diff().TableDiff(t1, t2)
+	changes, err := t.drv.TableDiff(t1, t2)
 	require.NoError(t, err)
 	return changes
 }
 
 func (t *pgTest) migrate(changes ...schema.Change) {
-	err := t.drv.Migrate().Exec(context.Background(), changes)
+	err := t.drv.Exec(context.Background(), changes)
 	require.NoError(t, err)
 }
 
diff --git a/internal/integration/sqlite_test.go b/internal/integration/sqlite_test.go
index fc435e17..069a63a9 100644
--- a/internal/integration/sqlite_test.go
+++ b/internal/integration/sqlite_test.go
@@ -644,9 +644,9 @@ func (t *liteTest) applyHcl(spec string) {
 	err := sqlite.UnmarshalSpec([]byte(spec), schemahcl.Unmarshal, &desired)
 	require.NoError(t, err)
 	existing := realm.Schemas[0]
-	diff, err := t.drv.Diff().SchemaDiff(existing, &desired)
+	diff, err := t.drv.SchemaDiff(existing, &desired)
 	require.NoError(t, err)
-	err = t.drv.Migrate().Exec(context.Background(), diff)
+	err = t.drv.Exec(context.Background(), diff)
 	require.NoError(t, err)
 }
 
@@ -748,13 +748,13 @@ func (t *liteTest) realm() *schema.Realm {
 }
 
 func (t *liteTest) diff(t1, t2 *schema.Table) []schema.Change {
-	changes, err := t.drv.Diff().TableDiff(t1, t2)
+	changes, err := t.drv.TableDiff(t1, t2)
 	require.NoError(t, err)
 	return changes
 }
 
 func (t *liteTest) migrate(changes ...schema.Change) {
-	err := t.drv.Migrate().Exec(context.Background(), changes)
+	err := t.drv.Exec(context.Background(), changes)
 	require.NoError(t, err)
 }
 
diff --git a/sql/mysql/diff_test.go b/sql/mysql/diff_test.go
index 47d3fef6..e2e79cf9 100644
--- a/sql/mysql/diff_test.go
+++ b/sql/mysql/diff_test.go
@@ -9,6 +9,7 @@ import (
 
 	"ariga.io/atlas/sql/schema"
 
+	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/stretchr/testify/require"
 )
 
@@ -266,9 +267,13 @@ func TestDiff_TableDiff(t *testing.T) {
 		}(),
 	}
 	for _, tt := range tests {
-		d := (&Driver{version: "8.0.19"}).Diff()
+		db, m, err := sqlmock.New()
+		require.NoError(t, err)
+		mock{m}.version("8.0.19")
+		drv, err := Open(db)
+		require.NoError(t, err)
 		t.Run(tt.name, func(t *testing.T) {
-			changes, err := d.TableDiff(tt.from, tt.to)
+			changes, err := drv.TableDiff(tt.from, tt.to)
 			require.Equal(t, tt.wantErr, err != nil)
 			require.EqualValues(t, tt.wantChanges, changes)
 		})
@@ -276,40 +281,42 @@ func TestDiff_TableDiff(t *testing.T) {
 }
 
 func TestDiff_SchemaDiff(t *testing.T) {
-	var (
-		d    = (&Driver{}).Diff()
-		from = &schema.Schema{
-			Realm: &schema.Realm{
-				Attrs: []schema.Attr{
-					&schema.Collation{V: "latin1"},
-				},
-			},
-			Tables: []*schema.Table{
-				{Name: "users"},
-				{Name: "pets"},
-			},
+	db, m, err := sqlmock.New()
+	require.NoError(t, err)
+	mock{m}.version("8.0.19")
+	drv, err := Open(db)
+	require.NoError(t, err)
+	from := &schema.Schema{
+		Realm: &schema.Realm{
 			Attrs: []schema.Attr{
 				&schema.Collation{V: "latin1"},
 			},
-		}
-		to = &schema.Schema{
-			Tables: []*schema.Table{
-				{
-					Name: "users",
-					Columns: []*schema.Column{
-						{Name: "t2_id", Type: &schema.ColumnType{Raw: "int", Type: &schema.IntegerType{T: "int"}}},
-					},
+		},
+		Tables: []*schema.Table{
+			{Name: "users"},
+			{Name: "pets"},
+		},
+		Attrs: []schema.Attr{
+			&schema.Collation{V: "latin1"},
+		},
+	}
+	to := &schema.Schema{
+		Tables: []*schema.Table{
+			{
+				Name: "users",
+				Columns: []*schema.Column{
+					{Name: "t2_id", Type: &schema.ColumnType{Raw: "int", Type: &schema.IntegerType{T: "int"}}},
 				},
-				{Name: "groups"},
 			},
-			Attrs: []schema.Attr{
-				&schema.Collation{V: "utf8"},
-			},
-		}
-	)
+			{Name: "groups"},
+		},
+		Attrs: []schema.Attr{
+			&schema.Collation{V: "utf8"},
+		},
+	}
 	from.Tables[0].Schema = from
 	from.Tables[1].Schema = from
-	changes, err := d.SchemaDiff(from, to)
+	changes, err := drv.SchemaDiff(from, to)
 	require.NoError(t, err)
 	require.EqualValues(t, []schema.Change{
 		&schema.ModifySchema{Changes: []schema.Change{&schema.ModifyAttr{From: from.Attrs[0], To: to.Attrs[0]}}},
@@ -320,60 +327,62 @@ func TestDiff_SchemaDiff(t *testing.T) {
 }
 
 func TestDiff_RealmDiff(t *testing.T) {
-	var (
-		d    = (&Driver{}).Diff()
-		from = &schema.Realm{
-			Schemas: []*schema.Schema{
-				{
-					Name: "public",
-					Tables: []*schema.Table{
-						{Name: "users"},
-						{Name: "pets"},
-					},
-					Attrs: []schema.Attr{
-						&schema.Collation{V: "latin1"},
-					},
+	db, m, err := sqlmock.New()
+	require.NoError(t, err)
+	mock{m}.version("8.0.19")
+	drv, err := Open(db)
+	require.NoError(t, err)
+	from := &schema.Realm{
+		Schemas: []*schema.Schema{
+			{
+				Name: "public",
+				Tables: []*schema.Table{
+					{Name: "users"},
+					{Name: "pets"},
 				},
-				{
-					Name: "internal",
-					Tables: []*schema.Table{
-						{Name: "pets"},
-					},
+				Attrs: []schema.Attr{
+					&schema.Collation{V: "latin1"},
 				},
 			},
-		}
-		to = &schema.Realm{
-			Schemas: []*schema.Schema{
-				{
-					Name: "public",
-					Tables: []*schema.Table{
-						{
-							Name: "users",
-							Columns: []*schema.Column{
-								{Name: "t2_id", Type: &schema.ColumnType{Raw: "int", Type: &schema.IntegerType{T: "int"}}},
-							},
+			{
+				Name: "internal",
+				Tables: []*schema.Table{
+					{Name: "pets"},
+				},
+			},
+		},
+	}
+	to := &schema.Realm{
+		Schemas: []*schema.Schema{
+			{
+				Name: "public",
+				Tables: []*schema.Table{
+					{
+						Name: "users",
+						Columns: []*schema.Column{
+							{Name: "t2_id", Type: &schema.ColumnType{Raw: "int", Type: &schema.IntegerType{T: "int"}}},
 						},
-						{Name: "pets"},
-					},
-					Attrs: []schema.Attr{
-						&schema.Collation{V: "utf8"},
 					},
+					{Name: "pets"},
 				},
-				{
-					Name: "test",
-					Tables: []*schema.Table{
-						{Name: "pets"},
-					},
+				Attrs: []schema.Attr{
+					&schema.Collation{V: "utf8"},
 				},
 			},
-		}
-	)
+			{
+				Name: "test",
+				Tables: []*schema.Table{
+					{Name: "pets"},
+				},
+			},
+		},
+	}
 	from.Schemas[0].Realm = from
 	from.Schemas[0].Tables[0].Schema = from.Schemas[0]
 	from.Schemas[0].Tables[1].Schema = from.Schemas[0]
 	to.Schemas[0].Realm = to
 	to.Schemas[0].Tables[0].Schema = to.Schemas[0]
-	changes, err := d.RealmDiff(from, to)
+	changes, err := drv.RealmDiff(from, to)
 	require.NoError(t, err)
 	require.EqualValues(t, []schema.Change{
 		&schema.ModifySchema{Changes: []schema.Change{&schema.ModifyAttr{From: from.Schemas[0].Attrs[0], To: to.Schemas[0].Attrs[0]}}},
diff --git a/sql/mysql/migrate_test.go b/sql/mysql/migrate_test.go
index 7ac58226..72df312b 100644
--- a/sql/mysql/migrate_test.go
+++ b/sql/mysql/migrate_test.go
@@ -177,5 +177,5 @@ func newMigrate(version string) (schema.Execer, *mock, error) {
 	if err != nil {
 		return nil, nil, err
 	}
-	return drv.Migrate(), mk, nil
+	return drv, mk, nil
 }
diff --git a/sql/postgres/diff_test.go b/sql/postgres/diff_test.go
index feb8b4de..46d2595b 100644
--- a/sql/postgres/diff_test.go
+++ b/sql/postgres/diff_test.go
@@ -7,6 +7,8 @@ package postgres
 import (
 	"testing"
 
+	"github.com/DATA-DOG/go-sqlmock"
+
 	"ariga.io/atlas/sql/schema"
 
 	"github.com/stretchr/testify/require"
@@ -207,9 +209,13 @@ func TestDiff_TableDiff(t *testing.T) {
 		}(),
 	}
 	for _, tt := range tests {
-		d := (&Driver{version: "13"}).Diff()
+		db, m, err := sqlmock.New()
+		require.NoError(t, err)
+		mock{m}.version("130000")
+		drv, err := Open(db)
+		require.NoError(t, err)
 		t.Run(tt.name, func(t *testing.T) {
-			changes, err := d.TableDiff(tt.from, tt.to)
+			changes, err := drv.TableDiff(tt.from, tt.to)
 			require.Equal(t, tt.wantErr, err != nil)
 			require.EqualValues(t, tt.wantChanges, changes)
 		})
@@ -217,29 +223,31 @@ func TestDiff_TableDiff(t *testing.T) {
 }
 
 func TestDiff_SchemaDiff(t *testing.T) {
-	var (
-		d    = (&Driver{version: "13"}).Diff()
-		from = &schema.Schema{
-			Tables: []*schema.Table{
-				{Name: "users"},
-				{Name: "pets"},
-			},
-		}
-		to = &schema.Schema{
-			Tables: []*schema.Table{
-				{
-					Name: "users",
-					Columns: []*schema.Column{
-						{Name: "t2_id", Type: &schema.ColumnType{Raw: "int", Type: &schema.IntegerType{T: "int"}}},
-					},
+	db, m, err := sqlmock.New()
+	require.NoError(t, err)
+	mock{m}.version("130000")
+	drv, err := Open(db)
+	require.NoError(t, err)
+	from := &schema.Schema{
+		Tables: []*schema.Table{
+			{Name: "users"},
+			{Name: "pets"},
+		},
+	}
+	to := &schema.Schema{
+		Tables: []*schema.Table{
+			{
+				Name: "users",
+				Columns: []*schema.Column{
+					{Name: "t2_id", Type: &schema.ColumnType{Raw: "int", Type: &schema.IntegerType{T: "int"}}},
 				},
-				{Name: "groups"},
 			},
-		}
-	)
+			{Name: "groups"},
+		},
+	}
 	from.Tables[0].Schema = from
 	from.Tables[1].Schema = from
-	changes, err := d.SchemaDiff(from, to)
+	changes, err := drv.SchemaDiff(from, to)
 	require.NoError(t, err)
 	require.EqualValues(t, []schema.Change{
 		&schema.ModifyTable{T: from.Tables[0], Changes: []schema.Change{&schema.AddColumn{C: to.Tables[0].Columns[0]}}},
diff --git a/sql/sqlite/diff_test.go b/sql/sqlite/diff_test.go
index fcc53be1..ff00d2d7 100644
--- a/sql/sqlite/diff_test.go
+++ b/sql/sqlite/diff_test.go
@@ -9,6 +9,7 @@ import (
 
 	"ariga.io/atlas/sql/schema"
 
+	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/stretchr/testify/require"
 )
 
@@ -199,9 +200,13 @@ func TestDiff_TableDiff(t *testing.T) {
 		}(),
 	}
 	for _, tt := range tests {
-		d := Driver{version: "13"}.Diff()
+		db, m, err := sqlmock.New()
+		require.NoError(t, err)
+		mock{m}.systemVars("3.36.0")
+		drv, err := Open(db)
+		require.NoError(t, err)
 		t.Run(tt.name, func(t *testing.T) {
-			changes, err := d.TableDiff(tt.from, tt.to)
+			changes, err := drv.TableDiff(tt.from, tt.to)
 			require.Equal(t, tt.wantErr, err != nil, err)
 			require.EqualValues(t, tt.wantChanges, changes)
 		})
@@ -209,29 +214,31 @@ func TestDiff_TableDiff(t *testing.T) {
 }
 
 func TestDiff_SchemaDiff(t *testing.T) {
-	var (
-		d    = Driver{}.Diff()
-		from = &schema.Schema{
-			Tables: []*schema.Table{
-				{Name: "users"},
-				{Name: "pets"},
-			},
-		}
-		to = &schema.Schema{
-			Tables: []*schema.Table{
-				{
-					Name: "users",
-					Columns: []*schema.Column{
-						{Name: "t2_id", Type: &schema.ColumnType{Raw: "int", Type: &schema.IntegerType{T: "int"}}},
-					},
+	db, m, err := sqlmock.New()
+	require.NoError(t, err)
+	mock{m}.systemVars("3.36.0")
+	drv, err := Open(db)
+	require.NoError(t, err)
+	from := &schema.Schema{
+		Tables: []*schema.Table{
+			{Name: "users"},
+			{Name: "pets"},
+		},
+	}
+	to := &schema.Schema{
+		Tables: []*schema.Table{
+			{
+				Name: "users",
+				Columns: []*schema.Column{
+					{Name: "t2_id", Type: &schema.ColumnType{Raw: "int", Type: &schema.IntegerType{T: "int"}}},
 				},
-				{Name: "groups"},
 			},
-		}
-	)
+			{Name: "groups"},
+		},
+	}
 	from.Tables[0].Schema = from
 	from.Tables[1].Schema = from
-	changes, err := d.SchemaDiff(from, to)
+	changes, err := drv.SchemaDiff(from, to)
 	require.NoError(t, err)
 	require.EqualValues(t, []schema.Change{
 		&schema.ModifyTable{T: from.Tables[0], Changes: []schema.Change{&schema.AddColumn{C: to.Tables[0].Columns[0]}}},