Repository: ariga/atlas
Pull request #1601 (2023-04-23 20:31:43)

PostGIS support
Hello,
while Atlas [supports using PostGIS in some capacity](https://github.com/ariga/atlas/issues/469#issuecomment-1088284374), we could improve it. This PR is an attempt to adds support to PostGIS extension to Atlas test suite.

PostGIS is the [most used extension](https://www.timescale.com/state-of-postgres/2022#ecosystem-tools) for Postgres.

<img width="1288" alt="image" src="https://user-images.githubusercontent.com/1251151/233859715-dac43008-d6d4-4528-85d2-cd4a019d6421.png">


The goals of this PR are small:
- Run the all the Postgres tests with the PostGIS image successful
- Create a pattern for testing Postgres extensions and be an example for other extensions (I'm looking at you TimescaleDB)

I think that been able to pass all the tests with PosGIS would solve the problems related to the schemas created by the extension (#1568 and #1577) and would pave the way to solve the problem related to geometry detection (#1577).

## The problem

PostGIS (can) create a few schema/tables, that Atlas recognize as user defined, this schema/tables are used to provide extra functionality and most of the time should not be changed by the user (or Atlas), there is ways to exclude them, but I think that Atlas should be able to identify schema/tables created by extensions and make the flow as easy as possible.

schemas:
- [topology](https://postgis.net/docs/Topology.html) (created by the extension [`postgis_topology`](https://github.com/postgis/docker-postgis/blob/8a3ed2b584cdbdc984e364f8b5ecf4ba73b82171/15-master/initdb-postgis.sh#L18))
- [tiger](https://postgis.net/docs/Extras.html#Tiger_Geocoder) (created by the extension [`postgis_tiger_geocoder`](https://github.com/postgis/docker-postgis/blob/8a3ed2b584cdbdc984e364f8b5ecf4ba73b82171/15-master/initdb-postgis.sh#L23))
- [tiger_data](https://postgis.net/docs/Extras.html#Tiger_Geocoder)

tables:
- [spatial_ref_sys](https://postgis.net/docs/manual-1.4/ch04.html#spatial_ref_sys)

#### Declarative:

When running inspect on an empty database, Atlas tries to define the table `spatial_ref_sys`

```shell
atlas schema inspect -u "postgres://localhost:15432/postgis_test?sslmode=disable" > schema.hcl
```

<details>
  <summary>schema.hcl</summary>

```hcl
table "spatial_ref_sys" {
  schema = schema.public
  column "srid" {
    null = false
    type = integer
  }
  column "auth_name" {
    null = true
    type = character_varying(256)
  }
  column "auth_srid" {
    null = true
    type = integer
  }
  column "srtext" {
    null = true
    type = character_varying(2048)
  }
  column "proj4text" {
    null = true
    type = character_varying(2048)
  }
  primary_key {
    columns = [column.srid]
  }
  check "spatial_ref_sys_srid_check" {
    expr = "((srid > 0) AND (srid <= 998999))"
  }
}
schema "public" {
}
```

</details>

It's possible to mitigate the problem using `exclude`:

```shell
atlas schema inspect -u "postgres://localhost:15432/postgis_test?sslmode=disable" --exclude "*.spatial_ref_sys" > schema.hcl
# or
atlas schema inspect -u "postgres://localhost:15432/postgis_test?search_path=public&sslmode=disable" --exclude "spatial_ref_sys" > schema.hcl
```

When applying the schema, we have to exclude it as well, otherwise we face an error:

```
Error: drop "spatial_ref_sys" table: pq: cannot drop table spatial_ref_sys because extension postgis requires it
```

#### Versioned:

The `atlas migrate diff` don't support the exclude flag, so when generating from a database with the PostGIS extension already enabled, we get the `spatial_ref_sys` table definition, one way to solve it would be to edit the file and hash the migration directory again. 

The `atlas migrate apply` works as expected as well, but the `atlas migrate lint` causes problems.

When running with the docker image, we get a `pq: extension "postgis" is not available` error as expected.

When using an externa database with PostGIS installed, we get different errors:

```
Error: drop "spatial_ref_sys" table: pq: cannot drop table spatial_ref_sys because extension postgis requires it
```

and

```
Error: taking database snapshot: sql/migrate: connected database is not clean: found table "spatial_ref_sys" in connected schema
```

Since the `atlas migrate` commands dont support the `exclude` flag, we don't really have a work around in this case.

## The solution (‽)

This PR tries to solve the problem by changing the query used to inspect tables to ignore tables created/used by extensions. This fixes the problem of the `spatial_ref_sys` table and mostly (if not all) the problems of the declarative way and the versioned way as well.

### But...

The tests are still failing due to the schemas created by PostGIS docker image and `Driver.CheckClean` checks.
See, on my machine I don't have that schemas (because I only use the base `postgis` extension), so the tests against my local Postgres instance works.

Tests error:
```
sql/migrate: connected database is not clean: found schema "tiger". baseline version or allow-dirty is required
```

I tried to change [the query used to inspect the schemas](https://github.com/ariga/atlas/blob/30a2c72536f86913891ba2ce642acee7ac5dbfc9/sql/postgres/inspect.go#L1076) as well, got one almost working, but the `tiger_data` still cause problems.

```text
test=# SELECT 
    schema_name 
FROM information_schema.schemata 
WHERE 
    schema_name NOT IN ('information_schema', 'pg_catalog', 'pg_toast', 'crdb_internal', 'pg_extension') 
        AND schema_name NOT LIKE 'pg_%temp_%'
        AND NOT EXISTS (
                SELECT 
                    1
                FROM pg_extension
                WHERE 
                    extnamespace = (
                        SELECT 
                            oid 
                        FROM pg_namespace 
                        WHERE 
                            nspname = schema_name
                    )
        )
ORDER BY 
    schema_name;
 schema_name 
-------------
 tiger_data
(1 row)
```

Tests error:
```
sql/migrate: connected database is not clean: found schema "tiger_data". baseline version or allow-dirty is required
```

The way I see, there is no perfect way to ensure a schema/table is used (only) by the extension in some cases (like this one), so while this PR make some progress, it's not the complete solution and tests are failing.

Maybe the best solution would be embrace that sometimes we can not exclude the schema/tables created by extensions automatically and let the user exclude it as required.

So that means adding a `exclude` to the `atlas migrate lint` and `atlas migrate diff` commands and to make the tests pass, we would need to allow providing `InspectOptions` to [`Driver.CheckClean`](https://github.com/ariga/atlas/blob/30a2c72536f86913891ba2ce642acee7ac5dbfc9/sql/postgres/driver.go#L255)/[`Driver.Snapshot`](https://github.com/ariga/atlas/blob/30a2c72536f86913891ba2ce642acee7ac5dbfc9/sql/postgres/driver.go#L203) and skip the PostGIS schemas.

Alternatively we could create a custom PostGIS image without the `postgis_topology` and `postgis_tiger_geocoder` extensions and handle them at a later stage.


--- 

Comments or ideas are appreciated.

plutino: @crossworth @a8m  I took a look of the CI failures, and they are all caused by the extra schemas added by `postgis`, namely `tiger`, `tiger_data`, and `topology`.

I'm not familiar with the code base enough to have the best way to fix this, but this change fixes the tests:

```diff
--- a/sql/postgres/inspect.go
+++ b/sql/postgres/inspect.go
@@ -1073,7 +1073,7 @@ const (
        // Query to list database schemas.
-       schemasQuery = "SELECT schema_name FROM information_schema.schemata WHERE schema_name NOT IN ('information_schema', 'pg_catalog', 'pg_toast', 'crdb_internal', 'pg_extension') AND schema_name NOT LIKE 'pg_%temp_%' ORDER BY schema_name"
+       schemasQuery = "SELECT schema_name FROM information_schema.schemata WHERE schema_name NOT IN ('information_schema', 'pg_catalog', 'pg_toast', 'crdb_internal', 'pg_extension', 'tiger', 'tiger_data', 'topology') AND schema_name NOT LIKE 'pg_%temp_%' ORDER BY schema_name"
 ```
crossworth: Hello @plutino, thanks for giving a look on the issue.

While this change fix this problem, it's not the best solution, this would limit the names `tiger`, `tiger_data` and `topology`, that are not that "unique" and can create hard to understood bugs. This wont help with future extensions either.

To make the CI pass, we only need to provide a way to exclude this schemas as needed (vs hardcoded on the query), Atlas already has support for it in a few places (the `exclude` flag) or use a lean PostGIS  Docker image.

In my opinion the best solution is to add support for the `exclude` flag on `atlas migrate lint` and `atlas migrate diff` commands.

I plan to take a look at it this week and see if I can make progress, but if you would like to give a try, please let me know.
plutino: > 



> Hello @plutino, thanks for giving a look on the issue.
> 
> While this change fix this problem, it's not the best solution, this would limit the names `tiger`, `tiger_data` and `topology`, that are not that "unique" and can create hard to understood bugs. This wont help with future extensions either.
> 
> To make the CI pass, we only need to provide a way to exclude this schemas as needed (vs hardcoded on the query), Atlas already has support for it in a few places (the `exclude` flag) or use a lean PostGIS Docker image like you mentioned.
> 
> In my opinion the best solution is to add support for the `exclude` flag on `atlas migrate lint` and `atlas migrate diff` commands.
> 
> I plan to take a look at it this week and see if I can make progress, but if you would like to give a try, please let me know.

Sounds good.  Thanks @crossworth .  This will help us a lot!
crossworth: Alright, I tried my ideia of adding support for the [exclude flag](https://github.com/crossworth/atlas/commits/postgis-tests), but seems like I had only two ways of doing (in a backward compatible way):
- Adding a optional argument to `Driver.CheckClean` and `Driver.Snapshot` (to allow controlling the `InspectSchema` and `InspectRealm`.
- Adding the options to the driver itself.

The first option seems to me the best way, since it creates a better API, but it requires changes to a lot of packages until we get to the packages used by the `cmd/atlas`, the second option seemed easier to me, but feels a lot more like hacked together.

After thinking a bit, I remembered that my goal with this PR was quite simple, making the CI pass with PostGIS, with as little code change as possible, so I took a step back and implemented a different solution, a little bit hacky as well, but at least only in the tests part.

We drop the [unused schemas](https://github.com/ariga/atlas/pull/1601/files#diff-74d23195890a3051b389f9273e4f9b90b58515b7ea01f0d1cde0a11d057edcf7R63-R74), would be better to have a lean docker image, but maintaining a custom docker image could not be worth for this case.

This means that the problem related to the `postgis_tiger_geocoder` extension still exists, we could always go back to it when/if the needs arise.

Now, with this change both the declarative and versioned migration strategies works, without the need of the `exclude` flag:

**NOTE:** Each test was done with a empty database running on docker:
```sh
docker run --rm -e POSTGRES_PASSWORD=pass -e POSTGRES_DB=test -p 25432:5432 postgis/postgis:latest
```

### Declarative

#### `atlas schema inspect`

```sh 
# note the search_path
atlas schema inspect --url "postgres://postgres:pass@localhost:25432/test?search_path=public&sslmode=disable"
```

<details>
<summary>atlas version v0.10.2-30a2c72-canary</summary>

```hcl
table "spatial_ref_sys" {
  schema = schema.public
  column "srid" {
    null = false
    type = integer
  }
  column "auth_name" {
    null = true
    type = character_varying(256)
  }
  column "auth_srid" {
    null = true
    type = integer
  }
  column "srtext" {
    null = true
    type = character_varying(2048)
  }
  column "proj4text" {
    null = true
    type = character_varying(2048)
  }
  primary_key {
    columns = [column.srid]
  }
  check "spatial_ref_sys_srid_check" {
    expr = "((srid > 0) AND (srid <= 998999))"
  }
}
schema "public" {
}
```
</details>

<details>
<summary>This PR</summary>

```hcl
schema "public" {
}
```
</details>

### Versioned

#### `atlas migrate diff`

```sh
# note the search_path
atlas migrate diff --dir "file://migrations" --to "postgres://postgres:pass@localhost:25432/test?search_path=public&sslmode=disable" --dev-url "docker://postgres/15"
```

<details>
<summary>atlas version v0.10.2-30a2c72-canary</summary>

File `migrations/20230510042843.sql` created.

```sql
-- Create "spatial_ref_sys" table
CREATE TABLE "public"."spatial_ref_sys" ("srid" integer NOT NULL, "auth_name" character varying(256) NULL, "auth_srid" integer NULL, "srtext" character varying(2048) NULL, "proj4text" character varying(2048) NULL, PRIMARY KEY ("srid"), CONSTRAINT "spatial_ref_sys_srid_check" CHECK ((srid > 0) AND (srid <= 998999)));

```
</details>

<details>
<summary>This PR</summary>

No file is created.

```text
The migration directory is synced with the desired state, no changes to be made
```
</details>

#### `atlas migrate apply`

```sh
# note the search_path
atlas migrate apply --dir "file://migrations" --url "postgres://postgres:pass@localhost:25432/test?search_path=public&sslmode=disable"
```

<details>
<summary>atlas version v0.10.2-30a2c72-canary</summary>

```text
Error: sql/migrate: connected database is not clean: found table "atlas_schema_revisions" in schema "public". baseline version or allow-dirty is required
```
</details>

<details>
<summary>This PR</summary>

```text
Migrating to version 20230510043225 (1 migrations in total):

  -- migrating version 20230510043225
    -> create table users (
id integer not null
);
  -- ok (6.722041ms)

  -------------------------
  -- 154.249ms
  -- 1 migrations
  -- 1 sql statements

```
</details>

#### `atlas migrate lint`
We are required to use a dev database with PostGIS extension, so no docker support 😔.

```sh
atlas migrate lint --dir "file://migrations" --dev-url "postgres://postgres:pass@localhost:25432/test?search_path=public&sslmode=disable" --latest 1
```

<details>
<summary>atlas version v0.10.2-30a2c72-canary</summary>

```text
Error: taking database snapshot: sql/migrate: connected database is not clean: found table "spatial_ref_sys" in connected schema
```
</details>

<details>
<summary>This PR</summary>

No output
</details>


Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 2ac52944..3cd6b83c 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -258,6 +258,30 @@ jobs:
         working-directory: internal/integration
         run: go test -race -count=2 -v -run="MySQL" -version="maria103" -timeout 15m ./...
 
+  integration-postgres-ext-postgis:
+    runs-on: ubuntu-latest
+    services:
+      postgres-ext-postgis:
+        image: postgis/postgis:latest
+        env:
+          POSTGRES_DB: test
+          POSTGRES_PASSWORD: pass
+        ports:
+          - 5429:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    steps:
+      - uses: actions/checkout@v2.3.4
+      - uses: actions/setup-go@v4
+        with:
+          go-version: '1.20'
+      - name: Run integration tests for postgres-ext-postgis
+        working-directory: internal/integration
+        run: go test -race -count=2 -v -run="Postgres" -version="postgres-ext-postgis" -timeout 15m ./...
+
   integration-postgres10:
     runs-on: ubuntu-latest
     services:
diff --git a/cmd/atlas/internal/cloudapi/client.go b/cmd/atlas/internal/cloudapi/client.go
index 2e7f26a6..219982cc 100644
--- a/cmd/atlas/internal/cloudapi/client.go
+++ b/cmd/atlas/internal/cloudapi/client.go
@@ -12,6 +12,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"net/url"
 	"time"
 
 	"ariga.io/atlas/sql/migrate"
@@ -77,7 +78,18 @@ func (c *Client) Dir(ctx context.Context, input DirInput) (migrate.Dir, error) {
 }
 
 type (
-	// ReportMigrationInput represents an input type for a reporting migration deployments.
+	// ReportMigrationSetInput represents the input type for reporting a set of migration deployments.
+	ReportMigrationSetInput struct {
+		ID        string                 `json:"id"`
+		Planned   int                    `json:"planned"`
+		StartTime time.Time              `json:"startTime"`
+		EndTime   time.Time              `json:"endTime"`
+		Error     *string                `json:"error,omitempty"`
+		Log       []ReportStep           `json:"log,omitempty"`
+		Completed []ReportMigrationInput `json:"completed,omitempty"`
+	}
+
+	// ReportMigrationInput represents an input type for reporting a migration deployments.
 	ReportMigrationInput struct {
 		ProjectName    string              `json:"projectName"`
 		EnvName        string              `json:"envName"`
@@ -91,6 +103,7 @@ type (
 		CurrentVersion string              `json:"currentVersion"`
 		Error          *string             `json:"error,omitempty"`
 		Files          []DeployedFileInput `json:"files"`
+		Log            string              `json:"log"`
 	}
 
 	// DeployedTargetInput represents the input type for a deployed target.
@@ -116,8 +129,45 @@ type (
 		Stmt string `json:"stmt"`
 		Text string `json:"text"`
 	}
+
+	// ReportStep is top-level step in a report.
+	ReportStep struct {
+		Text      string          `json:"text"`
+		StartTime time.Time       `json:"startTime"`
+		EndTime   time.Time       `json:"endTime"`
+		Error     bool            `json:"error,omitempty"`
+		Log       []ReportStepLog `json:"log,omitempty"`
+	}
+	// ReportStepLog is a log entry in a step.
+	ReportStepLog struct {
+		Text     string          `json:"text,omitempty"`
+		Children []ReportStepLog `json:"children,omitempty"`
+	}
 )
 
+// ReportMigrationSet reports a set of migration deployments to the Atlas Cloud API.
+func (c *Client) ReportMigrationSet(ctx context.Context, input ReportMigrationSetInput) error {
+	var (
+		payload struct {
+			ReportMigrationSet struct {
+				Success bool `json:"success"`
+			} `json:"reportMigrationSet"`
+		}
+		query = `
+		mutation ReportMigrationSet($input: ReportMigrationSetInput!) {
+		   reportMigrationSet(input: $input) {
+		     success
+		   }
+		}`
+		vars = struct {
+			Input ReportMigrationSetInput `json:"input"`
+		}{
+			Input: input,
+		}
+	)
+	return c.post(ctx, query, vars, &payload)
+}
+
 // ReportMigration reports a migration deployment to the Atlas Cloud API.
 func (c *Client) ReportMigration(ctx context.Context, input ReportMigrationInput) error {
 	var (
@@ -192,3 +242,12 @@ func (r *roundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
 	req.Header.Set("Content-Type", "application/json")
 	return http.DefaultTransport.RoundTrip(req)
 }
+
+// RedactedURL returns a URL string with the userinfo redacted.
+func RedactedURL(s string) (string, error) {
+	u, err := url.Parse(s)
+	if err != nil {
+		return "", err
+	}
+	return u.Redacted(), nil
+}
diff --git a/cmd/atlas/internal/cmdapi/migrate.go b/cmd/atlas/internal/cmdapi/migrate.go
index 8e24963e..95eddf5e 100644
--- a/cmd/atlas/internal/cmdapi/migrate.go
+++ b/cmd/atlas/internal/cmdapi/migrate.go
@@ -116,16 +116,23 @@ If run with the "--dry-run" flag, atlas will not execute any SQL.`,
   atlas migrate apply --dry-run --env dev 1`,
 			Args: cobra.MaximumNArgs(1),
 			RunE: func(cmd *cobra.Command, args []string) (err error) {
-				if GlobalFlags.SelectedEnv == "" {
+				switch {
+				case GlobalFlags.SelectedEnv == "":
+					if err := migrateFlagsFromConfig(cmd); err != nil {
+						return err
+					}
 					return migrateApplyRun(cmd, args, flags, &MigrateReport{}) // nop reporter
+				default:
+					project, envs, err := EnvByName(GlobalFlags.SelectedEnv, WithInput(GlobalFlags.Vars))
+					if err != nil {
+						return err
+					}
+					set := NewReportProvider(project, envs)
+					defer set.Flush(cmd)
+					return cmdEnvsRun(envs, setMigrateEnvFlags, cmd, func(env *Env) error {
+						return migrateApplyRun(cmd, args, flags, set.ReportFor(flags, env))
+					})
 				}
-				set := NewMigrateReportSet()
-				defer func() {
-					err = errors.Join(err, set.Flush(cmd.Context()))
-				}()
-				return cmdEnvsRun(setMigrateEnvFlags, cmd, func(env *Env) error {
-					return migrateApplyRun(cmd, args, flags, set.ReportFor(env))
-				})
 			},
 		}
 	)
@@ -281,28 +288,83 @@ type (
 	}
 	// MigrateReportSet is a set of reports.
 	MigrateReportSet struct {
-		id      string
-		reports []struct {
-			e *Env
-			r *cloudapi.ReportMigrationInput
-		}
+		cloudapi.ReportMigrationSetInput
+		client *cloudapi.Client
+		done   int // number of done migrations
 	}
 )
 
-// NewMigrateReportSet returns a new MigrateReportSet with a unique id set.
-func NewMigrateReportSet() *MigrateReportSet {
-	return &MigrateReportSet{id: uuid.NewString()}
+// NewReportProvider returns a new ReporterProvider.
+func NewReportProvider(project *Project, envs []*Env) *MigrateReportSet {
+	s := &MigrateReportSet{
+		client: project.cfg.Client,
+		ReportMigrationSetInput: cloudapi.ReportMigrationSetInput{
+			ID:        uuid.NewString(),
+			StartTime: time.Now(),
+			Planned:   len(envs),
+		},
+	}
+	s.Step("Start migration for %d targets", len(envs))
+	for _, e := range envs {
+		s.StepLog(s.RedactedURL(e.URL))
+	}
+	return s
+}
+
+// RedactedURL returns the redacted URL of the given environment at index i.
+func (*MigrateReportSet) RedactedURL(u string) string {
+	u, err := cloudapi.RedactedURL(u)
+	if err != nil {
+		return fmt.Sprintf("Error: redacting URL: %v", err)
+	}
+	return u
+}
+
+// Step starts a new reporting step.
+func (s *MigrateReportSet) Step(format string, args ...interface{}) {
+	if len(s.Log) > 0 && s.Log[len(s.Log)-1].EndTime.IsZero() {
+		s.Log[len(s.Log)-1].EndTime = time.Now()
+	}
+	s.Log = append(s.Log, cloudapi.ReportStep{
+		StartTime: time.Now(),
+		Text:      fmt.Sprintf(format, args...),
+	})
+}
+
+// StepLog logs a line to the current reporting step.
+func (s *MigrateReportSet) StepLog(format string, args ...interface{}) {
+	if len(s.Log) == 0 {
+		s.Step("Unnamed step") // Unexpected.
+	}
+	s.Log[len(s.Log)-1].Log = append(s.Log[len(s.Log)-1].Log, cloudapi.ReportStepLog{
+		Text: fmt.Sprintf(format, args...),
+	})
+}
+
+// StepLogError logs a line to the current reporting step.
+func (s *MigrateReportSet) StepLogError(text string) {
+	if !strings.HasPrefix(text, "Error") {
+		text = "Error: " + text
+	}
+	s.StepLog(text)
+	s.Error = &text
+	s.Log[len(s.Log)-1].Error = true
 }
 
 // ReportFor returns a new MigrateReport for the given environment.
-func (s *MigrateReportSet) ReportFor(e *Env) *MigrateReport {
+func (s *MigrateReportSet) ReportFor(flags migrateApplyFlags, e *Env) *MigrateReport {
+	s.Step("Run migration: %d", s.done+1)
+	s.StepLog("Target URL: %s", s.RedactedURL(e.URL))
+	s.StepLog("Migration directory: %s", s.RedactedURL(flags.dirURL))
 	return &MigrateReport{
 		env: e,
 		done: func(r *cloudapi.ReportMigrationInput) {
-			s.reports = append(s.reports, struct {
-				e *Env
-				r *cloudapi.ReportMigrationInput
-			}{e: e, r: r})
+			s.done++
+			s.Log[len(s.Log)-1].EndTime = time.Now()
+			if r.Error != nil && *r.Error != "" {
+				s.StepLogError(*r.Error)
+			}
+			s.Completed = append(s.Completed, *r)
 		},
 	}
 }
@@ -310,15 +372,31 @@ func (s *MigrateReportSet) ReportFor(e *Env) *MigrateReport {
 // Flush report the migration deployment to the cloud.
 // The current implementation is simplistic and sends each
 // report separately without marking them as part of a group.
-func (s *MigrateReportSet) Flush(ctx context.Context) error {
-	var errs []error
-	for _, r := range s.reports {
-		client := r.e.cfg.Client
-		if err := client.ReportMigration(ctx, *r.r); err != nil {
-			errs = append(errs, err)
+//
+// Note that reporting errors are logged, but not cause Atlas to fail.
+func (s *MigrateReportSet) Flush(cmd *cobra.Command) {
+	var err error
+	switch {
+	// Skip reporting if set is empty,
+	// or there is no cloud connectivity.
+	case s.Planned == 0, s.client == nil:
+		return
+	// Single migration.
+	case s.Planned == 1 && len(s.Completed) == 1:
+		err = s.client.ReportMigration(cmd.Context(), s.Completed[0])
+	// Multi environment migration (e.g., multi-tenancy).
+	case s.Planned > 1:
+		s.EndTime = time.Now()
+		err = s.client.ReportMigrationSet(cmd.Context(), s.ReportMigrationSetInput)
+	}
+	if err != nil {
+		txt := fmt.Sprintf("Error: %s", strings.TrimRight(err.Error(), "\n"))
+		// Ensure errors are printed in new lines.
+		if cmd.Flags().Changed(flagFormat) {
+			txt = "\n" + txt
 		}
+		cmd.PrintErrln(txt)
 	}
-	return errors.Join(errs...)
 }
 
 // Init the report if the necessary dependencies.
@@ -341,11 +419,14 @@ func (r *MigrateReport) RecordTargetID(ctx context.Context) error {
 
 // Done closes and flushes this report.
 func (r *MigrateReport) Done(cmd *cobra.Command, flags migrateApplyFlags) error {
-	err := logApply(cmd, flags, r.log)
 	if !r.CloudEnabled() {
-		return err
+		return logApply(cmd, cmd.OutOrStdout(), flags, r.log)
 	}
-	var ver string
+	var (
+		ver  string
+		clog bytes.Buffer
+		err  = logApply(cmd, io.MultiWriter(cmd.OutOrStdout(), &clog), flags, r.log)
+	)
 	switch rev, err1 := r.rrw.CurrentRevision(cmd.Context()); {
 	case ent.IsNotFound(err1):
 	case err1 != nil:
@@ -389,6 +470,7 @@ func (r *MigrateReport) Done(cmd *cobra.Command, flags migrateApplyFlags) error
 			}
 			return files
 		}(),
+		Log: clog.String(),
 	})
 	return err
 }
@@ -398,7 +480,7 @@ func (r *MigrateReport) CloudEnabled() bool {
 	return r.env != nil && r.env.cfg != nil && r.env.cfg.Client != nil && r.env.cfg.Project != ""
 }
 
-func logApply(cmd *cobra.Command, flags migrateApplyFlags, r *cmdlog.MigrateApply) error {
+func logApply(cmd *cobra.Command, w io.Writer, flags migrateApplyFlags, r *cmdlog.MigrateApply) error {
 	var (
 		err error
 		f   = cmdlog.MigrateApplyTemplate
@@ -409,7 +491,7 @@ func logApply(cmd *cobra.Command, flags migrateApplyFlags, r *cmdlog.MigrateAppl
 			return fmt.Errorf("parse format: %w", err)
 		}
 	}
-	if err = f.Execute(cmd.OutOrStdout(), r); err != nil {
+	if err = f.Execute(w, r); err != nil {
 		return fmt.Errorf("execute log template: %w", err)
 	}
 	// In case a custom logging was configured, avoid reporting errors twice.
@@ -446,7 +528,7 @@ directory state to the desired schema. The desired state can be another connecte
   atlas migrate diff --env dev --format '{{ sql . "  " }}'`,
 			Args: cobra.MaximumNArgs(1),
 			PreRunE: func(cmd *cobra.Command, args []string) error {
-				if err := migrateFlagsFromEnv(cmd); err != nil {
+				if err := migrateFlagsFromConfig(cmd); err != nil {
 					return err
 				}
 				if err := dirFormatBC(flags.dirFormat, &flags.dirURL); err != nil {
@@ -455,7 +537,7 @@ directory state to the desired schema. The desired state can be another connecte
 				return checkDir(cmd, flags.dirURL, true)
 			},
 			RunE: func(cmd *cobra.Command, args []string) error {
-				env, err := selectEnv(GlobalFlags.SelectedEnv)
+				env, err := selectEnv(cmd)
 				if err != nil {
 					return err
 				}
@@ -642,7 +724,7 @@ func migrateHashCmd() *cobra.Command {
 This command should be used whenever a manual change in the migration directory was made.`,
 			Example: `  atlas migrate hash`,
 			PreRunE: func(cmd *cobra.Command, args []string) error {
-				if err := migrateFlagsFromEnv(cmd); err != nil {
+				if err := migrateFlagsFromConfig(cmd); err != nil {
 					return err
 				}
 				return dirFormatBC(flags.dirFormat, &flags.dirURL)
@@ -680,7 +762,7 @@ func migrateImportCmd() *cobra.Command {
 			// Validate the source directory. Consider a directory with no sum file
 			// valid, since it might be an import from an existing project.
 			PreRunE: func(cmd *cobra.Command, _ []string) error {
-				if err := migrateFlagsFromEnv(cmd); err != nil {
+				if err := migrateFlagsFromConfig(cmd); err != nil {
 					return err
 				}
 				if err := dirFormatBC(flags.dirFormat, &flags.fromURL); err != nil {
@@ -806,7 +888,7 @@ func migrateLintCmd() *cobra.Command {
   atlas migrate lint --dir file:///path/to/migration/directory --dev-url mysql://root:pass@localhost:3306 --git-base master
   atlas migrate lint --dir file:///path/to/migration/directory --dev-url mysql://root:pass@localhost:3306 --format '{{ json .Files }}'`,
 			PreRunE: func(cmd *cobra.Command, args []string) error {
-				if err := migrateFlagsFromEnv(cmd); err != nil {
+				if err := migrateFlagsFromConfig(cmd); err != nil {
 					return err
 				}
 				return dirFormatBC(flags.dirFormat, &flags.dirURL)
@@ -866,7 +948,7 @@ func migrateLintRun(cmd *cobra.Command, _ []string, flags migrateLintFlags) erro
 			return fmt.Errorf("parse format: %w", err)
 		}
 	}
-	env, err := selectEnv(GlobalFlags.SelectedEnv)
+	env, err := selectEnv(cmd)
 	if err != nil {
 		return err
 	}
@@ -908,7 +990,7 @@ func migrateNewCmd() *cobra.Command {
 			Example: `  atlas migrate new my-new-migration`,
 			Args:    cobra.MaximumNArgs(1),
 			PreRunE: func(cmd *cobra.Command, _ []string) error {
-				if err := migrateFlagsFromEnv(cmd); err != nil {
+				if err := migrateFlagsFromConfig(cmd); err != nil {
 					return err
 				}
 				if err := dirFormatBC(flags.dirFormat, &flags.dirURL); err != nil {
@@ -970,7 +1052,7 @@ to be applied. This command is usually used after manually making changes to the
   atlas migrate set --env local
   atlas migrate set 1.2.4 --url mysql://user:pass@localhost:3306/my_db --revision-schema my_revisions`,
 			PreRunE: func(cmd *cobra.Command, _ []string) error {
-				if err := migrateFlagsFromEnv(cmd); err != nil {
+				if err := migrateFlagsFromConfig(cmd); err != nil {
 					return err
 				}
 				if err := dirFormatBC(flags.dirFormat, &flags.dirURL); err != nil {
@@ -1165,7 +1247,7 @@ func migrateStatusCmd() *cobra.Command {
 			Example: `  atlas migrate status --url mysql://user:pass@localhost:3306/
   atlas migrate status --url mysql://user:pass@localhost:3306/ --dir file:///path/to/migration/directory`,
 			PreRunE: func(cmd *cobra.Command, _ []string) error {
-				if err := migrateFlagsFromEnv(cmd); err != nil {
+				if err := migrateFlagsFromConfig(cmd); err != nil {
 					return err
 				}
 				if err := dirFormatBC(flags.dirFormat, &flags.dirURL); err != nil {
@@ -1239,7 +1321,7 @@ files are executed on the connected database in order to validate SQL semantics.
   atlas migrate validate --dir file:///path/to/migration/directory --dev-url mysql://user:pass@localhost:3306/dev
   atlas migrate validate --env dev`,
 			PreRunE: func(cmd *cobra.Command, _ []string) error {
-				if err := migrateFlagsFromEnv(cmd); err != nil {
+				if err := migrateFlagsFromConfig(cmd); err != nil {
 					return err
 				}
 				if err := dirFormatBC(flags.dirFormat, &flags.dirURL); err != nil {
@@ -1691,8 +1773,8 @@ func formatter(u *url.URL) (migrate.Formatter, error) {
 	}
 }
 
-func migrateFlagsFromEnv(cmd *cobra.Command) error {
-	env, err := selectEnv(GlobalFlags.SelectedEnv)
+func migrateFlagsFromConfig(cmd *cobra.Command) error {
+	env, err := selectEnv(cmd)
 	if err != nil {
 		return err
 	}
@@ -1789,14 +1871,11 @@ func isURL(s string) bool {
 
 // cmdEnvsRun executes a given command on each of the configured environment.
 func cmdEnvsRun(
+	envs []*Env,
 	setFlags func(*cobra.Command, *Env) error,
 	cmd *cobra.Command,
 	runCmd func(*Env) error,
 ) error {
-	envs, err := LoadEnv(GlobalFlags.SelectedEnv, WithInput(GlobalFlags.Vars))
-	if err != nil {
-		return err
-	}
 	var (
 		w     bytes.Buffer
 		out   = cmd.OutOrStdout()
diff --git a/cmd/atlas/internal/cmdapi/project.go b/cmd/atlas/internal/cmdapi/project.go
index a41a06dd..fd96ad2b 100644
--- a/cmd/atlas/internal/cmdapi/project.go
+++ b/cmd/atlas/internal/cmdapi/project.go
@@ -15,11 +15,12 @@ import (
 	"ariga.io/atlas/schemahcl"
 	"ariga.io/atlas/sql/schema"
 
+	"github.com/hashicorp/hcl/v2"
+	"github.com/hashicorp/hcl/v2/hclparse"
+	"github.com/hashicorp/hcl/v2/hclsyntax"
 	"github.com/zclconf/go-cty/cty"
 )
 
-const defaultConfigPath = "file://atlas.hcl"
-
 type loadConfig struct {
 	inputValues map[string]cty.Value
 }
@@ -35,11 +36,12 @@ func WithInput(values map[string]cty.Value) LoadOption {
 }
 
 type (
-	// Project represents an atlas.hcl project file.
+	// Project represents an atlas.hcl project config file.
 	Project struct {
 		Envs []*Env `spec:"env"`  // List of environments
 		Lint *Lint  `spec:"lint"` // Optional global lint policy
 		Diff *Diff  `spec:"diff"` // Optional global diff policy
+		cfg  *cmdext.AtlasConfig
 	}
 
 	// Env represents an Atlas environment.
@@ -329,81 +331,97 @@ func (e *Env) asMap() (map[string]string, error) {
 	return m, nil
 }
 
-// LoadEnv reads the project file in path, and loads
-// the environment instances with the provided name.
-func LoadEnv(name string, opts ...LoadOption) ([]*Env, error) {
-	cfg := &loadConfig{}
-	for _, f := range opts {
-		f(cfg)
-	}
+// EnvByName parses and returns the project configuration with selected environments.
+func EnvByName(name string, opts ...LoadOption) (*Project, []*Env, error) {
 	u, err := url.Parse(GlobalFlags.ConfigURL)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	if u.Scheme != "file" {
-		return nil, fmt.Errorf("unsupported project file driver %q", u.Scheme)
+		return nil, nil, fmt.Errorf("unsupported project file driver %q", u.Scheme)
 	}
 	path := filepath.Join(u.Host, u.Path)
 	if _, err := os.Stat(path); err != nil {
 		if os.IsNotExist(err) {
 			err = fmt.Errorf("project file %q was not found: %w", path, err)
 		}
-		return nil, err
+		return nil, nil, err
 	}
-	project, err := parseConfig(path, name, cfg.inputValues)
+	project, err := parseConfig(path, name, opts...)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	if err := project.Lint.remainedLog(); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	envs := make(map[string][]*Env)
 	for _, e := range project.Envs {
 		if e.Name == "" {
-			return nil, fmt.Errorf("all envs must have names on file %q", path)
+			return nil, nil, fmt.Errorf("all envs must have names on file %q", path)
 		}
 		if _, err := e.Sources(); err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		if e.Migration == nil {
 			e.Migration = &Migration{}
 		}
 		if err := e.remainedLog(); err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		e.Diff = e.Diff.Extend(project.Diff)
 		e.Lint = e.Lint.Extend(project.Lint)
 		if err := e.Lint.remainedLog(); err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		envs[e.Name] = append(envs[e.Name], e)
 	}
-	selected, ok := envs[name]
-	if !ok {
-		return nil, fmt.Errorf("env %q not defined in project file", name)
+	switch {
+	case name == "":
+		// If no env was selected,
+		// return only the project.
+		return project, nil, nil
+	case len(envs[name]) == 0:
+		return nil, nil, fmt.Errorf("env %q not defined in project file", name)
+	default:
+		return project, envs[name], nil
 	}
-	return selected, nil
 }
 
-func parseConfig(path, env string, values map[string]cty.Value) (*Project, error) {
+const (
+	blockEnv          = "env"
+	refAtlas          = "atlas"
+	defaultConfigPath = "file://atlas.hcl"
+)
+
+func parseConfig(path, env string, opts ...LoadOption) (*Project, error) {
+	loadCfg := &loadConfig{}
+	for _, f := range opts {
+		f(loadCfg)
+	}
+	pr, err := partialParse(path, env)
+	if err != nil {
+		return nil, err
+	}
 	cfg := &cmdext.AtlasConfig{}
-	opts := append(
-		cmdext.DataSources,
-		cfg.InitBlock(),
-		schemahcl.WithScopedEnums(
-			"env.migration.format",
-			formatAtlas, formatFlyway,
-			formatLiquibase, formatGoose,
-			formatGolangMigrate,
-		),
-		schemahcl.WithVariables(map[string]cty.Value{
-			"atlas": cty.ObjectVal(map[string]cty.Value{
-				"env": cty.StringVal(env),
+	state := schemahcl.New(
+		append(
+			cmdext.DataSources,
+			cfg.InitBlock(),
+			schemahcl.WithScopedEnums(
+				"env.migration.format",
+				formatAtlas, formatFlyway,
+				formatLiquibase, formatGoose,
+				formatGolangMigrate,
+			),
+			schemahcl.WithVariables(map[string]cty.Value{
+				refAtlas: cty.ObjectVal(map[string]cty.Value{
+					blockEnv: cty.StringVal(env),
+				}),
 			}),
-		}),
+		)...,
 	)
-	p := &Project{Lint: &Lint{}, Diff: &Diff{}}
-	if err := schemahcl.New(opts...).EvalFiles([]string{path}, p, values); err != nil {
+	p := &Project{Lint: &Lint{}, Diff: &Diff{}, cfg: cfg}
+	if err := state.Eval(pr, p, loadCfg.inputValues); err != nil {
 		return nil, err
 	}
 	for _, e := range p.Envs {
@@ -413,5 +431,46 @@ func parseConfig(path, env string, values map[string]cty.Value) (*Project, error
 }
 
 func init() {
-	schemahcl.Register("env", &Env{})
+	schemahcl.Register(blockEnv, &Env{})
+}
+
+func partialParse(path, env string) (*hclparse.Parser, error) {
+	parser := hclparse.NewParser()
+	fi, err := parser.ParseHCLFile(path)
+	if err != nil {
+		return nil, err
+	}
+	var used, datasrc []*hclsyntax.Block
+	for _, b := range fi.Body.(*hclsyntax.Body).Blocks {
+		switch b.Type {
+		case schemahcl.BlockData:
+			datasrc = append(datasrc, b)
+		case blockEnv:
+			switch n := len(b.Labels); {
+			// No env was selected.
+			case env == "" && n == 0:
+			// Exact env was selected.
+			case n == 1 && b.Labels[0] == env:
+				used = append(used, b)
+			// Dynamic env selection.
+			case n == 0 && b.Body != nil && b.Body.Attributes[schemahcl.AttrName] != nil:
+				t, ok := b.Body.Attributes[schemahcl.AttrName].Expr.(*hclsyntax.ScopeTraversalExpr)
+				if ok && len(t.Traversal) == 2 && t.Traversal.RootName() == refAtlas && t.Traversal[1].(hcl.TraverseAttr).Name == blockEnv {
+					used = append(used, b)
+				}
+			}
+		default:
+			used = append(used, b)
+		}
+	}
+	body := &hclsyntax.Body{
+		Blocks:     used,
+		Attributes: fi.Body.(*hclsyntax.Body).Attributes,
+	}
+	// If no attributes, locals, variables, or blocks reference data sources, we skip
+	// loading. Otherwise, body stays unchanged as data sources may depend on others.
+	if len(schemahcl.DynamicRefs(body).Data) == 0 {
+		fi.Body = body
+	}
+	return parser, nil
 }
diff --git a/cmd/atlas/internal/cmdapi/schema.go b/cmd/atlas/internal/cmdapi/schema.go
index cafe55b8..fc0bf6d3 100644
--- a/cmd/atlas/internal/cmdapi/schema.go
+++ b/cmd/atlas/internal/cmdapi/schema.go
@@ -89,12 +89,22 @@ migration.`,
   atlas schema apply --url "postgres://user:pass@host:port/dbname?sslmode=disable" --to file://schema.hcl
   atlas schema apply -u "sqlite://file:ex1.db?_fk=1" --to file://schema.hcl`,
 			RunE: func(cmd *cobra.Command, args []string) error {
-				if GlobalFlags.SelectedEnv == "" {
-					return schemaApplyRun(cmd, flags, nil)
+				switch {
+				case GlobalFlags.SelectedEnv == "":
+					env, err := selectEnv(cmd)
+					if err != nil {
+						return err
+					}
+					return schemaApplyRun(cmd, flags, env)
+				default:
+					_, envs, err := EnvByName(GlobalFlags.SelectedEnv, WithInput(GlobalFlags.Vars))
+					if err != nil {
+						return err
+					}
+					return cmdEnvsRun(envs, setSchemaEnvFlags, cmd, func(e *Env) error {
+						return schemaApplyRun(cmd, flags, e)
+					})
 				}
-				return cmdEnvsRun(setSchemaEnvFlags, cmd, func(e *Env) error {
-					return schemaApplyRun(cmd, flags, e)
-				})
 			},
 		}
 	)
@@ -238,7 +248,7 @@ As a safety feature, 'atlas schema clean' will ask for confirmation before attem
 			Example: `  atlas schema clean -u mysql://user:pass@localhost:3306/dbname
   atlas schema clean -u mysql://user:pass@localhost:3306/`,
 			PreRunE: func(cmd *cobra.Command, _ []string) error {
-				return schemaFlagsFromEnv(cmd)
+				return schemaFlagsFromConfig(cmd)
 			},
 			RunE: func(cmd *cobra.Command, args []string) error {
 				return schemaCleanRun(cmd, args, flags)
@@ -320,10 +330,10 @@ The database states can be read from a connected database, an HCL project or a m
   atlas schema diff --from mysql://user:pass@localhost:3306 --to file://schema_1.hcl --to file://schema_2.hcl
   atlas schema diff --from mysql://user:pass@localhost:3306 --to file://migrations --format '{{ sql . "  " }}'`,
 			PreRunE: func(cmd *cobra.Command, _ []string) error {
-				return schemaFlagsFromEnv(cmd)
+				return schemaFlagsFromConfig(cmd)
 			},
 			RunE: func(cmd *cobra.Command, args []string) error {
-				env, err := selectEnv(GlobalFlags.SelectedEnv)
+				env, err := selectEnv(cmd)
 				if err != nil {
 					return err
 				}
@@ -433,7 +443,7 @@ flag.
   atlas schema inspect --url "postgres://user:pass@host:port/dbname?sslmode=disable"
   atlas schema inspect -u "sqlite://file:ex1.db?_fk=1"`,
 			PreRunE: func(cmd *cobra.Command, _ []string) error {
-				return schemaFlagsFromEnv(cmd)
+				return schemaFlagsFromConfig(cmd)
 			},
 			RunE: func(cmd *cobra.Command, args []string) error {
 				return schemaInspectRun(cmd, args, flags)
@@ -538,26 +548,36 @@ func schemaFmtRun(cmd *cobra.Command, args []string) error {
 // selectEnv returns the Env from the current project file based on the selected
 // argument. If selected is "", or no project file exists in the current directory
 // a zero-value Env is returned.
-func selectEnv(name string) (*Env, error) {
-	env := &Env{
-		Lint:      &Lint{},
-		Migration: &Migration{},
-	}
-	if name == "" {
-		return env, nil
-	}
-	envs, err := LoadEnv(name, WithInput(GlobalFlags.Vars))
-	if err != nil {
-		return nil, err
-	}
-	if len(envs) > 1 {
-		return nil, fmt.Errorf("multiple envs found for %q", name)
+func selectEnv(cmd *cobra.Command) (*Env, error) {
+	switch name := GlobalFlags.SelectedEnv; {
+	// A config file was passed without an env.
+	case name == "" && cmd.Flags().Changed(flagConfig):
+		p, envs, err := EnvByName(name, WithInput(GlobalFlags.Vars))
+		if err != nil {
+			return nil, err
+		}
+		if len(envs) != 0 {
+			return nil, fmt.Errorf("unexpected number of envs found: %d", len(envs))
+		}
+		return &Env{Lint: p.Lint, Diff: p.Diff, cfg: p.cfg, Migration: &Migration{}}, nil
+	// No config nor env was passed.
+	case name == "":
+		return &Env{Lint: &Lint{}, Migration: &Migration{}}, nil
+	// Env was passed.
+	default:
+		_, envs, err := EnvByName(name, WithInput(GlobalFlags.Vars))
+		if err != nil {
+			return nil, err
+		}
+		if len(envs) > 1 {
+			return nil, fmt.Errorf("multiple envs found for %q", name)
+		}
+		return envs[0], nil
 	}
-	return envs[0], nil
 }
 
-func schemaFlagsFromEnv(cmd *cobra.Command) error {
-	env, err := selectEnv(GlobalFlags.SelectedEnv)
+func schemaFlagsFromConfig(cmd *cobra.Command) error {
+	env, err := selectEnv(cmd)
 	if err != nil {
 		return err
 	}
diff --git a/cmd/atlas/internal/cmdext/cmdext.go b/cmd/atlas/internal/cmdext/cmdext.go
index 264cc490..dc88bd09 100644
--- a/cmd/atlas/internal/cmdext/cmdext.go
+++ b/cmd/atlas/internal/cmdext/cmdext.go
@@ -466,10 +466,13 @@ type (
 	}
 )
 
-// States is a global registry for external state loaders.
-var States = registry{
-	"ent": EntLoader{},
-}
+var (
+	// States is a global registry for external state loaders.
+	States = registry{
+		"ent": EntLoader{},
+	}
+	errNotSchemaURL = errors.New("missing schema in --dev-url. See: https://atlasgo.io/url")
+)
 
 type registry map[string]StateLoader
 
@@ -495,8 +498,12 @@ type EntLoader struct{}
 // LoadState returns a migrate.StateReader that reads the schema from an ent.Schema.
 func (l EntLoader) LoadState(ctx context.Context, opts *LoadStateOptions) (migrate.StateReader, error) {
 	switch {
-	case opts.Dev == nil, len(opts.URLs) != 1:
-		return nil, errors.New("schema url and dev database are required")
+	case len(opts.URLs) != 1:
+		return nil, errors.New(`"ent://" requires exactly one schema URL`)
+	case opts.Dev == nil:
+		return nil, errors.New(`required flag "--dev-url" not set`)
+	case opts.Dev.URL.Schema == "":
+		return nil, errNotSchemaURL
 	case opts.URLs[0].Query().Has("globalid"):
 		return nil, errors.New("globalid is not supported by this command. Use 'migrate diff' instead")
 	}
@@ -517,6 +524,16 @@ func (l EntLoader) LoadState(ctx context.Context, opts *LoadStateOptions) (migra
 			return nil, err
 		}
 	}
+	if len(realm.Schemas) != 1 {
+		return nil, fmt.Errorf("expect exactly one schema, got %d", len(realm.Schemas))
+	}
+	// Use the dev-database schema name if the schema name is empty.
+	if realm.Schemas[0].Name == "" && opts.Dev.URL.Schema != "" {
+		realm.Schemas[0].Name = opts.Dev.URL.Schema
+	}
+	for _, t := range realm.Schemas[0].Tables {
+		t.Schema = realm.Schemas[0]
+	}
 	return migrate.Realm(realm), nil
 }
 
@@ -525,6 +542,9 @@ func (l EntLoader) MigrateDiff(ctx context.Context, opts *MigrateDiffOptions) er
 	if !l.needDiff(opts.To) {
 		return errors.New("invalid diff call")
 	}
+	if opts.Dev.URL.Schema == "" {
+		return errNotSchemaURL
+	}
 	u, err := url.Parse(opts.To[0])
 	if err != nil {
 		return nil
diff --git a/doc/md/atlas-schema/projects.mdx b/doc/md/atlas-schema/projects.mdx
index a65db1a8..3d735e0b 100644
--- a/doc/md/atlas-schema/projects.mdx
+++ b/doc/md/atlas-schema/projects.mdx
@@ -112,13 +112,47 @@ env "dev" {
 </TabItem>
 </Tabs>
 
-Once defined, a project's environment can be worked against using the `--env` flag.
-For example:
+### Flags
+
+Once the project configuration has been defined, you can interact with it using one of the following options:
+
+<Tabs>
+<TabItem value="env" label="Env">
+
+To run the `schema apply` command using the `prod` configuration defined in `atlas.hcl` file located in your working directory:
 
 ```shell
-atlas schema apply --env local
+atlas schema apply --env prod
+```
+
+</TabItem>
+<TabItem value="custom-path" label="Custom File">
+
+To run the `schema apply` command using the `prod` configuration defined in `atlas.hcl` in arbitrary location:
+
+```shell {2}
+atlas schema apply \
+  -c file://path/to/atlas.hcl \
+  --env prod
 ```
 
+</TabItem>
+<TabItem value="global-config" label="Global Config (without --env)">
+
+Some commands accept global configuration blocks such as [`lint`](#configure-migration-linting) and
+[`diff`](#configure-diff-policy) policies. If no `env` is defined, you can instruct Atlas to explicitly use the config
+file using the `-c` (or `--config`) flag:
+
+```shell {2}
+atlas migrate lint \
+  -c file://path/to/atlas.hcl \
+  --dir "file://path/to/migrations" \
+  --dev-url "sqlite://file?mode=memory"
+```
+
+</TabItem>
+</Tabs>
+
 Will run the `schema apply` command against the database that is defined for the `local`
 environment.
 
@@ -275,6 +309,11 @@ locals {
 Data sources enable users to retrieve information stored in an external service or database. The currently supported
 data sources are: [`sql`](#data-source-sql), [`runtimevar`](#data-source-runtimevar) and [`template_dir`](#data-source-template_dir).
 
+:::note
+Data sources are evaluated only if they are referenced by top-level blocks like `locals` or `variables`, or by the
+selected environment, for instance, `atlas schema apply --env dev`.
+:::
+
 #### Data source: `sql`
 
 ##### Arguments {#data-source-sql-arguments}
diff --git a/doc/md/atlas-schema/sql.mdx b/doc/md/atlas-schema/sql.mdx
index f71aeca9..b042e716 100644
--- a/doc/md/atlas-schema/sql.mdx
+++ b/doc/md/atlas-schema/sql.mdx
@@ -161,6 +161,30 @@ table "metrics" {
 }
 ```
 
+### Storage Engine
+
+The `engine` attribute allows for overriding the default storage engine of the table. Supported by MySQL and MariaDB.
+
+```hcl
+table "users" {
+  schema = schema.public
+  // highlight-next-line
+  engine = MyISAM
+}
+
+table "posts" {
+  schema = schema.public
+  // highlight-next-line
+  engine = InnoDB
+}
+
+table "orders" {
+  schema = schema.public
+  // highlight-next-line
+  engine = "MyRocks"
+}
+```
+
 ### Table Qualification
 
 In some cases, an Atlas DDL document may contain multiple tables of the same name. This usually happens
@@ -596,10 +620,10 @@ table "products" {
 schema "public" {}
 
 table "products" {
-    column "name" {
-        type    = text
-        collate = "es_ES"
-    }
+  column "name" {
+    type    = text
+    collate = "es_ES"
+  }
 }
 ```
 
diff --git a/doc/md/concepts/url.mdx b/doc/md/concepts/url.mdx
index fdc3354d..2dfb36b8 100644
--- a/doc/md/concepts/url.mdx
+++ b/doc/md/concepts/url.mdx
@@ -29,16 +29,17 @@ values={[
 ]}>
 <TabItem value="mysql">
 
+Connecting to a local MySQL server (all schemas/databases):
+```shell
+mysql://localhost:3306/
 ```
-mysql://localhost
 
-mysql://user:pass@localhost
-
-mysql://user:pass@localhost:3306/database
+Connecting to a specific MySQL schema (database) with a username and password:
+```shell
+mysql://user:pass@localhost:3306/schema
 ```
 
 Connecting using Unix Sockets:
-
 ```
 mysql+unix:///tmp/mysql.sock
 
@@ -50,16 +51,17 @@ mysql+unix://user@/tmp/mysql.sock?database=dbname
 </TabItem>
 <TabItem value="maria">
 
+Connecting to a local MariaDB server (all schemas/databases):
+```shell
+maria://localhost:3306/
 ```
-maria://localhost
 
-maria://user:pass@localhost
-
-maria://user:pass@localhost:3306/database
+Connecting to a specific MariaDB schema (database) with a username and password:
+```shell
+maria://user:pass@localhost:3306/schema
 ```
 
 Connecting using Unix Sockets:
-
 ```
 maria+unix:///tmp/mysql.sock
 
@@ -71,21 +73,32 @@ maria+unix://user@/tmp/mysql.sock?database=dbname
 </TabItem>
 <TabItem value="postgres">
 
-```
+Connecting to a local PostgreSQL database named `database` (all schemas):
+```shell
 postgres://localhost:5432/database
+```
 
-postgres://localhost:5432/database?search_path=schema
+Connecting to a specific PostgreSQL schema named `public`:
+```shell
+postgres://localhost:5432/database?search_path=public
+```
 
-postgres://postgres:pass@0.0.0.0:5432/database?sslmode=disable
+Connecting to a local PostgreSQL with credentials and SSL disabled:
+```shell
+postgres://postgres:pass@0.0.0.0:5432/database?search_path=public&sslmode=disable
 ```
 
 </TabItem>
 <TabItem value="sqlite">
 
-```
+Connecting to a local SQLite database (file):
+```shell
 sqlite://file.db
+```
 
-sqlite://file?cache=shared&mode=memory
+Connecting to an in-memory SQLite database (ephemeral). Useful for `--dev-url`:
+```shell
+sqlite://file?mode=memory&_fk=1
 ```
 
 </TabItem>
@@ -96,12 +109,24 @@ if you need a [dev database](../concepts/dev.mdx) for schema validation or diffi
 `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
+```shell
+# PostgreSQL database scope (all schemas).
+docker://postgres/15/test
+
+# PostgreSQL specific schema scope.
+docker://postgres/15/test?search_path=public
+
+# MySQL server scope (all schemas).
+docker://mysql/8
+
+# MySQL specific schema scope.
+docker://mysql/8/test
 
-docker://mariadb/latest
+# MySQL server scope (all schemas).
+docker://maria/latest
 
-docker://mysql/8/database
+# MySQL specific schema scope.
+docker://maria/latest/test
 ```
 
 </TabItem>
diff --git a/doc/md/getting-started/getting-started.mdx b/doc/md/getting-started/getting-started.mdx
index 62d49ce3..98bd52b4 100644
--- a/doc/md/getting-started/getting-started.mdx
+++ b/doc/md/getting-started/getting-started.mdx
@@ -308,7 +308,7 @@ chat on Discord.
 
 For web-based, free, and fun (GIFs included) support:
 
-<a href={"https://discord.gg/8mvDUG22"} className={"join-discord"}>
+<a href={"https://discord.gg/zZ6sWVg6NT"} className={"join-discord"}>
     <Discord />
     <span>
         Join our Discord server
diff --git a/doc/website/blog/2023-05-16-vscode-extension.md b/doc/website/blog/2023-05-16-vscode-extension.md
new file mode 100644
index 00000000..244a159c
--- /dev/null
+++ b/doc/website/blog/2023-05-16-vscode-extension.md
@@ -0,0 +1,101 @@
+---
+title: "Announcing the Atlas HCL VSCode Extension"
+authors: datdao
+tags: [vscode, hcl]
+---
+
+## Introduction
+
+Atlas supports many ways to describe database schemas. Using [Schema Loaders](https://atlasgo.io/blog/2023/01/05/atlas-v090#schema-loaders),
+Atlas gives users the flexibility to define schemas using plain SQL, a connection to another database or using 
+[Atlas HCL](https://atlasgo.io/atlas-schema/sql-resources). 
+
+## What is Atlas HCL?
+
+Atlas includes a data-definition language based on a Terraform-like [syntax](https://atlasgo.io/guides/ddl#hcl). This
+language, which was designed specifically for describing database schemas is based on HashiCorp's [HCL](https://github.com/hashicorp/hcl)
+toolkit for creating structured configuration languages.
+
+This data definition language is easy to read, edit, and visualize, and it allows us to organize our schema data in a
+efficient way that has many benefits to using plain SQL.
+
+## What is the Atlas HCL VSCode Extension?
+
+To make it easier for users to define schemas using Atlas HCL, we're happy to announce the release of the official
+[Atlas HCL VSCode Extension](https://marketplace.visualstudio.com/items?itemName=Ariga.atlas-hcl). 
+This extension provides a rich set of features that will help users define schemas using Atlas HCL.
+
+In this post, we'll take a closer look at the features of this extension
+and how it can benefit your work.
+
+## Features
+
+### Auto Completion
+
+The extension provides auto completion support for all SQL resources available within Atlas. For a comprehensive
+list of the supported resources, please visit this
+document: [SQL Resources](https://atlasgo.io/atlas-schema/sql-resources).
+
+![Completion](https://atlasgo.io/uploads/images/vscode/vscode-completion.gif)
+
+![Batch Completion](https://atlasgo.io/uploads/images/vscode/vscode-batch-completion.gif)
+
+### Referencing Qualified Tables
+
+This refers to a feature that allows for referencing tables and their columns in a structured manner, either relative to
+the current location (e.g. **_column.id_**) or absolute to the table (e.g. **_table.user.column.id_**).
+
+![Referencing Qualified Tables](https://atlasgo.io/uploads/images/vscode/vscode-references.gif)
+
+### Linting
+
+To ensure that errors are caught from the beginning, the Atlas HCL VSCode Extension offers a range of validations that
+enhance type safety for users. The specific validations provided will vary depending on the SQL dialect being used. This
+feature is designed to help users identify and fix issues early on, resulting in a more efficient and effective
+schema-building process.
+
+![Linting](https://atlasgo.io/uploads/images/vscode/vscode-validation.png)
+
+## Usage
+
+To use this extension, you can either install it directly from the VSCode desktop or download it from
+the [Marketplace](https://marketplace.visualstudio.com/items?itemName=Ariga.atlas-hcl). The extension will appear in the
+extensions list as **Atlas HCL**.
+
+![Extension](https://atlasgo.io/uploads/images/vscode/vscode-extension.png)
+
+After installing the Atlas HCL extension, it automatically registers for all the languages that Atlas supports. These
+languages include:
+
+| Name              | Language ID             | Description           
+|-------------------|-------------------------|-----------------------|
+| Atlas SQLite      | atlas-schema-sqlite     | SQLite dialect        
+| Atlas MySQL       | atlas-schema-mysql      | MySQL dialect         
+| Atlas PostgresQL  | atlas-schema-postgresql | PostgresQL dialect    
+| Atlas MariaDB     | atlas-schema-mariadb    | MariaDB dialect       
+| Atlas Config File | atlas-config            | Project Configuration 
+
+By default, the extension will auto-detect atlas.hcl files in your project.
+However, if you have any schema files in your project, you must choose the language for them. You can do this in three
+ways in vscode:
+
+1. Click on "Change Language" for your schema files in this section at bottom. (See image below)
+
+![Installation](https://atlasgo.io/uploads/images/vscode/vscode-installation.png)
+
+2. Use the Command Palette and choose "Change language mode" (See image below)
+
+![Installation by command palette](https://atlasgo.io/uploads/images/vscode/vscode-installation-2.png)
+
+3. Use [Language Identifiers](https://code.visualstudio.com/docs/languages/identifiers)
+
+## Wrapping Up
+
+In this post, we've introduced **The Atlas HCL Extension**. There's much more inside, and we invite you to install it
+today and give it a try.
+
+## What next?
+
+* Follow the [Getting Started](https://atlasgo.io/cli/getting-started/setting-up) guide.
+* Join our [Discord Server](https://discord.gg/zZ6sWVg6NT).
+* Follow us [on Twitter](https://twitter.com/ariga_io).
diff --git a/doc/website/blog/authors.yml b/doc/website/blog/authors.yml
index 46c91a5f..4c613851 100644
--- a/doc/website/blog/authors.yml
+++ b/doc/website/blog/authors.yml
@@ -29,4 +29,12 @@ giautm:
 crossworth:
   name: Pedro Henrique
   url: "https://github.com/crossworth"
-  image_url: "https://avatars.githubusercontent.com/u/1251151?v=4"
\ No newline at end of file
+  image_url: "https://avatars.githubusercontent.com/u/1251151?v=4"
+shani-a:
+  name: Shani Cohen
+  url: "https://github.com/shani-a"
+  image_url: "https://avatars.githubusercontent.com/u/117578627?v=4"
+datdao:
+  name: Dat Dao
+  url: "https://github.com/datdao"
+  image_url: "https://avatars.githubusercontent.com/u/16095902?v=4"
diff --git a/doc/website/package-lock.json b/doc/website/package-lock.json
index 1f92032e..5dc5ffe2 100644
--- a/doc/website/package-lock.json
+++ b/doc/website/package-lock.json
@@ -8,7 +8,7 @@
       "name": "website",
       "version": "0.0.0",
       "dependencies": {
-        "@ariga/atlas-website": "0.0.12",
+        "@ariga/atlas-website": "0.0.14",
         "@docusaurus/core": "^2.0.0-rc.1",
         "@docusaurus/plugin-client-redirects": "^2.0.0-rc.1",
         "@docusaurus/preset-classic": "^2.0.0-rc.1",
@@ -286,9 +286,9 @@
       }
     },
     "node_modules/@ariga/atlas-website": {
-      "version": "0.0.12",
-      "resolved": "https://registry.npmjs.org/@ariga/atlas-website/-/atlas-website-0.0.12.tgz",
-      "integrity": "sha512-3PwBc1c1oW9YxAFmMXFJVvrdoLkkL+QuH1aHNgtSEMbVLYWfS4Rdnd9U0GIjvWujVgFFCeWh5EuURHHfHuhccg==",
+      "version": "0.0.14",
+      "resolved": "https://registry.npmjs.org/@ariga/atlas-website/-/atlas-website-0.0.14.tgz",
+      "integrity": "sha512-QE6/folNZNd6A7z4HaEkr/m36hqZYcWZHy1AMwRALsBmgl5zkqDLc3RA1GqccDsDoah9tPLFeoLed3axG705pg==",
       "dependencies": {
         "@amplitude/analytics-browser": "^1.9.4",
         "@types/amplitude-js": "^8.16.2",
@@ -16038,9 +16038,9 @@
       }
     },
     "@ariga/atlas-website": {
-      "version": "0.0.12",
-      "resolved": "https://registry.npmjs.org/@ariga/atlas-website/-/atlas-website-0.0.12.tgz",
-      "integrity": "sha512-3PwBc1c1oW9YxAFmMXFJVvrdoLkkL+QuH1aHNgtSEMbVLYWfS4Rdnd9U0GIjvWujVgFFCeWh5EuURHHfHuhccg==",
+      "version": "0.0.14",
+      "resolved": "https://registry.npmjs.org/@ariga/atlas-website/-/atlas-website-0.0.14.tgz",
+      "integrity": "sha512-QE6/folNZNd6A7z4HaEkr/m36hqZYcWZHy1AMwRALsBmgl5zkqDLc3RA1GqccDsDoah9tPLFeoLed3axG705pg==",
       "requires": {
         "@amplitude/analytics-browser": "^1.9.4",
         "@types/amplitude-js": "^8.16.2",
diff --git a/doc/website/package.json b/doc/website/package.json
index b528a764..19e0fbfd 100644
--- a/doc/website/package.json
+++ b/doc/website/package.json
@@ -16,7 +16,7 @@
     "invalidate-cdn": "aws cloudfront create-invalidation --distribution-id E3VOJYSV9YO33D --paths \"/*\""
   },
   "dependencies": {
-    "@ariga/atlas-website": "0.0.12",
+    "@ariga/atlas-website": "0.0.14",
     "@docusaurus/core": "^2.0.0-rc.1",
     "@docusaurus/plugin-client-redirects": "^2.0.0-rc.1",
     "@docusaurus/preset-classic": "^2.0.0-rc.1",
diff --git a/doc/website/sidebars.js b/doc/website/sidebars.js
index 1f9b9b86..20e99614 100644
--- a/doc/website/sidebars.js
+++ b/doc/website/sidebars.js
@@ -35,7 +35,16 @@ module.exports = {
             collapsed: false,
             items: [
                 {type: 'doc', id: 'versioned/diff', label: 'Generating Migrations'},
-                {type: 'doc', id: 'versioned/lint', label: 'Migration Linting'},
+                {
+                    type: 'category', label: 'Migration Linting',
+                    link: {
+                        type: 'doc',
+                        id: 'versioned/lint',
+                    },
+                    items: [
+                        {type: 'doc', id: 'lint/analyzers', label: 'Analyzers and Checks'},
+                    ]
+                },
                 {type: 'doc', id: 'versioned/new', label: 'Manual Migrations'},
                 {type: 'doc', id: 'versioned/apply', label: 'Applying Migrations'},
                 {type: 'doc', id: 'versioned/troubleshoot', label: 'Migration Troubleshooting'},
diff --git a/internal/ci/main.go b/internal/ci/main.go
index a3adcc8f..ca6d71db 100644
--- a/internal/ci/main.go
+++ b/internal/ci/main.go
@@ -98,6 +98,14 @@ var (
 			Ports:   []string{"4308:3306"},
 			Options: mysqlOptions,
 		},
+		{
+			Version: "postgres-ext-postgis",
+			Image:   "postgis/postgis:latest",
+			Regex:   "Postgres",
+			Env:     pgEnv,
+			Ports:   []string{"5429:5432"},
+			Options: pgOptions,
+		},
 		{
 			Version: "postgres10",
 			Image:   "postgres:10",
diff --git a/internal/integration/README.md b/internal/integration/README.md
index 84be8d6b..f0c215bf 100644
--- a/internal/integration/README.md
+++ b/internal/integration/README.md
@@ -44,10 +44,10 @@ go test -run='MySQL_Executor/mysql56' ./...
 If you'd like to run the above for Postgres 10, change the name respectively:
 
 ```shell
-# Run TestMySQL_Executor for all mysql versions
+# Run TestPostgres_Executor for all postgres versions
 go test -run='Postgres_Executor' ./... 
 
-# Run TestMySQL_Executor for postgres 10 only
+# Run TestPostgres_Executor for postgres 10 only
 go test -run='Postgres_Executor/postgres10' ./...
 ```
 
diff --git a/internal/integration/docker-compose.yaml b/internal/integration/docker-compose.yaml
index 1ff09c08..9af49b2d 100644
--- a/internal/integration/docker-compose.yaml
+++ b/internal/integration/docker-compose.yaml
@@ -37,6 +37,18 @@ services:
     ports:
       - "3308:3306"
 
+  postgres-ext-postgis:
+    container_name: atlas-integration-postgres-ext-postgis
+    platform: linux/amd64
+    image: postgis/postgis:latest
+    environment:
+      POSTGRES_DB: test
+      POSTGRES_PASSWORD: pass
+    healthcheck:
+      test: pg_isready -U postgres
+    ports:
+      - "5429:5432"
+
   postgres10:
     container_name: atlas-integration-postgres10
     platform: linux/amd64
diff --git a/schemahcl/context.go b/schemahcl/context.go
index 1ea38708..971f04fb 100644
--- a/schemahcl/context.go
+++ b/schemahcl/context.go
@@ -86,7 +86,7 @@ func (s *State) evalReferences(ctx *hcl.EvalContext, body *hclsyntax.Body) error
 	)
 	for _, b := range body.Blocks {
 		switch b := b; {
-		case b.Type == dataBlock:
+		case b.Type == BlockData:
 			if len(b.Labels) < 2 {
 				return fmt.Errorf("data block %q must have exactly 2 labels", b.Type)
 			}
@@ -96,18 +96,18 @@ func (s *State) evalReferences(ctx *hcl.EvalContext, body *hclsyntax.Body) error
 			}
 			// Data references are combined from
 			// "data", "source" and "name" labels.
-			addr := [3]string{dataBlock, b.Labels[0], b.Labels[1]}
+			addr := [3]string{RefData, b.Labels[0], b.Labels[1]}
 			nodes[addr] = &node{
 				addr:  addr,
 				value: func() (cty.Value, error) { return h(ctx, b) },
 				edges: func() []hcl.Traversal { return bodyVars(b.Body) },
 			}
-		case b.Type == localsBlock:
+		case b.Type == BlockLocals:
 			for k, v := range b.Body.Attributes {
 				k, v := k, v
 				// Local references are combined from
 				// "local" and "name" labels.
-				addr := [3]string{localRef, k, ""}
+				addr := [3]string{RefLocal, k, ""}
 				nodes[addr] = &node{
 					addr:  addr,
 					edges: func() []hcl.Traversal { return hclsyntax.Variables(v.Expr) },
@@ -160,10 +160,10 @@ func (s *State) evalReferences(ctx *hcl.EvalContext, body *hclsyntax.Body) error
 		for _, e := range n.edges() {
 			var addr [3]string
 			switch root := e.RootName(); {
-			case root == localRef && len(e) == 2:
-				addr = [3]string{localRef, e[1].(hcl.TraverseAttr).Name, ""}
-			case root == dataBlock && len(e) > 2:
-				addr = [3]string{dataBlock, e[1].(hcl.TraverseAttr).Name, e[2].(hcl.TraverseAttr).Name}
+			case root == RefLocal && len(e) == 2:
+				addr = [3]string{RefLocal, e[1].(hcl.TraverseAttr).Name, ""}
+			case root == RefData && len(e) > 2:
+				addr = [3]string{RefData, e[1].(hcl.TraverseAttr).Name, e[2].(hcl.TraverseAttr).Name}
 			case s.config.initblk[root] != nil && len(e) == 1:
 				addr = [3]string{root, "", ""}
 			}
@@ -181,9 +181,9 @@ func (s *State) evalReferences(ctx *hcl.EvalContext, body *hclsyntax.Body) error
 			return err
 		}
 		switch n.addr[0] {
-		case dataBlock:
+		case RefData:
 			data := make(map[string]cty.Value)
-			if vv, ok := ctx.Variables[dataBlock]; ok {
+			if vv, ok := ctx.Variables[RefData]; ok {
 				data = vv.AsValueMap()
 			}
 			src := make(map[string]cty.Value)
@@ -192,14 +192,14 @@ func (s *State) evalReferences(ctx *hcl.EvalContext, body *hclsyntax.Body) error
 			}
 			src[n.addr[2]] = v
 			data[n.addr[1]] = cty.ObjectVal(src)
-			ctx.Variables[dataBlock] = cty.ObjectVal(data)
-		case localRef:
+			ctx.Variables[RefData] = cty.ObjectVal(data)
+		case RefLocal:
 			locals := make(map[string]cty.Value)
-			if vv, ok := ctx.Variables[localRef]; ok {
+			if vv, ok := ctx.Variables[RefLocal]; ok {
 				locals = vv.AsValueMap()
 			}
 			locals[n.addr[1]] = v
-			ctx.Variables[localRef] = cty.ObjectVal(locals)
+			ctx.Variables[RefLocal] = cty.ObjectVal(locals)
 		default:
 			ctx.Variables[n.addr[0]] = v
 		}
@@ -221,15 +221,15 @@ func (s *State) evalReferences(ctx *hcl.EvalContext, body *hclsyntax.Body) error
 	return nil
 }
 
-func mergeCtxVar(ctx *hcl.EvalContext, vals map[string]cty.Value) {
-	v, ok := ctx.Variables[varRef]
+func mergeCtxVar(ctx *hcl.EvalContext, values map[string]cty.Value) {
+	v, ok := ctx.Variables[RefVar]
 	if ok {
 		v.ForEachElement(func(key cty.Value, val cty.Value) (stop bool) {
-			vals[key.AsString()] = val
+			values[key.AsString()] = val
 			return false
 		})
 	}
-	ctx.Variables[varRef] = cty.ObjectVal(vals)
+	ctx.Variables[RefVar] = cty.ObjectVal(values)
 }
 
 func setBlockVars(ctx *hcl.EvalContext, b *hclsyntax.Body) (*hcl.EvalContext, error) {
@@ -374,15 +374,60 @@ var (
 
 // Built-in blocks.
 const (
-	varBlock    = "variable"
-	dataBlock   = "data"
-	localsBlock = "locals"
-	forEachAttr = "for_each"
-	eachRef     = "each"
-	varRef      = "var"
-	localRef    = "local"
+	BlockData     = "data"
+	BlockLocals   = "locals"
+	BlockVariable = "variable"
+	RefData       = "data"
+	RefVar        = "var"
+	RefLocal      = "local"
+	AttrName      = "name"
+	forEachAttr   = "for_each"
+	eachRef       = "each"
 )
 
+// Variables represents the dynamic variables used in a body.
+type Variables struct {
+	Var, Local map[string]bool
+	Data       map[string]map[string]bool
+}
+
+// DynamicRefs returns the references to dynamic variables.
+// i.e, variables, locals, and data sources.
+func DynamicRefs(b *hclsyntax.Body) *Variables {
+	var (
+		vars = &Variables{
+			Var:   make(map[string]bool),
+			Local: make(map[string]bool),
+			Data:  make(map[string]map[string]bool),
+		}
+		visit func(b *hclsyntax.Body, vars *Variables)
+	)
+	visit = func(b *hclsyntax.Body, vars *Variables) {
+		for _, a := range b.Attributes {
+			for _, v := range hclsyntax.Variables(a.Expr) {
+				switch {
+				case len(v) == 0:
+				case v.RootName() == RefVar && len(v) > 1:
+					vars.Var[v[1].(hcl.TraverseAttr).Name] = true
+				case v.RootName() == RefLocal && len(v) > 1:
+					vars.Local[v[1].(hcl.TraverseAttr).Name] = true
+				case v.RootName() == RefData && len(v) > 2:
+					n1, n2 := v[1].(hcl.TraverseAttr).Name, v[2].(hcl.TraverseAttr).Name
+					if _, ok := vars.Data[n1]; !ok {
+						vars.Data[n1] = make(map[string]bool)
+					}
+					vars.Data[n1][n2] = true
+				}
+			}
+		}
+		for _, b := range b.Blocks {
+			visit(b.Body, vars)
+		}
+	}
+	visit(b, vars)
+	return vars
+}
+
 // defRegistry returns a tree of blockDef structs representing the schema of the
 // blocks in the *hclsyntax.Body. The returned fields and children of each type
 // are an intersection of all existing blocks of the same type.
@@ -393,7 +438,7 @@ func defRegistry(b *hclsyntax.Body) *blockDef {
 	}
 	for _, blk := range b.Blocks {
 		// variable definition blocks are available in the HCL source but not reachable by reference.
-		if blk.Type == varBlock {
+		if blk.Type == BlockVariable {
 			continue
 		}
 		reg.child(extractDef(blk, reg))
diff --git a/schemahcl/hcl.go b/schemahcl/hcl.go
index 66595428..098110b1 100644
--- a/schemahcl/hcl.go
+++ b/schemahcl/hcl.go
@@ -90,7 +90,7 @@ func (s *State) Eval(parsed *hclparse.Parser, v any, input map[string]cty.Value)
 		for _, b := range body.Blocks {
 			switch {
 			// Variable blocks are not reachable by reference.
-			case b.Type == varBlock:
+			case b.Type == BlockVariable:
 				continue
 			// Semi-evaluate blocks with the for_each meta argument.
 			case b.Body != nil && b.Body.Attributes[forEachAttr] != nil:
@@ -222,7 +222,7 @@ func (s *State) resource(ctx *hcl.EvalContext, file *hcl.File) (*Resource, error
 	}
 	for _, blk := range body.Blocks {
 		// variable blocks may be included in the document but are skipped in unmarshaling.
-		if blk.Type == varBlock {
+		if blk.Type == BlockVariable {
 			continue
 		}
 		ctx, err := setBlockVars(ctx.NewChild(), blk.Body)
diff --git a/sql/mysql/diff.go b/sql/mysql/diff.go
index cf8f5649..05585bf2 100644
--- a/sql/mysql/diff.go
+++ b/sql/mysql/diff.go
@@ -68,6 +68,9 @@ func (d *diff) TableAttrDiff(from, to *schema.Table) ([]schema.Change, error) {
 	if change := d.collationChange(from.Attrs, from.Schema.Attrs, to.Attrs); change != noChange {
 		changes = append(changes, change)
 	}
+	if change := d.engineChange(from.Attrs, to.Attrs); change != noChange {
+		changes = append(changes, change)
+	}
 	if !d.SupportsCheck() && sqlx.Has(to.Attrs, &schema.Check{}) {
 		return nil, fmt.Errorf("version %q does not support CHECK constraints", d.V)
 	}
@@ -275,6 +278,35 @@ func (*diff) collationChange(from, top, to []schema.Attr) schema.Change {
 	return noChange
 }
 
+// engineChange returns the schema change for migrating the table engine in case
+// it was changed.
+func (*diff) engineChange(from, to []schema.Attr) schema.Change {
+	var fromE, toE Engine
+	switch fromHas, toHas := sqlx.Has(from, &fromE), sqlx.Has(to, &toE); {
+	// Both engines are defined but different.
+	case fromHas && toHas && strings.ToLower(fromE.V) != strings.ToLower(toE.V):
+		return &schema.ModifyAttr{
+			From: &fromE,
+			To:   &toE,
+		}
+	// If the engine attribute has been removed from the desired state (e.g., HCL), and the current state
+	// is not the default, we change the engine to InnoDB (the default for MySQL, unless configured otherwise).
+	case fromHas && !toHas && !fromE.Default && strings.ToLower(fromE.V) != strings.ToLower(EngineInnoDB):
+		return &schema.ModifyAttr{
+			From: &fromE,
+			To:   &Engine{V: EngineInnoDB, Default: true},
+		}
+	// In case the engine attribute was added to the desired state (e.g., HCL)
+	// and it is not the default, we modify the engine to the desired value.
+	case !fromHas && toHas && !toE.Default && strings.ToLower(fromE.V) != strings.ToLower(EngineInnoDB):
+		return &schema.ModifyAttr{
+			From: &Engine{V: EngineInnoDB, Default: true},
+			To:   &toE,
+		}
+	}
+	return noChange
+}
+
 // charsetChange returns the schema change for migrating the collation if
 // it was changed, and it is not the default attribute inherited from its parent.
 func (*diff) charsetChange(from, top, to []schema.Attr) schema.Change {
diff --git a/sql/mysql/driver.go b/sql/mysql/driver.go
index eb76c16d..efc5380d 100644
--- a/sql/mysql/driver.go
+++ b/sql/mysql/driver.go
@@ -363,6 +363,12 @@ const (
 	IndexTypeFullText = "FULLTEXT"
 	IndexTypeSpatial  = "SPATIAL"
 
+	EngineInnoDB = "InnoDB"
+	EngineMyISAM = "MyISAM"
+	EngineMemory = "Memory"
+	EngineCSV    = "CSV"
+	EngineNDB    = "NDB" // NDBCLUSTER
+
 	currentTS     = "current_timestamp"
 	defaultGen    = "default_generated"
 	autoIncrement = "auto_increment"
diff --git a/sql/mysql/inspect.go b/sql/mysql/inspect.go
index 73986822..6f6ecfd3 100644
--- a/sql/mysql/inspect.go
+++ b/sql/mysql/inspect.go
@@ -161,10 +161,11 @@ func (i *inspect) tables(ctx context.Context, realm *schema.Realm, opts *schema.
 	defer rows.Close()
 	for rows.Next() {
 		var (
-			autoinc                                             sql.NullInt64
-			tSchema, name, charset, collation, comment, options sql.NullString
+			defaultE                                                    sql.NullBool
+			autoinc                                                     sql.NullInt64
+			tSchema, name, charset, collation, comment, options, engine sql.NullString
 		)
-		if err := rows.Scan(&tSchema, &name, &charset, &collation, &autoinc, &comment, &options); err != nil {
+		if err := rows.Scan(&tSchema, &name, &charset, &collation, &autoinc, &comment, &options, &engine, &defaultE); err != nil {
 			return fmt.Errorf("scan table information: %w", err)
 		}
 		if !sqlx.ValidString(tSchema) || !sqlx.ValidString(name) {
@@ -196,6 +197,12 @@ func (i *inspect) tables(ctx context.Context, realm *schema.Realm, opts *schema.
 				V: options.String,
 			})
 		}
+		if sqlx.ValidString(engine) && defaultE.Valid {
+			t.Attrs = append(t.Attrs, &Engine{
+				V:       engine.String,
+				Default: defaultE.Bool,
+			})
+		}
 		if autoinc.Valid {
 			t.Attrs = append(t.Attrs, &AutoIncrement{
 				V: autoinc.Int64,
@@ -701,11 +708,15 @@ SELECT
 	t1.TABLE_COLLATION,
 	t1.AUTO_INCREMENT,
 	t1.TABLE_COMMENT,
-	t1.CREATE_OPTIONS
+	t1.CREATE_OPTIONS,
+	t1.ENGINE,
+	t3.SUPPORT = 'DEFAULT' AS DEFAULT_ENGINE
 FROM
 	INFORMATION_SCHEMA.TABLES AS t1
 	LEFT JOIN INFORMATION_SCHEMA.COLLATIONS AS t2
 	ON t1.TABLE_COLLATION = t2.COLLATION_NAME
+	LEFT JOIN INFORMATION_SCHEMA.ENGINES AS t3
+	ON t1.ENGINE = t3.ENGINE
 WHERE
 	TABLE_SCHEMA IN (%s)
 	AND TABLE_TYPE = 'BASE TABLE'
@@ -721,11 +732,15 @@ SELECT
 	t1.TABLE_COLLATION,
 	t1.AUTO_INCREMENT,
 	t1.TABLE_COMMENT,
-	t1.CREATE_OPTIONS
+	t1.CREATE_OPTIONS,
+	t1.ENGINE,
+	t3.SUPPORT = 'DEFAULT' AS DEFAULT_ENGINE
 FROM
 	INFORMATION_SCHEMA.TABLES AS t1
 	JOIN INFORMATION_SCHEMA.COLLATIONS AS t2
 	ON t1.TABLE_COLLATION = t2.COLLATION_NAME
+	LEFT JOIN INFORMATION_SCHEMA.ENGINES AS t3
+	ON t1.ENGINE = t3.ENGINE
 WHERE
 	TABLE_SCHEMA IN (%s)
 	AND TABLE_NAME IN (%s)
@@ -800,6 +815,13 @@ type (
 		S string
 	}
 
+	// Engine attribute describes the storage engine used to create a table.
+	Engine struct {
+		schema.Attr
+		V       string // InnoDB, MyISAM, etc.
+		Default bool   // The default engine used by the server.
+	}
+
 	// OnUpdate attribute for columns with "ON UPDATE CURRENT_TIMESTAMP" as a default.
 	OnUpdate struct {
 		schema.Attr
diff --git a/sql/mysql/migrate.go b/sql/mysql/migrate.go
index 98d598f1..dcde0f30 100644
--- a/sql/mysql/migrate.go
+++ b/sql/mysql/migrate.go
@@ -646,10 +646,15 @@ func (s *state) tableAttr(b *sqlx.Builder, c schema.Change, attrs ...schema.Attr
 		case *CreateOptions:
 			b.P(a.V)
 		case *AutoIncrement:
-			// Update the AUTO_INCREMENT if it is an update change or it is not the default.
+			// Update the AUTO_INCREMENT if it is a table modification, or it is not the default.
 			if _, ok := c.(*schema.ModifyAttr); ok || a.V > 1 {
 				b.P("AUTO_INCREMENT", strconv.FormatInt(a.V, 10))
 			}
+		case *Engine:
+			// Update the ENGINE if it is a table modification, or it is not the default.
+			if _, ok := c.(*schema.ModifyAttr); ok || !a.Default {
+				b.P("ENGINE", a.V)
+			}
 		case *schema.Check:
 			// Ignore CHECK constraints as they are not real attributes,
 			// and handled on CREATE or ALTER.
diff --git a/sql/mysql/sqlspec.go b/sql/mysql/sqlspec.go
index e36cf6ac..ba691950 100644
--- a/sql/mysql/sqlspec.go
+++ b/sql/mysql/sqlspec.go
@@ -79,6 +79,7 @@ func MarshalSpec(v any, marshaler schemahcl.Marshaler) ([]byte, error) {
 var (
 	hclState = schemahcl.New(
 		schemahcl.WithTypes("table.column.type", TypeRegistry.Specs()),
+		schemahcl.WithScopedEnums("table.engine", EngineInnoDB, EngineMyISAM, EngineMemory, EngineCSV, EngineNDB),
 		schemahcl.WithScopedEnums("table.index.type", IndexTypeBTree, IndexTypeHash, IndexTypeFullText, IndexTypeSpatial),
 		schemahcl.WithScopedEnums("table.primary_key.type", IndexTypeBTree, IndexTypeHash, IndexTypeFullText, IndexTypeSpatial),
 		schemahcl.WithScopedEnums("table.column.as.type", stored, persistent, virtual),
@@ -117,6 +118,13 @@ func convertTable(spec *sqlspec.Table, parent *schema.Schema) (*schema.Table, er
 		}
 		t.AddAttrs(&AutoIncrement{V: v})
 	}
+	if attr, ok := spec.Attr("engine"); ok {
+		v, err := attr.String()
+		if err != nil {
+			return nil, err
+		}
+		t.AddAttrs(&Engine{V: v})
+	}
 	return t, err
 }
 
@@ -255,6 +263,17 @@ func tableSpec(t *schema.Table) (*sqlspec.Table, error) {
 	if c, ok := hasCollate(t.Attrs, t.Schema.Attrs); ok {
 		ts.Extra.Attrs = append(ts.Extra.Attrs, schemahcl.StringAttr("collate", c))
 	}
+	// Marshal the engine attribute only if it is not InnoDB (default).
+	if e := (&Engine{}); sqlx.Has(t.Attrs, e) && e.V != "" && !e.Default {
+		attr := schemahcl.StringAttr("engine", e.V)
+		for _, e1 := range []string{EngineInnoDB, EngineMyISAM, EngineMemory, EngineCSV, EngineNDB} {
+			if strings.EqualFold(e.V, e1) {
+				attr = specutil.VarAttr("engine", e1)
+				break
+			}
+		}
+		ts.Extra.Attrs = append(ts.Extra.Attrs, attr)
+	}
 	return ts, nil
 }
 
diff --git a/sql/postgres/inspect.go b/sql/postgres/inspect.go
index 89ea7c41..5122a452 100644
--- a/sql/postgres/inspect.go
+++ b/sql/postgres/inspect.go
@@ -1092,10 +1092,12 @@ FROM
 	JOIN pg_catalog.pg_namespace AS t2 ON t2.nspname = t1.table_schema
 	JOIN pg_catalog.pg_class AS t3 ON t3.relnamespace = t2.oid AND t3.relname = t1.table_name
 	LEFT JOIN pg_catalog.pg_partitioned_table AS t4 ON t4.partrelid = t3.oid
+	LEFT JOIN pg_depend AS t5 ON t5.objid = t3.oid AND t5.deptype = 'e'
 WHERE
 	t1.table_type = 'BASE TABLE'
 	AND NOT COALESCE(t3.relispartition, false)
 	AND t1.table_schema IN (%s)
+	AND t5.objid IS NULL
 ORDER BY
 	t1.table_schema, t1.table_name
 `
@@ -1112,11 +1114,13 @@ FROM
 	JOIN pg_catalog.pg_namespace AS t2 ON t2.nspname = t1.table_schema
 	JOIN pg_catalog.pg_class AS t3 ON t3.relnamespace = t2.oid AND t3.relname = t1.table_name
 	LEFT JOIN pg_catalog.pg_partitioned_table AS t4 ON t4.partrelid = t3.oid
+	LEFT JOIN pg_depend AS t5 ON t5.objid = t3.oid AND t5.deptype = 'e'
 WHERE
 	t1.table_type = 'BASE TABLE'
 	AND NOT COALESCE(t3.relispartition, false)
 	AND t1.table_schema IN (%s)
 	AND t1.table_name IN (%s)
+	AND t5.objid IS NULL
 ORDER BY
 	t1.table_schema, t1.table_name
 `
diff --git a/sql/schema/inspect.go b/sql/schema/inspect.go
index 080794db..f5f04abd 100644
--- a/sql/schema/inspect.go
+++ b/sql/schema/inspect.go
@@ -83,7 +83,7 @@ type (
 		// Exclude defines a list of glob patterns used to filter resources from inspection.
 		// The syntax used by the different drivers is implemented as follows:
 		//
-		//	s     // exclude schema 't'.
+		//	s     // exclude schema 's'.
 		//	*     // exclude all schemas.
 		//	s.t   // exclude table 't' under schema 's'.
 		//	s.*   // the last item defines the filtering; all tables under 's' are excluded.
diff --git a/sql/sqltool/tool.go b/sql/sqltool/tool.go
index a0768751..e1529070 100644
--- a/sql/sqltool/tool.go
+++ b/sql/sqltool/tool.go
@@ -7,6 +7,7 @@ package sqltool
 import (
 	"bufio"
 	"bytes"
+	"errors"
 	"fmt"
 	"io/fs"
 	"path/filepath"
@@ -70,9 +71,9 @@ var (
 )
 
 type (
-	// GolangMigrateDir wraps migrate.LocalDir and provides a migrate.Scanner implementation able to understand files
+	// GolangMigrateDir wraps fs.FS and provides a migrate.Scanner implementation able to understand files
 	// generated by the GolangMigrateFormatter for migration directory replaying.
-	GolangMigrateDir struct{ *migrate.LocalDir }
+	GolangMigrateDir struct{ fs.FS }
 	// GolangMigrateFile wraps migrate.LocalFile with custom description function.
 	GolangMigrateFile struct{ *migrate.LocalFile }
 )
@@ -86,6 +87,14 @@ func NewGolangMigrateDir(path string) (*GolangMigrateDir, error) {
 	return &GolangMigrateDir{dir}, nil
 }
 
+// Path returns the local path used for opening this dir.
+func (d *GolangMigrateDir) Path() string {
+	if dir, ok := d.FS.(dirPath); ok {
+		return dir.Path()
+	}
+	return ""
+}
+
 // Files implements Scanner.Files. It looks for all files with up.sql suffix and orders them by filename.
 func (d *GolangMigrateDir) Files() ([]migrate.File, error) {
 	names, err := fs.Glob(d, "*.up.sql")
@@ -107,6 +116,26 @@ func (d *GolangMigrateDir) Files() ([]migrate.File, error) {
 	return ret, nil
 }
 
+// Checksum implements Dir.Checksum. By default, it calls Files() and creates a checksum from them.
+func (d *GolangMigrateDir) Checksum() (migrate.HashFile, error) {
+	if d, ok := d.FS.(migrate.Dir); ok {
+		return d.Checksum()
+	}
+	files, err := d.Files()
+	if err != nil {
+		return nil, err
+	}
+	return migrate.NewHashFile(files)
+}
+
+// WriteFile implements Dir.WriteFile.
+func (d *GolangMigrateDir) WriteFile(name string, b []byte) error {
+	if d, ok := d.FS.(migrate.Dir); ok {
+		return d.WriteFile(name, b)
+	}
+	return errors.New("sql/sqltool: write not supported")
+}
+
 // Desc implements File.Desc.
 func (f *GolangMigrateFile) Desc() string {
 	return strings.TrimSuffix(f.LocalFile.Desc(), ".up")
@@ -290,9 +319,9 @@ func (f *DBMateFile) Stmts() ([]string, error) {
 }
 
 type (
-	// FlywayDir wraps migrate.LocalDir and provides a migrate.Scanner implementation able to understand files
+	// FlywayDir wraps fs.FS and provides a migrate.Scanner implementation able to understand files
 	// generated by the FlywayFormatter for migration directory replaying.
-	FlywayDir struct{ *migrate.LocalDir }
+	FlywayDir struct{ fs.FS }
 	// FlywayFile wraps migrate.LocalFile with custom statements function.
 	FlywayFile struct{ *migrate.LocalFile }
 )
@@ -306,17 +335,29 @@ func NewFlywayDir(path string) (*FlywayDir, error) {
 	return &FlywayDir{dir}, nil
 }
 
+// Path returns the local path used for opening this dir.
+func (d *FlywayDir) Path() string {
+	if dir, ok := d.FS.(dirPath); ok {
+		return dir.Path()
+	}
+	return ""
+}
+
 // Files implements Scanner.Files. It looks for all files with .sql suffix. The given directory is recursively scanned
 // for non-hidden subdirectories. All found files will be ordered by migration type (Baseline, Versioned, Repeatable)
 // and filename.
 func (d *FlywayDir) Files() ([]migrate.File, error) {
 	var ff flywayFiles
-	if err := fs.WalkDir(d, "", func(path string, e fs.DirEntry, err error) error {
+	if err := fs.WalkDir(d, ".", func(path string, e fs.DirEntry, err error) error {
 		if err != nil {
 			return err
 		}
-		if path != "" && e.IsDir() {
-			h, err := hidden(filepath.Join(d.Path(), path))
+		if path != "." && e.IsDir() {
+			fullPath := path
+			if p, ok := d.FS.(dirPath); ok {
+				fullPath = filepath.Join(p.Path(), path)
+			}
+			h, err := hidden(fullPath)
 			if err != nil {
 				return err
 			}
@@ -351,6 +392,26 @@ func (d *FlywayDir) Files() ([]migrate.File, error) {
 	return ret, nil
 }
 
+// Checksum implements Dir.Checksum. By default, it calls Files() and creates a checksum from them.
+func (d *FlywayDir) Checksum() (migrate.HashFile, error) {
+	if d, ok := d.FS.(migrate.Dir); ok {
+		return d.Checksum()
+	}
+	files, err := d.Files()
+	if err != nil {
+		return nil, err
+	}
+	return migrate.NewHashFile(files)
+}
+
+// WriteFile implements Dir.WriteFile.
+func (d *FlywayDir) WriteFile(name string, b []byte) error {
+	if d, ok := d.FS.(migrate.Dir); ok {
+		return d.WriteFile(name, b)
+	}
+	return errors.New("sql/sqltool: write not supported")
+}
+
 // Desc implements File.Desc.
 func (f FlywayFile) Desc() string {
 	return flywayDesc(f.Name())
@@ -545,3 +606,15 @@ func reverse(changes []*migrate.Change) []*migrate.Change {
 	}
 	return rev
 }
+
+type dirPath interface {
+	Path() string
+}
+
+var (
+	_ dirPath = (*DBMateDir)(nil)
+	_ dirPath = (*FlywayDir)(nil)
+	_ dirPath = (*GolangMigrateDir)(nil)
+	_ dirPath = (*GooseDir)(nil)
+	_ dirPath = (*LiquibaseDir)(nil)
+)


Test patch:
diff --git a/cmd/atlas/internal/cloudapi/client_test.go b/cmd/atlas/internal/cloudapi/client_test.go
index f97ce34e..08f4ff51 100644
--- a/cmd/atlas/internal/cloudapi/client_test.go
+++ b/cmd/atlas/internal/cloudapi/client_test.go
@@ -71,3 +71,48 @@ func TestClient_ReportMigration(t *testing.T) {
 	})
 	require.NoError(t, err)
 }
+
+func TestClient_ReportMigrationSet(t *testing.T) {
+	const (
+		planned               = 2
+		id, log, project, env = "deployment-set-1", "started deployment", "atlas", "dev"
+	)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var input struct {
+			Variables struct {
+				Input ReportMigrationSetInput `json:"input"`
+			} `json:"variables"`
+		}
+		err := json.NewDecoder(r.Body).Decode(&input)
+		require.NoError(t, err)
+		require.Equal(t, id, input.Variables.Input.ID)
+		require.Equal(t, []ReportStep{{Text: log}}, input.Variables.Input.Log)
+		require.Equal(t, planned, input.Variables.Input.Planned)
+		require.Equal(t, env, input.Variables.Input.Completed[0].EnvName)
+		require.Equal(t, project, input.Variables.Input.Completed[0].ProjectName)
+		require.Equal(t, "dir-1", input.Variables.Input.Completed[0].DirName)
+		require.Equal(t, env, input.Variables.Input.Completed[1].EnvName)
+		require.Equal(t, project, input.Variables.Input.Completed[1].ProjectName)
+		require.Equal(t, "dir-2", input.Variables.Input.Completed[1].DirName)
+	}))
+	client := New(srv.URL, "atlas")
+	defer srv.Close()
+	err := client.ReportMigrationSet(context.Background(), ReportMigrationSetInput{
+		ID:      id,
+		Planned: planned,
+		Log:     []ReportStep{{Text: log}},
+		Completed: []ReportMigrationInput{
+			{
+				EnvName:     env,
+				ProjectName: project,
+				DirName:     "dir-1",
+			},
+			{
+				EnvName:     env,
+				ProjectName: project,
+				DirName:     "dir-2",
+			},
+		},
+	})
+	require.NoError(t, err)
+}
diff --git a/cmd/atlas/internal/cmdapi/migrate_test.go b/cmd/atlas/internal/cmdapi/migrate_test.go
index d5b76941..27f8ea8c 100644
--- a/cmd/atlas/internal/cmdapi/migrate_test.go
+++ b/cmd/atlas/internal/cmdapi/migrate_test.go
@@ -442,6 +442,12 @@ variable "url" {
   type = string
 }
 
+locals {
+  string = "%test"
+  bool   = true
+  int    = 1
+}
+
 data "sql" "tenants" {
   url   = var.url
   query = <<EOS
@@ -449,7 +455,7 @@ SELECT name FROM tenants
 	WHERE mode LIKE ? AND active = ? AND created = ?
 EOS
   # Pass all types of arguments.
-  args  = ["%test", true, 1]
+  args  = [local.string, local.bool, local.int]
 }
 
 env "local" {
@@ -843,6 +849,7 @@ env {
 		p    = t.TempDir()
 		path = filepath.Join(p, "atlas.hcl")
 	)
+	t.Cleanup(srv.Close)
 	require.NoError(t, os.WriteFile(path, []byte(h), 0600))
 
 	t.Run("NoPendingFiles", func(t *testing.T) {
@@ -875,6 +882,7 @@ env {
 				Schema: "main",
 				URL:    u,
 			},
+			Log: "No migration files to execute\n",
 		}, report)
 	})
 
@@ -907,7 +915,7 @@ env {
 		}
 	})
 
-	t.Run("Error", func(t *testing.T) {
+	t.Run("PrintError", func(t *testing.T) {
 		status = http.StatusInternalServerError
 		require.NoError(t, dir.WriteFile("3.sql", []byte("create table baz (id int)")))
 		cmd := migrateCmd()
@@ -919,9 +927,172 @@ env {
 			"--url", u,
 			"--var", "cloud_url="+srv.URL,
 		)
-		require.EqualError(t, err, "unexpected status code: 500")
+		require.NoError(t, err)
 		// Reporting error should not affect the migration execution.
 		require.True(t, strings.HasSuffix(s, "  -- 1 migrations \n  -- 1 sql statements\nError: unexpected status code: 500\n"))
+
+		// Custom logging.
+		cmd = migrateCmd()
+		cmd.AddCommand(migrateApplyCmd())
+		s, err = runCmd(
+			cmd, "apply",
+			"-c", "file://"+path,
+			"--env", "local",
+			"--url", u,
+			"--var", "cloud_url="+srv.URL,
+			"--format", "{{ .Env.Driver }}",
+		)
+		require.NoError(t, err)
+		require.Equal(t, "sqlite3\nError: unexpected status code: 500\n", s)
+	})
+}
+
+func TestMigrate_ApplyCloudReportSet(t *testing.T) {
+	var (
+		dir    migrate.MemDir
+		status int
+		report cloudapi.ReportMigrationSetInput
+		srv    = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			var m struct {
+				Query     string `json:"query"`
+				Variables struct {
+					Input json.RawMessage `json:"input"`
+				} `json:"variables"`
+			}
+			require.NoError(t, json.NewDecoder(r.Body).Decode(&m))
+			switch {
+			case strings.Contains(m.Query, "query"):
+				// Checksum before archiving.
+				hf, err := dir.Checksum()
+				require.NoError(t, err)
+				ht, err := hf.MarshalText()
+				require.NoError(t, err)
+				require.NoError(t, dir.WriteFile(migrate.HashFileName, ht))
+				// Archive and send.
+				arc, err := migrate.ArchiveDir(&dir)
+				require.NoError(t, err)
+				fmt.Fprintf(w, `{"data":{"dir":{"content":%q}}}`, base64.StdEncoding.EncodeToString(arc))
+			case strings.Contains(m.Query, "mutation"):
+				if status != 0 {
+					w.WriteHeader(status)
+				}
+				require.NoError(t, json.Unmarshal(m.Variables.Input, &report))
+			default:
+				t.Fatalf("unexpected query: %s", m.Query)
+			}
+		}))
+		h = `
+variable "cloud_url" {
+  type = string
+}
+
+atlas {
+  cloud {
+    token   = "token"
+    url     = var.cloud_url
+    project = "example"
+  }
+}
+
+data "remote_dir" "migration_set" {
+  name = "migration_set"
+}
+
+variable "urls" {
+  type    = list(string)
+  default = []
+}
+
+env {
+  name = atlas.env
+  for_each = toset(var.urls)
+  url = each.value
+  migration {
+    dir = data.remote_dir.migration_set.url
+  }
+}
+`
+		p    = t.TempDir()
+		path = filepath.Join(p, "atlas.hcl")
+	)
+	t.Cleanup(srv.Close)
+	require.NoError(t, os.WriteFile(path, []byte(h), 0600))
+
+	t.Run("MultipleEnvs", func(t *testing.T) {
+		cmd := migrateCmd()
+		cmd.AddCommand(migrateApplyCmd())
+		s, err := runCmd(
+			cmd, "apply",
+			"-c", "file://"+path,
+			"--env", "local",
+			"--var", fmt.Sprintf("urls=%s", openSQLite(t, "")),
+			"--var", fmt.Sprintf("urls=%s", openSQLite(t, "")),
+			"--var", "cloud_url="+srv.URL,
+		)
+		require.NoError(t, err)
+		require.Equal(t, "No migration files to execute\nNo migration files to execute\n", s)
+		require.NotEmpty(t, report.ID)
+		_, err = uuid.Parse(report.ID)
+		require.NoError(t, err, "set id is not a valid uuid")
+		require.False(t, report.StartTime.IsZero())
+		require.False(t, report.EndTime.IsZero())
+		require.Equal(t, 2, report.Planned)
+		require.Len(t, report.Completed, 2)
+		require.Nil(t, report.Error)
+
+		// Verify summary log.
+		require.Len(t, report.Log, 3)
+		top := report.Log[0]
+		require.Equal(t, top.Text, "Start migration for 2 targets")
+		require.Contains(t, top.Log[0].Text, "sqlite://file")
+		require.Contains(t, top.Log[1].Text, "sqlite://file")
+		require.Equal(t, report.Log[1].Text, "Run migration: 1")
+		require.Contains(t, report.Log[1].Log[0].Text, "Target URL: sqlite://file")
+		require.Contains(t, report.Log[1].Log[1].Text, "Migration directory: mem://migration_set")
+		require.Equal(t, report.Log[2].Text, "Run migration: 2")
+		require.Contains(t, report.Log[2].Log[0].Text, "Target URL: sqlite://file")
+		require.Contains(t, report.Log[2].Log[1].Text, "Migration directory: mem://migration_set")
+	})
+
+	t.Run("Error", func(t *testing.T) {
+		require.NoError(t, dir.WriteFile("1.sql", []byte("create table foo (id int)")))
+		require.NoError(t, dir.WriteFile("2.sql", []byte("create table bar (id int)")))
+
+		cmd := migrateCmd()
+		cmd.AddCommand(migrateApplyCmd())
+		s, err := runCmd(
+			cmd, "apply",
+			"-c", "file://"+path,
+			"--env", "local",
+			"--var", fmt.Sprintf("urls=%s", openSQLite(t, "create table bar (id int)")),
+			"--var", fmt.Sprintf("urls=%s", openSQLite(t, "create table bar (id int)")),
+			"--var", "cloud_url="+srv.URL,
+			"--allow-dirty",
+		)
+		require.EqualError(t, err, `sql/migrate: execute: executing statement "create table bar (id int)" from version "2": table bar already exists`)
+		require.Contains(t, s, "Migrating to version 2 (2 migrations in total):")
+		require.Contains(t, s, "Error: sql/migrate:")
+		require.NotEmpty(t, report.ID)
+		_, err = uuid.Parse(report.ID)
+		require.NoError(t, err, "set id is not a valid uuid")
+		require.False(t, report.StartTime.IsZero())
+		require.False(t, report.EndTime.IsZero())
+		require.Equal(t, 2, report.Planned)
+		require.Len(t, report.Completed, 1)
+		require.NotNil(t, report.Error)
+		require.Equal(t, `Error: sql/migrate: execute: executing statement "create table bar (id int)" from version "2": table bar already exists`, *report.Error)
+
+		// Verify summary log.
+		require.Len(t, report.Log, 2)
+		top := report.Log[0]
+		require.Equal(t, top.Text, "Start migration for 2 targets")
+		require.Contains(t, top.Log[0].Text, "sqlite://file")
+		require.Contains(t, top.Log[1].Text, "sqlite://file")
+		require.Equal(t, report.Log[1].Text, "Run migration: 1")
+		require.True(t, report.Log[1].Error)
+		require.Contains(t, report.Log[1].Log[0].Text, "Target URL: sqlite://file")
+		require.Contains(t, report.Log[1].Log[1].Text, "Migration directory: mem://migration_set")
+		require.Equal(t, `Error: sql/migrate: execute: executing statement "create table bar (id int)" from version "2": table bar already exists`, report.Log[1].Log[2].Text)
 	})
 }
 
@@ -1439,6 +1610,46 @@ func TestMigrate_Lint(t *testing.T) {
 	require.Error(t, err)
 	require.Equal(t, "2.sql", s)
 
+	t.Run("FromConfig", func(t *testing.T) {
+		cfg := filepath.Join(p, "atlas.hcl")
+		err := os.WriteFile(cfg, []byte(`
+variable "error" {
+  type    = bool
+  default = false
+}
+
+lint {
+  latest = 1
+  destructive {
+    error = var.error
+  }
+}
+`), 0600)
+		require.NoError(t, err)
+		cmd := migrateCmd()
+		cmd.AddCommand(migrateLintCmd())
+		s, err := runCmd(
+			cmd, "lint",
+			"--dir", "file://"+p,
+			"--dev-url", openSQLite(t, ""),
+			"-c", "file://"+cfg,
+		)
+		require.NoError(t, err)
+		require.Equal(t, "2.sql: destructive changes detected:\n\n\tL1: Dropping table \"t\"\n\n", s)
+
+		cmd = migrateCmd()
+		cmd.AddCommand(migrateLintCmd())
+		s, err = runCmd(
+			cmd, "lint",
+			"--dir", "file://"+p,
+			"--dev-url", openSQLite(t, ""),
+			"-c", "file://"+cfg,
+			"--var", "error=true",
+		)
+		require.Error(t, err)
+		require.Equal(t, "2.sql: destructive changes detected:\n\n\tL1: Dropping table \"t\"\n\n", s)
+	})
+
 	// Change files to golang-migrate format.
 	require.NoError(t, os.Rename(filepath.Join(p, "1.sql"), filepath.Join(p, "1.up.sql")))
 	require.NoError(t, os.Rename(filepath.Join(p, "2.sql"), filepath.Join(p, "1.down.sql")))
diff --git a/cmd/atlas/internal/cmdapi/project_test.go b/cmd/atlas/internal/cmdapi/project_test.go
index c07b9ffb..b9b12796 100644
--- a/cmd/atlas/internal/cmdapi/project_test.go
+++ b/cmd/atlas/internal/cmdapi/project_test.go
@@ -18,7 +18,7 @@ import (
 	"github.com/zclconf/go-cty/cty"
 )
 
-func TestLoadEnv(t *testing.T) {
+func TestEnvByName(t *testing.T) {
 	d := t.TempDir()
 	h := `
 variable "name" {
@@ -91,7 +91,7 @@ env "multi" {
 	require.NoError(t, err)
 	GlobalFlags.ConfigURL = "file://" + path
 	t.Run("ok", func(t *testing.T) {
-		envs, err := LoadEnv("local")
+		_, envs, err := EnvByName("local")
 		require.NoError(t, err)
 		require.Len(t, envs, 1)
 		env := envs[0]
@@ -147,7 +147,7 @@ env "multi" {
 		require.EqualValues(t, []string{"local/app.hcl"}, sources)
 	})
 	t.Run("multi", func(t *testing.T) {
-		envs, err := LoadEnv("multi")
+		_, envs, err := EnvByName("multi")
 		require.NoError(t, err)
 		require.Len(t, envs, 1)
 		srcs, err := envs[0].Sources()
@@ -155,7 +155,7 @@ env "multi" {
 		require.EqualValues(t, []string{"./a.hcl", "./b.hcl"}, srcs)
 	})
 	t.Run("with input", func(t *testing.T) {
-		envs, err := LoadEnv("local", WithInput(map[string]cty.Value{
+		_, envs, err := EnvByName("local", WithInput(map[string]cty.Value{
 			"name": cty.StringVal("goodbye"),
 		}))
 		require.NoError(t, err)
@@ -167,12 +167,12 @@ env "multi" {
 		require.EqualValues(t, "goodbye", val)
 	})
 	t.Run("wrong env", func(t *testing.T) {
-		_, err = LoadEnv("home")
+		_, _, err = EnvByName("home")
 		require.EqualError(t, err, `env "home" not defined in project file`)
 	})
 	t.Run("wrong dir", func(t *testing.T) {
 		GlobalFlags.ConfigURL = defaultConfigPath
-		_, err = LoadEnv("home")
+		_, _, err = EnvByName("home")
 		require.ErrorContains(t, err, `no such file or directory`)
 	})
 }
@@ -191,13 +191,100 @@ env {
 	err := os.WriteFile(path, []byte(h), 0600)
 	require.NoError(t, err)
 	GlobalFlags.ConfigURL = "file://" + path
-	envs, err := LoadEnv("local")
+	_, envs, err := EnvByName("local")
 	require.NoError(t, err)
 	require.Len(t, envs, 1)
 	require.Equal(t, "local", envs[0].Name)
 	require.Equal(t, "env: local", envs[0].Format.Schema.Apply)
 }
 
+func TestNoEnv(t *testing.T) {
+	h := `
+env "dev" {
+  log {
+    schema {
+      apply = "env: ${atlas.env}"
+    }
+  }
+}
+
+env {
+  name = atlas.env
+  log {
+    schema {
+      apply = "env: ${atlas.env}"
+    }
+  }
+}
+
+lint {
+  latest = 1
+}
+
+diff {
+  skip {
+    drop_column = true
+  }
+}
+`
+	path := filepath.Join(t.TempDir(), "atlas.hcl")
+	err := os.WriteFile(path, []byte(h), 0600)
+	require.NoError(t, err)
+	GlobalFlags.ConfigURL = "file://" + path
+	project, envs, err := EnvByName("")
+	require.NoError(t, err)
+	require.Len(t, envs, 0)
+	require.Equal(t, 1, project.Lint.Latest)
+	require.NotNil(t, project.Diff.SkipChanges)
+	require.True(t, project.Diff.SkipChanges.DropColumn)
+}
+
+func TestPartialParse(t *testing.T) {
+	h := `
+data "remote_dir" "ignored" {
+  name = "ignored"
+}
+
+locals {
+  a = "b"
+}
+
+env {
+  name = atlas.env
+  log {
+    schema {
+      diff  = local.a
+      apply = "env: ${atlas.env}"
+    }
+  }
+  a = local.a
+}
+
+env "dev" {
+  a = local.a
+}`
+	path := filepath.Join(t.TempDir(), "atlas.hcl")
+	err := os.WriteFile(path, []byte(h), 0600)
+	require.NoError(t, err)
+	GlobalFlags.ConfigURL = "file://" + path
+	_, envs, err := EnvByName("unnamed")
+	require.NoError(t, err)
+	require.Len(t, envs, 1)
+	require.Equal(t, "unnamed", envs[0].Name)
+	require.Equal(t, "b", envs[0].Format.Schema.Diff)
+	require.Equal(t, "env: unnamed", envs[0].Format.Schema.Apply)
+	_, envs, err = EnvByName("dev")
+	require.NoError(t, err)
+	require.Len(t, envs, 2)
+	for _, e := range envs {
+		attr, ok := e.Extra.Attr("a")
+		require.True(t, ok)
+		v, err := attr.String()
+		require.NoError(t, err)
+		require.Equal(t, "b", v)
+	}
+}
+
 func TestDiff_Options(t *testing.T) {
 	d := &Diff{}
 	require.Len(t, d.Options(), 1)
diff --git a/cmd/atlas/internal/cmdapi/schema_test.go b/cmd/atlas/internal/cmdapi/schema_test.go
index 1a12f860..00dc73c0 100644
--- a/cmd/atlas/internal/cmdapi/schema_test.go
+++ b/cmd/atlas/internal/cmdapi/schema_test.go
@@ -538,6 +538,7 @@ table "users" {
 	err = os.WriteFile(cfg, []byte(`
 variable "schema" {
   type = string
+  default = "dev"
 }
 
 variable "destructive" {
@@ -577,6 +578,25 @@ diff {
 		"CREATE TABLE `users` (`id` int NOT NULL);",
 	}, lines)
 
+	// Skip destructive changes by using project-level policy (no --env was passed).
+	cmd = schemaCmd()
+	cmd.AddCommand(schemaApplyCmd())
+	s, err = runCmd(
+		cmd, "apply",
+		"-u", openSQLite(t, "create table pets (id int);"),
+		"-c", "file://"+cfg, // Using the project-level policy.
+		"--to", "file://"+src,
+		"--dev-url", "sqlite://dev?mode=memory&_fk=1",
+		"--auto-approve",
+	)
+	require.NoError(t, err)
+	lines = strings.Split(strings.TrimSpace(s), "\n")
+	require.Equal(t, []string{
+		"-- Planned Changes:",
+		`-- Create "users" table`,
+		"CREATE TABLE `users` (`id` int NOT NULL);",
+	}, lines)
+
 	// Apply destructive changes.
 	cmd = schemaCmd()
 	cmd.AddCommand(schemaApplyCmd())
diff --git a/internal/integration/mysql_test.go b/internal/integration/mysql_test.go
index d5ac4021..66b40604 100644
--- a/internal/integration/mysql_test.go
+++ b/internal/integration/mysql_test.go
@@ -917,6 +917,7 @@ create table atlas_types_sanity
 				Attrs: []schema.Attr{
 					&schema.Charset{V: "latin1"},
 					&schema.Collation{V: "latin1_swedish_ci"},
+					&mysql.Engine{V: "InnoDB", Default: true},
 				},
 				Schema: realm.Schemas[0],
 				Columns: []*schema.Column{
@@ -1267,12 +1268,14 @@ create table atlas_types_sanity
 						return []schema.Attr{
 							&schema.Charset{V: "latin1"},
 							&schema.Collation{V: "latin1_swedish_ci"},
+							&mysql.Engine{V: "InnoDB", Default: true},
 							&schema.Check{Name: "tJSON", Expr: "json_valid(`tJSON`)"},
 						}
 					}
 					return []schema.Attr{
 						&schema.Charset{V: "latin1"},
 						&schema.Collation{V: "latin1_swedish_ci"},
+						&mysql.Engine{V: "InnoDB", Default: true},
 					}
 				}(),
 				Schema: realm.Schemas[0],
diff --git a/internal/integration/postgres_test.go b/internal/integration/postgres_test.go
index 9b325128..7f3fb125 100644
--- a/internal/integration/postgres_test.go
+++ b/internal/integration/postgres_test.go
@@ -35,12 +35,13 @@ type pgTest struct {
 }
 
 var pgTests = map[string]*pgTest{
-	"postgres10": {port: 5430},
-	"postgres11": {port: 5431},
-	"postgres12": {port: 5432},
-	"postgres13": {port: 5433},
-	"postgres14": {port: 5434},
-	"postgres15": {port: 5435},
+	"postgres-ext-postgis": {port: 5429},
+	"postgres10":           {port: 5430},
+	"postgres11":           {port: 5431},
+	"postgres12":           {port: 5432},
+	"postgres13":           {port: 5433},
+	"postgres14":           {port: 5434},
+	"postgres15":           {port: 5435},
 }
 
 func pgRun(t *testing.T, fn func(*pgTest)) {
@@ -56,6 +57,21 @@ func pgRun(t *testing.T, fn func(*pgTest)) {
 						log.Fatalln(err)
 					}
 					dbs = append(dbs, tt.db) // close connection after all tests have been run
+					// the postgis/postgis image enables the postgis_topology and postgis_tiger_geocoder extensions,
+					// this creates a few unwanted schemas, so we drop it.
+					// https://github.com/postgis/docker-postgis/issues/187
+					if tt.version == "postgres-ext-postgis" {
+						schemasToDrop := []string{
+							"tiger",      // created by postgis_tiger_geocoder
+							"tiger_data", // created by postgis_tiger_geocoder
+							"topology",   // created by postgis_topology
+						}
+						for _, s := range schemasToDrop {
+							if _, err := tt.db.Exec("DROP SCHEMA IF EXISTS " + s + " CASCADE;"); err != nil {
+								log.Fatalf("error dropping schema %q: %v", s, err)
+							}
+						}
+					}
 					tt.drv, err = postgres.Open(tt.db)
 					if err != nil {
 						log.Fatalln(err)
diff --git a/internal/integration/testdata/mysql/table-engine.txt b/internal/integration/testdata/mysql/table-engine.txt
new file mode 100644
index 00000000..da251ad7
--- /dev/null
+++ b/internal/integration/testdata/mysql/table-engine.txt
@@ -0,0 +1,99 @@
+only mysql8
+
+# InnoDB as the default storage engine.
+apply 1.hcl
+cmphcl 1.inspect.hcl
+
+# Change the InnoDB (default storage engine) to MyISAM.
+apply 2.hcl
+cmphcl 2.inspect.hcl
+
+# Drop MyISAM changes the engine to the default (InnoDB).
+apply 3.hcl
+cmphcl 3.inspect.hcl
+
+-- 1.hcl --
+schema "script_table_engine" {
+    charset = "$charset"
+    collate = "$collate"
+}
+
+table "users" {
+    schema = schema.$db
+    engine = InnoDB
+    column "name" {
+      null = false
+      type = varchar(255)
+    }
+    charset = "$charset"
+    collate = "$collate"
+}
+
+-- 1.inspect.hcl --
+table "users" {
+  schema = schema.script_table_engine
+  column "name" {
+    null = false
+    type = varchar(255)
+  }
+}
+schema "script_table_engine" {
+  charset = "utf8mb4"
+  collate = "utf8mb4_0900_ai_ci"
+}
+-- 2.hcl --
+schema "script_table_engine" {
+    charset = "$charset"
+    collate = "$collate"
+}
+
+table "users" {
+    schema = schema.$db
+    engine = MyISAM
+    column "name" {
+      null = false
+      type = varchar(255)
+    }
+    charset = "$charset"
+    collate = "$collate"
+}
+-- 2.inspect.hcl --
+table "users" {
+  schema = schema.script_table_engine
+  engine = MyISAM
+  column "name" {
+    null = false
+    type = varchar(255)
+  }
+}
+schema "script_table_engine" {
+  charset = "utf8mb4"
+  collate = "utf8mb4_0900_ai_ci"
+}
+-- 3.hcl --
+schema "script_table_engine" {
+    charset = "$charset"
+    collate = "$collate"
+}
+
+table "users" {
+    schema = schema.$db
+    column "name" {
+      null = false
+      type = varchar(255)
+    }
+    charset = "$charset"
+    collate = "$collate"
+}
+-- 3.inspect.hcl --
+table "users" {
+  schema = schema.script_table_engine
+  column "name" {
+    null = false
+    type = varchar(255)
+  }
+}
+schema "script_table_engine" {
+  charset = "utf8mb4"
+  collate = "utf8mb4_0900_ai_ci"
+}
\ No newline at end of file
diff --git a/internal/integration/tidb_test.go b/internal/integration/tidb_test.go
index 8ae46c92..c72e4ccb 100644
--- a/internal/integration/tidb_test.go
+++ b/internal/integration/tidb_test.go
@@ -704,9 +704,8 @@ create table atlas_types_sanity
 				Name: n,
 				Attrs: []schema.Attr{
 					&schema.Charset{V: "latin1"},
-					&schema.Collation{
-						V: "latin1_bin",
-					},
+					&schema.Collation{V: "latin1_bin"},
+					&mysql.Engine{V: "InnoDB", Default: true},
 				},
 				Schema: realm.Schemas[0],
 				Columns: []*schema.Column{
@@ -971,6 +970,7 @@ create table atlas_types_sanity
 					return []schema.Attr{
 						&schema.Charset{V: "latin1"},
 						&schema.Collation{V: "latin1_bin"},
+						&mysql.Engine{V: "InnoDB", Default: true},
 					}
 				}(),
 				Schema: realm.Schemas[0],
@@ -1008,6 +1008,7 @@ create table atlas_types_sanity
 					return []schema.Attr{
 						&schema.Collation{V: "utf8mb4_bin"},
 						&schema.Charset{V: "utf8mb4"},
+						&mysql.Engine{V: "InnoDB", Default: true},
 					}
 				}(),
 				Columns: []*schema.Column{
diff --git a/schemahcl/context_test.go b/schemahcl/context_test.go
index 6cf0df36..f174f85f 100644
--- a/schemahcl/context_test.go
+++ b/schemahcl/context_test.go
@@ -8,6 +8,8 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/hashicorp/hcl/v2/hclparse"
+	"github.com/hashicorp/hcl/v2/hclsyntax"
 	"github.com/stretchr/testify/require"
 	"github.com/zclconf/go-cty/cty"
 )
@@ -612,3 +614,77 @@ variable "name" {
 	err = New().EvalBytes([]byte(h), &struct{}{}, nil)
 	require.EqualError(t, err, `invalid type "boring" for variable "name". Valid types are: string, number, bool, list, map, or set`)
 }
+
+func TestDynamicRefs(t *testing.T) {
+	var (
+		h = `
+variable "name" {
+  type = string
+}
+
+locals {
+  name = "a8m"
+}
+
+data "remote_dir" "example1" {
+  name = "ent"
+}
+
+data "remote_dir" "example2" {
+  name = "atlas"
+}
+
+data "template_dir" "example1" {
+  name = "atlas"
+}
+
+env "var_only" {
+  v = var.name
+}
+
+env "local_only" {
+  v = local.name
+}
+
+env "data_only" {
+  v = data.remote_dir.example1.url
+}
+
+env "var_local" {
+  v = "${var.name}-${local.name}"
+}
+
+env "var_data" {
+  v = "${var.name}-${data.remote_dir.example1.url}"
+}
+
+env "all" {
+  v = "${var.name}-${local.name}-${data.remote_dir.example1.url}-${data.remote_dir.example2.url}-${data.template_dir.example1.url}"
+}
+`
+		zero1, zero2 = make(map[string]bool), make(map[string]map[string]bool)
+	)
+	for name, vars := range map[string]Variables{
+		"var_only":   {Var: map[string]bool{"name": true}, Local: zero1, Data: zero2},
+		"local_only": {Var: zero1, Local: map[string]bool{"name": true}, Data: zero2},
+		"data_only":  {Var: zero1, Local: zero1, Data: map[string]map[string]bool{"remote_dir": {"example1": true}}},
+		"var_local":  {Var: map[string]bool{"name": true}, Local: map[string]bool{"name": true}, Data: zero2},
+		"var_data":   {Var: map[string]bool{"name": true}, Local: zero1, Data: map[string]map[string]bool{"remote_dir": {"example1": true}}},
+		"all":        {Var: map[string]bool{"name": true}, Local: map[string]bool{"name": true}, Data: map[string]map[string]bool{"remote_dir": {"example1": true, "example2": true}, "template_dir": {"example1": true}}},
+	} {
+		parser := hclparse.NewParser()
+		fi, diags := parser.ParseHCL([]byte(h), "test")
+		require.False(t, diags.HasErrors())
+		env := func() *hclsyntax.Body {
+			for _, b := range fi.Body.(*hclsyntax.Body).Blocks {
+				if b.Type == "env" && b.Labels[0] == name {
+					return b.Body
+				}
+			}
+			require.Failf(t, "env %q block not found", name)
+			return nil
+		}()
+		got := DynamicRefs(env)
+		require.Equal(t, vars, *got)
+	}
+}
diff --git a/sql/mysql/diff_test.go b/sql/mysql/diff_test.go
index f6b3714e..9e586bdf 100644
--- a/sql/mysql/diff_test.go
+++ b/sql/mysql/diff_test.go
@@ -48,6 +48,72 @@ func TestDiff_TableDiff(t *testing.T) {
 				},
 			},
 		},
+		// Attributes are specified and the same.
+		{
+			name: "no engine changes",
+			from: &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}, Attrs: []schema.Attr{&Engine{V: EngineInnoDB}}},
+			to:   &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}, Attrs: []schema.Attr{&Engine{V: EngineInnoDB}}},
+		},
+		// Attributes are specified and represent the same engine.
+		{
+			name: "no engine changes",
+			from: &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}, Attrs: []schema.Attr{&Engine{V: EngineInnoDB}}},
+			to:   &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}, Attrs: []schema.Attr{&Engine{V: "INNODB"}}},
+		},
+		// Attribute was dropped from the desired state, but the current state if the default (assumed one).
+		{
+			name: "no engine changes",
+			from: &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}, Attrs: []schema.Attr{&Engine{V: EngineInnoDB}}},
+			to:   &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}},
+		},
+		// Attribute was dropped from the desired state, but the current state if the default (explicitly specific as default).
+		{
+			name: "no engine changes",
+			from: &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}, Attrs: []schema.Attr{&Engine{V: EngineMyISAM, Default: true}}},
+			to:   &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}},
+		},
+		// The current state has no engine specified (unlikely case), and the desired state is set to the default (assumed one).
+		{
+			name: "no engine changes",
+			from: &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}},
+			to:   &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}, Attrs: []schema.Attr{&Engine{V: EngineInnoDB, Default: true}}},
+		},
+		// Attributes are specified, but they do not represent the same engine.
+		{
+			name: "engine changed",
+			from: &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}, Attrs: []schema.Attr{&Engine{V: EngineInnoDB}}},
+			to:   &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}, Attrs: []schema.Attr{&Engine{V: EngineMyISAM}}},
+			wantChanges: []schema.Change{
+				&schema.ModifyAttr{
+					From: &Engine{V: EngineInnoDB},
+					To:   &Engine{V: EngineMyISAM},
+				},
+			},
+		},
+		// The desired state has no engine specified, and the current state is not the default.
+		{
+			name: "engine changed",
+			from: &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}, Attrs: []schema.Attr{&Engine{V: EngineMyISAM}}},
+			to:   &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}},
+			wantChanges: []schema.Change{
+				&schema.ModifyAttr{
+					From: &Engine{V: EngineMyISAM},
+					To:   &Engine{V: EngineInnoDB, Default: true}, // Assume InnoDB is the default.
+				},
+			},
+		},
+		// The current state has no engine specified (unlikely case), and the desired state is not the default.
+		{
+			name: "engine changed",
+			from: &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}},
+			to:   &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}, Attrs: []schema.Attr{&Engine{V: EngineMyISAM}}},
+			wantChanges: []schema.Change{
+				&schema.ModifyAttr{
+					From: &Engine{V: EngineInnoDB, Default: true}, // Assume InnoDB is the default.
+					To:   &Engine{V: EngineMyISAM},
+				},
+			},
+		},
 		{
 			name: "add collation",
 			from: &schema.Table{Name: "users", Schema: &schema.Schema{Name: "public"}, Attrs: []schema.Attr{&schema.Charset{V: "latin1"}}},
diff --git a/sql/mysql/inspect_test.go b/sql/mysql/inspect_test.go
index 7f9b3d4f..fc7cf001 100644
--- a/sql/mysql/inspect_test.go
+++ b/sql/mysql/inspect_test.go
@@ -42,11 +42,11 @@ func TestDriver_InspectTable(t *testing.T) {
 				m.ExpectQuery(queryTable).
 					WithArgs("public").
 					WillReturnRows(sqltest.Rows(`
-+--------------+--------------+--------------------+--------------------+----------------+---------------+--------------------+
-| TABLE_SCHEMA | TABLE_NAME   | CHARACTER_SET_NAME | TABLE_COLLATION    | AUTO_INCREMENT | TABLE_COMMENT | CREATE_OPTIONS     |
-+--------------+--------------+--------------------+--------------------+----------------+---------------+--------------------+
-| public       | users        | utf8mb4            | utf8mb4_0900_ai_ci | nil            | Comment       | COMPRESSION="ZLIB" |
-+--------------+--------------+--------------------+--------------------+----------------+---------------+--------------------+
++--------------+--------------+--------------------+--------------------+----------------+---------------+-------------------+------------------+------------------+
+| TABLE_SCHEMA | TABLE_NAME   | CHARACTER_SET_NAME | TABLE_COLLATION    | AUTO_INCREMENT | TABLE_COMMENT | CREATE_OPTIONS    |      ENGINE      |  DEFAULT_ENGINE  |
++--------------+--------------+--------------------+--------------------+----------------+---------------+-------------------+------------------+------------------+
+| public       | users        | utf8mb4            | utf8mb4_0900_ai_ci | nil            | Comment       | COMPRESSION="ZLIB"|       InnoDB     |       1          |
++--------------+--------------+--------------------+--------------------+----------------+---------------+-------------------+------------------+------------------+
 `))
 				m.ExpectQuery(queryColumns).
 					WithArgs("public", "users").
@@ -84,6 +84,7 @@ func TestDriver_InspectTable(t *testing.T) {
 					&schema.Collation{V: "utf8mb4_0900_ai_ci"},
 					&schema.Comment{Text: "Comment"},
 					&CreateOptions{V: `COMPRESSION="ZLIB"`},
+					&Engine{V: "InnoDB", Default: true},
 					&CreateStmt{S: "CREATE TABLE users (id bigint NOT NULL AUTO_INCREMENT) ENGINE=InnoDB AUTO_INCREMENT=55834574848 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"},
 					&AutoIncrement{V: 55834574848},
 				}, t.Attrs)
@@ -100,11 +101,11 @@ func TestDriver_InspectTable(t *testing.T) {
 				m.ExpectQuery(queryTable).
 					WithArgs("public").
 					WillReturnRows(sqltest.Rows(`
-+--------------+--------------+--------------------+--------------------+----------------+---------------+--------------------+
-| TABLE_SCHEMA | TABLE_NAME   | CHARACTER_SET_NAME | TABLE_COLLATION    | AUTO_INCREMENT | TABLE_COMMENT | CREATE_OPTIONS     |
-+--------------+--------------+--------------------+--------------------+----------------+---------------+--------------------+
-| public       | users        | utf8mb4            | utf8mb4_0900_ai_ci | nil            | Comment       | COMPRESSION="ZLIB" |
-+--------------+--------------+--------------------+--------------------+----------------+---------------+--------------------+
++--------------+--------------+--------------------+--------------------+----------------+---------------+--------------------+------------------+------------------+
+| TABLE_SCHEMA | TABLE_NAME   | CHARACTER_SET_NAME | TABLE_COLLATION    | AUTO_INCREMENT | TABLE_COMMENT | CREATE_OPTIONS     |      ENGINE      |  DEFAULT_ENGINE  |
++--------------+--------------+--------------------+--------------------+----------------+---------------+--------------------+------------------+------------------+
+| public       | users        | utf8mb4            | utf8mb4_0900_ai_ci | nil            | Comment       | COMPRESSION="ZLIB" |       InnoDB     |       1          |
++--------------+--------------+--------------------+--------------------+----------------+---------------+--------------------+------------------+------------------+
 `))
 				m.ExpectQuery(queryColumns).
 					WithArgs("public", "users").
@@ -1069,9 +1070,9 @@ func (m mock) noFKs() {
 }
 
 func (m mock) tableExists(schema, table string, exists bool) {
-	rows := sqlmock.NewRows([]string{"table_schema", "table_name", "table_collation", "character_set", "auto_increment", "table_comment", "create_options"})
+	rows := sqlmock.NewRows([]string{"table_schema", "table_name", "table_collation", "character_set", "auto_increment", "table_comment", "create_options", "engine", "default_engine"})
 	if exists {
-		rows.AddRow(schema, table, nil, nil, nil, nil, nil)
+		rows.AddRow(schema, table, nil, nil, nil, nil, nil, nil, nil)
 	}
 	m.ExpectQuery(queryTable).
 		WithArgs(schema).
@@ -1079,9 +1080,9 @@ func (m mock) tableExists(schema, table string, exists bool) {
 }
 
 func (m mock) tables(schema string, tables ...string) {
-	rows := sqlmock.NewRows([]string{"schema", "table", "charset", "collate", "inc", "comment", "options"})
+	rows := sqlmock.NewRows([]string{"schema", "table", "charset", "collate", "inc", "comment", "options", "engine", "default_engine"})
 	for _, t := range tables {
-		rows.AddRow(schema, t, nil, nil, nil, nil, nil)
+		rows.AddRow(schema, t, nil, nil, nil, nil, nil, nil, nil)
 	}
 	m.ExpectQuery(queryTable).
 		WithArgs(schema).
diff --git a/sql/mysql/migrate_test.go b/sql/mysql/migrate_test.go
index 08d65872..acd86341 100644
--- a/sql/mysql/migrate_test.go
+++ b/sql/mysql/migrate_test.go
@@ -228,6 +228,33 @@ func TestPlanChanges(t *testing.T) {
 				},
 			},
 		},
+		{
+			changes: []schema.Change{
+				&schema.AddTable{
+					T: schema.NewTable("users").
+						AddColumns(schema.NewIntColumn("id", "int")).
+						AddAttrs(&Engine{V: EngineInnoDB, Default: true}),
+				},
+				&schema.AddTable{
+					T: schema.NewTable("pets").
+						AddColumns(schema.NewIntColumn("id", "int")).
+						AddAttrs(&Engine{V: EngineMyISAM, Default: false}),
+				},
+			},
+			wantPlan: &migrate.Plan{
+				Reversible: true,
+				Changes: []*migrate.Change{
+					{
+						Cmd:     "CREATE TABLE `users` (`id` int NOT NULL)",
+						Reverse: "DROP TABLE `users`",
+					},
+					{
+						Cmd:     "CREATE TABLE `pets` (`id` int NOT NULL) ENGINE MyISAM",
+						Reverse: "DROP TABLE `pets`",
+					},
+				},
+			},
+		},
 		{
 			changes: []schema.Change{
 				func() schema.Change {
diff --git a/sql/mysql/sqlspec_test.go b/sql/mysql/sqlspec_test.go
index c5effdeb..3be38e67 100644
--- a/sql/mysql/sqlspec_test.go
+++ b/sql/mysql/sqlspec_test.go
@@ -541,6 +541,114 @@ schema "test" {
 	require.EqualValues(t, expected, string(buf))
 }
 
+func TestMarshalSpec_TableEngine(t *testing.T) {
+	s := schema.New("a8m").
+		AddTables(
+			// InnoDB (default) is not printed.
+			schema.NewTable("repos").AddAttrs(&Engine{V: EngineInnoDB, Default: true}).AddColumns(schema.NewIntColumn("id", TypeBigInt)),
+			schema.NewTable("stars").AddAttrs(&Engine{V: "INNODB", Default: true}).AddColumns(schema.NewIntColumn("id", TypeBigInt)),
+			schema.NewTable("prs").AddAttrs(&Engine{V: EngineMyISAM}).AddColumns(schema.NewIntColumn("id", TypeBigInt)),
+			schema.NewTable("issues").AddAttrs(&Engine{V: "MYISAM"}).AddColumns(schema.NewIntColumn("id", TypeBigInt)),
+			schema.NewTable("commits").AddAttrs(&Engine{V: "MyRocks"}).AddColumns(schema.NewIntColumn("id", TypeBigInt)),
+		)
+	buf, err := MarshalSpec(s, hclState)
+	require.NoError(t, err)
+	const expected = `table "repos" {
+  schema = schema.a8m
+  column "id" {
+    null = false
+    type = bigint
+  }
+}
+table "stars" {
+  schema = schema.a8m
+  column "id" {
+    null = false
+    type = bigint
+  }
+}
+table "prs" {
+  schema = schema.a8m
+  engine = MyISAM
+  column "id" {
+    null = false
+    type = bigint
+  }
+}
+table "issues" {
+  schema = schema.a8m
+  engine = MyISAM
+  column "id" {
+    null = false
+    type = bigint
+  }
+}
+table "commits" {
+  schema = schema.a8m
+  engine = "MyRocks"
+  column "id" {
+    null = false
+    type = bigint
+  }
+}
+schema "a8m" {
+}
+`
+	require.EqualValues(t, expected, string(buf))
+}
+
+func TestUnmarshalSpec_TableEngine(t *testing.T) {
+	var (
+		s schema.Schema
+		f = `table "repos" {
+  schema = schema.a8m
+  engine = InnoDB
+  column "id" {
+    null = false
+    type = bigint
+  }
+}
+table "stars" {
+  schema = schema.a8m
+  engine = "INNODB"
+  column "id" {
+    null = false
+    type = bigint
+  }
+}
+table "prs" {
+  schema = schema.a8m
+  engine = MyISAM
+  column "id" {
+    null = false
+    type = bigint
+  }
+}
+table "issues" {
+  schema = schema.a8m
+  engine = "MYISAM"
+  column "id" {
+    null = false
+    type = bigint
+  }
+}
+table "commits" {
+  schema = schema.a8m
+  engine = "MyRocks"
+  column "id" {
+    null = false
+    type = bigint
+  }
+}
+schema "a8m" {}
+`
+	)
+	require.NoError(t, EvalHCLBytes([]byte(f), &s, nil))
+	for i, e := range []string{EngineInnoDB, "INNODB", EngineMyISAM, "MYISAM", "MyRocks"} {
+		require.EqualValues(t, e, s.Tables[i].Attrs[0].(*Engine).V)
+	}
+}
+
 func TestUnmarshalSpec_IndexParts(t *testing.T) {
 	var (
 		s schema.Schema
diff --git a/sql/sqltool/testdata/flyway/v3/V3_1__fourth_migration.sql b/sql/sqltool/testdata/flyway/v3/V3_1__fourth_migration.sql
new file mode 100644
index 00000000..bd094311
--- /dev/null
+++ b/sql/sqltool/testdata/flyway/v3/V3_1__fourth_migration.sql
@@ -0,0 +1,1 @@
+ALTER TABLE tbl_2 ADD col_2 INTEGER NOT NULL;
\ No newline at end of file
diff --git a/sql/sqltool/tool_test.go b/sql/sqltool/tool_test.go
index 83cea39b..3f121eb1 100644
--- a/sql/sqltool/tool_test.go
+++ b/sql/sqltool/tool_test.go
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io/fs"
 	"testing"
+	"testing/fstest"
 	"time"
 
 	"ariga.io/atlas/sql/migrate"
@@ -206,14 +207,15 @@ func TestScanners(t *testing.T) {
 				require.NoError(t, err)
 				return d
 			}(),
-			versions:     []string{"2", "3", ""},
-			descriptions: []string{"baseline", "third_migration", "views"},
+			versions:     []string{"2", "3", "3_1", ""},
+			descriptions: []string{"baseline", "third_migration", "fourth_migration", "views"},
 			stmts: [][]string{
 				{
 					"CREATE TABLE post\n(\n    id    int NOT NULL,\n    title text,\n    body  text,\n    created_at TIMESTAMP NOT NULL\n    PRIMARY KEY (id)\n);",
 					"INSERT INTO post (title, created_at) VALUES (\n'This is\nmy multiline\n\nvalue', NOW());",
 				},
 				{"ALTER TABLE tbl_2 ADD col_1 INTEGER NOT NULL;"},
+				{"ALTER TABLE tbl_2 ADD col_2 INTEGER NOT NULL;"},
 				{"CREATE VIEW `my_view` AS SELECT * FROM `post`;"},
 			},
 		},
@@ -292,6 +294,22 @@ func TestChecksum(t *testing.T) {
 				"2_second_migration.up.sql",
 			},
 		},
+		{
+			name: "golang-migrate non-local directory",
+			dir: func() migrate.Dir {
+				fs := fstest.MapFS{
+					"1_initial.down.sql":          &fstest.MapFile{Data: []byte("1_initial.down")},
+					"1_initial.up.sql":            &fstest.MapFile{Data: []byte("1_initial.up")},
+					"2_second_migration.down.sql": &fstest.MapFile{Data: []byte("2_second_migration.down")},
+					"2_second_migration.up.sql":   &fstest.MapFile{Data: []byte("2_second_migration.up")},
+				}
+				return &sqltool.GolangMigrateDir{&fs}
+			}(),
+			files: []string{
+				"1_initial.up.sql",
+				"2_second_migration.up.sql",
+			},
+		},
 		{
 			name: "goose",
 			dir: func() migrate.Dir {
@@ -320,6 +338,27 @@ func TestChecksum(t *testing.T) {
 				"V3__third_migration.sql",
 			},
 		},
+		{
+			name: "flyway non-local directory",
+			dir: func() migrate.Dir {
+				fs := fstest.MapFS{
+					"U1__initial.sql":               &fstest.MapFile{Data: []byte("U1__initial")},
+					"V1__initial.sql":               &fstest.MapFile{Data: []byte("V1__initial")},
+					"R__views.sql":                  &fstest.MapFile{Data: []byte("R__views")},
+					"B2__baseline.sql":              &fstest.MapFile{Data: []byte("B2__baseline")},
+					"V2__second_migration.sql":      &fstest.MapFile{Data: []byte("V2__second_migration")},
+					"V3__third_migration.sql":       &fstest.MapFile{Data: []byte("V3__third_migration")},
+					"v3/V3_1__fourth_migration.sql": &fstest.MapFile{Data: []byte("V3__1_fourth_migration")},
+				}
+				return &sqltool.FlywayDir{&fs}
+			}(),
+			files: []string{
+				"B2__baseline.sql",
+				"V3__third_migration.sql",
+				"v3/V3_1__fourth_migration.sql",
+				"R__views.sql",
+			},
+		},
 		{
 			name: "liquibase",
 			dir: func() migrate.Dir {