diff --git a/internal/db/schema/manager.go b/internal/db/schema/manager.go
index ca24ade31..12f81d71c 100644
--- a/internal/db/schema/manager.go
+++ b/internal/db/schema/manager.go
@@ -18,10 +18,17 @@ type driver interface {
 	Lock(context.Context) error
 	Unlock(context.Context) error
 	UnlockShared(context.Context) error
-	Run(context.Context, io.Reader) error
-	// A value of -1 indicates no version is set.
-	SetVersion(context.Context, int, bool) error
-	// A value of -1 indicates no version is set.
+	// Either starts a transactioon internal to the driver or sets a dirty
+	// bit so if the Run fails the CurrentState reflects it.
+	StartRun(context.Context) error
+	// Either commits the transaction or clears the dirty bit.
+	CommitRun() error
+	// Performs the mutation on the driver.  This should always be
+	// wrapped by StartRun and CommitRun.  The driver must properly
+	// handle the transaction or dirty bit in case of error when
+	// executing Run.
+	Run(context.Context, io.Reader, int) error
+	// A version of -1 indicates no version is set.
 	CurrentState(context.Context) (int, bool, error)
 }
 
@@ -157,6 +164,10 @@ func (b *Manager) RollForward(ctx context.Context) error {
 // through the passed in context.
 func (b *Manager) runMigrations(ctx context.Context, qp *statementProvider) error {
 	const op = "schema.(Manager).runMigrations"
+
+	if err := b.driver.StartRun(ctx); err != nil {
+		return err
+	}
 	for qp.Next() {
 		select {
 		case <-ctx.Done():
@@ -164,22 +175,12 @@ func (b *Manager) runMigrations(ctx context.Context, qp *statementProvider) erro
 		default:
 			// context is not done yet. Continue on to the next query to execute.
 		}
-
-		// set version with dirty state
-		if err := b.driver.SetVersion(ctx, qp.Version(), true); err != nil {
-			return errors.Wrap(err, op)
-		}
-
-		// TODO: Wrap all the queries provided by the statementProvider into a
-		//  single transaction which includes setting the version number.
-		if err := b.driver.Run(ctx, bytes.NewReader(qp.ReadUp())); err != nil {
-			return errors.Wrap(err, op)
-		}
-
-		// set clean state
-		if err := b.driver.SetVersion(ctx, qp.Version(), false); err != nil {
+		if err := b.driver.Run(ctx, bytes.NewReader(qp.ReadUp()), qp.Version()); err != nil {
 			return errors.Wrap(err, op)
 		}
 	}
+	if err := b.driver.CommitRun(); err != nil {
+		return err
+	}
 	return nil
 }
diff --git a/internal/db/schema/manager_test.go b/internal/db/schema/manager_test.go
index 1a32229fc..760d0c4c1 100644
--- a/internal/db/schema/manager_test.go
+++ b/internal/db/schema/manager_test.go
@@ -3,6 +3,7 @@ package schema
 import (
 	"context"
 	"database/sql"
+	"strings"
 	"testing"
 
 	"github.com/hashicorp/boundary/internal/db/schema/postgres"
@@ -61,12 +62,11 @@ func TestCurrentState(t *testing.T) {
 
 	testDriver, err := postgres.New(ctx, d)
 	require.NoError(t, err)
-	require.NoError(t, testDriver.SetVersion(ctx, 2, true))
+	require.NoError(t, testDriver.Run(ctx, strings.NewReader("SELECT 1"), 2))
 
 	want = &State{
 		InitializationStarted: true,
 		BinarySchemaVersion:   BinarySchemaVersion(dialect),
-		Dirty:                 true,
 		DatabaseSchemaVersion: 2,
 	}
 	s, err = m.CurrentState(ctx)
@@ -87,10 +87,13 @@ func TestRollForward(t *testing.T) {
 	ctx := context.Background()
 	m, err := NewManager(ctx, dialect, d)
 	require.NoError(t, err)
+	assert.NoError(t, m.RollForward(ctx))
+
 	// Now set to dirty at an early version
-	testDriver, err := postgres.New(ctx, d)
+	_, err = postgres.New(ctx, d)
 	require.NoError(t, err)
-	testDriver.SetVersion(ctx, 0, true)
+	// TODO: Extract out a way to mock the db to test failing rollforwards.
+	_, err = d.ExecContext(ctx, "TRUNCATE boundary_schema_version; INSERT INTO boundary_schema_version (Version, dirty) VALUES (2, true)")
 	assert.Error(t, m.RollForward(ctx))
 }
 
@@ -115,10 +118,10 @@ func TestRollForward_NotFromFresh(t *testing.T) {
 	require.NoError(t, err)
 	assert.NoError(t, m.RollForward(ctx))
 
-	ver, dirty, err := m.driver.CurrentState(ctx)
-	assert.NoError(t, err)
-	assert.Equal(t, nState.binarySchemaVersion, ver)
-	assert.False(t, dirty)
+	state, err := m.CurrentState(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, nState.binarySchemaVersion, state.DatabaseSchemaVersion)
+	assert.False(t, state.Dirty)
 
 	// Restore the full set of sql scripts and roll the rest of the way forward.
 	migrationStates[dialect] = oState
@@ -126,10 +129,40 @@ func TestRollForward_NotFromFresh(t *testing.T) {
 	newM, err := NewManager(ctx, dialect, d)
 	require.NoError(t, err)
 	assert.NoError(t, newM.RollForward(ctx))
-	ver, dirty, err = newM.driver.CurrentState(ctx)
-	assert.NoError(t, err)
-	assert.Equal(t, oState.binarySchemaVersion, ver)
-	assert.False(t, dirty)
+	state, err = newM.CurrentState(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, oState.binarySchemaVersion, state.DatabaseSchemaVersion)
+	assert.False(t, state.Dirty)
+}
+
+func TestRollForward_BadSQL(t *testing.T) {
+	dialect := "postgres"
+	oState := migrationStates[dialect]
+	defer func() { migrationStates[dialect] = oState }()
+
+	nState := createPartialMigrationState(oState, 8)
+	nState.binarySchemaVersion = 10
+	nState.upMigrations[10] = []byte("SELECT 1 FROM NonExistantTable;")
+	migrationStates[dialect] = nState
+
+	c, u, _, err := docker.StartDbInDocker(dialect)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, c())
+	})
+	d, err := sql.Open(dialect, u)
+	require.NoError(t, err)
+
+	// Initialize the DB with only a portion of the current sql scripts.
+	ctx := context.Background()
+	m, err := NewManager(ctx, dialect, d)
+	require.NoError(t, err)
+	assert.Error(t, m.RollForward(ctx))
+
+	state, err := m.CurrentState(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, nilVersion, state.DatabaseSchemaVersion)
+	assert.False(t, state.Dirty)
 }
 
 func TestManager_ExclusiveLock(t *testing.T) {
diff --git a/internal/db/schema/migrations/generate/generate.go b/internal/db/schema/migrations/generate/generate.go
index c74563657..8d4aad41e 100644
--- a/internal/db/schema/migrations/generate/generate.go
+++ b/internal/db/schema/migrations/generate/generate.go
@@ -75,12 +75,6 @@ func generate(dialect string) {
 				continue
 			}
 
-			contents, err := ioutil.ReadFile(fmt.Sprintf("%s/%s/%s/%s", srcDir, dialect, ver, name))
-			if err != nil {
-				fmt.Printf("error opening file %s with dialect %s: %v", name, dialect, err)
-				os.Exit(1)
-			}
-
 			nameParts := strings.SplitN(name, "_", 2)
 			if len(nameParts) != 2 {
 				continue
@@ -96,9 +90,25 @@ func generate(dialect string) {
 			if fullV > largestSchemaVersion {
 				largestSchemaVersion = fullV
 			}
+
+			cbts, err := ioutil.ReadFile(fmt.Sprintf("%s/%s/%s/%s", srcDir, dialect, ver, name))
+			if err != nil {
+				fmt.Printf("error opening file %s with dialect %s: %v", name, dialect, err)
+				os.Exit(1)
+			}
+
+			contents := strings.TrimSpace(string(cbts))
+			if strings.ToLower(contents[:len("begin;")]) == "begin;" {
+				contents = contents[len("begin;"):]
+			}
+			if strings.ToLower(contents[len(contents)-len("commit;"):]) == "commit;" {
+				contents = contents[:len(contents)-len("commit;")]
+			}
+			contents = strings.TrimSpace(contents)
+
 			cv := ContentValues{
 				Name:    fmt.Sprint(fullV),
-				Content: string(contents),
+				Content: contents,
 			}
 			switch {
 			case strings.Contains(nameParts[1], ".up."):
diff --git a/internal/db/schema/migrations/postgres/0/41_targets.up.sql b/internal/db/schema/migrations/postgres/0/41_targets.up.sql
index 13ebfd80f..0a19f2d5c 100644
--- a/internal/db/schema/migrations/postgres/0/41_targets.up.sql
+++ b/internal/db/schema/migrations/postgres/0/41_targets.up.sql
@@ -1,3 +1,5 @@
+begin;
+
 /*
                                    ┌─────────────────┐
 ┌─────────────────┐                │   target_tcp    │
@@ -33,8 +35,6 @@
 
 */
 
-begin;
-
 create table target (
   public_id wt_public_id primary key,
   scope_id wt_scope_id not null 
diff --git a/internal/db/schema/postgres/postgres.go b/internal/db/schema/postgres/postgres.go
index 872eaf67c..feccbc3e4 100644
--- a/internal/db/schema/postgres/postgres.go
+++ b/internal/db/schema/postgres/postgres.go
@@ -57,6 +57,8 @@ type Postgres struct {
 	// Locking and unlocking need to use the same connection
 	conn *sql.Conn
 	db   *sql.DB
+
+	tx *sql.Tx
 }
 
 // New returns a postgres pointer with the provided db verified as
@@ -154,10 +156,41 @@ func (p *Postgres) UnlockShared(ctx context.Context) error {
 	return nil
 }
 
+// StartRun starts a transaction that all subsequent calls to Run will use.
+func (p *Postgres) StartRun(ctx context.Context) error {
+	tx, err := p.conn.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	p.tx = tx
+	return nil
+}
+
+// CommitRun commits the pending transaction if there is one
+func (p *Postgres) CommitRun() error {
+	const op = "postgres.(Postgres).CommitRun"
+	defer func() {
+		p.tx = nil
+	}()
+	if p.tx == nil {
+		return errors.New(errors.MigrationIntegrity, op, "no pending transaction")
+	}
+	if err := p.tx.Commit(); err != nil {
+		if errRollback := p.tx.Rollback(); errRollback != nil {
+			return multierror.Append(err, errRollback)
+		}
+	}
+	return nil
+}
+
+type execContexter interface {
+	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
+}
+
 // Executes the sql provided in the passed in io.Reader.  The contents of the reader must
 // fit in memory as the full content is read into a string before being passed to the
 // backing database.
-func (p *Postgres) Run(ctx context.Context, migration io.Reader) error {
+func (p *Postgres) Run(ctx context.Context, migration io.Reader, version int) error {
 	const op = "postgres.(Postgres).Run"
 	migr, err := ioutil.ReadAll(migration)
 	if err != nil {
@@ -165,7 +198,21 @@ func (p *Postgres) Run(ctx context.Context, migration io.Reader) error {
 	}
 	// Run migration
 	query := string(migr)
-	if _, err := p.conn.ExecContext(ctx, query); err != nil {
+
+	var extr execContexter = p.conn
+	rollback := func() error { return nil }
+	if p.tx != nil {
+		extr = p.tx
+		rollback = func() error {
+			defer func() { p.tx = nil }()
+			return p.tx.Rollback()
+		}
+	}
+
+	if _, err := extr.ExecContext(ctx, query); err != nil {
+		if rollbackErr := rollback(); rollbackErr != nil {
+			err = multierror.Append(err, rollbackErr)
+		}
 		if pgErr, ok := err.(*pq.Error); ok {
 			var line uint
 			var col uint
@@ -187,6 +234,9 @@ func (p *Postgres) Run(ctx context.Context, migration io.Reader) error {
 		}
 		return errors.Wrap(err, op, errors.WithMsg(fmt.Sprintf("migration failed: %s", migr)))
 	}
+	if err := p.setVersion(ctx, version, false); err != nil {
+		return errors.Wrap(err, op)
+	}
 
 	return nil
 }
@@ -226,18 +276,26 @@ func runesLastIndex(input []rune, target rune) int {
 	return -1
 }
 
-// SetVersion sets the version number, and whether the database is in a dirty state.
+// setVersion sets the version number, and whether the database is in a dirty state.
 // A version value of -1 indicates no version is set.
-func (p *Postgres) SetVersion(ctx context.Context, version int, dirty bool) error {
+func (p *Postgres) setVersion(ctx context.Context, version int, dirty bool) error {
 	const op = "postgres.(Postgres).SetVersion"
-	tx, err := p.conn.BeginTx(ctx, &sql.TxOptions{})
-	if err != nil {
-		return err
+	tx := p.tx
+	var err error
+	if tx == nil {
+		tx, err = p.conn.BeginTx(ctx, nil)
+		if err != nil {
+			return err
+		}
+	}
+	rollback := func() error {
+		defer func() { p.tx = nil }()
+		return tx.Rollback()
 	}
 
 	query := `TRUNCATE ` + pq.QuoteIdentifier(defaultMigrationsTable)
 	if _, err := tx.ExecContext(ctx, query); err != nil {
-		if errRollback := tx.Rollback(); errRollback != nil {
+		if errRollback := rollback(); errRollback != nil {
 			err = multierror.Append(err, errRollback)
 		}
 		return errors.Wrap(err, op)
@@ -250,24 +308,25 @@ func (p *Postgres) SetVersion(ctx context.Context, version int, dirty bool) erro
 		query = `INSERT INTO ` + pq.QuoteIdentifier(defaultMigrationsTable) +
 			` (Version, dirty) VALUES ($1, $2)`
 		if _, err := tx.ExecContext(ctx, query, version, dirty); err != nil {
-			if errRollback := tx.Rollback(); errRollback != nil {
+			if errRollback := rollback(); errRollback != nil {
 				err = multierror.Append(err, errRollback)
 			}
 			return errors.Wrap(err, op)
 		}
 	}
 
-	if err := tx.Commit(); err != nil {
-		return errors.Wrap(err, op)
+	if p.tx == nil {
+		if err := tx.Commit(); err != nil {
+			return errors.Wrap(err, op)
+		}
 	}
-
 	return nil
 }
 
 // CurrentState returns the version, if the database is currently in a dirty state, and any error.
 // A version value of -1 indicates no version is set.
 func (p *Postgres) CurrentState(ctx context.Context) (version int, dirty bool, err error) {
-	const op = "postgres.(Postgres).Version"
+	const op = "postgres.(Postgres).CurrentState"
 	query := `SELECT Version, dirty FROM ` + pq.QuoteIdentifier(defaultMigrationsTable) + ` LIMIT 1`
 	err = p.conn.QueryRowContext(ctx, query).Scan(&version, &dirty)
 	switch {
diff --git a/internal/db/schema/postgres/postgres_test.go b/internal/db/schema/postgres/postgres_test.go
index 85af44922..b4be6e343 100644
--- a/internal/db/schema/postgres/postgres_test.go
+++ b/internal/db/schema/postgres/postgres_test.go
@@ -163,7 +163,7 @@ func TestMultiStatement(t *testing.T) {
 				t.Error(err)
 			}
 		}()
-		if err := d.Run(ctx, strings.NewReader("CREATE TABLE foo (foo text); CREATE TABLE bar (bar text);")); err != nil {
+		if err := d.Run(ctx, strings.NewReader("CREATE TABLE foo (foo text); CREATE TABLE bar (bar text);"), 2); err != nil {
 			t.Fatalf("expected err to be nil, got %v", err)
 		}
 
@@ -178,6 +178,45 @@ func TestMultiStatement(t *testing.T) {
 	})
 }
 
+func TestTransaction(t *testing.T) {
+	dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
+		ctx := context.Background()
+		ip, port, err := c.FirstPort()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		addr := pgConnectionString(ip, port)
+		d, err := open(t, ctx, addr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer func() {
+			if err := d.close(t); err != nil {
+				t.Error(err)
+			}
+		}()
+
+		assert.NoError(t, d.StartRun(ctx))
+		assert.NoError(t, d.Run(ctx, strings.NewReader("CREATE TABLE foo (foo text);"), 2))
+		assert.NoError(t, d.Run(ctx, strings.NewReader("SELECT 1"), 3))
+		assert.NoError(t, d.CommitRun())
+		v, dirty, err := d.CurrentState(ctx)
+		assert.NoError(t, err)
+		assert.False(t, dirty)
+		assert.Equal(t, 3, v)
+
+		assert.NoError(t, d.StartRun(ctx))
+		assert.NoError(t, d.Run(ctx, strings.NewReader("CREATE TABLE bar (bar text);"), 20))
+		assert.Error(t, d.Run(ctx, strings.NewReader("SELECT 1 FROM NonExistingTable"), 30))
+		assert.Error(t, d.CommitRun())
+		v, dirty, err = d.CurrentState(ctx)
+		assert.NoError(t, err)
+		assert.False(t, dirty)
+		assert.Equal(t, 3, v)
+	})
+}
+
 func TestWithSchema(t *testing.T) {
 	dktesting.ParallelTest(t, specs, func(t *testing.T, c dktest.ContainerInfo) {
 		ctx := context.Background()
@@ -198,10 +237,7 @@ func TestWithSchema(t *testing.T) {
 		}()
 
 		// create foobar schema
-		if err := d.Run(ctx, strings.NewReader("CREATE SCHEMA foobar AUTHORIZATION postgres")); err != nil {
-			t.Fatal(err)
-		}
-		if err := d.SetVersion(ctx, 1, false); err != nil {
+		if err := d.Run(ctx, strings.NewReader("CREATE SCHEMA foobar AUTHORIZATION postgres"), 1); err != nil {
 			t.Fatal(err)
 		}
 
@@ -226,7 +262,7 @@ func TestWithSchema(t *testing.T) {
 		}
 
 		// now update CurrentState and compare
-		if err := d2.SetVersion(ctx, 2, false); err != nil {
+		if err := d2.setVersion(ctx, 2, false); err != nil {
 			t.Fatal(err)
 		}
 		version, _, err = d2.CurrentState(ctx)
@@ -303,7 +339,7 @@ func TestRun_Error(t *testing.T) {
 			require.NoError(t, p.close(t))
 		})
 
-		err = p.Run(ctx, bytes.NewReader([]byte("SELECT *\nFROM foo")))
+		err = p.Run(ctx, bytes.NewReader([]byte("SELECT *\nFROM foo")), 2)
 		assert.Error(t, err)
 	})
 }
diff --git a/internal/db/schema/postgres/testing.go b/internal/db/schema/postgres/testing.go
index b4cc148d5..c736ae7a0 100644
--- a/internal/db/schema/postgres/testing.go
+++ b/internal/db/schema/postgres/testing.go
@@ -98,7 +98,7 @@ func testRun(t *testing.T, d *Postgres, migration io.Reader) {
 		t.Fatal("migration can't be nil")
 	}
 
-	if err := d.Run(ctx, migration); err != nil {
+	if err := d.Run(ctx, migration, 1); err != nil {
 		t.Fatal(err)
 	}
 }
@@ -132,7 +132,7 @@ func testSetVersion(t *testing.T, d *Postgres) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			err := d.SetVersion(ctx, tc.version, tc.dirty)
+			err := d.setVersion(ctx, tc.version, tc.dirty)
 			if err != tc.expectedErr {
 				t.Fatal("Got unexpected error:", err, "!=", tc.expectedErr)
 			}
diff --git a/internal/db/schema/postgres_migration.gen.go b/internal/db/schema/postgres_migration.gen.go
index 2b84eb79e..6f47cae8a 100644
--- a/internal/db/schema/postgres_migration.gen.go
+++ b/internal/db/schema/postgres_migration.gen.go
@@ -8,8 +8,6 @@ func init() {
 		binarySchemaVersion: 1086,
 		upMigrations: map[int][]byte{
 			1: []byte(`
-begin;
-
 create domain wt_public_id as text
 check(
   length(trim(value)) > 10
@@ -164,13 +162,8 @@ comment on function
   immutable_columns()
 is
   'function used in before update triggers to make columns immutable';
-
-commit;
-
 `),
 			2: []byte(`
-begin;
-
 -- TODO (jimlambrt 7/2020) remove update_time
 create table if not exists oplog_entry (
   id bigint generated always as identity primary key,
@@ -278,15 +271,8 @@ values
   ('db_test_scooter', 1),
   ('auth_account', 1),
   ('iam_principal_role', 1);
-  
-
-commit;
-
-
 `),
 			3: []byte(`
-begin;
-
 -- create test tables used in the unit tests for the internal/db package 
 -- these tables (db_test_user, db_test_car, db_test_rental, db_test_scooter) are
 -- not part of the boundary domain model... they are simply used for testing
@@ -414,13 +400,8 @@ create trigger
 before
 insert on db_test_scooter
   for each row execute procedure default_create_time();
-
-commit;
-
 `),
 			6: []byte(`
-begin;
-
 create table iam_scope_type_enm (
   string text not null primary key
     constraint only_predefined_scope_types_allowed
@@ -1123,14 +1104,8 @@ from
 where
   gm.member_id = u.public_id and
   gm.group_id = g.public_id;
-  
-
-commit;
-
 `),
 			7: []byte(`
-begin;
-
 /*
 
   ┌────────────────┐               ┌────────────────┐
@@ -1293,13 +1268,8 @@ begin;
     before update of iam_user_id on auth_account
     for each row
     execute procedure update_iam_user_auth_account();
-
-commit;
-
 `),
 			8: []byte(`
-begin;
-
 -- For now at least the IDs will be the same as the name, because this allows us
 -- to not have to persist some generated ID to worker and controller nodes.
 -- Eventually we may want them to diverge, so we have both here for now.
@@ -1346,14 +1316,9 @@ create trigger
 before
 update on recovery_nonces
   for each row execute procedure immutable_columns('nonce', 'create_time');
-
-commit;
-
 `),
 			11: []byte(`
-begin;
-
-  -- an auth token belongs to 1 and only 1 auth account
+-- an auth token belongs to 1 and only 1 auth account
   -- an auth account can have 0 to many auth tokens
   create table auth_token (
     public_id wt_public_id primary key,
@@ -1485,13 +1450,8 @@ begin;
   before
   update on auth_token
     for each row execute procedure immutable_columns('public_id', 'auth_account_id', 'create_time');
-
-commit;
-
 `),
 			12: []byte(`
-begin;
-
 /*
 
        ┌────────────────┐                 ┌──────────────────────┐             ┌────────────────────────────┐
@@ -1784,14 +1744,9 @@ begin;
     ('auth_password_method', 1),
     ('auth_password_account', 1),
     ('auth_password_credential', 1);
-
-commit;
-
 `),
 			13: []byte(`
-begin;
-
-  create table auth_password_argon2_conf (
+create table auth_password_argon2_conf (
     private_id wt_private_id primary key
       references auth_password_conf (private_id)
       on delete cascade
@@ -1936,14 +1891,9 @@ begin;
   values
     ('auth_password_argon2_conf', 1),
     ('auth_password_argon2_cred', 1);
-
-commit;
-
 `),
 			14: []byte(`
-begin;
-
-  -- auth_password_conf_union is a union of the configuration settings
+-- auth_password_conf_union is a union of the configuration settings
   -- of all supported key derivation functions.
   -- It will be updated as new key derivation functions are supported.
   create or replace view auth_password_conf_union as
@@ -1964,13 +1914,8 @@ begin;
         from auth_password_method pm
   inner join auth_password_conf_union c
           on pm.password_conf_id = c.password_conf_id;
-
-commit;
-
 `),
 			20: []byte(`
-begin;
-
 /*
 
                                ┌─────────────────┐
@@ -2135,13 +2080,8 @@ begin;
     ('host_catalog', 1),
     ('host', 1),
     ('host_set', 1);
-
-commit;
-
 `),
 			22: []byte(`
-begin;
-
 /*
 
   ┌─────────────────┐          ┌─────────────────────┐
@@ -2339,13 +2279,8 @@ begin;
     ('static_host', 1),
     ('static_host_set', 1),
     ('static_host_set_member', 1);
-
-commit;
-
 `),
 			30: []byte(`
-begin;
-
 -- kms_version_column() will increment the version column whenever row data
 -- is inserted and should only be used in an before insert trigger.  This
 -- function will overwrite any explicit values to the version column.
@@ -2371,12 +2306,8 @@ comment on function
   kms_version_column()
 is
   'function used in before insert triggers to properly set version columns for kms_* tables with a version column';
-  
-  commit;
 `),
 			31: []byte(`
-begin;
-
 /*
              ┌────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐            
              ├────────────────────────────────────────────────────────────────────────────────────────────┐                           ○            
@@ -2698,13 +2629,8 @@ before insert on kms_token_key_version
   values
     ('kms_root_key', 1),
     ('kms_root_key_version', 1);
-    
-commit;
 `),
 			40: []byte(`
-begin;
-
-
 -- insert_target_subtype() is a before insert trigger
 -- function for subtypes of target
 create or replace function
@@ -2770,8 +2696,6 @@ end if;
 return new;
 end;
 $$ language plpgsql;
-
-commit;
 `),
 			41: []byte(`
 /*
@@ -2809,8 +2733,6 @@ commit;
 
 */
 
-begin;
-
 create table target (
   public_id wt_public_id primary key,
   scope_id wt_scope_id not null 
@@ -2963,13 +2885,8 @@ insert into oplog_ticket
   (name, version)
 values
   ('target_tcp', 1);
-
-commit;
-
 `),
 			50: []byte(`
-begin;
-
 /*
 
              ┌─────────────────┐               ┌─────────────────┐   ┌─────────────────┐
@@ -3455,13 +3372,8 @@ begin;
     session_state ss
   where 
     s.public_id = ss.session_id;
-
-commit;
-
 `),
 			51: []byte(`
-begin;
-
 /*
 
                ┌────────────────┐
@@ -3835,15 +3747,9 @@ create or replace function
     );
  end;
   $$ language plpgsql;
-
-
-commit;
-
 `),
 			60: []byte(`
-begin;
-
-  create extension if not exists "pgcrypto";
+create extension if not exists "pgcrypto";
 
   create domain wh_inet_port as integer
   check(
@@ -3923,14 +3829,9 @@ begin;
   as $$
     select wh_time_id(current_timestamp);
   $$ language sql;
-
-commit;
-
 `),
 			62: []byte(`
-begin;
-
-  create table wh_date_dimension (
+create table wh_date_dimension (
     id                            integer      primary key,
     date                          date         not null,
     calendar_quarter              wh_dim_text,
@@ -4023,14 +3924,9 @@ begin;
            date_trunc('day', current_timestamp) + interval '24 hours' - interval '1 second',
            interval '1 second'
          ) as t(second);
-
-commit;
-
 `),
 			65: []byte(`
-begin;
-
-  create table wh_host_dimension (
+create table wh_host_dimension (
     -- random id generated using encode(digest(gen_random_bytes(16), 'sha256'), 'base64')
     -- this is done to prevent conflicts with rows in other clusters
     -- which enables warehouse data from multiple clusters to be loaded into a
@@ -4255,14 +4151,9 @@ begin;
       from wh_user_dimension
      where current_row_indicator = 'Current'
   ;
-
-commit;
-
 `),
 			66: []byte(`
-begin;
-
-  -- wh_upsert_host returns the wh_host_dimension id for p_host_id,
+-- wh_upsert_host returns the wh_host_dimension id for p_host_id,
   -- p_host_set_id, and p_target_id. wh_upsert_host compares the current values
   -- in the wh_host_dimension with the current values in the operational tables
   -- for the provide parameters. If the values between the operational tables
@@ -4397,13 +4288,9 @@ begin;
 
   end;
   $$ language plpgsql;
-commit;
-
 `),
 			68: []byte(`
-begin;
-
-  -- Column names for numeric fields that are not a measurement end in id or
+-- Column names for numeric fields that are not a measurement end in id or
   -- number. This naming convention enables automatic field type detection in
   -- certain data analysis tools.
   -- https://help.tableau.com/current/pro/desktop/en-us/data_clean_adm.htm
@@ -4585,14 +4472,9 @@ begin;
     'additive measurement.';
 
   create index on wh_session_connection_accumulating_fact(session_id);
-
-commit;
-
 `),
 			69: []byte(`
-begin;
-
-  -- wh_rollup_connections calculates the aggregate values from
+-- wh_rollup_connections calculates the aggregate values from
   -- wh_session_connection_accumulating_fact for p_session_id and updates
   -- wh_session_accumulating_fact for p_session_id with those values.
   create or replace function wh_rollup_connections(p_session_id wt_public_id)
@@ -4861,13 +4743,8 @@ begin;
     after insert on session_connection_state
     for each row
     execute function wh_insert_session_connection_state();
-
-commit;
-
 `),
 			1001: []byte(`
-begin;
-
 -- This series of expressions fixes the primary key on the server table
 alter table session
   drop constraint session_server_id_server_type_fkey;
@@ -4987,8 +4864,6 @@ create table server_tag (
   value wt_tagpair,
   primary key(server_id, key, value)
 );
-
-commit;
 `),
 			1080: []byte(`
 begin;
diff --git a/internal/db/schema/schema_test.go b/internal/db/schema/schema_test.go
index 210831e9b..a877350a1 100644
--- a/internal/db/schema/schema_test.go
+++ b/internal/db/schema/schema_test.go
@@ -2,7 +2,6 @@ package schema
 
 import (
 	"context"
-	"database/sql"
 	"testing"
 
 	"github.com/hashicorp/boundary/internal/docker"
@@ -42,24 +41,3 @@ func TestMigrateStore(t *testing.T) {
 	assert.NoError(t, err)
 	assert.False(t, ran)
 }
-
-func TestMigrateStore_Dirty(t *testing.T) {
-	dialect := "postgres"
-	ctx := context.Background()
-
-	c, u, _, err := docker.StartDbInDocker(dialect)
-	require.NoError(t, err)
-	t.Cleanup(func() {
-		require.NoError(t, c())
-	})
-
-	// Mark the db as dirty indicating a previously run failed migration
-	db, err := sql.Open(dialect, u)
-	require.NoError(t, err)
-	m, err := NewManager(ctx, dialect, db)
-	m.driver.SetVersion(ctx, -1, true)
-
-	b, err := MigrateStore(ctx, dialect, u)
-	assert.Error(t, err)
-	assert.False(t, b)
-}