Repository: ariga/atlas
Pull request #2408 (2023-12-28 10:25:43)

cmd/atlas/internal/cmdext: make schema reader errors friendlier


Patch:
diff --git a/cmd/atlas/internal/cmdext/cmdext.go b/cmd/atlas/internal/cmdext/cmdext.go
index 75f169b2..a6186793 100644
--- a/cmd/atlas/internal/cmdext/cmdext.go
+++ b/cmd/atlas/internal/cmdext/cmdext.go
@@ -463,7 +463,7 @@ func TemplateDir(ctx *hcl.EvalContext, block *hclsyntax.Block) (cty.Value, error
 	u := fmt.Sprintf("mem://%s", dirname)
 	// Allow using reading the computed dir as a state source.
 	memLoader.states[u] = StateLoaderFunc(func(ctx context.Context, config *StateReaderConfig) (*StateReadCloser, error) {
-		return stateReaderSQL(ctx, config, dir)
+		return stateReaderSQL(ctx, config, dir, nil, nil)
 	})
 	return cty.ObjectVal(map[string]cty.Value{
 		"url": cty.StringVal(u),
@@ -635,7 +635,10 @@ func SchemaExternal(ctx *hcl.EvalContext, block *hclsyntax.Block) (cty.Value, er
 		}
 		return cty.NilVal, errorf("running program %v: %v", cmd.Path, msg)
 	}
-	dir, err := filesAsDir(migrate.NewLocalFile("schema.sql", out))
+	// Directory files must have an .sql extension to be read by the executor.
+	// The "schema" word is added to indicate that unlike data-source errors, load error
+	// comes from the output of the data-source (SQL representation of the state/schema).
+	dir, err := filesAsDir(migrate.NewLocalFile(fmt.Sprintf("%s/schema.sql", block.Labels[1]), out))
 	if err != nil {
 		return cty.NilVal, errorf("converting output to migration: %v", err)
 	}
@@ -644,7 +647,7 @@ func SchemaExternal(ctx *hcl.EvalContext, block *hclsyntax.Block) (cty.Value, er
 		return cty.NilVal, errorf("build url: %v", err)
 	}
 	memLoader.states[u] = StateLoaderFunc(func(ctx context.Context, config *StateReaderConfig) (*StateReadCloser, error) {
-		return stateReaderSQL(ctx, config, dir)
+		return stateSchemaSQL(ctx, config, dir)
 	})
 	return cty.ObjectVal(map[string]cty.Value{
 		"url": cty.StringVal(u),
diff --git a/cmd/atlas/internal/cmdext/reader.go b/cmd/atlas/internal/cmdext/reader.go
index 7c2ef5e8..1b14b01b 100644
--- a/cmd/atlas/internal/cmdext/reader.go
+++ b/cmd/atlas/internal/cmdext/reader.go
@@ -13,6 +13,7 @@ import (
 	"os"
 	"path/filepath"
 
+	"ariga.io/atlas/cmd/atlas/internal/cmdlog"
 	cmdmigrate "ariga.io/atlas/cmd/atlas/internal/migrate"
 	"ariga.io/atlas/sql/migrate"
 	"ariga.io/atlas/sql/schema"
@@ -55,7 +56,6 @@ func StateReaderSQL(ctx context.Context, config *StateReaderConfig) (*StateReadC
 	}
 	var (
 		dir  migrate.Dir
-		opts []migrate.ReplayOption
 		path = filepath.Join(config.URLs[0].Host, config.URLs[0].Path)
 	)
 	switch fi, err := os.Stat(path); {
@@ -70,6 +70,7 @@ func StateReaderSQL(ctx context.Context, config *StateReaderConfig) (*StateReadC
 		if dir, err = filesAsDir(migrate.NewLocalFile(fi.Name(), b)); err != nil {
 			return nil, err
 		}
+		return stateSchemaSQL(ctx, config, dir)
 	// The sum file is optional when reading the directory state.
 	case isSchemaDir(config.URLs[0], path):
 		dirs, err := os.ReadDir(path)
@@ -87,16 +88,18 @@ func StateReaderSQL(ctx context.Context, config *StateReaderConfig) (*StateReadC
 		if dir, err = filesAsDir(files...); err != nil {
 			return nil, err
 		}
+		return stateSchemaSQL(ctx, config, dir)
 	// A migration directory.
 	default:
+		var opts []migrate.ReplayOption
 		if dir, err = cmdmigrate.DirURL(ctx, config.URLs[0], false); err != nil {
 			return nil, err
 		}
 		if v := config.URLs[0].Query().Get("version"); v != "" {
 			opts = append(opts, migrate.ReplayToVersion(v))
 		}
+		return stateReaderSQL(ctx, config, dir, nil, opts)
 	}
-	return stateReaderSQL(ctx, config, dir, opts...)
 }
 
 // isSchemaDir returns true if the given path is a schema directory (not a migration directory).
@@ -108,31 +111,49 @@ func isSchemaDir(u *url.URL, path string) bool {
 	return errors.Is(err, os.ErrNotExist)
 }
 
+// errNoDevURL is returned when trying to read an SQL schema file/directory or replay a migration directory,
+// the dev-url was not set.
+var errNoDevURL = errors.New("--dev-url cannot be empty. See: https://atlasgo.io/atlas-schema/sql#dev-database")
+
+// stateSchemaSQL wraps stateReaderSQL for SQL schema files or directories to control errors when replay/read fails.
+func stateSchemaSQL(ctx context.Context, cfg *StateReaderConfig, dir migrate.Dir) (*StateReadCloser, error) {
+	if cfg.Dev == nil {
+		return nil, errNoDevURL
+	}
+	log := cmdlog.NewMigrateApply(ctx, cfg.Dev, dir)
+	r, err := stateReaderSQL(ctx, cfg, dir, []migrate.ExecutorOption{migrate.WithLogger(log)}, nil)
+	if n := len(log.Applied); err != nil && n > 0 {
+		if serr := log.Applied[n-1].Error; serr != nil && serr.Stmt != "" && serr.Text != "" {
+			err = fmt.Errorf("read state from %q: executing statement: %q: %s", log.Applied[n-1].Name(), serr.Stmt, serr.Text)
+		}
+	}
+	return r, err
+}
+
 // stateReaderSQL returns a migrate.StateReader from an SQL file or a directory of migrations.
-func stateReaderSQL(ctx context.Context, config *StateReaderConfig, dir migrate.Dir, opts ...migrate.ReplayOption) (*StateReadCloser, error) {
-	// Replaying a migration directory requires a dev connection.
-	if config.Dev == nil {
-		return nil, errors.New("--dev-url cannot be empty. See: https://atlasgo.io/atlas-schema/sql#dev-database")
+func stateReaderSQL(ctx context.Context, cfg *StateReaderConfig, dir migrate.Dir, optsExec []migrate.ExecutorOption, optsReplay []migrate.ReplayOption) (*StateReadCloser, error) {
+	if cfg.Dev == nil {
+		return nil, errNoDevURL
 	}
-	ex, err := migrate.NewExecutor(config.Dev.Driver, dir, migrate.NopRevisionReadWriter{})
+	ex, err := migrate.NewExecutor(cfg.Dev.Driver, dir, migrate.NopRevisionReadWriter{}, optsExec...)
 	if err != nil {
 		return nil, err
 	}
 	sr, err := ex.Replay(ctx, func() migrate.StateReader {
-		if config.Dev.URL.Schema != "" {
-			return migrate.SchemaConn(config.Dev, "", nil)
+		if cfg.Dev.URL.Schema != "" {
+			return migrate.SchemaConn(cfg.Dev, "", nil)
 		}
-		return migrate.RealmConn(config.Dev, &schema.InspectRealmOption{
-			Schemas: config.Schemas,
-			Exclude: config.Exclude,
+		return migrate.RealmConn(cfg.Dev, &schema.InspectRealmOption{
+			Schemas: cfg.Schemas,
+			Exclude: cfg.Exclude,
 		})
-	}(), opts...)
+	}(), optsReplay...)
 	if err != nil && !errors.Is(err, migrate.ErrNoPendingFiles) {
 		return nil, err
 	}
 	return &StateReadCloser{
 		StateReader: migrate.Realm(sr),
-		Schema:      config.Dev.URL.Schema,
+		Schema:      cfg.Dev.URL.Schema,
 	}, nil
 }
 
diff --git a/doc/md/versioned/troubleshoot.mdx b/doc/md/versioned/troubleshoot.mdx
index 68b140ee..05fb6f02 100644
--- a/doc/md/versioned/troubleshoot.mdx
+++ b/doc/md/versioned/troubleshoot.mdx
@@ -66,7 +66,7 @@ Error 1062: Duplicate entry '1-2' for key 'user_groups.PRIMARY'
 
 ```text
 $ atlas --env local migrate apply 1 --tx-mode none
-Error: sql/migrate: execute: executing statement "INSERT INTO `user_groups` (`user_id`, `group_id`) VALUES
+Error: sql/migrate: executing statement "INSERT INTO `user_groups` (`user_id`, `group_id`) VALUES
     -- Founders\n    (2, 1), (3, 1),
     -- Seniors\n    (1, 2), (2, 2), (3, 2),
     -- Constraint error (adding masseelch twice to seniors)
diff --git a/doc/website/blog/2022-09-05-announcing-migration-execution.mdx b/doc/website/blog/2022-09-05-announcing-migration-execution.mdx
index 3fe0b154..7052d4b7 100644
--- a/doc/website/blog/2022-09-05-announcing-migration-execution.mdx
+++ b/doc/website/blog/2022-09-05-announcing-migration-execution.mdx
@@ -177,7 +177,7 @@ Migrating to version 3 from 1 (2 migrations in total):
 Error: Execution had errors:
 Error 1046: No database selected
 
-Error: sql/migrate: execute: executing statement "CREATE TABLE `tbl_2` (`col` int NOT NULL);" from version "3": Error 1046: No database selected
+Error: sql/migrate: executing statement "CREATE TABLE `tbl_2` (`col` int NOT NULL);" from version "3": Error 1046: No database selected
 exit status 1
 
 ```
diff --git a/sql/migrate/migrate.go b/sql/migrate/migrate.go
index ee380b5f..8ce9b5f1 100644
--- a/sql/migrate/migrate.go
+++ b/sql/migrate/migrate.go
@@ -558,13 +558,13 @@ func (e MissingMigrationError) Error() string {
 // NewExecutor creates a new Executor with default values.
 func NewExecutor(drv Driver, dir Dir, rrw RevisionReadWriter, opts ...ExecutorOption) (*Executor, error) {
 	if drv == nil {
-		return nil, errors.New("sql/migrate: execute: no driver given")
+		return nil, errors.New("sql/migrate: no driver given")
 	}
 	if dir == nil {
-		return nil, errors.New("sql/migrate: execute: no dir given")
+		return nil, errors.New("sql/migrate: no dir given")
 	}
 	if rrw == nil {
-		return nil, errors.New("sql/migrate: execute: no revision storage given")
+		return nil, errors.New("sql/migrate: no revision storage given")
 	}
 	ex := &Executor{drv: drv, dir: dir, rrw: rrw}
 	for _, opt := range opts {
@@ -582,7 +582,7 @@ func NewExecutor(drv Driver, dir Dir, rrw RevisionReadWriter, opts ...ExecutorOp
 		return nil, ErrCleanCheckerUnsupported
 	}
 	if ex.baselineVer != "" && ex.allowDirty {
-		return nil, errors.New("sql/migrate: execute: baseline and allow-dirty are mutually exclusive")
+		return nil, errors.New("sql/migrate: baseline and allow-dirty are mutually exclusive")
 	}
 	return ex, nil
 }
@@ -653,16 +653,16 @@ func WithOperatorVersion(v string) ExecutorOption {
 func (e *Executor) Pending(ctx context.Context) ([]File, error) {
 	// Don't operate with a broken migration directory.
 	if err := Validate(e.dir); err != nil {
-		return nil, fmt.Errorf("sql/migrate: execute: validate migration directory: %w", err)
+		return nil, fmt.Errorf("sql/migrate: validate migration directory: %w", err)
 	}
 	// Read all applied database revisions.
 	revs, err := e.rrw.ReadRevisions(ctx)
 	if err != nil {
-		return nil, fmt.Errorf("sql/migrate: execute: read revisions: %w", err)
+		return nil, fmt.Errorf("sql/migrate: read revisions: %w", err)
 	}
 	all, err := e.dir.Files()
 	if err != nil {
-		return nil, fmt.Errorf("sql/migrate: execute: select migration files: %w", err)
+		return nil, fmt.Errorf("sql/migrate: read migration directory files: %w", err)
 	}
 	migrations := SkipCheckpointFiles(all)
 	var pending []File
@@ -774,15 +774,15 @@ func (e *Executor) Pending(ctx context.Context) ([]File, error) {
 func (e *Executor) Execute(ctx context.Context, m File) (err error) {
 	hf, err := e.dir.Checksum()
 	if err != nil {
-		return fmt.Errorf("sql/migrate: execute: compute hash: %w", err)
+		return fmt.Errorf("sql/migrate: compute hash: %w", err)
 	}
 	hash, err := hf.SumByName(m.Name())
 	if err != nil {
-		return fmt.Errorf("sql/migrate: execute: scanning checksum from %q: %w", m.Name(), err)
+		return fmt.Errorf("sql/migrate: scanning checksum from %q: %w", m.Name(), err)
 	}
 	stmts, err := FileStmts(e.drv, m)
 	if err != nil {
-		return fmt.Errorf("sql/migrate: execute: scanning statements from %q: %w", m.Name(), err)
+		return fmt.Errorf("sql/migrate: scanning statements from %q: %w", m.Name(), err)
 	}
 	// Create checksums for the statements.
 	var (
@@ -800,7 +800,7 @@ func (e *Executor) Execute(ctx context.Context, m File) (err error) {
 	// and it is partially applied, continue where the last attempt was left off.
 	r, err := e.rrw.ReadRevision(ctx, version)
 	if err != nil && !errors.Is(err, ErrRevisionNotExist) {
-		return fmt.Errorf("sql/migrate: execute: read revision: %w", err)
+		return fmt.Errorf("sql/migrate: read revision: %w", err)
 	}
 	if errors.Is(err, ErrRevisionNotExist) {
 		// Haven't seen this file before, create a new revision.
@@ -841,7 +841,7 @@ func (e *Executor) Execute(ctx context.Context, m File) (err error) {
 			r.done()
 			r.ErrorStmt = stmt
 			r.Error = err.Error()
-			return fmt.Errorf("sql/migrate: execute: executing statement %q from version %q: %w", stmt, r.Version, err)
+			return fmt.Errorf("sql/migrate: executing statement %q from version %q: %w", stmt, r.Version, err)
 		}
 		r.PartialHashes = append(r.PartialHashes, "h1:"+sums[r.Applied])
 		r.Applied++
@@ -857,7 +857,7 @@ func (e *Executor) writeRevision(ctx context.Context, r *Revision) error {
 	r.ExecutedAt = time.Now()
 	r.OperatorVersion = e.operator
 	if err := e.rrw.WriteRevision(ctx, r); err != nil {
-		return fmt.Errorf("sql/migrate: execute: write revision: %w", err)
+		return fmt.Errorf("sql/migrate: write revision: %w", err)
 	}
 	return nil
 }
@@ -869,7 +869,7 @@ type HistoryChangedError struct {
 }
 
 func (e HistoryChangedError) Error() string {
-	return fmt.Sprintf("sql/migrate: execute: history changed: statement %d from file %q changed", e.Stmt, e.File)
+	return fmt.Sprintf("sql/migrate: history changed: statement %d from file %q changed", e.Stmt, e.File)
 }
 
 // HistoryNonLinearError is returned if the migration history is not linear. Means, a file was added out of order.
@@ -919,7 +919,7 @@ func (e *Executor) ExecuteTo(ctx context.Context, version string) (err error) {
 		return file.Version() == version
 	}); idx {
 	case -1:
-		return fmt.Errorf("sql/migrate: execute: migration with version %q not found", version)
+		return fmt.Errorf("sql/migrate: migration with version %q not found", version)
 	default:
 		pending = pending[:idx+1]
 	}
@@ -929,7 +929,7 @@ func (e *Executor) ExecuteTo(ctx context.Context, version string) (err error) {
 func (e *Executor) exec(ctx context.Context, files []File) error {
 	revs, err := e.rrw.ReadRevisions(ctx)
 	if err != nil {
-		return fmt.Errorf("sql/migrate: execute: read revisions: %w", err)
+		return fmt.Errorf("sql/migrate: read revisions: %w", err)
 	}
 	LogIntro(e.log, revs, files)
 	for _, m := range files {


Test patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate_test.go b/cmd/atlas/internal/cmdapi/migrate_test.go
index 7b11509c..3eb36efe 100644
--- a/cmd/atlas/internal/cmdapi/migrate_test.go
+++ b/cmd/atlas/internal/cmdapi/migrate_test.go
@@ -684,7 +684,7 @@ func TestMigrate_ApplyTxModeDirective(t *testing.T) {
 			"--url", u,
 			"--tx-mode", mode,
 		)
-		require.EqualError(t, err, `sql/migrate: execute: executing statement "INSERT INTO t1 VALUES (1), (1);" from version "20220925094021": UNIQUE constraint failed: t1.a`)
+		require.EqualError(t, err, `sql/migrate: executing statement "INSERT INTO t1 VALUES (1), (1);" from version "20220925094021": UNIQUE constraint failed: t1.a`)
 		db, err := sql.Open("sqlite3", strings.TrimPrefix(u, "sqlite://"))
 		require.NoError(t, err)
 		var n int
@@ -1270,7 +1270,7 @@ env {
 			"--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.EqualError(t, err, `sql/migrate: 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)
@@ -1281,7 +1281,7 @@ env {
 		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)
+		require.Equal(t, `Error: sql/migrate: 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)
@@ -1293,7 +1293,7 @@ env {
 		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)
+		require.Equal(t, `Error: sql/migrate: executing statement "create table bar (id int)" from version "2": table bar already exists`, report.Log[1].Log[2].Text)
 	})
 }
 
diff --git a/cmd/atlas/internal/cmdext/cmdext_test.go b/cmd/atlas/internal/cmdext/cmdext_test.go
index a8e0689d..3b262a8c 100644
--- a/cmd/atlas/internal/cmdext/cmdext_test.go
+++ b/cmd/atlas/internal/cmdext/cmdext_test.go
@@ -372,6 +372,27 @@ schema = data.external_schema.a8m.url
 schema "main" {
 }
 `, string(buf))
+
+	// Read state error.
+	err = state.EvalBytes([]byte(`
+data "external_schema" "a8m" {
+  program = [
+    "echo",
+    "CREATE TABLE t(c int);",
+    "CREATE UNKNOWN",
+  ]
+}
+
+schema = data.external_schema.a8m.url
+`), &v, nil)
+	require.NoError(t, err)
+	loader, ok = cmdext.States.Loader(u.Scheme)
+	require.True(t, ok)
+	sr, err = loader.LoadState(ctx, &cmdext.StateReaderConfig{
+		Dev:  drv,
+		URLs: []*url.URL{u},
+	})
+	require.EqualError(t, err, `read state from "a8m/schema.sql": executing statement: "CREATE UNKNOWN": near "UNKNOWN": syntax error`)
 }
 
 func TestExternal(t *testing.T) {
diff --git a/sql/migrate/migrate_test.go b/sql/migrate/migrate_test.go
index 8f44e98b..f514c49c 100644
--- a/sql/migrate/migrate_test.go
+++ b/sql/migrate/migrate_test.go
@@ -502,17 +502,17 @@ func TestExecutor_ExecOrderLinear(t *testing.T) {
 func TestExecutor(t *testing.T) {
 	// Passing nil raises error.
 	ex, err := migrate.NewExecutor(nil, nil, nil)
-	require.EqualError(t, err, "sql/migrate: execute: no driver given")
+	require.EqualError(t, err, "sql/migrate: no driver given")
 	require.Nil(t, ex)
 
 	ex, err = migrate.NewExecutor(&mockDriver{}, nil, nil)
-	require.EqualError(t, err, "sql/migrate: execute: no dir given")
+	require.EqualError(t, err, "sql/migrate: no dir given")
 	require.Nil(t, ex)
 
 	dir, err := migrate.NewLocalDir(t.TempDir())
 	require.NoError(t, err)
 	ex, err = migrate.NewExecutor(&mockDriver{}, dir, nil)
-	require.EqualError(t, err, "sql/migrate: execute: no revision storage given")
+	require.EqualError(t, err, "sql/migrate: no revision storage given")
 	require.Nil(t, ex)
 
 	// Does not operate on invalid migration dir.
@@ -648,7 +648,7 @@ func TestExecutor(t *testing.T) {
 	// Test ExecuteTo.
 	*rrw = []*migrate.Revision{}
 	*drv = mockDriver{}
-	require.EqualError(t, ex.ExecuteTo(context.Background(), ""), "sql/migrate: execute: migration with version \"\" not found")
+	require.EqualError(t, ex.ExecuteTo(context.Background(), ""), "sql/migrate: migration with version \"\" not found")
 	require.NoError(t, ex.ExecuteTo(context.Background(), "2.10.x-20"))
 	requireEqualRevisions(t, []*migrate.Revision{rev1, rev2}, *rrw)
 }