Repository: ariga/atlas
Pull request #808 (2022-05-18 13:27:24)

sql/migrate: remove GlobStateReader and replace by Executor
This removes the deprecated GlobStateReader and replaces it by the Executor. The Executor is not a StateReader (it replays the migration directory on the dev-db and inspects it).

This enables us in the future to support reading other formats like Liquibase etc as well and ease transition from those tools to Atlas. We will also be able to "rewrite" the Liquibase to the Atlas format (like creating a snapshot e.g.).

This PR is also needed for the CLI to work since the Executor can support multi-schema setups while this was not possible before.

a8m: Please, fix CI failures and I'll approve it 😄 

Patch:
diff --git a/cmd/atlascmd/atlascmd.go b/cmd/atlascmd/atlascmd.go
index 201e0ab1..7789d0e8 100644
--- a/cmd/atlascmd/atlascmd.go
+++ b/cmd/atlascmd/atlascmd.go
@@ -29,8 +29,15 @@ var (
 	// Should be set by build script "-X 'ariga.io/atlas/cmd/action.version=${version}'"
 	version string
 
-	// selectedEnv stores the value for the global --env flag.
-	selectedEnv string
+	// GlobalFlags contains flags common to many Atlas sub-commands.
+	GlobalFlags struct {
+		// SelectedEnv contains the environment selected from the active
+		// project via the --env flag.
+		SelectedEnv string
+		// Vars contains the input variables passed from the CLI to
+		// Atlas DDL or project files.
+		Vars map[string]string
+	}
 
 	// schemaCmd represents the subcommand 'atlas version'.
 	versionCmd = &cobra.Command{
@@ -78,7 +85,38 @@ func init() {
 
 // receivesEnv configures cmd to receive the common '--env' flag.
 func receivesEnv(cmd *cobra.Command) {
-	cmd.PersistentFlags().StringVarP(&selectedEnv, "env", "", "", "set which env from the project file to use")
+	cmd.PersistentFlags().StringVarP(&GlobalFlags.SelectedEnv, "env", "", "", "set which env from the project file to use")
+	cmd.PersistentFlags().StringToStringVarP(&GlobalFlags.Vars, varFlag, "", nil, "input variables")
+}
+
+// inputValsFromEnv populates GlobalFlags.Vars from the active environment. If we are working
+// inside a project, the "var" flag is not propagated to the schema definition. Instead, it
+// is used to evaluate the project file which can pass input values via the "values" block
+// to the schema.
+func inputValsFromEnv(cmd *cobra.Command) error {
+	activeEnv, err := selectEnv(GlobalFlags.SelectedEnv)
+	if err != nil {
+		return err
+	}
+	if fl := cmd.Flag(varFlag); fl == nil {
+		return nil
+	}
+	values, err := activeEnv.asMap()
+	if err != nil {
+		return err
+	}
+	if len(values) == 0 {
+		return nil
+	}
+	pairs := make([]string, 0, len(values))
+	for k, v := range values {
+		pairs = append(pairs, fmt.Sprintf("%s=%s", k, v))
+	}
+	vars := strings.Join(pairs, ",")
+	if err := cmd.Flags().Set(varFlag, vars); err != nil {
+		return err
+	}
+	return nil
 }
 
 // parse returns a user facing version and release notes url
diff --git a/cmd/atlascmd/migrate.go b/cmd/atlascmd/migrate.go
index 173976ea..b2be3f41 100644
--- a/cmd/atlascmd/migrate.go
+++ b/cmd/atlascmd/migrate.go
@@ -6,6 +6,7 @@ package atlascmd
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -242,12 +243,16 @@ func CmdMigrateValidateRun(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 	defer dev.Close()
-	// We are using a migrate.GlobStateReader to replay the migration directory.
+	// Currently, only our own migration file format is supported.
 	dir, err := dir()
 	if err != nil {
 		return err
 	}
-	if _, err := migrate.GlobStateReader(dir, dev.Driver, "*.sql").ReadState(cmd.Context()); err != nil {
+	ex, err := migrate.NewExecutor(dev.Driver, dir, migrate.NoopRevisionReadWriter{})
+	if err != nil {
+		return err
+	}
+	if _, err := ex.ReadState(cmd.Context()); err != nil && !errors.Is(err, migrate.ErrNoPendingFiles) {
 		return fmt.Errorf("replaying the migration directory: %w", err)
 	}
 	return nil
@@ -347,10 +352,13 @@ func formatter() (migrate.Formatter, error) {
 }
 
 func migrateFlagsFromEnv(cmd *cobra.Command, _ []string) error {
-	activeEnv, err := selectEnv(selectedEnv)
+	activeEnv, err := selectEnv(GlobalFlags.SelectedEnv)
 	if err != nil {
 		return err
 	}
+	if err := inputValsFromEnv(cmd); err != nil {
+		return err
+	}
 	if err := maySetFlag(cmd, migrateFlagDevURL, activeEnv.DevURL); err != nil {
 		return err
 	}
diff --git a/cmd/atlascmd/project.go b/cmd/atlascmd/project.go
index cc212d12..c881871f 100644
--- a/cmd/atlascmd/project.go
+++ b/cmd/atlascmd/project.go
@@ -14,6 +14,20 @@ import (
 
 const projectFileName = "atlas.hcl"
 
+type loadConfig struct {
+	inputVals map[string]string
+}
+
+// LoadOption configures the LoadEnv function.
+type LoadOption func(*loadConfig)
+
+// WithInput is a LoadOption that sets the input values for the LoadEnv function.
+func WithInput(vals map[string]string) LoadOption {
+	return func(config *loadConfig) {
+		config.inputVals = vals
+	}
+}
+
 // projectFile represents an atlas.hcl file.
 type projectFile struct {
 	Envs []*Env `spec:"env"`
@@ -25,6 +39,22 @@ type MigrationDir struct {
 	Format string `spec:"format"`
 }
 
+// asMap returns the extra attributes stored in the Env as a map[string]string.
+func (e *Env) asMap() (map[string]string, error) {
+	m := make(map[string]string, len(e.Extra.Attrs))
+	for _, attr := range e.Extra.Attrs {
+		if v, err := attr.String(); err == nil {
+			m[attr.K] = v
+			continue
+		}
+		if lv, ok := attr.V.(*schemaspec.LiteralValue); ok {
+			m[attr.K] = lv.V
+		}
+		return nil, fmt.Errorf("expecting attr %q to be a literal, got: %T", attr.K, attr.V)
+	}
+	return m, nil
+}
+
 // Env represents an Atlas environment.
 type Env struct {
 	// Name for this environment.
@@ -55,13 +85,17 @@ var hclState = schemahcl.New(
 
 // LoadEnv reads the project file in path, and loads the environment
 // with the provided name into env.
-func LoadEnv(path string, name string) (*Env, error) {
+func LoadEnv(path string, name string, opts ...LoadOption) (*Env, error) {
+	cfg := &loadConfig{}
+	for _, f := range opts {
+		f(cfg)
+	}
 	b, err := os.ReadFile(path)
 	if err != nil {
 		return nil, err
 	}
 	var project projectFile
-	if err := hclState.Eval(b, &project, nil); err != nil {
+	if err := hclState.Eval(b, &project, cfg.inputVals); err != nil {
 		return nil, fmt.Errorf("error reading project file: %w", err)
 	}
 	projEnvs := make(map[string]*Env)
diff --git a/cmd/atlascmd/schema.go b/cmd/atlascmd/schema.go
index 87622c8c..3707e15e 100644
--- a/cmd/atlascmd/schema.go
+++ b/cmd/atlascmd/schema.go
@@ -28,6 +28,7 @@ const (
 	devURLFlag = "dev-url"
 	fileFlag   = "file"
 	dsnFlag    = "dsn"
+	varFlag    = "var"
 )
 
 var (
@@ -56,7 +57,6 @@ var (
 		DryRun      bool
 		AutoApprove bool
 		Verbose     bool
-		Vars        map[string]string
 	}
 	// SchemaApply represents the 'atlas schema apply' subcommand command.
 	SchemaApply = &cobra.Command{
@@ -145,7 +145,6 @@ func init() {
 	SchemaApply.Flags().BoolVarP(&ApplyFlags.Web, "web", "w", false, "Open in a local Atlas UI.")
 	SchemaApply.Flags().StringVarP(&ApplyFlags.Addr, "addr", "", ":5800", "used with -w, local address to bind the server to.")
 	SchemaApply.Flags().BoolVarP(&ApplyFlags.Verbose, migrateDiffFlagVerbose, "", false, "enable verbose logging")
-	SchemaApply.Flags().StringToStringVarP(&ApplyFlags.Vars, "var", "", nil, "input variables")
 	SchemaApply.Flags().StringVarP(&SchemaFlags.DSN, dsnFlag, "d", "", "")
 	cobra.CheckErr(SchemaApply.Flags().MarkHidden(dsnFlag))
 	cobra.CheckErr(SchemaApply.MarkFlagRequired(urlFlag))
@@ -178,14 +177,17 @@ func selectEnv(selected string) (*Env, error) {
 	if _, err := os.Stat(projectFileName); os.IsNotExist(err) {
 		return env, nil
 	}
-	return LoadEnv(projectFileName, selected)
+	return LoadEnv(projectFileName, selected, WithInput(GlobalFlags.Vars))
 }
 
 func schemaFlagsFromEnv(cmd *cobra.Command, _ []string) error {
-	activeEnv, err := selectEnv(selectedEnv)
+	activeEnv, err := selectEnv(GlobalFlags.SelectedEnv)
 	if err != nil {
 		return err
 	}
+	if err := inputValsFromEnv(cmd); err != nil {
+		return err
+	}
 	if err := dsn2url(cmd); err != nil {
 		return err
 	}
@@ -272,11 +274,8 @@ func CmdApplyRun(cmd *cobra.Command, _ []string) error {
 	if err != nil {
 		return err
 	}
-	if err != nil {
-		return err
-	}
 	defer c.Close()
-	return applyRun(cmd.Context(), c, ApplyFlags.DevURL, ApplyFlags.File, ApplyFlags.DryRun, ApplyFlags.AutoApprove, ApplyFlags.Vars)
+	return applyRun(cmd.Context(), c, ApplyFlags.DevURL, ApplyFlags.File, ApplyFlags.DryRun, ApplyFlags.AutoApprove, GlobalFlags.Vars)
 }
 
 // CmdFmtRun formats all HCL files in a given directory using canonical HCL formatting
diff --git a/doc/md/cli/reference.md b/doc/md/cli/reference.md
index 8358b803..217e8c15 100644
--- a/doc/md/cli/reference.md
+++ b/doc/md/cli/reference.md
@@ -60,11 +60,12 @@ atlas migrate
 
 #### Flags
 ```
-      --dir string       select migration directory using URL format (default "file://migrations")
-      --env string       set which env from the project file to use
-      --force            force a command to run on a broken migration directory state
-      --format string    set migration file format (default "atlas")
-      --schema strings   set schema names
+      --dir string           select migration directory using URL format (default "file://migrations")
+      --env string           set which env from the project file to use
+      --force                force a command to run on a broken migration directory state
+      --format string        set migration file format (default "atlas")
+      --schema strings       set schema names
+      --var stringToString   input variables (default [])
 
 ```
 
@@ -177,7 +178,8 @@ The `atlas schema` command groups subcommands for working with Atlas schemas.
 
 #### Flags
 ```
-      --env string   set which env from the project file to use
+      --env string           set which env from the project file to use
+      --var stringToString   input variables (default [])
 
 ```
 
@@ -211,18 +213,17 @@ migration.
 ```
 #### Flags
 ```
-  -f, --file string          [/path/to/file] file containing the HCL schema.
-  -u, --url string           URL to the database using the format:
-                             [driver://username:password@address/dbname?param=value]
-  -s, --schema strings       Set schema names.
-      --dev-url string       URL for the dev database. Used to validate schemas and calculate diffs
-                             before running migration.
-      --dry-run              Dry-run. Print SQL plan without prompting for execution.
-      --auto-approve         Auto approve. Apply the schema changes without prompting for approval.
-  -w, --web                  Open in a local Atlas UI.
-      --addr string          used with -w, local address to bind the server to. (default ":5800")
-      --verbose              enable verbose logging
-      --var stringToString   input variables (default [])
+  -f, --file string      [/path/to/file] file containing the HCL schema.
+  -u, --url string       URL to the database using the format:
+                         [driver://username:password@address/dbname?param=value]
+  -s, --schema strings   Set schema names.
+      --dev-url string   URL for the dev database. Used to validate schemas and calculate diffs
+                         before running migration.
+      --dry-run          Dry-run. Print SQL plan without prompting for execution.
+      --auto-approve     Auto approve. Apply the schema changes without prompting for approval.
+  -w, --web              Open in a local Atlas UI.
+      --addr string      used with -w, local address to bind the server to. (default ":5800")
+      --verbose          enable verbose logging
 
 ```
 
diff --git a/go.mod b/go.mod
index 341477a9..06afbfd1 100644
--- a/go.mod
+++ b/go.mod
@@ -32,5 +32,7 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e // indirect
 	golang.org/x/text v0.3.7 // indirect
+	golang.org/x/tools v0.1.9-0.20211216111533-8d383106f7e7 // indirect
+	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
 )
diff --git a/go.sum b/go.sum
index 1a759659..095ef9ad 100644
--- a/go.sum
+++ b/go.sum
@@ -1,5 +1,3 @@
-entgo.io/ent v0.10.2-0.20220428114225-0c8f9a977c99 h1:++Y53GODlpz4OZOJ4VjANc9OxKUR6NZRgXcFg+CEtOc=
-entgo.io/ent v0.10.2-0.20220428114225-0c8f9a977c99/go.mod h1:rUH4yiUVADR2XHnv0ZERDnwF/YjR42m3Qra6CZ4ZxBA=
 entgo.io/ent v0.10.2-0.20220502113020-4ac82f5bb3f0 h1:qHA4+ANAzDj6BcDLxNgZuzKxFre/RI9r5wwsI2O+1M4=
 entgo.io/ent v0.10.2-0.20220502113020-4ac82f5bb3f0/go.mod h1:Zh61BPvB+cL6VWEyN8f1YoDacrMjQf2KDlDeX26xq2k=
 github.com/DATA-DOG/go-sqlmock v1.5.0 h1:Shsta01QNfFxHCfpW6YH2STWB0MudeXXEWMr20OEh60=
@@ -16,7 +14,6 @@ github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e h1:fY5BOSpyZCqRo5O
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1 h1:q763qf9huN11kDQavWsoZXJNW3xEE4JJyHa5Q25/sd8=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
-github.com/cpuguy83/go-md2man/v2 v2.0.1 h1:r/myEWzV9lfsM1tFLgDyu0atFtJ1fXn261LKYj/3DxU=
 github.com/cpuguy83/go-md2man/v2 v2.0.1/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -60,7 +57,6 @@ github.com/mitchellh/go-wordwrap v0.0.0-20150314170334-ad45545899c7 h1:DpOJ2HYzC
 github.com/mitchellh/go-wordwrap v0.0.0-20150314170334-ad45545899c7/go.mod h1:ZXFpozHsX6DPmq2I0TCekCxypsnAUbP2oI0UX1GXzOo=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/sergi/go-diff v1.0.0 h1:Kpca3qRNrduNnOQeazBd0ysaKrUJiIuISHxogkT9RPQ=
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
@@ -102,14 +98,15 @@ golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.1.9-0.20211216111533-8d383106f7e7 h1:M1gcVrIb2lSn2FIL19DG0+/b8nNVKJ7W7b4WcAGZAYM=
+golang.org/x/tools v0.1.9-0.20211216111533-8d383106f7e7/go.mod h1:nABZi5QlRsZVlzPpHl034qft6wpY4eDcsTt5AaioBiU=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.6.5/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
diff --git a/schema/schemaspec/extension.go b/schema/schemaspec/extension.go
index c3f8edb8..638cd7a4 100644
--- a/schema/schemaspec/extension.go
+++ b/schema/schemaspec/extension.go
@@ -649,6 +649,9 @@ func isSingleResource(t reflect.Type) bool {
 		if _, ok := f.Tag.Lookup("spec"); ok {
 			return true
 		}
+		if f.Type == reflect.TypeOf(DefaultExtension{}) {
+			return true
+		}
 	}
 	return false
 }
diff --git a/sql/migrate/migrate.go b/sql/migrate/migrate.go
index aaa295de..cd860a63 100644
--- a/sql/migrate/migrate.go
+++ b/sql/migrate/migrate.go
@@ -183,11 +183,11 @@ type (
 	// Planner can plan the steps to take to migrate from one state to another. It uses the enclosed FS to write
 	// those changes to versioned migration files.
 	Planner struct {
-		drv Driver      // driver to use
-		dir Dir         // where migration files are stored and read from
-		fmt Formatter   // how to format a plan to migration files
-		dsr StateReader // how to read a state from the migration directory
-		sum bool        // whether to create a sum file for the migration directory
+		drv Driver    // driver to use
+		dir Dir       // where migration files are stored and read from
+		fmt Formatter // how to format a plan to migration files
+		sc  Scanner   // how to interpret a migration dir
+		sum bool      // whether to create a sum file for the migration directory
 	}
 
 	// PlannerOption allows managing a Planner using functional arguments.
@@ -242,9 +242,6 @@ func NewPlanner(drv Driver, dir Dir, opts ...PlannerOption) *Planner {
 	if p.fmt == nil {
 		p.fmt = DefaultFormatter
 	}
-	if p.dsr == nil {
-		p.dsr = GlobStateReader(p.dir, p.drv, "*.sql")
-	}
 	return p
 }
 
@@ -255,13 +252,6 @@ func WithFormatter(fmt Formatter) PlannerOption {
 	}
 }
 
-// WithStateReader sets the StateReader of a Planner.
-func WithStateReader(dsr StateReader) PlannerOption {
-	return func(p *Planner) {
-		p.dsr = dsr
-	}
-}
-
 // DisableChecksum disables the hash-sum functionality for the migration directory.
 func DisableChecksum() PlannerOption {
 	return func(p *Planner) {
@@ -272,7 +262,11 @@ func DisableChecksum() PlannerOption {
 // Plan calculates the migration Plan required for moving the current state (from) state to
 // the next state (to). A StateReader can be a directory, static schema elements or a Driver connection.
 func (p *Planner) Plan(ctx context.Context, name string, to StateReader) (*Plan, error) {
-	current, err := p.dsr.ReadState(ctx)
+	ex, err := NewExecutor(p.drv, p.dir, NoopRevisionReadWriter{})
+	if err != nil {
+		return nil, err
+	}
+	current, err := ex.ReadState(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -392,9 +386,13 @@ func (e *Executor) Execute(ctx context.Context, n int) (err error) {
 	}
 	// Check if the existing revisions did come from the migration directory and not from somewhere else.
 	hf, err := readHashFile(e.dir)
-	if err != nil {
+	if err != nil && !errors.Is(err, os.ErrNotExist) {
 		return fmt.Errorf("sql/migrate: execute: read atlas.sum file: %w", err)
 	}
+	// If there is no atlas.sum file there are no migration files.
+	if errors.Is(err, os.ErrNotExist) {
+		return ErrNoPendingFiles
+	}
 	// For up to len(revisions) revisions the migration files must both match in order and content.
 	if len(revisions) > len(migrations) { // TODO(masseelch): keep compaction in mind.
 		return errors.New("sql/migrate: execute: revisions and migrations mismatch: more revisions than migrations")
@@ -488,103 +486,87 @@ func (e *Executor) Execute(ctx context.Context, n int) (err error) {
 	return
 }
 
-// GlobStateReader creates a StateReader that loads all files from Dir matching
-// glob in lexicographic order and uses the Driver to create a migration state.
-//
-// If the given Driver implements the Emptier interface the IsEmpty method will be used to determine if the Driver is
-// connected to an "empty" database. This behavior was added to support SQLite flavors.
+// ErrNotClean is returned when the connected dev-db is not in a clean state (aka it has schemas and tables).
+// This check is done to ensure no data is lost by overriding it when working on the dev-dn.
+var ErrNotClean = errors.New("sql/migrate: connected database is not clean")
+
+// ReadState implements the StateReader interface.
 //
-// Deprecated: GlobStateReader will be removed once the Executor is functional.
-func GlobStateReader(dir Dir, drv Driver, glob string) StateReaderFunc {
-	var errNotClean = errors.New("sql/migrate: connected database is not clean")
-	return func(ctx context.Context) (realm *schema.Realm, err error) {
-		// Collect the migration files.
-		names, err := fs.Glob(dir, glob)
+// It does so by calling Execute and inspecting the database thereafter.
+func (e *Executor) ReadState(ctx context.Context) (realm *schema.Realm, err error) {
+	unlock, err := e.drv.(schema.Locker).Lock(ctx, "atlas_migration_directory_state", 0)
+	if err != nil {
+		return nil, fmt.Errorf("sql/migrate: acquiring database lock: %w", err)
+	}
+	defer unlock()
+	// We need an empty database state to reliably replay the migration directory.
+	if c, ok := e.drv.(interface {
+		// The IsClean method can be added to a Driver to override how to
+		// determine if a connected database is in a clean state.
+		// This interface exists only to support SQLite favors.
+		IsClean(context.Context) (bool, error)
+	}); ok {
+		e, err := c.IsClean(ctx)
+		if err != nil {
+			return nil, fmt.Errorf("sql/migrate: checking database state: %w", err)
+		}
+		if !e {
+			return nil, ErrNotClean
+		}
+	} else {
+		realm, err = e.drv.InspectRealm(ctx, nil)
 		if err != nil {
 			return nil, err
 		}
-		// We need an empty database state to reliably replay the migration directory.
-		if c, ok := drv.(interface {
-			// The IsClean method can be added to a Driver to override how to
-			// determine if a connected database is in a clean state.
+		if len(realm.Schemas) > 0 {
+			return nil, ErrNotClean
+		}
+	}
+	// Clean up after ourselves.
+	defer func() {
+		if e, ok := e.drv.(interface {
+			// The Clean method can be added to a Driver to change how to clean a database.
 			// This interface exists only to support SQLite favors.
-			IsClean(context.Context) (bool, error)
+			Clean(context.Context) error
 		}); ok {
-			e, err := c.IsClean(ctx)
-			if err != nil {
-				return nil, fmt.Errorf("sql/migrate: checking database state: %w", err)
-			}
-			if !e {
-				return nil, errNotClean
-			}
-		} else {
-			realm, err = drv.InspectRealm(ctx, nil)
-			if err != nil {
-				return nil, err
-			}
-			if len(realm.Schemas) > 0 {
-				return nil, errNotClean
-			}
-		}
-		// Sort files lexicographically.
-		sort.Slice(names, func(i, j int) bool {
-			return names[i] < names[j]
-		})
-		// If the driver supports it, acquire a lock while replaying the migration changes.
-		if l, ok := drv.(schema.Locker); ok {
-			unlock, err := l.Lock(ctx, "atlas_migration_directory_state", 0)
-			if err != nil {
-				return nil, fmt.Errorf("sql/migrate: acquiring database lock: %w", err)
-			}
-			defer unlock()
-		}
-		// Clean up after ourselves.
-		defer func() {
-			if e, ok := drv.(interface {
-				// The Clean method can be added to a Driver to change how to clean a database.
-				// This interface exists only to support SQLite favors.
-				Clean(context.Context) error
-			}); ok {
-				if derr := e.Clean(ctx); derr != nil {
-					err = wrap(derr, err)
-				}
-				return
-			}
-			realm, derr := drv.InspectRealm(ctx, nil)
-			if derr != nil {
-				err = wrap(derr, err)
-			}
-			del := make([]schema.Change, len(realm.Schemas))
-			for i, s := range realm.Schemas {
-				del[i] = &schema.DropSchema{S: s, Extra: []schema.Clause{&schema.IfExists{}}}
-			}
-			if derr := drv.ApplyChanges(ctx, del); derr != nil {
+			if derr := e.Clean(ctx); derr != nil {
 				err = wrap(derr, err)
 			}
-		}()
-		for _, n := range names {
-			f, err := dir.Open(n)
-			if err != nil {
-				return nil, err
-			}
-			sc := bufio.NewScanner(f)
-			for sc.Scan() {
-				t := sc.Text()
-				if !strings.HasPrefix(strings.TrimSpace(t), "--") {
-					if _, err := drv.ExecContext(ctx, t); err != nil {
-						f.Close()
-						return nil, err
-					}
-				}
-			}
-			f.Close()
-			if err := sc.Err(); err != nil {
-				return nil, err
-			}
+			return
+		}
+		realm, derr := e.drv.InspectRealm(ctx, nil)
+		if derr != nil {
+			err = wrap(derr, err)
 		}
-		realm, err = drv.InspectRealm(ctx, nil)
-		return
+		del := make([]schema.Change, len(realm.Schemas))
+		for i, s := range realm.Schemas {
+			del[i] = &schema.DropSchema{S: s, Extra: []schema.Clause{&schema.IfExists{}}}
+		}
+		if derr := e.drv.ApplyChanges(ctx, del); derr != nil {
+			err = wrap(derr, err)
+		}
+	}()
+	// Replay the migration directory on the database.
+	if err := e.Execute(ctx, 0); err != nil && !errors.Is(err, ErrNoPendingFiles) {
+		return nil, fmt.Errorf("sql/migrate: read migration directory state: %w", err)
 	}
+	// Inspect the database back and return the result.
+	realm, err = e.drv.InspectRealm(ctx, nil)
+	return
+}
+
+// NoopRevisionReadWriter is a RevisionsReadWriter that does nothing.
+// It is useful for one-time replay of the migration directory.
+type NoopRevisionReadWriter struct{}
+
+// ReadRevisions implements RevisionsReadWriter.ReadRevisions,
+func (NoopRevisionReadWriter) ReadRevisions(context.Context) (Revisions, error) {
+	return nil, nil
+}
+
+// WriteRevisions implements RevisionsReadWriter.WriteRevisions,
+func (NoopRevisionReadWriter) WriteRevisions(context.Context, Revisions) error {
+	return nil
 }
 
 // done computes and sets the ExecutionTime.


Test patch:
diff --git a/cmd/atlascmd/migrate_test.go b/cmd/atlascmd/migrate_test.go
index 891872e9..93fcdf79 100644
--- a/cmd/atlascmd/migrate_test.go
+++ b/cmd/atlascmd/migrate_test.go
@@ -45,7 +45,7 @@ func TestMigrate_Diff(t *testing.T) {
 	require.True(t, strings.HasPrefix(s, "Error: sql/migrate: connected database is not clean"))
 	require.EqualError(t, err, "sql/migrate: connected database is not clean")
 
-	// Works.
+	// Works (on empty directory).
 	s, err = runCmd(
 		Root, "migrate", "diff",
 		"name",
@@ -53,8 +53,8 @@ func TestMigrate_Diff(t *testing.T) {
 		"--dev-url", openSQLite(t, ""),
 		"--to", hclURL(t),
 	)
-	require.Zero(t, s)
 	require.NoError(t, err)
+	require.Zero(t, s)
 	require.FileExists(t, filepath.Join(p, fmt.Sprintf("%s_name.sql", time.Now().UTC().Format("20060102150405"))))
 	require.FileExists(t, filepath.Join(p, "atlas.sum"))
 
diff --git a/cmd/atlascmd/project_test.go b/cmd/atlascmd/project_test.go
index 27deb192..5f655009 100644
--- a/cmd/atlascmd/project_test.go
+++ b/cmd/atlascmd/project_test.go
@@ -7,14 +7,21 @@ package atlascmd
 import (
 	"os"
 	"path/filepath"
+	"sort"
 	"testing"
 
+	"ariga.io/atlas/schema/schemaspec"
 	"github.com/stretchr/testify/require"
 )
 
 func TestLoadEnv(t *testing.T) {
 	d := t.TempDir()
 	h := `
+variable "name" {
+	type = string
+	default = "hello"
+}
+
 env "local" {
 	url = "mysql://root:pass@localhost:3306/"
 	dev = "docker://mysql/8"
@@ -25,14 +32,21 @@ env "local" {
 		url = "file://migrations"
 		format = atlas
 	}
+	
+	bool = true
+	integer = 42
+	str = var.name
 }
 `
 	err := os.WriteFile(filepath.Join(d, projectFileName), []byte(h), 0600)
 	require.NoError(t, err)
 	path := filepath.Join(d, projectFileName)
 	t.Run("ok", func(t *testing.T) {
-		env := &Env{}
+		var env *Env
 		env, err = LoadEnv(path, "local")
+		sort.Slice(env.Extra.Attrs, func(i, j int) bool {
+			return env.Extra.Attrs[i].K < env.Extra.Attrs[j].K
+		})
 		require.NoError(t, err)
 		require.EqualValues(t, &Env{
 			Name:    "local",
@@ -44,8 +58,28 @@ env "local" {
 				URL:    "file://migrations",
 				Format: formatAtlas,
 			},
+			DefaultExtension: schemaspec.DefaultExtension{
+				Extra: schemaspec.Resource{
+					Attrs: []*schemaspec.Attr{
+						{K: "bool", V: &schemaspec.LiteralValue{V: "true"}},
+						{K: "integer", V: &schemaspec.LiteralValue{V: "42"}},
+						{K: "str", V: &schemaspec.LiteralValue{V: `"hello"`}},
+					},
+				},
+			},
 		}, env)
 	})
+	t.Run("with input", func(t *testing.T) {
+		env, err := LoadEnv(path, "local", WithInput(map[string]string{
+			"name": "goodbye",
+		}))
+		require.NoError(t, err)
+		str, ok := env.Attr("str")
+		require.True(t, ok)
+		val, err := str.String()
+		require.NoError(t, err)
+		require.EqualValues(t, "goodbye", val)
+	})
 	t.Run("wrong env", func(t *testing.T) {
 		_, err = LoadEnv(path, "home")
 		require.EqualError(t, err, `env "home" not defined in project file`)
diff --git a/internal/integration/testdata/sqlite/cli-apply-vars.txt b/internal/integration/testdata/sqlite/cli-apply-vars.txt
new file mode 100644
index 00000000..2c90afb4
--- /dev/null
+++ b/internal/integration/testdata/sqlite/cli-apply-vars.txt
@@ -0,0 +1,37 @@
+! atlas schema apply --env local --auto-approve
+stderr 'input value "def_val" expected but missing'
+
+atlas schema apply --env local_with_vals --auto-approve
+cmpshow users expected.sql
+
+-- atlas.hcl --
+env "local" {
+    url = "URL"
+    src = "./1.hcl"
+}
+env "local_with_vals" {
+    url = "URL"
+    src = "./1.hcl"
+
+    def_val = "hello"
+}
+-- 1.hcl --
+variable "def_val" {
+    type = string
+}
+table "users" {
+  schema = schema.main
+  column "id" {
+    null = false
+    type = int
+  }
+  column "status" {
+    null = true
+    type = text
+    default = var.def_val
+  }
+}
+schema "main" {
+}
+-- expected.sql --
+CREATE TABLE `users` (`id` int NOT NULL, `status` text NULL DEFAULT 'hello')
\ No newline at end of file
diff --git a/internal/integration/testdata/sqlite/cli-project-vars.txt b/internal/integration/testdata/sqlite/cli-project-vars.txt
new file mode 100644
index 00000000..9159c69b
--- /dev/null
+++ b/internal/integration/testdata/sqlite/cli-project-vars.txt
@@ -0,0 +1,35 @@
+! atlas schema apply --env local --auto-approve
+stderr '"user_status_default" expected but missing'
+
+atlas schema apply --env local --auto-approve --var user_status_default=hello
+cmpshow users expected.sql
+
+-- atlas.hcl --
+variable "user_status_default" {
+    type = string
+}
+env "local" {
+    url = "URL"
+    src = "./1.hcl"
+    def_val = var.user_status_default
+}
+-- 1.hcl --
+variable "def_val" {
+    type = string
+}
+table "users" {
+  schema = schema.main
+  column "id" {
+    null = false
+    type = int
+  }
+  column "status" {
+    null = true
+    type = text
+    default = var.def_val
+  }
+}
+schema "main" {
+}
+-- expected.sql --
+CREATE TABLE `users` (`id` int NOT NULL, `status` text NULL DEFAULT 'hello')
\ No newline at end of file
diff --git a/sql/migrate/migrate_test.go b/sql/migrate/migrate_test.go
index e237b634..e759970f 100644
--- a/sql/migrate/migrate_test.go
+++ b/sql/migrate/migrate_test.go
@@ -61,7 +61,7 @@ func TestPlanner_WritePlan(t *testing.T) {
 
 func TestPlanner_Plan(t *testing.T) {
 	var (
-		drv = &mockDriver{}
+		drv = &lockMockDriver{&mockDriver{}}
 		ctx = context.Background()
 	)
 	d, err := migrate.NewLocalDir(t.TempDir())
@@ -184,41 +184,45 @@ func TestHash_UnmarshalText(t *testing.T) {
 }
 
 // Deprecated: GlobStateReader will be removed once the Executor is functional.
-func TestGlobStateReader(t *testing.T) {
-	var (
-		drv = &lockMockDriver{&mockDriver{}}
-		ctx = context.Background()
-	)
+func TestExecutor_ReadState(t *testing.T) {
+	ctx := context.Background()
 	d, err := migrate.NewLocalDir("testdata")
 	require.NoError(t, err)
 
-	_, err = migrate.GlobStateReader(d, drv, "*.up.sql").ReadState(ctx)
+	// Locking not supported.
+	_, err = migrate.NewExecutor(&mockDriver{}, d, migrate.NoopRevisionReadWriter{})
+	require.ErrorIs(t, err, migrate.ErrLockUnsupported)
+
+	drv := &lockMockDriver{&mockDriver{}}
+	ex, err := migrate.NewExecutor(drv, d, migrate.NoopRevisionReadWriter{})
 	require.NoError(t, err)
-	require.Equal(t, []string{"CREATE TABLE t(c int);"}, drv.executed)
-	require.Equal(t, 1, drv.lockCounter)
-	require.Equal(t, 1, drv.unlockCounter)
 
-	_, err = migrate.GlobStateReader(d, drv, "*.down.sql").ReadState(ctx)
+	_, err = ex.ReadState(ctx)
 	require.NoError(t, err)
-	require.Equal(t, []string{"CREATE TABLE t(c int);", "DROP TABLE IF EXISTS t;"}, drv.executed)
+	require.Equal(t, []string{"DROP TABLE IF EXISTS t;", "CREATE TABLE t(c int);"}, drv.executed)
 	require.Equal(t, 2, drv.lockCounter)
 	require.Equal(t, 2, drv.unlockCounter)
 
-	drv.locked = true
-	_, err = migrate.GlobStateReader(d, drv, "").ReadState(ctx)
+	// Does not work if locked.
+	drv.locks = map[string]struct{}{"atlas_migration_directory_state": {}}
+	_, err = ex.ReadState(ctx)
 	require.EqualError(t, err, "sql/migrate: acquiring database lock: lockErr")
 	require.Equal(t, 2, drv.lockCounter)
 	require.Equal(t, 2, drv.unlockCounter)
-	drv.locked = false
+	drv.locks = make(map[string]struct{})
 
+	// Does not work if database is not clean.
 	drv.realm = schema.Realm{Schemas: []*schema.Schema{{Name: "schema"}}}
-	_, err = migrate.GlobStateReader(d, drv, "*.up.sql").ReadState(ctx)
+	_, err = ex.ReadState(ctx)
 	require.EqualError(t, err, "sql/migrate: connected database is not clean")
-	require.Equal(t, 2, drv.lockCounter)
-	require.Equal(t, 2, drv.unlockCounter)
+	require.Equal(t, 3, drv.lockCounter)
+	require.Equal(t, 3, drv.unlockCounter)
 
+	// Works.
 	edrv := &emptyMockDriver{drv}
-	_, err = migrate.GlobStateReader(d, edrv, "*.up.sql").ReadState(ctx)
+	ex, err = migrate.NewExecutor(edrv, d, migrate.NoopRevisionReadWriter{})
+	require.NoError(t, err)
+	_, err = ex.ReadState(ctx)
 	require.NoError(t, err)
 	require.Equal(t, []schema.Change{
 		&schema.DropSchema{
@@ -226,8 +230,8 @@ func TestGlobStateReader(t *testing.T) {
 			Extra: []schema.Clause{&schema.IfExists{}},
 		},
 	}, edrv.applied)
-	require.Equal(t, 3, drv.lockCounter)
-	require.Equal(t, 3, drv.unlockCounter)
+	require.Equal(t, 5, drv.lockCounter)
+	require.Equal(t, 5, drv.unlockCounter)
 }
 
 func TestLocalDir(t *testing.T) {
@@ -403,7 +407,7 @@ type (
 		applied       []schema.Change
 		realm         schema.Realm
 		executed      []string
-		locked        bool
+		locks         map[string]struct{}
 		lockCounter   int
 		unlockCounter int
 	}
@@ -429,15 +433,18 @@ func (m *mockDriver) ApplyChanges(_ context.Context, changes []schema.Change) er
 	m.applied = changes
 	return nil
 }
-func (m *lockMockDriver) Lock(context.Context, string, time.Duration) (schema.UnlockFunc, error) {
-	if m.locked {
+func (m *lockMockDriver) Lock(_ context.Context, name string, _ time.Duration) (schema.UnlockFunc, error) {
+	if _, ok := m.locks[name]; ok {
 		return nil, errors.New("lockErr")
 	}
 	m.lockCounter++
-	m.locked = true
+	if m.locks == nil {
+		m.locks = make(map[string]struct{})
+	}
+	m.locks[name] = struct{}{}
 	return func() error {
 		m.unlockCounter++
-		m.locked = false
+		delete(m.locks, name)
 		return nil
 	}, nil
 }