Repository: ariga/atlas
Pull request #1176 (2022-10-06 10:42:58)

doc/md/guides: Add article "Indexes with Included Columns in PostgreSQL"


Patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate.go b/cmd/atlas/internal/cmdapi/migrate.go
index d6a5723d..b6e65e20 100644
--- a/cmd/atlas/internal/cmdapi/migrate.go
+++ b/cmd/atlas/internal/cmdapi/migrate.go
@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"io"
 	"io/fs"
+	"net/url"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -62,7 +63,7 @@ var (
 		ToURLs         []string
 		Schemas        []string
 		DirURL         string
-		DirFormat      string
+		DirFormat      string // Deprecated: format is enclosed in the dir URL now.
 		RevisionSchema string
 		Apply          struct {
 			DryRun          bool
@@ -98,7 +99,10 @@ var (
 			if err := migrateFlagsFromEnv(cmd, nil); err != nil {
 				return err
 			}
-			dir, err := dir(false)
+			if err := dirFormatBC(); err != nil {
+				return err
+			}
+			dir, err := dir(MigrateFlags.DirURL, false)
 			if err != nil {
 				return err
 			}
@@ -147,7 +151,10 @@ directory state to the desired schema. The desired state can be another connecte
 			if err := migrateFlagsFromEnv(cmd, nil); err != nil {
 				return err
 			}
-			dir, err := dir(true)
+			if err := dirFormatBC(); err != nil {
+				return err
+			}
+			dir, err := dir(MigrateFlags.DirURL, true)
 			if err != nil {
 				return err
 			}
@@ -166,23 +173,33 @@ directory state to the desired schema. The desired state can be another connecte
 		Short: "Hash (re-)creates an integrity hash file for the migration directory.",
 		Long: `'atlas migrate hash' computes the integrity hash sum of the migration directory and stores it in the atlas.sum file.
 This command should be used whenever a manual change in the migration directory was made.`,
-		Example:           `  atlas migrate hash`,
-		PersistentPreRunE: migrateFlagsFromEnv,
-		RunE:              CmdMigrateHashRun,
+		Example: `  atlas migrate hash`,
+		PersistentPreRunE: func(cmd *cobra.Command, _ []string) error {
+			if err := migrateFlagsFromEnv(cmd, nil); err != nil {
+				return err
+			}
+			if err := dirFormatBC(); err != nil {
+				return err
+			}
+			return nil
+		},
+		RunE: CmdMigrateHashRun,
 	}
 	// MigrateImportCmd represents the 'atlas migrate import' command.
 	MigrateImportCmd = &cobra.Command{
 		Use:     "import",
 		Short:   "Import a migration directory from another migration management tool to the Atlas format.",
-		Example: `  atlas migrate import --dir-format liquibase --from file:///path/to/source/directory --to file:///path/to/migration/directory`,
+		Example: `  atlas migrate import --from file:///path/to/source/directory?format=liquibase --to file:///path/to/migration/directory`,
 		// Validate the source directory. Consider a directory with no sum file valid, since it might be an import
 		// from an existing project.
 		PersistentPreRunE: func(cmd *cobra.Command, _ []string) error {
 			if err := migrateFlagsFromEnv(cmd, nil); err != nil {
 				return err
 			}
-			MigrateFlags.DirURL = MigrateFlags.Import.FromURL
-			dir, err := dir(false)
+			if err := dirFormatBC(); err != nil {
+				return err
+			}
+			dir, err := dir(MigrateFlags.Import.FromURL, false)
 			if err != nil {
 				return err
 			}
@@ -248,8 +265,16 @@ files are executed on the connected database in order to validate SQL semantics.
   atlas migrate lint --dir file:///path/to/migration/directory --dev-url mysql://root:pass@localhost:3306 --log '{{ json .Files }}'`,
 		// Override the parent 'migrate' pre-run function to allow executing
 		// 'migrate lint' on directories that are not maintained by Atlas.
-		PersistentPreRunE: migrateFlagsFromEnv,
-		RunE:              CmdMigrateLintRun,
+		PersistentPreRunE: func(cmd *cobra.Command, _ []string) error {
+			if err := migrateFlagsFromEnv(cmd, nil); err != nil {
+				return err
+			}
+			if err := dirFormatBC(); err != nil {
+				return err
+			}
+			return nil
+		},
+		RunE: CmdMigrateLintRun,
 	}
 )
 
@@ -292,7 +317,6 @@ func init() {
 	MigrateApplyCmd.Flags().SortFlags = false
 	cobra.CheckErr(MigrateApplyCmd.MarkFlagRequired(migrateFlagURL))
 	MigrateApplyCmd.MarkFlagsMutuallyExclusive(migrateFlagFrom, migrateApplyBaselineVersion)
-	cobra.CheckErr(MigrateApplyCmd.Flags().MarkHidden(migrateFlagDirFormat))
 	cobra.CheckErr(MigrateApplyCmd.Flags().MarkHidden(migrateFlagSchema))
 	// Diff flags.
 	urlFlag(&MigrateFlags.DevURL, migrateFlagDevURL, "", MigrateDiffCmd.Flags())
@@ -344,7 +368,7 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 		}
 	}
 	// Open the migration directory.
-	dir, err := dir(false)
+	dir, err := dir(MigrateFlags.DirURL, false)
 	if err != nil {
 		return err
 	}
@@ -620,20 +644,24 @@ func CmdMigrateDiffRun(cmd *cobra.Command, args []string) error {
 		defer cobra.CheckErr(unlock())
 	}
 	// Open the migration directory.
-	dir, err := dir(false)
+	u, err := url.Parse(MigrateFlags.DirURL)
 	if err != nil {
 		return err
 	}
-	// Get a state reader for the desired state.
-	desired, err := to(cmd.Context(), dev)
+	dir, err := dirURL(u, false)
 	if err != nil {
 		return err
 	}
-	defer desired.Close()
-	f, err := formatter()
+	f, err := formatter(u)
 	if err != nil {
 		return err
 	}
+	// Get a state reader for the desired state.
+	desired, err := to(cmd.Context(), dev)
+	if err != nil {
+		return err
+	}
+	defer desired.Close()
 	opts := []migrate.PlannerOption{migrate.PlanFormat(f)}
 	if dev.URL.Schema != "" {
 		// Disable tables qualifier in schema-mode.
@@ -668,7 +696,7 @@ func CmdMigrateDiffRun(cmd *cobra.Command, args []string) error {
 
 // CmdMigrateHashRun is the command executed when running the CLI with 'migrate hash' args.
 func CmdMigrateHashRun(*cobra.Command, []string) error {
-	dir, err := dir(false)
+	dir, err := dir(MigrateFlags.DirURL, false)
 	if err != nil {
 		return err
 	}
@@ -681,17 +709,19 @@ func CmdMigrateHashRun(*cobra.Command, []string) error {
 
 // CmdMigrateImportRun is the command executed when running the CLI with 'migrate import' args.
 func CmdMigrateImportRun(cmd *cobra.Command, _ []string) error {
-	if MigrateFlags.DirFormat == formatAtlas {
+	p, err := url.Parse(MigrateFlags.Import.FromURL)
+	if err != nil {
+		return err
+	}
+	if f := p.Query().Get("format"); f == "" || f == formatAtlas {
 		return fmt.Errorf("cannot import a migration directory already in %q format", formatAtlas)
 	}
-	MigrateFlags.DirURL = MigrateFlags.Import.FromURL
-	src, err := dir(false)
+	src, err := dir(MigrateFlags.Import.FromURL, false)
 	if err != nil {
 		return err
 	}
-	MigrateFlags.DirFormat = formatAtlas
-	MigrateFlags.DirURL = MigrateFlags.Import.ToURL
-	trgt, err := dir(true)
+	MigrateFlags.DirFormat = formatAtlas // remove assignment once the --dir-format flag is removed
+	trgt, err := dir(MigrateFlags.Import.ToURL, true)
 	if err != nil {
 		return err
 	}
@@ -759,11 +789,15 @@ func CmdMigrateImportRun(cmd *cobra.Command, _ []string) error {
 
 // CmdMigrateNewRun is the command executed when running the CLI with 'migrate new' args.
 func CmdMigrateNewRun(_ *cobra.Command, args []string) error {
-	dir, err := dir(true)
+	u, err := url.Parse(MigrateFlags.DirURL)
 	if err != nil {
 		return err
 	}
-	f, err := formatter()
+	dir, err := dirURL(u, true)
+	if err != nil {
+		return err
+	}
+	f, err := formatter(u)
 	if err != nil {
 		return err
 	}
@@ -776,7 +810,7 @@ func CmdMigrateNewRun(_ *cobra.Command, args []string) error {
 
 // CmdMigrateSetRun is the command executed when running the CLI with 'migrate set' args.
 func CmdMigrateSetRun(cmd *cobra.Command, args []string) error {
-	dir, err := dir(false)
+	dir, err := dir(MigrateFlags.DirURL, false)
 	if err != nil {
 		return err
 	}
@@ -910,7 +944,7 @@ func CmdMigrateSetRun(cmd *cobra.Command, args []string) error {
 
 // CmdMigrateStatusRun is the command executed when running the CLI with 'migrate status' args.
 func CmdMigrateStatusRun(cmd *cobra.Command, _ []string) error {
-	dir, err := dir(false)
+	dir, err := dir(MigrateFlags.DirURL, false)
 	if err != nil {
 		return err
 	}
@@ -951,7 +985,7 @@ func CmdMigrateValidateRun(cmd *cobra.Command, _ []string) error {
 	}
 	defer dev.Close()
 	// Currently, only our own migration file format is supported.
-	dir, err := dir(false)
+	dir, err := dir(MigrateFlags.DirURL, false)
 	if err != nil {
 		return err
 	}
@@ -977,7 +1011,7 @@ func CmdMigrateLintRun(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 	defer dev.Close()
-	dir, err := dir(false)
+	dir, err := dir(MigrateFlags.DirURL, false)
 	if err != nil {
 		return err
 	}
@@ -1049,41 +1083,75 @@ to re-hash the contents and resolve the error
 `)
 }
 
-// dir returns a migrate.Dir to use as migration directory. For now only local directories are supported.
-func dir(create bool) (migrate.Dir, error) {
-	parts := strings.SplitN(MigrateFlags.DirURL, "://", 2)
-	if len(parts) != 2 {
-		return nil, fmt.Errorf("invalid dir url %q", MigrateFlags.DirURL)
+// dir parses u and calls dirURL.
+func dir(u string, create bool) (migrate.Dir, error) {
+	parsed, err := url.Parse(u)
+	if err != nil {
+		return nil, err
+	}
+	return dirURL(parsed, create)
+}
+
+// dirURL returns a migrate.Dir to use as migration directory. For now only local directories are supported.
+func dirURL(u *url.URL, create bool) (migrate.Dir, error) {
+	if u.Scheme != "file" {
+		return nil, fmt.Errorf("unsupported driver %q", u.Scheme)
 	}
-	if parts[0] != "file" {
-		return nil, fmt.Errorf("unsupported driver %q", parts[0])
+	path := filepath.Join(u.Host, u.Path)
+	if path == "" {
+		path = "migrations"
 	}
-	f := func() (migrate.Dir, error) { return migrate.NewLocalDir(parts[1]) }
-	switch MigrateFlags.DirFormat {
+	fn := func() (migrate.Dir, error) { return migrate.NewLocalDir(path) }
+	switch f := u.Query().Get("format"); f {
 	case formatAtlas:
+		// this is the default
 	case formatGolangMigrate:
-		f = func() (migrate.Dir, error) { return sqltool.NewGolangMigrateDir(parts[1]) }
+		fn = func() (migrate.Dir, error) { return sqltool.NewGolangMigrateDir(path) }
 	case formatGoose:
-		f = func() (migrate.Dir, error) { return sqltool.NewGooseDir(parts[1]) }
+		fn = func() (migrate.Dir, error) { return sqltool.NewGooseDir(path) }
 	case formatFlyway:
-		f = func() (migrate.Dir, error) { return sqltool.NewFlywayDir(parts[1]) }
+		fn = func() (migrate.Dir, error) { return sqltool.NewFlywayDir(path) }
 	case formatLiquibase:
-		f = func() (migrate.Dir, error) { return sqltool.NewLiquibaseDir(parts[1]) }
+		fn = func() (migrate.Dir, error) { return sqltool.NewLiquibaseDir(path) }
 	case formatDBMate:
-		f = func() (migrate.Dir, error) { return sqltool.NewDBMateDir(parts[1]) }
+		fn = func() (migrate.Dir, error) { return sqltool.NewDBMateDir(path) }
 	default:
-		return nil, fmt.Errorf("unknown dir format %q", MigrateFlags.DirFormat)
+		return nil, fmt.Errorf("unknown dir format %q", f)
 	}
-	d, err := f()
+	d, err := fn()
 	if create && errors.Is(err, fs.ErrNotExist) {
-		if err := os.MkdirAll(parts[1], 0755); err != nil {
+		if err := os.MkdirAll(path, 0755); err != nil {
+			return nil, err
+		}
+		d, err = fn()
+		if err != nil {
 			return nil, err
 		}
-		d, err = f()
 	}
 	return d, err
 }
 
+// dirFormatBC ensures the soon-to-be deprecated --dir-format flag gets set on all migration directory URLs.
+func dirFormatBC() error {
+	for _, s := range []*string{
+		&MigrateFlags.DirURL,
+		&MigrateFlags.Import.FromURL,
+		&MigrateFlags.Import.ToURL,
+	} {
+		u, err := url.Parse(*s)
+		if err != nil {
+			return err
+		}
+		if !u.Query().Has("format") && MigrateFlags.DirFormat != "" {
+			q := u.Query()
+			q.Set("format", MigrateFlags.DirFormat)
+			u.RawQuery = q.Encode()
+			*s = u.String()
+		}
+	}
+	return nil
+}
+
 type target struct {
 	migrate.StateReader        // desired state.
 	io.Closer                  // optional close function.
@@ -1262,8 +1330,8 @@ const (
 	formatDBMate        = "dbmate"
 )
 
-func formatter() (migrate.Formatter, error) {
-	switch MigrateFlags.DirFormat {
+func formatter(u *url.URL) (migrate.Formatter, error) {
+	switch f := u.Query().Get("format"); f {
 	case formatAtlas:
 		return migrate.DefaultFormatter, nil
 	case formatGolangMigrate:
@@ -1277,7 +1345,7 @@ func formatter() (migrate.Formatter, error) {
 	case formatDBMate:
 		return sqltool.DBMateFormatter, nil
 	default:
-		return nil, fmt.Errorf("unknown format %q", MigrateFlags.DirFormat)
+		return nil, fmt.Errorf("unknown format %q", f)
 	}
 }
 
diff --git a/doc/md/guides/guides.md b/doc/md/guides/guides.md
index 1efc8c0f..939d9ceb 100644
--- a/doc/md/guides/guides.md
+++ b/doc/md/guides/guides.md
@@ -20,6 +20,8 @@ Welcome to the Atlas Guides.
 
 [Partial Indexes](postgres/partial-indexes.md)
 
+[Included Columns](postgres/included-columns.md)
+
 ## Migration tools
 
 [`golang-migrate`](migration-tools/golang-migrate.md)
diff --git a/doc/md/guides/migration-tools/golang-migrate.md b/doc/md/guides/migration-tools/golang-migrate.md
index adaa335a..a39cec07 100644
--- a/doc/md/guides/migration-tools/golang-migrate.md
+++ b/doc/md/guides/migration-tools/golang-migrate.md
@@ -174,11 +174,14 @@ Now, let's use Atlas's `migrate diff` command to plan a migration from the curre
 as it exists in the migrations directory to the desired state that is defined by the `schema.hcl`
 file:
 
-```text
-atlas migrate diff --dir file://migrations --dev-url mysql://root:pass@localhost:3306/dev --to file://schema.hcl --dir-format golang-migrate add_blog_posts 
+```shell
+atlas migrate diff add_blog_posts \
+  --dir "file://migrations?format=golang-migrate" \
+  --dev-url "mysql://root:pass@localhost:3306/dev" \
+  --to "file://schema.hcl" 
 ```
 
-Notice that we used the `dir-format` flag to specify that we're using `golang-migrate` as the directory format.
+Notice that we used the `format` query parameter to specify that we're using `golang-migrate` as the directory format.
 
 Hooray! Two new files were created in the migrations directory:
 ```text {5-6}
@@ -210,8 +213,11 @@ plan a migration from your current migration directory state to an existing sche
 database was available at `mysql://root:pass@some.db.io:3306/db`, a migration to the state
 of that database could be planned by running:
 
-```text
-atlas migrate diff --dir file://migrations --dev-url mysql://root:pass@localhost:3306/dev --to mysql://root:pass@some.db.io:3306/db --dir-format golang-migrate migration_name
+```shell
+atlas migrate diff migration_name \
+  --dir "file://migrations?format=golang-migrate" \
+  --dev-url "mysql://root:pass@localhost:3306/dev" \
+  --to "mysql://root:pass@some.db.io:3306/db"
 ```
 
 ## Conclusion
diff --git a/doc/md/guides/postgres/included-columns.md b/doc/md/guides/postgres/included-columns.md
new file mode 100644
index 00000000..4244c9f1
--- /dev/null
+++ b/doc/md/guides/postgres/included-columns.md
@@ -0,0 +1,319 @@
+---
+id: included-columns
+title: Indexes with Included Columns in PostgreSQL
+slug: /guides/postgres/included-columns
+---
+
+With PostgreSQL, we can create indexes with `INCLUDE` clauses, which are types of indexes that specify a list of columns to be included in the index as non-key columns. If used correctly, indexes with included columns improve performance and reduce total costs.
+
+### Basic PostgreSQL syntax for using `INCLUDE` clause with an index:
+
+```sql
+CREATE [UNIQUE] INDEX index_name
+ON table_name(key_column_list)
+INCLUDE(included_column_list);
+```
+
+### How do they work?
+
+A B-Tree index is made of two structures:
+1. The B-tree
+2. The doubly linked list at the leaf node level of the B-tree
+In an index with included columns, records of the columns mentioned in the `INCLUDE` clause are included in the doubly linked list at the leaf node level of the B-tree, corresponding to the heap (rows/tuples stored in a table).
+
+:::info
+Each index is stored separately from the table's main data area, which in PostgreSQL this is known as the table's _heap_.  Leaf pages are the pages on the lowest level of the index tree. To learn more about the B-tree index structure in PostgreSQL, visit the PostgreSQL documentation [here](https://www.postgresql.org/docs/current/btree-implementation.html#BTREE-STRUCTURE). 
+:::
+
+### When do we need them?
+
+Let's demonstrate an example where an index with an `INCLUDE` clause may be useful, by contrasting it with a unique index without an `INCLUDE` clause.
+
+First, create a table with the following command:
+
+```sql
+DROP TABLE IF EXISTS "bankdb";
+
+CREATE TABLE "bankdb" (
+  id SERIAL PRIMARY KEY,
+  savings varchar(100),
+  first_name varchar(255),
+  last_name varchar(255),
+  email varchar(255),
+  bank varchar(34)
+);
+```
+
+Here is how a portion of the table might look like after inserting values:
+
+```console title="Output"
+-[ RECORD 1 ]--------------------------------------
+id         | 1
+savings    | 28 497
+first_name | Amena
+last_name  | Gardner
+email      | a_gardner@aol.edu
+bank       | GE77159307648978112812
+-[ RECORD 2 ]--------------------------------------
+id         | 2
+savings    | 71 279
+first_name | Joan
+last_name  | Kaufman
+email      | k-joan3559@google.couk
+bank       | DK8023212366607361
+
+.
+.
+.
+-[ RECORD 1499 ]------------------------------------
+id         | 1499
+savings    | 4 880
+first_name | Ramona
+last_name  | Wilkins
+email      | r.wilkins@google.net
+bank       | BA928132235277210873
+-[ RECORD 1500 ]------------------------------------
+id         | 1500
+savings    | 69 873
+first_name | Imani
+last_name  | Noble
+email      | imaninoble@hotmail.net
+bank       | BG45LBAX41796917951361
+```
+
+Now, suppose we want to find the ID of a user by their email address. Let’s check the performance of the query with a WHERE clause without any index, with the following command:
+
+```sql
+EXPLAIN ANALYZE
+SELECT    
+    first_name,
+    last_name,
+    email
+FROM    
+    "bankdb"
+WHERE 
+    email = 'd-abbott3425@google.edu';
+```
+
+```console title="Output"
+QUERY PLAN                            
+----------------
+ Seq Scan on bankdb  (cost=0.00..38.75 rows=1 width=37) (actual time=0.180..0.181 
+rows=1 loops=1)
+   Filter: ((email)::text = 'd-abbott3425@google.edu'::text)
+   Rows Removed by Filter: 1499
+ Planning Time: 0.053 ms
+ Execution Time: 0.195 ms
+(5 rows)
+
+Time: 0.626 ms
+```
+
+Notice that the total cost is 38.75 units. If we want to use a unique index to accelerate the query, we can create it on the `email` column with the following command:
+
+```sql
+CREATE UNIQUE INDEX emails_idx 
+ON bankdb(email);
+```
+```console title="Output"
+CREATE INDEX
+Time: 10.316 ms
+```
+
+Now, let's check the performance of querying data of first and last names based on their email addresses, with the following command:
+
+```sql
+EXPLAIN ANALYZE
+SELECT    
+    first_name,
+    last_name,
+    email
+FROM    
+    "bankdb"
+WHERE 
+    email = 'd-abbott3425@google.edu';
+```
+```console title="Output"
+QUERY PLAN                   
+----------------------------------
+ Index Scan using emails_idx on bankdb  (cost=0.28..8.29 rows=1 width=37) (actual 
+time=0.200..0.203 rows=1 loops=1)
+   Index Cond: ((email)::text = 'd-abbott3425@google.edu'::text)
+ Planning Time: 0.081 ms
+ Execution Time: 0.259 ms
+(4 rows)
+
+Time: 1.470 ms
+```
+
+Notice that the total cost is now 8.29 units. The performance of the query has improved by creating a primary key index on `email` column, compared to 38.75 units without using any index. The engine still has to fetch the `first_name` and `last_name` columns from the table (also known as "heap fetches").
+Let's drop the existing index to demonstrate the next section in the article:
+
+```sql
+DROP INDEX emails_idx;
+```
+```console title="Output"
+DROP INDEX
+Time: 3.856 ms
+```
+
+Suppose we want to accelerate the same query using the `INCLUDE` clause.
+In the following command, we will create an index with an `INCLUDE` clause that precisely covers `first_name` and `last_name` columns which are part of the query for which we are trying to improve performance.
+
+```sql
+CREATE UNIQUE INDEX emails_idx 
+ON bankdb(email)
+INCLUDE(first_name,last_name);
+```
+
+```console title="Output"
+CREATE INDEX
+Time: 8.942 ms
+```
+
+Now, let's check the performance of querying data of first and last names based on their email addresses, with the following command:
+
+```sql
+EXPLAIN ANALYZE
+SELECT    
+    first_name,
+    last_name,
+    email
+FROM    
+    "bankdb"
+WHERE 
+    email = 'd-abbott3425@google.edu';
+```
+```console title="Output"
+QUERY PLAN                 
+---------------------------------------
+ Index Only Scan using emails_idx on bankdb  (cost=0.28..4.29 rows=1 width=37) (ac
+tual time=0.228..0.231 rows=1 loops=1)
+   Index Cond: (email = 'd-abbott3425@google.edu'::text)
+   Heap Fetches: 0
+ Planning Time: 0.233 ms
+ Execution Time: 0.283 ms
+(5 rows)
+```
+Notice that the total cost is now 4.29, which is significantly lower, compared to 8.29 which we got while using a unique index without the `INCLUDE` clause. We were able to reduce the total cost because the query only scanned the index in order to get the data. As a result, `heap fetches` is also zero, which means the query does not access any tables to retrieve the records. 
+
+:::info
+You might be wondering why we didn’t just use `CREATE INDEX ON bankdb(email,first_name,last_name)` instead of using the `INCLUDE` clause. One of the advantages of using the `INCLUDE` clause is having fewer levels in a  B-tree. All `INCLUDE` columns are stored in the doubly linked list of the B-tree index.
+:::
+
+### Advantages of using Indexes with an INCLUDE clause:
+1. The B-tree has fewer levels because they do not contain include columns
+2. Greatly improves performance
+3. Has the ability to return the contents of non-key columns without having to visit the index's table
+
+### Limitation of using Indexes with included columns
+- Expressions are not supported as included columns since they cannot be used in index-only scans.
+
+## Managing indexes with included columns is easy with Atlas
+
+Managing indexes and database schemas in PostgreSQL can be confusing and error-prone. Atlas is an open-source project which allows us to manage our database using a simple and easy-to-understand declarative syntax (similar to Terraform). We will now learn how to manage indexes with included columns using Atlas.
+
+:::info
+If you are just getting started, install the latest version of Atlas using the guide to [setting up Atlas](https://atlasgo.io/cli/getting-started/setting-up).
+:::
+
+### Managing Indexes with included columns in Atlas
+
+We will first use the `atlas schema inspect` command to get an HCL representation of the table which we created earlier by using the Atlas CLI:
+
+```console
+atlas schema inspect -u "postgres://postgres:mysecretpassword@localhost:5432/bankdb?sslmode=disable" > schema.hcl
+```
+```hcl title="schema.hcl"
+table "bankdb" {
+  schema = schema.public
+  column "id" {
+    null = false
+    type = serial
+  }
+  column "savings" {
+    null    = true
+    type    = varchar(100)
+  }
+  column "first_name" {
+    null    = true
+    type    = varchar(255)
+  }
+  column "last_name" {
+    null    = true
+    type    = varchar(255)
+  }
+  column "email" {
+    null    = true
+    type    = varchar(255)
+  }
+  column "bank" {
+    null = true
+    type = varchar(34)
+  }
+  primary_key {
+    columns = [column.id]
+  }
+}
+schema "public" {
+}
+```
+
+Now, lets add the following index definition to the file:
+
+```hcl
+  index "emails_idx" {
+    unique  = true
+    columns = [column.email]
+    include = [column.first_name, column.last_name]
+  }
+```
+
+Save and apply the schema changes on the database by using the `apply` command:
+
+```console
+atlas schema apply -u "postgres://postgres:mysecretpassword@localhost:5432/bankdb?sslmode=disable" -f schema.hcl
+```
+
+Atlas generates the necessary SQL statements to add the new index to the database schema. Press Enter while the Apply option is highlighted to apply the changes:
+
+```console title="Output"
+-- Planned Changes:
+-- Create index "emails_idx" to table: "bankdb"
+CREATE UNIQUE INDEX "emails_idx" ON "public"."bankdb" ("email") INCLUDE ("first_name", "last_name")
+Use the arrow keys to navigate: ↓ ↑ → ← 
+? Are you sure?: 
+  ▸ Apply
+    Abort
+```
+
+To verify that our new index was created, open the database command line tool from the previous step and run:
+
+```console
+\d bankdb;
+```
+
+```console title="Output"
+                                    Table "public.bankdb"
+ Column     |          Type         |Collation| Nullable |              Default   
+            
+------------+-----------------------+---------+----------+------------------------------------
+ id         | integer               |         | not null | nextval('bankdb_id_seq'::regclass)
+ savings    | character varying(100)|         |          | NULL::character varying
+ first_name | character varying(255)|         |          | NULL::character varying
+ last_name  | character varying(255)|         |          | NULL::character varying
+ email      | character varying(255)|         |          | NULL::character varying
+ bank       | character varying(34) |         |          | 
+Indexes:
+    "bankdb_pkey" PRIMARY KEY, btree (id)
+    "emails_idx" UNIQUE, btree (email) INCLUDE (first_name, last_name)
+```
+Amazing! Our new index with included columns is now created!
+
+## Conclusion
+
+In this section, we learned about PostgreSQL indexes with included columns and how we can easily create them in our database by using Atlas.
+
+## Need More Help?​
+
+[Join the Ariga Discord Server](https://discord.com/invite/zZ6sWVg6NT) for early access to features and the ability to provide exclusive feedback that improves your Database Management Tooling. [Sign up](https://www.getrevue.co/profile/ariga) to our newsletter to stay up to date about [Atlas](http://atlasgo.io), our OSS database schema management tool, and our cloud platform [Ariga Cloud](http://Ariga.cloud).
diff --git a/doc/md/reference.md b/doc/md/reference.md
index 3ee88eb0..4ae5cc5e 100644
--- a/doc/md/reference.md
+++ b/doc/md/reference.md
@@ -49,6 +49,7 @@ atlas migrate
 #### Flags
 ```
       --dir string           select migration directory using URL format (default "file://migrations")
+      --dir-format string    set migration file format (default "atlas")
       --env string           set which env from the project file to use
       --var <name>=<value>   input variables (default [])
 
@@ -158,7 +159,7 @@ atlas migrate import [flags]
 #### Example
 
 ```
-  atlas migrate import --dir-format liquibase --from file:///path/to/source/directory --to file:///path/to/migration/directory
+  atlas migrate import --from file:///path/to/source/directory?format=liquibase --to file:///path/to/migration/directory
 ```
 #### Flags
 ```
diff --git a/doc/md/versioned/diff.mdx b/doc/md/versioned/diff.mdx
index 592f1e32..2b75c174 100644
--- a/doc/md/versioned/diff.mdx
+++ b/doc/md/versioned/diff.mdx
@@ -57,7 +57,7 @@ values={[
 <TabItem value="migration_file">
 
 By default, migration files are named with the following format `{{ now }}_{{ name }}.sql`.
-If you wish to use a different file format, use the `--dir-format` option.
+If you wish to use a different file format, use the `format` query parameter in the directory URL.
 
 ```sql
 -- create "users" table
@@ -184,7 +184,7 @@ values={[
 <TabItem value="migration_file">
 
 By default, migration files are named with the following format `{{ now }}_{{ name }}.sql`.
-If you wish to use a different file format, use the `--dir-format` option.
+If you wish to use a different file format, use the `format` query parameter in the directory URL.
 
 ```sql
 -- create "users" table
@@ -233,15 +233,14 @@ CREATE TABLE `market`.`users` (`id` int NOT NULL, `name` varchar(255) NOT NULL)
 
 ### Generate migrations with custom formats
 
-Some migration tools use a different file format than the one used by Atlas. Therefore, Atlas provides
-an option named `--dir-format` that allows controlling the format of the migration directory.
+Some migration tools use a different file format than the one used by Atlas. You can control the format of the migration
+directory by passing in the `format` query parameter to the migration directory URL.
 
 ```shell
 atlas migrate diff create_users \
-  --dir "file://migrations" \
+  --dir "file://migrations?format=golang-migrate" \
   --to "file://schema.hcl" \
-  --dev-url "mysql://root:pass@:3306/test" \
-  --dir-format "golang-migrate"
+  --dev-url "mysql://root:pass@:3306/test"
 ```
 
 Run `ls migrations`, and you will notice Atlas created 3 files:
diff --git a/doc/md/versioned/import.mdx b/doc/md/versioned/import.mdx
index 49996366..5a1a96fa 100644
--- a/doc/md/versioned/import.mdx
+++ b/doc/md/versioned/import.mdx
@@ -14,10 +14,10 @@ a convenient command to import existing migration directories of supported tools
 
 ### Flags
 When using `atlas migrate import` to import a migration directory, users must supply multiple parameters:
-* `--from` the [URL](/concepts/url) to the migration directory to import.
+* `--from` the [URL](/concepts/url) to the migration directory to import, the `format` query parameter controls the
+migration directory format, e.g. `file://migrations?format=flyway`. Supported formats are `atlas` (default),
+`golang-migrate`, `goose`, `flyway`, `liquibase` and `dbmate`.
 * `--to` the URL of the migration directory to save imported migration files into, by default it is `file://migrations`.
-* `--dir-format` one of `[golang-migrate, goose, flyway, liquibase, dbmate]` to specifiy the migration file format of the
-migration directory to import.
 
 ### Limitations
 
@@ -174,15 +174,17 @@ CREATE VIEW `users_over_30` AS SELECT * FROM `users` where `age` > 30;
 Import existing `golang-migrate/migrate` migration directory:
 ```shell
 atlas migrate import \
-  --from "file://migrations" \
-  --to "file://atlas-migrations" \
-  -- dir-format "golang-migrate"
+  --from "file://migrations?format=golang-migrate" \
+  --to "file://atlas-migrations"
 ```
 
 Import existing Flyway migration directory:
 ```shell
 atlas migrate import \
-  --from "file://migrations" \
-  --to "file://atlas-migrations" \
-  -- dir-format "flyway"
+  --from "file://migrations?format=flyway" \
+  --to "file://atlas-migrations"
 ```
+
+### Reference
+
+[CLI Command Reference](/cli-reference#atlas-migrate-import)
diff --git a/doc/website/docusaurus.config.js b/doc/website/docusaurus.config.js
index 0842d6a2..4de87592 100644
--- a/doc/website/docusaurus.config.js
+++ b/doc/website/docusaurus.config.js
@@ -193,7 +193,7 @@ module.exports = {
           {
             to: '/guides/ddl',
             from: '/knowledge/ddl',
-          },          
+          },
         ],
       },
     ],
diff --git a/doc/website/sidebars.js b/doc/website/sidebars.js
index affd6e39..efc067e3 100644
--- a/doc/website/sidebars.js
+++ b/doc/website/sidebars.js
@@ -122,6 +122,11 @@ module.exports = {
                     id: 'guides/postgres/partial-indexes',
                     label: 'Partial Indexes'
                 },
+                {
+                    type: 'doc',
+                    id: 'guides/postgres/included-columns',
+                    label: 'Included Columns'
+                },
             ],
         },
         {
diff --git a/schemahcl/context.go b/schemahcl/context.go
index 566717b3..d23d847c 100644
--- a/schemahcl/context.go
+++ b/schemahcl/context.go
@@ -14,7 +14,6 @@ import (
 	"github.com/hashicorp/hcl/v2/hclsyntax"
 	"github.com/zclconf/go-cty/cty"
 	"github.com/zclconf/go-cty/cty/convert"
-	"github.com/zclconf/go-cty/cty/function"
 )
 
 // varDef is an HCL resource that defines an input variable to the Atlas DDL document.
@@ -37,26 +36,8 @@ func (s *State) setInputVals(ctx *hcl.EvalContext, body hcl.Body, input map[stri
 			Vars   []*varDef `hcl:"variable,block"`
 			Remain hcl.Body  `hcl:",remain"`
 		}
-		nctx = ctx.NewChild()
+		nctx = varBlockContext(ctx)
 	)
-	nctx.Variables = map[string]cty.Value{
-		"string": cty.CapsuleVal(ctyNilType, &cty.String),
-		"bool":   cty.CapsuleVal(ctyNilType, &cty.Bool),
-		"int":    cty.CapsuleVal(ctyNilType, &cty.Number),
-	}
-	nctx.Functions = map[string]function.Function{
-		"list": function.New(&function.Spec{
-			Params: []function.Parameter{
-				{Name: "type", Type: ctyNilType},
-			},
-			Type: function.StaticReturnType(ctyNilType),
-			Impl: func(args []cty.Value, _ cty.Type) (cty.Value, error) {
-				argT := args[0].EncapsulatedValue().(*cty.Type)
-				listT := cty.List(*argT)
-				return cty.CapsuleVal(ctyNilType, &listT), nil
-			},
-		}),
-	}
 	if diag := gohcl.DecodeBody(body, nctx, &doc); diag.HasErrors() {
 		return diag
 	}
@@ -245,8 +226,12 @@ var (
 	ctyRawExpr  = cty.Capsule("raw_expr", reflect.TypeOf(RawExpr{}))
 )
 
-// varBlock is the block type for variable declarations.
-const varBlock = "variable"
+// Built-in blocks.
+const (
+	varBlock     = "variable"
+	dynamicBlock = "dynamic"
+	forEachAttr  = "for_each"
+)
 
 // 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
diff --git a/schemahcl/hcl.go b/schemahcl/hcl.go
index 0282b8a7..f532c0a6 100644
--- a/schemahcl/hcl.go
+++ b/schemahcl/hcl.go
@@ -222,11 +222,33 @@ func (s *State) resource(ctx *hcl.EvalContext, file *hcl.File) (*Resource, error
 		if err != nil {
 			return nil, err
 		}
-		resource, err := s.toResource(ctx, blk, []string{blk.Type})
-		if err != nil {
-			return nil, err
+		switch {
+		case blk.Body != nil && blk.Body.Attributes[forEachAttr] != nil:
+			forEach, diags := blk.Body.Attributes[forEachAttr].Expr.Value(ctx)
+			if diags.HasErrors() {
+				return nil, diags
+			}
+			if t := forEach.Type(); !t.IsSetType() && !t.IsObjectType() {
+				return nil, fmt.Errorf("schemahcl: for_each does not support %s type", t.FriendlyName())
+			}
+			delete(blk.Body.Attributes, forEachAttr)
+			// Iterate over the set and create a resource for each element.
+			for it := forEach.ElementIterator(); it.Next(); {
+				k, v := it.Element()
+				ctx.Variables["each"] = cty.ObjectVal(map[string]cty.Value{"key": k, "value": v})
+				resource, err := s.toResource(ctx, blk, []string{blk.Type})
+				if err != nil {
+					return nil, err
+				}
+				res.Children = append(res.Children, resource)
+			}
+		default:
+			resource, err := s.toResource(ctx, blk, []string{blk.Type})
+			if err != nil {
+				return nil, err
+			}
+			res.Children = append(res.Children, resource)
 		}
-		res.Children = append(res.Children, resource)
 	}
 	return res, nil
 }
@@ -372,7 +394,7 @@ func (s *State) toResource(ctx *hcl.EvalContext, block *hclsyntax.Block, scope [
 	spec.Attrs = attrs
 	for _, blk := range block.Body.Blocks {
 		switch blk.Type {
-		case "dynamic":
+		case dynamicBlock:
 			dec, err := dynamicSpec(blk)
 			if err != nil {
 				return nil, err
diff --git a/schemahcl/opts.go b/schemahcl/opts.go
index d5da3573..73fc1c0f 100644
--- a/schemahcl/opts.go
+++ b/schemahcl/opts.go
@@ -32,8 +32,8 @@ func New(opts ...Option) *State {
 		pathVars: make(map[string]map[string]cty.Value),
 		newCtx: func() *hcl.EvalContext {
 			return &hcl.EvalContext{
+				Functions: stdFuncs(),
 				Variables: make(map[string]cty.Value),
-				Functions: make(map[string]function.Function),
 			}
 		},
 	}
diff --git a/schemahcl/stdlib.go b/schemahcl/stdlib.go
new file mode 100644
index 00000000..83b01733
--- /dev/null
+++ b/schemahcl/stdlib.go
@@ -0,0 +1,235 @@
+// Copyright 2021-present The Atlas Authors. All rights reserved.
+// This source code is licensed under the Apache 2.0 license found
+// in the LICENSE file in the root directory of this source tree.
+
+package schemahcl
+
+import (
+	"strconv"
+
+	"github.com/hashicorp/hcl/v2"
+	"github.com/hashicorp/hcl/v2/ext/tryfunc"
+	"github.com/zclconf/go-cty/cty"
+	"github.com/zclconf/go-cty/cty/convert"
+	"github.com/zclconf/go-cty/cty/function"
+	"github.com/zclconf/go-cty/cty/function/stdlib"
+)
+
+func varBlockContext(ctx *hcl.EvalContext) *hcl.EvalContext {
+	ctx = ctx.NewChild()
+	ctx.Variables = map[string]cty.Value{
+		"string": cty.CapsuleVal(ctyNilType, &cty.String),
+		"bool":   cty.CapsuleVal(ctyNilType, &cty.Bool),
+		"number": cty.CapsuleVal(ctyNilType, &cty.Number),
+		// Exists for backwards compatibility.
+		"int": cty.CapsuleVal(ctyNilType, &cty.Number),
+	}
+	ctx.Functions = map[string]function.Function{
+		"list": function.New(&function.Spec{
+			Params: []function.Parameter{
+				{Name: "elem_type", Type: ctyNilType},
+			},
+			Type: function.StaticReturnType(ctyNilType),
+			Impl: func(args []cty.Value, _ cty.Type) (cty.Value, error) {
+				argT := args[0].EncapsulatedValue().(*cty.Type)
+				listT := cty.List(*argT)
+				return cty.CapsuleVal(ctyNilType, &listT), nil
+			},
+		}),
+		"set": function.New(&function.Spec{
+			Params: []function.Parameter{
+				{Name: "elem_type", Type: ctyNilType},
+			},
+			Type: function.StaticReturnType(ctyNilType),
+			Impl: func(args []cty.Value, retType cty.Type) (cty.Value, error) {
+				argT := args[0].EncapsulatedValue().(*cty.Type)
+				setT := cty.Set(*argT)
+				return cty.CapsuleVal(ctyNilType, &setT), nil
+			},
+		}),
+		"map": function.New(&function.Spec{
+			Params: []function.Parameter{
+				{Name: "elem_type", Type: ctyNilType},
+			},
+			Type: function.StaticReturnType(ctyNilType),
+			Impl: func(args []cty.Value, retType cty.Type) (cty.Value, error) {
+				argT := args[0].EncapsulatedValue().(*cty.Type)
+				mapT := cty.Map(*argT)
+				return cty.CapsuleVal(ctyNilType, &mapT), nil
+			},
+		}),
+		"tuple": function.New(&function.Spec{
+			Params: []function.Parameter{
+				{Name: "elem_type", Type: cty.List(ctyNilType)},
+			},
+			Type: function.StaticReturnType(ctyNilType),
+			Impl: func(args []cty.Value, retType cty.Type) (cty.Value, error) {
+				argV := args[0]
+				argsT := make([]cty.Type, 0, argV.LengthInt())
+				for it := argV.ElementIterator(); it.Next(); {
+					_, ev := it.Element()
+					argsT = append(argsT, *ev.EncapsulatedValue().(*cty.Type))
+				}
+				tupleT := cty.Tuple(argsT)
+				return cty.CapsuleVal(ctyNilType, &tupleT), nil
+			},
+		}),
+		"object": function.New(&function.Spec{
+			Params: []function.Parameter{
+				{Name: "attr_type", Type: cty.Map(ctyNilType)},
+			},
+			Type: function.StaticReturnType(ctyNilType),
+			Impl: func(args []cty.Value, retType cty.Type) (cty.Value, error) {
+				argV := args[0]
+				argsT := make(map[string]cty.Type)
+				for it := argV.ElementIterator(); it.Next(); {
+					nameV, typeV := it.Element()
+					name := nameV.AsString()
+					argsT[name] = *typeV.EncapsulatedValue().(*cty.Type)
+				}
+				objT := cty.Object(argsT)
+				return cty.CapsuleVal(ctyNilType, &objT), nil
+			},
+		}),
+	}
+	return ctx
+}
+
+// standard functions exist in schemahcl language.
+func stdFuncs() map[string]function.Function {
+	return map[string]function.Function{
+		"abs":             stdlib.AbsoluteFunc,
+		"ceil":            stdlib.CeilFunc,
+		"chomp":           stdlib.ChompFunc,
+		"chunklist":       stdlib.ChunklistFunc,
+		"coalescelist":    stdlib.CoalesceListFunc,
+		"compact":         stdlib.CompactFunc,
+		"concat":          stdlib.ConcatFunc,
+		"contains":        stdlib.ContainsFunc,
+		"csvdecode":       stdlib.CSVDecodeFunc,
+		"distinct":        stdlib.DistinctFunc,
+		"element":         stdlib.ElementFunc,
+		"flatten":         stdlib.FlattenFunc,
+		"floor":           stdlib.FloorFunc,
+		"format":          stdlib.FormatFunc,
+		"formatdate":      stdlib.FormatDateFunc,
+		"formatlist":      stdlib.FormatListFunc,
+		"indent":          stdlib.IndentFunc,
+		"join":            stdlib.JoinFunc,
+		"jsondecode":      stdlib.JSONDecodeFunc,
+		"jsonencode":      stdlib.JSONEncodeFunc,
+		"keys":            stdlib.KeysFunc,
+		"log":             stdlib.LogFunc,
+		"lower":           stdlib.LowerFunc,
+		"max":             stdlib.MaxFunc,
+		"merge":           stdlib.MergeFunc,
+		"min":             stdlib.MinFunc,
+		"parseint":        stdlib.ParseIntFunc,
+		"pow":             stdlib.PowFunc,
+		"range":           stdlib.RangeFunc,
+		"regex":           stdlib.RegexFunc,
+		"regexall":        stdlib.RegexAllFunc,
+		"reverse":         stdlib.ReverseListFunc,
+		"setintersection": stdlib.SetIntersectionFunc,
+		"setproduct":      stdlib.SetProductFunc,
+		"setsubtract":     stdlib.SetSubtractFunc,
+		"setunion":        stdlib.SetUnionFunc,
+		"signum":          stdlib.SignumFunc,
+		"slice":           stdlib.SliceFunc,
+		"sort":            stdlib.SortFunc,
+		"split":           stdlib.SplitFunc,
+		"strrev":          stdlib.ReverseFunc,
+		"substr":          stdlib.SubstrFunc,
+		"timeadd":         stdlib.TimeAddFunc,
+		"title":           stdlib.TitleFunc,
+		"tostring":        makeToFunc(cty.String),
+		"tonumber":        makeToFunc(cty.Number),
+		"tobool":          makeToFunc(cty.Bool),
+		"toset":           makeToFunc(cty.Set(cty.DynamicPseudoType)),
+		"tolist":          makeToFunc(cty.List(cty.DynamicPseudoType)),
+		"trim":            stdlib.TrimFunc,
+		"trimprefix":      stdlib.TrimPrefixFunc,
+		"trimspace":       stdlib.TrimSpaceFunc,
+		"trimsuffix":      stdlib.TrimSuffixFunc,
+		"try":             tryfunc.TryFunc,
+		"upper":           stdlib.UpperFunc,
+		"values":          stdlib.ValuesFunc,
+		"zipmap":          stdlib.ZipmapFunc,
+	}
+}
+
+// makeToFunc constructs a "to..." function, like "tostring", which converts
+// its argument to a specific type or type kind. Code was copied from:
+// github.com/hashicorp/terraform/blob/master/internal/lang/funcs/conversion.go
+func makeToFunc(wantTy cty.Type) function.Function {
+	return function.New(&function.Spec{
+		Params: []function.Parameter{
+			{
+				Name: "v",
+				// We use DynamicPseudoType rather than wantTy here so that
+				// all values will pass through the function API verbatim and
+				// we can handle the conversion logic within the Type and
+				// Impl functions. This allows us to customize the error
+				// messages to be more appropriate for an explicit type
+				// conversion, whereas the cty function system produces
+				// messages aimed at _implicit_ type conversions.
+				Type:             cty.DynamicPseudoType,
+				AllowNull:        true,
+				AllowMarked:      true,
+				AllowDynamicType: true,
+			},
+		},
+		Type: func(args []cty.Value) (cty.Type, error) {
+			gotTy := args[0].Type()
+			if gotTy.Equals(wantTy) {
+				return wantTy, nil
+			}
+			conv := convert.GetConversionUnsafe(args[0].Type(), wantTy)
+			if conv == nil {
+				// We'll use some specialized errors for some trickier cases,
+				// but most we can handle in a simple way.
+				switch {
+				case gotTy.IsTupleType() && wantTy.IsTupleType():
+					return cty.NilType, function.NewArgErrorf(0, "incompatible tuple type for conversion: %s", convert.MismatchMessage(gotTy, wantTy))
+				case gotTy.IsObjectType() && wantTy.IsObjectType():
+					return cty.NilType, function.NewArgErrorf(0, "incompatible object type for conversion: %s", convert.MismatchMessage(gotTy, wantTy))
+				default:
+					return cty.NilType, function.NewArgErrorf(0, "cannot convert %s to %s", gotTy.FriendlyName(), wantTy.FriendlyNameForConstraint())
+				}
+			}
+			// If a conversion is available then everything is fine.
+			return wantTy, nil
+		},
+		Impl: func(args []cty.Value, retType cty.Type) (cty.Value, error) {
+			// We didn't set "AllowUnknown" on our argument, so it is guaranteed
+			// to be known here but may still be null.
+			ret, err := convert.Convert(args[0], retType)
+			if err != nil {
+				val, _ := args[0].UnmarkDeep()
+				// Because we used GetConversionUnsafe above, conversion can
+				// still potentially fail in here. For example, if the user
+				// asks to convert the string "a" to bool then we'll
+				// optimistically permit it during type checking but fail here
+				// once we note that the value isn't either "true" or "false".
+				gotTy := val.Type()
+				switch {
+				case gotTy == cty.String && wantTy == cty.Bool:
+					what := "string"
+					if !val.IsNull() {
+						what = strconv.Quote(val.AsString())
+					}
+					return cty.NilVal, function.NewArgErrorf(0, `cannot convert %s to bool; only the strings "true" or "false" are allowed`, what)
+				case gotTy == cty.String && wantTy == cty.Number:
+					what := "string"
+					if !val.IsNull() {
+						what = strconv.Quote(val.AsString())
+					}
+					return cty.NilVal, function.NewArgErrorf(0, `cannot convert %s to number; given string must be a decimal representation of a number`, what)
+				default:
+					return cty.NilVal, function.NewArgErrorf(0, "cannot convert %s to %s", gotTy.FriendlyName(), wantTy.FriendlyNameForConstraint())
+				}
+			}
+			return ret, nil
+		},
+	})
+}


Test patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate_test.go b/cmd/atlas/internal/cmdapi/migrate_test.go
index 3baa2f99..b6da6f99 100644
--- a/cmd/atlas/internal/cmdapi/migrate_test.go
+++ b/cmd/atlas/internal/cmdapi/migrate_test.go
@@ -39,7 +39,7 @@ func TestMigrate(t *testing.T) {
 func TestMigrate_Import(t *testing.T) {
 	for _, tool := range []string{"dbmate", "flyway", "golang-migrate", "goose", "liquibase"} {
 		p := t.TempDir()
-		t.Run(tool, func(t *testing.T) {
+		t.Run(tool, func(t *testing.T) { // remove this once --dir-format is removed. Test is kept to ensure BC.
 			path := filepath.FromSlash("testdata/import/" + tool)
 			out, err := runCmd(
 				Root, "migrate", "import",
@@ -57,6 +57,33 @@ func TestMigrate_Import(t *testing.T) {
 			require.NoError(t, err)
 			require.Equal(t, len(ex)+1, len(ac)) // sum file
 
+			for i := range ex {
+				e, err := os.ReadFile(filepath.Join(path, ex[i].Name()))
+				require.NoError(t, err)
+				a, err := os.ReadFile(filepath.Join(p, ex[i].Name()))
+				require.NoError(t, err)
+
+				require.Equal(t, string(e), string(a))
+			}
+		})
+		p = t.TempDir()
+		t.Run(tool, func(t *testing.T) {
+			path := filepath.FromSlash("testdata/import/" + tool)
+			out, err := runCmd(
+				Root, "migrate", "import",
+				"--from", fmt.Sprintf("file://%s?format=%s", path, tool),
+				"--to", "file://"+p,
+			)
+			require.NoError(t, err)
+			require.Zero(t, out)
+
+			path += "_gold"
+			ex, err := os.ReadDir(path)
+			require.NoError(t, err)
+			ac, err := os.ReadDir(p)
+			require.NoError(t, err)
+			require.Equal(t, len(ex)+1, len(ac)) // sum file
+
 			for i := range ex {
 				e, err := os.ReadFile(filepath.Join(path, ex[i].Name()))
 				require.NoError(t, err)
@@ -554,14 +581,14 @@ func TestMigrate_New(t *testing.T) {
 	require.Equal(t, 3, countFiles(t, p))
 
 	p = t.TempDir()
-	s, err = runCmd(Root, "migrate", "new", "goose", "--dir", "file://"+p, "--dir-format", formatGoose)
+	s, err = runCmd(Root, "migrate", "new", "goose", "--dir", "file://"+p+"?format="+formatGoose)
 	require.Zero(t, s)
 	require.NoError(t, err)
 	require.FileExists(t, filepath.Join(p, v+"_goose.sql"))
 	require.Equal(t, 2, countFiles(t, p))
 
 	p = t.TempDir()
-	s, err = runCmd(Root, "migrate", "new", "flyway", "--dir", "file://"+p, "--dir-format", formatFlyway)
+	s, err = runCmd(Root, "migrate", "new", "flyway", "--dir", "file://"+p+"?format="+formatFlyway)
 	require.Zero(t, s)
 	require.NoError(t, err)
 	require.FileExists(t, filepath.Join(p, fmt.Sprintf("V%s__%s.sql", v, formatFlyway)))
@@ -569,14 +596,14 @@ func TestMigrate_New(t *testing.T) {
 	require.Equal(t, 3, countFiles(t, p))
 
 	p = t.TempDir()
-	s, err = runCmd(Root, "migrate", "new", "liquibase", "--dir", "file://"+p, "--dir-format", formatLiquibase)
+	s, err = runCmd(Root, "migrate", "new", "liquibase", "--dir", "file://"+p+"?format="+formatLiquibase)
 	require.Zero(t, s)
 	require.NoError(t, err)
 	require.FileExists(t, filepath.Join(p, v+"_liquibase.sql"))
 	require.Equal(t, 2, countFiles(t, p))
 
 	p = t.TempDir()
-	s, err = runCmd(Root, "migrate", "new", "dbmate", "--dir", "file://"+p, "--dir-format", formatDBMate)
+	s, err = runCmd(Root, "migrate", "new", "dbmate", "--dir", "file://"+p+"?format="+formatDBMate)
 	require.Zero(t, s)
 	require.NoError(t, err)
 	require.FileExists(t, filepath.Join(p, v+"_dbmate.sql"))
@@ -701,8 +728,7 @@ func TestMigrate_Lint(t *testing.T) {
 	require.NoError(t, os.Rename(filepath.Join(p, "2.sql"), filepath.Join(p, "1.down.sql")))
 	s, err = runCmd(
 		Root, "migrate", "lint",
-		"--dir", "file://"+p,
-		"--dir-format", "golang-migrate",
+		"--dir", "file://"+p+"?format="+formatGolangMigrate,
 		"--dev-url", openSQLite(t, ""),
 		"--latest", "2",
 		"--log", "{{ range .Files }}{{ .Name }}:{{ len .Reports }}{{ end }}",
@@ -716,8 +742,7 @@ func TestMigrate_Lint(t *testing.T) {
 	require.NoError(t, err)
 	s, err = runCmd(
 		Root, "migrate", "lint",
-		"--dir", "file://"+p,
-		"--dir-format", "golang-migrate",
+		"--dir", "file://"+p+"?format="+formatGolangMigrate,
 		"--dev-url", openSQLite(t, ""),
 		"--latest", "1",
 	)
diff --git a/cmd/atlas/internal/cmdapi/schema_test.go b/cmd/atlas/internal/cmdapi/schema_test.go
index 31b7b652..91b3fcd2 100644
--- a/cmd/atlas/internal/cmdapi/schema_test.go
+++ b/cmd/atlas/internal/cmdapi/schema_test.go
@@ -115,6 +115,7 @@ func TestSchema_Clean(t *testing.T) {
 	require.NoError(t, err)
 
 	// Apply migrations onto database.
+	MigrateFlags.Apply.BaselineVersion = ""
 	_, err = runCmd(Root, "migrate", "apply", "--dir", "file://testdata/sqlite", "--url", u)
 	require.NoError(t, err)
 
diff --git a/internal/integration/testdata/mysql/cli-migrate-diff-format.txt b/internal/integration/testdata/mysql/cli-migrate-diff-format.txt
index 6ed058a4..942201fd 100644
--- a/internal/integration/testdata/mysql/cli-migrate-diff-format.txt
+++ b/internal/integration/testdata/mysql/cli-migrate-diff-format.txt
@@ -2,6 +2,7 @@ only maria103
 
 mkdir migrations
 
+# old behavior still works
 atlas migrate diff --dev-url URL --to file://1.hcl --dir-format golang-migrate first
 cmpmig 0 golang-migrate/1.down.sql
 cmpmig 1 golang-migrate/1.up.sql
@@ -13,39 +14,50 @@ cmpmig 3 golang-migrate/2.up.sql
 rm migrations
 mkdir migrations
 
-atlas migrate diff --dev-url URL --to file://1.hcl --dir-format goose first
+atlas migrate diff --dev-url URL --to file://1.hcl --dir file://migrations?format=golang-migrate first
+cmpmig 0 golang-migrate/1.down.sql
+cmpmig 1 golang-migrate/1.up.sql
+
+atlas migrate diff --dev-url URL --to file://2.hcl --dir file://migrations?format=golang-migrate second
+cmpmig 2 golang-migrate/2.down.sql
+cmpmig 3 golang-migrate/2.up.sql
+
+rm migrations
+mkdir migrations
+
+atlas migrate diff --dev-url URL --to file://1.hcl --dir file://migrations?format=goose first
 cmpmig 0 goose/1.sql
 
-atlas migrate diff --dev-url URL --to file://2.hcl --dir-format goose second
+atlas migrate diff --dev-url URL --to file://2.hcl --dir file://migrations?format=goose second
 cmpmig 1 goose/2.sql
 
 rm migrations
 mkdir migrations
 
-atlas migrate diff --dev-url URL --to file://1.hcl --dir-format dbmate first
+atlas migrate diff --dev-url URL --to file://1.hcl --dir file://migrations?format=dbmate first
 cmpmig 0 dbmate/1.sql
 
-atlas migrate diff --dev-url URL --to file://2.hcl --dir-format dbmate second
+atlas migrate diff --dev-url URL --to file://2.hcl --dir file://migrations?format=dbmate second
 cmpmig 1 dbmate/2.sql
 
 rm migrations
 mkdir migrations
 
-atlas migrate diff --dev-url URL --to file://1.hcl --dir-format flyway first
+atlas migrate diff --dev-url URL --to file://1.hcl --dir file://migrations?format=flyway first
 cmpmig 0 flyway/U1.sql
 cmpmig 1 flyway/V1.sql
 
-atlas migrate diff --dev-url URL --to file://2.hcl --dir-format flyway second
+atlas migrate diff --dev-url URL --to file://2.hcl --dir file://migrations?format=flyway second
 cmpmig 1 flyway/U2.sql
 cmpmig 3 flyway/V2.sql
 
 rm migrations
 mkdir migrations
 
-atlas migrate diff --dev-url URL --to file://1.hcl --dir-format liquibase first
+atlas migrate diff --dev-url URL --to file://1.hcl --dir file://migrations?format=liquibase first
 cmpmig 0 liquibase/1.sql
 
-atlas migrate diff --dev-url URL --to file://2.hcl --dir-format liquibase second
+atlas migrate diff --dev-url URL --to file://2.hcl --dir file://migrations?format=liquibase second
 cmpmig 1 liquibase/2.sql
 
 -- golang-migrate/1.up.sql --
diff --git a/schemahcl/hcl_test.go b/schemahcl/hcl_test.go
index 62b1eb82..2b9d7fff 100644
--- a/schemahcl/hcl_test.go
+++ b/schemahcl/hcl_test.go
@@ -423,3 +423,83 @@ env "staging" {
 	})
 	require.EqualError(t, err, `variable "domains": list of string required`)
 }
+
+func TestForEachResources(t *testing.T) {
+	type (
+		Env struct {
+			Name string `spec:",name"`
+			URL  string `spec:"url"`
+		}
+	)
+	var (
+		doc struct {
+			Envs []*Env `spec:"env"`
+		}
+		b = []byte(`
+variable "tenants" {
+  type    = list(string)
+  default = ["atlas", "ent"]
+}
+
+variable "domains" {
+  type = list(object({
+    name = string
+    port = number
+  }))
+  default = [
+    {
+      name = "atlasgo.io"
+      port = 443
+    },
+    {
+      name = "entgo.io"
+      port = 443
+    },
+  ]
+}
+
+env "prod" {
+  for_each = toset(var.tenants)
+  url = "mysql://root:pass@:3306/${each.value}"
+}
+
+env "staging" {
+  for_each = toset(var.domains)
+  url = "${each.value.name}:${each.value.port}"
+}
+
+env "dev" {
+  for_each = {
+    atlas = "atlasgo.io"
+    ent   = "entgo.io"
+  }
+  url = "${each.value}/${each.key}"
+}
+`)
+	)
+	require.NoError(t, New().EvalBytes(b, &doc, nil))
+	require.Len(t, doc.Envs, 6)
+	require.Equal(t, "prod", doc.Envs[0].Name)
+	require.EqualValues(t, doc.Envs[0].URL, "mysql://root:pass@:3306/atlas")
+	require.Equal(t, "prod", doc.Envs[1].Name)
+	require.EqualValues(t, doc.Envs[1].URL, "mysql://root:pass@:3306/ent")
+	require.Equal(t, "staging", doc.Envs[2].Name)
+	require.EqualValues(t, doc.Envs[2].URL, "atlasgo.io:443")
+	require.Equal(t, "staging", doc.Envs[3].Name)
+	require.EqualValues(t, doc.Envs[3].URL, "entgo.io:443")
+	require.Equal(t, "dev", doc.Envs[4].Name)
+	require.EqualValues(t, doc.Envs[4].URL, "atlasgo.io/atlas")
+	require.Equal(t, "dev", doc.Envs[5].Name)
+	require.EqualValues(t, doc.Envs[5].URL, "entgo.io/ent")
+
+	// Mismatched element types.
+	err := New().EvalBytes(b, &doc, map[string]cty.Value{
+		"domains": cty.ListVal([]cty.Value{
+			cty.ObjectVal(map[string]cty.Value{
+				"name": cty.StringVal("a"),
+				"port": cty.StringVal("b"),
+			}),
+		}),
+	})
+	require.EqualError(t, err, `variable "domains": a number is required`)
+}
diff --git a/sql/migrate/migrate_test.go b/sql/migrate/migrate_test.go
index 16f7f9a1..75aabb53 100644
--- a/sql/migrate/migrate_test.go
+++ b/sql/migrate/migrate_test.go
@@ -147,13 +147,14 @@ func TestPlanner_PlanSchema(t *testing.T) {
 
 func TestExecutor_Replay(t *testing.T) {
 	ctx := context.Background()
-	d, err := migrate.NewLocalDir("testdata/migrate")
+	d, err := migrate.NewLocalDir(filepath.FromSlash("testdata/migrate"))
 	require.NoError(t, err)
 
 	drv := &mockDriver{}
 	ex, err := migrate.NewExecutor(drv, d, migrate.NopRevisionReadWriter{})
 	require.NoError(t, err)
 
+	// No options.
 	_, err = ex.Replay(ctx, migrate.RealmConn(drv, nil))
 	require.NoError(t, err)
 	require.Equal(t, []string{"DROP TABLE IF EXISTS t;", "CREATE TABLE t(c int);"}, drv.executed)
@@ -318,7 +319,7 @@ func TestExecutor(t *testing.T) {
 			OperatorVersion: "op",
 		}
 	)
-	dir, err = migrate.NewLocalDir(filepath.Join("testdata/migrate", "sub"))
+	dir, err = migrate.NewLocalDir(filepath.Join("testdata", "migrate", "sub"))
 	require.NoError(t, err)
 	ex, err = migrate.NewExecutor(drv, dir, rrw, migrate.WithLogger(log), migrate.WithOperatorVersion("op"))
 	require.NoError(t, err)