Repository: vitessio/vitess
Pull request #6225 (2020-05-26 13:18:22)

Support uppercase user defined variables
This handled uppercase UDV which causing issues of missing bindvars from the vttablet
#6218

Similar to mysql behaviour.
```
mysql > select @myvalue;
+----------+
| @myvalue |
+----------+
| NULL     |
+----------+
1 row in set (0.00 sec)

mysql > set @MYVALUE = 42;
Query OK, 0 rows affected (0.00 sec)

mysql > select @myvalue;
+----------+
| @myvalue |
+----------+
|       42 |
+----------+
1 row in set (0.00 sec)

```

Signed-off-by: Harshit Gangal <harshit@planetscale.com>


Patch:
diff --git a/go/mysql/fakesqldb/server.go b/go/mysql/fakesqldb/server.go
index b71a5c8369..ada111ffd4 100644
--- a/go/mysql/fakesqldb/server.go
+++ b/go/mysql/fakesqldb/server.go
@@ -98,8 +98,6 @@ type DB struct {
 	patternData []exprResult
 	// queryCalled keeps track of how many times a query was called.
 	queryCalled map[string]int
-	// querylog keeps track of all called queries
-	querylog []string
 
 	// This next set of fields is used when ordering of the queries matters.
 
@@ -342,7 +340,6 @@ func (db *DB) HandleQuery(c *mysql.Conn, query string, callback func(*sqltypes.R
 	db.mu.Lock()
 	defer db.mu.Unlock()
 	db.queryCalled[key]++
-	db.querylog = append(db.querylog, key)
 
 	// Check if we should close the connection and provoke errno 2013.
 	if db.shouldClose {
@@ -526,11 +523,6 @@ func (db *DB) GetQueryCalledNum(query string) int {
 	return num
 }
 
-//QueryLog returns the query log in a semicomma separated string
-func (db *DB) QueryLog() string {
-	return strings.Join(db.querylog, ";")
-}
-
 // EnableConnFail makes connection to this fake DB fail.
 func (db *DB) EnableConnFail() {
 	db.mu.Lock()
diff --git a/go/pools/numbered.go b/go/pools/numbered.go
index f1e7648816..ef4e3ea306 100644
--- a/go/pools/numbered.go
+++ b/go/pools/numbered.go
@@ -51,7 +51,6 @@ func (u *unregistered) Size() int {
 	return 1
 }
 
-//NewNumbered creates a new numbered
 func NewNumbered() *Numbered {
 	n := &Numbered{
 		resources:            make(map[int64]*numberedWrapper),
@@ -87,7 +86,7 @@ func (nu *Numbered) Register(id int64, val interface{}, enforceTimeout bool) err
 
 // Unregister forgets the specified resource.  If the resource is not present, it's ignored.
 func (nu *Numbered) Unregister(id int64, reason string) {
-	success := nu.unregister(id)
+	success := nu.unregister(id, reason)
 	if success {
 		nu.recentlyUnregistered.Set(
 			fmt.Sprintf("%v", id), &unregistered{reason: reason, timeUnregistered: time.Now()})
@@ -96,7 +95,7 @@ func (nu *Numbered) Unregister(id int64, reason string) {
 
 // unregister forgets the resource, if it exists. Returns whether or not the resource existed at
 // time of Unregister.
-func (nu *Numbered) unregister(id int64) bool {
+func (nu *Numbered) unregister(id int64, reason string) bool {
 	nu.mu.Lock()
 	defer nu.mu.Unlock()
 
@@ -200,13 +199,11 @@ func (nu *Numbered) WaitForEmpty() {
 	}
 }
 
-//StatsJSON returns stats in JSON format
 func (nu *Numbered) StatsJSON() string {
 	return fmt.Sprintf("{\"Size\": %v}", nu.Size())
 }
 
-//Size returns the current size
-func (nu *Numbered) Size() int64 {
+func (nu *Numbered) Size() (size int64) {
 	nu.mu.Lock()
 	defer nu.mu.Unlock()
 	return int64(len(nu.resources))
diff --git a/go/vt/sqlparser/ast.go b/go/vt/sqlparser/ast.go
index 355a42e68a..9c86eb21c6 100644
--- a/go/vt/sqlparser/ast.go
+++ b/go/vt/sqlparser/ast.go
@@ -1705,9 +1705,7 @@ func (node *CaseExpr) Format(buf *TrackedBuffer) {
 func (node *Default) Format(buf *TrackedBuffer) {
 	buf.astPrintf(node, "default")
 	if node.ColName != "" {
-		buf.WriteString("(")
-		formatID(buf, node.ColName, strings.ToLower(node.ColName), NoAt)
-		buf.WriteString(")")
+		buf.astPrintf(node, "(%s)", node.ColName)
 	}
 }
 
diff --git a/go/vt/sqlparser/expression_rewriting.go b/go/vt/sqlparser/expression_rewriting.go
index fed9e5dd8a..ece90c80fa 100644
--- a/go/vt/sqlparser/expression_rewriting.go
+++ b/go/vt/sqlparser/expression_rewriting.go
@@ -17,6 +17,8 @@ limitations under the License.
 package sqlparser
 
 import (
+	"strings"
+
 	"vitess.io/vitess/go/vt/log"
 	querypb "vitess.io/vitess/go/vt/proto/query"
 	"vitess.io/vitess/go/vt/proto/vtrpc"
@@ -155,7 +157,7 @@ func (er *expressionRewriter) goingDown(cursor *Cursor) bool {
 		er.funcRewrite(cursor, node)
 	case *ColName:
 		if node.Name.at == SingleAt {
-			cursor.Replace(bindVarExpression(UserDefinedVariableName + node.Name.CompliantName()))
+			cursor.Replace(bindVarExpression(UserDefinedVariableName + strings.ToLower(node.Name.CompliantName())))
 			er.needBindVarFor(UserDefinedVariableName)
 		}
 	}
diff --git a/go/vt/vtgate/vindexes/lookup_internal.go b/go/vt/vtgate/vindexes/lookup_internal.go
index 178922675a..2ea8a77e45 100644
--- a/go/vt/vtgate/vindexes/lookup_internal.go
+++ b/go/vt/vtgate/vindexes/lookup_internal.go
@@ -36,7 +36,6 @@ type lookupInternal struct {
 	To            string   `json:"to"`
 	Autocommit    bool     `json:"autocommit,omitempty"`
 	Upsert        bool     `json:"upsert,omitempty"`
-	IgnoreNulls   bool     `json:"ignore_nulls,omitempty"`
 	sel, ver, del string
 }
 
@@ -49,12 +48,6 @@ func (lkp *lookupInternal) Init(lookupQueryParams map[string]string, autocommit,
 	}
 	lkp.FromColumns = fromColumns
 
-	var err error
-	lkp.IgnoreNulls, err = boolFromMap(lookupQueryParams, "ignore_nulls")
-	if err != nil {
-		return err
-	}
-
 	lkp.Autocommit = autocommit
 	lkp.Upsert = upsert
 
@@ -165,32 +158,15 @@ func (lkp *lookupInternal) Create(vcursor VCursor, rowsColValues [][]sqltypes.Va
 }
 
 func (lkp *lookupInternal) createCustom(vcursor VCursor, rowsColValues [][]sqltypes.Value, toValues []sqltypes.Value, ignoreMode bool, co vtgatepb.CommitOrder) error {
-	// Trim rows with null values
-	trimmedRowsCols := make([][]sqltypes.Value, 0, len(rowsColValues))
-	trimmedToValues := make([]sqltypes.Value, 0, len(toValues))
-nextRow:
-	for i, row := range rowsColValues {
-		for j, col := range row {
-			if col.IsNull() {
-				if !lkp.IgnoreNulls {
-					return fmt.Errorf("lookup.Create: input has null values: row: %d, col: %d", i, j)
-				}
-				continue nextRow
-			}
-		}
-		trimmedRowsCols = append(trimmedRowsCols, row)
-		trimmedToValues = append(trimmedToValues, toValues[i])
-	}
-	if len(trimmedRowsCols) == 0 {
+	if len(rowsColValues) == 0 {
+		// This code is unreachable. It's just a failsafe.
 		return nil
 	}
 	// We only need to check the first row. Number of cols per row
 	// is guaranteed by the engine to be uniform.
-	if len(trimmedRowsCols[0]) != len(lkp.FromColumns) {
-		return fmt.Errorf("lookup.Create: column vindex count does not match the columns in the lookup: %d vs %v", len(trimmedRowsCols[0]), lkp.FromColumns)
+	if len(rowsColValues[0]) != len(lkp.FromColumns) {
+		return fmt.Errorf("lookup.Create: column vindex count does not match the columns in the lookup: %d vs %v", len(rowsColValues[0]), lkp.FromColumns)
 	}
-	sort.Sort(&sorter{rowsColValues: trimmedRowsCols, toValues: trimmedToValues})
-
 	buf := new(bytes.Buffer)
 	if ignoreMode {
 		fmt.Fprintf(buf, "insert ignore into %s(", lkp.Table)
@@ -202,9 +178,13 @@ nextRow:
 	}
 	fmt.Fprintf(buf, "%s) values(", lkp.To)
 
-	bindVars := make(map[string]*querypb.BindVariable, 2*len(trimmedRowsCols))
-	for rowIdx := range trimmedToValues {
-		colIds := trimmedRowsCols[rowIdx]
+	bindVars := make(map[string]*querypb.BindVariable, 2*len(rowsColValues))
+	// Make a copy before sorting.
+	rowsColValues = append([][]sqltypes.Value(nil), rowsColValues...)
+	toValues = append([]sqltypes.Value(nil), toValues...)
+	sort.Sort(&sorter{rowsColValues: rowsColValues, toValues: toValues})
+	for rowIdx := range toValues {
+		colIds := rowsColValues[rowIdx]
 		if rowIdx != 0 {
 			buf.WriteString(", (")
 		}
@@ -215,7 +195,7 @@ nextRow:
 		}
 		toStr := lkp.To + strconv.Itoa(rowIdx)
 		buf.WriteString(":" + toStr + ")")
-		bindVars[toStr] = sqltypes.ValueBindVariable(trimmedToValues[rowIdx])
+		bindVars[toStr] = sqltypes.ValueBindVariable(toValues[rowIdx])
 	}
 
 	if lkp.Upsert {
diff --git a/go/vt/vttablet/endtoend/framework/eventcatcher.go b/go/vt/vttablet/endtoend/framework/eventcatcher.go
index 429a40ffb8..3579709b80 100644
--- a/go/vt/vttablet/endtoend/framework/eventcatcher.go
+++ b/go/vt/vttablet/endtoend/framework/eventcatcher.go
@@ -44,12 +44,12 @@ func (tc *TxCatcher) Close() {
 
 // Next fetches the next captured transaction.
 // If the wait is longer than one second, it returns an error.
-func (tc *TxCatcher) Next() (*tabletserver.StatefulConnection, error) {
+func (tc *TxCatcher) Next() (*tabletserver.TxConnection, error) {
 	event, err := tc.catcher.next()
 	if err != nil {
 		return nil, err
 	}
-	return event.(*tabletserver.StatefulConnection), nil
+	return event.(*tabletserver.TxConnection), nil
 }
 
 // QueryCatcher allows you to capture and fetch queries that are being
diff --git a/go/vt/vttablet/tabletserver/query_executor.go b/go/vt/vttablet/tabletserver/query_executor.go
index d637a52fd4..55ff6abf2d 100644
--- a/go/vt/vttablet/tabletserver/query_executor.go
+++ b/go/vt/vttablet/tabletserver/query_executor.go
@@ -22,8 +22,6 @@ import (
 	"strings"
 	"time"
 
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
 	"vitess.io/vitess/go/vt/vtgate/evalengine"
 
 	"golang.org/x/net/context"
@@ -115,11 +113,11 @@ func (qre *QueryExecutor) Execute() (reply *sqltypes.Result, err error) {
 
 	if qre.transactionID != 0 {
 		// Need upfront connection for DMLs and transactions
-		conn, err := qre.tsv.te.txPool.GetAndLock(qre.transactionID, "for query")
+		conn, err := qre.tsv.te.txPool.Get(qre.transactionID, "for query")
 		if err != nil {
 			return nil, err
 		}
-		defer conn.Unlock()
+		defer conn.Recycle()
 		return qre.txConnExec(conn)
 	}
 
@@ -147,51 +145,50 @@ func (qre *QueryExecutor) Execute() (reply *sqltypes.Result, err error) {
 	return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "%s unexpected plan type", qre.plan.PlanID.String())
 }
 
-func (qre *QueryExecutor) execAutocommit(f func(conn tx.IStatefulConnection) (*sqltypes.Result, error)) (reply *sqltypes.Result, err error) {
+func (qre *QueryExecutor) execAutocommit(f func(conn *TxConnection) (*sqltypes.Result, error)) (reply *sqltypes.Result, err error) {
 	if qre.options == nil {
 		qre.options = &querypb.ExecuteOptions{}
 	}
 	qre.options.TransactionIsolation = querypb.ExecuteOptions_AUTOCOMMIT
-
-	conn, _, err := qre.tsv.te.txPool.Begin(qre.ctx, qre.options)
-
+	conn, _, err := qre.tsv.te.txPool.LocalBegin(qre.ctx, qre.options)
 	if err != nil {
 		return nil, err
 	}
-	defer qre.tsv.te.txPool.RollbackAndRelease(qre.ctx, conn)
+	defer qre.tsv.te.txPool.LocalConclude(qre.ctx, conn)
 
 	return f(conn)
 }
 
-func (qre *QueryExecutor) execAsTransaction(f func(conn tx.IStatefulConnection) (*sqltypes.Result, error)) (*sqltypes.Result, error) {
-	conn, beginSQL, err := qre.tsv.te.txPool.Begin(qre.ctx, qre.options)
+func (qre *QueryExecutor) execAsTransaction(f func(conn *TxConnection) (*sqltypes.Result, error)) (reply *sqltypes.Result, err error) {
+	conn, beginSQL, err := qre.tsv.te.txPool.LocalBegin(qre.ctx, qre.options)
 	if err != nil {
 		return nil, err
 	}
-	defer qre.tsv.te.txPool.RollbackAndRelease(qre.ctx, conn)
+	defer qre.tsv.te.txPool.LocalConclude(qre.ctx, conn)
 	qre.logStats.AddRewrittenSQL(beginSQL, time.Now())
 
-	result, err := f(conn)
+	reply, err = f(conn)
 	if err != nil {
 		// dbConn is nil, it means the transaction was aborted.
 		// If so, we should not relog the rollback.
 		// TODO(sougou): these txPool functions should take the logstats
 		// and log any statements they issue. This needs to be done as
 		// a separate refactor because it impacts lot of code.
-		if conn.IsInTransaction() {
+		if conn.dbConn != nil {
 			defer qre.logStats.AddRewrittenSQL("rollback", time.Now())
+			qre.tsv.te.txPool.LocalConclude(qre.ctx, conn)
 		}
 		return nil, err
 	}
 
 	defer qre.logStats.AddRewrittenSQL("commit", time.Now())
-	if _, err := qre.tsv.te.txPool.Commit(qre.ctx, conn); err != nil {
+	if _, err := qre.tsv.te.txPool.LocalCommit(qre.ctx, conn); err != nil {
 		return nil, err
 	}
-	return result, nil
+	return reply, nil
 }
 
-func (qre *QueryExecutor) txConnExec(conn tx.IStatefulConnection) (*sqltypes.Result, error) {
+func (qre *QueryExecutor) txConnExec(conn *TxConnection) (*sqltypes.Result, error) {
 	switch qre.plan.PlanID {
 	case planbuilder.PlanInsert, planbuilder.PlanUpdate, planbuilder.PlanDelete:
 		return qre.txFetch(conn, true)
@@ -235,12 +232,12 @@ func (qre *QueryExecutor) Stream(callback func(*sqltypes.Result) error) error {
 	// if we have a transaction id, let's use the txPool for this query
 	var conn *connpool.DBConn
 	if qre.transactionID != 0 {
-		txConn, err := qre.tsv.te.txPool.GetAndLock(qre.transactionID, "for streaming query")
+		txConn, err := qre.tsv.te.txPool.Get(qre.transactionID, "for streaming query")
 		if err != nil {
 			return err
 		}
-		defer txConn.Unlock()
-		conn = txConn.UnderlyingdDBConn()
+		defer txConn.Recycle()
+		conn = txConn.dbConn
 	} else {
 		dbConn, err := qre.getStreamConn()
 		if err != nil {
@@ -364,7 +361,7 @@ func (qre *QueryExecutor) checkAccess(authorized *tableacl.ACLResult, tableName
 	return nil
 }
 
-func (qre *QueryExecutor) execDDL(conn tx.IStatefulConnection) (*sqltypes.Result, error) {
+func (qre *QueryExecutor) execDDL(conn *TxConnection) (*sqltypes.Result, error) {
 	defer func() {
 		if err := qre.tsv.se.Reload(qre.ctx); err != nil {
 			log.Errorf("failed to reload schema %v", err)
@@ -375,27 +372,13 @@ func (qre *QueryExecutor) execDDL(conn tx.IStatefulConnection) (*sqltypes.Result
 	if err != nil {
 		return nil, err
 	}
-	err = qre.BeginAgain(qre.ctx, conn)
+	err = conn.BeginAgain(qre.ctx)
 	if err != nil {
 		return nil, err
 	}
 	return result, nil
 }
 
-// BeginAgain commits the existing transaction and begins a new one
-func (*QueryExecutor) BeginAgain(ctx context.Context, dc tx.IStatefulConnection) error {
-	if dc.IsClosed() || dc.TxProperties().Autocommit {
-		return nil
-	}
-	if _, err := dc.Exec(ctx, "commit", 1, false); err != nil {
-		return err
-	}
-	if _, err := dc.Exec(ctx, "begin", 1, false); err != nil {
-		return err
-	}
-	return nil
-}
-
 func (qre *QueryExecutor) execNextval() (*sqltypes.Result, error) {
 	inc, err := resolveNumber(qre.plan.NextCount, qre.bindVars)
 	if err != nil {
@@ -410,7 +393,7 @@ func (qre *QueryExecutor) execNextval() (*sqltypes.Result, error) {
 	t.SequenceInfo.Lock()
 	defer t.SequenceInfo.Unlock()
 	if t.SequenceInfo.NextVal == 0 || t.SequenceInfo.NextVal+inc > t.SequenceInfo.LastVal {
-		_, err := qre.execAsTransaction(func(conn tx.IStatefulConnection) (*sqltypes.Result, error) {
+		_, err := qre.execAsTransaction(func(conn *TxConnection) (*sqltypes.Result, error) {
 			query := fmt.Sprintf("select next_id, cache from %s where id = 0 for update", sqlparser.String(tableName))
 			qr, err := qre.execSQL(conn, query, false)
 			if err != nil {
@@ -446,7 +429,7 @@ func (qre *QueryExecutor) execNextval() (*sqltypes.Result, error) {
 				newLast += cache
 			}
 			query = fmt.Sprintf("update %s set next_id = %d where id = 0", sqlparser.String(tableName), newLast)
-			conn.TxProperties().RecordQuery(query)
+			conn.RecordQuery(query)
 			_, err = qre.execSQL(conn, query, false)
 			if err != nil {
 				return nil, err
@@ -490,7 +473,7 @@ func (qre *QueryExecutor) execSelect() (*sqltypes.Result, error) {
 	return qre.dbConnFetch(conn, qre.plan.FullQuery, qre.bindVars)
 }
 
-func (qre *QueryExecutor) execDMLLimit(conn tx.IStatefulConnection) (*sqltypes.Result, error) {
+func (qre *QueryExecutor) execDMLLimit(conn *TxConnection) (*sqltypes.Result, error) {
 	maxrows := qre.tsv.qe.maxResultSize.Get()
 	qre.bindVars["#maxLimit"] = sqltypes.Int64BindVariable(maxrows + 1)
 	result, err := qre.txFetch(conn, true)
@@ -499,7 +482,7 @@ func (qre *QueryExecutor) execDMLLimit(conn tx.IStatefulConnection) (*sqltypes.R
 	}
 	if err := qre.verifyRowCount(int64(result.RowsAffected), maxrows); err != nil {
 		defer qre.logStats.AddRewrittenSQL("rollback", time.Now())
-		_ = qre.tsv.te.txPool.Rollback(qre.ctx, conn)
+		qre.tsv.te.txPool.LocalConclude(qre.ctx, conn)
 		return nil, err
 	}
 	return result, nil
@@ -602,7 +585,7 @@ func (qre *QueryExecutor) qFetch(logStats *tabletenv.LogStats, parsedQuery *sqlp
 }
 
 // txFetch fetches from a TxConnection.
-func (qre *QueryExecutor) txFetch(conn tx.IStatefulConnection, record bool) (*sqltypes.Result, error) {
+func (qre *QueryExecutor) txFetch(conn *TxConnection, record bool) (*sqltypes.Result, error) {
 	sql, _, err := qre.generateFinalSQL(qre.plan.FullQuery, qre.bindVars)
 	if err != nil {
 		return nil, err
@@ -613,7 +596,7 @@ func (qre *QueryExecutor) txFetch(conn tx.IStatefulConnection, record bool) (*sq
 	}
 	// Only record successful queries.
 	if record {
-		conn.TxProperties().RecordQuery(sql)
+		conn.RecordQuery(sql)
 	}
 	return qr, nil
 }
@@ -660,12 +643,12 @@ func (qre *QueryExecutor) getSelectLimit() int64 {
 	return maxRows
 }
 
-// executor is an abstraction for reusing code in execSQL.
-type executor interface {
+// poolConn is an abstraction for reusing code in execSQL.
+type poolConn interface {
 	Exec(ctx context.Context, query string, maxrows int, wantfields bool) (*sqltypes.Result, error)
 }
 
-func (qre *QueryExecutor) execSQL(conn executor, sql string, wantfields bool) (*sqltypes.Result, error) {
+func (qre *QueryExecutor) execSQL(conn poolConn, sql string, wantfields bool) (*sqltypes.Result, error) {
 	span, ctx := trace.NewSpan(qre.ctx, "QueryExecutor.execSQL")
 	defer span.Finish()
 
diff --git a/go/vt/vttablet/tabletserver/stateful_connection.go b/go/vt/vttablet/tabletserver/stateful_connection.go
deleted file mode 100644
index b698cf392e..0000000000
--- a/go/vt/vttablet/tabletserver/stateful_connection.go
+++ /dev/null
@@ -1,162 +0,0 @@
-/*
-Copyright 2019 The Vitess Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package tabletserver
-
-import (
-	"fmt"
-
-	"vitess.io/vitess/go/mysql"
-	"vitess.io/vitess/go/sqltypes"
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/connpool"
-
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
-	"golang.org/x/net/context"
-
-	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
-	"vitess.io/vitess/go/vt/vterrors"
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
-)
-
-// StatefulConnection is used in the situations where we need a dedicated connection for a vtgate session.
-// This is used for transactions and reserved connections.
-// NOTE: After use, if must be returned either by doing a Unlock() or a Release().
-type StatefulConnection struct {
-	pool   *StatefulConnectionPool
-	dbConn *connpool.DBConn
-	ConnID tx.ConnID
-	env    tabletenv.Env
-
-	txProps *tx.Properties
-}
-
-// Close closes the underlying connection. When the connection is Unblocked, it will be Released
-func (sc *StatefulConnection) Close() {
-	if sc.dbConn != nil {
-		sc.dbConn.Close()
-	}
-}
-
-//IsClosed returns true when the connection is still operational
-func (sc *StatefulConnection) IsClosed() bool {
-	return sc.dbConn == nil || sc.dbConn.IsClosed()
-}
-
-//IsInTransaction returns true when the connection has tx state
-func (sc *StatefulConnection) IsInTransaction() bool {
-	return sc.txProps != nil
-}
-
-// Exec executes the statement in the dedicated connection
-func (sc *StatefulConnection) Exec(ctx context.Context, query string, maxrows int, wantfields bool) (*sqltypes.Result, error) {
-	if sc.IsClosed() {
-		if sc.IsInTransaction() {
-			return nil, vterrors.Errorf(vtrpcpb.Code_ABORTED, "transaction was aborted: %v", sc.txProps.Conclusion)
-		}
-		return nil, vterrors.New(vtrpcpb.Code_ABORTED, "connection was aborted")
-	}
-	r, err := sc.dbConn.ExecOnce(ctx, query, maxrows, wantfields)
-	if err != nil {
-		if mysql.IsConnErr(err) {
-			select {
-			case <-ctx.Done():
-				// If the context is done, the query was killed.
-				// So, don't trigger a mysql check.
-			default:
-				sc.env.CheckMySQL()
-			}
-		}
-		return nil, err
-	}
-	return r, nil
-}
-
-func (sc *StatefulConnection) execWithRetry(ctx context.Context, query string, maxrows int, wantfields bool) error {
-	if sc.IsClosed() {
-		return nil
-	}
-	if _, err := sc.dbConn.Exec(ctx, query, maxrows, wantfields); err != nil {
-		return err
-	}
-	return nil
-}
-
-// Unlock returns the connection to the pool. The connection remains active.
-// This method is idempotent and can be called multiple times
-func (sc *StatefulConnection) Unlock() {
-	if sc.dbConn == nil {
-		return
-	}
-	if sc.dbConn.IsClosed() {
-		sc.Releasef("unlocked closed connection")
-	} else {
-		sc.pool.markAsNotInUse(sc.ConnID)
-	}
-}
-
-//Release is used when the connection will not be used ever again.
-//The underlying dbConn is removed so that this connection cannot be used by mistake.
-func (sc *StatefulConnection) Release(reason tx.ReleaseReason) {
-	sc.Releasef(reason.String())
-}
-
-//Releasef is used when the connection will not be used ever again.
-//The underlying dbConn is removed so that this connection cannot be used by mistake.
-func (sc *StatefulConnection) Releasef(reasonFormat string, a ...interface{}) {
-	if sc.dbConn == nil {
-		return
-	}
-	sc.pool.unregister(sc.ConnID, fmt.Sprintf(reasonFormat, a...))
-	sc.dbConn.Recycle()
-	sc.dbConn = nil
-}
-
-// String returns a printable version of the connection info.
-func (sc *StatefulConnection) String() string {
-	return fmt.Sprintf(
-		"%v\t%s",
-		sc.ConnID,
-		sc.txProps.String(),
-	)
-}
-
-//TxProperties returns the transactional properties of the connection
-func (sc *StatefulConnection) TxProperties() *tx.Properties {
-	return sc.txProps
-}
-
-//ID returns the identifier for this connection
-func (sc *StatefulConnection) ID() tx.ConnID {
-	return sc.ConnID
-}
-
-//UnderlyingdDBConn returns the underlying database connection
-func (sc *StatefulConnection) UnderlyingdDBConn() *connpool.DBConn {
-	return sc.dbConn
-}
-
-//CleanTxState cleans out the current transaction state
-func (sc *StatefulConnection) CleanTxState() {
-	sc.txProps = nil
-}
-
-//Stats implements the tx.IStatefulConnection interface
-func (sc *StatefulConnection) Stats() *tabletenv.Stats {
-	return sc.env.Stats()
-}
-
-var _ tx.IStatefulConnection = (*StatefulConnection)(nil)
diff --git a/go/vt/vttablet/tabletserver/stateful_connection_pool.go b/go/vt/vttablet/tabletserver/stateful_connection_pool.go
deleted file mode 100644
index f2d2903ee6..0000000000
--- a/go/vt/vttablet/tabletserver/stateful_connection_pool.go
+++ /dev/null
@@ -1,194 +0,0 @@
-/*
-Copyright 2020 The Vitess Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package tabletserver
-
-import (
-	"time"
-
-	"vitess.io/vitess/go/pools"
-
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
-	"golang.org/x/net/context"
-
-	"vitess.io/vitess/go/sync2"
-	"vitess.io/vitess/go/vt/dbconfigs"
-	"vitess.io/vitess/go/vt/log"
-	querypb "vitess.io/vitess/go/vt/proto/query"
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/connpool"
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
-)
-
-//StatefulConnectionPool keeps track of currently and future active connections
-//it's used whenever the session has some state that requires a dedicated connection
-type StatefulConnectionPool struct {
-	// conns is the 'regular' pool. By default, connections
-	// are pulled from here for starting transactions.
-	conns *connpool.Pool
-
-	// foundRowsPool is the alternate pool that creates
-	// connections with CLIENT_FOUND_ROWS flag set. A separate
-	// pool is needed because this option can only be set at
-	// connection time.
-	foundRowsPool *connpool.Pool
-	active        *pools.Numbered
-	lastID        sync2.AtomicInt64
-	env           tabletenv.Env
-}
-
-//NewStatefulConnPool creates an ActivePool
-func NewStatefulConnPool(env tabletenv.Env) *StatefulConnectionPool {
-	config := env.Config()
-
-	return &StatefulConnectionPool{
-		env:           env,
-		conns:         connpool.NewPool(env, "TransactionPool", config.TxPool),
-		foundRowsPool: connpool.NewPool(env, "FoundRowsPool", config.TxPool),
-		active:        pools.NewNumbered(),
-		lastID:        sync2.NewAtomicInt64(time.Now().UnixNano()),
-	}
-}
-
-// Open makes the TxPool operational. This also starts the transaction killer
-// that will kill long-running transactions.
-func (sf *StatefulConnectionPool) Open(appParams, dbaParams, appDebugParams dbconfigs.Connector) {
-	log.Infof("Starting transaction id: %d", sf.lastID)
-	sf.conns.Open(appParams, dbaParams, appDebugParams)
-	foundRowsParam, _ := appParams.MysqlParams()
-	foundRowsParam.EnableClientFoundRows()
-	appParams = dbconfigs.New(foundRowsParam)
-	sf.foundRowsPool.Open(appParams, dbaParams, appDebugParams)
-}
-
-// Close closes the TxPool. A closed pool can be reopened.
-func (sf *StatefulConnectionPool) Close() {
-	for _, v := range sf.active.GetOutdated(time.Duration(0), "for closing") {
-		conn := v.(*StatefulConnection)
-		thing := "connection"
-		if conn.IsInTransaction() {
-			thing = "transaction"
-		}
-		log.Warningf("killing %s for shutdown: %s", thing, conn.String())
-		sf.env.Stats().InternalErrors.Add("StrayTransactions", 1)
-		conn.Close()
-		conn.Releasef("pool closed")
-	}
-	sf.conns.Close()
-	sf.foundRowsPool.Close()
-}
-
-// AdjustLastID adjusts the last transaction id to be at least
-// as large as the input value. This will ensure that there are
-// no dtid collisions with future transactions.
-func (sf *StatefulConnectionPool) AdjustLastID(id int64) {
-	if current := sf.lastID.Get(); current < id {
-		log.Infof("Adjusting transaction id to: %d", id)
-		sf.lastID.Set(id)
-	}
-}
-
-// GetOutdated returns a list of connections that are older than age.
-// It does not return any connections that are in use.
-func (sf *StatefulConnectionPool) GetOutdated(age time.Duration, purpose string) []*StatefulConnection {
-	return mapToTxConn(sf.active.GetOutdated(age, purpose))
-}
-
-func mapToTxConn(outdated []interface{}) []*StatefulConnection {
-	result := make([]*StatefulConnection, len(outdated))
-	for i, el := range outdated {
-		result[i] = el.(*StatefulConnection)
-	}
-	return result
-}
-
-// WaitForEmpty returns as soon as the pool becomes empty
-func (sf *StatefulConnectionPool) WaitForEmpty() {
-	sf.active.WaitForEmpty()
-}
-
-// GetAndLock locks the connection for use. It accepts a purpose as a string.
-// If it cannot be found, it returns a "not found" error. If in use,
-// it returns a "in use: purpose" error.
-func (sf *StatefulConnectionPool) GetAndLock(id int64, reason string) (*StatefulConnection, error) {
-	conn, err := sf.active.Get(id, reason)
-	if err != nil {
-		return nil, err
-	}
-	return conn.(*StatefulConnection), nil
-}
-
-//NewConn creates a new StatefulConnection. It will be created from either the normal pool or
-//the found_rows pool, depending on the options provided
-func (sf *StatefulConnectionPool) NewConn(ctx context.Context, options *querypb.ExecuteOptions) (*StatefulConnection, error) {
-
-	var conn *connpool.DBConn
-	var err error
-
-	if options.GetClientFoundRows() {
-		conn, err = sf.foundRowsPool.Get(ctx)
-	} else {
-		conn, err = sf.conns.Get(ctx)
-	}
-	if err != nil {
-		return nil, err
-	}
-
-	connID := sf.lastID.Add(1)
-	sfConn := &StatefulConnection{
-		dbConn: conn,
-		ConnID: connID,
-		pool:   sf,
-		env:    sf.env,
-	}
-
-	err = sf.active.Register(
-		sfConn.ConnID,
-		sfConn,
-		options.GetWorkload() != querypb.ExecuteOptions_DBA,
-	)
-	if err != nil {
-		sfConn.Release(tx.ConnInitFail)
-		return nil, err
-	}
-
-	return sf.GetAndLock(sfConn.ConnID, "new connection")
-}
-
-//ForAllTxProperties executes a function an every connection that has a not-nil TxProperties
-func (sf *StatefulConnectionPool) ForAllTxProperties(f func(*tx.Properties)) {
-	for _, connection := range mapToTxConn(sf.active.GetAll()) {
-		props := connection.txProps
-		if props != nil {
-			f(props)
-		}
-	}
-}
-
-// Unregister forgets the specified connection.  If the connection is not present, it's ignored.
-func (sf *StatefulConnectionPool) unregister(id tx.ConnID, reason string) {
-	sf.active.Unregister(id, reason)
-}
-
-//markAsNotInUse marks the connection as not in use at the moment
-func (sf *StatefulConnectionPool) markAsNotInUse(id tx.ConnID) {
-	sf.active.Put(id)
-}
-
-// Capacity returns the pool capacity.
-func (sf *StatefulConnectionPool) Capacity() int {
-	return int(sf.conns.Capacity())
-}
diff --git a/go/vt/vttablet/tabletserver/tabletserver.go b/go/vt/vttablet/tabletserver/tabletserver.go
index 6a7c9ea5d9..be94d0f5d8 100644
--- a/go/vt/vttablet/tabletserver/tabletserver.go
+++ b/go/vt/vttablet/tabletserver/tabletserver.go
@@ -29,8 +29,6 @@ import (
 	"syscall"
 	"time"
 
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
 	"golang.org/x/net/context"
 	"vitess.io/vitess/go/acl"
 	"vitess.io/vitess/go/history"
@@ -170,8 +168,6 @@ type TabletServer struct {
 	vstreamer *vstreamer.Engine
 	messager  *messager.Engine
 
-	txController tx.EngineStateMachine
-
 	// checkMySQLThrottler is used to throttle the number of
 	// requests sent to CheckMySQL.
 	checkMySQLThrottler *sync2.Semaphore
@@ -230,7 +226,6 @@ func NewTabletServer(name string, config *tabletenv.TabletConfig, topoServer *to
 	tsv.se = schema.NewEngine(tsv)
 	tsv.qe = NewQueryEngine(tsv, tsv.se)
 	tsv.te = NewTxEngine(tsv)
-	tsv.txController = tsv.te
 	tsv.hw = heartbeat.NewWriter(tsv, alias)
 	tsv.hr = heartbeat.NewReader(tsv)
 	tsv.txThrottler = txthrottler.NewTxThrottler(tsv.config, topoServer)
@@ -523,7 +518,7 @@ func (tsv *TabletServer) fullStart() (err error) {
 	if err := tsv.qe.Open(); err != nil {
 		return err
 	}
-	if err := tsv.txController.Init(); err != nil {
+	if err := tsv.te.Init(); err != nil {
 		return err
 	}
 	if err := tsv.hw.Init(tsv.target); err != nil {
@@ -541,7 +536,7 @@ func (tsv *TabletServer) serveNewType() (err error) {
 	// transactional requests are not allowed. So, we can
 	// be sure that the tx pool won't change after the wait.
 	if tsv.target.TabletType == topodatapb.TabletType_MASTER {
-		tsv.txController.AcceptReadWrite()
+		tsv.te.AcceptReadWrite()
 		if err := tsv.txThrottler.Open(tsv.target.Keyspace, tsv.target.Shard); err != nil {
 			return err
 		}
@@ -549,7 +544,7 @@ func (tsv *TabletServer) serveNewType() (err error) {
 		tsv.hr.Close()
 		tsv.hw.Open()
 	} else {
-		tsv.txController.AcceptReadOnly()
+		tsv.te.AcceptReadOnly()
 		tsv.messager.Close()
 		tsv.hr.Open()
 		tsv.hw.Close()
@@ -605,7 +600,7 @@ func (tsv *TabletServer) waitForShutdown() {
 	// will be allowed. They will enable the conclusion of outstanding
 	// transactions.
 	tsv.messager.Close()
-	tsv.txController.StopGently()
+	tsv.te.StopGently()
 	tsv.qe.streamQList.TerminateAll()
 	tsv.watcher.Close()
 	tsv.requests.Wait()
@@ -620,7 +615,7 @@ func (tsv *TabletServer) closeAll() {
 	tsv.vstreamer.Close()
 	tsv.hr.Close()
 	tsv.hw.Close()
-	tsv.txController.StopGently()
+	tsv.te.StopGently()
 	tsv.watcher.Close()
 	tsv.qe.Close()
 	tsv.se.Close()
@@ -958,7 +953,7 @@ func (tsv *TabletServer) Execute(ctx context.Context, target *querypb.Target, sq
 	trace.AnnotateSQL(span, sql)
 	defer span.Finish()
 
-	allowOnShutdown := transactionID != 0
+	allowOnShutdown := (transactionID != 0)
 	err = tsv.execRequest(
 		ctx, tsv.QueryTimeout.Get(),
 		"Execute", sql, bindVariables,
@@ -1850,12 +1845,12 @@ func (tsv *TabletServer) StreamPoolSize() int {
 // SetTxPoolSize changes the tx pool size to the specified value.
 // This function should only be used for testing.
 func (tsv *TabletServer) SetTxPoolSize(val int) {
-	tsv.te.txPool.scp.conns.SetCapacity(val)
+	tsv.te.txPool.conns.SetCapacity(val)
 }
 
 // TxPoolSize returns the tx pool size.
 func (tsv *TabletServer) TxPoolSize() int {
-	return tsv.te.txPool.scp.Capacity()
+	return int(tsv.te.txPool.conns.Capacity())
 }
 
 // SetTxTimeout changes the transaction timeout to the specified value.
diff --git a/go/vt/vttablet/tabletserver/twopc.go b/go/vt/vttablet/tabletserver/twopc.go
index df03db170e..4c365afb5d 100644
--- a/go/vt/vttablet/tabletserver/twopc.go
+++ b/go/vt/vttablet/tabletserver/twopc.go
@@ -20,8 +20,6 @@ import (
 	"fmt"
 	"time"
 
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
 	"vitess.io/vitess/go/vt/vtgate/evalengine"
 
 	"golang.org/x/net/context"
@@ -211,7 +209,7 @@ func (tpc *TwoPC) Close() {
 }
 
 // SaveRedo saves the statements in the redo log using the supplied connection.
-func (tpc *TwoPC) SaveRedo(ctx context.Context, conn tx.IStatefulConnection, dtid string, queries []string) error {
+func (tpc *TwoPC) SaveRedo(ctx context.Context, conn *TxConnection, dtid string, queries []string) error {
 	bindVars := map[string]*querypb.BindVariable{
 		"dtid":         sqltypes.StringBindVariable(dtid),
 		"state":        sqltypes.Int64BindVariable(RedoStatePrepared),
@@ -242,7 +240,7 @@ func (tpc *TwoPC) SaveRedo(ctx context.Context, conn tx.IStatefulConnection, dti
 }
 
 // UpdateRedo changes the state of the redo log for the dtid.
-func (tpc *TwoPC) UpdateRedo(ctx context.Context, conn tx.IStatefulConnection, dtid string, state int) error {
+func (tpc *TwoPC) UpdateRedo(ctx context.Context, conn *TxConnection, dtid string, state int) error {
 	bindVars := map[string]*querypb.BindVariable{
 		"dtid":  sqltypes.StringBindVariable(dtid),
 		"state": sqltypes.Int64BindVariable(int64(state)),
@@ -252,7 +250,7 @@ func (tpc *TwoPC) UpdateRedo(ctx context.Context, conn tx.IStatefulConnection, d
 }
 
 // DeleteRedo deletes the redo log for the dtid.
-func (tpc *TwoPC) DeleteRedo(ctx context.Context, conn tx.IStatefulConnection, dtid string) error {
+func (tpc *TwoPC) DeleteRedo(ctx context.Context, conn *TxConnection, dtid string) error {
 	bindVars := map[string]*querypb.BindVariable{
 		"dtid": sqltypes.StringBindVariable(dtid),
 	}
@@ -264,8 +262,15 @@ func (tpc *TwoPC) DeleteRedo(ctx context.Context, conn tx.IStatefulConnection, d
 	return err
 }
 
+// PreparedTx represents a displayable version of a prepared transaction.
+type PreparedTx struct {
+	Dtid    string
+	Queries []string
+	Time    time.Time
+}
+
 // ReadAllRedo returns all the prepared transactions from the redo logs.
-func (tpc *TwoPC) ReadAllRedo(ctx context.Context) (prepared, failed []*tx.PreparedTx, err error) {
+func (tpc *TwoPC) ReadAllRedo(ctx context.Context) (prepared, failed []*PreparedTx, err error) {
 	conn, err := tpc.readPool.Get(ctx)
 	if err != nil {
 		return nil, nil, err
@@ -277,7 +282,7 @@ func (tpc *TwoPC) ReadAllRedo(ctx context.Context) (prepared, failed []*tx.Prepa
 		return nil, nil, err
 	}
 
-	var curTx *tx.PreparedTx
+	var curTx *PreparedTx
 	for _, row := range qr.Rows {
 		dtid := row[0].ToString()
 		if curTx == nil || dtid != curTx.Dtid {
@@ -285,7 +290,7 @@ func (tpc *TwoPC) ReadAllRedo(ctx context.Context) (prepared, failed []*tx.Prepa
 			// A failure in time parsing will show up as a very old time,
 			// which is harmless.
 			tm, _ := evalengine.ToInt64(row[2])
-			curTx = &tx.PreparedTx{
+			curTx = &PreparedTx{
 				Dtid: dtid,
 				Time: time.Unix(0, tm),
 			}
@@ -331,7 +336,7 @@ func (tpc *TwoPC) CountUnresolvedRedo(ctx context.Context, unresolvedTime time.T
 }
 
 // CreateTransaction saves the metadata of a 2pc transaction as Prepared.
-func (tpc *TwoPC) CreateTransaction(ctx context.Context, conn tx.IStatefulConnection, dtid string, participants []*querypb.Target) error {
+func (tpc *TwoPC) CreateTransaction(ctx context.Context, conn *TxConnection, dtid string, participants []*querypb.Target) error {
 	bindVars := map[string]*querypb.BindVariable{
 		"dtid":     sqltypes.StringBindVariable(dtid),
 		"state":    sqltypes.Int64BindVariable(int64(DTStatePrepare)),
@@ -364,7 +369,7 @@ func (tpc *TwoPC) CreateTransaction(ctx context.Context, conn tx.IStatefulConnec
 
 // Transition performs a transition from Prepare to the specified state.
 // If the transaction is not a in the Prepare state, an error is returned.
-func (tpc *TwoPC) Transition(ctx context.Context, conn tx.IStatefulConnection, dtid string, state querypb.TransactionState) error {
+func (tpc *TwoPC) Transition(ctx context.Context, conn *TxConnection, dtid string, state querypb.TransactionState) error {
 	bindVars := map[string]*querypb.BindVariable{
 		"dtid":    sqltypes.StringBindVariable(dtid),
 		"state":   sqltypes.Int64BindVariable(int64(state)),
@@ -381,7 +386,7 @@ func (tpc *TwoPC) Transition(ctx context.Context, conn tx.IStatefulConnection, d
 }
 
 // DeleteTransaction deletes the metadata for the specified transaction.
-func (tpc *TwoPC) DeleteTransaction(ctx context.Context, conn tx.IStatefulConnection, dtid string) error {
+func (tpc *TwoPC) DeleteTransaction(ctx context.Context, conn *TxConnection, dtid string) error {
 	bindVars := map[string]*querypb.BindVariable{
 		"dtid": sqltypes.StringBindVariable(dtid),
 	}
@@ -469,8 +474,17 @@ func (tpc *TwoPC) ReadAbandoned(ctx context.Context, abandonTime time.Time) (map
 	return txs, nil
 }
 
+// DistributedTx is similar to querypb.TransactionMetadata, but
+// is display friendly.
+type DistributedTx struct {
+	Dtid         string
+	State        string
+	Created      time.Time
+	Participants []querypb.Target
+}
+
 // ReadAllTransactions returns info about all distributed transactions.
-func (tpc *TwoPC) ReadAllTransactions(ctx context.Context) ([]*tx.DistributedTx, error) {
+func (tpc *TwoPC) ReadAllTransactions(ctx context.Context) ([]*DistributedTx, error) {
 	conn, err := tpc.readPool.Get(ctx)
 	if err != nil {
 		return nil, err
@@ -482,8 +496,8 @@ func (tpc *TwoPC) ReadAllTransactions(ctx context.Context) ([]*tx.DistributedTx,
 		return nil, err
 	}
 
-	var curTx *tx.DistributedTx
-	var distributed []*tx.DistributedTx
+	var curTx *DistributedTx
+	var distributed []*DistributedTx
 	for _, row := range qr.Rows {
 		dtid := row[0].ToString()
 		if curTx == nil || dtid != curTx.Dtid {
@@ -501,7 +515,7 @@ func (tpc *TwoPC) ReadAllTransactions(ctx context.Context) ([]*tx.DistributedTx,
 			if protostate < querypb.TransactionState_UNKNOWN || protostate > querypb.TransactionState_ROLLBACK {
 				log.Errorf("Unexpected state for dtid %s: %v.", dtid, protostate)
 			}
-			curTx = &tx.DistributedTx{
+			curTx = &DistributedTx{
 				Dtid:    dtid,
 				State:   querypb.TransactionState(st).String(),
 				Created: time.Unix(0, tm),
@@ -516,7 +530,7 @@ func (tpc *TwoPC) ReadAllTransactions(ctx context.Context) ([]*tx.DistributedTx,
 	return distributed, nil
 }
 
-func (tpc *TwoPC) exec(ctx context.Context, conn tx.IStatefulConnection, pq *sqlparser.ParsedQuery, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
+func (tpc *TwoPC) exec(ctx context.Context, conn *TxConnection, pq *sqlparser.ParsedQuery, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
 	q, err := pq.GenerateQuery(bindVars, nil)
 	if err != nil {
 		return nil, err
diff --git a/go/vt/vttablet/tabletserver/twopcz.go b/go/vt/vttablet/tabletserver/twopcz.go
index dd8768844c..6f93d05e58 100644
--- a/go/vt/vttablet/tabletserver/twopcz.go
+++ b/go/vt/vttablet/tabletserver/twopcz.go
@@ -22,8 +22,6 @@ import (
 	"html/template"
 	"net/http"
 
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
 	"vitess.io/vitess/go/acl"
 	"vitess.io/vitess/go/vt/log"
 )
@@ -162,8 +160,8 @@ func twopczHandler(txe *TxExecutor, w http.ResponseWriter, r *http.Request) {
 	if format == "json" {
 		w.Header().Set("Content-Type", "application/json")
 		js, err := json.Marshal(struct {
-			Distributed      []*tx.DistributedTx
-			Prepared, Failed []*tx.PreparedTx
+			Distributed      []*DistributedTx
+			Prepared, Failed []*PreparedTx
 		}{
 			Distributed: distributed,
 			Prepared:    prepared,
diff --git a/go/vt/vttablet/tabletserver/tx/api.go b/go/vt/vttablet/tabletserver/tx/api.go
deleted file mode 100644
index 5b98f97fba..0000000000
--- a/go/vt/vttablet/tabletserver/tx/api.go
+++ /dev/null
@@ -1,170 +0,0 @@
-/*
-Copyright 2020 The Vitess Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package tx
-
-import (
-	"context"
-	"fmt"
-	"strings"
-	"time"
-
-	"vitess.io/vitess/go/sqltypes"
-	querypb "vitess.io/vitess/go/vt/proto/query"
-	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
-	"vitess.io/vitess/go/vt/servenv"
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/connpool"
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
-)
-
-type (
-	// ConnID as type int64
-	ConnID = int64
-
-	//DTID as type string
-	DTID = string
-
-	//EngineStateMachine is used to control the state the transactional engine -
-	//whether new connections and/or transactions are allowed or not.
-	EngineStateMachine interface {
-		Init() error
-		AcceptReadWrite() error
-		AcceptReadOnly() error
-		StopGently()
-	}
-
-	//IStatefulConnection is a connection where the user is trusted to clean things up after using the connection
-	IStatefulConnection interface {
-		// Executes a query on the connection
-		Exec(ctx context.Context, query string, maxrows int, wantfields bool) (*sqltypes.Result, error)
-
-		// Release is used after we are done with the connection and will not use it again
-		Release(reason ReleaseReason)
-
-		// Releasef is used after we are done with the connection and will not use it again
-		Releasef(format string, params ...interface{})
-
-		// Unlock marks the connection as not in use. The connection remains active.
-		Unlock()
-
-		IsInTransaction() bool
-
-		Close()
-
-		IsClosed() bool
-
-		String() string
-
-		TxProperties() *Properties
-
-		ID() ConnID
-
-		UnderlyingdDBConn() *connpool.DBConn
-
-		CleanTxState()
-
-		Stats() *tabletenv.Stats
-	}
-
-	// ReleaseReason as type int
-	ReleaseReason int
-
-	//Properties contains all information that is related to the currently running
-	//transaction on the connection
-	Properties struct {
-		EffectiveCaller *vtrpcpb.CallerID
-		ImmediateCaller *querypb.VTGateCallerID
-		StartTime       time.Time
-		EndTime         time.Time
-		Queries         []string
-		Autocommit      bool
-		Conclusion      string
-		LogToFile       bool
-
-		Stats *servenv.TimingsWrapper
-	}
-)
-
-const (
-	// TxClose - connection released on close.
-	TxClose ReleaseReason = iota
-
-	// TxCommit - connection released on commit.
-	TxCommit
-
-	// TxRollback - connection released on rollback.
-	TxRollback
-
-	// TxKill - connection released on tx kill.
-	TxKill
-
-	// ConnInitFail - connection released on failed to start tx.
-	ConnInitFail
-)
-
-func (r ReleaseReason) String() string {
-	return txResolutions[r]
-}
-
-//Name return the name of enum.
-func (r ReleaseReason) Name() string {
-	return txNames[r]
-}
-
-var txResolutions = map[ReleaseReason]string{
-	TxClose:      "closed",
-	TxCommit:     "transaction committed",
-	TxRollback:   "transaction rolled back",
-	TxKill:       "kill",
-	ConnInitFail: "initFail",
-}
-
-var txNames = map[ReleaseReason]string{
-	TxClose:      "close",
-	TxCommit:     "commit",
-	TxRollback:   "rollback",
-	TxKill:       "kill",
-	ConnInitFail: "initFail",
-}
-
-// RecordQuery records the query against this transaction.
-func (p *Properties) RecordQuery(query string) {
-	if p == nil {
-		return
-	}
-	p.Queries = append(p.Queries, query)
-}
-
-// InTransaction returns true as soon as this struct is not nil
-func (p *Properties) InTransaction() bool { return p != nil }
-
-// String returns a printable version of the transaction
-func (p *Properties) String() string {
-	if p == nil {
-		return ""
-	}
-
-	return fmt.Sprintf(
-		"'%v'\t'%v'\t%v\t%v\t%.6f\t%v\t%v\t\n",
-		p.EffectiveCaller,
-		p.ImmediateCaller,
-		p.StartTime.Format(time.StampMicro),
-		p.EndTime.Format(time.StampMicro),
-		p.EndTime.Sub(p.StartTime).Seconds(),
-		p.Conclusion,
-		strings.Join(p.Queries, ";"),
-	)
-}
diff --git a/go/vt/vttablet/tabletserver/tx/twopc.go b/go/vt/vttablet/tabletserver/tx/twopc.go
deleted file mode 100644
index 56cfbd1a51..0000000000
--- a/go/vt/vttablet/tabletserver/tx/twopc.go
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
-Copyright 2020 The Vitess Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package tx
-
-import (
-	"time"
-
-	querypb "vitess.io/vitess/go/vt/proto/query"
-)
-
-// DistributedTx is similar to querypb.TransactionMetadata, but
-// is display friendly.
-type DistributedTx struct {
-	Dtid         string
-	State        string
-	Created      time.Time
-	Participants []querypb.Target
-}
-
-// PreparedTx represents a displayable version of a prepared transaction.
-type PreparedTx struct {
-	Dtid    string
-	Queries []string
-	Time    time.Time
-}
diff --git a/go/vt/vttablet/tabletserver/tx_engine.go b/go/vt/vttablet/tabletserver/tx_engine.go
index c6580b199e..51ede88c3e 100644
--- a/go/vt/vttablet/tabletserver/tx_engine.go
+++ b/go/vt/vttablet/tabletserver/tx_engine.go
@@ -21,8 +21,6 @@ import (
 	"sync"
 	"time"
 
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
 	"golang.org/x/net/context"
 
 	"vitess.io/vitess/go/timer"
@@ -142,6 +140,39 @@ func NewTxEngine(env tabletenv.Env) *TxEngine {
 	return te
 }
 
+// Stop will stop accepting any new transactions. Transactions are immediately aborted.
+func (te *TxEngine) Stop() error {
+	te.beginRequests.Wait()
+	te.stateLock.Lock()
+
+	switch te.state {
+	case NotServing:
+		// Nothing to do. We are already stopped or stopping
+		te.stateLock.Unlock()
+		return nil
+
+	case AcceptingReadAndWrite:
+		return te.transitionTo(NotServing)
+
+	case AcceptingReadOnly:
+		// We are not master, so it's safe to kill all read-only transactions
+		te.close(true)
+		te.state = NotServing
+		te.stateLock.Unlock()
+		return nil
+
+	case Transitioning:
+		te.nextState = NotServing
+		te.stateLock.Unlock()
+		te.blockUntilEndOfTransition()
+		return nil
+
+	default:
+		te.stateLock.Unlock()
+		return te.unknownStateError()
+	}
+}
+
 // AcceptReadWrite will start accepting all transactions.
 // If transitioning from RO mode, transactions might need to be
 // rolled back before new transactions can be accepts.
@@ -241,28 +272,14 @@ func (te *TxEngine) Begin(ctx context.Context, options *querypb.ExecuteOptions)
 	te.stateLock.Unlock()
 
 	defer te.beginRequests.Done()
-	conn, beginSQL, err := te.txPool.Begin(ctx, options)
-	if err != nil {
-		return 0, "", err
-	}
-	defer conn.Unlock()
-	return conn.ID(), beginSQL, err
+	return te.txPool.Begin(ctx, options)
 }
 
 // Commit commits the specified transaction.
 func (te *TxEngine) Commit(ctx context.Context, transactionID int64) (string, error) {
 	span, ctx := trace.NewSpan(ctx, "TxEngine.Commit")
 	defer span.Finish()
-	conn, err := te.txPool.GetAndLock(transactionID, "for commit")
-	if err != nil {
-		return "", err
-	}
-	defer conn.Release(tx.TxCommit)
-	queries, err := te.txPool.Commit(ctx, conn)
-	if err != nil {
-		return "", err
-	}
-	return queries, nil
+	return te.txPool.Commit(ctx, transactionID)
 }
 
 // Rollback rolls back the specified transaction.
@@ -270,12 +287,7 @@ func (te *TxEngine) Rollback(ctx context.Context, transactionID int64) error {
 	span, ctx := trace.NewSpan(ctx, "TxEngine.Rollback")
 	defer span.Finish()
 
-	conn, err := te.txPool.GetAndLock(transactionID, "for rollback")
-	if err != nil {
-		return err
-	}
-	defer conn.Release(tx.TxRollback)
-	return te.txPool.Rollback(ctx, conn)
+	return te.txPool.Rollback(ctx, transactionID)
 }
 
 func (te *TxEngine) unknownStateError() error {
@@ -430,45 +442,45 @@ func (te *TxEngine) prepareFromRedo() error {
 
 	maxid := int64(0)
 outer:
-	for _, preparedTx := range prepared {
-		txid, err := dtids.TransactionID(preparedTx.Dtid)
+	for _, tx := range prepared {
+		txid, err := dtids.TransactionID(tx.Dtid)
 		if err != nil {
 			log.Errorf("Error extracting transaction ID from ditd: %v", err)
 		}
 		if txid > maxid {
 			maxid = txid
 		}
-		conn, _, err := te.txPool.Begin(ctx, &querypb.ExecuteOptions{})
+		conn, _, err := te.txPool.LocalBegin(ctx, &querypb.ExecuteOptions{})
 		if err != nil {
 			allErr.RecordError(err)
 			continue
 		}
-		for _, stmt := range preparedTx.Queries {
-			conn.TxProperties().RecordQuery(stmt)
+		for _, stmt := range tx.Queries {
+			conn.RecordQuery(stmt)
 			_, err := conn.Exec(ctx, stmt, 1, false)
 			if err != nil {
 				allErr.RecordError(err)
-				te.txPool.RollbackAndRelease(ctx, conn)
+				te.txPool.LocalConclude(ctx, conn)
 				continue outer
 			}
 		}
 		// We should not use the external Prepare because
 		// we don't want to write again to the redo log.
-		err = te.preparedPool.Put(conn, preparedTx.Dtid)
+		err = te.preparedPool.Put(conn, tx.Dtid)
 		if err != nil {
 			allErr.RecordError(err)
 			continue
 		}
 	}
-	for _, preparedTx := range failed {
-		txid, err := dtids.TransactionID(preparedTx.Dtid)
+	for _, tx := range failed {
+		txid, err := dtids.TransactionID(tx.Dtid)
 		if err != nil {
 			log.Errorf("Error extracting transaction ID from ditd: %v", err)
 		}
 		if txid > maxid {
 			maxid = txid
 		}
-		te.preparedPool.SetFailed(preparedTx.Dtid)
+		te.preparedPool.SetFailed(tx.Dtid)
 	}
 	te.txPool.AdjustLastID(maxid)
 	log.Infof("Prepared %d transactions, and registered %d failures.", len(prepared), len(failed))
@@ -480,8 +492,10 @@ outer:
 // This is used for transitioning from a master to a non-master
 // serving type.
 func (te *TxEngine) rollbackTransactions() {
-	te.rollbackPrepared()
 	ctx := tabletenv.LocalContext()
+	for _, c := range te.preparedPool.FetchAll() {
+		te.txPool.LocalConclude(ctx, c)
+	}
 	// The order of rollbacks is currently not material because
 	// we don't allow new statements or commits during
 	// this function. In case of any such change, this will
@@ -491,9 +505,8 @@ func (te *TxEngine) rollbackTransactions() {
 
 func (te *TxEngine) rollbackPrepared() {
 	ctx := tabletenv.LocalContext()
-	for _, conn := range te.preparedPool.FetchAll() {
-		te.txPool.Rollback(ctx, conn)
-		conn.Release(tx.TxRollback)
+	for _, c := range te.preparedPool.FetchAll() {
+		te.txPool.LocalConclude(ctx, c)
 	}
 }
 
diff --git a/go/vt/vttablet/tabletserver/tx_executor.go b/go/vt/vttablet/tabletserver/tx_executor.go
index 3eb067f099..351de1dd80 100644
--- a/go/vt/vttablet/tabletserver/tx_executor.go
+++ b/go/vt/vttablet/tabletserver/tx_executor.go
@@ -19,8 +19,6 @@ package tabletserver
 import (
 	"time"
 
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
 	"golang.org/x/net/context"
 
 	"vitess.io/vitess/go/trace"
@@ -33,7 +31,6 @@ import (
 )
 
 // TxExecutor is used for executing a transactional request.
-// TODO: merge this with tx_engine
 type TxExecutor struct {
 	// TODO(sougou): Parameterize this.
 	ctx      context.Context
@@ -52,27 +49,40 @@ func (txe *TxExecutor) Prepare(transactionID int64, dtid string) error {
 	defer txe.te.env.Stats().QueryTimings.Record("PREPARE", time.Now())
 	txe.logStats.TransactionID = transactionID
 
-	conn, err := txe.te.txPool.GetAndLock(transactionID, "for prepare")
+	conn, err := txe.te.txPool.Get(transactionID, "for prepare")
 	if err != nil {
 		return err
 	}
 
 	// If no queries were executed, we just rollback.
-	if len(conn.TxProperties().Queries) == 0 {
-		conn.Release(tx.TxRollback)
+	if len(conn.Queries) == 0 {
+		txe.te.txPool.LocalConclude(txe.ctx, conn)
 		return nil
 	}
 
 	err = txe.te.preparedPool.Put(conn, dtid)
 	if err != nil {
-		txe.te.txPool.RollbackAndRelease(txe.ctx, conn)
+		txe.te.txPool.localRollback(txe.ctx, conn)
 		return vterrors.Errorf(vtrpcpb.Code_RESOURCE_EXHAUSTED, "prepare failed for transaction %d: %v", transactionID, err)
 	}
 
-	return txe.inTransaction(func(localConn tx.IStatefulConnection) error {
-		return txe.te.twoPC.SaveRedo(txe.ctx, localConn, dtid, conn.TxProperties().Queries)
-	})
+	localConn, _, err := txe.te.txPool.LocalBegin(txe.ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		return err
+	}
+	defer txe.te.txPool.LocalConclude(txe.ctx, localConn)
 
+	err = txe.te.twoPC.SaveRedo(txe.ctx, localConn, dtid, conn.Queries)
+	if err != nil {
+		return err
+	}
+
+	_, err = txe.te.txPool.LocalCommit(txe.ctx, localConn)
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // CommitPrepared commits a prepared transaction. If the operation
@@ -93,13 +103,13 @@ func (txe *TxExecutor) CommitPrepared(dtid string) error {
 	// We have to use a context that will never give up,
 	// even if the original context expires.
 	ctx := trace.CopySpan(context.Background(), txe.ctx)
-	defer txe.te.txPool.RollbackAndRelease(ctx, conn)
+	defer txe.te.txPool.LocalConclude(ctx, conn)
 	err = txe.te.twoPC.DeleteRedo(ctx, conn, dtid)
 	if err != nil {
 		txe.markFailed(ctx, dtid)
 		return err
 	}
-	_, err = txe.te.txPool.Commit(ctx, conn)
+	_, err = txe.te.txPool.LocalCommit(ctx, conn)
 	if err != nil {
 		txe.markFailed(ctx, dtid)
 		return err
@@ -118,19 +128,19 @@ func (txe *TxExecutor) CommitPrepared(dtid string) error {
 func (txe *TxExecutor) markFailed(ctx context.Context, dtid string) {
 	txe.te.env.Stats().InternalErrors.Add("TwopcCommit", 1)
 	txe.te.preparedPool.SetFailed(dtid)
-	conn, _, err := txe.te.txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	conn, _, err := txe.te.txPool.LocalBegin(ctx, &querypb.ExecuteOptions{})
 	if err != nil {
 		log.Errorf("markFailed: Begin failed for dtid %s: %v", dtid, err)
 		return
 	}
-	defer txe.te.txPool.RollbackAndRelease(ctx, conn)
+	defer txe.te.txPool.LocalConclude(ctx, conn)
 
 	if err = txe.te.twoPC.UpdateRedo(ctx, conn, dtid, RedoStateFailed); err != nil {
 		log.Errorf("markFailed: UpdateRedo failed for dtid %s: %v", dtid, err)
 		return
 	}
 
-	if _, err = txe.te.txPool.Commit(ctx, conn); err != nil {
+	if _, err = txe.te.txPool.LocalCommit(ctx, conn); err != nil {
 		log.Errorf("markFailed: Commit failed for dtid %s: %v", dtid, err)
 	}
 }
@@ -158,17 +168,28 @@ func (txe *TxExecutor) RollbackPrepared(dtid string, originalID int64) error {
 		return vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "2pc is not enabled")
 	}
 	defer txe.te.env.Stats().QueryTimings.Record("ROLLBACK_PREPARED", time.Now())
-	defer func() {
-		if preparedConn := txe.te.preparedPool.FetchForRollback(dtid); preparedConn != nil {
-			txe.te.txPool.RollbackAndRelease(txe.ctx, preparedConn)
-		}
-		if originalID != 0 {
-			txe.te.Rollback(txe.ctx, originalID)
-		}
-	}()
-	return txe.inTransaction(func(conn tx.IStatefulConnection) error {
-		return txe.te.twoPC.DeleteRedo(txe.ctx, conn, dtid)
-	})
+	conn, _, err := txe.te.txPool.LocalBegin(txe.ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		goto returnConn
+	}
+	defer txe.te.txPool.LocalConclude(txe.ctx, conn)
+
+	err = txe.te.twoPC.DeleteRedo(txe.ctx, conn, dtid)
+	if err != nil {
+		goto returnConn
+	}
+
+	_, err = txe.te.txPool.LocalCommit(txe.ctx, conn)
+
+returnConn:
+	if preparedConn := txe.te.preparedPool.FetchForRollback(dtid); preparedConn != nil {
+		txe.te.txPool.LocalConclude(txe.ctx, preparedConn)
+	}
+	if originalID != 0 {
+		txe.te.txPool.Rollback(txe.ctx, originalID)
+	}
+
+	return err
 }
 
 // CreateTransaction creates the metadata for a 2PC transaction.
@@ -177,9 +198,18 @@ func (txe *TxExecutor) CreateTransaction(dtid string, participants []*querypb.Ta
 		return vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "2pc is not enabled")
 	}
 	defer txe.te.env.Stats().QueryTimings.Record("CREATE_TRANSACTION", time.Now())
-	return txe.inTransaction(func(conn tx.IStatefulConnection) error {
-		return txe.te.twoPC.CreateTransaction(txe.ctx, conn, dtid, participants)
-	})
+	conn, _, err := txe.te.txPool.LocalBegin(txe.ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		return err
+	}
+	defer txe.te.txPool.LocalConclude(txe.ctx, conn)
+
+	err = txe.te.twoPC.CreateTransaction(txe.ctx, conn, dtid, participants)
+	if err != nil {
+		return err
+	}
+	_, err = txe.te.txPool.LocalCommit(txe.ctx, conn)
+	return err
 }
 
 // StartCommit atomically commits the transaction along with the
@@ -191,17 +221,17 @@ func (txe *TxExecutor) StartCommit(transactionID int64, dtid string) error {
 	defer txe.te.env.Stats().QueryTimings.Record("START_COMMIT", time.Now())
 	txe.logStats.TransactionID = transactionID
 
-	conn, err := txe.te.txPool.GetAndLock(transactionID, "for 2pc commit")
+	conn, err := txe.te.txPool.Get(transactionID, "for 2pc commit")
 	if err != nil {
 		return err
 	}
-	defer txe.te.txPool.RollbackAndRelease(txe.ctx, conn)
+	defer txe.te.txPool.LocalConclude(txe.ctx, conn)
 
 	err = txe.te.twoPC.Transition(txe.ctx, conn, dtid, querypb.TransactionState_COMMIT)
 	if err != nil {
 		return err
 	}
-	_, err = txe.te.txPool.Commit(txe.ctx, conn)
+	_, err = txe.te.txPool.LocalCommit(txe.ctx, conn)
 	return err
 }
 
@@ -215,12 +245,26 @@ func (txe *TxExecutor) SetRollback(dtid string, transactionID int64) error {
 	txe.logStats.TransactionID = transactionID
 
 	if transactionID != 0 {
-		txe.te.Rollback(txe.ctx, transactionID)
+		txe.te.txPool.Rollback(txe.ctx, transactionID)
 	}
 
-	return txe.inTransaction(func(conn tx.IStatefulConnection) error {
-		return txe.te.twoPC.Transition(txe.ctx, conn, dtid, querypb.TransactionState_ROLLBACK)
-	})
+	conn, _, err := txe.te.txPool.LocalBegin(txe.ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		return err
+	}
+	defer txe.te.txPool.LocalConclude(txe.ctx, conn)
+
+	err = txe.te.twoPC.Transition(txe.ctx, conn, dtid, querypb.TransactionState_ROLLBACK)
+	if err != nil {
+		return err
+	}
+
+	_, err = txe.te.txPool.LocalCommit(txe.ctx, conn)
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // ConcludeTransaction deletes the 2pc transaction metadata
@@ -231,9 +275,18 @@ func (txe *TxExecutor) ConcludeTransaction(dtid string) error {
 	}
 	defer txe.te.env.Stats().QueryTimings.Record("RESOLVE", time.Now())
 
-	return txe.inTransaction(func(conn tx.IStatefulConnection) error {
-		return txe.te.twoPC.DeleteTransaction(txe.ctx, conn, dtid)
-	})
+	conn, _, err := txe.te.txPool.LocalBegin(txe.ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		return err
+	}
+	defer txe.te.txPool.LocalConclude(txe.ctx, conn)
+
+	err = txe.te.twoPC.DeleteTransaction(txe.ctx, conn, dtid)
+	if err != nil {
+		return err
+	}
+	_, err = txe.te.txPool.LocalCommit(txe.ctx, conn)
+	return err
 }
 
 // ReadTransaction returns the metadata for the sepcified dtid.
@@ -245,7 +298,7 @@ func (txe *TxExecutor) ReadTransaction(dtid string) (*querypb.TransactionMetadat
 }
 
 // ReadTwopcInflight returns info about all in-flight 2pc transactions.
-func (txe *TxExecutor) ReadTwopcInflight() (distributed []*tx.DistributedTx, prepared, failed []*tx.PreparedTx, err error) {
+func (txe *TxExecutor) ReadTwopcInflight() (distributed []*DistributedTx, prepared, failed []*PreparedTx, err error) {
 	if !txe.te.twopcEnabled {
 		return nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "2pc is not enabled")
 	}
@@ -259,22 +312,3 @@ func (txe *TxExecutor) ReadTwopcInflight() (distributed []*tx.DistributedTx, pre
 	}
 	return distributed, prepared, failed, nil
 }
-
-func (txe *TxExecutor) inTransaction(f func(tx.IStatefulConnection) error) error {
-	conn, _, err := txe.te.txPool.Begin(txe.ctx, &querypb.ExecuteOptions{})
-	if err != nil {
-		return err
-	}
-	defer txe.te.txPool.RollbackAndRelease(txe.ctx, conn)
-
-	err = f(conn)
-	if err != nil {
-		return err
-	}
-
-	_, err = txe.te.txPool.Commit(txe.ctx, conn)
-	if err != nil {
-		return err
-	}
-	return nil
-}
diff --git a/go/vt/vttablet/tabletserver/tx_pool.go b/go/vt/vttablet/tabletserver/tx_pool.go
index 27849d5da0..ac03877ef4 100644
--- a/go/vt/vttablet/tabletserver/tx_pool.go
+++ b/go/vt/vttablet/tabletserver/tx_pool.go
@@ -17,24 +17,25 @@ limitations under the License.
 package tabletserver
 
 import (
+	"fmt"
+	"strings"
 	"sync"
 	"time"
 
-	"vitess.io/vitess/go/pools"
-
-	"vitess.io/vitess/go/vt/servenv"
-
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
 	"golang.org/x/net/context"
 
+	"vitess.io/vitess/go/mysql"
+	"vitess.io/vitess/go/pools"
+	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/sync2"
 	"vitess.io/vitess/go/timer"
 	"vitess.io/vitess/go/trace"
 	"vitess.io/vitess/go/vt/callerid"
 	"vitess.io/vitess/go/vt/dbconfigs"
 	"vitess.io/vitess/go/vt/log"
+	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/vterrors"
+	"vitess.io/vitess/go/vt/vttablet/tabletserver/connpool"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/txlimiter"
 
@@ -42,45 +43,68 @@ import (
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 )
 
+// These consts identify how a transaction was resolved.
+const (
+	TxClose    = "close"
+	TxCommit   = "commit"
+	TxRollback = "rollback"
+	TxKill     = "kill"
+)
+
 const txLogInterval = 1 * time.Minute
 
-var txIsolations = map[querypb.ExecuteOptions_TransactionIsolation]queries{
-	querypb.ExecuteOptions_DEFAULT:                       {setIsolationLevel: "", openTransaction: "begin"},
-	querypb.ExecuteOptions_REPEATABLE_READ:               {setIsolationLevel: "REPEATABLE READ", openTransaction: "begin"},
-	querypb.ExecuteOptions_READ_COMMITTED:                {setIsolationLevel: "READ COMMITTED", openTransaction: "begin"},
-	querypb.ExecuteOptions_READ_UNCOMMITTED:              {setIsolationLevel: "READ UNCOMMITTED", openTransaction: "begin"},
-	querypb.ExecuteOptions_SERIALIZABLE:                  {setIsolationLevel: "SERIALIZABLE", openTransaction: "begin"},
-	querypb.ExecuteOptions_CONSISTENT_SNAPSHOT_READ_ONLY: {setIsolationLevel: "REPEATABLE READ", openTransaction: "start transaction with consistent snapshot, read only"},
+type queries struct {
+	setIsolationLevel string
+	openTransaction   string
 }
 
-type (
-	// TxPool does a lot of the transactional operations on StatefulConnections. It does not, with two exceptions,
-	// concern itself with a connections life cycle. The two exceptions are Begin, which creates a new StatefulConnection,
-	// and RollbackAndRelease, which does a Release after doing the rollback.
-	TxPool struct {
-		env                tabletenv.Env
-		scp                *StatefulConnectionPool
-		transactionTimeout sync2.AtomicDuration
-		ticks              *timer.Timer
-		limiter            txlimiter.TxLimiter
-
-		logMu   sync.Mutex
-		lastLog time.Time
-		txStats *servenv.TimingsWrapper
-	}
-	queries struct {
-		setIsolationLevel string
-		openTransaction   string
+var (
+	txIsolations = map[querypb.ExecuteOptions_TransactionIsolation]queries{
+		querypb.ExecuteOptions_DEFAULT:                       {setIsolationLevel: "", openTransaction: "begin"},
+		querypb.ExecuteOptions_REPEATABLE_READ:               {setIsolationLevel: "REPEATABLE READ", openTransaction: "begin"},
+		querypb.ExecuteOptions_READ_COMMITTED:                {setIsolationLevel: "READ COMMITTED", openTransaction: "begin"},
+		querypb.ExecuteOptions_READ_UNCOMMITTED:              {setIsolationLevel: "READ UNCOMMITTED", openTransaction: "begin"},
+		querypb.ExecuteOptions_SERIALIZABLE:                  {setIsolationLevel: "SERIALIZABLE", openTransaction: "begin"},
+		querypb.ExecuteOptions_CONSISTENT_SNAPSHOT_READ_ONLY: {setIsolationLevel: "REPEATABLE READ", openTransaction: "start transaction with consistent snapshot, read only"},
 	}
 )
 
+// TxPool is the transaction pool for the query service.
+type TxPool struct {
+	env tabletenv.Env
+
+	// conns is the 'regular' pool. By default, connections
+	// are pulled from here for starting transactions.
+	conns *connpool.Pool
+
+	// foundRowsPool is the alternate pool that creates
+	// connections with CLIENT_FOUND_ROWS flag set. A separate
+	// pool is needed because this option can only be set at
+	// connection time.
+	foundRowsPool      *connpool.Pool
+	activePool         *pools.Numbered
+	lastID             sync2.AtomicInt64
+	transactionTimeout sync2.AtomicDuration
+	ticks              *timer.Timer
+	limiter            txlimiter.TxLimiter
+
+	txStats *servenv.TimingsWrapper
+
+	// Tracking culprits that cause tx pool full errors.
+	logMu   sync.Mutex
+	lastLog time.Time
+}
+
 // NewTxPool creates a new TxPool. It's not operational until it's Open'd.
 func NewTxPool(env tabletenv.Env, limiter txlimiter.TxLimiter) *TxPool {
 	config := env.Config()
 	transactionTimeout := time.Duration(config.Oltp.TxTimeoutSeconds * 1e9)
 	axp := &TxPool{
 		env:                env,
-		scp:                NewStatefulConnPool(env),
+		conns:              connpool.NewPool(env, "TransactionPool", config.TxPool),
+		foundRowsPool:      connpool.NewPool(env, "FoundRowsPool", config.TxPool),
+		activePool:         pools.NewNumbered(),
+		lastID:             sync2.NewAtomicInt64(time.Now().UnixNano()),
 		transactionTimeout: sync2.NewAtomicDuration(transactionTimeout),
 		ticks:              timer.NewTimer(transactionTimeout / 10),
 		limiter:            limiter,
@@ -95,172 +119,237 @@ func NewTxPool(env tabletenv.Env, limiter txlimiter.TxLimiter) *TxPool {
 // Open makes the TxPool operational. This also starts the transaction killer
 // that will kill long-running transactions.
 func (tp *TxPool) Open(appParams, dbaParams, appDebugParams dbconfigs.Connector) {
-	tp.scp.Open(appParams, dbaParams, appDebugParams)
+	log.Infof("Starting transaction id: %d", tp.lastID)
+	tp.conns.Open(appParams, dbaParams, appDebugParams)
+	foundRowsParam, _ := appParams.MysqlParams()
+	foundRowsParam.EnableClientFoundRows()
+	appParams = dbconfigs.New(foundRowsParam)
+	tp.foundRowsPool.Open(appParams, dbaParams, appDebugParams)
 	tp.ticks.Start(func() { tp.transactionKiller() })
 }
 
 // Close closes the TxPool. A closed pool can be reopened.
 func (tp *TxPool) Close() {
 	tp.ticks.Stop()
-	tp.scp.Close()
+	for _, v := range tp.activePool.GetOutdated(time.Duration(0), "for closing") {
+		conn := v.(*TxConnection)
+		log.Warningf("killing transaction for shutdown: %s", conn.Format())
+		tp.env.Stats().InternalErrors.Add("StrayTransactions", 1)
+		conn.Close()
+		conn.conclude(TxClose, "pool closed")
+	}
+	tp.conns.Close()
+	tp.foundRowsPool.Close()
 }
 
 // AdjustLastID adjusts the last transaction id to be at least
 // as large as the input value. This will ensure that there are
 // no dtid collisions with future transactions.
 func (tp *TxPool) AdjustLastID(id int64) {
-	tp.scp.AdjustLastID(id)
+	if current := tp.lastID.Get(); current < id {
+		log.Infof("Adjusting transaction id to: %d", id)
+		tp.lastID.Set(id)
+	}
 }
 
 // RollbackNonBusy rolls back all transactions that are not in use.
 // Transactions can be in use for situations like executing statements
 // or in prepared state.
 func (tp *TxPool) RollbackNonBusy(ctx context.Context) {
-	for _, v := range tp.scp.GetOutdated(time.Duration(0), "for transition") {
-		tp.RollbackAndRelease(ctx, v)
+	for _, v := range tp.activePool.GetOutdated(time.Duration(0), "for transition") {
+		tp.LocalConclude(ctx, v.(*TxConnection))
 	}
 }
 
 func (tp *TxPool) transactionKiller() {
 	defer tp.env.LogError()
-	for _, conn := range tp.scp.GetOutdated(tp.Timeout(), "for tx killer rollback") {
-		log.Warningf("killing transaction (exceeded timeout: %v): %s", tp.Timeout(), conn.String())
+	for _, v := range tp.activePool.GetOutdated(tp.Timeout(), "for tx killer rollback") {
+		conn := v.(*TxConnection)
+		log.Warningf("killing transaction (exceeded timeout: %v): %s", tp.Timeout(), conn.Format())
 		tp.env.Stats().KillCounters.Add("Transactions", 1)
 		conn.Close()
-		conn.Releasef("exceeded timeout: %v", tp.Timeout())
+		conn.conclude(TxKill, fmt.Sprintf("exceeded timeout: %v", tp.Timeout()))
 	}
 }
 
 // WaitForEmpty waits until all active transactions are completed.
 func (tp *TxPool) WaitForEmpty() {
-	tp.scp.WaitForEmpty()
+	tp.activePool.WaitForEmpty()
 }
 
-//NewTxProps creates a new TxProperties struct
-func (tp *TxPool) NewTxProps(immediateCaller *querypb.VTGateCallerID, effectiveCaller *vtrpcpb.CallerID, autocommit bool) *tx.Properties {
-	return &tx.Properties{
-		StartTime:       time.Now(),
-		EffectiveCaller: effectiveCaller,
-		ImmediateCaller: immediateCaller,
-		Autocommit:      autocommit,
-		Stats:           tp.txStats,
+// Begin begins a transaction, and returns the associated transaction id and
+// the statements (if any) executed to initiate the transaction. In autocommit
+// mode the statement will be "".
+//
+// Subsequent statements can access the connection through the transaction id.
+func (tp *TxPool) Begin(ctx context.Context, options *querypb.ExecuteOptions) (int64, string, error) {
+	span, ctx := trace.NewSpan(ctx, "TxPool.Begin")
+	defer span.Finish()
+	var conn *connpool.DBConn
+	var err error
+	immediateCaller := callerid.ImmediateCallerIDFromContext(ctx)
+	effectiveCaller := callerid.EffectiveCallerIDFromContext(ctx)
+
+	if !tp.limiter.Get(immediateCaller, effectiveCaller) {
+		return 0, "", vterrors.Errorf(vtrpcpb.Code_RESOURCE_EXHAUSTED, "per-user transaction pool connection limit exceeded")
 	}
-}
 
-// GetAndLock fetches the connection associated to the transactionID and blocks it from concurrent use
-// You must call Unlock on TxConnection once done.
-func (tp *TxPool) GetAndLock(connID tx.ConnID, reason string) (tx.IStatefulConnection, error) {
-	conn, err := tp.scp.GetAndLock(connID, reason)
+	var beginSucceeded bool
+	defer func() {
+		if beginSucceeded {
+			return
+		}
+
+		if conn != nil {
+			conn.Recycle()
+		}
+		tp.limiter.Release(immediateCaller, effectiveCaller)
+	}()
+
+	if options.GetClientFoundRows() {
+		conn, err = tp.foundRowsPool.Get(ctx)
+	} else {
+		conn, err = tp.conns.Get(ctx)
+	}
 	if err != nil {
-		return nil, vterrors.Errorf(vtrpcpb.Code_ABORTED, "transaction %d: %v", connID, err)
+		switch err {
+		case connpool.ErrConnPoolClosed:
+			return 0, "", err
+		case pools.ErrCtxTimeout:
+			tp.LogActive()
+			return 0, "", vterrors.Errorf(vtrpcpb.Code_RESOURCE_EXHAUSTED, "transaction pool aborting request due to already expired context")
+		case pools.ErrTimeout:
+			tp.LogActive()
+			return 0, "", vterrors.Errorf(vtrpcpb.Code_RESOURCE_EXHAUSTED, "transaction pool connection limit exceeded")
+		}
+		return 0, "", err
 	}
-	return conn, nil
-}
 
-// Commit commits the transaction on the connection.
-func (tp *TxPool) Commit(ctx context.Context, txConn tx.IStatefulConnection) (string, error) {
-	if !txConn.IsInTransaction() {
-		return "", vterrors.New(vtrpcpb.Code_INTERNAL, "not in a transaction")
+	autocommitTransaction := false
+	beginQueries := ""
+	if queries, ok := txIsolations[options.GetTransactionIsolation()]; ok {
+		if queries.setIsolationLevel != "" {
+			if _, err := conn.Exec(ctx, "set transaction isolation level "+queries.setIsolationLevel, 1, false); err != nil {
+				return 0, "", err
+			}
+
+			beginQueries = queries.setIsolationLevel + "; "
+		}
+
+		if _, err := conn.Exec(ctx, queries.openTransaction, 1, false); err != nil {
+			return 0, "", err
+		}
+		beginQueries = beginQueries + queries.openTransaction
+	} else if options.GetTransactionIsolation() == querypb.ExecuteOptions_AUTOCOMMIT {
+		autocommitTransaction = true
+	} else {
+		return 0, "", fmt.Errorf("don't know how to open a transaction of this type: %v", options.GetTransactionIsolation())
 	}
+
+	beginSucceeded = true
+	transactionID := tp.lastID.Add(1)
+	tp.activePool.Register(
+		transactionID,
+		newTxConnection(
+			conn,
+			transactionID,
+			tp,
+			immediateCaller,
+			effectiveCaller,
+			autocommitTransaction,
+		),
+		options.GetWorkload() != querypb.ExecuteOptions_DBA,
+	)
+	return transactionID, beginQueries, nil
+}
+
+// Commit commits the specified transaction.
+func (tp *TxPool) Commit(ctx context.Context, transactionID int64) (string, error) {
 	span, ctx := trace.NewSpan(ctx, "TxPool.Commit")
 	defer span.Finish()
-	defer tp.txComplete(txConn, tx.TxCommit)
-	if txConn.TxProperties().Autocommit {
-		return "", nil
+	conn, err := tp.Get(transactionID, "for commit")
+	if err != nil {
+		return "", err
 	}
+	return tp.LocalCommit(ctx, conn)
+}
 
-	if _, err := txConn.Exec(ctx, "commit", 1, false); err != nil {
-		txConn.Close()
-		return "", err
+// Rollback rolls back the specified transaction.
+func (tp *TxPool) Rollback(ctx context.Context, transactionID int64) error {
+	span, ctx := trace.NewSpan(ctx, "TxPool.Rollback")
+	defer span.Finish()
+
+	conn, err := tp.Get(transactionID, "for rollback")
+	if err != nil {
+		return err
 	}
-	return "commit", nil
+	return tp.localRollback(ctx, conn)
 }
 
-// RollbackAndRelease rolls back the transaction on the specified connection, and releases the connection when done
-func (tp *TxPool) RollbackAndRelease(ctx context.Context, txConn tx.IStatefulConnection) {
-	defer txConn.Release(tx.TxRollback)
-	rollbackError := tp.Rollback(ctx, txConn)
-	if rollbackError != nil {
-		log.Errorf("tried to rollback, but failed with: %v", rollbackError.Error())
+// Get fetches the connection associated to the transactionID.
+// You must call Recycle on TxConnection once done.
+func (tp *TxPool) Get(transactionID int64, reason string) (*TxConnection, error) {
+	v, err := tp.activePool.Get(transactionID, reason)
+	if err != nil {
+		return nil, vterrors.Errorf(vtrpcpb.Code_ABORTED, "transaction %d: %v", transactionID, err)
 	}
+	return v.(*TxConnection), nil
 }
 
-// Rollback rolls back the transaction on the specified connection.
-func (tp *TxPool) Rollback(ctx context.Context, txConn tx.IStatefulConnection) error {
-	span, ctx := trace.NewSpan(ctx, "TxPool.Rollback")
+// LocalBegin is equivalent to Begin->Get.
+// It's used for executing transactions within a request. It's safe
+// to always call LocalConclude at the end.
+func (tp *TxPool) LocalBegin(ctx context.Context, options *querypb.ExecuteOptions) (*TxConnection, string, error) {
+	span, ctx := trace.NewSpan(ctx, "TxPool.LocalBegin")
 	defer span.Finish()
-	if txConn.IsClosed() || !txConn.IsInTransaction() {
-		return nil
-	}
-	if txConn.TxProperties().Autocommit {
-		tp.txComplete(txConn, tx.TxCommit)
-		return nil
-	}
-	defer tp.txComplete(txConn, tx.TxRollback)
-	if _, err := txConn.Exec(ctx, "rollback", 1, false); err != nil {
-		txConn.Close()
-		return err
+
+	transactionID, beginSQL, err := tp.Begin(ctx, options)
+	if err != nil {
+		return nil, "", err
 	}
-	return nil
+	conn, err := tp.Get(transactionID, "for local query")
+	return conn, beginSQL, err
 }
 
-// Begin begins a transaction, and returns the associated connection and
-// the statements (if any) executed to initiate the transaction. In autocommit
-// mode the statement will be "".
-// The connection returned is locked for the callee and its responsibility is to unlock the connection.
-func (tp *TxPool) Begin(ctx context.Context, options *querypb.ExecuteOptions) (tx.IStatefulConnection, string, error) {
-	span, ctx := trace.NewSpan(ctx, "TxPool.Begin")
+// LocalCommit is the commit function for LocalBegin.
+func (tp *TxPool) LocalCommit(ctx context.Context, conn *TxConnection) (string, error) {
+	span, ctx := trace.NewSpan(ctx, "TxPool.LocalCommit")
 	defer span.Finish()
-	beginQueries := ""
+	defer conn.conclude(TxCommit, "transaction committed")
 
-	immediateCaller := callerid.ImmediateCallerIDFromContext(ctx)
-	effectiveCaller := callerid.EffectiveCallerIDFromContext(ctx)
+	if conn.Autocommit {
+		return "", nil
+	}
 
-	if !tp.limiter.Get(immediateCaller, effectiveCaller) {
-		return nil, "", vterrors.Errorf(vtrpcpb.Code_RESOURCE_EXHAUSTED, "per-user transaction pool connection limit exceeded")
+	if _, err := conn.Exec(ctx, "commit", 1, false); err != nil {
+		conn.Close()
+		return "", err
 	}
+	return "commit", nil
+}
 
-	conn, err := tp.scp.NewConn(ctx, options)
-	if err != nil {
-		switch err {
-		case pools.ErrCtxTimeout:
-			tp.LogActive()
-			err = vterrors.Errorf(vtrpcpb.Code_RESOURCE_EXHAUSTED, "transaction pool aborting request due to already expired context")
-		case pools.ErrTimeout:
-			tp.LogActive()
-			err = vterrors.Errorf(vtrpcpb.Code_RESOURCE_EXHAUSTED, "transaction pool connection limit exceeded")
-		}
-		return nil, "", err
+// LocalConclude concludes a transaction started by LocalBegin.
+// If the transaction was not previously concluded, it's rolled back.
+func (tp *TxPool) LocalConclude(ctx context.Context, conn *TxConnection) {
+	if conn.dbConn == nil {
+		return
 	}
-	err = func() error {
-		autocommitTransaction := false
-		if queries, ok := txIsolations[options.GetTransactionIsolation()]; ok {
-			if queries.setIsolationLevel != "" {
-				txQuery := "set transaction isolation level " + queries.setIsolationLevel
-				if err := conn.execWithRetry(ctx, txQuery, 1, false); err != nil {
-					return vterrors.Wrap(err, txQuery)
-				}
-				beginQueries = queries.setIsolationLevel + "; "
-			}
-			if err := conn.execWithRetry(ctx, queries.openTransaction, 1, false); err != nil {
-				return vterrors.Wrap(err, queries.openTransaction)
-			}
-			beginQueries = beginQueries + queries.openTransaction
-		} else if options.GetTransactionIsolation() == querypb.ExecuteOptions_AUTOCOMMIT {
-			autocommitTransaction = true
-		} else {
-			return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "don't know how to open a transaction of this type: %v", options.GetTransactionIsolation())
-		}
-		conn.txProps = tp.NewTxProps(immediateCaller, effectiveCaller, autocommitTransaction)
+	span, ctx := trace.NewSpan(ctx, "TxPool.LocalConclude")
+	defer span.Finish()
+	_ = tp.localRollback(ctx, conn)
+}
+
+func (tp *TxPool) localRollback(ctx context.Context, conn *TxConnection) error {
+	if conn.Autocommit {
+		conn.conclude(TxCommit, "returned to pool")
 		return nil
-	}()
-	if err != nil {
+	}
+	defer conn.conclude(TxRollback, "transaction rolled back")
+	if _, err := conn.Exec(ctx, "rollback", 1, false); err != nil {
 		conn.Close()
-		conn.Release(tx.ConnInitFail)
-		return nil, "", err
+		return err
 	}
-
-	return conn, beginQueries, nil
+	return nil
 }
 
 // LogActive causes all existing transactions to be logged when they complete.
@@ -272,9 +361,10 @@ func (tp *TxPool) LogActive() {
 		return
 	}
 	tp.lastLog = time.Now()
-	tp.scp.ForAllTxProperties(func(props *tx.Properties) {
-		props.LogToFile = true
-	})
+	conns := tp.activePool.GetAll()
+	for _, c := range conns {
+		c.(*TxConnection).LogToFile.Set(1)
+	}
 }
 
 // Timeout returns the transaction timeout.
@@ -288,29 +378,140 @@ func (tp *TxPool) SetTimeout(timeout time.Duration) {
 	tp.ticks.SetInterval(timeout / 10)
 }
 
-func (tp *TxPool) txComplete(conn tx.IStatefulConnection, reason tx.ReleaseReason) {
-	tp.log(conn, reason)
-	tp.limiter.Release(conn.TxProperties().ImmediateCaller, conn.TxProperties().EffectiveCaller)
-	conn.CleanTxState()
+// TxConnection is meant for executing transactions. It can return itself to
+// the tx pool correctly. It also does not retry statements if there
+// are failures.
+type TxConnection struct {
+	dbConn            *connpool.DBConn
+	TransactionID     int64
+	pool              *TxPool
+	StartTime         time.Time
+	EndTime           time.Time
+	Queries           []string
+	Conclusion        string
+	LogToFile         sync2.AtomicInt32
+	ImmediateCallerID *querypb.VTGateCallerID
+	EffectiveCallerID *vtrpcpb.CallerID
+	Autocommit        bool
 }
 
-func (tp *TxPool) log(txc tx.IStatefulConnection, reason tx.ReleaseReason) {
-	if txc.TxProperties() == nil {
-		return //Nothing to log as no transaction exists on this connection.
+func newTxConnection(conn *connpool.DBConn, transactionID int64, pool *TxPool, immediate *querypb.VTGateCallerID, effective *vtrpcpb.CallerID, autocommit bool) *TxConnection {
+	return &TxConnection{
+		dbConn:            conn,
+		TransactionID:     transactionID,
+		pool:              pool,
+		StartTime:         time.Now(),
+		ImmediateCallerID: immediate,
+		EffectiveCallerID: effective,
+		Autocommit:        autocommit,
 	}
-	txc.TxProperties().Conclusion = reason.Name()
-	txc.TxProperties().EndTime = time.Now()
+}
 
-	username := callerid.GetPrincipal(txc.TxProperties().EffectiveCaller)
+// Close closes the connection.
+func (txc *TxConnection) Close() {
+	if txc.dbConn != nil {
+		txc.dbConn.Close()
+	}
+}
+
+// Exec executes the statement for the current transaction.
+func (txc *TxConnection) Exec(ctx context.Context, query string, maxrows int, wantfields bool) (*sqltypes.Result, error) {
+	if txc.dbConn == nil {
+		return nil, vterrors.Errorf(vtrpcpb.Code_ABORTED, "transaction was aborted: %v", txc.Conclusion)
+	}
+	r, err := txc.dbConn.ExecOnce(ctx, query, maxrows, wantfields)
+	if err != nil {
+		if mysql.IsConnErr(err) {
+			select {
+			case <-ctx.Done():
+				// If the context is done, the query was killed.
+				// So, don't trigger a mysql check.
+			default:
+				txc.pool.env.CheckMySQL()
+			}
+		}
+		return nil, err
+	}
+	return r, nil
+}
+
+// BeginAgain commits the existing transaction and begins a new one
+func (txc *TxConnection) BeginAgain(ctx context.Context) error {
+	if txc.dbConn == nil || txc.Autocommit {
+		return nil
+	}
+	if _, err := txc.dbConn.Exec(ctx, "commit", 1, false); err != nil {
+		return err
+	}
+	if _, err := txc.dbConn.Exec(ctx, "begin", 1, false); err != nil {
+		return err
+	}
+	return nil
+}
+
+// Recycle returns the connection to the pool. The transaction remains
+// active.
+func (txc *TxConnection) Recycle() {
+	if txc.dbConn == nil {
+		return
+	}
+	if txc.dbConn.IsClosed() {
+		txc.conclude(TxClose, "closed")
+	} else {
+		txc.pool.activePool.Put(txc.TransactionID)
+	}
+}
+
+// RecordQuery records the query against this transaction.
+func (txc *TxConnection) RecordQuery(query string) {
+	txc.Queries = append(txc.Queries, query)
+}
+
+func (txc *TxConnection) conclude(conclusion, reason string) {
+	if txc.dbConn == nil {
+		return
+	}
+	txc.pool.activePool.Unregister(txc.TransactionID, reason)
+	txc.dbConn.Recycle()
+	txc.dbConn = nil
+	txc.pool.limiter.Release(txc.ImmediateCallerID, txc.EffectiveCallerID)
+	txc.log(conclusion)
+}
+
+func (txc *TxConnection) log(conclusion string) {
+	txc.Conclusion = conclusion
+	txc.EndTime = time.Now()
+
+	username := callerid.GetPrincipal(txc.EffectiveCallerID)
 	if username == "" {
-		username = callerid.GetUsername(txc.TxProperties().ImmediateCaller)
+		username = callerid.GetUsername(txc.ImmediateCallerID)
 	}
-	duration := txc.TxProperties().EndTime.Sub(txc.TxProperties().StartTime)
-	txc.Stats().UserTransactionCount.Add([]string{username, reason.Name()}, 1)
-	txc.Stats().UserTransactionTimesNs.Add([]string{username, reason.Name()}, int64(duration))
-	txc.TxProperties().Stats.Add(reason.Name(), duration)
-	if txc.TxProperties().LogToFile {
-		log.Infof("Logged transaction: %s", txc.String())
+	duration := txc.EndTime.Sub(txc.StartTime)
+	txc.pool.env.Stats().UserTransactionCount.Add([]string{username, conclusion}, 1)
+	txc.pool.env.Stats().UserTransactionTimesNs.Add([]string{username, conclusion}, int64(duration))
+	txc.pool.txStats.Add(conclusion, duration)
+	if txc.LogToFile.Get() != 0 {
+		log.Infof("Logged transaction: %s", txc.Format())
 	}
 	tabletenv.TxLogger.Send(txc)
 }
+
+// EventTime returns the time the event was created.
+func (txc *TxConnection) EventTime() time.Time {
+	return txc.EndTime
+}
+
+// Format returns a printable version of the connection info.
+func (txc *TxConnection) Format() string {
+	return fmt.Sprintf(
+		"%v\t'%v'\t'%v'\t%v\t%v\t%.6f\t%v\t%v\t\n",
+		txc.TransactionID,
+		callerid.GetPrincipal(txc.EffectiveCallerID),
+		callerid.GetUsername(txc.ImmediateCallerID),
+		txc.StartTime.Format(time.StampMicro),
+		txc.EndTime.Format(time.StampMicro),
+		txc.EndTime.Sub(txc.StartTime).Seconds(),
+		txc.Conclusion,
+		strings.Join(txc.Queries, ";"),
+	)
+}
diff --git a/go/vt/vttablet/tabletserver/tx_prep_pool.go b/go/vt/vttablet/tabletserver/tx_prep_pool.go
index 1be341eef1..0ec7d51c3c 100644
--- a/go/vt/vttablet/tabletserver/tx_prep_pool.go
+++ b/go/vt/vttablet/tabletserver/tx_prep_pool.go
@@ -20,8 +20,6 @@ import (
 	"errors"
 	"fmt"
 	"sync"
-
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
 )
 
 var (
@@ -34,7 +32,7 @@ var (
 // is done by TxPool.
 type TxPreparedPool struct {
 	mu       sync.Mutex
-	conns    map[string]tx.IStatefulConnection
+	conns    map[string]*TxConnection
 	reserved map[string]error
 	capacity int
 }
@@ -46,7 +44,7 @@ func NewTxPreparedPool(capacity int) *TxPreparedPool {
 		capacity = 0
 	}
 	return &TxPreparedPool{
-		conns:    make(map[string]tx.IStatefulConnection, capacity),
+		conns:    make(map[string]*TxConnection, capacity),
 		reserved: make(map[string]error),
 		capacity: capacity,
 	}
@@ -54,7 +52,7 @@ func NewTxPreparedPool(capacity int) *TxPreparedPool {
 
 // Put adds the connection to the pool. It returns an error
 // if the pool is full or on duplicate key.
-func (pp *TxPreparedPool) Put(c tx.IStatefulConnection, dtid string) error {
+func (pp *TxPreparedPool) Put(c *TxConnection, dtid string) error {
 	pp.mu.Lock()
 	defer pp.mu.Unlock()
 	if _, ok := pp.reserved[dtid]; ok {
@@ -75,7 +73,7 @@ func (pp *TxPreparedPool) Put(c tx.IStatefulConnection, dtid string) error {
 // is in the reserved list, it means that an operator is trying
 // to resolve a previously failed commit. So, it removes the entry
 // and returns nil.
-func (pp *TxPreparedPool) FetchForRollback(dtid string) tx.IStatefulConnection {
+func (pp *TxPreparedPool) FetchForRollback(dtid string) *TxConnection {
 	pp.mu.Lock()
 	defer pp.mu.Unlock()
 	if _, ok := pp.reserved[dtid]; ok {
@@ -94,7 +92,7 @@ func (pp *TxPreparedPool) FetchForRollback(dtid string) tx.IStatefulConnection {
 // reserved list by calling Forget. If the commit failed, SetFailed
 // must be called. This will inform future retries that the previous
 // commit failed.
-func (pp *TxPreparedPool) FetchForCommit(dtid string) (tx.IStatefulConnection, error) {
+func (pp *TxPreparedPool) FetchForCommit(dtid string) (*TxConnection, error) {
 	pp.mu.Lock()
 	defer pp.mu.Unlock()
 	if err, ok := pp.reserved[dtid]; ok {
@@ -125,14 +123,14 @@ func (pp *TxPreparedPool) Forget(dtid string) {
 
 // FetchAll removes all connections and returns them as a list.
 // It also forgets all reserved dtids.
-func (pp *TxPreparedPool) FetchAll() []tx.IStatefulConnection {
+func (pp *TxPreparedPool) FetchAll() []*TxConnection {
 	pp.mu.Lock()
 	defer pp.mu.Unlock()
-	conns := make([]tx.IStatefulConnection, 0, len(pp.conns))
+	conns := make([]*TxConnection, 0, len(pp.conns))
 	for _, c := range pp.conns {
 		conns = append(conns, c)
 	}
-	pp.conns = make(map[string]tx.IStatefulConnection, pp.capacity)
+	pp.conns = make(map[string]*TxConnection, pp.capacity)
 	pp.reserved = make(map[string]error)
 	return conns
 }
diff --git a/go/vt/vttablet/tabletserver/txlogz.go b/go/vt/vttablet/tabletserver/txlogz.go
index bd9f3076ff..f393941add 100644
--- a/go/vt/vttablet/tabletserver/txlogz.go
+++ b/go/vt/vttablet/tabletserver/txlogz.go
@@ -111,16 +111,16 @@ func txlogzHandler(w http.ResponseWriter, req *http.Request) {
 	for i := 0; i < limit; i++ {
 		select {
 		case out := <-ch:
-			txc, ok := out.(*StatefulConnection)
+			txc, ok := out.(*TxConnection)
 			if !ok {
-				err := fmt.Errorf("unexpected value in %s: %#v (expecting value of type %T)", tabletenv.TxLogger.Name(), out, &StatefulConnection{})
+				err := fmt.Errorf("unexpected value in %s: %#v (expecting value of type %T)", tabletenv.TxLogger.Name(), out, &TxConnection{})
 				io.WriteString(w, `<tr class="error">`)
 				io.WriteString(w, err.Error())
 				io.WriteString(w, "</tr>")
 				log.Error(err)
 				continue
 			}
-			duration := txc.txProps.EndTime.Sub(txc.txProps.StartTime).Seconds()
+			duration := txc.EndTime.Sub(txc.StartTime).Seconds()
 			var level string
 			if duration < 0.1 {
 				level = "low"
@@ -130,7 +130,7 @@ func txlogzHandler(w http.ResponseWriter, req *http.Request) {
 				level = "high"
 			}
 			tmplData := struct {
-				*StatefulConnection
+				*TxConnection
 				Duration   float64
 				ColorLevel string
 			}{txc, duration, level}
diff --git a/go/vt/wrangler/materializer.go b/go/vt/wrangler/materializer.go
index cd3ccf014a..4cab0c0a9c 100644
--- a/go/vt/wrangler/materializer.go
+++ b/go/vt/wrangler/materializer.go
@@ -22,7 +22,6 @@ import (
 	"sync"
 	"text/template"
 
-	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/vtgate/evalengine"
 
 	"github.com/gogo/protobuf/proto"
@@ -51,10 +50,6 @@ type materializer struct {
 	targetShards  []*topo.ShardInfo
 }
 
-const (
-	createDDLAsCopy = "copy"
-)
-
 // MoveTables initiates moving table(s) over to another keyspace
 func (wr *Wrangler) MoveTables(ctx context.Context, workflow, sourceKeyspace, targetKeyspace, tableSpecs, cell, tabletTypes string) error {
 	var tables []string
@@ -117,7 +112,7 @@ func (wr *Wrangler) MoveTables(ctx context.Context, workflow, sourceKeyspace, ta
 		ms.TableSettings = append(ms.TableSettings, &vtctldatapb.TableMaterializeSettings{
 			TargetTable:      table,
 			SourceExpression: buf.String(),
-			CreateDdl:        createDDLAsCopy,
+			CreateDdl:        "copy",
 		})
 	}
 	return wr.Materialize(ctx, ms)
@@ -588,77 +583,46 @@ func (wr *Wrangler) buildMaterializer(ctx context.Context, ms *vtctldatapb.Mater
 }
 
 func (mz *materializer) deploySchema(ctx context.Context) error {
-
 	return mz.forAllTargets(func(target *topo.ShardInfo) error {
-		allTables := []string{"/.*/"}
-
-		hasTargetTable := map[string]bool{}
-		{
-			targetSchema, err := mz.wr.GetSchema(ctx, target.MasterAlias, allTables, nil, false)
-			if err != nil {
-				return err
-			}
-
-			for _, td := range targetSchema.TableDefinitions {
-				hasTargetTable[td.Name] = true
-			}
-		}
-
-		sourceDDL := map[string]string{}
-		{
-			sourceMaster := mz.sourceShards[0].MasterAlias
-			if sourceMaster == nil {
-				return fmt.Errorf("source shard must have a master for copying schema: %v", mz.sourceShards[0].ShardName())
-			}
-
-			log.Infof("getting table schemas from source master...")
-			var err error
-			sourceSchema, err := mz.wr.GetSchema(ctx, sourceMaster, allTables, nil, false)
+		for _, ts := range mz.ms.TableSettings {
+			tableSchema, err := mz.wr.GetSchema(ctx, target.MasterAlias, []string{ts.TargetTable}, nil, false)
 			if err != nil {
 				return err
 			}
-
-			for _, td := range sourceSchema.TableDefinitions {
-				sourceDDL[td.Name] = td.Schema
-			}
-		}
-
-		for _, ts := range mz.ms.TableSettings {
-			if hasTargetTable[ts.TargetTable] {
+			if len(tableSchema.TableDefinitions) != 0 {
 				// Table already exists.
 				continue
 			}
 			if ts.CreateDdl == "" {
 				return fmt.Errorf("target table %v does not exist and there is no create ddl defined", ts.TargetTable)
 			}
-			createDDL := ts.CreateDdl
-			if createDDL == createDDLAsCopy {
-				if ts.SourceExpression != "" {
-					// Check for table if non-empty SourceExpression.
-					sourceTableName, err := sqlparser.TableFromStatement(ts.SourceExpression)
-					if err != nil {
-						return err
-					}
-					if sourceTableName.Name.String() != ts.TargetTable {
-						return fmt.Errorf("source and target table names must match for copying schema: %v vs %v", sqlparser.String(sourceTableName), ts.TargetTable)
-
-					}
-
+			createddl := ts.CreateDdl
+			if createddl == "copy" {
+				sourceTableName, err := sqlparser.TableFromStatement(ts.SourceExpression)
+				if err != nil {
+					return err
 				}
-
-				ddl, ok := sourceDDL[ts.TargetTable]
-				if !ok {
+				if sourceTableName.Name.String() != ts.TargetTable {
+					return fmt.Errorf("source and target table names must match for copying schema: %v vs %v", sqlparser.String(sourceTableName), ts.TargetTable)
+				}
+				sourceMaster := mz.sourceShards[0].MasterAlias
+				if sourceMaster == nil {
+					return fmt.Errorf("source shard must have a master for copying schema: %v", mz.sourceShards[0].ShardName())
+				}
+				sourceSchema, err := mz.wr.GetSchema(ctx, sourceMaster, []string{ts.TargetTable}, nil, false)
+				if err != nil {
+					return err
+				}
+				if len(sourceSchema.TableDefinitions) == 0 {
 					return fmt.Errorf("source table %v does not exist", ts.TargetTable)
 				}
-				createDDL = ddl
+				createddl = sourceSchema.TableDefinitions[0].Schema
 			}
-
 			targetTablet, err := mz.wr.ts.GetTablet(ctx, target.MasterAlias)
 			if err != nil {
 				return err
 			}
-
-			if _, err := mz.wr.tmc.ExecuteFetchAsDba(ctx, targetTablet.Tablet, false, []byte(createDDL), 0, false, true); err != nil {
+			if _, err := mz.wr.tmc.ExecuteFetchAsDba(ctx, targetTablet.Tablet, false, []byte(createddl), 0, false, true); err != nil {
 				return err
 			}
 		}
@@ -680,13 +644,7 @@ func (mz *materializer) generateInserts(ctx context.Context) (string, error) {
 			rule := &binlogdatapb.Rule{
 				Match: ts.TargetTable,
 			}
-
-			if ts.SourceExpression == "" {
-				bls.Filter.Rules = append(bls.Filter.Rules, rule)
-				continue
-			}
-
-			// Validate non-empty query.
+			// Validate the query.
 			stmt, err := sqlparser.Parse(ts.SourceExpression)
 			if err != nil {
 				return "", err
@@ -695,8 +653,6 @@ func (mz *materializer) generateInserts(ctx context.Context) (string, error) {
 			if !ok {
 				return "", fmt.Errorf("unrecognized statement: %s", ts.SourceExpression)
 			}
-
-			filter := ts.SourceExpression
 			if mz.targetVSchema.Keyspace.Sharded && mz.targetVSchema.Tables[ts.TargetTable].Type != vindexes.TypeReference {
 				cv, err := vindexes.FindBestColVindex(mz.targetVSchema.Tables[ts.TargetTable])
 				if err != nil {
@@ -724,12 +680,10 @@ func (mz *materializer) generateInserts(ctx context.Context) (string, error) {
 						Exprs: subExprs,
 					},
 				}
-
-				filter = sqlparser.String(sel)
+				rule.Filter = sqlparser.String(sel)
+			} else {
+				rule.Filter = ts.SourceExpression
 			}
-
-			rule.Filter = filter
-
 			bls.Filter.Rules = append(bls.Filter.Rules, rule)
 		}
 		ig.AddRow(mz.ms.Workflow, bls, "", mz.ms.Cell, mz.ms.TabletTypes)


Test patch:
diff --git a/go/test/endtoend/vtgate/setstatement/udv_test.go b/go/test/endtoend/vtgate/setstatement/udv_test.go
index 19ae64c7bb..a60a6a1646 100644
--- a/go/test/endtoend/vtgate/setstatement/udv_test.go
+++ b/go/test/endtoend/vtgate/setstatement/udv_test.go
@@ -47,8 +47,17 @@ func TestSetUDV(t *testing.T) {
 		query:        "set @foo = 'abc', @bar = 42, @baz = 30.5, @tablet = concat('foo','bar')",
 		expectedRows: "", rowsAffected: 0,
 	}, {
+		query:        "/*!40101 SET @OLD_SQL_MODE=@@SQL_MODE */",
+		expectedRows: "", rowsAffected: 0,
+	}, { // This is handled at vtgate.
 		query:        "select @foo, @bar, @baz, @tablet",
 		expectedRows: `[[VARBINARY("abc") INT64(42) FLOAT64(30.5) VARBINARY("foobar")]]`, rowsAffected: 1,
+	}, { // Cannot really check a specific value for sql_mode as it will differ based on database selected to run these tests.
+		query:        "select @OLD_SQL_MODE = @@SQL_MODE",
+		expectedRows: `[[INT64(1)]]`, rowsAffected: 1,
+	}, { // This one is sent to tablet.
+		query:        "select @foo, @bar, @baz, @tablet, @OLD_SQL_MODE = @@SQL_MODE",
+		expectedRows: `[[VARCHAR("abc") INT64(42) DECIMAL(30.5) VARCHAR("foobar") INT64(1)]]`, rowsAffected: 1,
 	}, {
 		query:        "insert into test(id, val1, val2, val3) values(1, @foo, null, null), (2, null, @bar, null), (3, null, null, @baz)",
 		expectedRows: ``, rowsAffected: 3,
diff --git a/go/vt/sqlparser/ast_test.go b/go/vt/sqlparser/ast_test.go
index 968492bcec..16089143fb 100644
--- a/go/vt/sqlparser/ast_test.go
+++ b/go/vt/sqlparser/ast_test.go
@@ -792,9 +792,3 @@ func TestTypeConversion(t *testing.T) {
 	ct2 := &ColumnType{Type: "bigint"}
 	assert.Equal(t, ct1.SQLType(), ct2.SQLType())
 }
-
-func TestDefaultStatus(t *testing.T) {
-	assert.Equal(t,
-		String(&Default{ColName: "status"}),
-		"default(`status`)")
-}
diff --git a/go/vt/sqlparser/parse_test.go b/go/vt/sqlparser/parse_test.go
index aa56a019dd..b5884f78a1 100644
--- a/go/vt/sqlparser/parse_test.go
+++ b/go/vt/sqlparser/parse_test.go
@@ -697,8 +697,6 @@ var (
 		input: "insert /* bool in on duplicate */ into a values (1, 2, 3) on duplicate key update b = values(a.b), c = d",
 	}, {
 		input: "insert /* bool expression on duplicate */ into a values (1, 2) on duplicate key update b = func(a), c = a > d",
-	}, {
-		input: "insert into user(username, `status`) values ('Chuck', default(`status`))",
 	}, {
 		input: "update /* simple */ a set b = 3",
 	}, {
diff --git a/go/vt/vtgate/engine/insert_test.go b/go/vt/vtgate/engine/insert_test.go
index c51d4f37a6..3c46e5510a 100644
--- a/go/vt/vtgate/engine/insert_test.go
+++ b/go/vt/vtgate/engine/insert_test.go
@@ -574,10 +574,9 @@ func TestInsertShardedOwnedWithNull(t *testing.T) {
 					"onecol": {
 						Type: "lookup",
 						Params: map[string]string{
-							"table":        "lkp1",
-							"from":         "from",
-							"to":           "toc",
-							"ignore_nulls": "true",
+							"table": "lkp1",
+							"from":  "from",
+							"to":    "toc",
 						},
 						Owner: "t1",
 					},
@@ -637,6 +636,8 @@ func TestInsertShardedOwnedWithNull(t *testing.T) {
 		t.Fatal(err)
 	}
 	vc.ExpectLog(t, []string{
+		`Execute insert into lkp1(from, toc) values(:from0, :toc0) from0: toc0: type:VARBINARY ` +
+			`value:"\026k@\264J\272K\326"  true`,
 		`ResolveDestinations sharded [value:"0" ] Destinations:DestinationKeyspaceID(166b40b44aba4bd6)`,
 		`ExecuteMultiShard sharded.20-: prefix mid1 suffix ` +
 			`{_c30: _id0: type:INT64 value:"1" } true true`,
@@ -948,10 +949,9 @@ func TestInsertShardedIgnoreOwnedWithNull(t *testing.T) {
 					"onecol": {
 						Type: "lookup",
 						Params: map[string]string{
-							"table":        "lkp1",
-							"from":         "from",
-							"to":           "toc",
-							"ignore_nulls": "true",
+							"table": "lkp1",
+							"from":  "from",
+							"to":    "toc",
 						},
 						Owner: "t1",
 					},
@@ -1023,6 +1023,8 @@ func TestInsertShardedIgnoreOwnedWithNull(t *testing.T) {
 		t.Fatal(err)
 	}
 	vc.ExpectLog(t, []string{
+		`Execute insert ignore into lkp1(from, toc) values(:from0, :toc0) from0: toc0: type:VARBINARY ` +
+			`value:"\026k@\264J\272K\326"  true`,
 		`Execute select from from lkp1 where from = :from and toc = :toc from: toc: type:VARBINARY value:"\026k@\264J\272K\326"  false`,
 		`ResolveDestinations sharded [value:"0" ] Destinations:DestinationKeyspaceID(166b40b44aba4bd6)`,
 		`ExecuteMultiShard sharded.-20: prefix mid1 suffix ` +
diff --git a/go/vt/vtgate/plan_executor_set_test.go b/go/vt/vtgate/plan_executor_set_test.go
index b5bfe15fae..ffb91ed37d 100644
--- a/go/vt/vtgate/plan_executor_set_test.go
+++ b/go/vt/vtgate/plan_executor_set_test.go
@@ -56,7 +56,7 @@ func TestPlanExecutorSetUDV(t *testing.T) {
 		out *vtgatepb.Session
 		err string
 	}{{
-		in:  "set @foo = 'bar'",
+		in:  "set @FOO = 'bar'",
 		out: &vtgatepb.Session{UserDefinedVariables: createMap([]string{"foo"}, []interface{}{"bar"}), Autocommit: true},
 	}, {
 		in:  "set @foo = 2",
diff --git a/go/vt/vtgate/vindexes/lookup_hash_test.go b/go/vt/vtgate/vindexes/lookup_hash_test.go
index f94204a1a4..ba6ff9223e 100644
--- a/go/vt/vtgate/vindexes/lookup_hash_test.go
+++ b/go/vt/vtgate/vindexes/lookup_hash_test.go
@@ -220,22 +220,15 @@ func TestLookupHashCreate(t *testing.T) {
 		t.Errorf("vc.queries length: %v, want %v", got, want)
 	}
 
-	err = lookuphash.(Lookup).Create(vc, [][]sqltypes.Value{{sqltypes.NULL}}, [][]byte{[]byte("\x16k@\xb4J\xbaK\xd6")}, false /* ignoreMode */)
-	want := "lookup.Create: input has null values: row: 0, col: 0"
-	if err == nil || err.Error() != want {
-		t.Errorf("lookuphash.Create(NULL) err: %v, want %s", err, want)
-	}
-
 	vc.queries = nil
-	lookuphash.(*LookupHash).lkp.IgnoreNulls = true
 	err = lookuphash.(Lookup).Create(vc, [][]sqltypes.Value{{sqltypes.NULL}}, [][]byte{[]byte("\x16k@\xb4J\xbaK\xd6")}, false /* ignoreMode */)
 	require.NoError(t, err)
-	if got, want := len(vc.queries), 0; got != want {
+	if got, want := len(vc.queries), 1; got != want {
 		t.Errorf("vc.queries length: %v, want %v", got, want)
 	}
 
 	err = lookuphash.(Lookup).Create(vc, [][]sqltypes.Value{{sqltypes.NewInt64(1)}}, [][]byte{[]byte("bogus")}, false /* ignoreMode */)
-	want = "lookup.Create.vunhash: invalid keyspace id: 626f677573"
+	want := "lookup.Create.vunhash: invalid keyspace id: 626f677573"
 	if err == nil || err.Error() != want {
 		t.Errorf("lookuphash.Create(bogus) err: %v, want %s", err, want)
 	}
diff --git a/go/vt/vtgate/vindexes/lookup_test.go b/go/vt/vtgate/vindexes/lookup_test.go
index 6784dcac5a..827e0708dd 100644
--- a/go/vt/vtgate/vindexes/lookup_test.go
+++ b/go/vt/vtgate/vindexes/lookup_test.go
@@ -371,30 +371,8 @@ func TestLookupNonUniqueCreate(t *testing.T) {
 	vc.queries = nil
 	err = lookupNonUnique.(Lookup).Create(vc, [][]sqltypes.Value{{sqltypes.NewInt64(2)}, {sqltypes.NewInt64(1)}}, [][]byte{[]byte("test2"), []byte("test1")}, true /* ignoreMode */)
 	require.NoError(t, err)
-	wantqueries[0].Sql = "insert ignore into t(fromc, toc) values(:fromc0, :toc0), (:fromc1, :toc1)"
-	if !reflect.DeepEqual(vc.queries, wantqueries) {
-		t.Errorf("lookup.Create queries:\n%v, want\n%v", vc.queries, wantqueries)
-	}
-
-	// With ignore_nulls off
-	err = lookupNonUnique.(Lookup).Create(vc, [][]sqltypes.Value{{sqltypes.NewInt64(2)}, {sqltypes.NULL}}, [][]byte{[]byte("test2"), []byte("test1")}, true /* ignoreMode */)
-	want := "lookup.Create: input has null values: row: 1, col: 0"
-	if err == nil || err.Error() != want {
-		t.Errorf("lookupNonUnique(query fail) err: %v, want %s", err, want)
-	}
 
-	// With ignore_nulls on
-	vc.queries = nil
-	lookupNonUnique.(*LookupNonUnique).lkp.IgnoreNulls = true
-	err = lookupNonUnique.(Lookup).Create(vc, [][]sqltypes.Value{{sqltypes.NewInt64(2)}, {sqltypes.NULL}}, [][]byte{[]byte("test2"), []byte("test1")}, true /* ignoreMode */)
-	require.NoError(t, err)
-	wantqueries = []*querypb.BoundQuery{{
-		Sql: "insert ignore into t(fromc, toc) values(:fromc0, :toc0)",
-		BindVariables: map[string]*querypb.BindVariable{
-			"fromc0": sqltypes.Int64BindVariable(2),
-			"toc0":   sqltypes.BytesBindVariable([]byte("test2")),
-		},
-	}}
+	wantqueries[0].Sql = "insert ignore into t(fromc, toc) values(:fromc0, :toc0), (:fromc1, :toc1)"
 	if !reflect.DeepEqual(vc.queries, wantqueries) {
 		t.Errorf("lookup.Create queries:\n%v, want\n%v", vc.queries, wantqueries)
 	}
@@ -402,7 +380,7 @@ func TestLookupNonUniqueCreate(t *testing.T) {
 	// Test query fail.
 	vc.mustFail = true
 	err = lookupNonUnique.(Lookup).Create(vc, [][]sqltypes.Value{{sqltypes.NewInt64(1)}}, [][]byte{[]byte("\x16k@\xb4J\xbaK\xd6")}, false /* ignoreMode */)
-	want = "lookup.Create: execute failed"
+	want := "lookup.Create: execute failed"
 	if err == nil || err.Error() != want {
 		t.Errorf("lookupNonUnique(query fail) err: %v, want %s", err, want)
 	}
diff --git a/go/vt/vttablet/endtoend/batch_test.go b/go/vt/vttablet/endtoend/batch_test.go
index 7f4dac26ca..40118d1e6c 100644
--- a/go/vt/vttablet/endtoend/batch_test.go
+++ b/go/vt/vttablet/endtoend/batch_test.go
@@ -20,7 +20,6 @@ import (
 	"reflect"
 	"testing"
 
-	"github.com/stretchr/testify/require"
 	"vitess.io/vitess/go/sqltypes"
 	querypb "vitess.io/vitess/go/vt/proto/query"
 	"vitess.io/vitess/go/vt/vttablet/endtoend/framework"
@@ -119,14 +118,16 @@ func TestBatchRead(t *testing.T) {
 	want := []sqltypes.Result{qr1, qr2}
 
 	qrl, err := client.ExecuteBatch(queries, false)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	if !reflect.DeepEqual(qrl, want) {
 		t.Errorf("ExecueBatch: \n%#v, want \n%#v", prettyPrintArr(qrl), prettyPrintArr(want))
 	}
 }
 
 func TestBatchTransaction(t *testing.T) {
-
 	client := framework.NewClient()
 	queries := []*querypb.BoundQuery{{
 		Sql: "insert into vitess_test values(4, null, null, null)",
@@ -147,14 +148,20 @@ func TestBatchTransaction(t *testing.T) {
 
 	// Not in transaction, AsTransaction false
 	qrl, err := client.ExecuteBatch(queries, false)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	if !reflect.DeepEqual(qrl[1].Rows, wantRows) {
 		t.Errorf("Rows: \n%#v, want \n%#v", qrl[1].Rows, wantRows)
 	}
 
 	// Not in transaction, AsTransaction true
 	qrl, err = client.ExecuteBatch(queries, true)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	if !reflect.DeepEqual(qrl[1].Rows, wantRows) {
 		t.Errorf("Rows: \n%#v, want \n%#v", qrl[1].Rows, wantRows)
 	}
@@ -162,10 +169,16 @@ func TestBatchTransaction(t *testing.T) {
 	// In transaction, AsTransaction false
 	func() {
 		err = client.Begin(false)
-		require.NoError(t, err)
+		if err != nil {
+			t.Error(err)
+			return
+		}
 		defer client.Commit()
 		qrl, err = client.ExecuteBatch(queries, false)
-		require.NoError(t, err)
+		if err != nil {
+			t.Error(err)
+			return
+		}
 		if !reflect.DeepEqual(qrl[1].Rows, wantRows) {
 			t.Errorf("Rows: \n%#v, want \n%#v", qrl[1].Rows, wantRows)
 		}
@@ -174,10 +187,15 @@ func TestBatchTransaction(t *testing.T) {
 	// In transaction, AsTransaction true
 	func() {
 		err = client.Begin(false)
-		require.NoError(t, err)
+		if err != nil {
+			t.Error(err)
+			return
+		}
 		defer client.Rollback()
 		qrl, err = client.ExecuteBatch(queries, true)
 		want := "cannot start a new transaction in the scope of an existing one"
-		require.EqualError(t, err, want)
+		if err == nil || err.Error() != want {
+			t.Errorf("Error: %v, want %s", err, want)
+		}
 	}()
 }
diff --git a/go/vt/vttablet/endtoend/config_test.go b/go/vt/vttablet/endtoend/config_test.go
index 851a45a6b3..026c2855e6 100644
--- a/go/vt/vttablet/endtoend/config_test.go
+++ b/go/vt/vttablet/endtoend/config_test.go
@@ -17,13 +17,12 @@ limitations under the License.
 package endtoend
 
 import (
+	"fmt"
 	"strings"
 	"sync"
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/require"
-
 	"vitess.io/vitess/go/sqltypes"
 	querypb "vitess.io/vitess/go/vt/proto/query"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
@@ -34,15 +33,88 @@ import (
 )
 
 // compareIntDiff returns an error if end[tag] != start[tag]+diff.
-func compareIntDiff(t *testing.T, end map[string]interface{}, tag string, start map[string]interface{}, diff int) {
-	t.Helper()
-	verifyIntValue(t, end, tag, framework.FetchInt(start, tag)+diff)
+func compareIntDiff(end map[string]interface{}, tag string, start map[string]interface{}, diff int) error {
+	return verifyIntValue(end, tag, framework.FetchInt(start, tag)+diff)
 }
 
 // verifyIntValue returns an error if values[tag] != want.
-func verifyIntValue(t *testing.T, values map[string]interface{}, tag string, want int) {
-	t.Helper()
-	require.Equal(t, want, framework.FetchInt(values, tag), tag)
+func verifyIntValue(values map[string]interface{}, tag string, want int) error {
+	got := framework.FetchInt(values, tag)
+	if got != want {
+		return fmt.Errorf("%s: %d, want %d", tag, got, want)
+	}
+	return nil
+}
+
+func TestConfigVars(t *testing.T) {
+	currentConfig := tabletenv.NewCurrentConfig()
+	vars := framework.DebugVars()
+	cases := []struct {
+		tag string
+		val int
+	}{{
+		tag: "ConnPoolAvailable",
+		val: currentConfig.OltpReadPool.Size,
+	}, {
+		tag: "ConnPoolCapacity",
+		val: currentConfig.OltpReadPool.Size,
+	}, {
+		tag: "ConnPoolIdleTimeout",
+		val: currentConfig.OltpReadPool.IdleTimeoutSeconds * 1e9,
+	}, {
+		tag: "ConnPoolMaxCap",
+		val: currentConfig.OltpReadPool.Size,
+	}, {
+		tag: "MaxResultSize",
+		val: currentConfig.Oltp.MaxRows,
+	}, {
+		tag: "WarnResultSize",
+		val: currentConfig.Oltp.WarnRows,
+	}, {
+		tag: "QueryCacheCapacity",
+		val: currentConfig.QueryCacheSize,
+	}, {
+		tag: "QueryTimeout",
+		val: int(currentConfig.Oltp.QueryTimeoutSeconds * 1e9),
+	}, {
+		tag: "SchemaReloadTime",
+		val: int(currentConfig.SchemaReloadIntervalSeconds * 1e9),
+	}, {
+		tag: "StreamBufferSize",
+		val: currentConfig.StreamBufferSize,
+	}, {
+		tag: "StreamConnPoolAvailable",
+		val: currentConfig.OlapReadPool.Size,
+	}, {
+		tag: "StreamConnPoolCapacity",
+		val: currentConfig.OlapReadPool.Size,
+	}, {
+		tag: "StreamConnPoolIdleTimeout",
+		val: currentConfig.OlapReadPool.IdleTimeoutSeconds * 1e9,
+	}, {
+		tag: "StreamConnPoolMaxCap",
+		val: currentConfig.OlapReadPool.Size,
+	}, {
+		tag: "TransactionPoolAvailable",
+		val: currentConfig.TxPool.Size,
+	}, {
+		tag: "TransactionPoolCapacity",
+		val: currentConfig.TxPool.Size,
+	}, {
+		tag: "TransactionPoolIdleTimeout",
+		val: currentConfig.TxPool.IdleTimeoutSeconds * 1e9,
+	}, {
+		tag: "TransactionPoolMaxCap",
+		val: currentConfig.TxPool.Size,
+	}, {
+		tag: "TransactionTimeout",
+		val: int(currentConfig.Oltp.TxTimeoutSeconds * 1e9),
+	}}
+	for _, tcase := range cases {
+		if err := verifyIntValue(vars, tcase.tag, int(tcase.val)); err != nil {
+			t.Error(err)
+		}
+	}
 }
 
 func TestPoolSize(t *testing.T) {
@@ -50,7 +122,9 @@ func TestPoolSize(t *testing.T) {
 	framework.Server.SetPoolSize(1)
 
 	vstart := framework.DebugVars()
-	verifyIntValue(t, vstart, "ConnPoolCapacity", 1)
+	if err := verifyIntValue(vstart, "ConnPoolCapacity", 1); err != nil {
+		t.Error(err)
+	}
 
 	var wg sync.WaitGroup
 	wg.Add(2)
@@ -196,19 +270,34 @@ func TestQueryPlanCache(t *testing.T) {
 	_, _ = client.Execute("select * from vitess_test where intval=:ival1", bindVars)
 	_, _ = client.Execute("select * from vitess_test where intval=:ival2", bindVars)
 	vend := framework.DebugVars()
-	verifyIntValue(t, vend, "QueryCacheLength", 1)
-	verifyIntValue(t, vend, "QueryCacheSize", 1)
-	verifyIntValue(t, vend, "QueryCacheCapacity", 1)
+	if err := verifyIntValue(vend, "QueryCacheLength", 1); err != nil {
+		t.Error(err)
+	}
+	if err := verifyIntValue(vend, "QueryCacheSize", 1); err != nil {
+		t.Error(err)
+	}
+	if err := verifyIntValue(vend, "QueryCacheCapacity", 1); err != nil {
+		t.Error(err)
+	}
 
 	framework.Server.SetQueryPlanCacheCap(10)
 	_, _ = client.Execute("select * from vitess_test where intval=:ival1", bindVars)
 	vend = framework.DebugVars()
-	verifyIntValue(t, vend, "QueryCacheLength", 2)
-	verifyIntValue(t, vend, "QueryCacheSize", 2)
+	if err := verifyIntValue(vend, "QueryCacheLength", 2); err != nil {
+		t.Error(err)
+	}
+	if err := verifyIntValue(vend, "QueryCacheSize", 2); err != nil {
+		t.Error(err)
+	}
+
 	_, _ = client.Execute("select * from vitess_test where intval=1", bindVars)
 	vend = framework.DebugVars()
-	verifyIntValue(t, vend, "QueryCacheLength", 3)
-	verifyIntValue(t, vend, "QueryCacheSize", 3)
+	if err := verifyIntValue(vend, "QueryCacheLength", 3); err != nil {
+		t.Error(err)
+	}
+	if err := verifyIntValue(vend, "QueryCacheSize", 3); err != nil {
+		t.Error(err)
+	}
 }
 
 func TestMaxResultSize(t *testing.T) {
@@ -222,7 +311,10 @@ func TestMaxResultSize(t *testing.T) {
 	if err == nil || !strings.HasPrefix(err.Error(), want) {
 		t.Errorf("Error: %v, must start with %s", err, want)
 	}
-	verifyIntValue(t, framework.DebugVars(), "MaxResultSize", 2)
+	if err := verifyIntValue(framework.DebugVars(), "MaxResultSize", 2); err != nil {
+		t.Error(err)
+	}
+
 	framework.Server.SetMaxResultSize(10)
 	_, err = client.Execute(query, nil)
 	if err != nil {
@@ -245,7 +337,10 @@ func TestWarnResultSize(t *testing.T) {
 		t.Errorf("Warnings.ResultsExceeded counter should have increased by 1, instead got %v", exceededCountDiff)
 	}
 
-	verifyIntValue(t, framework.DebugVars(), "WarnResultSize", 2)
+	if err := verifyIntValue(framework.DebugVars(), "WarnResultSize", 2); err != nil {
+		t.Error(err)
+	}
+
 	framework.Server.SetWarnResultSize(10)
 	_, _ = client.Execute(query, nil)
 	newerWarningsResultsExceededCount := framework.FetchInt(framework.DebugVars(), "Warnings/ResultsExceeded")
@@ -275,6 +370,10 @@ func TestQueryTimeout(t *testing.T) {
 		t.Errorf("Error code: %v, want %v", code, vtrpcpb.Code_ABORTED)
 	}
 	vend := framework.DebugVars()
-	verifyIntValue(t, vend, "QueryTimeout", int(100*time.Millisecond))
-	compareIntDiff(t, vend, "Kills/Queries", vstart, 1)
+	if err := verifyIntValue(vend, "QueryTimeout", int(100*time.Millisecond)); err != nil {
+		t.Error(err)
+	}
+	if err := compareIntDiff(vend, "Kills/Queries", vstart, 1); err != nil {
+		t.Error(err)
+	}
 }
diff --git a/go/vt/vttablet/endtoend/misc_test.go b/go/vt/vttablet/endtoend/misc_test.go
index 6b21c05ee6..a6f41805c8 100644
--- a/go/vt/vttablet/endtoend/misc_test.go
+++ b/go/vt/vttablet/endtoend/misc_test.go
@@ -48,8 +48,12 @@ func TestSimpleRead(t *testing.T) {
 		return
 	}
 	vend := framework.DebugVars()
-	compareIntDiff(t, vend, "Queries/TotalCount", vstart, 1)
-	compareIntDiff(t, vend, "Queries/Histograms/Select/Count", vstart, 1)
+	if err := compareIntDiff(vend, "Queries/TotalCount", vstart, 1); err != nil {
+		t.Error(err)
+	}
+	if err := compareIntDiff(vend, "Queries/Histograms/Select/Count", vstart, 1); err != nil {
+		t.Error(err)
+	}
 }
 
 func TestBinary(t *testing.T) {
@@ -184,7 +188,9 @@ func TestIntegrityError(t *testing.T) {
 	if err == nil || !strings.HasPrefix(err.Error(), want) {
 		t.Errorf("Error: %v, want prefix %s", err, want)
 	}
-	compareIntDiff(t, framework.DebugVars(), "Errors/ALREADY_EXISTS", vstart, 1)
+	if err := compareIntDiff(framework.DebugVars(), "Errors/ALREADY_EXISTS", vstart, 1); err != nil {
+		t.Error(err)
+	}
 }
 
 func TestTrailingComment(t *testing.T) {
@@ -289,7 +295,10 @@ func TestConsolidation(t *testing.T) {
 		wg.Wait()
 
 		vend := framework.DebugVars()
-		compareIntDiff(t, vend, "Waits/Histograms/Consolidations/Count", vstart, 1)
+		if err := compareIntDiff(vend, "Waits/Histograms/Consolidations/Count", vstart, 1); err != nil {
+			t.Logf("DebugVars Waits/Histograms/Consolidations/Count not incremented with sleep=%v", sleep)
+			continue
+		}
 		t.Logf("DebugVars properly incremented with sleep=%v", sleep)
 		return
 	}
@@ -492,9 +501,15 @@ func TestQueryStats(t *testing.T) {
 		t.Errorf("stat: %+v, want %+v", stat, want)
 	}
 	vend := framework.DebugVars()
-	compareIntDiff(t, vend, "QueryCounts/vitess_a.Select", vstart, 2)
-	compareIntDiff(t, vend, "QueryRowCounts/vitess_a.Select", vstart, 2)
-	compareIntDiff(t, vend, "QueryErrorCounts/vitess_a.Select", vstart, 1)
+	if err := compareIntDiff(vend, "QueryCounts/vitess_a.Select", vstart, 2); err != nil {
+		t.Error(err)
+	}
+	if err := compareIntDiff(vend, "QueryRowCounts/vitess_a.Select", vstart, 2); err != nil {
+		t.Error(err)
+	}
+	if err := compareIntDiff(vend, "QueryErrorCounts/vitess_a.Select", vstart, 1); err != nil {
+		t.Error(err)
+	}
 
 	// Ensure BeginExecute also updates the stats and strips comments.
 	query = "select /* begin_execute */ 1 /* trailing comment */"
diff --git a/go/vt/vttablet/endtoend/transaction_test.go b/go/vt/vttablet/endtoend/transaction_test.go
index e588ea03bb..715c2d0790 100644
--- a/go/vt/vttablet/endtoend/transaction_test.go
+++ b/go/vt/vttablet/endtoend/transaction_test.go
@@ -18,12 +18,11 @@ package endtoend
 
 import (
 	"fmt"
+	"reflect"
 	"strings"
 	"testing"
 	"time"
 
-	"vitess.io/vitess/go/test/utils"
-
 	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/context"
@@ -41,28 +40,63 @@ func TestCommit(t *testing.T) {
 	client := framework.NewClient()
 	defer client.Execute("delete from vitess_test where intval=4", nil)
 
+	catcher := framework.NewTxCatcher()
+	defer catcher.Close()
 	vstart := framework.DebugVars()
 
-	query := "insert into vitess_test (intval, floatval, charval, binval) values (4, null, null, null)"
+	query := "insert into vitess_test (intval, floatval, charval, binval) " +
+		"values(4, null, null, null)"
 	err := client.Begin(false)
-	require.NoError(t, err)
-
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	_, err = client.Execute(query, nil)
-	require.NoError(t, err)
-
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	err = client.Commit()
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	tx, err := catcher.Next()
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	want := []string{"insert into vitess_test(intval, floatval, charval, binval) values (4, null, null, null)"}
+	if !reflect.DeepEqual(tx.Queries, want) {
+		t.Errorf("queries: %v, want %v", tx.Queries, want)
+	}
+	if !reflect.DeepEqual(tx.Conclusion, "commit") {
+		t.Errorf("conclusion: %s, want commit", tx.Conclusion)
+	}
 
 	qr, err := client.Execute("select * from vitess_test", nil)
-	require.NoError(t, err)
-	require.Equal(t, uint64(4), qr.RowsAffected, "rows affected")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	if qr.RowsAffected != 4 {
+		t.Errorf("rows affected: %d, want 4", qr.RowsAffected)
+	}
 
 	_, err = client.Execute("delete from vitess_test where intval=4", nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 
 	qr, err = client.Execute("select * from vitess_test", nil)
-	require.NoError(t, err)
-	require.Equal(t, uint64(3), qr.RowsAffected, "rows affected")
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	if qr.RowsAffected != 3 {
+		t.Errorf("rows affected: %d, want 4", qr.RowsAffected)
+	}
 
 	expectedDiffs := []struct {
 		tag  string
@@ -94,25 +128,53 @@ func TestCommit(t *testing.T) {
 	}}
 	vend := framework.DebugVars()
 	for _, expected := range expectedDiffs {
-		compareIntDiff(t, vend, expected.tag, vstart, expected.diff)
+		if err := compareIntDiff(vend, expected.tag, vstart, expected.diff); err != nil {
+			t.Error(err)
+		}
 	}
 }
 
 func TestRollback(t *testing.T) {
 	client := framework.NewClient()
 
+	catcher := framework.NewTxCatcher()
+	defer catcher.Close()
 	vstart := framework.DebugVars()
 
 	query := "insert into vitess_test values(4, null, null, null)"
 	err := client.Begin(false)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	_, err = client.Execute(query, nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	err = client.Rollback()
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	tx, err := catcher.Next()
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	want := []string{"insert into vitess_test values (4, null, null, null)"}
+	if !reflect.DeepEqual(tx.Queries, want) {
+		t.Errorf("queries: %v, want %v", tx.Queries, want)
+	}
+	if !reflect.DeepEqual(tx.Conclusion, "rollback") {
+		t.Errorf("conclusion: %s, want rollback", tx.Conclusion)
+	}
 
 	qr, err := client.Execute("select * from vitess_test", nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	if qr.RowsAffected != 3 {
 		t.Errorf("rows affected: %d, want 3", qr.RowsAffected)
 	}
@@ -138,7 +200,9 @@ func TestRollback(t *testing.T) {
 	}}
 	vend := framework.DebugVars()
 	for _, expected := range expectedDiffs {
-		compareIntDiff(t, vend, expected.tag, vstart, expected.diff)
+		if err := compareIntDiff(vend, expected.tag, vstart, expected.diff); err != nil {
+			t.Error(err)
+		}
 	}
 }
 
@@ -146,23 +210,53 @@ func TestAutoCommit(t *testing.T) {
 	client := framework.NewClient()
 	defer client.Execute("delete from vitess_test where intval=4", nil)
 
+	catcher := framework.NewTxCatcher()
+	defer catcher.Close()
 	vstart := framework.DebugVars()
 
-	query := "insert into vitess_test (intval, floatval, charval, binval) values (4, null, null, null)"
+	query := "insert into vitess_test (intval, floatval, charval, binval) " +
+		"values(4, null, null, null)"
 	_, err := client.Execute(query, nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	tx, err := catcher.Next()
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	want := []string{"insert into vitess_test(intval, floatval, charval, binval) values (4, null, null, null)"}
+	// Sometimes, no queries will be returned by the querylog because reliability
+	// is not guaranteed. If so, just move on without verifying. The subsequent
+	// rowcount check will anyway verify that the insert succeeded.
+	if len(tx.Queries) != 0 && !reflect.DeepEqual(tx.Queries, want) {
+		t.Errorf("queries: %v, want %v", tx.Queries, want)
+	}
+	if !reflect.DeepEqual(tx.Conclusion, "commit") {
+		t.Errorf("conclusion: %s, want commit", tx.Conclusion)
+	}
 
 	qr, err := client.Execute("select * from vitess_test", nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	if qr.RowsAffected != 4 {
 		t.Errorf("rows affected: %d, want 4", qr.RowsAffected)
 	}
 
 	_, err = client.Execute("delete from vitess_test where intval=4", nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 
 	qr, err = client.Execute("select * from vitess_test", nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	if qr.RowsAffected != 3 {
 		t.Errorf("rows affected: %d, want 4", qr.RowsAffected)
 	}
@@ -212,21 +306,34 @@ func TestTxPoolSize(t *testing.T) {
 
 	client1 := framework.NewClient()
 	err := client1.Begin(false)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	defer client1.Rollback()
-	verifyIntValue(t, framework.DebugVars(), "TransactionPoolAvailable", tabletenv.NewCurrentConfig().TxPool.Size-1)
+	if err := verifyIntValue(framework.DebugVars(), "TransactionPoolAvailable", tabletenv.NewCurrentConfig().TxPool.Size-1); err != nil {
+		t.Error(err)
+	}
 
 	defer framework.Server.SetTxPoolSize(framework.Server.TxPoolSize())
 	framework.Server.SetTxPoolSize(1)
 	vend := framework.DebugVars()
-	verifyIntValue(t, vend, "TransactionPoolAvailable", 0)
-	verifyIntValue(t, vend, "TransactionPoolCapacity", 1)
+	if err := verifyIntValue(vend, "TransactionPoolAvailable", 0); err != nil {
+		t.Error(err)
+	}
+	if err := verifyIntValue(vend, "TransactionPoolCapacity", 1); err != nil {
+		t.Error(err)
+	}
 
 	client2 := framework.NewClient()
 	err = client2.Begin(false)
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "connection limit exceeded")
-	compareIntDiff(t, framework.DebugVars(), "Errors/RESOURCE_EXHAUSTED", vstart, 1)
+	want := "connection limit exceeded"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("%v, must contain %s", err, want)
+	}
+	if err := compareIntDiff(framework.DebugVars(), "Errors/RESOURCE_EXHAUSTED", vstart, 1); err != nil {
+		t.Error(err)
+	}
 }
 
 func TestForUpdate(t *testing.T) {
@@ -241,11 +348,20 @@ func TestForUpdate(t *testing.T) {
 
 		// We should not get errors here
 		err = client.Begin(false)
-		require.NoError(t, err)
+		if err != nil {
+			t.Error(err)
+			return
+		}
 		_, err = client.Execute(query, nil)
-		require.NoError(t, err)
+		if err != nil {
+			t.Error(err)
+			return
+		}
 		err = client.Commit()
-		require.NoError(t, err)
+		if err != nil {
+			t.Error(err)
+			return
+		}
 	}
 }
 
@@ -256,18 +372,31 @@ func TestPrepareRollback(t *testing.T) {
 	query := "insert into vitess_test (intval, floatval, charval, binval) " +
 		"values(4, null, null, null)"
 	err := client.Begin(false)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	_, err = client.Execute(query, nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	err = client.Prepare("aa")
 	if err != nil {
 		client.RollbackPrepared("aa", 0)
-		t.Fatalf(err.Error())
+		t.Error(err)
+		return
 	}
 	err = client.RollbackPrepared("aa", 0)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	qr, err := client.Execute("select * from vitess_test", nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	if qr.RowsAffected != 3 {
 		t.Errorf("rows affected: %d, want 3", qr.RowsAffected)
 	}
@@ -280,18 +409,31 @@ func TestPrepareCommit(t *testing.T) {
 	query := "insert into vitess_test (intval, floatval, charval, binval) " +
 		"values(4, null, null, null)"
 	err := client.Begin(false)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	_, err = client.Execute(query, nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	err = client.Prepare("aa")
 	if err != nil {
 		client.RollbackPrepared("aa", 0)
-		t.Fatal(err)
+		t.Error(err)
+		return
 	}
 	err = client.CommitPrepared("aa")
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	qr, err := client.Execute("select * from vitess_test", nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	if qr.RowsAffected != 4 {
 		t.Errorf("rows affected: %d, want 4", qr.RowsAffected)
 	}
@@ -304,24 +446,43 @@ func TestPrepareReparentCommit(t *testing.T) {
 	query := "insert into vitess_test (intval, floatval, charval, binval) " +
 		"values(4, null, null, null)"
 	err := client.Begin(false)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	_, err = client.Execute(query, nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	err = client.Prepare("aa")
 	if err != nil {
 		client.RollbackPrepared("aa", 0)
-		t.Fatal(err)
+		t.Error(err)
+		return
 	}
 	// Rollback all transactions
 	err = client.SetServingType(topodatapb.TabletType_REPLICA)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	// This should resurrect the prepared transaction.
 	err = client.SetServingType(topodatapb.TabletType_MASTER)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	err = client.CommitPrepared("aa")
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	qr, err := client.Execute("select * from vitess_test", nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	if qr.RowsAffected != 4 {
 		t.Errorf("rows affected: %d, want 4", qr.RowsAffected)
 	}
@@ -348,14 +509,19 @@ func TestMMCommitFlow(t *testing.T) {
 	require.NoError(t, err)
 
 	err = client.CreateTransaction("aa", []*querypb.Target{})
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "Duplicate entry")
+	want := "Duplicate entry"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("%v, must contain %s", err, want)
+	}
 
 	err = client.StartCommit("aa")
 	require.NoError(t, err)
 
 	err = client.SetRollback("aa", 0)
-	require.EqualError(t, err, "could not transition to ROLLBACK: aa (CallerID: dev)")
+	want = "could not transition to ROLLBACK: aa (CallerID: dev)"
+	if err == nil || err.Error() != want {
+		t.Errorf("%v, must contain %s", err, want)
+	}
 
 	info, err := client.ReadTransaction("aa")
 	require.NoError(t, err)
@@ -373,7 +539,9 @@ func TestMMCommitFlow(t *testing.T) {
 			TabletType: topodatapb.TabletType_MASTER,
 		}},
 	}
-	utils.MustMatch(t, wantInfo, info, "ReadTransaction")
+	if !proto.Equal(info, wantInfo) {
+		t.Errorf("ReadTransaction: %#v, want %#v", info, wantInfo)
+	}
 
 	err = client.ConcludeTransaction("aa")
 	require.NoError(t, err)
@@ -493,12 +661,21 @@ func TestUnresolvedTracking(t *testing.T) {
 	query := "insert into vitess_test (intval, floatval, charval, binval) " +
 		"values(4, null, null, null)"
 	err := client.Begin(false)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	_, err = client.Execute(query, nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	err = client.Prepare("aa")
 	defer client.RollbackPrepared("aa", 0)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	time.Sleep(10 * time.Second)
 	vars := framework.DebugVars()
 	if val := framework.FetchInt(vars, "Unresolved/Prepares"); val != 1 {
@@ -517,30 +694,57 @@ func TestManualTwopcz(t *testing.T) {
 
 	ctx := context.Background()
 	conn, err := mysql.Connect(ctx, &connParams)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	defer conn.Close()
 
 	// Successful prepare.
 	err = client.Begin(false)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	_, err = client.Execute("insert into vitess_test (intval, floatval, charval, binval) values(4, null, null, null)", nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	_, err = client.Execute("insert into vitess_test (intval, floatval, charval, binval) values(5, null, null, null)", nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	err = client.Prepare("dtidsuccess")
 	defer client.RollbackPrepared("dtidsuccess", 0)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 
 	// Failed transaction.
 	err = client.Begin(false)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	_, err = client.Execute("insert into vitess_test (intval, floatval, charval, binval) values(6, null, null, null)", nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	_, err = client.Execute("insert into vitess_test (intval, floatval, charval, binval) values(7, null, null, null)", nil)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	err = client.Prepare("dtidfail")
 	defer client.RollbackPrepared("dtidfail", 0)
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	conn.ExecuteFetch(fmt.Sprintf("update _vt.redo_state set state = %d where dtid = 'dtidfail'", tabletserver.RedoStateFailed), 10, false)
 	conn.ExecuteFetch("commit", 10, false)
 
@@ -554,7 +758,10 @@ func TestManualTwopcz(t *testing.T) {
 	}})
 	defer client.ConcludeTransaction("distributed")
 
-	require.NoError(t, err)
+	if err != nil {
+		t.Error(err)
+		return
+	}
 	fmt.Printf("%s/twopcz\n", framework.ServerAddress)
 	fmt.Print("Sleeping for 30 seconds\n")
 	time.Sleep(30 * time.Second)
diff --git a/go/vt/vttablet/tabletserver/query_executor_test.go b/go/vt/vttablet/tabletserver/query_executor_test.go
index 1b54e198a6..c58947bf23 100644
--- a/go/vt/vttablet/tabletserver/query_executor_test.go
+++ b/go/vt/vttablet/tabletserver/query_executor_test.go
@@ -24,8 +24,6 @@ import (
 	"strings"
 	"testing"
 
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/context"
@@ -379,8 +377,8 @@ func TestQueryExecutorLimitFailure(t *testing.T) {
 		logWant:     "begin; delete from test_table limit 3; rollback",
 		inTxWant:    "delete from test_table limit 3",
 	}}
-	for i, tcase := range testcases {
-		t.Run(fmt.Sprintf("%d - %s", i, tcase.input), func(t *testing.T) {
+	for _, tcase := range testcases {
+		func() {
 			db := setUpQueryExecutorTest(t)
 			defer db.Close()
 			for _, dbr := range tcase.dbResponses {
@@ -395,21 +393,21 @@ func TestQueryExecutorLimitFailure(t *testing.T) {
 			// Test outside a transaction.
 			qre := newTestQueryExecutor(ctx, tsv, tcase.input, 0)
 			_, err := qre.Execute()
-			assert.Error(t, err)
-			assert.Contains(t, err.Error(), tcase.err)
+			if err == nil || !strings.Contains(err.Error(), tcase.err) {
+				t.Errorf("Execute(%v): %v, must contain %v", tcase.input, err, tcase.err)
+			}
 			assert.Equal(t, tcase.logWant, qre.logStats.RewrittenSQL(), tcase.input)
 
 			// Test inside a transaction.
 			txid, err := tsv.Begin(ctx, &tsv.target, nil)
-
 			require.NoError(t, err)
 			defer tsv.Commit(ctx, &tsv.target, txid)
 
 			qre = newTestQueryExecutor(ctx, tsv, tcase.input, txid)
 			_, err = qre.Execute()
-			assert.Error(t, err)
-			assert.Contains(t, err.Error(), tcase.err)
-
+			if err == nil || !strings.Contains(err.Error(), tcase.err) {
+				t.Errorf("Execute(%v): %v, must contain %v", tcase.input, err, tcase.err)
+			}
 			want := tcase.logWant
 			if tcase.inTxWant != "" {
 				want = tcase.inTxWant
@@ -420,12 +418,13 @@ func TestQueryExecutorLimitFailure(t *testing.T) {
 				return
 			}
 			// Ensure transaction was rolled back.
-			conn, err := tsv.te.txPool.GetAndLock(txid, "")
-			require.NoError(t, err)
-			defer conn.Release(tx.TxClose)
-
-			require.False(t, conn.IsInTransaction(), "connection is still in a transaction")
-		})
+			qre = newTestQueryExecutor(ctx, tsv, "update test_table set a=1", txid)
+			_, err = qre.Execute()
+			notxError := "ended at"
+			if err == nil || !strings.Contains(err.Error(), notxError) {
+				t.Errorf("Execute(%v): %v, must contain %v", tcase.input, err, notxError)
+			}
+		}()
 	}
 }
 
@@ -600,7 +599,7 @@ func TestQueryExecutorMessageStreamACL(t *testing.T) {
 	db := setUpQueryExecutorTest(t)
 	defer db.Close()
 
-	tsv := newTestTabletServer(ctx, enableStrictTableACL, db)
+	tsv := newTestTabletServer(context.Background(), enableStrictTableACL, db)
 	defer tsv.StopService()
 
 	plan, err := tsv.qe.GetMessageStreamPlan("msg")
diff --git a/go/vt/vttablet/tabletserver/stateful_connection_pool_test.go b/go/vt/vttablet/tabletserver/stateful_connection_pool_test.go
deleted file mode 100644
index 5ae132a04a..0000000000
--- a/go/vt/vttablet/tabletserver/stateful_connection_pool_test.go
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
-Copyright 2020 The Vitess Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package tabletserver
-
-import (
-	"context"
-	"testing"
-
-	"github.com/stretchr/testify/require"
-	"gotest.tools/assert"
-	"vitess.io/vitess/go/mysql/fakesqldb"
-	"vitess.io/vitess/go/sqltypes"
-	querypb "vitess.io/vitess/go/vt/proto/query"
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-)
-
-var ctx = context.Background()
-
-func TestActivePoolClientRowsFound(t *testing.T) {
-	db := fakesqldb.New(t)
-	defer db.Close()
-	db.AddQuery("begin", &sqltypes.Result{})
-
-	pool := newActivePool()
-	pool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
-
-	startNormalSize := pool.conns.Available()
-	startFoundRowsSize := pool.foundRowsPool.Available()
-
-	conn1, err := pool.NewConn(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	assert.Equal(t, startNormalSize-1, pool.conns.Available(), "default pool not used")
-
-	conn2, err := pool.NewConn(ctx, &querypb.ExecuteOptions{ClientFoundRows: true})
-	require.NoError(t, err)
-	assert.Equal(t, startFoundRowsSize-1, pool.conns.Available(), "foundRows pool not used")
-
-	conn1.Release(tx.TxClose)
-	assert.Equal(t, startNormalSize, pool.conns.Available(), "default pool not restored after release")
-
-	conn2.Release(tx.TxClose)
-	assert.Equal(t, startFoundRowsSize, pool.conns.Available(), "default pool not restored after release")
-}
-
-func TestActivePoolForAllTxProps(t *testing.T) {
-	db := fakesqldb.New(t)
-	defer db.Close()
-	pool := newActivePool()
-	pool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
-	conn1, err := pool.NewConn(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	conn1.txProps = &tx.Properties{}
-
-	conn2, err := pool.NewConn(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	// for the second connection, we are not going to set a tx state
-
-	conn3, err := pool.NewConn(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	conn3.txProps = &tx.Properties{}
-
-	pool.ForAllTxProperties(func(p *tx.Properties) {
-		p.LogToFile = true
-	})
-
-	require.True(t, conn1.txProps.LogToFile, "connection missed")
-	require.Nil(t, conn2.txProps)
-	require.True(t, conn3.txProps.LogToFile, "connection missed")
-}
-
-func TestActivePoolGetConnNonExistentTransaction(t *testing.T) {
-	db := fakesqldb.New(t)
-	defer db.Close()
-	pool := newActivePool()
-	pool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
-	_, err := pool.GetAndLock(12345, "for query")
-	require.EqualError(t, err, "not found")
-}
-
-func TestExecWithAbortedCtx(t *testing.T) {
-	ctx, cancel := context.WithCancel(ctx)
-	db := fakesqldb.New(t)
-	defer db.Close()
-	pool := newActivePool()
-	pool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
-	conn, err := pool.NewConn(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	cancel()
-	_, err = conn.Exec(ctx, "", 0, false)
-	require.Error(t, err)
-}
-
-func TestExecWithDbconnClosed(t *testing.T) {
-	db := fakesqldb.New(t)
-	defer db.Close()
-	pool := newActivePool()
-	pool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
-	conn, err := pool.NewConn(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	conn.Close()
-
-	_, err = conn.Exec(ctx, "", 0, false)
-	require.EqualError(t, err, "connection was aborted")
-}
-
-func TestExecWithDbconnClosedHavingTx(t *testing.T) {
-	db := fakesqldb.New(t)
-	defer db.Close()
-	pool := newActivePool()
-	pool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
-	conn, err := pool.NewConn(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	conn.txProps = &tx.Properties{Conclusion: "foobar"}
-	conn.Close()
-
-	_, err = conn.Exec(ctx, "", 0, false)
-	require.EqualError(t, err, "transaction was aborted: foobar")
-}
-
-func newActivePool() *StatefulConnectionPool {
-	env := newEnv("ActivePoolTest")
-
-	return NewStatefulConnPool(env)
-}
diff --git a/go/vt/vttablet/tabletserver/tabletserver_test.go b/go/vt/vttablet/tabletserver/tabletserver_test.go
index 8f1ebdc2b3..e1225aebea 100644
--- a/go/vt/vttablet/tabletserver/tabletserver_test.go
+++ b/go/vt/vttablet/tabletserver/tabletserver_test.go
@@ -29,8 +29,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
-
 	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/context"
@@ -516,17 +514,16 @@ func TestTabletServerMasterToReplica(t *testing.T) {
 	target := querypb.Target{TabletType: topodatapb.TabletType_MASTER}
 	txid1, err := tsv.Begin(ctx, &target, nil)
 	require.NoError(t, err)
-
-	_, err = tsv.Execute(ctx, &target, "update test_table set name = 2 where pk = 1", nil, txid1, nil)
-	require.NoError(t, err)
-
-	err = tsv.Prepare(ctx, &target, txid1, "aa")
-	require.NoError(t, err)
-
+	if _, err := tsv.Execute(ctx, &target, "update test_table set name = 2 where pk = 1", nil, txid1, nil); err != nil {
+		t.Error(err)
+	}
+	if err = tsv.Prepare(ctx, &target, txid1, "aa"); err != nil {
+		t.Error(err)
+	}
 	txid2, err := tsv.Begin(ctx, &target, nil)
 	require.NoError(t, err)
 	// This makes txid2 busy
-	conn2, err := tsv.te.txPool.GetAndLock(txid2, "for query")
+	conn2, err := tsv.te.txPool.Get(txid2, "for query")
 	require.NoError(t, err)
 	ch := make(chan bool)
 	go func() {
@@ -541,10 +538,12 @@ func TestTabletServerMasterToReplica(t *testing.T) {
 		t.Fatal("ch should not fire")
 	case <-time.After(10 * time.Millisecond):
 	}
-	assert.Equal(t, int64(1), tsv.te.txPool.scp.active.Size(), "tsv.te.txPool.scp.active.Size(): ")
+	if tsv.te.txPool.activePool.Size() != 1 {
+		t.Errorf("len(tsv.te.txPool.activePool.Size()): %d, want 1", len(tsv.te.preparedPool.conns))
+	}
 
 	// Concluding conn2 will allow the transition to go through.
-	tsv.te.txPool.RollbackAndRelease(ctx, conn2)
+	tsv.te.txPool.LocalConclude(ctx, conn2)
 	<-ch
 }
 
@@ -589,7 +588,7 @@ func TestTabletServerRedoLogIsKeptBetweenRestarts(t *testing.T) {
 	if len(tsv.te.preparedPool.conns) != 1 {
 		t.Errorf("len(tsv.te.preparedPool.conns): %d, want 1", len(tsv.te.preparedPool.conns))
 	}
-	got := tsv.te.preparedPool.conns["dtid0"].TxProperties().Queries
+	got := tsv.te.preparedPool.conns["dtid0"].Queries
 	want := []string{"update test_table set name = 2 where pk = 1 limit 10001"}
 	if !reflect.DeepEqual(got, want) {
 		t.Errorf("Prepared queries: %v, want %v", got, want)
@@ -599,7 +598,7 @@ func TestTabletServerRedoLogIsKeptBetweenRestarts(t *testing.T) {
 		t.Errorf("len(tsv.te.preparedPool.conns): %d, want 0", v)
 	}
 
-	tsv.te.txPool.scp.lastID.Set(1)
+	tsv.te.txPool.lastID.Set(1)
 	// Ensure we continue past errors.
 	db.AddQuery(tpc.readAllRedo, &sqltypes.Result{
 		Fields: []*querypb.Field{
@@ -629,7 +628,7 @@ func TestTabletServerRedoLogIsKeptBetweenRestarts(t *testing.T) {
 	if len(tsv.te.preparedPool.conns) != 1 {
 		t.Errorf("len(tsv.te.preparedPool.conns): %d, want 1", len(tsv.te.preparedPool.conns))
 	}
-	got = tsv.te.preparedPool.conns["a:b:10"].TxProperties().Queries
+	got = tsv.te.preparedPool.conns["a:b:10"].Queries
 	want = []string{"update test_table set name = 2 where pk = 1 limit 10001"}
 	if !reflect.DeepEqual(got, want) {
 		t.Errorf("Prepared queries: %v, want %v", got, want)
@@ -639,7 +638,7 @@ func TestTabletServerRedoLogIsKeptBetweenRestarts(t *testing.T) {
 		t.Errorf("Failed dtids: %v, want %v", tsv.te.preparedPool.reserved, wantFailed)
 	}
 	// Verify last id got adjusted.
-	if v := tsv.te.txPool.scp.lastID.Get(); v != 20 {
+	if v := tsv.te.txPool.lastID.Get(); v != 20 {
 		t.Errorf("tsv.te.txPool.lastID.Get(): %d, want 20", v)
 	}
 	turnOffTxEngine()
@@ -2481,7 +2480,7 @@ func TestConfigChanges(t *testing.T) {
 	if val := tsv.TxPoolSize(); val != newSize {
 		t.Errorf("TxPoolSize: %d, want %d", val, newSize)
 	}
-	if val := int(tsv.te.txPool.scp.Capacity()); val != newSize {
+	if val := int(tsv.te.txPool.conns.Capacity()); val != newSize {
 		t.Errorf("tsv.te.txPool.pool.Capacity: %d, want %d", val, newSize)
 	}
 
@@ -2490,7 +2489,7 @@ func TestConfigChanges(t *testing.T) {
 		t.Errorf("tsv.TxTimeout: %v, want %v", val, newDuration)
 	}
 	if val := tsv.te.txPool.Timeout(); val != newDuration {
-		t.Errorf("tsv.te.Pool().Timeout: %v, want %v", val, newDuration)
+		t.Errorf("tsv.te.txPool.Timeout: %v, want %v", val, newDuration)
 	}
 
 	tsv.SetQueryPlanCacheCap(newSize)
diff --git a/go/vt/vttablet/tabletserver/twopc_test.go b/go/vt/vttablet/tabletserver/twopc_test.go
index 6790773ab6..c64231be73 100644
--- a/go/vt/vttablet/tabletserver/twopc_test.go
+++ b/go/vt/vttablet/tabletserver/twopc_test.go
@@ -22,8 +22,6 @@ import (
 	"testing"
 	"time"
 
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
 	"golang.org/x/net/context"
 
 	"vitess.io/vitess/go/sqltypes"
@@ -50,7 +48,7 @@ func TestReadAllRedo(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	var want []*tx.PreparedTx
+	var want []*PreparedTx
 	if !reflect.DeepEqual(prepared, want) {
 		t.Errorf("ReadAllRedo: %s, want %s", jsonStr(prepared), jsonStr(want))
 	}
@@ -76,7 +74,7 @@ func TestReadAllRedo(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	want = []*tx.PreparedTx{{
+	want = []*PreparedTx{{
 		Dtid:    "dtid0",
 		Queries: []string{"stmt01"},
 		Time:    time.Unix(0, 1),
@@ -111,7 +109,7 @@ func TestReadAllRedo(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	want = []*tx.PreparedTx{{
+	want = []*PreparedTx{{
 		Dtid:    "dtid0",
 		Queries: []string{"stmt01", "stmt02"},
 		Time:    time.Unix(0, 1),
@@ -151,7 +149,7 @@ func TestReadAllRedo(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	want = []*tx.PreparedTx{{
+	want = []*PreparedTx{{
 		Dtid:    "dtid0",
 		Queries: []string{"stmt01", "stmt02"},
 		Time:    time.Unix(0, 1),
@@ -210,7 +208,7 @@ func TestReadAllRedo(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	want = []*tx.PreparedTx{{
+	want = []*PreparedTx{{
 		Dtid:    "dtid0",
 		Queries: []string{"stmt01", "stmt02"},
 		Time:    time.Unix(0, 1),
@@ -222,7 +220,7 @@ func TestReadAllRedo(t *testing.T) {
 	if !reflect.DeepEqual(prepared, want) {
 		t.Errorf("ReadAllRedo: %s, want %s", jsonStr(prepared), jsonStr(want))
 	}
-	wantFailed := []*tx.PreparedTx{{
+	wantFailed := []*PreparedTx{{
 		Dtid:    "dtid1",
 		Queries: []string{"stmt11"},
 		Time:    time.Unix(0, 1),
@@ -254,7 +252,7 @@ func TestReadAllTransactions(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	var want []*tx.DistributedTx
+	var want []*DistributedTx
 	if !reflect.DeepEqual(distributed, want) {
 		t.Errorf("ReadAllTransactions: %s, want %s", jsonStr(distributed), jsonStr(want))
 	}
@@ -279,7 +277,7 @@ func TestReadAllTransactions(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	want = []*tx.DistributedTx{{
+	want = []*DistributedTx{{
 		Dtid:    "dtid0",
 		State:   "PREPARE",
 		Created: time.Unix(0, 1),
@@ -318,7 +316,7 @@ func TestReadAllTransactions(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	want = []*tx.DistributedTx{{
+	want = []*DistributedTx{{
 		Dtid:    "dtid0",
 		State:   "PREPARE",
 		Created: time.Unix(0, 1),
@@ -366,7 +364,7 @@ func TestReadAllTransactions(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	want = []*tx.DistributedTx{{
+	want = []*DistributedTx{{
 		Dtid:    "dtid0",
 		State:   "PREPARE",
 		Created: time.Unix(0, 1),
diff --git a/go/vt/vttablet/tabletserver/tx_engine_test.go b/go/vt/vttablet/tabletserver/tx_engine_test.go
index 58d16cd81c..85e80d3455 100644
--- a/go/vt/vttablet/tabletserver/tx_engine_test.go
+++ b/go/vt/vttablet/tabletserver/tx_engine_test.go
@@ -23,10 +23,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
-
-	"github.com/stretchr/testify/require"
-
 	"vitess.io/vitess/go/mysql/fakesqldb"
 	"vitess.io/vitess/go/sqltypes"
 
@@ -57,10 +53,14 @@ func TestTxEngineClose(t *testing.T) {
 
 	// Normal close with timeout wait.
 	te.open()
-	c, beginSQL, err := te.txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	require.Equal(t, "begin", beginSQL)
-	c.Unlock()
+	c, beginSQL, err := te.txPool.LocalBegin(ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if beginSQL != "begin" {
+		t.Errorf("beginSQL: %q, want 'begin'", beginSQL)
+	}
+	c.Recycle()
 	start = time.Now()
 	te.close(false)
 	if diff := time.Since(start); diff < 500*time.Millisecond {
@@ -69,11 +69,11 @@ func TestTxEngineClose(t *testing.T) {
 
 	// Immediate close.
 	te.open()
-	c, _, err = te.txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	c, _, err = te.txPool.LocalBegin(ctx, &querypb.ExecuteOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	c.Unlock()
+	c.Recycle()
 	start = time.Now()
 	te.close(true)
 	if diff := time.Since(start); diff > 500*time.Millisecond {
@@ -83,11 +83,11 @@ func TestTxEngineClose(t *testing.T) {
 	// Normal close with short grace period.
 	te.shutdownGracePeriod = 250 * time.Millisecond
 	te.open()
-	c, _, err = te.txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	c, _, err = te.txPool.LocalBegin(ctx, &querypb.ExecuteOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	c.Unlock()
+	c.Recycle()
 	start = time.Now()
 	te.close(false)
 	if diff := time.Since(start); diff > 500*time.Millisecond {
@@ -100,16 +100,18 @@ func TestTxEngineClose(t *testing.T) {
 	// Normal close with short grace period, but pool gets empty early.
 	te.shutdownGracePeriod = 250 * time.Millisecond
 	te.open()
-	c, _, err = te.txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	c, _, err = te.txPool.LocalBegin(ctx, &querypb.ExecuteOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
-	c.Unlock()
+	c.Recycle()
 	go func() {
 		time.Sleep(100 * time.Millisecond)
-		_, err := te.txPool.GetAndLock(c.ID(), "return")
-		assert.NoError(t, err)
-		te.txPool.RollbackAndRelease(ctx, c)
+		_, err := te.txPool.Get(c.TransactionID, "return")
+		if err != nil {
+			t.Error(err)
+		}
+		te.txPool.LocalConclude(ctx, c)
 	}()
 	start = time.Now()
 	te.close(false)
@@ -122,11 +124,13 @@ func TestTxEngineClose(t *testing.T) {
 
 	// Immediate close, but connection is in use.
 	te.open()
-	c, _, err = te.txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
+	c, _, err = te.txPool.LocalBegin(ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
 	go func() {
 		time.Sleep(100 * time.Millisecond)
-		te.txPool.RollbackAndRelease(ctx, c)
+		te.txPool.LocalConclude(ctx, c)
 	}()
 	start = time.Now()
 	te.close(true)
@@ -193,8 +197,7 @@ func changeState(te *TxEngine, state txEngineState) error {
 	case AcceptingReadOnly:
 		return te.AcceptReadOnly()
 	case NotServing:
-		te.StopGently()
-		return nil
+		return te.Stop()
 	default:
 		return fmt.Errorf("don't know how to do that: %v", state)
 	}
@@ -406,24 +409,28 @@ func TestWithInnerTests(outerT *testing.T) {
 			defer db.Close()
 			te := setupTxEngine(db)
 
-			require.NoError(t,
+			failIfError(t,
 				changeState(te, test.startState))
 
 			switch test.tx {
 			case NoTx:
 				// nothing to do
 			case WriteAccepted:
-				require.NoError(t,
+				failIfError(t,
 					startTransaction(te, true))
 			case ReadOnlyAccepted:
-				require.NoError(t,
+				failIfError(t,
 					startTransaction(te, false))
 			case WriteRejected:
 				err := startTransaction(te, true)
-				require.Error(t, err)
+				if err == nil {
+					t.Fatalf("expected an error to be returned when opening write transaction, but got nil")
+				}
 			case ReadOnlyRejected:
 				err := startTransaction(te, false)
-				require.Error(t, err)
+				if err == nil {
+					t.Fatalf("expected an error to be returned when opening read transaction, but got nil")
+				}
 			default:
 				t.Fatalf("don't know how to [%v]", test.tx)
 			}
@@ -434,7 +441,7 @@ func TestWithInnerTests(outerT *testing.T) {
 				go func(s txEngineState) {
 					defer wg.Done()
 
-					require.NoError(t,
+					failIfError(t,
 						changeState(te, s))
 				}(newState)
 
@@ -445,7 +452,7 @@ func TestWithInnerTests(outerT *testing.T) {
 			// Let's wait for all transitions to wrap up
 			wg.Wait()
 
-			require.NoError(t,
+			failIfError(t,
 				test.stateAssertion(te.state))
 		})
 	}
@@ -461,6 +468,13 @@ func setupTxEngine(db *fakesqldb.DB) *TxEngine {
 	return te
 }
 
+func failIfError(t *testing.T, err error) {
+	if err != nil {
+		t.Logf("%+v", err)
+		t.FailNow()
+	}
+}
+
 func assertEndStateIs(expected txEngineState) func(actual txEngineState) error {
 	return func(actual txEngineState) error {
 		if actual != expected {
diff --git a/go/vt/vttablet/tabletserver/tx_executor_test.go b/go/vt/vttablet/tabletserver/tx_executor_test.go
index b8c2c02aa8..2e5f775d39 100644
--- a/go/vt/vttablet/tabletserver/tx_executor_test.go
+++ b/go/vt/vttablet/tabletserver/tx_executor_test.go
@@ -20,11 +20,10 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"strings"
 	"testing"
 	"time"
 
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
 	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/context"
@@ -47,7 +46,9 @@ func TestTxExecutorEmptyPrepare(t *testing.T) {
 	err := txe.Prepare(txid, "aa")
 	require.NoError(t, err)
 	// Nothing should be prepared.
-	require.Empty(t, txe.te.preparedPool.conns, "txe.te.preparedPool.conns")
+	if len(txe.te.preparedPool.conns) != 0 {
+		t.Errorf("len(txe.te.preparedPool.conns): %d, want 0", len(txe.te.preparedPool.conns))
+	}
 }
 
 func TestTxExecutorPrepare(t *testing.T) {
@@ -72,7 +73,10 @@ func TestTxExecutorPrepareNotInTx(t *testing.T) {
 	defer db.Close()
 	defer tsv.StopService()
 	err := txe.Prepare(0, "aa")
-	require.EqualError(t, err, "transaction 0: not found")
+	want := "transaction 0: not found"
+	if err == nil || err.Error() != want {
+		t.Errorf("Prepare err: %v, want %s", err, want)
+	}
 }
 
 func TestTxExecutorPreparePoolFail(t *testing.T) {
@@ -85,8 +89,10 @@ func TestTxExecutorPreparePoolFail(t *testing.T) {
 	require.NoError(t, err)
 	defer txe.RollbackPrepared("aa", 0)
 	err = txe.Prepare(txid2, "bb")
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "prepared transactions exceeded limit")
+	want := "prepared transactions exceeded limit"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("Prepare err: %v, must contain %s", err, want)
+	}
 }
 
 func TestTxExecutorPrepareRedoBeginFail(t *testing.T) {
@@ -97,8 +103,10 @@ func TestTxExecutorPrepareRedoBeginFail(t *testing.T) {
 	db.AddRejectedQuery("begin", errors.New("begin fail"))
 	err := txe.Prepare(txid, "aa")
 	defer txe.RollbackPrepared("aa", 0)
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "begin fail")
+	want := "begin fail"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("Prepare err: %v, want %s", err, want)
+	}
 }
 
 func TestTxExecutorPrepareRedoFail(t *testing.T) {
@@ -108,8 +116,10 @@ func TestTxExecutorPrepareRedoFail(t *testing.T) {
 	txid := newTxForPrep(tsv)
 	err := txe.Prepare(txid, "bb")
 	defer txe.RollbackPrepared("bb", 0)
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "is not supported")
+	want := "is not supported"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("Prepare err: %v, must contain %s", err, want)
+	}
 }
 
 func TestTxExecutorPrepareRedoCommitFail(t *testing.T) {
@@ -120,8 +130,10 @@ func TestTxExecutorPrepareRedoCommitFail(t *testing.T) {
 	db.AddRejectedQuery("commit", errors.New("commit fail"))
 	err := txe.Prepare(txid, "aa")
 	defer txe.RollbackPrepared("aa", 0)
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "commit fail")
+	want := "commit fail"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("Prepare err: %v, want %s", err, want)
+	}
 }
 
 func TestTxExecutorCommit(t *testing.T) {
@@ -150,12 +162,16 @@ func TestTxExecutorCommitRedoFail(t *testing.T) {
 	defer txe.RollbackPrepared("bb", 0)
 	db.AddQuery("update _vt.redo_state set state = 'Failed' where dtid = 'bb'", &sqltypes.Result{})
 	err = txe.CommitPrepared("bb")
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "is not supported")
+	want := "is not supported"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("txe.CommitPrepared err: %v, must contain %s", err, want)
+	}
 	// A retry should fail differently.
 	err = txe.CommitPrepared("bb")
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "cannot commit dtid bb, state: failed")
+	want = "cannot commit dtid bb, state: failed"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("txe.CommitPrepared err: %v, must contain %s", err, want)
+	}
 }
 
 func TestTxExecutorCommitRedoCommitFail(t *testing.T) {
@@ -168,8 +184,10 @@ func TestTxExecutorCommitRedoCommitFail(t *testing.T) {
 	defer txe.RollbackPrepared("aa", 0)
 	db.AddRejectedQuery("commit", errors.New("commit fail"))
 	err = txe.CommitPrepared("aa")
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "commit fail")
+	want := "commit fail"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("Prepare err: %v, want %s", err, want)
+	}
 }
 
 func TestTxExecutorRollbackBeginFail(t *testing.T) {
@@ -181,8 +199,10 @@ func TestTxExecutorRollbackBeginFail(t *testing.T) {
 	require.NoError(t, err)
 	db.AddRejectedQuery("begin", errors.New("begin fail"))
 	err = txe.RollbackPrepared("aa", txid)
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "begin fail")
+	want := "begin fail"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("Prepare err: %v, want %s", err, want)
+	}
 }
 
 func TestTxExecutorRollbackRedoFail(t *testing.T) {
@@ -195,8 +215,10 @@ func TestTxExecutorRollbackRedoFail(t *testing.T) {
 	err := txe.Prepare(txid, "bb")
 	require.NoError(t, err)
 	err = txe.RollbackPrepared("bb", txid)
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "is not supported")
+	want := "is not supported"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("Prepare err: %v, must contain %s", err, want)
+	}
 }
 
 func TestExecutorCreateTransaction(t *testing.T) {
@@ -227,8 +249,10 @@ func TestExecutorStartCommit(t *testing.T) {
 	db.AddQuery(commitTransition, &sqltypes.Result{})
 	txid = newTxForPrep(tsv)
 	err = txe.StartCommit(txid, "aa")
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "could not transition to COMMIT: aa")
+	want := "could not transition to COMMIT: aa"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("Prepare err: %v, want %s", err, want)
+	}
 }
 
 func TestExecutorSetRollback(t *testing.T) {
@@ -245,8 +269,10 @@ func TestExecutorSetRollback(t *testing.T) {
 	db.AddQuery(rollbackTransition, &sqltypes.Result{})
 	txid = newTxForPrep(tsv)
 	err = txe.SetRollback("aa", txid)
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "could not transition to ROLLBACK: aa")
+	want := "could not transition to ROLLBACK: aa"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("Prepare err: %v, want %s", err, want)
+	}
 }
 
 func TestExecutorConcludeTransaction(t *testing.T) {
@@ -383,7 +409,7 @@ func TestExecutorReadAllTransactions(t *testing.T) {
 	})
 	got, _, _, err := txe.ReadTwopcInflight()
 	require.NoError(t, err)
-	want := []*tx.DistributedTx{{
+	want := []*DistributedTx{{
 		Dtid:    "dtid0",
 		State:   "PREPARE",
 		Created: time.Unix(0, 1),
@@ -488,12 +514,15 @@ func TestNoTwopc(t *testing.T) {
 	want := "2pc is not enabled"
 	for _, tc := range testcases {
 		err := tc.fun()
-		require.EqualError(t, err, want)
+		if err == nil || err.Error() != want {
+			t.Errorf("%s: %v, want %s", tc.desc, err, want)
+		}
 	}
 }
 
 func newTestTxExecutor(t *testing.T) (txe *TxExecutor, tsv *TabletServer, db *fakesqldb.DB) {
 	db = setUpQueryExecutorTest(t)
+	ctx := context.Background()
 	logStats := tabletenv.NewLogStats(ctx, "TestTxExecutor")
 	tsv = newTestTabletServer(ctx, smallTxPool, db)
 	db.AddQueryPattern("insert into _vt\\.redo_state\\(dtid, state, time_created\\) values \\('aa', 1,.*", &sqltypes.Result{})
@@ -511,6 +540,7 @@ func newTestTxExecutor(t *testing.T) (txe *TxExecutor, tsv *TabletServer, db *fa
 // newShortAgeExecutor is same as newTestTxExecutor, but shorter transaction abandon age.
 func newShortAgeExecutor(t *testing.T) (txe *TxExecutor, tsv *TabletServer, db *fakesqldb.DB) {
 	db = setUpQueryExecutorTest(t)
+	ctx := context.Background()
 	logStats := tabletenv.NewLogStats(ctx, "TestTxExecutor")
 	tsv = newTestTabletServer(ctx, smallTxPool|shortTwopcAge, db)
 	db.AddQueryPattern("insert into _vt\\.redo_state\\(dtid, state, time_created\\) values \\('aa', 1,.*", &sqltypes.Result{})
@@ -528,6 +558,7 @@ func newShortAgeExecutor(t *testing.T) (txe *TxExecutor, tsv *TabletServer, db *
 // newNoTwopcExecutor is same as newTestTxExecutor, but 2pc disabled.
 func newNoTwopcExecutor(t *testing.T) (txe *TxExecutor, tsv *TabletServer, db *fakesqldb.DB) {
 	db = setUpQueryExecutorTest(t)
+	ctx := context.Background()
 	logStats := tabletenv.NewLogStats(ctx, "TestTxExecutor")
 	tsv = newTestTabletServer(ctx, noTwopc, db)
 	return &TxExecutor{
@@ -541,7 +572,7 @@ func newNoTwopcExecutor(t *testing.T) (txe *TxExecutor, tsv *TabletServer, db *f
 func newTxForPrep(tsv *TabletServer) int64 {
 	txid := newTransaction(tsv, nil)
 	target := querypb.Target{TabletType: topodatapb.TabletType_MASTER}
-	_, err := tsv.Execute(ctx, &target, "update test_table set name = 2 where pk = 1", nil, txid, nil)
+	_, err := tsv.Execute(context.Background(), &target, "update test_table set name = 2 where pk = 1", nil, txid, nil)
 	if err != nil {
 		panic(err)
 	}
diff --git a/go/vt/vttablet/tabletserver/tx_pool_test.go b/go/vt/vttablet/tabletserver/tx_pool_test.go
index 26e24eb626..df49e45f26 100644
--- a/go/vt/vttablet/tabletserver/tx_pool_test.go
+++ b/go/vt/vttablet/tabletserver/tx_pool_test.go
@@ -17,155 +17,330 @@ limitations under the License.
 package tabletserver
 
 import (
-	"context"
 	"fmt"
+	"strings"
 	"testing"
 	"time"
 
-	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
-	"vitess.io/vitess/go/vt/vterrors"
-
-	"gotest.tools/assert"
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
-	"github.com/stretchr/testify/require"
-
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/txlimiter"
 
+	"golang.org/x/net/context"
+
+	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/mysql/fakesqldb"
 	"vitess.io/vitess/go/sqltypes"
+	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
 
+	"regexp"
+
 	querypb "vitess.io/vitess/go/vt/proto/query"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 )
 
 func TestTxPoolExecuteCommit(t *testing.T) {
-	db, txPool, closer := setup(t)
-	defer closer()
-
-	sql := "select 'this is a query'"
-	// begin a transaction and then return the connection
-	conn, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
+	sql := "update test_column set x=1 where 1!=1"
+	db := fakesqldb.New(t)
+	defer db.Close()
+	db.AddQuery(sql, &sqltypes.Result{})
+	db.AddQuery("begin", &sqltypes.Result{})
+	db.AddQuery("commit", &sqltypes.Result{})
 
-	id := conn.ID()
-	conn.Unlock()
+	txPool := newTxPool()
+	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	defer txPool.Close()
+	ctx := context.Background()
+	transactionID, beginSQL, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if beginSQL != "begin" {
+		t.Errorf("beginSQL got %q want 'begin'", beginSQL)
+	}
+	txConn, err := txPool.Get(transactionID, "for query")
+	if err != nil {
+		t.Fatal(err)
+	}
+	txConn.RecordQuery(sql)
+	_, _ = txConn.Exec(ctx, sql, 1, true)
+	txConn.Recycle()
 
-	// get the connection and execute a query on it
-	conn2, err := txPool.GetAndLock(id, "")
-	require.NoError(t, err)
-	_, _ = conn2.Exec(ctx, sql, 1, true)
-	conn2.Unlock()
+	commitSQL, err := txPool.Commit(ctx, transactionID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if commitSQL != "commit" {
+		t.Errorf("commitSQL got %q want 'commit'", commitSQL)
+	}
+}
 
-	// get the connection again and now commit it
-	conn3, err := txPool.GetAndLock(id, "")
-	require.NoError(t, err)
+func TestTxPoolExecuteRollback(t *testing.T) {
+	sql := "alter table test_table add test_column int"
+	db := fakesqldb.New(t)
+	defer db.Close()
+	db.AddQuery(sql, &sqltypes.Result{})
+	db.AddQuery("begin", &sqltypes.Result{})
+	db.AddQuery("rollback", &sqltypes.Result{})
 
-	_, err = txPool.Commit(ctx, conn3)
-	require.NoError(t, err)
+	txPool := newTxPool()
+	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	defer txPool.Close()
+	ctx := context.Background()
+	transactionID, beginSQL, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if beginSQL != "begin" {
+		t.Errorf("beginSQL got %q want 'begin'", beginSQL)
+	}
+	txConn, err := txPool.Get(transactionID, "for query")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer txPool.Rollback(ctx, transactionID)
+	txConn.RecordQuery(sql)
+	_, err = txConn.Exec(ctx, sql, 1, true)
+	txConn.Recycle()
+	if err != nil {
+		t.Fatalf("got error: %v", err)
+	}
+}
 
-	// try committing again. this should fail
-	_, err = txPool.Commit(ctx, conn)
-	require.EqualError(t, err, "not in a transaction")
+func TestTxPoolRollbackNonBusy(t *testing.T) {
+	db := fakesqldb.New(t)
+	defer db.Close()
+	db.AddQuery("begin", &sqltypes.Result{})
+	db.AddQuery("rollback", &sqltypes.Result{})
 
-	// wrap everything up and assert
-	assert.Equal(t, "begin;"+sql+";commit", db.QueryLog())
-	conn3.Release(tx.TxCommit)
+	txPool := newTxPool()
+	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	defer txPool.Close()
+	ctx := context.Background()
+	txid1, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, _, err = txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	conn1, err := txPool.Get(txid1, "for query")
+	if err != nil {
+		t.Fatal(err)
+	}
+	// This should rollback only txid2.
+	txPool.RollbackNonBusy(ctx)
+	if sz := txPool.activePool.Size(); sz != 1 {
+		t.Errorf("txPool.activePool.Size(): %d, want 1", sz)
+	}
+	conn1.Recycle()
+	// This should rollback txid1.
+	txPool.RollbackNonBusy(ctx)
+	if sz := txPool.activePool.Size(); sz != 0 {
+		t.Errorf("txPool.activePool.Size(): %d, want 0", sz)
+	}
 }
 
-func TestTxPoolExecuteRollback(t *testing.T) {
-	db, txPool, closer := setup(t)
-	defer closer()
+func TestTxPoolTransactionKillerEnforceTimeoutEnabled(t *testing.T) {
+	sqlWithTimeout := "alter table test_table add test_column int"
+	sqlWithoutTimeout := "alter table test_table add test_column_no_timeout int"
+	db := fakesqldb.New(t)
+	defer db.Close()
+	db.AddQuery(sqlWithTimeout, &sqltypes.Result{})
+	db.AddQuery(sqlWithoutTimeout, &sqltypes.Result{})
+	db.AddQuery("begin", &sqltypes.Result{})
+	db.AddQuery("rollback", &sqltypes.Result{})
 
-	conn, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	defer conn.Release(tx.TxRollback)
+	txPool := newTxPool()
+	// make sure transaction killer will run frequent enough
+	txPool.SetTimeout(1 * time.Millisecond)
+	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	defer txPool.Close()
+	ctx := context.Background()
+	killCount := txPool.env.Stats().KillCounters.Counts()["Transactions"]
 
-	err = txPool.Rollback(ctx, conn)
-	require.NoError(t, err)
+	txWithoutTimeout, err := addQuery(ctx, sqlWithoutTimeout, txPool, querypb.ExecuteOptions_DBA)
+	if err != nil {
+		t.Fatal(err)
+	}
 
-	// try rolling back again, this should be no-op.
-	err = txPool.Rollback(ctx, conn)
-	require.NoError(t, err, "not in a transaction")
+	if _, err := addQuery(ctx, sqlWithTimeout, txPool, querypb.ExecuteOptions_UNSPECIFIED); err != nil {
+		t.Fatal(err)
+	}
 
-	assert.Equal(t, "begin;rollback", db.QueryLog())
-}
+	var (
+		killCountDiff int64
+		expectedKills = int64(1)
+		timeoutCh     = time.After(5 * time.Second)
+	)
+
+	// transaction killer should kill the query the second query
+	for {
+		killCountDiff = txPool.env.Stats().KillCounters.Counts()["Transactions"] - killCount
+		if killCountDiff >= expectedKills {
+			break
+		}
 
-func TestTxPoolExecuteRollbackOnClosedConn(t *testing.T) {
-	db, txPool, closer := setup(t)
-	defer closer()
+		select {
+		case <-timeoutCh:
+			txPool.Rollback(ctx, txWithoutTimeout)
+			t.Fatal("waited too long for timed transaction to be killed by transaction killer")
+		default:
+		}
+	}
 
-	conn, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	defer conn.Release(tx.TxRollback)
+	if killCountDiff > expectedKills {
+		t.Fatalf("expected only %v query to be killed, but got %v killed", expectedKills, killCountDiff)
+	}
 
-	conn.Close()
+	txPool.Rollback(ctx, txWithoutTimeout)
+	txPool.WaitForEmpty()
 
-	// rollback should not be logged.
-	err = txPool.Rollback(ctx, conn)
-	require.NoError(t, err)
+	if got, expected := db.GetQueryCalledNum("begin"), 2; got != expected {
+		t.Fatalf("'begin' called: got=%v, expected=%v", got, expected)
+	}
+	if got, expected := db.GetQueryCalledNum(sqlWithoutTimeout), 1; got != expected {
+		t.Fatalf("'%v' called: got=%v, expected=%v", sqlWithoutTimeout, got, expected)
+	}
+	if got, expected := db.GetQueryCalledNum(sqlWithTimeout), 1; got != expected {
+		t.Fatalf("'%v' called: got=%v, expected=%v", sqlWithTimeout, got, expected)
+	}
+	if got, expected := db.GetQueryCalledNum("rollback"), 1; got != expected {
+		t.Fatalf("'rollback' called: got=%v, expected=%v", got, expected)
+	}
 
-	assert.Equal(t, "begin", db.QueryLog())
+}
+func addQuery(ctx context.Context, sql string, txPool *TxPool, workload querypb.ExecuteOptions_Workload) (int64, error) {
+	transactionID, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{Workload: workload})
+	if err != nil {
+		return 0, err
+	}
+	txConn, err := txPool.Get(transactionID, "for query")
+	if err != nil {
+		return 0, err
+	}
+	txConn.Exec(ctx, sql, 1, false)
+	txConn.Recycle()
+	return transactionID, nil
 }
 
-func TestTxPoolRollbackNonBusy(t *testing.T) {
-	db, txPool, closer := setup(t)
-	defer closer()
+func TestTxPoolClientRowsFound(t *testing.T) {
+	db := fakesqldb.New(t)
+	defer db.Close()
+	db.AddQuery("begin", &sqltypes.Result{})
+	txPool := newTxPool()
+	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	ctx := context.Background()
 
-	// start two transactions, and mark one of them as unused
-	conn1, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	conn2, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	conn2.Unlock() // this marks conn2 as NonBusy
+	startNormalSize := txPool.conns.Available()
+	startFoundRowsSize := txPool.foundRowsPool.Available()
 
-	// This should rollback only txid2.
-	txPool.RollbackNonBusy(ctx)
+	// Start a 'normal' transaction. It should take a connection
+	// for the normal 'conns' pool.
+	id1, beginSQL, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if beginSQL != "begin" {
+		t.Errorf("beginSQL got %q want 'begin'", beginSQL)
+	}
+	if got, want := txPool.conns.Available(), startNormalSize-1; got != want {
+		t.Errorf("Normal pool size: %d, want %d", got, want)
+	}
+	if got, want := txPool.foundRowsPool.Available(), startFoundRowsSize; got != want {
+		t.Errorf("foundRows pool size: %d, want %d", got, want)
+	}
 
-	// committing tx1 should not be an issue
-	_, err = txPool.Commit(ctx, conn1)
-	require.NoError(t, err)
+	// Start a 'foundRows' transaction. It should take a connection
+	// from the foundRows pool.
+	id2, beginSQL, err := txPool.Begin(ctx, &querypb.ExecuteOptions{ClientFoundRows: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if beginSQL != "begin" {
+		t.Errorf("beginSQL got %q want 'begin'", beginSQL)
+	}
+	if got, want := txPool.conns.Available(), startNormalSize-1; got != want {
+		t.Errorf("Normal pool size: %d, want %d", got, want)
+	}
+	if got, want := txPool.foundRowsPool.Available(), startFoundRowsSize-1; got != want {
+		t.Errorf("foundRows pool size: %d, want %d", got, want)
+	}
 
-	// Trying to get back to conn2 should not work since the transaction has been rolled back
-	_, err = txPool.GetAndLock(conn2.ID(), "")
-	require.Error(t, err)
+	// Rollback the first transaction. The conn should be returned to
+	// the conns pool.
+	txPool.Rollback(ctx, id1)
+	if got, want := txPool.conns.Available(), startNormalSize; got != want {
+		t.Errorf("Normal pool size: %d, want %d", got, want)
+	}
+	if got, want := txPool.foundRowsPool.Available(), startFoundRowsSize-1; got != want {
+		t.Errorf("foundRows pool size: %d, want %d", got, want)
+	}
 
-	conn1.Release(tx.TxCommit)
-	assert.Equal(t, "begin;begin;rollback;commit", db.QueryLog())
+	// Rollback the second transaction. The conn should be returned to
+	// the foundRows pool.
+	txPool.Rollback(ctx, id2)
+	if got, want := txPool.conns.Available(), startNormalSize; got != want {
+		t.Errorf("Normal pool size: %d, want %d", got, want)
+	}
+	if got, want := txPool.foundRowsPool.Available(), startFoundRowsSize; got != want {
+		t.Errorf("foundRows pool size: %d, want %d", got, want)
+	}
 }
 
 func TestTxPoolTransactionIsolation(t *testing.T) {
-	db, txPool, closer := setup(t)
-	defer closer()
+	db := fakesqldb.New(t)
+	defer db.Close()
+	db.AddQuery("begin", &sqltypes.Result{})
+	txPool := newTxPool()
+	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	ctx := context.Background()
 
-	c2, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{TransactionIsolation: querypb.ExecuteOptions_READ_COMMITTED})
-	require.NoError(t, err)
-	c2.Release(tx.TxClose)
+	// Start a transaction with default. It should not change isolation.
+	_, beginSQL, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if beginSQL != "begin" {
+		t.Errorf("beginSQL got %q want 'begin'", beginSQL)
+	}
 
-	assert.Equal(t, "set transaction isolation level read committed;begin", db.QueryLog())
+	db.AddQuery("set transaction isolation level READ COMMITTED", &sqltypes.Result{})
+	_, beginSQL, err = txPool.Begin(ctx, &querypb.ExecuteOptions{TransactionIsolation: querypb.ExecuteOptions_READ_COMMITTED})
+	if err != nil {
+		t.Fatal(err)
+	}
+	wantBeginSQL := "READ COMMITTED; begin"
+	if beginSQL != wantBeginSQL {
+		t.Errorf("beginSQL got %q want %q", beginSQL, wantBeginSQL)
+	}
 }
 
 func TestTxPoolAutocommit(t *testing.T) {
-	db, txPool, closer := setup(t)
-	defer closer()
+	db := fakesqldb.New(t)
+	defer db.Close()
+	txPool := newTxPool()
+	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	ctx := context.Background()
 
 	// Start a transaction with autocommit. This will ensure that the executor does not send begin/commit statements
 	// to mysql.
 	// This test is meaningful because if txPool.Begin were to send a BEGIN statement to the connection, it will fatal
 	// because is not in the list of expected queries (i.e db.AddQuery hasn't been called).
-	conn1, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{TransactionIsolation: querypb.ExecuteOptions_AUTOCOMMIT})
-	require.NoError(t, err)
-
-	// run a query to see it in the query log
-	query := "select 3"
-	conn1.Exec(ctx, query, 1, false)
-
-	_, err = txPool.Commit(ctx, conn1)
-	require.NoError(t, err)
-	conn1.Release(tx.TxCommit)
-
-	// finally, we should only see the query, no begin/commit
-	assert.Equal(t, query, db.QueryLog())
+	txid, beginSQL, err := txPool.Begin(ctx, &querypb.ExecuteOptions{TransactionIsolation: querypb.ExecuteOptions_AUTOCOMMIT})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if beginSQL != "" {
+		t.Errorf("beginSQL got %q want ''", beginSQL)
+	}
+	commitSQL, err := txPool.Commit(ctx, txid)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if commitSQL != "" {
+		t.Errorf("commitSQL got %q want ''", commitSQL)
+	}
 }
 
 // TestTxPoolBeginWithPoolConnectionError_TransientErrno2006 tests the case
@@ -173,131 +348,214 @@ func TestTxPoolAutocommit(t *testing.T) {
 // db connection. DBConn.Exec() is going to reconnect and retry automatically
 // due to this connection error and the BEGIN will succeed.
 func TestTxPoolBeginWithPoolConnectionError_Errno2006_Transient(t *testing.T) {
-	db, txPool := primeTxPoolWithConnection(t)
+	db, txPool, err := primeTxPoolWithConnection(t)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	defer txPool.Close()
+
+	// Close the connection on the server side.
+	db.CloseAllConnections()
+	if err := db.WaitForClose(2 * time.Second); err != nil {
+		t.Fatal(err)
+	}
+
+	ctx := context.Background()
+	txConn, _, err := txPool.LocalBegin(ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		t.Fatalf("Begin should have succeeded after the retry in DBConn.Exec(): %v", err)
+	}
+	txPool.LocalConclude(ctx, txConn)
+}
+
+// TestTxPoolBeginWithPoolConnectionError_Errno2006_Permanent tests the case
+// where a transient errno 2006 is followed by permanent connection rejections.
+// For example, if all open connections are killed and new connections are
+// rejected.
+func TestTxPoolBeginWithPoolConnectionError_Errno2006_Permanent(t *testing.T) {
+	db, txPool, err := primeTxPoolWithConnection(t)
+	if err != nil {
+		t.Fatal(err)
+	}
 	defer db.Close()
 	defer txPool.Close()
 
 	// Close the connection on the server side.
 	db.CloseAllConnections()
-	err := db.WaitForClose(2 * time.Second)
-	require.NoError(t, err)
+	if err := db.WaitForClose(2 * time.Second); err != nil {
+		t.Fatal(err)
+	}
+	// Prevent new connections as well.
+	db.EnableConnFail()
+
+	// This Begin will error with 2006.
+	// After that, vttablet will automatically try to reconnect and this fail.
+	// DBConn.Exec() will return the reconnect error as final error and not the
+	// initial connection error.
+	_, _, err = txPool.LocalBegin(context.Background(), &querypb.ExecuteOptions{})
+	if err == nil || !strings.Contains(err.Error(), "(errno 2013)") {
+		t.Fatalf("Begin did not return the reconnect error: %v", err)
+	}
+	sqlErr, ok := err.(*mysql.SQLError)
+	if !ok {
+		t.Fatalf("Unexpected error type: %T, want %T", err, &mysql.SQLError{})
+	}
+	if got, want := sqlErr.Number(), mysql.CRServerLost; got != want {
+		t.Errorf("Unexpected error code: %d, want %d", got, want)
+	}
+}
 
-	txConn, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err, "Begin should have succeeded after the retry in DBConn.Exec()")
-	txConn.Release(tx.TxCommit)
+func TestTxPoolBeginWithPoolConnectionError_Errno2013(t *testing.T) {
+	db, txPool, err := primeTxPoolWithConnection(t)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// No db.Close() needed. We close it below.
+	defer txPool.Close()
+
+	// Close the connection *after* the server received the query.
+	// This will provoke a MySQL client error with errno 2013.
+	db.EnableShouldClose()
+
+	// 2013 is not retryable. DBConn.Exec() fails after the first attempt.
+	_, _, err = txPool.Begin(context.Background(), &querypb.ExecuteOptions{})
+	if err == nil || !strings.Contains(err.Error(), "(errno 2013)") {
+		t.Fatalf("Begin must return connection error with MySQL errno 2013: %v", err)
+	}
+	if got, want := vterrors.Code(err), vtrpcpb.Code_UNKNOWN; got != want {
+		t.Errorf("wrong error code for Begin error: got = %v, want = %v", got, want)
+	}
 }
 
 // primeTxPoolWithConnection is a helper function. It reconstructs the
 // scenario where future transactions are going to reuse an open db connection.
-func primeTxPoolWithConnection(t *testing.T) (*fakesqldb.DB, *TxPool) {
-	t.Helper()
+func primeTxPoolWithConnection(t *testing.T) (*fakesqldb.DB, *TxPool, error) {
 	db := fakesqldb.New(t)
 	txPool := newTxPool()
 	// Set the capacity to 1 to ensure that the db connection is reused.
-	txPool.scp.conns.SetCapacity(1)
+	txPool.conns.SetCapacity(1)
 	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
 
 	// Run a query to trigger a database connection. That connection will be
 	// reused by subsequent transactions.
 	db.AddQuery("begin", &sqltypes.Result{})
 	db.AddQuery("rollback", &sqltypes.Result{})
-	txConn, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	txConn.Release(tx.TxCommit)
+	ctx := context.Background()
+	txConn, _, err := txPool.LocalBegin(ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		return nil, nil, err
+	}
+	txPool.LocalConclude(ctx, txConn)
 
-	return db, txPool
+	return db, txPool, nil
 }
 
 func TestTxPoolBeginWithError(t *testing.T) {
-	db, txPool, closer := setup(t)
-	defer closer()
+	db := fakesqldb.New(t)
+	defer db.Close()
 	db.AddRejectedQuery("begin", errRejected)
+	txPool := newTxPool()
+	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	defer txPool.Close()
+	ctx := context.Background()
 	_, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "error: rejected")
-	require.Equal(t, vtrpcpb.Code_UNKNOWN, vterrors.Code(err), "wrong error code for Begin error")
+	want := "error: rejected"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("Begin: %v, want %s", err, want)
+	}
+	if got, want := vterrors.Code(err), vtrpcpb.Code_UNKNOWN; got != want {
+		t.Errorf("wrong error code for Begin error: got = %v, want = %v", got, want)
+	}
 }
 
 func TestTxPoolCancelledContextError(t *testing.T) {
-	// given
-	db, txPool, closer := setup(t)
-	defer closer()
-	ctx, cancel := context.WithCancel(ctx)
+	db := fakesqldb.New(t)
+	defer db.Close()
+	db.AddRejectedQuery("begin", errRejected)
+	txPool := newTxPool()
+	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	defer txPool.Close()
+	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
-
-	// when
 	_, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
-
-	// then
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "transaction pool aborting request due to already expired context")
-	require.Equal(t, vtrpcpb.Code_RESOURCE_EXHAUSTED, vterrors.Code(err))
-	require.Empty(t, db.QueryLog())
-}
-
-func TestTxPoolWaitTimeoutError(t *testing.T) {
-	env := newEnv("TabletServerTest")
-	env.Config().TxPool.Size = 1
-	env.Config().TxPool.MaxWaiters = 0
-	env.Config().TxPool.TimeoutSeconds = 1
-	// given
-	db, txPool, closer := setupWithEnv(t, env)
-	defer closer()
-
-	// lock the only connection in the pool.
-	conn, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	defer conn.Unlock()
-
-	// try locking one more connection.
-	_, _, err = txPool.Begin(ctx, &querypb.ExecuteOptions{})
-
-	// then
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "transaction pool connection limit exceeded")
-	require.Equal(t, vtrpcpb.Code_RESOURCE_EXHAUSTED, vterrors.Code(err))
-	require.Equal(t, "begin", db.QueryLog())
-	require.True(t, conn.TxProperties().LogToFile)
+	want := "transaction pool aborting request due to already expired context"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("Unexpected error: %v, want %s", err, want)
+	}
+	if got, want := vterrors.Code(err), vtrpcpb.Code_RESOURCE_EXHAUSTED; got != want {
+		t.Errorf("wrong error code error: got = %v, want = %v", got, want)
+	}
 }
 
-func TestTxPoolRollbackFailIsPassedThrough(t *testing.T) {
+func TestTxPoolRollbackFail(t *testing.T) {
 	sql := "alter table test_table add test_column int"
-	db, txPool, closer := setup(t)
-	defer closer()
+	db := fakesqldb.New(t)
+	defer db.Close()
+	db.AddQuery(sql, &sqltypes.Result{})
+	db.AddQuery("begin", &sqltypes.Result{})
 	db.AddRejectedQuery("rollback", errRejected)
 
-	conn1, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-
-	_, err = conn1.Exec(ctx, sql, 1, true)
-	require.NoError(t, err)
-
-	// rollback is refused by the underlying db and the error is passed on
-	err = txPool.Rollback(ctx, conn1)
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "error: rejected")
+	txPool := newTxPool()
+	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	defer txPool.Close()
+	ctx := context.Background()
+	transactionID, _, err := txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	txConn, err := txPool.Get(transactionID, "for query")
+	if err != nil {
+		t.Fatal(err)
+	}
+	txConn.RecordQuery(sql)
+	_, err = txConn.Exec(ctx, sql, 1, true)
+	txConn.Recycle()
+	if err != nil {
+		t.Fatalf("got error: %v", err)
+	}
+	err = txPool.Rollback(ctx, transactionID)
+	want := "error: rejected"
+	if err == nil || !strings.Contains(err.Error(), want) {
+		t.Errorf("Begin: %v, want %s", err, want)
+	}
+}
 
-	conn1.Unlock()
+func TestTxPoolGetConnNonExistentTransaction(t *testing.T) {
+	db := fakesqldb.New(t)
+	defer db.Close()
+	txPool := newTxPool()
+	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	defer txPool.Close()
+	_, err := txPool.Get(12345, "for query")
+	want := "transaction 12345: not found"
+	if err == nil || err.Error() != want {
+		t.Errorf("Get: %v, want %s", err, want)
+	}
 }
 
 func TestTxPoolGetConnRecentlyRemovedTransaction(t *testing.T) {
-	db, txPool, _ := setup(t)
+	db := fakesqldb.New(t)
 	defer db.Close()
-	conn1, _, _ := txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	id := conn1.ID()
-	conn1.Unlock()
+	ctx := context.Background()
+	db.AddQuery("begin", &sqltypes.Result{})
+	db.AddQuery("commit", &sqltypes.Result{})
+	db.AddQuery("rollback", &sqltypes.Result{})
+	txPool := newTxPool()
+	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	id, _, _ := txPool.Begin(ctx, &querypb.ExecuteOptions{})
 	txPool.Close()
 
 	assertErrorMatch := func(id int64, reason string) {
-		conn, err := txPool.GetAndLock(id, "for query")
-		if err == nil { //
-			conn.Releasef("fail")
-			t.Errorf("expected to get an error")
-			return
+		conn, err := txPool.Get(id, "for query")
+		if err == nil {
+			conn.Recycle()
+			t.Fatalf("expected error, got nil")
 		}
-
 		want := fmt.Sprintf("transaction %v: ended at .* \\(%v\\)", id, reason)
-		require.Regexp(t, want, err.Error())
+		if m, _ := regexp.MatchString(want, err.Error()); !m {
+			t.Errorf("Get: %v, want match %s", err, want)
+		}
 	}
 
 	assertErrorMatch(id, "pool closed")
@@ -305,89 +563,149 @@ func TestTxPoolGetConnRecentlyRemovedTransaction(t *testing.T) {
 	txPool = newTxPool()
 	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
 
-	conn1, _, _ = txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	id = conn1.ID()
-	_, err := txPool.Commit(ctx, conn1)
-	require.NoError(t, err)
-
-	conn1.Releasef("transaction committed")
+	id, _, _ = txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	if _, err := txPool.Commit(ctx, id); err != nil {
+		t.Fatalf("got error: %v", err)
+	}
 
 	assertErrorMatch(id, "transaction committed")
 
+	id, _, _ = txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	if err := txPool.Rollback(ctx, id); err != nil {
+		t.Fatalf("got error: %v", err)
+	}
+
+	assertErrorMatch(id, "transaction rolled back")
+
+	txPool.Close()
 	txPool = newTxPool()
 	txPool.SetTimeout(1 * time.Millisecond)
 	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
 	defer txPool.Close()
 
-	conn1, _, _ = txPool.Begin(ctx, &querypb.ExecuteOptions{})
-	conn1.Unlock()
-	id = conn1.ID()
+	id, _, _ = txPool.Begin(ctx, &querypb.ExecuteOptions{})
 	time.Sleep(20 * time.Millisecond)
 
 	assertErrorMatch(id, "exceeded timeout: 1ms")
-}
 
-func TestTxPoolCloseKillsStrayTransactions(t *testing.T) {
-	_, txPool, closer := setup(t)
-	defer closer()
-
-	startingStray := txPool.env.Stats().InternalErrors.Counts()["StrayTransactions"]
+	txPool.SetTimeout(1 * time.Hour)
+	id, _, _ = txPool.Begin(ctx, &querypb.ExecuteOptions{})
+	txc, err := txPool.Get(id, "for close")
+	if err != nil {
+		t.Fatalf("got error: %v", err)
+	}
 
-	// Start stray transaction.
-	conn, _, err := txPool.Begin(context.Background(), &querypb.ExecuteOptions{})
-	require.NoError(t, err)
-	conn.Unlock()
+	txc.Close()
+	txc.Recycle()
 
-	// Close kills stray transaction.
-	txPool.Close()
-	require.Equal(t, int64(1), txPool.env.Stats().InternalErrors.Counts()["StrayTransactions"]-startingStray)
-	require.Equal(t, 0, txPool.scp.Capacity())
+	assertErrorMatch(id, "closed")
 }
 
-func newTxPool() *TxPool {
-	return newTxPoolWithEnv(newEnv("TabletServerTest"))
-}
+func TestTxPoolExecFailDueToConnFail_Errno2006(t *testing.T) {
+	db := fakesqldb.New(t)
+	defer db.Close()
+	db.AddQuery("begin", &sqltypes.Result{})
 
-func newTxPoolWithEnv(env tabletenv.Env) *TxPool {
-	limiter := &txlimiter.TxAllowAll{}
-	return NewTxPool(env, limiter)
-}
+	txPool := newTxPool()
+	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	defer txPool.Close()
+	ctx := context.Background()
 
-func newEnv(exporterName string) tabletenv.Env {
-	config := tabletenv.NewDefaultConfig()
-	config.TxPool.Size = 300
-	config.Oltp.TxTimeoutSeconds = 30
-	config.TxPool.TimeoutSeconds = 40
-	config.TxPool.MaxWaiters = 500000
-	config.OltpReadPool.IdleTimeoutSeconds = 30
-	config.OlapReadPool.IdleTimeoutSeconds = 30
-	config.TxPool.IdleTimeoutSeconds = 30
-	env := tabletenv.NewEnv(config, exporterName)
-	return env
+	// Start the transaction.
+	txConn, _, err := txPool.LocalBegin(ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Close the connection on the server side. Future queries will fail.
+	db.CloseAllConnections()
+	if err := db.WaitForClose(2 * time.Second); err != nil {
+		t.Fatal(err)
+	}
+
+	// Query is going to fail with connection error because the connection was closed.
+	sql := "alter table test_table add test_column int"
+	_, err = txConn.Exec(ctx, sql, 1, true)
+	txConn.Recycle()
+	if err == nil || !strings.Contains(err.Error(), "(errno 2006)") {
+		t.Fatalf("Exec must return connection error with MySQL errno 2006: %v", err)
+	}
+	sqlErr, ok := err.(*mysql.SQLError)
+	if !ok {
+		t.Fatalf("Unexpected error type: %T, want %T", err, &mysql.SQLError{})
+	}
+	if num := sqlErr.Number(); num != mysql.CRServerGone {
+		t.Errorf("Unexpected error code: %d, want %d", num, mysql.CRServerGone)
+	}
 }
 
-func setup(t *testing.T) (*fakesqldb.DB, *TxPool, func()) {
+func TestTxPoolExecFailDueToConnFail_Errno2013(t *testing.T) {
 	db := fakesqldb.New(t)
-	db.AddQueryPattern(".*", &sqltypes.Result{})
+	// No db.Close() needed. We close it below.
+	db.AddQuery("begin", &sqltypes.Result{})
 
 	txPool := newTxPool()
 	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	defer txPool.Close()
+	ctx := context.Background()
 
-	return db, txPool, func() {
-		txPool.Close()
-		db.Close()
+	// Start the transaction.
+	txConn, _, err := txPool.LocalBegin(ctx, &querypb.ExecuteOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Close the connection *after* the server received the query.
+	// This will provoke a MySQL client error with errno 2013.
+	db.EnableShouldClose()
+
+	// Query is going to fail with connection error because the connection was closed.
+	sql := "alter table test_table add test_column int"
+	db.AddQuery(sql, &sqltypes.Result{})
+	_, err = txConn.Exec(ctx, sql, 1, true)
+	txConn.Recycle()
+	if err == nil || !strings.Contains(err.Error(), "(errno 2013)") {
+		t.Fatalf("Exec must return connection error with MySQL errno 2013: %v", err)
+	}
+	if got, want := vterrors.Code(err), vtrpcpb.Code_UNKNOWN; got != want {
+		t.Errorf("wrong error code for Exec error: got = %v, want = %v", got, want)
 	}
 }
 
-func setupWithEnv(t *testing.T, env tabletenv.Env) (*fakesqldb.DB, *TxPool, func()) {
+func TestTxPoolCloseKillsStrayTransactions(t *testing.T) {
 	db := fakesqldb.New(t)
-	db.AddQueryPattern(".*", &sqltypes.Result{})
+	defer db.Close()
+	db.AddQuery("begin", &sqltypes.Result{})
 
-	txPool := newTxPoolWithEnv(env)
+	txPool := newTxPool()
 	txPool.Open(db.ConnParams(), db.ConnParams(), db.ConnParams())
+	startingStray := txPool.env.Stats().InternalErrors.Counts()["StrayTransactions"]
+
+	// Start stray transaction.
+	_, _, err := txPool.Begin(context.Background(), &querypb.ExecuteOptions{})
+	if err != nil {
+		t.Fatal(err)
+	}
 
-	return db, txPool, func() {
-		txPool.Close()
-		db.Close()
+	// Close kills stray transaction.
+	txPool.Close()
+	if got, want := txPool.env.Stats().InternalErrors.Counts()["StrayTransactions"]-startingStray, int64(1); got != want {
+		t.Fatalf("internal error count for stray transactions not increased: got = %v, want = %v", got, want)
+	}
+	if got, want := txPool.conns.Capacity(), int64(0); got != want {
+		t.Fatalf("resource pool was not closed. capacity: got = %v, want = %v", got, want)
 	}
 }
+
+func newTxPool() *TxPool {
+	config := tabletenv.NewDefaultConfig()
+	config.TxPool.Size = 300
+	config.Oltp.TxTimeoutSeconds = 30
+	config.TxPool.TimeoutSeconds = 40
+	config.TxPool.MaxWaiters = 500000
+	config.OltpReadPool.IdleTimeoutSeconds = 30
+	config.OlapReadPool.IdleTimeoutSeconds = 30
+	config.TxPool.IdleTimeoutSeconds = 30
+	limiter := &txlimiter.TxAllowAll{}
+	return NewTxPool(tabletenv.NewEnv(config, "TabletServerTest"), limiter)
+}
diff --git a/go/vt/vttablet/tabletserver/tx_prep_pool_test.go b/go/vt/vttablet/tabletserver/tx_prep_pool_test.go
index a1cf50edb5..7d6724643b 100644
--- a/go/vt/vttablet/tabletserver/tx_prep_pool_test.go
+++ b/go/vt/vttablet/tabletserver/tx_prep_pool_test.go
@@ -61,7 +61,7 @@ func TestPrepPut(t *testing.T) {
 
 func TestPrepFetchForRollback(t *testing.T) {
 	pp := NewTxPreparedPool(2)
-	conn := &StatefulConnection{}
+	conn := &TxConnection{}
 	pp.Put(conn, "aa")
 	got := pp.FetchForRollback("bb")
 	if got != nil {
@@ -79,7 +79,7 @@ func TestPrepFetchForRollback(t *testing.T) {
 
 func TestPrepFetchForCommit(t *testing.T) {
 	pp := NewTxPreparedPool(2)
-	conn := &StatefulConnection{}
+	conn := &TxConnection{}
 	got, err := pp.FetchForCommit("aa")
 	require.NoError(t, err)
 	if got != nil {
@@ -118,8 +118,8 @@ func TestPrepFetchForCommit(t *testing.T) {
 
 func TestPrepFetchAll(t *testing.T) {
 	pp := NewTxPreparedPool(2)
-	conn1 := &StatefulConnection{}
-	conn2 := &StatefulConnection{}
+	conn1 := &TxConnection{}
+	conn2 := &TxConnection{}
 	pp.Put(conn1, "aa")
 	pp.Put(conn2, "bb")
 	got := pp.FetchAll()
diff --git a/go/vt/vttablet/tabletserver/txlogz_test.go b/go/vt/vttablet/tabletserver/txlogz_test.go
index bde1325254..2f3564c9dd 100644
--- a/go/vt/vttablet/tabletserver/txlogz_test.go
+++ b/go/vt/vttablet/tabletserver/txlogz_test.go
@@ -23,11 +23,9 @@ import (
 	"testing"
 	"time"
 
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
-	"vitess.io/vitess/go/vt/callerid"
-
 	"vitess.io/vitess/go/streamlog"
+	"vitess.io/vitess/go/sync2"
+	"vitess.io/vitess/go/vt/callerid"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
 )
 
@@ -53,27 +51,26 @@ func testHandler(req *http.Request, t *testing.T) {
 	if !strings.Contains(response.Body.String(), "error") {
 		t.Fatalf("should show an error page since transaction log format is invalid.")
 	}
-	txConn := &StatefulConnection{
-		ConnID: 123456,
-		txProps: &tx.Properties{
-			EffectiveCaller: callerid.NewEffectiveCallerID("effective-caller", "component", "subcomponent"),
-			ImmediateCaller: callerid.NewImmediateCallerID("immediate-caller"),
-			StartTime:       time.Now(),
-			Conclusion:      "unknown",
-			Queries:         []string{"select * from test"},
-		},
+	txConn := &TxConnection{
+		TransactionID:     123456,
+		StartTime:         time.Now(),
+		Queries:           []string{"select * from test"},
+		Conclusion:        "unknown",
+		LogToFile:         sync2.AtomicInt32{},
+		EffectiveCallerID: callerid.NewEffectiveCallerID("effective-caller", "component", "subcomponent"),
+		ImmediateCallerID: callerid.NewImmediateCallerID("immediate-caller"),
 	}
-	txConn.txProps.EndTime = txConn.txProps.StartTime
+	txConn.EndTime = txConn.StartTime
 	response = httptest.NewRecorder()
 	tabletenv.TxLogger.Send(txConn)
 	txlogzHandler(response, req)
 	testNotRedacted(t, response)
-	txConn.txProps.EndTime = txConn.txProps.StartTime.Add(time.Duration(2) * time.Second)
+	txConn.EndTime = txConn.StartTime.Add(time.Duration(2) * time.Second)
 	response = httptest.NewRecorder()
 	tabletenv.TxLogger.Send(txConn)
 	txlogzHandler(response, req)
 	testNotRedacted(t, response)
-	txConn.txProps.EndTime = txConn.txProps.StartTime.Add(time.Duration(500) * time.Millisecond)
+	txConn.EndTime = txConn.StartTime.Add(time.Duration(500) * time.Millisecond)
 	response = httptest.NewRecorder()
 	tabletenv.TxLogger.Send(txConn)
 	txlogzHandler(response, req)
diff --git a/go/vt/wrangler/materializer_env_test.go b/go/vt/wrangler/materializer_env_test.go
index 949a1f4dcd..ab88b81e3b 100644
--- a/go/vt/wrangler/materializer_env_test.go
+++ b/go/vt/wrangler/materializer_env_test.go
@@ -19,7 +19,6 @@ package wrangler
 import (
 	"fmt"
 	"regexp"
-	"strings"
 	"sync"
 	"testing"
 
@@ -77,20 +76,18 @@ func newTestMaterializerEnv(t *testing.T, ms *vtctldatapb.MaterializeSettings, s
 	}
 
 	for _, ts := range ms.TableSettings {
-		tableName := ts.TargetTable
+		tablename := ts.TargetTable
 		table, err := sqlparser.TableFromStatement(ts.SourceExpression)
 		if err == nil {
-			tableName = table.Name.String()
+			tablename = table.Name.String()
 		}
-		env.tmc.schema[ms.SourceKeyspace+"."+tableName] = &tabletmanagerdatapb.SchemaDefinition{
+		env.tmc.schema[ms.SourceKeyspace+"."+tablename] = &tabletmanagerdatapb.SchemaDefinition{
 			TableDefinitions: []*tabletmanagerdatapb.TableDefinition{{
-				Name:   tableName,
-				Schema: fmt.Sprintf("%s_schema", tableName),
+				Schema: fmt.Sprintf("%s_schema", tablename),
 			}},
 		}
 		env.tmc.schema[ms.TargetKeyspace+"."+ts.TargetTable] = &tabletmanagerdatapb.SchemaDefinition{
 			TableDefinitions: []*tabletmanagerdatapb.TableDefinition{{
-				Name:   ts.TargetTable,
 				Schema: fmt.Sprintf("%s_schema", ts.TargetTable),
 			}},
 		}
@@ -172,27 +169,11 @@ func newTestMaterializerTMClient() *testMaterializerTMClient {
 }
 
 func (tmc *testMaterializerTMClient) GetSchema(ctx context.Context, tablet *topodatapb.Tablet, tables, excludeTables []string, includeViews bool) (*tabletmanagerdatapb.SchemaDefinition, error) {
-	schemaDefn := &tabletmanagerdatapb.SchemaDefinition{}
-	for _, table := range tables {
-		// TODO: Add generalized regexps if needed for test purposes.
-		if table == "/.*/" {
-			// Special case of all tables in keyspace.
-			for key, tableDefn := range tmc.schema {
-				if strings.HasPrefix(key, tablet.Keyspace+".") {
-					schemaDefn.TableDefinitions = append(schemaDefn.TableDefinitions, tableDefn.TableDefinitions...)
-				}
-			}
-			break
-		}
-
-		key := tablet.Keyspace + "." + table
-		tableDefn := tmc.schema[key]
-		if tableDefn == nil {
-			continue
-		}
-		schemaDefn.TableDefinitions = append(schemaDefn.TableDefinitions, tableDefn.TableDefinitions...)
+	key := tablet.Keyspace + "." + tables[0]
+	if tmc.schema[key] == nil {
+		return &tabletmanagerdatapb.SchemaDefinition{}, nil
 	}
-	return schemaDefn, nil
+	return tmc.schema[key], nil
 }
 
 func (tmc *testMaterializerTMClient) expectVRQuery(tabletID int, query string, result *sqltypes.Result) {
diff --git a/go/vt/wrangler/materializer_test.go b/go/vt/wrangler/materializer_test.go
index 6e8b549602..85370056e3 100644
--- a/go/vt/wrangler/materializer_test.go
+++ b/go/vt/wrangler/materializer_test.go
@@ -1418,23 +1418,15 @@ func TestMaterializerOneToOne(t *testing.T) {
 		Workflow:       "workflow",
 		SourceKeyspace: "sourceks",
 		TargetKeyspace: "targetks",
-		TableSettings: []*vtctldatapb.TableMaterializeSettings{
-			{
-				TargetTable:      "t1",
-				SourceExpression: "select * from t1",
-				CreateDdl:        "t1ddl",
-			},
-			{
-				TargetTable:      "t2",
-				SourceExpression: "select * from t3",
-				CreateDdl:        "t2ddl",
-			},
-			{
-				TargetTable:      "t4",
-				SourceExpression: "", // empty
-				CreateDdl:        "t4ddl",
-			},
-		},
+		TableSettings: []*vtctldatapb.TableMaterializeSettings{{
+			TargetTable:      "t1",
+			SourceExpression: "select * from t1",
+			CreateDdl:        "t1ddl",
+		}, {
+			TargetTable:      "t2",
+			SourceExpression: "select * from t3",
+			CreateDdl:        "t2ddl",
+		}},
 	}
 	env := newTestMaterializerEnv(t, ms, []string{"0"}, []string{"0"})
 	defer env.close()
@@ -1442,16 +1434,8 @@ func TestMaterializerOneToOne(t *testing.T) {
 	env.tmc.expectVRQuery(
 		200,
 		insertPrefix+
-			`\(`+
-			`'workflow', `+
-			(`'keyspace:\\"sourceks\\" shard:\\"0\\" `+
-				`filter:<`+
-				`rules:<match:\\"t1\\" filter:\\"select.*t1\\" > `+
-				`rules:<match:\\"t2\\" filter:\\"select.*t3\\" > `+
-				`rules:<match:\\"t4\\" > `+
-				`> ', `)+
-			`'', [0-9]*, [0-9]*, '', '', [0-9]*, 0, 'Stopped', 'vt_targetks'`+
-			`\)`+eol,
+			`\('workflow', 'keyspace:\\"sourceks\\" shard:\\"0\\" filter:<rules:<match:\\"t1\\" filter:\\"select.*t1\\" > rules:<match:\\"t2\\" filter:\\"select.*t3\\" > > ', '', [0-9]*, [0-9]*, '', '', [0-9]*, 0, 'Stopped', 'vt_targetks'\)`+
+			eol,
 		&sqltypes.Result{},
 	)
 	env.tmc.expectVRQuery(200, mzUpdateQuery, &sqltypes.Result{})
@@ -1969,27 +1953,7 @@ func TestMaterializerNoSourceMaster(t *testing.T) {
 	assert.EqualError(t, err, "source shard must have a master for copying schema: 0")
 }
 
-func TestMaterializerTableMismatchNonCopy(t *testing.T) {
-	ms := &vtctldatapb.MaterializeSettings{
-		Workflow:       "workflow",
-		SourceKeyspace: "sourceks",
-		TargetKeyspace: "targetks",
-		TableSettings: []*vtctldatapb.TableMaterializeSettings{{
-			TargetTable:      "t1",
-			SourceExpression: "select * from t2",
-			CreateDdl:        "",
-		}},
-	}
-	env := newTestMaterializerEnv(t, ms, []string{"0"}, []string{"0"})
-	defer env.close()
-
-	delete(env.tmc.schema, "targetks.t1")
-
-	err := env.wr.Materialize(context.Background(), ms)
-	assert.EqualError(t, err, "target table t1 does not exist and there is no create ddl defined")
-}
-
-func TestMaterializerTableMismatchCopy(t *testing.T) {
+func TestMaterializerTableMismatch(t *testing.T) {
 	ms := &vtctldatapb.MaterializeSettings{
 		Workflow:       "workflow",
 		SourceKeyspace: "sourceks",