Repository: ariga/atlas
Pull request #677 (2022-03-31 10:59:28)

sql/internal/specutil: support references to qualified tables

rotemtam: Found a bug. Hold review

Patch:
diff --git a/cmd/action/apply.go b/cmd/action/apply.go
index 25ff95ea..8cf76395 100644
--- a/cmd/action/apply.go
+++ b/cmd/action/apply.go
@@ -27,6 +27,7 @@ var (
 		DryRun      bool
 		Schema      []string
 		AutoApprove bool
+		Verbose     bool
 	}
 	// ApplyCmd represents the apply command.
 	ApplyCmd = &cobra.Command{
@@ -65,6 +66,7 @@ func init() {
 	ApplyCmd.Flags().BoolVarP(&ApplyFlags.AutoApprove, "auto-approve", "", false, "Auto approve. Apply the schema changes without prompting for approval.")
 	ApplyCmd.Flags().BoolVarP(&ApplyFlags.Web, "web", "w", false, "Open in a local Atlas UI.")
 	ApplyCmd.Flags().StringVarP(&ApplyFlags.Addr, "addr", "", ":5800", "used with -w, local address to bind the server to.")
+	ApplyCmd.Flags().BoolVarP(&ApplyFlags.Verbose, migrateDiffFlagVerbose, "", false, "enable verbose logging")
 	cobra.CheckErr(ApplyCmd.MarkFlagRequired("url"))
 	cobra.CheckErr(ApplyCmd.MarkFlagRequired("file"))
 	dsn2url(ApplyCmd, &ApplyFlags.URL)
@@ -79,10 +81,10 @@ func CmdApplyRun(cmd *cobra.Command, _ []string) {
 	d, err := DefaultMux.OpenAtlas(cmd.Context(), ApplyFlags.URL)
 	cobra.CheckErr(err)
 	defer d.Close()
-	applyRun(cmd.Context(), d, ApplyFlags.URL, ApplyFlags.File, ApplyFlags.DryRun, ApplyFlags.AutoApprove)
+	applyRun(cmd.Context(), d, ApplyFlags.URL, ApplyFlags.File, ApplyFlags.DryRun, ApplyFlags.AutoApprove, ApplyFlags.Verbose)
 }
 
-func applyRun(ctx context.Context, d *Driver, url string, file string, dryRun bool, autoApprove bool) {
+func applyRun(ctx context.Context, d *Driver, url string, file string, dryRun bool, autoApprove bool, verbose bool) {
 	schemas := ApplyFlags.Schema
 	if n, err := SchemaNameFromURL(ctx, url); n != "" {
 		cobra.CheckErr(err)
@@ -110,7 +112,11 @@ func applyRun(ctx context.Context, d *Driver, url string, file string, dryRun bo
 		}
 	}
 	if _, ok := d.Driver.(schema.Normalizer); ok && ApplyFlags.DevURL != "" {
-		dev, err := DefaultMux.OpenAtlas(ctx, ApplyFlags.DevURL)
+		var opts []ProviderOption
+		if verbose {
+			opts = append(opts, &VerboseLogging{})
+		}
+		dev, err := DefaultMux.OpenAtlas(ctx, ApplyFlags.DevURL, opts...)
 		cobra.CheckErr(err)
 		defer d.Close()
 		desired, err = dev.Driver.(schema.Normalizer).NormalizeRealm(ctx, desired)
diff --git a/cmd/action/docker.go b/cmd/action/docker.go
index da17202e..591a46aa 100644
--- a/cmd/action/docker.go
+++ b/cmd/action/docker.go
@@ -28,6 +28,7 @@ const pass = "pass"
 type (
 	// DockerConfig is used to configure container creation.
 	DockerConfig struct {
+		setup []string // contains statements to execute once the service is up
 		// Image is the name of the image to pull and run.
 		Image string
 		// Env vars to pass to the docker container.
@@ -65,14 +66,35 @@ func NewConfig(opts ...DockerConfigOption) (*DockerConfig, error) {
 
 // MySQL returns a new DockerConfig for a MySQL image.
 func MySQL(version string, opts ...DockerConfigOption) (*DockerConfig, error) {
-	return NewConfig(append(
-		[]DockerConfigOption{
-			Image("mysql:" + version),
-			Port("3306"),
-			Env("MYSQL_ROOT_PASSWORD=" + pass),
-		},
-		opts...,
-	)...,
+	return NewConfig(
+		append(
+			[]DockerConfigOption{
+				Image("mysql:" + version),
+				Port("3306"),
+				Env("MYSQL_ROOT_PASSWORD=" + pass),
+			},
+			opts...,
+		)...,
+	)
+}
+
+// MariaDB returns a new DockerConfig for a MariaDB image.
+func MariaDB(version string, opts ...DockerConfigOption) (*DockerConfig, error) {
+	return MySQL(version, append([]DockerConfigOption{Image("mariadb:" + version)}, opts...)...)
+}
+
+// PostgreSQL returns a new DockerConfig for a PostgreSQL image.
+func PostgreSQL(version string, opts ...DockerConfigOption) (*DockerConfig, error) {
+	return NewConfig(
+		append(
+			[]DockerConfigOption{
+				Image("postgres:" + version),
+				Port("5432"),
+				Env("POSTGRES_PASSWORD=" + pass),
+				setup("DROP SCHEMA IF EXISTS public CASCADE;"),
+			},
+			opts...,
+		)...,
 	)
 }
 
@@ -83,7 +105,7 @@ func MySQL(version string, opts ...DockerConfigOption) (*DockerConfig, error) {
 //
 func Image(i string) DockerConfigOption {
 	return func(c *DockerConfig) error {
-		c.Image = i
+		c.Image = strings.TrimSuffix(i, ":")
 		return nil
 	}
 }
@@ -124,6 +146,17 @@ func Out(w io.Writer) DockerConfigOption {
 	}
 }
 
+// setup adds statements to execute once the service is ready. For example:
+//
+//  setup("DROP SCHEMA IF EXISTS public CASCADE;")
+//
+func setup(s ...string) DockerConfigOption {
+	return func(c *DockerConfig) error {
+		c.setup = s
+		return nil
+	}
+}
+
 // Run pulls and starts a new docker container from the DockerConfig.
 func (c *DockerConfig) Run(ctx context.Context) (*Container, error) {
 	// Make sure the configuration is not missing critical values.
@@ -133,13 +166,13 @@ func (c *DockerConfig) Run(ctx context.Context) (*Container, error) {
 	// Get a free host TCP port the container can bind its exposed service port on.
 	p, err := freePort()
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("getting open port: %w", err)
 	}
 	// Make sure the image is up-to-date.
 	cmd := exec.CommandContext(ctx, "docker", "pull", c.Image) //nolint:gosec
 	cmd.Stdout = c.Out
 	if err := cmd.Run(); err != nil {
-		return nil, err
+		return nil, fmt.Errorf("pulling image: %w", err)
 	}
 	// Run the container.
 	args := []string{"docker", "run", "--rm", "--detach"}
@@ -163,8 +196,8 @@ func (c *DockerConfig) Run(ctx context.Context) (*Container, error) {
 }
 
 // Down stops and removes this container.
-func (c *Container) Down(ctx context.Context) error {
-	return exec.CommandContext(ctx, "docker", "stop", c.ID).Run() //nolint:gosec
+func (c *Container) Down() error {
+	return exec.Command("docker", "stop", c.ID).Run() //nolint:gosec
 }
 
 // Wait waits for this container to be ready.
@@ -188,11 +221,18 @@ func (c *Container) Wait(ctx context.Context, timeout time.Duration) error {
 		case <-time.After(100 * time.Millisecond):
 			db, err := sql.Open(drv, dsn)
 			if err != nil {
-				continue
+				return err
 			}
 			if err := db.PingContext(ctx); err != nil {
 				continue
 			}
+			for _, s := range c.cfg.setup {
+				if _, err := db.ExecContext(ctx, s); err != nil {
+					_ = db.Close()
+					return fmt.Errorf("%q: %w", s, err)
+				}
+			}
+			_ = db.Close()
 			fmt.Fprintln(c.out, "Service is ready to connect!")
 			return nil
 		case <-ctx.Done():
@@ -205,16 +245,37 @@ func (c *Container) Wait(ctx context.Context, timeout time.Duration) error {
 
 // Driver returns the database/sql driver name.
 func (c *Container) Driver() string {
-	return strings.SplitN(c.cfg.Image, ":", 2)[0]
+	switch img := strings.SplitN(c.cfg.Image, ":", 2)[0]; img {
+	case "mysql", "mariadb":
+		return "mysql"
+	default:
+		return img
+	}
 }
 
 // DSN returns a DSN to connect to the Container.
 func (c *Container) DSN() (string, error) {
-	switch drv := c.Driver(); drv {
+	switch img := strings.SplitN(c.cfg.Image, ":", 2)[0]; img {
 	case "mysql", "mariadb":
 		return fmt.Sprintf("root:%s@tcp(localhost:%s)/", c.Passphrase, c.Port), nil
+	case "postgres":
+		return fmt.Sprintf("host=localhost port=%s dbname=postgres user=postgres password=%s sslmode=disable", c.Port, c.Passphrase), nil
+	default:
+		return "", fmt.Errorf("unsupported image: %q", c.cfg.Image)
+	}
+}
+
+// URL returns a URL to connect to the Container.
+func (c *Container) URL() (string, error) {
+	switch img := strings.SplitN(c.cfg.Image, ":", 2)[0]; img {
+	case "postgres":
+		return fmt.Sprintf("postgres://postgres:%s@localhost:%s/postgres?sslmode=disable", c.Passphrase, c.Port), nil
 	default:
-		return "", fmt.Errorf("unknown driver: %q", drv)
+		dsn, err := c.DSN()
+		if err != nil {
+			return "", err
+		}
+		return fmt.Sprintf("%s://%s", strings.SplitN(c.cfg.Image, ":", 2)[0], dsn), nil
 	}
 }
 
diff --git a/cmd/action/provider.go b/cmd/action/provider.go
index 0212fa6b..7227f8a0 100644
--- a/cmd/action/provider.go
+++ b/cmd/action/provider.go
@@ -82,16 +82,15 @@ func sqliteProvider(_ context.Context, dsn string, _ ...ProviderOption) (*Driver
 	}, nil
 }
 
-var reDockerConfig = regexp.MustCompile("(mysql|mariadb)(?::([0-9a-zA-Z.-]+)?(\\?.*)?)?")
+var reDockerConfig = regexp.MustCompile("(mysql|mariadb|tidb|postgres)(?::([0-9a-zA-Z.-]+)?(\\?.*)?)?")
 
 type dockerCloser struct {
 	c   *Container
 	drv *Driver
-	ctx context.Context
 }
 
 func (dc *dockerCloser) Close() (err error) {
-	derr, cerr := dc.drv.Close(), dc.c.Down(dc.ctx)
+	derr, cerr := dc.drv.Close(), dc.c.Down()
 	if derr != nil {
 		err = derr
 	}
@@ -105,14 +104,23 @@ func dockerProvider(ctx context.Context, dsn string, opts ...ProviderOption) (*D
 	// The DSN has the driver part (docker:// removed already.
 	// Get rid of the query arguments, and we have the image name.
 	m := reDockerConfig.FindStringSubmatch(dsn)
+	if len(m) != 4 {
+		return nil, fmt.Errorf("invalid docker url %q", dsn)
+	}
 	img, v := m[1], m[2]
 	var (
 		cfg *DockerConfig
 		err error
 	)
 	switch img {
-	case "mysql", "mariadb":
+	case "mysql":
 		cfg, err = MySQL(v)
+	case "mariadb":
+		cfg, err = MariaDB(v)
+	case "postgres":
+		cfg, err = PostgreSQL(v)
+	default:
+		return nil, fmt.Errorf("unsupported docker image %q", img)
 	}
 	if err != nil {
 		return nil, err
@@ -131,22 +139,23 @@ func dockerProvider(ctx context.Context, dsn string, opts ...ProviderOption) (*D
 		return nil, err
 	}
 	if err := c.Wait(ctx, time.Minute); err != nil {
+		_ = c.Down()
 		return nil, err
 	}
-	d, err := c.DSN()
+	u, err := c.URL()
 	if err != nil {
-		_ = c.Down(ctx)
+		_ = c.Down()
 		return nil, err
 	}
-	drv, err := DefaultMux.OpenAtlas(ctx, fmt.Sprintf("%s://%s", img, d))
+	drv, err := DefaultMux.OpenAtlas(ctx, u)
 	if err != nil {
-		_ = c.Down(ctx)
+		_ = c.Down()
 		return nil, err
 	}
 	return &Driver{
 		Driver:      drv.Driver,
 		Marshaler:   drv.Marshaler,
 		Unmarshaler: drv.Unmarshaler,
-		Closer:      &dockerCloser{c, drv, ctx},
+		Closer:      &dockerCloser{c, drv},
 	}, nil
 }
diff --git a/doc/md/cli/dev.md b/doc/md/cli/dev.md
index 40350ca6..02fca4ad 100644
--- a/doc/md/cli/dev.md
+++ b/doc/md/cli/dev.md
@@ -7,8 +7,9 @@ slug: /cli/dev-database
 
 Various Atlas commands accept the `--dev-url` flag. This option uses the [URL](url.mdx) format, and allows passing
 a development database (a twin environment) to validate schemas, simulate migrations and calculate the state of the
-migration directory by replaying the historical changes. Let's go over a few examples to explain the benefits of using
-a dev/twin database.
+migration directory by replaying the historical changes. Let's go over a few examples to explain the benefits of using a
+dev/twin database. For a one-time use Atlas can spin up an ephemeral local docker container for you with a special
+[docker driver](url.mdx).
 
 ## Validation
 
diff --git a/doc/md/cli/reference.md b/doc/md/cli/reference.md
index 5135fdeb..be5e60b8 100644
--- a/doc/md/cli/reference.md
+++ b/doc/md/cli/reference.md
@@ -187,6 +187,7 @@ migration.
       --auto-approve     Auto approve. Apply the schema changes without prompting for approval.
   -w, --web              Open in a local Atlas UI.
       --addr string      used with -w, local address to bind the server to. (default ":5800")
+      --verbose          enable verbose logging
 
 ```
 
diff --git a/doc/md/cli/url.mdx b/doc/md/cli/url.mdx
index def5f437..5db7433d 100644
--- a/doc/md/cli/url.mdx
+++ b/doc/md/cli/url.mdx
@@ -25,6 +25,7 @@ values={[
 {label: 'MariaDB', value: 'maria'},
 {label: 'PostgreSQL', value: 'postgres'},
 {label: 'SQLite', value: 'sqlite'},
+{label: 'Docker', value: 'docker'},
 ]}>
 <TabItem value="mysql">
 
@@ -69,6 +70,22 @@ sqlite://file.db
 sqlite://file?cache=shared&mode=memory
 ```
 
+</TabItem>
+<TabItem value="docker">
+
+Atlas can spin up an ephemeral local docker container for you by specifying a special URL like below. This can be useful
+if you need a [dev database](dev.md) for schema validation or diffing. However, some images like `mysql` /
+`mariadb` take quite some time to "boot", before they are ready to be used. For a smoother developing experience
+consider spinning up a longer lived container by yourself.
+
+```
+docker://postgres
+
+docker://mysql:8
+
+docker://mariadb:10.8.2-rc-focal
+```
+
 </TabItem>
 </Tabs>
 
diff --git a/sql/internal/specutil/convert.go b/sql/internal/specutil/convert.go
index 4c616aa9..5df942db 100644
--- a/sql/internal/specutil/convert.go
+++ b/sql/internal/specutil/convert.go
@@ -33,50 +33,53 @@ type (
 	CheckSpecFunc         func(*schema.Check) *sqlspec.Check
 )
 
-// Realm converts the schemas and tables into a schema.Realm.
-func Realm(schemas []*sqlspec.Schema, tables []*sqlspec.Table, convertTable ConvertTableFunc) (*schema.Realm, error) {
-	r := &schema.Realm{}
+// Scan populates the Realm from the schemas and table specs.
+func Scan(r *schema.Realm, schemas []*sqlspec.Schema, tables []*sqlspec.Table, convertTable ConvertTableFunc) error {
+	// Build the schemas.
 	for _, schemaSpec := range schemas {
-		var schemaTables []*sqlspec.Table
+		sch := &schema.Schema{Name: schemaSpec.Name, Realm: r}
 		for _, tableSpec := range tables {
 			name, err := SchemaName(tableSpec.Schema)
 			if err != nil {
-				return nil, fmt.Errorf("specutil: cannot extract schema name for table %q: %w", tableSpec.Name, err)
+				return fmt.Errorf("specutil: cannot extract schema name for table %q: %w", tableSpec.Name, err)
 			}
 			if name == schemaSpec.Name {
-				schemaTables = append(schemaTables, tableSpec)
+				tbl, err := convertTable(tableSpec, sch)
+				if err != nil {
+					return err
+				}
+				sch.Tables = append(sch.Tables, tbl)
 			}
 		}
-		sch, err := Schema(schemaSpec, schemaTables, convertTable)
-		if err != nil {
-			return nil, err
-		}
 		r.Schemas = append(r.Schemas, sch)
 	}
-	return r, nil
+	// Link the foreign keys.
+	for _, sch := range r.Schemas {
+		for _, tbl := range sch.Tables {
+			tableSpec, err := findTableSpec(tables, sch.Name, tbl.Name)
+			if err != nil {
+				return err
+			}
+			if err := linkForeignKeys(tbl, sch, tableSpec); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
 }
 
-// Schema converts a sqlspec.Schema with its relevant []sqlspec.Tables
-// into a schema.Schema.
-func Schema(spec *sqlspec.Schema, tables []*sqlspec.Table, convertTable ConvertTableFunc) (*schema.Schema, error) {
-	sch := &schema.Schema{
-		Name: spec.Name,
-	}
-	m := make(map[*schema.Table]*sqlspec.Table)
-	for _, ts := range tables {
-		table, err := convertTable(ts, sch)
+// findTableSpec searches tableSpecs for a spec of a table named tableName in a schema named schemaName.
+func findTableSpec(tableSpecs []*sqlspec.Table, schemaName, tableName string) (*sqlspec.Table, error) {
+	for _, tbl := range tableSpecs {
+		n, err := SchemaName(tbl.Schema)
 		if err != nil {
 			return nil, err
 		}
-		sch.Tables = append(sch.Tables, table)
-		m[table] = ts
-	}
-	for _, tbl := range sch.Tables {
-		if err := LinkForeignKeys(tbl, sch, m[tbl]); err != nil {
-			return nil, err
+		if n == schemaName && tbl.Name == tableName {
+			return tbl, nil
 		}
 	}
-	return sch, nil
+	return nil, fmt.Errorf("table %s.%s not found", schemaName, tableName)
 }
 
 // Table converts a sqlspec.Table to a schema.Table. Table conversion is done without converting
@@ -235,10 +238,10 @@ func PrimaryKey(spec *sqlspec.PrimaryKey, parent *schema.Table) (*schema.Index,
 	}, nil
 }
 
-// LinkForeignKeys creates the foreign keys defined in the Table's spec by creating references
-// to column in the provided Schema. It is assumed that the schema contains all of the tables
-// referenced by the FK definitions in the spec.
-func LinkForeignKeys(tbl *schema.Table, sch *schema.Schema, table *sqlspec.Table) error {
+// linkForeignKeys creates the foreign keys defined in the Table's spec by creating references
+// to column in the provided Schema. It is assumed that all tables referenced FK definitions in the spec
+// are reachable from the provided schema or its connected realm.
+func linkForeignKeys(tbl *schema.Table, sch *schema.Schema, table *sqlspec.Table) error {
 	for _, spec := range table.ForeignKeys {
 		fk := &schema.ForeignKey{Symbol: spec.Symbol, Table: tbl}
 		if spec.OnUpdate != nil {
@@ -528,13 +531,9 @@ func column(t *schema.Table, ref *schemaspec.Ref) (*schema.Column, error) {
 }
 
 func externalRef(ref *schemaspec.Ref, sch *schema.Schema) (*schema.Table, *schema.Column, error) {
-	t, err := tableName(ref)
+	tbl, err := findTable(ref, sch)
 	if err != nil {
-		return nil, nil, fmt.Errorf("sqlspec: table %q not found", ref.V)
-	}
-	tbl, ok := sch.Table(t)
-	if !ok {
-		return nil, nil, fmt.Errorf("sqlspec: table %q not found", t)
+		return nil, nil, err
 	}
 	c, err := column(tbl, ref)
 	if err != nil {
@@ -543,17 +542,53 @@ func externalRef(ref *schemaspec.Ref, sch *schema.Schema) (*schema.Table, *schem
 	return tbl, c, nil
 }
 
-func tableName(ref *schemaspec.Ref) (string, error) {
+// findTable finds the table referenced by ref in the provided schema. If the table
+// is not in the provided schema.Schema other schemas in the connected schema.Realm
+// are searched as well.
+func findTable(ref *schemaspec.Ref, sch *schema.Schema) (*schema.Table, error) {
+	qualifier, tblName, err := tableName(ref)
+	if err != nil {
+		return nil, err
+	}
+	// Search the same schema.
+	if qualifier == "" || qualifier == sch.Name {
+		tbl, ok := sch.Table(tblName)
+		if !ok {
+			return tbl, fmt.Errorf("sqlspec: table %q not found", tblName)
+		}
+		return tbl, nil
+	}
+	if sch.Realm == nil {
+		return nil, fmt.Errorf("sqlspec: table %s.%s not found", qualifier, tblName)
+	}
+	// Search for the table in another schemas in the realm.
+	sch, ok := sch.Realm.Schema(qualifier)
+	if !ok {
+		return nil, fmt.Errorf("sqlspec: schema %q not found", qualifier)
+	}
+	tbl, ok := sch.Table(tblName)
+	if !ok {
+		return tbl, fmt.Errorf("sqlspec: table %q not found", tblName)
+	}
+	return tbl, nil
+}
+
+func tableName(ref *schemaspec.Ref) (qualifier, name string, err error) {
 	s := strings.Split(ref.V, "$column.")
 	if len(s) != 2 {
-		return "", fmt.Errorf("sqlspec: failed to split by column name from %q", ref)
-
-	}
-	s = strings.Split(s[0], ".")
-	if len(s) != 3 {
-		return "", fmt.Errorf("sqlspec: failed to extract table name from %q", s)
+		return "", "", fmt.Errorf("sqlspec: failed to split by column name from %q", ref)
+	}
+	table := strings.TrimSuffix(s[0], ".")
+	s = strings.Split(table, ".")
+	switch len(s) {
+	case 2:
+		name = s[1]
+	case 3:
+		qualifier, name = s[1], s[2]
+	default:
+		return "", "", fmt.Errorf("sqlspec: failed to extract table name from %q", s)
 	}
-	return s[1], nil
+	return
 }
 
 func isLocalRef(r *schemaspec.Ref) bool {
diff --git a/sql/internal/specutil/spec.go b/sql/internal/specutil/spec.go
index 0018e4f4..ce0f5806 100644
--- a/sql/internal/specutil/spec.go
+++ b/sql/internal/specutil/spec.go
@@ -116,20 +116,20 @@ func Unmarshal(data []byte, unmarshaler schemaspec.Unmarshaler, v interface{},
 	}
 	switch v := v.(type) {
 	case *schema.Realm:
-		realm, err := Realm(d.Schemas, d.Tables, convertTable)
+		err := Scan(v, d.Schemas, d.Tables, convertTable)
 		if err != nil {
 			return fmt.Errorf("specutil: failed converting to *schema.Realm: %w", err)
 		}
-		*v = *realm
 	case *schema.Schema:
 		if len(d.Schemas) != 1 {
 			return fmt.Errorf("specutil: expecting document to contain a single schema, got %d", len(d.Schemas))
 		}
-		conv, err := Schema(d.Schemas[0], d.Tables, convertTable)
-		if err != nil {
-			return fmt.Errorf("specutil: failed converting to *schema.Schema: %w", err)
+		var r schema.Realm
+		if err := Scan(&r, d.Schemas, d.Tables, convertTable); err != nil {
+			return err
 		}
-		*v = *conv
+		r.Schemas[0].Realm = nil
+		*v = *r.Schemas[0]
 	default:
 		return fmt.Errorf("specutil: failed unmarshaling spec. %T is not supported", v)
 	}
diff --git a/sql/mysql/sqlspec.go b/sql/mysql/sqlspec.go
index e94e075e..0a4fe6c9 100644
--- a/sql/mysql/sqlspec.go
+++ b/sql/mysql/sqlspec.go
@@ -27,12 +27,12 @@ func UnmarshalSpec(data []byte, unmarshaler schemaspec.Unmarshaler, v interface{
 	}
 	switch v := v.(type) {
 	case *schema.Realm:
-		realm, err := specutil.Realm(d.Schemas, d.Tables, convertTable)
+		err := specutil.Scan(v, d.Schemas, d.Tables, convertTable)
 		if err != nil {
 			return fmt.Errorf("mysql: failed converting to *schema.Realm: %w", err)
 		}
 		for _, schemaSpec := range d.Schemas {
-			schm, ok := realm.Schema(schemaSpec.Name)
+			schm, ok := v.Schema(schemaSpec.Name)
 			if !ok {
 				return fmt.Errorf("could not find schema: %q", schemaSpec.Name)
 			}
@@ -40,19 +40,19 @@ func UnmarshalSpec(data []byte, unmarshaler schemaspec.Unmarshaler, v interface{
 				return err
 			}
 		}
-		*v = *realm
 	case *schema.Schema:
 		if len(d.Schemas) != 1 {
 			return fmt.Errorf("mysql: expecting document to contain a single schema, got %d", len(d.Schemas))
 		}
-		conv, err := specutil.Schema(d.Schemas[0], d.Tables, convertTable)
-		if err != nil {
-			return fmt.Errorf("mysql: failed converting to *schema.Schema: %w", err)
+		var r schema.Realm
+		if err := specutil.Scan(&r, d.Schemas, d.Tables, convertTable); err != nil {
+			return err
 		}
-		if err := convertCharset(d.Schemas[0], &conv.Attrs); err != nil {
+		if err := convertCharset(d.Schemas[0], &r.Schemas[0].Attrs); err != nil {
 			return err
 		}
-		*v = *conv
+		r.Schemas[0].Realm = nil
+		*v = *r.Schemas[0]
 	default:
 		return fmt.Errorf("mysql: failed unmarshaling spec. %T is not supported", v)
 	}
diff --git a/sql/postgres/sqlspec.go b/sql/postgres/sqlspec.go
index 2fa0b80f..e465f61c 100644
--- a/sql/postgres/sqlspec.go
+++ b/sql/postgres/sqlspec.go
@@ -41,20 +41,29 @@ func UnmarshalSpec(data []byte, unmarshaler schemaspec.Unmarshaler, v interface{
 	}
 	switch v := v.(type) {
 	case *schema.Realm:
-		realm, err := Realm(d.Schemas, d.Tables, d.Enums)
-		if err != nil {
+		if err := specutil.Scan(v, d.Schemas, d.Tables, convertTable); err != nil {
 			return fmt.Errorf("specutil: failed converting to *schema.Realm: %w", err)
 		}
-		*v = *realm
+		if len(d.Enums) > 0 {
+			for _, sch := range v.Schemas {
+				if err := convertEnums(d.Tables, d.Enums, sch); err != nil {
+					return err
+				}
+			}
+		}
 	case *schema.Schema:
 		if len(d.Schemas) != 1 {
 			return fmt.Errorf("specutil: expecting document to contain a single schema, got %d", len(d.Schemas))
 		}
-		conv, err := Schema(d.Schemas[0], d.Tables, d.Enums)
-		if err != nil {
-			return fmt.Errorf("specutil: failed converting to *schema.Schema: %w", err)
+		var r schema.Realm
+		if err := specutil.Scan(&r, d.Schemas, d.Tables, convertTable); err != nil {
+			return err
+		}
+		if err := convertEnums(d.Tables, d.Enums, r.Schemas[0]); err != nil {
+			return err
 		}
-		*v = *conv
+		r.Schemas[0].Realm = nil
+		*v = *r.Schemas[0]
 	default:
 		return fmt.Errorf("specutil: failed unmarshaling spec. %T is not supported", v)
 	}
@@ -108,68 +117,6 @@ var (
 	})
 )
 
-// Realm converts the schemas and tables of the doc into a schema.Realm.
-func Realm(schemas []*sqlspec.Schema, tables []*sqlspec.Table, enums []*Enum) (*schema.Realm, error) {
-	r := &schema.Realm{}
-	for _, schemaSpec := range schemas {
-		var (
-			schemaTables []*sqlspec.Table
-			schemaEnums  []*Enum
-		)
-		for _, tableSpec := range tables {
-			name, err := specutil.SchemaName(tableSpec.Schema)
-			if err != nil {
-				return nil, fmt.Errorf("specutil: cannot extract schema name for table %q: %w", tableSpec.Name, err)
-			}
-			if name == schemaSpec.Name {
-				schemaTables = append(schemaTables, tableSpec)
-			}
-		}
-		for _, enum := range enums {
-			name, err := specutil.SchemaName(enum.Schema)
-			if err != nil {
-				return nil, fmt.Errorf("specutil: cannot extract schema name for table %q: %w", enum.Name, err)
-			}
-			if name == schemaSpec.Name {
-				schemaEnums = append(schemaEnums, enum)
-			}
-		}
-		sch, err := Schema(schemaSpec, schemaTables, schemaEnums)
-		if err != nil {
-			return nil, err
-		}
-		r.Schemas = append(r.Schemas, sch)
-	}
-	return r, nil
-}
-
-// Schema converts a sqlspec.Schema with its relevant []sqlspec.Tables and []Enum into a schema.Schema.
-func Schema(spec *sqlspec.Schema, tables []*sqlspec.Table, enums []*Enum) (*schema.Schema, error) {
-	sch := &schema.Schema{
-		Name: spec.Name,
-	}
-	m := make(map[*schema.Table]*sqlspec.Table)
-	for _, ts := range tables {
-		table, err := convertTable(ts, sch)
-		if err != nil {
-			return nil, err
-		}
-		sch.Tables = append(sch.Tables, table)
-		m[table] = ts
-	}
-	for _, tbl := range sch.Tables {
-		if err := specutil.LinkForeignKeys(tbl, sch, m[tbl]); err != nil {
-			return nil, err
-		}
-	}
-	if len(enums) > 0 {
-		if err := convertEnums(tables, enums, sch); err != nil {
-			return nil, err
-		}
-	}
-	return sch, nil
-}
-
 // convertTable converts a sqlspec.Table to a schema.Table. Table conversion is done without converting
 // ForeignKeySpecs into ForeignKeys, as the target tables do not necessarily exist in the schema
 // at this point. Instead, the linking is done by the convertSchema function.
@@ -284,7 +231,7 @@ func convertEnums(tbls []*sqlspec.Table, enums []*Enum, sch *schema.Schema) erro
 				}
 				t, ok := sch.Table(tbl.Name)
 				if !ok {
-					return fmt.Errorf("postgrs: table %q not found in schema %q", tbl.Name, sch.Name)
+					return fmt.Errorf("postgres: table %q not found in schema %q", tbl.Name, sch.Name)
 				}
 				c, ok := t.Column(col.Name)
 				if !ok {
diff --git a/sql/sqlspec/sqlspec.go b/sql/sqlspec/sqlspec.go
index 18142754..0a053efa 100644
--- a/sql/sqlspec/sqlspec.go
+++ b/sql/sqlspec/sqlspec.go
@@ -15,6 +15,7 @@ type (
 	// Table holds a specification for an SQL table.
 	Table struct {
 		Name        string          `spec:",name"`
+		Qualifier   string          `spec:",qualifier"`
 		Schema      *schemaspec.Ref `spec:"schema"`
 		Columns     []*Column       `spec:"column"`
 		PrimaryKey  *PrimaryKey     `spec:"primary_key"`


Test patch:
diff --git a/cmd/action/docker_test.go b/cmd/action/docker_test.go
index faa76d89..bc0f22f3 100644
--- a/cmd/action/docker_test.go
+++ b/cmd/action/docker_test.go
@@ -2,14 +2,13 @@
 // 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 action_test
+package action
 
 import (
 	"context"
 	"io/ioutil"
 	"testing"
 
-	"ariga.io/atlas/cmd/action"
 	"github.com/stretchr/testify/require"
 )
 
@@ -17,16 +16,37 @@ func TestDockerConfig(t *testing.T) {
 	ctx := context.Background()
 
 	// invalid config
-	_, err := (&action.DockerConfig{}).Run(ctx)
+	_, err := (&DockerConfig{}).Run(ctx)
 	require.Error(t, err)
 
 	// MySQL
-	cfg, err := action.MySQL("latest", action.Out(ioutil.Discard))
+	cfg, err := MySQL("latest", Out(ioutil.Discard))
 	require.NoError(t, err)
-	require.Equal(t, &action.DockerConfig{
+	require.Equal(t, &DockerConfig{
 		Image: "mysql:latest",
 		Env:   []string{"MYSQL_ROOT_PASSWORD=pass"},
 		Port:  "3306",
 		Out:   ioutil.Discard,
 	}, cfg)
+
+	// MariaDB
+	cfg, err = MariaDB("latest", Out(ioutil.Discard))
+	require.NoError(t, err)
+	require.Equal(t, &DockerConfig{
+		Image: "mariadb:latest",
+		Env:   []string{"MYSQL_ROOT_PASSWORD=pass"},
+		Port:  "3306",
+		Out:   ioutil.Discard,
+	}, cfg)
+
+	// PostgreSQL
+	cfg, err = PostgreSQL("latest", Out(ioutil.Discard))
+	require.NoError(t, err)
+	require.Equal(t, &DockerConfig{
+		Image: "postgres:latest",
+		Env:   []string{"POSTGRES_PASSWORD=pass"},
+		Port:  "5432",
+		Out:   ioutil.Discard,
+		setup: []string{"DROP SCHEMA IF EXISTS public CASCADE;"},
+	}, cfg)
 }
diff --git a/cmd/action/mux_test.go b/cmd/action/mux_test.go
index dc927480..ae15f8fc 100644
--- a/cmd/action/mux_test.go
+++ b/cmd/action/mux_test.go
@@ -174,7 +174,7 @@ func TestMux_OpenAtlas(t *testing.T) {
 			require.NoError(t, mysqld.SetLogger(log.New(ioutil.Discard, "", 1)))
 			_, err := action.DefaultMux.OpenAtlas(context.Background(), fmt.Sprintf(u, l.Addr()))
 			require.Error(t, err, "mock server rejects all incoming connections")
-			require.NotZero(t, *calls)
+			require.NotZero(t, atomic.LoadInt64(calls))
 		}
 	})
 }
diff --git a/internal/integration/hclsqlspec/hclsqlspec_test.go b/internal/integration/hclsqlspec/hclsqlspec_test.go
index 033f93e9..c9e83a0f 100644
--- a/internal/integration/hclsqlspec/hclsqlspec_test.go
+++ b/internal/integration/hclsqlspec/hclsqlspec_test.go
@@ -1,7 +1,6 @@
 package hclsqlspec
 
 import (
-	"strconv"
 	"testing"
 
 	"ariga.io/atlas/schema/schemaspec"
@@ -370,13 +369,15 @@ table "t2" {
 			exp := &schema.Realm{
 				Schemas: []*schema.Schema{
 					{
-						Name: "account_a",
+						Name:  "account_a",
+						Realm: &r,
 						Tables: []*schema.Table{
 							{Name: "t1"},
 						},
 					},
 					{
-						Name: "account_b",
+						Name:  "account_b",
+						Realm: &r,
 						Tables: []*schema.Table{
 							{Name: "t2"},
 						},
@@ -398,6 +399,7 @@ table "t2" {
 
 func TestUnsignedImmutability(t *testing.T) {
 	f := `table "users" {
+	schema = schema.test
 	column "id" {
 		type = bigint
 		unsigned = true
@@ -416,6 +418,47 @@ schema "test" {
 	require.EqualValues(t, &schema.IntegerType{T: "bigint", Unsigned: false}, tbl.Columns[1].Type.Type)
 }
 
+func TestTablesWithQualifiers(t *testing.T) {
+	h := `
+schema "a" {}
+schema "b" {}
+
+table "a" "users" {
+	schema = schema.a
+	column "id" {
+		type = int
+	}
+	column "friend_id" {
+		type = int
+	}
+	foreign_key "friend_b" {
+		columns = [column.friend_id]
+		ref_columns = [table.b.users.column.id]
+	}
+}
+
+table "b" "users" {
+	schema = schema.b
+	column "id" {
+		type = int
+	}
+	column "friend_id" {
+		type = int
+	}
+	foreign_key "friend_a" {
+		columns = [column.friend_id]
+		ref_columns = [table.a.users.column.id]
+	}
+}
+`
+	var r schema.Realm
+	err := mysql.UnmarshalHCL([]byte(h), &r)
+	require.NoError(t, err)
+
+	require.EqualValues(t, r.Schemas[0].Tables[0].Columns[0], r.Schemas[1].Tables[0].ForeignKeys[0].RefColumns[0])
+	require.EqualValues(t, "b", r.Schemas[0].Tables[0].ForeignKeys[0].RefTable.Schema.Name)
+}
+
 func decode(f string) (*db, error) {
 	d := &db{}
 	if err := hcl.UnmarshalSpec([]byte(f), d); err != nil {
@@ -428,21 +471,3 @@ type db struct {
 	Schemas []*sqlspec.Schema `spec:"schema"`
 	Tables  []*sqlspec.Table  `spec:"table"`
 }
-
-func attrs(attrs ...*schemaspec.Attr) []*schemaspec.Attr {
-	return attrs
-}
-
-func size(n int) *schemaspec.Attr {
-	return &schemaspec.Attr{
-		K: "size",
-		V: &schemaspec.LiteralValue{V: strconv.Itoa(n)},
-	}
-}
-
-func unsigned(b bool) *schemaspec.Attr {
-	return &schemaspec.Attr{
-		K: "unsigned",
-		V: &schemaspec.LiteralValue{V: strconv.FormatBool(b)},
-	}
-}
diff --git a/sql/internal/specutil/convert_test.go b/sql/internal/specutil/convert_test.go
index 163b6a6d..568a1930 100644
--- a/sql/internal/specutil/convert_test.go
+++ b/sql/internal/specutil/convert_test.go
@@ -10,13 +10,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestRef_TableName(t *testing.T) {
-	ref := &schemaspec.Ref{V: "$table.accounts.$column.user_active"}
-	c, err := tableName(ref)
-	require.NoError(t, err)
-	require.Equal(t, "accounts", c)
-}
-
 func TestFromSpec_SchemaName(t *testing.T) {
 	sc := &schema.Schema{
 		Name: "schema",
diff --git a/sql/mysql/sqlspec_test.go b/sql/mysql/sqlspec_test.go
index d2f06707..4a29ec6b 100644
--- a/sql/mysql/sqlspec_test.go
+++ b/sql/mysql/sqlspec_test.go
@@ -15,6 +15,7 @@ schema "schema" {
 }
 
 table "table" {
+	schema = schema.schema
 	column "col" {
 		type = int
 		comment = "column comment"
@@ -78,6 +79,7 @@ table "table" {
 }
 
 table "accounts" {
+	schema = schema.schema
 	column "name" {
 		type = varchar(32)
 	}
diff --git a/sql/postgres/sqlspec_test.go b/sql/postgres/sqlspec_test.go
index 51545133..9fd6bba4 100644
--- a/sql/postgres/sqlspec_test.go
+++ b/sql/postgres/sqlspec_test.go
@@ -16,6 +16,7 @@ schema "schema" {
 }
 
 table "table" {
+	schema = schema.schema
 	column "col" {
 		type = integer
 		comment = "column comment"
@@ -75,6 +76,7 @@ table "table" {
 }
 
 table "accounts" {
+	schema = schema.schema
 	column "name" {
 		type = varchar(32)
 	}
diff --git a/sql/sqlite/sqlspec_test.go b/sql/sqlite/sqlspec_test.go
index d044acdd..8cdcf2a5 100644
--- a/sql/sqlite/sqlspec_test.go
+++ b/sql/sqlite/sqlspec_test.go
@@ -19,6 +19,7 @@ schema "schema" {
 }
 
 table "table" {
+	schema = schema.schema
 	column "id" {
 		type = integer
 		auto_increment = true
@@ -58,6 +59,7 @@ table "table" {
 }
 
 table "accounts" {
+	schema = schema.schema
 	column "name" {
 		type = varchar(32)
 	}