Repository: ariga/atlas
Pull request #1042 (2022-08-10 14:56:21)

sql/migrate: remove locking from executor
Keep it simple

- Reduce the number of locks from 3 to 1; each lock opens and acquires a connection if it's not running inside a transaction.
- Both diff and apply should acquire one lock and it should be done by the caller and not be done as a side effect. e.g. setting `MaxConnection(X)` can cause the CLI to block.
- Also, moving it out from the executor allows us to use external locks similar to terraform (e.g. DynamoDB or Consul)


Patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate.go b/cmd/atlas/internal/cmdapi/migrate.go
index c742691b..4403d37c 100644
--- a/cmd/atlas/internal/cmdapi/migrate.go
+++ b/cmd/atlas/internal/cmdapi/migrate.go
@@ -45,10 +45,10 @@ const (
 	migrateFlagDryRun           = "dry-run"
 	migrateFlagTo               = "to"
 	migrateFlagSchema           = "schema"
-	migrateDiffFlagVerbose      = "verbose"
 	migrateLintLatest           = "latest"
 	migrateLintGitDir           = "git-dir"
 	migrateLintGitBase          = "git-base"
+	migrateDiffQualifier        = "qualifier"
 	migrateApplyAllowDirty      = "allow-dirty"
 	migrateApplyFromVersion     = "from"
 	migrateApplyBaselineVersion = "baseline"
@@ -65,7 +65,6 @@ var (
 		DirFormat      string
 		RevisionSchema string
 		Force          bool
-		Verbose        bool
 		Apply          struct {
 			DryRun          bool
 			LogFormat       string
@@ -73,6 +72,9 @@ var (
 			FromVersion     string
 			BaselineVersion string
 		}
+		Diff struct {
+			Qualifier string // optional table qualifier
+		}
 		Lint struct {
 			Format  string // log formatting
 			Latest  uint   // latest N migration files
@@ -247,7 +249,7 @@ func init() {
 	// Diff flags.
 	urlFlag(&MigrateFlags.DevURL, migrateFlagDevURL, "", MigrateDiffCmd.Flags())
 	MigrateDiffCmd.Flags().StringSliceVarP(&MigrateFlags.ToURLs, migrateFlagTo, "", nil, "[driver://username:password@address/dbname?param=value ...] select a desired state using the URL format")
-	MigrateDiffCmd.Flags().BoolVarP(&MigrateFlags.Verbose, migrateDiffFlagVerbose, "", false, "enable verbose logging")
+	MigrateDiffCmd.Flags().StringVarP(&MigrateFlags.Diff.Qualifier, migrateDiffQualifier, "", "", "qualify tables with custom qualifier when working on a single schema")
 	MigrateDiffCmd.Flags().SortFlags = false
 	cobra.CheckErr(MigrateDiffCmd.MarkFlagRequired(migrateFlagDevURL))
 	cobra.CheckErr(MigrateDiffCmd.MarkFlagRequired(migrateFlagTo))
@@ -289,6 +291,15 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 		return err
 	}
 	defer c.Close()
+	// Acquire a lock.
+	if l, ok := c.Driver.(schema.Locker); ok {
+		unlock, err := l.Lock(cmd.Context(), "atlas_migrate_execute", 0)
+		if err != nil {
+			return fmt.Errorf("acquiring database lock: %w", err)
+		}
+		// If unlocking fails notify the user about it.
+		defer cobra.CheckErr(unlock())
+	}
 	// Get the correct log format and destination. Currently, only os.Stdout is supported.
 	l, err := logFormat(cmd.OutOrStdout())
 	if err != nil {
@@ -315,11 +326,6 @@ func CmdMigrateApplyRun(cmd *cobra.Command, args []string) error {
 	if err != nil {
 		return err
 	}
-	unlock, err := ex.Lock(cmd.Context())
-	if err != nil {
-		return err
-	}
-	defer unlock()
 	pending, err := ex.Pending(cmd.Context())
 	if err != nil && !errors.Is(err, migrate.ErrNoPendingFiles) {
 		return err
@@ -407,7 +413,7 @@ func CmdMigrateDiffRun(cmd *cobra.Command, args []string) error {
 	if l, ok := dev.Driver.(schema.Locker); ok {
 		unlock, err := l.Lock(cmd.Context(), "atlas_migrate_diff", 0)
 		if err != nil {
-			return err
+			return fmt.Errorf("acquiring database lock: %w", err)
 		}
 		// If unlocking fails notify the user about it.
 		defer cobra.CheckErr(unlock())
@@ -430,7 +436,7 @@ func CmdMigrateDiffRun(cmd *cobra.Command, args []string) error {
 	opts := []migrate.PlannerOption{migrate.PlanFormat(f)}
 	if dev.URL.Schema != "" {
 		// Disable tables qualifier in schema-mode.
-		opts = append(opts, migrate.PlanWithSchemaQualifier(""))
+		opts = append(opts, migrate.PlanWithSchemaQualifier(MigrateFlags.Diff.Qualifier))
 	}
 	// Plan the changes and create a new migration file.
 	pl := migrate.NewPlanner(dev.Driver, dir, opts...)
diff --git a/cmd/atlas/internal/cmdapi/schema.go b/cmd/atlas/internal/cmdapi/schema.go
index 49da6d84..517a9a05 100644
--- a/cmd/atlas/internal/cmdapi/schema.go
+++ b/cmd/atlas/internal/cmdapi/schema.go
@@ -55,7 +55,6 @@ var (
 		Paths       []string
 		DryRun      bool
 		AutoApprove bool
-		Verbose     bool
 	}
 	// SchemaApply represents the 'atlas schema apply' subcommand command.
 	SchemaApply = &cobra.Command{
@@ -143,7 +142,6 @@ func init() {
 	SchemaApply.Flags().StringVarP(&ApplyFlags.DevURL, devURLFlag, "", "", "URL for the dev database. Used to validate schemas and calculate diffs\nbefore running migration.")
 	SchemaApply.Flags().BoolVarP(&ApplyFlags.DryRun, "dry-run", "", false, "Dry-run. Print SQL plan without prompting for execution.")
 	SchemaApply.Flags().BoolVarP(&ApplyFlags.AutoApprove, "auto-approve", "", false, "Auto approve. Apply the schema changes without prompting for approval.")
-	SchemaApply.Flags().BoolVarP(&ApplyFlags.Verbose, migrateDiffFlagVerbose, "", false, "enable verbose logging")
 	SchemaApply.Flags().StringVarP(&SchemaFlags.DSN, dsnFlag, "d", "", "")
 	cobra.CheckErr(SchemaApply.Flags().MarkHidden(dsnFlag))
 	cobra.CheckErr(SchemaApply.MarkFlagRequired(urlFlag))
diff --git a/doc/md/reference.md b/doc/md/reference.md
index c0f6aff7..7d38bb32 100644
--- a/doc/md/reference.md
+++ b/doc/md/reference.md
@@ -142,9 +142,9 @@ directory state to the desired schema. The desired state can be another connecte
 ```
 #### Flags
 ```
-      --dev-url string   [driver://username:password@address/dbname?param=value] select a database using the URL format
-      --to strings       [driver://username:password@address/dbname?param=value ...] select a desired state using the URL format
-      --verbose          enable verbose logging
+      --dev-url string     [driver://username:password@address/dbname?param=value] select a database using the URL format
+      --to strings         [driver://username:password@address/dbname?param=value ...] select a desired state using the URL format
+      --qualifier string   qualify tables with custom qualifier when working on a single schema
 
 ```
 
@@ -333,7 +333,6 @@ migration.
                          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.
-      --verbose          enable verbose logging
 
 ```
 
diff --git a/sql/migrate/migrate.go b/sql/migrate/migrate.go
index 48e03f8a..241e7318 100644
--- a/sql/migrate/migrate.go
+++ b/sql/migrate/migrate.go
@@ -387,8 +387,6 @@ func (p *Planner) WritePlan(plan *Plan) error {
 var (
 	// ErrNoPendingFiles is returned if there are no pending migration files to execute on the managed database.
 	ErrNoPendingFiles = errors.New("sql/migrate: execute: nothing to do")
-	// ErrLockUnsupported is returned if there is no schema.Locker given.
-	ErrLockUnsupported = errors.New("sql/migrate: driver does not support locking")
 	// ErrSnapshotUnsupported is returned if there is no Snapshoter given.
 	ErrSnapshotUnsupported = errors.New("sql/migrate: driver does not support taking a database snapshot")
 	// ErrCleanCheckerUnsupported is returned if there is no CleanChecker given.
@@ -417,9 +415,6 @@ func NewExecutor(drv Driver, dir Dir, rrw RevisionReadWriter, opts ...ExecutorOp
 	if ex.log == nil {
 		ex.log = NopLogger{}
 	}
-	if _, ok := drv.(schema.Locker); !ok {
-		return nil, ErrLockUnsupported
-	}
 	if _, ok := drv.(Snapshoter); !ok {
 		return nil, ErrSnapshotUnsupported
 	}
@@ -467,16 +462,6 @@ func WithAllowDirty(b bool) ExecutorOption {
 	}
 }
 
-// Lock acquires a lock for the executor.
-// It is considered a user error to not call Lock before the Pending and Execute methods.
-func (e *Executor) Lock(ctx context.Context) (schema.UnlockFunc, error) {
-	unlock, err := e.drv.(schema.Locker).Lock(ctx, "atlas_migration_execute", 0)
-	if err != nil {
-		return nil, fmt.Errorf("sql/migrate: acquiring database lock: %w", err)
-	}
-	return unlock, nil
-}
-
 // Pending returns all pending (not fully applied) migration files in the migration directory.
 func (e *Executor) Pending(ctx context.Context) ([]File, error) {
 	// Don't operate with a broken migration directory.
@@ -662,15 +647,6 @@ func (e HistoryChangedError) Error() string {
 // ExecuteN executes n pending migration files. If n<=0 all pending migration files are executed. It will not attempt
 // an execution if the database is not "clean" (has only successfully applied migrations).
 func (e *Executor) ExecuteN(ctx context.Context, n int) (err error) {
-	unlock, err := e.Lock(ctx)
-	if err != nil {
-		return err
-	}
-	defer func() {
-		if err2 := unlock(); err2 != nil {
-			err = wrap(err2, err)
-		}
-	}()
 	pending, err := e.Pending(ctx)
 	if err != nil {
 		return err
@@ -699,15 +675,6 @@ func (e *Executor) ExecuteN(ctx context.Context, n int) (err error) {
 
 // Replay the migration directory and invoke the state to get back the inspection result.
 func (e *Executor) Replay(ctx context.Context, r StateReader) (_ *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 func() {
-		if err2 := unlock(); err2 != nil {
-			err = wrap(err2, err)
-		}
-	}()
 	// Clean up after ourselves.
 	restore, err := e.drv.(Snapshoter).Snapshot(ctx)
 	if err != nil {


Test patch:
diff --git a/cmd/atlas/internal/cmdapi/migrate_test.go b/cmd/atlas/internal/cmdapi/migrate_test.go
index ff5198ed..34a9af8d 100644
--- a/cmd/atlas/internal/cmdapi/migrate_test.go
+++ b/cmd/atlas/internal/cmdapi/migrate_test.go
@@ -101,7 +101,7 @@ func TestMigrate_Apply(t *testing.T) {
 		"--url", fmt.Sprintf("sqlitelockapply://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
 	)
 	require.ErrorIs(t, err, errLock)
-	require.True(t, strings.HasPrefix(s, "Error: sql/migrate: acquiring database lock: "+errLock.Error()))
+	require.True(t, strings.HasPrefix(s, "Error: acquiring database lock: "+errLock.Error()))
 
 	// Will work and print stuff to the console.
 	s, err = runCmd(
@@ -328,7 +328,7 @@ func TestMigrate_Diff(t *testing.T) {
 		"--dev-url", fmt.Sprintf("sqlitelockdiff://file:%s?cache=shared&_fk=1", filepath.Join(p, "test.db")),
 		"--to", to,
 	)
-	require.True(t, strings.HasPrefix(s, "Error: "+errLock.Error()))
+	require.True(t, strings.HasPrefix(s, "Error: acquiring database lock: "+errLock.Error()))
 	require.ErrorIs(t, err, errLock)
 }
 
diff --git a/internal/integration/postgres_test.go b/internal/integration/postgres_test.go
index f4b858ff..6c3ed7a1 100644
--- a/internal/integration/postgres_test.go
+++ b/internal/integration/postgres_test.go
@@ -709,9 +709,9 @@ table "posts" {
 }
 `
 		err = os.WriteFile(filepath.Join(dir, "schema.hcl"), []byte(hcl), 0600)
-		diff := func() string {
+		diff := func(name string) string {
 			out, err := exec.Command(
-				bin, "migrate", "diff",
+				bin, "migrate", "diff", name,
 				"--dir", fmt.Sprintf("file://%s", filepath.Join(dir, "migrations")),
 				"--to", fmt.Sprintf("file://%s", filepath.Join(dir, "schema.hcl")),
 				"--dev-url", fmt.Sprintf("postgres://postgres:pass@localhost:%d/migrate_diff?sslmode=disable", t.port),
@@ -719,7 +719,7 @@ table "posts" {
 			require.NoError(t, err, string(out))
 			return strings.TrimSpace(string(out))
 		}
-		require.Empty(t, diff())
+		require.Empty(t, diff("initial"))
 
 		// Expect one file and read its contents.
 		files, err := os.ReadDir(filepath.Join(dir, "migrations"))
@@ -736,7 +736,26 @@ CREATE TABLE "public"."users" ("id" integer NOT NULL);
 -- create "posts" table
 CREATE TABLE "other"."posts" ("id" integer NOT NULL);
 `, string(b))
-		require.Equal(t, "The migration directory is synced with the desired state, no changes to be made", diff())
+		require.Equal(t, "The migration directory is synced with the desired state, no changes to be made", diff("no_change"))
+
+		// Append a change to the schema and expect a migration to be created.
+		hcl += `
+table "other" "users" {
+	schema = schema.other
+	column "id" { type = integer }
+}`
+		err = os.WriteFile(filepath.Join(dir, "schema.hcl"), []byte(hcl), 0600)
+		require.Empty(t, diff("second"))
+		require.Equal(t, "The migration directory is synced with the desired state, no changes to be made", diff("no_change"))
+		files, err = os.ReadDir(filepath.Join(dir, "migrations"))
+		require.NoError(t, err)
+		require.Equal(t, 3, len(files), dir)
+		b, err = os.ReadFile(filepath.Join(dir, "migrations", files[1].Name()))
+		require.NoError(t, err)
+		require.Equal(t,
+			`-- create "users" table
+CREATE TABLE "other"."users" ("id" integer NOT NULL);
+`, string(b))
 	})
 }
 
diff --git a/internal/integration/testdata/mysql/cli-migrate-diff.txt b/internal/integration/testdata/mysql/cli-migrate-diff.txt
index d59a9b77..b63ffa2c 100644
--- a/internal/integration/testdata/mysql/cli-migrate-diff.txt
+++ b/internal/integration/testdata/mysql/cli-migrate-diff.txt
@@ -14,6 +14,11 @@ cmpmig 0 1.sql
 atlas migrate diff --dev-url URL --to file://./2.hcl second
 cmpmig 1 2.sql
 
+# Clean migration directory and run diff with a custom qualifier.
+exec rm -rf migrations
+atlas migrate diff --dev-url URL --to file://./1.hcl --qualifier test first
+cmpmig 0 1-qualifier.sql
+
 -- 1.hcl --
 schema "script_cli_migrate_diff" {}
 
@@ -35,6 +40,10 @@ table "users" {
 -- create "users" table
 CREATE TABLE `users` (`id` bigint NOT NULL AUTO_INCREMENT, PRIMARY KEY (`id`)) CHARSET utf8mb4 COLLATE utf8mb4_general_ci;
 
+-- 1-qualifier.sql --
+-- create "users" table
+CREATE TABLE `test`.`users` (`id` bigint NOT NULL AUTO_INCREMENT, PRIMARY KEY (`id`)) CHARSET utf8mb4 COLLATE utf8mb4_general_ci;
+
 -- 2.hcl --
 schema "script_cli_migrate_diff" {}
 
diff --git a/sql/migrate/migrate_test.go b/sql/migrate/migrate_test.go
index c15837d7..06495d74 100644
--- a/sql/migrate/migrate_test.go
+++ b/sql/migrate/migrate_test.go
@@ -59,7 +59,7 @@ func TestPlanner_WritePlan(t *testing.T) {
 
 func TestPlanner_Plan(t *testing.T) {
 	var (
-		drv = &lockMockDriver{&mockDriver{}}
+		drv = &mockDriver{}
 		ctx = context.Background()
 	)
 	d, err := migrate.NewLocalDir(t.TempDir())
@@ -89,7 +89,7 @@ func TestPlanner_Plan(t *testing.T) {
 
 func TestPlanner_PlanSchema(t *testing.T) {
 	var (
-		drv = &lockMockDriver{&mockDriver{}}
+		drv = &mockDriver{}
 		ctx = context.Background()
 	)
 	d, err := migrate.NewLocalDir(t.TempDir())
@@ -125,43 +125,24 @@ func TestExecutor_Replay(t *testing.T) {
 	d, err := migrate.NewLocalDir("testdata/migrate")
 	require.NoError(t, err)
 
-	// Locking not supported.
-	_, err = migrate.NewExecutor(&mockDriver{}, d, migrate.NopRevisionReadWriter{})
-	require.ErrorIs(t, err, migrate.ErrLockUnsupported)
-
-	drv := &lockMockDriver{&mockDriver{}}
+	drv := &mockDriver{}
 	ex, err := migrate.NewExecutor(drv, d, migrate.NopRevisionReadWriter{})
 	require.NoError(t, err)
 
 	_, 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)
-	require.Equal(t, 2, drv.lockCounter)
-	require.Equal(t, 2, drv.unlockCounter)
-	require.True(t, drv.released())
-
-	// Does not work if locked.
-	drv.locks = map[string]struct{}{"atlas_migration_directory_state": {}}
-	_, err = ex.Replay(ctx, migrate.RealmConn(drv, nil))
-	require.EqualError(t, err, "sql/migrate: acquiring database lock: lockErr")
-	require.Equal(t, 2, drv.lockCounter)
-	require.Equal(t, 2, drv.unlockCounter)
-	require.False(t, drv.released())
-	drv.locks = make(map[string]struct{})
 
 	// Does not work if database is not clean.
 	drv.dirty = true
 	drv.realm = schema.Realm{Schemas: []*schema.Schema{{Name: "schema"}}}
 	_, err = ex.Replay(ctx, migrate.RealmConn(drv, nil))
 	require.ErrorAs(t, err, &migrate.NotCleanError{})
-	require.Equal(t, 3, drv.lockCounter)
-	require.Equal(t, 3, drv.unlockCounter)
-	require.True(t, drv.released())
 }
 
 func TestExecutor_Pending(t *testing.T) {
 	var (
-		drv  = &lockMockDriver{&mockDriver{}}
+		drv  = &mockDriver{}
 		rrw  = &mockRevisionReadWriter{}
 		log  = &mockLogger{}
 		rev1 = &migrate.Revision{
@@ -232,23 +213,18 @@ func TestExecutor(t *testing.T) {
 	require.EqualError(t, err, "sql/migrate: execute: no revision storage given")
 	require.Nil(t, ex)
 
-	// Does not work if no locking mechanism is provided.
-	ex, err = migrate.NewExecutor(&mockDriver{}, dir, &mockRevisionReadWriter{})
-	require.ErrorIs(t, err, migrate.ErrLockUnsupported)
-	require.Nil(t, ex)
-
 	// Does not operate on invalid migration dir.
 	dir, err = migrate.NewLocalDir(t.TempDir())
 	require.NoError(t, err)
 	require.NoError(t, dir.WriteFile("atlas.sum", hash))
-	ex, err = migrate.NewExecutor(&lockMockDriver{&mockDriver{}}, dir, &mockRevisionReadWriter{})
+	ex, err = migrate.NewExecutor(&mockDriver{}, dir, &mockRevisionReadWriter{})
 	require.NoError(t, err)
 	require.NotNil(t, ex)
 	require.ErrorIs(t, ex.ExecuteN(context.Background(), 0), migrate.ErrChecksumMismatch)
 
 	// Prerequisites.
 	var (
-		drv  = &lockMockDriver{&mockDriver{}}
+		drv  = &mockDriver{}
 		rrw  = &mockRevisionReadWriter{}
 		log  = &mockLogger{}
 		rev1 = &migrate.Revision{
@@ -288,9 +264,6 @@ func TestExecutor(t *testing.T) {
 		migrate.LogStmt{SQL: "ALTER TABLE t_sub ADD c2 int;"},
 		migrate.LogDone{},
 	}, []migrate.LogEntry(*log))
-	require.Equal(t, drv.lockCounter, 1)
-	require.Equal(t, drv.unlockCounter, 1)
-	require.True(t, drv.released())
 
 	// Partly is pending.
 	p, err := ex.Pending(context.Background())
@@ -300,20 +273,17 @@ func TestExecutor(t *testing.T) {
 
 	// Apply one by one.
 	*rrw = mockRevisionReadWriter{}
-	*drv = lockMockDriver{&mockDriver{}}
+	*drv = mockDriver{}
 
 	require.NoError(t, ex.ExecuteN(context.Background(), 1))
-	require.Equal(t, drv.executed, []string{"CREATE TABLE t_sub(c int);", "ALTER TABLE t_sub ADD c1 int;"})
+	require.Equal(t, []string{"CREATE TABLE t_sub(c int);", "ALTER TABLE t_sub ADD c1 int;"}, drv.executed)
 	requireEqualRevisions(t, migrate.Revisions{rev1}, migrate.Revisions(*rrw))
 
 	require.NoError(t, ex.ExecuteN(context.Background(), 1))
-	require.Equal(t, drv.executed, []string{
+	require.Equal(t, []string{
 		"CREATE TABLE t_sub(c int);", "ALTER TABLE t_sub ADD c1 int;", "ALTER TABLE t_sub ADD c2 int;",
-	})
+	}, drv.executed)
 	requireEqualRevisions(t, migrate.Revisions{rev1, rev2}, migrate.Revisions(*rrw))
-	require.Equal(t, 2, drv.lockCounter)
-	require.Equal(t, 2, drv.unlockCounter)
-	require.True(t, drv.released())
 
 	// Partly is pending.
 	p, err = ex.Pending(context.Background())
@@ -323,7 +293,7 @@ func TestExecutor(t *testing.T) {
 
 	// Suppose first revision is already executed, only execute second migration file.
 	*rrw = mockRevisionReadWriter(migrate.Revisions{rev1})
-	*drv = lockMockDriver{&mockDriver{}}
+	*drv = mockDriver{}
 
 	require.NoError(t, ex.ExecuteN(context.Background(), 1))
 	require.Equal(t, []string{"ALTER TABLE t_sub ADD c2 int;"}, drv.executed)
@@ -335,13 +305,9 @@ func TestExecutor(t *testing.T) {
 	require.Len(t, p, 1)
 	require.Equal(t, "3_partly.sql", p[0].Name())
 
-	require.Equal(t, 1, drv.lockCounter)
-	require.Equal(t, 1, drv.unlockCounter)
-	require.True(t, drv.released())
-
 	// Failing, counter will be correct.
 	*rrw = mockRevisionReadWriter(migrate.Revisions{rev1, rev2})
-	*drv = lockMockDriver{&mockDriver{}}
+	*drv = mockDriver{}
 	drv.failOn(2, errors.New("this is an error"))
 	require.ErrorContains(t, ex.ExecuteN(context.Background(), 1), "this is an error")
 	revs, err := rrw.ReadRevisions(context.Background())
@@ -362,7 +328,7 @@ func TestExecutor(t *testing.T) {
 
 	// Re-attempting to migrate will pick up where the execution was left off.
 	revs[len(revs)-1].PartialHashes[0] = h
-	*drv = lockMockDriver{&mockDriver{}}
+	*drv = mockDriver{}
 	require.NoError(t, ex.ExecuteN(context.Background(), 1))
 	require.Equal(t, []string{"ALTER TABLE t_sub ADD c4 int;"}, drv.executed)
 
@@ -373,7 +339,7 @@ func TestExecutor(t *testing.T) {
 func TestExecutor_Baseline(t *testing.T) {
 	var (
 		rrw mockRevisionReadWriter
-		drv = &lockMockDriver{&mockDriver{dirty: true}}
+		drv = &mockDriver{dirty: true}
 		log = &mockLogger{}
 	)
 	dir, err := migrate.NewLocalDir(filepath.Join("testdata/migrate", "sub"))
@@ -417,7 +383,7 @@ func TestExecutor_Baseline(t *testing.T) {
 
 func TestExecutor_FromVersion(t *testing.T) {
 	var (
-		drv = &lockMockDriver{&mockDriver{}}
+		drv = &mockDriver{}
 		log = &mockLogger{}
 		rrw = &mockRevisionReadWriter{
 			{
@@ -455,19 +421,15 @@ func TestExecutor_FromVersion(t *testing.T) {
 type (
 	mockDriver struct {
 		migrate.Driver
-		plan          *migrate.Plan
-		changes       []schema.Change
-		applied       []schema.Change
-		realm         schema.Realm
-		executed      []string
-		locks         map[string]struct{}
-		lockCounter   int
-		unlockCounter int
-		failCounter   int
-		failWith      error
-		dirty         bool
+		plan        *migrate.Plan
+		changes     []schema.Change
+		applied     []schema.Change
+		realm       schema.Realm
+		executed    []string
+		failCounter int
+		failWith    error
+		dirty       bool
 	}
-	lockMockDriver struct{ *mockDriver }
 )
 
 // the nth call to ExecContext will fail with the given error.
@@ -533,26 +495,6 @@ func (m *mockDriver) CheckClean(context.Context, *migrate.TableIdent) error {
 	return nil
 }
 
-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++
-	if m.locks == nil {
-		m.locks = make(map[string]struct{})
-	}
-	m.locks[name] = struct{}{}
-	return func() error {
-		m.unlockCounter++
-		delete(m.locks, name)
-		return nil
-	}, nil
-}
-
-func (m *lockMockDriver) released() bool {
-	return len(m.locks) == 0
-}
-
 type mockRevisionReadWriter migrate.Revisions
 
 func (mockRevisionReadWriter) Ident() *migrate.TableIdent {