Merge remote-tracking branch 'origin/master' into aaron/run-altertests-separate-in-ci

.github/workflows/ci-sql-correctness-release.yaml

@@ -1,3 +1,5 @@
+name: SQL Correctness
+
 on:
   repository_dispatch:
     types: [ release-dolt ]

go/cmd/dolt/commands/sql.go

@@ -365,7 +365,7 @@ func parseCommitSpec(dEnv *env.DoltEnv, apr *argparser.ArgParseResults) (*doltdb
 }
 
 func execShell(sqlCtx *sql.Context, readOnly bool, mrEnv env.MultiRepoEnv, roots map[string]*doltdb.RootValue, format resultFormat) errhand.VerboseError {
-	dbs := CollectDBs(mrEnv, newDatabase)
+	dbs := CollectDBs(mrEnv)
 	se, err := newSqlEngine(sqlCtx, readOnly, mrEnv, roots, format, dbs...)
 	if err != nil {
 		return errhand.VerboseErrorFromError(err)
@@ -379,32 +379,44 @@ func execShell(sqlCtx *sql.Context, readOnly bool, mrEnv env.MultiRepoEnv, roots
 }
 
 func execBatch(sqlCtx *sql.Context, readOnly bool, mrEnv env.MultiRepoEnv, roots map[string]*doltdb.RootValue, batchInput io.Reader, format resultFormat) errhand.VerboseError {
-	dbs := CollectDBs(mrEnv, newBatchedDatabase)
+	dbs := CollectDBs(mrEnv)
 	se, err := newSqlEngine(sqlCtx, readOnly, mrEnv, roots, format, dbs...)
 	if err != nil {
 		return errhand.VerboseErrorFromError(err)
 	}
 
+	// In batch mode, we need to set a couple flags on the session to prevent flushes to disk after every commit
+	dsqle.DSessFromSess(sqlCtx.Session).EnableBatchedMode()
+	err = sqlCtx.Session.SetSessionVariable(sqlCtx, sql.AutoCommitSessionVar, true)
+	if err != nil {
+		return errhand.VerboseErrorFromError(err)
+	}
+
 	err = runBatchMode(sqlCtx, se, batchInput)
 	if err != nil {
+		// If we encounter an error, flush what we have so far to disk before exiting, except in the case of merge
+		// errors, which have already updated the repo state all they're going to (and writing session root on top of
+		// them would overwrite these changes)
+		// TODO: this is a mess, merge conflicts need to follow the same code path as everything else
+		if err == doltdb.ErrUnresolvedConflicts || err == doltdb.ErrMergeActive {
+			return errhand.BuildDError("Error processing batch").Build()
+		}
+
+		_ = flushBatchedEdits(sqlCtx, se)
+		_ = writeRoots(sqlCtx, se, mrEnv, roots)
+
 		return errhand.BuildDError("Error processing batch").Build()
 	}
 
 	return writeRoots(sqlCtx, se, mrEnv, roots)
 }
 
-type createDBFunc func(name string, dEnv *env.DoltEnv) dsqle.Database
-
 func newDatabase(name string, dEnv *env.DoltEnv) dsqle.Database {
 	return dsqle.NewDatabase(name, dEnv.DbData())
 }
 
-func newBatchedDatabase(name string, dEnv *env.DoltEnv) dsqle.Database {
-	return dsqle.NewBatchedDatabase(name, dEnv.DbData())
-}
-
 func execQuery(sqlCtx *sql.Context, readOnly bool, mrEnv env.MultiRepoEnv, roots map[string]*doltdb.RootValue, query string, format resultFormat) errhand.VerboseError {
-	dbs := CollectDBs(mrEnv, newDatabase)
+	dbs := CollectDBs(mrEnv)
 	se, err := newSqlEngine(sqlCtx, readOnly, mrEnv, roots, format, dbs...)
 	if err != nil {
 		return errhand.VerboseErrorFromError(err)
@@ -427,10 +439,10 @@ func execQuery(sqlCtx *sql.Context, readOnly bool, mrEnv env.MultiRepoEnv, roots
 
 // CollectDBs takes a MultiRepoEnv and creates Database objects from each environment and returns a slice of these
 // objects.
-func CollectDBs(mrEnv env.MultiRepoEnv, createDB createDBFunc) []dsqle.Database {
+func CollectDBs(mrEnv env.MultiRepoEnv) []dsqle.Database {
 	dbs := make([]dsqle.Database, 0, len(mrEnv))
 	_ = mrEnv.Iter(func(name string, dEnv *env.DoltEnv) (stop bool, err error) {
-		db := createDB(name, dEnv)
+		db := newDatabase(name, dEnv)
 		dbs = append(dbs, db)
 		return false, nil
 	})
@@ -1095,7 +1107,7 @@ func processNonBatchableQuery(ctx *sql.Context, se *sqlEngine, query string, sql
 	if rowIter != nil {
 		err = mergeResultIntoStats(sqlStatement, rowIter, batchEditStats)
 		if err != nil {
-			return fmt.Errorf("error executing statement: %v", err.Error())
+			return err
 		}
 
 		// Some statement types should print results, even in batch mode.
@@ -1125,7 +1137,7 @@ func processNonBatchableQuery(ctx *sql.Context, se *sqlEngine, query string, sql
 func processBatchableEditQuery(ctx *sql.Context, se *sqlEngine, query string, sqlStatement sqlparser.Statement) (returnErr error) {
 	_, rowIter, err := se.query(ctx, query)
 	if err != nil {
-		return fmt.Errorf("Error inserting rows: %v", err.Error())
+		return err
 	}
 
 	if rowIter != nil {
@@ -1137,7 +1149,7 @@ func processBatchableEditQuery(ctx *sql.Context, se *sqlEngine, query string, sq
 		}()
 		err = mergeResultIntoStats(sqlStatement, rowIter, batchEditStats)
 		if err != nil {
-			return fmt.Errorf("Error inserting rows: %v", err.Error())
+			return err
 		}
 	}
 

go/cmd/dolt/commands/sqlserver/server.go

@@ -116,7 +116,7 @@ func Serve(ctx context.Context, version string, serverConfig ServerConfig, serve
 		}
 	}
 
-	dbs := commands.CollectDBs(mrEnv, newDatabase)
+	dbs := commands.CollectDBs(mrEnv)
 
 	for _, db := range dbs {
 		sqlEngine.AddDatabase(db)

go/cmd/dolt/dolt.go

@@ -49,7 +49,7 @@ import (
 )
 
 const (
-	Version = "0.26.6"
+	Version = "0.26.7"
 )
 
 var dumpDocsCommand = &commands.DumpDocsCmd{}

go/go.mod

@@ -18,11 +18,11 @@ require (
 	github.com/denisbrodbeck/machineid v1.0.1
 	github.com/dolthub/dolt/go/gen/proto/dolt/services/eventsapi v0.0.0-20201005193433-3ee972b1d078
 	github.com/dolthub/fslock v0.0.2
-	github.com/dolthub/go-mysql-server v0.10.1-0.20210608215915-0d05d07f7f23
+	github.com/dolthub/go-mysql-server v0.10.1-0.20210610023714-4bbd1ce883f9
 	github.com/dolthub/ishell v0.0.0-20210205014355-16a4ce758446
 	github.com/dolthub/mmap-go v1.0.4-0.20201107010347-f9f2a9588a66
 	github.com/dolthub/sqllogictest/go v0.0.0-20201107003712-816f3ae12d81
-	github.com/dolthub/vitess v0.0.0-20210608004428-4e83b958d1ed
+	github.com/dolthub/vitess v0.0.0-20210610023035-d2579eac208d
 	github.com/dustin/go-humanize v1.0.0
 	github.com/fatih/color v1.9.0
 	github.com/flynn-archive/go-shlex v0.0.0-20150515145356-3f9db97f8568

go/go.sum

@@ -142,24 +142,16 @@ github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZm
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
 github.com/dolthub/fslock v0.0.2 h1:8vUh47iKovgrtXNrXVIzsIoWLlspoXg+3nslhUzgKSw=
 github.com/dolthub/fslock v0.0.2/go.mod h1:0i7bsNkK+XHwFL3dIsSWeXSV7sykVzzVr6+jq8oeEo0=
-github.com/dolthub/go-mysql-server v0.10.1-0.20210603222011-4c1a2422c236 h1:CJ1hxi3di8i+Wv9Cko0AJAnZrbtaiXlP0pPWbsifwzk=
-github.com/dolthub/go-mysql-server v0.10.1-0.20210603222011-4c1a2422c236/go.mod h1:nY+4DzVnlryGHlnvkeDjplQH3SLBSymHs0woPHrBIfw=
-github.com/dolthub/go-mysql-server v0.10.1-0.20210608182718-5ce80eb27ec1 h1:Az2XOZcS9ExcLCdwasdHeaRaUqYCCQpLTd+jkyTv5yY=
-github.com/dolthub/go-mysql-server v0.10.1-0.20210608182718-5ce80eb27ec1/go.mod h1:HTh0n35wJXBN1ltNYLlAOhUEY3PmALNosMQpTXfLuKA=
-github.com/dolthub/go-mysql-server v0.10.1-0.20210608212241-d1af8fcdbd22 h1:5PmKH5ziWAs0IkdGIb1U77Gjo/001aX/rBoBHD/MgC0=
-github.com/dolthub/go-mysql-server v0.10.1-0.20210608212241-d1af8fcdbd22/go.mod h1:nY+4DzVnlryGHlnvkeDjplQH3SLBSymHs0woPHrBIfw=
-github.com/dolthub/go-mysql-server v0.10.1-0.20210608215915-0d05d07f7f23 h1:nSzyx2Qv+aIPATvNlJ4U5Q/qU+5WMRqxicv9Uhlkjsk=
-github.com/dolthub/go-mysql-server v0.10.1-0.20210608215915-0d05d07f7f23/go.mod h1:HTh0n35wJXBN1ltNYLlAOhUEY3PmALNosMQpTXfLuKA=
-github.com/dolthub/go-mysql-server v0.10.1-0.20210608211752-782d278b4e07 h1:2cWaX0LEB2hKB9lAfv2I0DaQDhtU8q4bKuucQC+wXUI=
-github.com/dolthub/go-mysql-server v0.10.1-0.20210608211752-782d278b4e07/go.mod h1:HTh0n35wJXBN1ltNYLlAOhUEY3PmALNosMQpTXfLuKA=
+github.com/dolthub/go-mysql-server v0.10.1-0.20210610023714-4bbd1ce883f9 h1:MlkjKEvZylziuYKp3YTIr1cQi97Q33/iF+u0z9qAb3Q=
+github.com/dolthub/go-mysql-server v0.10.1-0.20210610023714-4bbd1ce883f9/go.mod h1:EN0uqWlk59rt6Lmlvf+UO2A+4iIm5A5GHMwGD0Zg01g=
 github.com/dolthub/ishell v0.0.0-20210205014355-16a4ce758446 h1:0ol5pj+QlKUKAtqs1LiPM3ZJKs+rHPgLSsMXmhTrCAM=
 github.com/dolthub/ishell v0.0.0-20210205014355-16a4ce758446/go.mod h1:dhGBqcCEfK5kuFmeO5+WOx3hqc1k3M29c1oS/R7N4ms=
 github.com/dolthub/mmap-go v1.0.4-0.20201107010347-f9f2a9588a66 h1:WRPDbpJWEnPxPmiuOTndT+lUWUeGjx6eoNOK9O4tQQQ=
 github.com/dolthub/mmap-go v1.0.4-0.20201107010347-f9f2a9588a66/go.mod h1:N5ZIbMGuDUpTpOFQ7HcsN6WSIpTGQjHP+Mz27AfmAgk=
 github.com/dolthub/sqllogictest/go v0.0.0-20201107003712-816f3ae12d81 h1:7/v8q9XGFa6q5Ap4Z/OhNkAMBaK5YeuEzwJt+NZdhiE=
 github.com/dolthub/sqllogictest/go v0.0.0-20201107003712-816f3ae12d81/go.mod h1:siLfyv2c92W1eN/R4QqG/+RjjX5W2+gCTRjZxBjI3TY=
-github.com/dolthub/vitess v0.0.0-20210608004428-4e83b958d1ed h1:zwUqqNhyFDus/BrV9519n4DyATCOn6yv3xm32p3UEEg=
-github.com/dolthub/vitess v0.0.0-20210608004428-4e83b958d1ed/go.mod h1:hUE8oSk2H5JZnvtlLBhJPYC8WZCA5AoSntdLTcBvdBM=
+github.com/dolthub/vitess v0.0.0-20210610023035-d2579eac208d h1:N5MvVTgc6eZhpPB1w2oL1H/USdMjseGUBrw2QD90ZHA=
+github.com/dolthub/vitess v0.0.0-20210610023035-d2579eac208d/go.mod h1:hUE8oSk2H5JZnvtlLBhJPYC8WZCA5AoSntdLTcBvdBM=
 github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=

go/libraries/doltcore/doltdb/errors.go

@@ -47,6 +47,9 @@ var ErrUpToDate = errors.New("up to date")
 var ErrIsAhead = errors.New("current fast forward from a to b. a is ahead of b already")
 var ErrIsBehind = errors.New("cannot reverse from b to a. b is a is behind a already")
 
+var ErrUnresolvedConflicts = errors.New("merge has unresolved conflicts")
+var ErrMergeActive = errors.New("merging is not possible because you have not committed an active merge")
+
 type ErrClientOutOfDate struct {
 	RepoVer   FeatureVersion
 	ClientVer FeatureVersion

go/libraries/doltcore/doltdb/root_val.go

@@ -15,6 +15,7 @@
 package doltdb
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -1189,3 +1190,37 @@ func validateTagUniqueness(ctx context.Context, root *RootValue, tableName strin
 
 	return nil
 }
+
+// DebugString returns a human readable string with the contents of this root. If |transitive| is true, row data from
+// all tables is also included. This method is very expensive for large root values, so |transitive| should only be used
+// when debugging tests.
+func (root *RootValue) DebugString(ctx context.Context, transitive bool) string {
+	var buf bytes.Buffer
+	err := types.WriteEncodedValue(ctx, &buf, root.valueSt)
+	if err != nil {
+		panic(err)
+	}
+
+	if transitive {
+		buf.WriteString("\nTables:")
+		root.IterTables(ctx, func(name string, table *Table, sch schema.Schema) (stop bool, err error) {
+			buf.WriteString("\nName:")
+			buf.WriteString(name)
+			buf.WriteString("\n")
+
+			buf.WriteString("Data:\n")
+			data, err := table.GetRowData(ctx)
+			if err != nil {
+				panic(err)
+			}
+
+			err = types.WriteEncodedValue(ctx, &buf, data)
+			if err != nil {
+				panic(err)
+			}
+			return false, nil
+		})
+	}
+
+	return buf.String()
+}

go/libraries/doltcore/ref/workingset_ref.go

@@ -54,6 +54,10 @@ func (r WorkingSetRef) GetPath() string {
 	return r.name
 }
 
+func (r WorkingSetRef) ToHeadRef() (DoltRef, error) {
+	return Parse(r.GetPath())
+}
+
 // String returns the fully qualified reference name e.g.
 // refs/workingSets/my-branch
 func (r WorkingSetRef) String() string {

go/libraries/doltcore/sqle/database.go

@@ -30,6 +30,7 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
 	"github.com/dolthub/dolt/go/libraries/doltcore/env"
 	"github.com/dolthub/dolt/go/libraries/doltcore/env/actions/commitwalk"
+	"github.com/dolthub/dolt/go/libraries/doltcore/ref"
 	"github.com/dolthub/dolt/go/libraries/doltcore/row"
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema/alterschema"
@@ -40,17 +41,10 @@ import (
 	"github.com/dolthub/dolt/go/store/types"
 )
 
-type commitBehavior int8
-
 var ErrInvalidTableName = errors.NewKind("Invalid table name %s. Table names must match the regular expression " + doltdb.TableNameRegexStr)
 var ErrReservedTableName = errors.NewKind("Invalid table name %s. Table names beginning with `dolt_` are reserved for internal use")
 var ErrSystemTableAlter = errors.NewKind("Cannot alter table %s: system tables cannot be dropped or altered")
 
-const (
-	batched commitBehavior = iota
-	single
-)
-
 type SqlDatabase interface {
 	sql.Database
 	GetRoot(*sql.Context) (*doltdb.RootValue, error)
@@ -59,12 +53,11 @@ type SqlDatabase interface {
 
 // Database implements sql.Database for a dolt DB.
 type Database struct {
-	name      string
-	ddb       *doltdb.DoltDB
-	rsr       env.RepoStateReader
-	rsw       env.RepoStateWriter
-	drw       env.DocsReadWriter
-	batchMode commitBehavior
+	name string
+	ddb  *doltdb.DoltDB
+	rsr  env.RepoStateReader
+	rsw  env.RepoStateWriter
+	drw  env.DocsReadWriter
 }
 
 var _ sql.Database = (*Database)(nil)
@@ -80,7 +73,7 @@ func (d DisabledTransaction) String() string {
 }
 
 func (db Database) StartTransaction(ctx *sql.Context) (sql.Transaction, error) {
-	if !TransactionsEnabled {
+	if !TransactionsEnabled(ctx) {
 		return DisabledTransaction{}, nil
 	}
 
@@ -105,7 +98,7 @@ func (db Database) StartTransaction(ctx *sql.Context) (sql.Transaction, error) {
 		return nil, err
 	}
 
-	err = db.setHeadHash(ctx)
+	err = db.setHeadHash(ctx, wsRef)
 	if err != nil {
 		return nil, err
 	}
@@ -113,7 +106,13 @@ func (db Database) StartTransaction(ctx *sql.Context) (sql.Transaction, error) {
 	return NewDoltTransaction(root, wsRef, db.DbData()), nil
 }
 
-func (db Database) setHeadHash(ctx *sql.Context) error {
+func (db Database) setHeadHash(ctx *sql.Context, ref ref.WorkingSetRef) error {
+	// TODO: use the session HEAD ref here instead of the repo state one
+	// headRef, err := ref.ToHeadRef()
+	// if err != nil {
+	// 	return err
+	// }
+
 	headCommit, err := db.ddb.Resolve(ctx, db.rsr.CWBHeadSpec(), db.rsr.CWBHeadRef())
 	if err != nil {
 		return err
@@ -167,25 +166,11 @@ var _ sql.TransactionDatabase = Database{}
 // NewDatabase returns a new dolt database to use in queries.
 func NewDatabase(name string, dbData env.DbData) Database {
 	return Database{
-		name:      name,
-		ddb:       dbData.Ddb,
-		rsr:       dbData.Rsr,
-		rsw:       dbData.Rsw,
-		drw:       dbData.Drw,
-		batchMode: single,
-	}
-}
-
-// NewBatchedDatabase returns a new dolt database executing in batch insert mode. Integrators must call Flush() to
-// commit any outstanding edits.
-func NewBatchedDatabase(name string, dbData env.DbData) Database {
-	return Database{
-		name:      name,
-		ddb:       dbData.Ddb,
-		rsr:       dbData.Rsr,
-		rsw:       dbData.Rsw,
-		drw:       dbData.Drw,
-		batchMode: batched,
+		name: name,
+		ddb:  dbData.Ddb,
+		rsr:  dbData.Rsr,
+		rsw:  dbData.Rsw,
+		drw:  dbData.Drw,
 	}
 }
 
@@ -307,6 +292,7 @@ func (db Database) GetTableInsensitiveWithRoot(ctx *sql.Context, root *doltdb.Ro
 // GetTableInsensitiveAsOf implements sql.VersionedDatabase
 func (db Database) GetTableInsensitiveAsOf(ctx *sql.Context, tableName string, asOf interface{}) (sql.Table, bool, error) {
 	root, err := db.rootAsOf(ctx, asOf)
+
 	if err != nil {
 		return nil, false, err
 	} else if root == nil {

go/libraries/doltcore/sqle/dfunctions/commit.go

@@ -47,7 +47,11 @@ func (cf *CommitFunc) Eval(ctx *sql.Context, row sql.Row) (interface{}, error) {
 	if err != nil {
 		return nil, err
 	}
-	apr := cli.ParseArgsOrDie(ap, args, nil)
+
+	apr, err := ap.Parse(args)
+	if err != nil {
+		return nil, err
+	}
 
 	var name, email string
 	if authorStr, ok := apr.GetValue(cli.AuthorParam); ok {
@@ -95,25 +99,21 @@ func (cf *CommitFunc) Eval(ctx *sql.Context, row sql.Row) (interface{}, error) {
 	}
 
 	h, err := ddb.WriteRootValue(ctx, root)
-
 	if err != nil {
 		return nil, err
 	}
 
 	meta, err := doltdb.NewCommitMeta(name, email, commitMessage)
-
 	if err != nil {
 		return nil, err
 	}
 
 	cm, err := ddb.CommitDanglingWithParentCommits(ctx, h, []*doltdb.Commit{parent}, meta)
-
 	if err != nil {
 		return nil, err
 	}
 
 	h, err = cm.HashOf()
-
 	if err != nil {
 		return nil, err
 	}

go/libraries/doltcore/sqle/dfunctions/dolt_add.go

@@ -52,7 +52,7 @@ func (d DoltAddFunc) Eval(ctx *sql.Context, row sql.Row) (interface{}, error) {
 		return 1, err
 	}
 
-	apr, err := cli.ParseArgs(ap, args, nil)
+	apr, err := ap.Parse(args)
 	if err != nil {
 		return 1, err
 	}

go/libraries/doltcore/sqle/dfunctions/dolt_checkout.go

@@ -58,7 +58,7 @@ func (d DoltCheckoutFunc) Eval(ctx *sql.Context, row sql.Row) (interface{}, erro
 		return 1, err
 	}
 
-	apr, err := cli.ParseArgs(ap, args, nil)
+	apr, err := ap.Parse(args)
 	if err != nil {
 		return 1, err
 	}

go/libraries/doltcore/sqle/dfunctions/dolt_commit.go

@@ -62,7 +62,7 @@ func (d DoltCommitFunc) Eval(ctx *sql.Context, row sql.Row) (interface{}, error)
 		return nil, err
 	}
 
-	apr, err := cli.ParseArgs(ap, args, nil)
+	apr, err := ap.Parse(args)
 	if err != nil {
 		return nil, err
 	}

go/libraries/doltcore/sqle/dfunctions/dolt_merge.go

@@ -58,7 +58,7 @@ func (d DoltMergeFunc) Eval(ctx *sql.Context, row sql.Row) (interface{}, error)
 		return nil, err
 	}
 
-	apr, err := cli.ParseArgs(ap, args, nil)
+	apr, err := ap.Parse(args)
 	if err != nil {
 		return nil, err
 	}
@@ -100,11 +100,11 @@ func (d DoltMergeFunc) Eval(ctx *sql.Context, row sql.Row) (interface{}, error)
 	}
 
 	if hasConflicts {
-		return 1, errors.New("error: merge has unresolved conflicts")
+		return 1, doltdb.ErrUnresolvedConflicts
 	}
 
 	if dbData.Rsr.IsMergeActive() {
-		return 1, errors.New("error: merging is not possible because you have not committed an active merge")
+		return 1, doltdb.ErrMergeActive
 	}
 
 	head, hh, headRoot, err := getHead(ctx, sess, dbName)
@@ -333,7 +333,7 @@ func mergeRootToWorking(
 	hasConflicts := checkForConflicts(mergeStats)
 
 	if hasConflicts {
-		return errors.New("merge has conflicts. use the dolt_conflicts table to resolve")
+		return doltdb.ErrUnresolvedConflicts
 	}
 
 	_, err = env.UpdateStagedRoot(ctx, dbData.Ddb, dbData.Rsw, workingRoot)

go/libraries/doltcore/sqle/dfunctions/dolt_reset.go

@@ -53,7 +53,7 @@ func (d DoltResetFunc) Eval(ctx *sql.Context, row sql.Row) (interface{}, error)
 		return 1, err
 	}
 
-	apr, err := cli.ParseArgs(ap, args, nil)
+	apr, err := ap.Parse(args)
 	if err != nil {
 		return 1, err
 	}

go/libraries/doltcore/sqle/dfunctions/merge.go

@@ -51,7 +51,7 @@ func (cf *MergeFunc) Eval(ctx *sql.Context, row sql.Row) (interface{}, error) {
 		return nil, err
 	}
 
-	apr, err := cli.ParseArgs(ap, args, nil)
+	apr, err := ap.Parse(args)
 	if err != nil {
 		return nil, err
 	}

go/libraries/doltcore/sqle/dolt_session.go

@@ -46,15 +46,21 @@ const (
 	DoltCommitOnTransactionCommit = "dolt_transaction_commit"
 )
 
-// TransactionsEnabled controls whether to use SQL transactions
-// Exported only for testing
-var TransactionsEnabled = false
+type batchMode int8
+
+const (
+	single batchMode = iota
+	batched
+)
+
+const TransactionsEnabledSysVar = "dolt_transactions_enabled"
 
 func init() {
+	txEnabledSessionVar := int8(0)
 	enableTx, ok := os.LookupEnv(EnableTransactionsEnvKey)
 	if ok {
 		if strings.ToLower(enableTx) == "true" {
-			TransactionsEnabled = true
+			txEnabledSessionVar = int8(1)
 		}
 	}
 	sql.SystemVariables.AddSystemVariables([]sql.SystemVariable{
@@ -66,9 +72,33 @@ func init() {
 			Type:              sql.NewSystemBoolType(DoltCommitOnTransactionCommit),
 			Default:           int8(0),
 		},
+		{
+			Name:              TransactionsEnabledSysVar,
+			Scope:             sql.SystemVariableScope_Session,
+			Dynamic:           true,
+			SetVarHintApplies: false,
+			Type:              sql.NewSystemBoolType(TransactionsEnabledSysVar),
+			Default:           txEnabledSessionVar,
+		},
 	})
 }
 
+func TransactionsEnabled(ctx *sql.Context) bool {
+	enabled, err := ctx.GetSessionVariable(ctx, TransactionsEnabledSysVar)
+	if err != nil {
+		panic(err)
+	}
+
+	switch enabled.(int8) {
+	case 0:
+		return false
+	case 1:
+		return true
+	default:
+		panic(fmt.Sprintf("Unexpected value %v", enabled))
+	}
+}
+
 func IsHeadKey(key string) (bool, string) {
 	if strings.HasSuffix(key, HeadKeySuffix) {
 		return true, key[:len(key)-len(HeadKeySuffix)]
@@ -93,6 +123,7 @@ type DoltSession struct {
 	dbDatas               map[string]env.DbData
 	editSessions          map[string]*editor.TableEditSession
 	dirty                 map[string]bool
+	batchMode             batchMode
 	Username              string
 	Email                 string
 	tempTableRoots        map[string]*doltdb.RootValue
@@ -151,13 +182,40 @@ func NewDoltSession(ctx *sql.Context, sqlSess sql.Session, username, email strin
 	return sess, nil
 }
 
+// EnableBatchedMode enables batched mode for this session. This is only safe to do during initialization.
+// Sessions operating in batched mode don't flush any edit buffers except when told to do so explicitly, or when a
+// transaction commits. Disable @@autocommit to prevent edit buffers from being flushed prematurely in this mode.
+func (sess *DoltSession) EnableBatchedMode() {
+	sess.batchMode = batched
+}
+
 // DSessFromSess retrieves a dolt session from a standard sql.Session
 func DSessFromSess(sess sql.Session) *DoltSession {
 	return sess.(*DoltSession)
 }
 
+// Flush flushes all changes sitting in edit sessions to the session root for the database named. This normally
+// happens automatically as part of statement execution, and is only necessary when the session is manually batched (as
+// for bulk SQL import)
+func (sess *DoltSession) Flush(ctx *sql.Context, dbName string) error {
+	editSession := sess.editSessions[dbName]
+	newRoot, err := editSession.Flush(ctx)
+	if err != nil {
+		return err
+	}
+
+	return sess.SetRoot(ctx, dbName, newRoot)
+}
+
 // CommitTransaction commits the in-progress transaction for the database named
 func (sess *DoltSession) CommitTransaction(ctx *sql.Context, dbName string, tx sql.Transaction) error {
+	if sess.batchMode == batched {
+		err := sess.Flush(ctx, dbName)
+		if err != nil {
+			return err
+		}
+	}
+
 	if !sess.dirty[dbName] {
 		return nil
 	}
@@ -177,7 +235,7 @@ func (sess *DoltSession) CommitTransaction(ctx *sql.Context, dbName string, tx s
 
 	// Old "commit" path, which just writes whatever the root for this session is to the repo state file with no care
 	// for concurrency. Over time we will disable this path.
-	if !TransactionsEnabled {
+	if !TransactionsEnabled(ctx) {
 		dbData := sess.dbDatas[dbName]
 
 		h, err := dbData.Ddb.WriteRootValue(ctx, dbRoot.root)
@@ -261,7 +319,7 @@ func (sess *DoltSession) CommitWorkingSetToDolt(ctx *sql.Context, dbData env.DbD
 
 // RollbackTransaction rolls the given transaction back
 func (sess *DoltSession) RollbackTransaction(ctx *sql.Context, dbName string, tx sql.Transaction) error {
-	if !TransactionsEnabled || dbName == "" {
+	if !TransactionsEnabled(ctx) || dbName == "" {
 		return nil
 	}
 
@@ -286,7 +344,7 @@ func (sess *DoltSession) RollbackTransaction(ctx *sql.Context, dbName string, tx
 // CreateSavepoint creates a new savepoint for this transaction with the name given. A previously created savepoint
 // with the same name will be overwritten.
 func (sess *DoltSession) CreateSavepoint(ctx *sql.Context, savepointName, dbName string, tx sql.Transaction) error {
-	if !TransactionsEnabled || dbName == "" {
+	if !TransactionsEnabled(ctx) || dbName == "" {
 		return nil
 	}
 
@@ -302,7 +360,7 @@ func (sess *DoltSession) CreateSavepoint(ctx *sql.Context, savepointName, dbName
 // RollbackToSavepoint sets this session's root to the one saved in the savepoint name. It's an error if no savepoint
 // with that name exists.
 func (sess *DoltSession) RollbackToSavepoint(ctx *sql.Context, savepointName, dbName string, tx sql.Transaction) error {
-	if !TransactionsEnabled || dbName == "" {
+	if !TransactionsEnabled(ctx) || dbName == "" {
 		return nil
 	}
 
@@ -327,7 +385,7 @@ func (sess *DoltSession) RollbackToSavepoint(ctx *sql.Context, savepointName, db
 // ReleaseSavepoint removes the savepoint name from the transaction. It's an error if no savepoint with that name
 // exists.
 func (sess *DoltSession) ReleaseSavepoint(ctx *sql.Context, savepointName, dbName string, tx sql.Transaction) error {
-	if !TransactionsEnabled || dbName == "" {
+	if !TransactionsEnabled(ctx) || dbName == "" {
 		return nil
 	}
 
@@ -667,7 +725,7 @@ func (sess *DoltSession) AddDB(ctx *sql.Context, db sql.Database, dbData env.DbD
 		}
 	}
 
-	if TransactionsEnabled {
+	if TransactionsEnabled(ctx) {
 		// Not all dolt commands update the working set ref yet. So until that's true, we update it here with the contents
 		// of the repo_state.json file
 		workingSetRef, err := ref.WorkingSetRefForHead(headRef)

go/libraries/doltcore/sqle/dtables/query_catalog_table_test.go

@@ -56,6 +56,9 @@ func TestInsertIntoQueryCatalogTable(t *testing.T) {
 	require.NoError(t, err)
 	require.True(t, ok)
 
+	err = dEnv.UpdateWorkingRoot(ctx, root)
+	require.NoError(t, err)
+
 	rows, err := sqle.ExecuteSelect(dEnv, dEnv.DoltDB, root, "select display_order, query, name, description from "+doltdb.DoltQueryCatalogTableName)
 	require.NoError(t, err)
 	expectedRows := []sql.Row{
@@ -76,6 +79,9 @@ func TestInsertIntoQueryCatalogTable(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, sq2, retrieved2)
 
+	err = dEnv.UpdateWorkingRoot(ctx, root)
+	require.NoError(t, err)
+
 	rows, err = sqle.ExecuteSelect(dEnv, dEnv.DoltDB, root, "select display_order, query, name, description from "+doltdb.DoltQueryCatalogTableName+" order by display_order")
 	require.NoError(t, err)
 	expectedRows = []sql.Row{

go/libraries/doltcore/sqle/enginetest/dolt_engine_test.go

@@ -19,7 +19,6 @@ import (
 
 	"github.com/dolthub/go-mysql-server/enginetest"
 	"github.com/dolthub/go-mysql-server/sql"
-	"github.com/stretchr/testify/assert"
 
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle"
 )
@@ -29,7 +28,12 @@ func init() {
 }
 
 func TestQueries(t *testing.T) {
-	enginetest.TestQueries(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestQueries(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestQueries(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestSingleQuery(t *testing.T) {
@@ -89,7 +93,12 @@ func TestSingleScript(t *testing.T) {
 }
 
 func TestVersionedQueries(t *testing.T) {
-	enginetest.TestVersionedQueries(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestVersionedQueries(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestVersionedQueries(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 // Tests of choosing the correct execution plan independent of result correctness. Mostly useful for confirming that
@@ -113,36 +122,75 @@ func TestQueryPlans(t *testing.T) {
 }
 
 func TestQueryErrors(t *testing.T) {
-	enginetest.TestQueryErrors(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestQueryErrors(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestQueryErrors(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestInfoSchema(t *testing.T) {
-	enginetest.TestInfoSchema(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestInfoSchema(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestInfoSchema(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestColumnAliases(t *testing.T) {
-	assert.Equal(t, 1, 1, "msg", 1, 2)
-	enginetest.TestColumnAliases(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestColumnAliases(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestColumnAliases(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestOrderByGroupBy(t *testing.T) {
-	enginetest.TestOrderByGroupBy(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestOrderByGroupBy(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestOrderByGroupBy(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestAmbiguousColumnResolution(t *testing.T) {
-	enginetest.TestAmbiguousColumnResolution(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestAmbiguousColumnResolution(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestAmbiguousColumnResolution(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestInsertInto(t *testing.T) {
-	enginetest.TestInsertInto(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestInsertInto(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestInsertInto(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestInsertIgnoreInto(t *testing.T) {
-	enginetest.TestInsertIgnoreInto(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestInsertIgnoreInto(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestInsertIgnoreInto(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestInsertIntoErrors(t *testing.T) {
-	enginetest.TestInsertIntoErrors(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestInsertIntoErrors(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestInsertIntoErrors(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestReplaceInto(t *testing.T) {
@@ -151,87 +199,192 @@ func TestReplaceInto(t *testing.T) {
 }
 
 func TestReplaceIntoErrors(t *testing.T) {
-	enginetest.TestReplaceIntoErrors(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestReplaceIntoErrors(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestReplaceIntoErrors(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestUpdate(t *testing.T) {
-	enginetest.TestUpdate(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestUpdate(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestUpdate(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestUpdateErrors(t *testing.T) {
-	enginetest.TestUpdateErrors(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestUpdateErrors(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestUpdateErrors(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestDeleteFrom(t *testing.T) {
-	enginetest.TestDelete(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestDelete(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestDelete(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestDeleteFromErrors(t *testing.T) {
-	enginetest.TestDeleteErrors(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestDeleteErrors(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestDeleteErrors(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestTruncate(t *testing.T) {
-	enginetest.TestTruncate(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestTruncate(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestTruncate(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestScripts(t *testing.T) {
-	enginetest.TestScripts(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestScripts(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestScripts(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestCreateTable(t *testing.T) {
-	enginetest.TestCreateTable(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestCreateTable(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestCreateTable(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestDropTable(t *testing.T) {
-	enginetest.TestDropTable(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestDropTable(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestDropTable(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestRenameTable(t *testing.T) {
-	enginetest.TestRenameTable(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestRenameTable(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestRenameTable(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestRenameColumn(t *testing.T) {
-	enginetest.TestRenameColumn(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestRenameColumn(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestRenameColumn(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestAddColumn(t *testing.T) {
-	enginetest.TestAddColumn(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestAddColumn(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestAddColumn(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestModifyColumn(t *testing.T) {
-	enginetest.TestModifyColumn(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestModifyColumn(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestModifyColumn(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestDropColumn(t *testing.T) {
-	enginetest.TestDropColumn(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestDropColumn(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestDropColumn(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestCreateForeignKeys(t *testing.T) {
-	enginetest.TestCreateForeignKeys(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestCreateForeignKeys(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestCreateForeignKeys(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestDropForeignKeys(t *testing.T) {
-	enginetest.TestDropForeignKeys(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestDropForeignKeys(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestDropForeignKeys(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestCreateCheckConstraints(t *testing.T) {
-	enginetest.TestCreateCheckConstraints(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestCreateCheckConstraints(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestCreateCheckConstraints(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestChecksOnInsert(t *testing.T) {
-	enginetest.TestChecksOnInsert(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestChecksOnInsert(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestChecksOnInsert(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestChecksOnUpdate(t *testing.T) {
-	enginetest.TestChecksOnUpdate(t, enginetest.NewDefaultMemoryHarness())
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestChecksOnUpdate(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestChecksOnUpdate(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
-func TestTestDisallowedCheckConstraints(t *testing.T) {
-	enginetest.TestDisallowedCheckConstraints(t, newDoltHarness(t))
+func TestDisallowedCheckConstraints(t *testing.T) {
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestDisallowedCheckConstraints(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestDisallowedCheckConstraints(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestDropCheckConstraints(t *testing.T) {
-	enginetest.TestDropCheckConstraints(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestDropCheckConstraints(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestDropCheckConstraints(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestExplode(t *testing.T) {
@@ -240,51 +393,108 @@ func TestExplode(t *testing.T) {
 }
 
 func TestReadOnly(t *testing.T) {
-	enginetest.TestReadOnly(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestReadOnly(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestReadOnly(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestViews(t *testing.T) {
-	enginetest.TestViews(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestViews(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestViews(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestVersionedViews(t *testing.T) {
-	enginetest.TestVersionedViews(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestVersionedViews(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestVersionedViews(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestNaturalJoin(t *testing.T) {
-	enginetest.TestNaturalJoin(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestNaturalJoin(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestNaturalJoin(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestNaturalJoinEqual(t *testing.T) {
-	enginetest.TestNaturalJoinEqual(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestNaturalJoinEqual(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestNaturalJoinEqual(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestNaturalJoinDisjoint(t *testing.T) {
-	enginetest.TestNaturalJoinDisjoint(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestNaturalJoinEqual(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestNaturalJoinEqual(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestInnerNestedInNaturalJoins(t *testing.T) {
-	enginetest.TestInnerNestedInNaturalJoins(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestInnerNestedInNaturalJoins(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestInnerNestedInNaturalJoins(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestColumnDefaults(t *testing.T) {
-	enginetest.TestColumnDefaults(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestColumnDefaults(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestColumnDefaults(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestVariables(t *testing.T) {
+	// Can't run these tests more than once because they set and make assertions about global vars, which obviously
+	// persist outside sessions.
 	enginetest.TestVariables(t, newDoltHarness(t))
 }
 
 func TestVariableErrors(t *testing.T) {
-	enginetest.TestVariableErrors(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestVariableErrors(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestVariableErrors(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestJsonScripts(t *testing.T) {
-	enginetest.TestJsonScripts(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestJsonScripts(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestJsonScripts(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestTriggers(t *testing.T) {
-	enginetest.TestTriggers(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestTriggers(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestTriggers(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestStoredProcedures(t *testing.T) {
@@ -297,19 +507,17 @@ func TestStoredProcedures(t *testing.T) {
 	}
 	enginetest.ProcedureLogicTests = tests
 
-	enginetest.TestStoredProcedures(t, newDoltHarness(t))
+	t.Run("no transactions", func(t *testing.T) {
+		enginetest.TestStoredProcedures(t, newDoltHarness(t))
+	})
+	t.Run("with transactions", func(t *testing.T) {
+		enginetest.TestStoredProcedures(t, newDoltHarness(t).withTransactionsEnabled(true))
+	})
 }
 
 func TestTransactions(t *testing.T) {
-	if !sqle.TransactionsEnabled {
-		sqle.TransactionsEnabled = true
-		defer func() {
-			sqle.TransactionsEnabled = false
-		}()
-	}
-	enginetest.TestTransactionScripts(t, newDoltHarness(t))
-
+	enginetest.TestTransactionScripts(t, newDoltHarness(t).withTransactionsEnabled(true))
 	for _, script := range DoltTransactionTests {
-		enginetest.TestTransactionScript(t, newDoltHarness(t), script)
+		enginetest.TestTransactionScript(t, newDoltHarness(t).withTransactionsEnabled(true), script)
 	}
 }

go/libraries/doltcore/sqle/enginetest/dolt_harness.go

@@ -30,11 +30,12 @@ import (
 )
 
 type DoltHarness struct {
-	t              *testing.T
-	session        *sqle.DoltSession
-	databases      []sqle.Database
-	parallelism    int
-	skippedQueries []string
+	t                   *testing.T
+	session             *sqle.DoltSession
+	transactionsEnabled bool
+	databases           []sqle.Database
+	parallelism         int
+	skippedQueries      []string
 }
 
 var _ enginetest.Harness = (*DoltHarness)(nil)
@@ -54,6 +55,18 @@ func newDoltHarness(t *testing.T) *DoltHarness {
 	}
 }
 
+// withTransactionsEnabled returns a copy of this harness with transactions enabled or not for all sessions
+func (d DoltHarness) withTransactionsEnabled(enabled bool) *DoltHarness {
+	d.transactionsEnabled = enabled
+	d.setTransactionSessionVar(d.session, enabled)
+	return &d
+}
+
+func (d DoltHarness) setTransactionSessionVar(session *sqle.DoltSession, enabled bool) {
+	err := session.SetSessionVariable(sql.NewEmptyContext(), sqle.TransactionsEnabledSysVar, enabled)
+	require.NoError(d.t, err)
+}
+
 var defaultSkippedQueries = []string{
 	"show variables",           // we set extra variables
 	"show create table fk_tbl", // we create an extra key for the FK that vanilla gms does not
@@ -114,6 +127,8 @@ func (d DoltHarness) NewSession() *sql.Context {
 	session, err := sqle.NewDoltSession(sql.NewEmptyContext(), enginetest.NewBaseSession(), "test", "email@test.com")
 	require.NoError(d.t, err)
 
+	d.setTransactionSessionVar(session, d.transactionsEnabled)
+
 	ctx := sql.NewContext(
 		context.Background(),
 		sql.WithSession(session))
@@ -153,6 +168,8 @@ func (d *DoltHarness) NewDatabases(names ...string) []sql.Database {
 	d.session, err = sqle.NewDoltSession(sql.NewEmptyContext(), enginetest.NewBaseSession(), "test", "email@test.com")
 	require.NoError(d.t, err)
 
+	d.setTransactionSessionVar(d.session, d.transactionsEnabled)
+
 	var dbs []sql.Database
 	d.databases = nil
 	for _, name := range names {
@@ -207,6 +224,7 @@ func (d *DoltHarness) SnapshotTable(db sql.VersionedDatabase, name string, asOf
 
 	asOfString, ok := asOf.(string)
 	require.True(d.t, ok)
+
 	ctx := enginetest.NewContext(d)
 	_, iter, err := e.Query(ctx,
 		"set @@"+sqle.HeadKey(ddb.Name())+" = COMMIT('-m', 'test commit');")
@@ -214,9 +232,17 @@ func (d *DoltHarness) SnapshotTable(db sql.VersionedDatabase, name string, asOf
 	_, err = sql.RowIterToRows(ctx, iter)
 	require.NoError(d.t, err)
 
+	headHash, err := ctx.GetSessionVariable(ctx, sqle.HeadKey(ddb.Name()))
+	require.NoError(d.t, err)
+
 	ctx = enginetest.NewContext(d)
+	// TODO: there's a bug in test setup with transactions, where the HEAD session var gets overwritten on transaction
+	//  start, so we quote it here instead
+	// query := "insert into dolt_branches (name, hash) values ('" + asOfString + "', @@" + sqle.HeadKey(ddb.Name()) + ")"
+	query := "insert into dolt_branches (name, hash) values ('" + asOfString + "', '" + headHash.(string) + "')"
+
 	_, iter, err = e.Query(ctx,
-		"insert into dolt_branches (name, hash) values ('"+asOfString+"', @@"+sqle.HeadKey(ddb.Name())+")")
+		query)
 	require.NoError(d.t, err)
 	_, err = sql.RowIterToRows(ctx, iter)
 	require.NoError(d.t, err)

go/libraries/doltcore/sqle/enginetest/dolt_transaction_commit_test.go

@@ -23,19 +23,12 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
-	"github.com/dolthub/dolt/go/libraries/doltcore/sqle"
 )
 
 func TestDoltTransactionCommitOneClient(t *testing.T) {
 	// In this test, we're setting only one client to match transaction commits to dolt commits.
 	// Autocommit is disabled for the enabled client, as it's the recommended way to use this feature.
-	ote := sqle.TransactionsEnabled
-	sqle.TransactionsEnabled = true
-	defer func() {
-		sqle.TransactionsEnabled = ote
-	}()
-
-	harness := newDoltHarness(t)
+	harness := newDoltHarness(t).withTransactionsEnabled(true)
 	enginetest.TestTransactionScript(t, harness, enginetest.TransactionTest{
 		Name: "dolt commit after transaction commit one client",
 		SetUpScript: []string{
@@ -164,13 +157,7 @@ func TestDoltTransactionCommitOneClient(t *testing.T) {
 func TestDoltTransactionCommitTwoClients(t *testing.T) {
 	// In this test, we're setting both clients to match transaction commits to dolt commits.
 	// Autocommit is disabled, as it's the recommended way to use this feature.
-	ote := sqle.TransactionsEnabled
-	sqle.TransactionsEnabled = true
-	defer func() {
-		sqle.TransactionsEnabled = ote
-	}()
-
-	harness := newDoltHarness(t)
+	harness := newDoltHarness(t).withTransactionsEnabled(true)
 	enginetest.TestTransactionScript(t, harness, enginetest.TransactionTest{
 		Name: "dolt commit after transaction commit two clients",
 		SetUpScript: []string{
@@ -293,13 +280,7 @@ func TestDoltTransactionCommitTwoClients(t *testing.T) {
 func TestDoltTransactionCommitAutocommit(t *testing.T) {
 	// In this test, each insertion from both clients cause a commit as autocommit is enabled.
 	// Not the recommended way to use the feature, but it's permitted.
-	ote := sqle.TransactionsEnabled
-	sqle.TransactionsEnabled = true
-	defer func() {
-		sqle.TransactionsEnabled = ote
-	}()
-
-	harness := newDoltHarness(t)
+	harness := newDoltHarness(t).withTransactionsEnabled(true)
 	enginetest.TestTransactionScript(t, harness, enginetest.TransactionTest{
 		Name: "dolt commit after transaction commit autocommit",
 		SetUpScript: []string{

go/libraries/doltcore/sqle/logictest/dolt/doltharness.go

@@ -21,6 +21,7 @@ import (
 	"math/rand"
 	"strconv"
 	"strings"
+	"time"
 
 	sqle "github.com/dolthub/go-mysql-server"
 	"github.com/dolthub/go-mysql-server/sql"
@@ -34,10 +35,16 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/env"
 	dsql "github.com/dolthub/dolt/go/libraries/doltcore/sqle"
 	"github.com/dolthub/dolt/go/libraries/utils/filesys"
+	"github.com/dolthub/dolt/go/store/types"
 )
 
 var _ logictest.Harness = &DoltHarness{}
 
+const (
+	name  = "sqllogictest runner"
+	email = "sqllogictestrunner@dolthub.com"
+)
+
 type DoltHarness struct {
 	Version string
 	engine  *sqle.Engine
@@ -53,10 +60,6 @@ func (h *DoltHarness) EngineStr() string {
 
 func (h *DoltHarness) Init() error {
 	dEnv := env.Load(context.Background(), env.GetCurrentUserHomeDir, filesys.LocalFS, doltdb.LocalDirDoltDB, "test")
-	if !dEnv.HasDoltDir() {
-		panic("Current directory must be a valid dolt repository")
-	}
-
 	return innerInit(h, dEnv)
 }
 
@@ -116,6 +119,18 @@ func (h *DoltHarness) ExecuteQuery(statement string) (schema string, results []s
 }
 
 func innerInit(h *DoltHarness, dEnv *env.DoltEnv) error {
+	if !dEnv.HasDoltDir() {
+		err := dEnv.InitRepoWithTime(context.Background(), types.Format_Default, name, email, time.Now())
+		if err != nil {
+			return err
+		}
+	} else {
+		err := dEnv.InitDBAndRepoState(context.Background(), types.Format_Default, name, email, time.Now())
+		if err != nil {
+			return err
+		}
+	}
+
 	var err error
 	h.engine, err = sqlNewEngine(dEnv)
 

go/libraries/doltcore/sqle/logictest/dolt/doltharness_test.go

@@ -16,6 +16,7 @@ package dolt
 
 import (
 	"context"
+	"fmt"
 	"path/filepath"
 	"testing"
 
@@ -97,26 +98,30 @@ func TestDoltHarness(t *testing.T) {
 		},
 	}
 
-	t.Run("should execute simple sql queries against Dolt", func(t *testing.T) {
-		h := &DoltHarness{}
-		fs := filesys.NewInMemFS([]string{}, nil, tmp)
-		dEnv := createTestEnvWithFS(fs, wd)
+	fs := filesys.NewInMemFS([]string{}, nil, tmp)
+	dEnv := createTestEnvWithFS(fs, wd)
 
-		err := innerInit(h, dEnv)
-		assert.Equal(t, nil, err)
+	// We run this several times in a row to make sure that the same dolt env can be used in multiple setup / teardown
+	// cycles
+	for i := 0; i < 10; i++ {
+		t.Run(fmt.Sprintf("dolt harness runner %d", i), func(t *testing.T) {
+			h := &DoltHarness{}
+			err := innerInit(h, dEnv)
+			assert.Equal(t, nil, err)
 
-		// setup repo with statements
-		for _, test := range statementTests {
-			err = h.ExecuteStatement(test.statement)
-			assert.Equal(t, test.expErr, err)
-		}
+			// setup repo with statements
+			for _, test := range statementTests {
+				err = h.ExecuteStatement(test.statement)
+				assert.Equal(t, test.expErr, err)
+			}
 
-		// test queries
-		for _, test := range queryTests {
-			schema, results, err := h.ExecuteQuery(test.query)
-			assert.Equal(t, test.expErr, err)
-			assert.Equal(t, test.expSchema, schema)
-			assert.Equal(t, test.expResults, results)
-		}
-	})
+			// test queries
+			for _, test := range queryTests {
+				schema, results, err := h.ExecuteQuery(test.query)
+				assert.Equal(t, test.expErr, err)
+				assert.Equal(t, test.expSchema, schema)
+				assert.Equal(t, test.expResults, results)
+			}
+		})
+	}
 }

go/libraries/doltcore/sqle/sqlbatch_test.go

@@ -63,9 +63,10 @@ func TestSqlBatchInserts(t *testing.T) {
 	CreateTestDatabase(dEnv, t)
 	root, _ := dEnv.WorkingRoot(ctx)
 
-	db := NewBatchedDatabase("dolt", dEnv.DbData())
+	db := NewDatabase("dolt", dEnv.DbData())
 	engine, sqlCtx, err := NewTestEngine(ctx, db, root)
 	require.NoError(t, err)
+	DSessFromSess(sqlCtx.Session).EnableBatchedMode()
 
 	for _, stmt := range insertStatements {
 		_, rowIter, err := engine.Query(sqlCtx, stmt)
@@ -151,9 +152,10 @@ func TestSqlBatchInsertIgnoreReplace(t *testing.T) {
 	CreateTestDatabase(dEnv, t)
 	root, _ := dEnv.WorkingRoot(ctx)
 
-	db := NewBatchedDatabase("dolt", dEnv.DbData())
+	db := NewDatabase("dolt", dEnv.DbData())
 	engine, sqlCtx, err := NewTestEngine(ctx, db, root)
 	require.NoError(t, err)
+	DSessFromSess(sqlCtx.Session).EnableBatchedMode()
 
 	for _, stmt := range insertStatements {
 		_, rowIter, err := engine.Query(sqlCtx, stmt)
@@ -189,9 +191,10 @@ func TestSqlBatchInsertErrors(t *testing.T) {
 	CreateTestDatabase(dEnv, t)
 	root, _ := dEnv.WorkingRoot(ctx)
 
-	db := NewBatchedDatabase("dolt", dEnv.DbData())
+	db := NewDatabase("dolt", dEnv.DbData())
 	engine, sqlCtx, err := NewTestEngine(ctx, db, root)
 	require.NoError(t, err)
+	DSessFromSess(sqlCtx.Session).EnableBatchedMode()
 
 	_, rowIter, err := engine.Query(sqlCtx, `insert into people (id, first_name, last_name, is_married, age, rating, uuid, num_episodes) values
 					(0, "Maggie", "Simpson", false, 1, 5.1, '00000000-0000-0000-0000-000000000007', 677)`)

go/libraries/doltcore/sqle/table_editor.go

@@ -41,7 +41,6 @@ type sqlTableEditor struct {
 	dbName            string
 	sch               schema.Schema
 	autoIncCol        schema.Column
-	batchMode         commitBehavior
 	vrw               types.ValueReadWriter
 	autoIncrementType typeinfo.TypeInfo
 	kvToSQLRow        *KVToSqlRowConverter
@@ -69,7 +68,6 @@ func newSqlTableEditor(ctx *sql.Context, t *WritableDoltTable) (*sqlTableEditor,
 		dbName:      t.db.Name(),
 		sch:         t.sch,
 		autoIncCol:  t.autoIncCol,
-		batchMode:   t.db.batchMode,
 		vrw:         t.db.ddb.ValueReadWriter(),
 		kvToSQLRow:  conv,
 		tableEditor: tableEditor,
@@ -147,8 +145,10 @@ func (te *sqlTableEditor) SetAutoIncrementValue(ctx *sql.Context, val interface{
 
 // Close implements Closer
 func (te *sqlTableEditor) Close(ctx *sql.Context) error {
-	// If we're running in batched mode, don't flush the edits until explicitly told to do so by the parent table.
-	if te.batchMode == batched {
+	sess := DSessFromSess(ctx.Session)
+
+	// If we're running in batched mode, don't flush the edits until explicitly told to do so
+	if sess.batchMode == batched {
 		return nil
 	}
 	return te.flush(ctx)

go/libraries/doltcore/sqle/table_editor_fk_test.go

@@ -34,16 +34,13 @@ import (
 	"github.com/dolthub/dolt/go/store/types"
 )
 
-var fk_dEnv *env.DoltEnv
-var fk_initialRoot *doltdb.RootValue
-
-func init() {
-	fk_dEnv = dtestutils.CreateTestEnv()
-	root, err := fk_dEnv.WorkingRoot(context.Background())
+func setupEditorFkTest(t *testing.T) (*env.DoltEnv, *doltdb.RootValue) {
+	dEnv := dtestutils.CreateTestEnv()
+	root, err := dEnv.WorkingRoot(context.Background())
 	if err != nil {
 		panic(err)
 	}
-	fk_initialRoot, err = ExecuteSql(fk_dEnv, root, `
+	initialRoot, err := ExecuteSql(dEnv, root, `
 CREATE TABLE one (
   pk BIGINT PRIMARY KEY,
   v1 BIGINT,
@@ -74,18 +71,12 @@ CREATE TABLE child (
   v2 BIGINT
 );
 `)
-	if err != nil {
-		panic(err)
-	}
+	require.NoError(t, err)
+
+	return dEnv, initialRoot
 }
 
 func TestTableEditorForeignKeyCascade(t *testing.T) {
-	testRoot, err := ExecuteSql(fk_dEnv, fk_initialRoot, `
-ALTER TABLE two ADD FOREIGN KEY (v1) REFERENCES one(v1) ON DELETE CASCADE ON UPDATE CASCADE;
-ALTER TABLE three ADD FOREIGN KEY (v1, v2) REFERENCES two(v1, v2) ON DELETE CASCADE ON UPDATE CASCADE;
-`)
-	require.NoError(t, err)
-
 	tests := []struct {
 		name          string
 		sqlStatement  string
@@ -163,10 +154,18 @@ ALTER TABLE three ADD FOREIGN KEY (v1, v2) REFERENCES two(v1, v2) ON DELETE CASC
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
+			dEnv, initialRoot := setupEditorFkTest(t)
+
+			testRoot, err := ExecuteSql(dEnv, initialRoot, `
+ALTER TABLE two ADD FOREIGN KEY (v1) REFERENCES one(v1) ON DELETE CASCADE ON UPDATE CASCADE;
+ALTER TABLE three ADD FOREIGN KEY (v1, v2) REFERENCES two(v1, v2) ON DELETE CASCADE ON UPDATE CASCADE;
+`)
+			require.NoError(t, err)
+
 			root := testRoot
 			for _, sqlStatement := range strings.Split(test.sqlStatement, ";") {
 				var err error
-				root, err = executeModify(context.Background(), fk_dEnv, root, sqlStatement)
+				root, err = executeModify(context.Background(), dEnv, root, sqlStatement)
 				require.NoError(t, err)
 			}
 
@@ -178,10 +177,6 @@ ALTER TABLE three ADD FOREIGN KEY (v1, v2) REFERENCES two(v1, v2) ON DELETE CASC
 }
 
 func TestTableEditorForeignKeySetNull(t *testing.T) {
-	testRoot, err := ExecuteSql(fk_dEnv, fk_initialRoot, `
-ALTER TABLE two ADD FOREIGN KEY (v1) REFERENCES one(v1) ON DELETE SET NULL ON UPDATE SET NULL;`)
-	require.NoError(t, err)
-
 	tests := []struct {
 		sqlStatement string
 		expectedOne  []sql.Row
@@ -208,10 +203,16 @@ ALTER TABLE two ADD FOREIGN KEY (v1) REFERENCES one(v1) ON DELETE SET NULL ON UP
 
 	for _, test := range tests {
 		t.Run(test.sqlStatement, func(t *testing.T) {
+			dEnv, initialRoot := setupEditorFkTest(t)
+
+			testRoot, err := ExecuteSql(dEnv, initialRoot, `
+ALTER TABLE two ADD FOREIGN KEY (v1) REFERENCES one(v1) ON DELETE SET NULL ON UPDATE SET NULL;`)
+			require.NoError(t, err)
+
 			root := testRoot
 			for _, sqlStatement := range strings.Split(test.sqlStatement, ";") {
 				var err error
-				root, err = executeModify(context.Background(), fk_dEnv, root, sqlStatement)
+				root, err = executeModify(context.Background(), dEnv, root, sqlStatement)
 				require.NoError(t, err)
 			}
 
@@ -232,12 +233,6 @@ func TestTableEditorForeignKeyRestrict(t *testing.T) {
 		"",
 	} {
 		t.Run(referenceOption, func(t *testing.T) {
-			testRoot, err := ExecuteSql(fk_dEnv, fk_initialRoot, fmt.Sprintf(`
-			ALTER TABLE two ADD FOREIGN KEY (v1) REFERENCES one(v1) %s;
-			INSERT INTO one VALUES (1, 1, 1), (2, 2, 2), (3, 3, 3);
-			INSERT INTO two VALUES (1, 1, 1), (2, 2, 2), (3, 3, 3);`, referenceOption))
-			require.NoError(t, err)
-
 			tests := []struct {
 				setup       string
 				trigger     string
@@ -290,17 +285,25 @@ func TestTableEditorForeignKeyRestrict(t *testing.T) {
 
 			for _, test := range tests {
 				t.Run(test.setup+test.trigger, func(t *testing.T) {
+					dEnv, initialRoot := setupEditorFkTest(t)
+
+					testRoot, err := ExecuteSql(dEnv, initialRoot, fmt.Sprintf(`
+			ALTER TABLE two ADD FOREIGN KEY (v1) REFERENCES one(v1) %s;
+			INSERT INTO one VALUES (1, 1, 1), (2, 2, 2), (3, 3, 3);
+			INSERT INTO two VALUES (1, 1, 1), (2, 2, 2), (3, 3, 3);`, referenceOption))
+					require.NoError(t, err)
+
 					root := testRoot
 					for _, sqlStatement := range strings.Split(test.setup, ";") {
 						var err error
-						root, err = executeModify(context.Background(), fk_dEnv, root, sqlStatement)
+						root, err = executeModify(context.Background(), dEnv, root, sqlStatement)
 						require.NoError(t, err)
 					}
 					if test.expectedErr {
-						root, err = executeModify(context.Background(), fk_dEnv, root, test.trigger)
+						root, err = executeModify(context.Background(), dEnv, root, test.trigger)
 						assert.Error(t, err)
 					} else {
-						root, err = executeModify(context.Background(), fk_dEnv, root, test.trigger)
+						root, err = executeModify(context.Background(), dEnv, root, test.trigger)
 						assert.NoError(t, err)
 					}
 				})
@@ -310,12 +313,6 @@ func TestTableEditorForeignKeyRestrict(t *testing.T) {
 }
 
 func TestTableEditorForeignKeyViolations(t *testing.T) {
-	testRoot, err := ExecuteSql(fk_dEnv, fk_initialRoot, `
-ALTER TABLE two ADD FOREIGN KEY (v1) REFERENCES one(v1) ON DELETE CASCADE ON UPDATE CASCADE;
-ALTER TABLE three ADD FOREIGN KEY (v1, v2) REFERENCES two(v1, v2) ON DELETE CASCADE ON UPDATE CASCADE;
-`)
-	require.NoError(t, err)
-
 	tests := []struct {
 		setup   string
 		trigger string
@@ -359,21 +356,31 @@ ALTER TABLE three ADD FOREIGN KEY (v1, v2) REFERENCES two(v1, v2) ON DELETE CASC
 
 	for _, test := range tests {
 		t.Run(test.setup+test.trigger, func(t *testing.T) {
+			dEnv, initialRoot := setupEditorFkTest(t)
+
+			testRoot, err := ExecuteSql(dEnv, initialRoot, `
+ALTER TABLE two ADD FOREIGN KEY (v1) REFERENCES one(v1) ON DELETE CASCADE ON UPDATE CASCADE;
+ALTER TABLE three ADD FOREIGN KEY (v1, v2) REFERENCES two(v1, v2) ON DELETE CASCADE ON UPDATE CASCADE;
+`)
+			require.NoError(t, err)
+
 			root := testRoot
 			for _, sqlStatement := range strings.Split(test.setup, ";") {
 				var err error
-				root, err = executeModify(context.Background(), fk_dEnv, root, sqlStatement)
+				root, err = executeModify(context.Background(), dEnv, root, sqlStatement)
 				require.NoError(t, err)
 			}
-			root, err = executeModify(context.Background(), fk_dEnv, root, test.trigger)
+			root, err = executeModify(context.Background(), dEnv, root, test.trigger)
 			assert.Error(t, err)
 		})
 	}
 }
 
 func TestTableEditorSelfReferentialForeignKeyRestrict(t *testing.T) {
+	dEnv, initialRoot := setupEditorFkTest(t)
+
 	ctx := context.Background()
-	root := fk_initialRoot
+	root := initialRoot
 
 	sequentialTests := []struct {
 		statement   string
@@ -428,7 +435,7 @@ func TestTableEditorSelfReferentialForeignKeyRestrict(t *testing.T) {
 	}
 
 	for _, test := range sequentialTests {
-		newRoot, err := executeModify(ctx, fk_dEnv, root, test.statement)
+		newRoot, err := executeModify(ctx, dEnv, root, test.statement)
 		if test.expectedErr {
 			require.Error(t, err)
 			continue
@@ -440,8 +447,10 @@ func TestTableEditorSelfReferentialForeignKeyRestrict(t *testing.T) {
 }
 
 func TestTableEditorSelfReferentialForeignKeyCascade(t *testing.T) {
+	dEnv, initialRoot := setupEditorFkTest(t)
+
 	ctx := context.Background()
-	root := fk_initialRoot
+	root := initialRoot
 
 	sequentialTests := []struct {
 		statement   string
@@ -526,7 +535,7 @@ func TestTableEditorSelfReferentialForeignKeyCascade(t *testing.T) {
 	}
 
 	for _, test := range sequentialTests {
-		newRoot, err := executeModify(ctx, fk_dEnv, root, test.statement)
+		newRoot, err := executeModify(ctx, dEnv, root, test.statement)
 		if test.expectedErr {
 			require.Error(t, err)
 			continue
@@ -538,8 +547,10 @@ func TestTableEditorSelfReferentialForeignKeyCascade(t *testing.T) {
 }
 
 func TestTableEditorSelfReferentialForeignKeySetNull(t *testing.T) {
+	dEnv, initialRoot := setupEditorFkTest(t)
+
 	ctx := context.Background()
-	root := fk_initialRoot
+	root := initialRoot
 
 	sequentialTests := []struct {
 		statement   string
@@ -624,7 +635,7 @@ func TestTableEditorSelfReferentialForeignKeySetNull(t *testing.T) {
 	}
 
 	for _, test := range sequentialTests {
-		newRoot, err := executeModify(ctx, fk_dEnv, root, test.statement)
+		newRoot, err := executeModify(ctx, dEnv, root, test.statement)
 		if test.expectedErr {
 			require.Error(t, err)
 			continue

go/libraries/doltcore/sqle/table_editor_index_test.go

@@ -31,16 +31,12 @@ import (
 	"github.com/dolthub/dolt/go/store/types"
 )
 
-var index_dEnv *env.DoltEnv
-var index_initialRoot *doltdb.RootValue
-
-func init() {
-	index_dEnv = dtestutils.CreateTestEnv()
+func setupEditorIndexTest(t *testing.T) (*env.DoltEnv, *doltdb.RootValue) {
+	index_dEnv := dtestutils.CreateTestEnv()
 	root, err := index_dEnv.WorkingRoot(context.Background())
-	if err != nil {
-		panic(err)
-	}
-	index_initialRoot, err = ExecuteSql(index_dEnv, root, `
+	require.NoError(t, err)
+
+	index_initialRoot, err := ExecuteSql(index_dEnv, root, `
 CREATE TABLE onepk (
   pk1 BIGINT PRIMARY KEY,
   v1 BIGINT,
@@ -70,9 +66,10 @@ CREATE INDEX idx_v2v1 ON twopk(v2, v1);
 CREATE UNIQUE INDEX idx_v1 ON oneuni(v1);
 CREATE UNIQUE INDEX idx_v1v2 ON twouni(v1, v2);
 `)
-	if err != nil {
-		panic(err)
-	}
+
+	require.NoError(t, err)
+
+	return index_dEnv, index_initialRoot
 }
 
 func TestTableEditorIndexResults(t *testing.T) {
@@ -122,10 +119,12 @@ UPDATE onepk SET pk1 = v1 + pk1 ORDER BY pk1 DESC;
 
 	for _, test := range tests {
 		t.Run(test.sqlStatement, func(t *testing.T) {
-			root := index_initialRoot
+			dEnv, initialRoot := setupEditorIndexTest(t)
+
+			root := initialRoot
 			for _, sqlStatement := range strings.Split(test.sqlStatement, ";") {
 				var err error
-				root, err = executeModify(context.Background(), index_dEnv, root, sqlStatement)
+				root, err = executeModify(context.Background(), dEnv, root, sqlStatement)
 				require.NoError(t, err)
 			}
 
@@ -278,10 +277,12 @@ UPDATE oneuni SET v1 = v1 + pk1;
 
 	for _, test := range tests {
 		t.Run(test.sqlStatement, func(t *testing.T) {
-			root := index_initialRoot
+			dEnv, initialRoot := setupEditorIndexTest(t)
+
+			root := initialRoot
 			var err error
 			for _, sqlStatement := range strings.Split(test.sqlStatement, ";") {
-				root, err = executeModify(context.Background(), index_dEnv, root, sqlStatement)
+				root, err = executeModify(context.Background(), dEnv, root, sqlStatement)
 				if err != nil {
 					break
 				}

go/libraries/doltcore/sqle/tables.go

@@ -501,7 +501,10 @@ func (t *WritableDoltTable) Inserter(ctx *sql.Context) sql.RowInserter {
 }
 
 func (t *WritableDoltTable) getTableEditor(ctx *sql.Context) (*sqlTableEditor, error) {
-	if t.db.batchMode == batched {
+	sess := DSessFromSess(ctx.Session)
+
+	// In batched mode, reuse the same table editor. Otherwise, hand out a new one
+	if sess.batchMode == batched {
 		if t.ed != nil {
 			return t.ed, nil
 		}

go/libraries/doltcore/sqle/testutil.go

@@ -29,12 +29,14 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/env"
 )
 
-// Executes all the SQL non-select statements given in the string against the root value given and returns the updated
-// root, or an error. Statements in the input string are split by `;\n`
+// ExecuteSql executes all the SQL non-select statements given in the string against the root value given and returns
+// the updated root, or an error. Statements in the input string are split by `;\n`
 func ExecuteSql(dEnv *env.DoltEnv, root *doltdb.RootValue, statements string) (*doltdb.RootValue, error) {
-	db := NewBatchedDatabase("dolt", dEnv.DbData())
-	engine, ctx, err := NewTestEngine(context.Background(), db, root)
+	db := NewDatabase("dolt", dEnv.DbData())
 
+	engine, ctx, err := NewTestEngine(context.Background(), db, root)
+	DSessFromSess(ctx.Session).EnableBatchedMode()
+	err = ctx.Session.SetSessionVariable(ctx, sql.AutoCommitSessionVar, true)
 	if err != nil {
 		return nil, err
 	}
@@ -122,8 +124,7 @@ func NewTestEngine(ctx context.Context, db Database, root *doltdb.RootValue) (*s
 	return engine, sqlCtx, nil
 }
 
-// Executes the select statement given and returns the resulting rows, or an error if one is encountered.
-// This uses the index functionality, which is not ready for prime time. Use with caution.
+// ExecuteSelect executes the select statement given and returns the resulting rows, or an error if one is encountered.
 func ExecuteSelect(dEnv *env.DoltEnv, ddb *doltdb.DoltDB, root *doltdb.RootValue, query string) ([]sql.Row, error) {
 
 	dbData := env.DbData{

go/libraries/doltcore/sqle/transactions.go

@@ -112,6 +112,8 @@ func (tx *DoltTransaction) Commit(ctx *sql.Context, newRoot *doltdb.RootValue) (
 		err = tx.dbData.Ddb.UpdateWorkingSet(ctx, tx.workingSet, mergedRoot, hash)
 		if err == datas.ErrOptimisticLockFailed {
 			continue
+		} else if err != nil {
+			return nil, err
 		}
 
 		// TODO: this is not thread safe, but will not be necessary after migrating all clients away from using the

integration-tests/bats/1pk5col-ints.bats

@@ -126,12 +126,10 @@ teardown() {
 }
 
 @test "1pk5col-ints: dolt sql with insert ignore" {
-    skip "New engine does not support insert ignore"
     dolt sql -q "insert into test (pk,c1,c2,c3,c4,c5) values (0,6,6,6,6,6)"
     run dolt sql -q "insert ignore into test (pk,c1,c2,c3,c4,c5) values (0,6,6,6,6,6),(11,111,111,111,111,111)"
     [ "$status" -eq 0 ]
-    [ "${lines[0]}" = "Rows inserted: 1" ]
-    [ "${lines[1]}" = "Errors ignored: 1" ]
+    [[ "$output" = "Query OK, 1 row affected" ]] || false
     run dolt sql -q "select * from test"
     [[ "$output" =~ "111" ]] || false
 }

integration-tests/bats/sql-batch.bats

@@ -48,6 +48,21 @@ SQL
     [[ "$output" =~ "Rows inserted: 2" ]] || false
 }
 
+@test "sql-batch: script commits up until error" {
+    run dolt sql <<SQL
+insert into test values (0,0,0,0,0,0);
+insert into test values (1,0,0,0,0,0);
+insert into test values (a,b,c);
+insert into test values (2,0,0,0,0,0); -- will not run
+SQL
+    [ "$status" -eq 1 ]
+    [[ "$output" =~ "error on line 3 for query" ]] || false
+
+    run dolt sql -q "select count(*) from test" -r csv
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "2" ]] || false
+}
+
 @test "sql-batch: Line number and bad query displayed on error in batch sql" {
     run dolt sql <<SQL
 insert into test values (0,0,0,0,0,0);
@@ -60,13 +75,13 @@ SQL
     [[ "$output" =~ "insert into test values poop" ]] || false
 
     run dolt sql <<SQL
-insert into test values (0,0,0,0,0,0);
+insert into test values (2,0,0,0,0,0);
 
-insert into test values (1,0,
+insert into test values (3,0,
 0,0,0,0);
 
 insert into 
-test values (2,0,0,0,0,0)
+test values (4,0,0,0,0,0)
 ;
 
 insert into 
@@ -175,4 +190,4 @@ SQL
   run dolt sql -r csv -q 'SELECT pk FROM TEST ORDER BY pk;'
   [ "$status" -eq 0 ]
   [[ "$output" =~ "$EXPECTED" ]] || false
-}
+}

integration-tests/bats/sql-create-tables.bats

@@ -774,4 +774,4 @@ SELECT * FROM myTempTable;
 SQL
     [ "$status" -eq 1 ]
     [[ "$output" =~ "table not found: myTempTable" ]] || false
-}
+}

integration-tests/bats/sql-merge.bats

@@ -209,7 +209,7 @@ SELECT DOLT_CHECKOUT('master');
 SELECT DOLT_MERGE('feature-branch');
 SQL
     [ $status -eq 1 ]
-    [[ $output =~ "merge has conflicts" ]] || false
+    [[ $output =~ "merge has unresolved conflicts" ]] || false
 
     run dolt sql -q "SELECT DOLT_MERGE('--abort');"
     [ $status -eq 0 ]
@@ -244,8 +244,9 @@ SELECT DOLT_COMMIT('-a', '-m', 'changed feature branch');
 SELECT DOLT_CHECKOUT('master');
 SELECT DOLT_MERGE('feature-branch');
 SQL
+
     [ $status -eq 1 ]
-    [[ $output =~ "merge has conflicts" ]] || false
+    [[ $output =~ "merge has unresolved conflicts" ]] || false
 
     run dolt status
     [ $status -eq 0 ]
@@ -304,7 +305,7 @@ SELECT DOLT_CHECKOUT('master');
 SELECT DOLT_MERGE('feature-branch');
 SQL
     [ $status -eq 1 ]
-    [[ $output =~ "merge has conflicts" ]] || false
+    [[ $output =~ "merge has unresolved conflicts" ]] || false
 
     run dolt sql -q "SELECT DOLT_MERGE('feature-branch');"
     [ $status -eq 1 ]