Repository: vitessio/vitess
Pull request #6370 (2020-06-24 15:45:21)

Support locking functions
Closes: #6367


Patch:
diff --git a/go/vt/sqlparser/analyzer.go b/go/vt/sqlparser/analyzer.go
index 58b866ae05..03b7d196b1 100644
--- a/go/vt/sqlparser/analyzer.go
+++ b/go/vt/sqlparser/analyzer.go
@@ -406,3 +406,21 @@ func NewPlanValue(node Expr) (sqltypes.PlanValue, error) {
 	}
 	return sqltypes.PlanValue{}, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "expression is too complex '%v'", String(node))
 }
+
+//IsLockingFunc returns true for all functions that are used to work with mysql advisory locks
+func IsLockingFunc(node Expr) bool {
+	switch p := node.(type) {
+	case *FuncExpr:
+		_, found := lockingFunctions[p.Name.Lowered()]
+		return found
+	}
+	return false
+}
+
+var lockingFunctions = map[string]interface{}{
+	"get_lock":          nil,
+	"is_free_lock":      nil,
+	"is_used_lock":      nil,
+	"release_all_locks": nil,
+	"release_lock":      nil,
+}
diff --git a/go/vt/vtgate/engine/primitive.go b/go/vt/vtgate/engine/primitive.go
index 45470a270b..d094ce6ed0 100644
--- a/go/vt/vtgate/engine/primitive.go
+++ b/go/vt/vtgate/engine/primitive.go
@@ -96,6 +96,9 @@ type (
 		SetUDV(key string, value interface{}) error
 
 		SetSysVar(name string, expr string)
+
+		// NeedsReservedConn marks this session as needing a dedicated connection to mysql
+		NeedsReservedConn()
 	}
 
 	// Plan represents the execution strategy for a given query.
@@ -129,7 +132,7 @@ type (
 		GetKeyspaceName() string
 		GetTableName() string
 		Execute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool) (*sqltypes.Result, error)
-		StreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantields bool, callback func(*sqltypes.Result) error) error
+		StreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error
 		GetFields(vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error)
 		NeedsTransaction() bool
 
diff --git a/go/vt/vtgate/engine/reserve.go b/go/vt/vtgate/engine/reserve.go
new file mode 100644
index 0000000000..a4f4425542
--- /dev/null
+++ b/go/vt/vtgate/engine/reserve.go
@@ -0,0 +1,78 @@
+/*
+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 engine
+
+import (
+	"vitess.io/vitess/go/sqltypes"
+	querypb "vitess.io/vitess/go/vt/proto/query"
+)
+
+var _ Primitive = (*Reserve)(nil)
+
+//Reserve will mark the session as needing a
+//reserved connection and then execute the inner Primitive
+type Reserve struct {
+	Input Primitive
+}
+
+// RouteType is part of the Primitive interface
+func (r *Reserve) RouteType() string {
+	return "reserve"
+}
+
+// GetKeyspaceName is part of the Primitive interface
+func (r *Reserve) GetKeyspaceName() string {
+	return r.Input.GetKeyspaceName()
+}
+
+// GetTableName is part of the Primitive interface
+func (r *Reserve) GetTableName() string {
+	return r.Input.GetTableName()
+}
+
+// Execute is part of the Primitive interface
+func (r *Reserve) Execute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool) (*sqltypes.Result, error) {
+	vcursor.Session().NeedsReservedConn()
+	return r.Input.Execute(vcursor, bindVars, wantfields)
+}
+
+// StreamExecute is part of the Primitive interface
+func (r *Reserve) StreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
+	vcursor.Session().NeedsReservedConn()
+	return r.Input.StreamExecute(vcursor, bindVars, wantfields, callback)
+}
+
+// GetFields is part of the Primitive interface
+func (r *Reserve) GetFields(vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
+	return r.Input.GetFields(vcursor, bindVars)
+}
+
+// NeedsTransaction is part of the Primitive interface
+func (r *Reserve) NeedsTransaction() bool {
+	return r.Input.NeedsTransaction()
+}
+
+// Inputs is part of the Primitive interface
+func (r *Reserve) Inputs() []Primitive {
+	return []Primitive{}
+}
+
+func (r *Reserve) description() PrimitiveDescription {
+	return PrimitiveDescription{
+		OperatorType: "Reserve",
+	}
+}
diff --git a/go/vt/vtgate/planbuilder/builder.go b/go/vt/vtgate/planbuilder/builder.go
index 6adfffdd97..6b73b22be8 100644
--- a/go/vt/vtgate/planbuilder/builder.go
+++ b/go/vt/vtgate/planbuilder/builder.go
@@ -131,6 +131,7 @@ type ContextVSchema interface {
 	TabletType() topodatapb.TabletType
 	TargetDestination(qualifier string) (key.Destination, *vindexes.Keyspace, topodatapb.TabletType, error)
 	AnyKeyspace() (*vindexes.Keyspace, error)
+	FirstSortedKeyspace() (*vindexes.Keyspace, error)
 }
 
 //-------------------------------------------------------------------------
diff --git a/go/vt/vtgate/planbuilder/select.go b/go/vt/vtgate/planbuilder/select.go
index cacb6de0e1..9ccb1eb9c9 100644
--- a/go/vt/vtgate/planbuilder/select.go
+++ b/go/vt/vtgate/planbuilder/select.go
@@ -20,6 +20,8 @@ import (
 	"errors"
 	"fmt"
 
+	"vitess.io/vitess/go/vt/key"
+
 	"vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/vterrors"
 
@@ -33,7 +35,10 @@ import (
 func buildSelectPlan(stmt sqlparser.Statement, vschema ContextVSchema) (engine.Primitive, error) {
 	sel := stmt.(*sqlparser.Select)
 
-	p := tryAtVtgate(sel)
+	p, err := handleDualSelects(sel, vschema)
+	if err != nil {
+		return nil, err
+	}
 	if p != nil {
 		return p, nil
 	}
@@ -136,9 +141,9 @@ func (pb *primitiveBuilder) processSelect(sel *sqlparser.Select, outer *symtab)
 	return nil
 }
 
-func tryAtVtgate(sel *sqlparser.Select) engine.Primitive {
+func handleDualSelects(sel *sqlparser.Select, vschema ContextVSchema) (engine.Primitive, error) {
 	if !isOnlyDual(sel.From) {
-		return nil
+		return nil, nil
 	}
 
 	exprs := make([]evalengine.Expr, len(sel.SelectExprs))
@@ -146,12 +151,17 @@ func tryAtVtgate(sel *sqlparser.Select) engine.Primitive {
 	for i, e := range sel.SelectExprs {
 		expr, ok := e.(*sqlparser.AliasedExpr)
 		if !ok {
-			return nil
+			return nil, nil
 		}
 		var err error
+		if sqlparser.IsLockingFunc(expr.Expr) {
+			// if we are using any locking functions, we bail out here and send the whole query to a single destination
+			return buildLockingPrimitive(sel, vschema)
+
+		}
 		exprs[i], err = sqlparser.Convert(expr.Expr)
 		if err != nil {
-			return nil
+			return nil, nil
 		}
 		cols[i] = expr.As.String()
 		if cols[i] == "" {
@@ -162,7 +172,23 @@ func tryAtVtgate(sel *sqlparser.Select) engine.Primitive {
 		Exprs: exprs,
 		Cols:  cols,
 		Input: &engine.SingleRow{},
+	}, nil
+}
+
+func buildLockingPrimitive(sel *sqlparser.Select, vschema ContextVSchema) (engine.Primitive, error) {
+	ks, err := vschema.FirstSortedKeyspace()
+	if err != nil {
+		return nil, err
 	}
+	return &engine.Reserve{
+		Input: &engine.Send{
+			Keyspace:          ks,
+			TargetDestination: key.DestinationKeyspaceID{0},
+			Query:             sqlparser.String(sel),
+			IsDML:             false,
+			SingleShardOnly:   true,
+		},
+	}, nil
 }
 
 func isOnlyDual(from sqlparser.TableExprs) bool {
diff --git a/go/vt/vtgate/safe_session.go b/go/vt/vtgate/safe_session.go
index 29329cc44f..69821a5b98 100644
--- a/go/vt/vtgate/safe_session.go
+++ b/go/vt/vtgate/safe_session.go
@@ -193,7 +193,7 @@ func addOrUpdate(shardSession *vtgatepb.Session_ShardSession, sessions []*vtgate
 			sess.Target.TabletType == shardSession.Target.TabletType &&
 			sess.Target.Shard == shardSession.Target.Shard
 		if targetedAtSameTablet {
-			if sess.TabletAlias != shardSession.TabletAlias {
+			if sess.TabletAlias.Cell != shardSession.TabletAlias.Cell || sess.TabletAlias.Uid != shardSession.TabletAlias.Uid {
 				return nil, vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, "got a different alias for the same target")
 			}
 			// replace the old info with the new one
@@ -340,3 +340,10 @@ func (session *SafeSession) InReservedConn() bool {
 	defer session.mu.Unlock()
 	return session.Session.InReservedConn
 }
+
+//SetReservedConn set the InReservedConn setting.
+func (session *SafeSession) SetReservedConn(reservedConn bool) {
+	session.mu.Lock()
+	defer session.mu.Unlock()
+	session.Session.InReservedConn = reservedConn
+}
diff --git a/go/vt/vtgate/scatter_conn.go b/go/vt/vtgate/scatter_conn.go
index 14e24b166e..b01043871e 100644
--- a/go/vt/vtgate/scatter_conn.go
+++ b/go/vt/vtgate/scatter_conn.go
@@ -245,12 +245,12 @@ func (stc *ScatterConn) ExecuteMultiShard(
 
 func getQueryService(rs *srvtopo.ResolvedShard, info *shardActionInfo) (queryservice.QueryService, error) {
 	_, usingLegacyGw := rs.Gateway.(*DiscoveryGateway)
-	//if usingLegacyGw {
-	//	switch info.actionNeeded {
-	//	case reserve, reserveBegin:
-	//		return nil, vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, "reserved connections are not supported on old gen gateway")
-	//	}
-	//}
+	if usingLegacyGw {
+		switch info.actionNeeded {
+		case reserve, reserveBegin:
+			return nil, vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, "reserved connections are not supported on old gen gateway")
+		}
+	}
 	if usingLegacyGw || info.alias == nil {
 		return rs.Gateway, nil
 	}
diff --git a/go/vt/vtgate/tabletgateway.go b/go/vt/vtgate/tabletgateway.go
index 423b0f53a9..6d846761f3 100644
--- a/go/vt/vtgate/tabletgateway.go
+++ b/go/vt/vtgate/tabletgateway.go
@@ -183,8 +183,8 @@ func (gw *TabletGateway) CacheStatus() TabletCacheStatusList {
 func (gw *TabletGateway) withRetry(ctx context.Context, target *querypb.Target, _ queryservice.QueryService,
 	_ string, inTransaction bool, inner func(ctx context.Context, target *querypb.Target, conn queryservice.QueryService) (bool, error)) error {
 	// for transactions, we connect to a specific tablet instead of letting gateway choose one
-	if inTransaction {
-		return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "gateway's query service can only be used for non-transactional queries")
+	if inTransaction && target.TabletType != topodatapb.TabletType_MASTER {
+		return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "gateway's query service can only be used for non-transactional queries on replicas")
 	}
 	var tabletLastUsed *topodatapb.Tablet
 	var err error
diff --git a/go/vt/vtgate/tx_conn.go b/go/vt/vtgate/tx_conn.go
index 2dd50364c3..b3ee328342 100644
--- a/go/vt/vtgate/tx_conn.go
+++ b/go/vt/vtgate/tx_conn.go
@@ -134,6 +134,7 @@ func (txc *TxConn) commitNormal(ctx context.Context, session *SafeSession) error
 	return nil
 }
 
+// commit2PC will not used the pinned tablets - to make sure we use the current source, we need to use the gateway's queryservice
 func (txc *TxConn) commit2PC(ctx context.Context, session *SafeSession) error {
 	if len(session.PreSessions) != 0 || len(session.PostSessions) != 0 {
 		_ = txc.Rollback(ctx, session)
@@ -159,13 +160,7 @@ func (txc *TxConn) commit2PC(ctx context.Context, session *SafeSession) error {
 	}
 
 	err = txc.runSessions(ctx, session.ShardSessions[1:], func(ctx context.Context, s *vtgatepb.Session_ShardSession) error {
-		var qs queryservice.QueryService
-		var err error
-		qs, err = txc.queryService(s.TabletAlias)
-		if err != nil {
-			return err
-		}
-		return qs.Prepare(ctx, s.Target, s.TransactionId, dtid)
+		return txc.gateway.Prepare(ctx, s.Target, s.TransactionId, dtid)
 	})
 	if err != nil {
 		// TODO(sougou): Perform a more fine-grained cleanup
@@ -177,11 +172,7 @@ func (txc *TxConn) commit2PC(ctx context.Context, session *SafeSession) error {
 		return err
 	}
 
-	qs, err := txc.queryService(mmShard.TabletAlias)
-	if err != nil {
-		return err
-	}
-	err = qs.StartCommit(ctx, mmShard.Target, mmShard.TransactionId, dtid)
+	err = txc.gateway.StartCommit(ctx, mmShard.Target, mmShard.TransactionId, dtid)
 	if err != nil {
 		return err
 	}
diff --git a/go/vt/vtgate/vcursor_impl.go b/go/vt/vtgate/vcursor_impl.go
index d76f29d7c2..344b58eb73 100644
--- a/go/vt/vtgate/vcursor_impl.go
+++ b/go/vt/vtgate/vcursor_impl.go
@@ -270,6 +270,20 @@ func (vc *vcursorImpl) AnyKeyspace() (*vindexes.Keyspace, error) {
 	return keyspace, nil
 }
 
+func (vc *vcursorImpl) FirstSortedKeyspace() (*vindexes.Keyspace, error) {
+	if len(vc.vschema.Keyspaces) == 0 {
+		return nil, vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, "no keyspaces available")
+	}
+	kss := vc.vschema.Keyspaces
+	keys := make([]string, 0, len(kss))
+	for ks := range kss {
+		keys = append(keys, ks)
+	}
+	sort.Strings(keys)
+
+	return kss[keys[0]].Keyspace, nil
+}
+
 // TargetString returns the current TargetString of the session.
 func (vc *vcursorImpl) TargetString() string {
 	return vc.safeSession.TargetString
@@ -389,6 +403,11 @@ func (vc *vcursorImpl) SetSysVar(name string, expr string) {
 	vc.safeSession.SetSystemVariable(name, expr)
 }
 
+//NeedsReservedConn implements the SessionActions interface
+func (vc *vcursorImpl) NeedsReservedConn() {
+	vc.safeSession.SetReservedConn(true)
+}
+
 // Destination implements the ContextVSchema interface
 func (vc *vcursorImpl) Destination() key.Destination {
 	return vc.destination
diff --git a/go/vt/vttablet/queryservice/wrapped.go b/go/vt/vttablet/queryservice/wrapped.go
index 039d095851..66a88188ac 100644
--- a/go/vt/vttablet/queryservice/wrapped.go
+++ b/go/vt/vttablet/queryservice/wrapped.go
@@ -203,7 +203,7 @@ func (ws *wrappedService) StreamExecute(ctx context.Context, target *querypb.Tar
 }
 
 func (ws *wrappedService) ExecuteBatch(ctx context.Context, target *querypb.Target, queries []*querypb.BoundQuery, asTransaction bool, transactionID int64, options *querypb.ExecuteOptions) (qrs []sqltypes.Result, err error) {
-	inTransaction := (transactionID != 0)
+	inTransaction := transactionID != 0
 	err = ws.wrapper(ctx, target, ws.impl, "ExecuteBatch", inTransaction, func(ctx context.Context, target *querypb.Target, conn QueryService) (bool, error) {
 		var innerErr error
 		qrs, innerErr = conn.ExecuteBatch(ctx, target, queries, asTransaction, transactionID, options)
@@ -294,14 +294,14 @@ func (ws *wrappedService) ReserveBeginExecute(ctx context.Context, target *query
 }
 
 func (ws *wrappedService) ReserveExecute(ctx context.Context, target *querypb.Target, preQueries []string, sql string, bindVariables map[string]*querypb.BindVariable, transactionID int64, options *querypb.ExecuteOptions) (*sqltypes.Result, int64, *topodatapb.TabletAlias, error) {
-	notInTransaction := (transactionID == 0)
+	inTransaction := transactionID != 0
 	var res *sqltypes.Result
 	var reservedID int64
 	var alias *topodatapb.TabletAlias
-	err := ws.wrapper(ctx, target, ws.impl, "ReserveExecute", false, func(ctx context.Context, target *querypb.Target, conn QueryService) (bool, error) {
+	err := ws.wrapper(ctx, target, ws.impl, "ReserveExecute", inTransaction, func(ctx context.Context, target *querypb.Target, conn QueryService) (bool, error) {
 		var err error
 		res, reservedID, alias, err = conn.ReserveExecute(ctx, target, preQueries, sql, bindVariables, transactionID, options)
-		return canRetry(ctx, err) && notInTransaction, err
+		return canRetry(ctx, err) && !inTransaction, err
 	})
 
 	return res, reservedID, alias, err
diff --git a/go/vt/vttablet/sandboxconn/sandboxconn.go b/go/vt/vttablet/sandboxconn/sandboxconn.go
index 1c1b000a96..15a2b4d7df 100644
--- a/go/vt/vttablet/sandboxconn/sandboxconn.go
+++ b/go/vt/vttablet/sandboxconn/sandboxconn.go
@@ -460,8 +460,10 @@ func (sbc *SandboxConn) reserve(ctx context.Context, target *querypb.Target, pre
 	for _, query := range preQueries {
 		sbc.Execute(ctx, target, query, bindVariables, transactionID, 0, options)
 	}
-	reservedID := sbc.ReserveID.Add(1)
-	return reservedID
+	if transactionID != 0 {
+		return transactionID
+	}
+	return sbc.ReserveID.Add(1)
 }
 
 //Release implements the QueryService interface
diff --git a/go/vt/vttablet/tabletserver/planbuilder/plan.go b/go/vt/vttablet/tabletserver/planbuilder/plan.go
index 7714e36ad6..fda8de64ab 100644
--- a/go/vt/vttablet/tabletserver/planbuilder/plan.go
+++ b/go/vt/vttablet/tabletserver/planbuilder/plan.go
@@ -151,12 +151,12 @@ func (plan *Plan) TableName() sqlparser.TableIdent {
 }
 
 // Build builds a plan based on the schema.
-func Build(statement sqlparser.Statement, tables map[string]*schema.Table) (*Plan, error) {
-	var plan *Plan
-
-	err := checkForPoolingUnsafeConstructs(statement)
-	if err != nil {
-		return nil, err
+func Build(statement sqlparser.Statement, tables map[string]*schema.Table, isReservedConn bool) (plan *Plan, err error) {
+	if !isReservedConn {
+		err = checkForPoolingUnsafeConstructs(statement)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	switch stmt := statement.(type) {
@@ -203,15 +203,17 @@ func Build(statement sqlparser.Statement, tables map[string]*schema.Table) (*Pla
 }
 
 // BuildStreaming builds a streaming plan based on the schema.
-func BuildStreaming(sql string, tables map[string]*schema.Table) (*Plan, error) {
+func BuildStreaming(sql string, tables map[string]*schema.Table, isReservedConn bool) (*Plan, error) {
 	statement, err := sqlparser.Parse(sql)
 	if err != nil {
 		return nil, err
 	}
 
-	err = checkForPoolingUnsafeConstructs(statement)
-	if err != nil {
-		return nil, err
+	if !isReservedConn {
+		err = checkForPoolingUnsafeConstructs(statement)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	plan := &Plan{
@@ -260,8 +262,8 @@ func BuildMessageStreaming(name string, tables map[string]*schema.Table) (*Plan,
 func checkForPoolingUnsafeConstructs(expr sqlparser.SQLNode) error {
 	return sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
 		if f, ok := node.(*sqlparser.FuncExpr); ok {
-			if f.Name.Lowered() == "get_lock" {
-				return false, vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, "get_lock() not allowed")
+			if sqlparser.IsLockingFunc(f) {
+				return false, vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "%s() not allowed", f.Name.String())
 			}
 		}
 
diff --git a/go/vt/vttablet/tabletserver/query_engine.go b/go/vt/vttablet/tabletserver/query_engine.go
index 1ce97b70b8..8f52d5e222 100644
--- a/go/vt/vttablet/tabletserver/query_engine.go
+++ b/go/vt/vttablet/tabletserver/query_engine.go
@@ -285,7 +285,7 @@ func (qe *QueryEngine) Close() {
 }
 
 // GetPlan returns the TabletPlan that for the query. Plans are cached in a cache.LRUCache.
-func (qe *QueryEngine) GetPlan(ctx context.Context, logStats *tabletenv.LogStats, sql string, skipQueryPlanCache bool) (*TabletPlan, error) {
+func (qe *QueryEngine) GetPlan(ctx context.Context, logStats *tabletenv.LogStats, sql string, skipQueryPlanCache bool, isReservedConn bool) (*TabletPlan, error) {
 	span, ctx := trace.NewSpan(ctx, "QueryEngine.GetPlan")
 	defer span.Finish()
 
@@ -305,7 +305,7 @@ func (qe *QueryEngine) GetPlan(ctx context.Context, logStats *tabletenv.LogStats
 	if err != nil {
 		return nil, err
 	}
-	splan, err := planbuilder.Build(statement, qe.tables)
+	splan, err := planbuilder.Build(statement, qe.tables, isReservedConn)
 	if err != nil {
 		return nil, err
 	}
@@ -340,10 +340,10 @@ func (qe *QueryEngine) GetPlan(ctx context.Context, logStats *tabletenv.LogStats
 
 // GetStreamPlan is similar to GetPlan, but doesn't use the cache
 // and doesn't enforce a limit. It just returns the parsed query.
-func (qe *QueryEngine) GetStreamPlan(sql string) (*TabletPlan, error) {
+func (qe *QueryEngine) GetStreamPlan(sql string, isReservedConn bool) (*TabletPlan, error) {
 	qe.mu.RLock()
 	defer qe.mu.RUnlock()
-	splan, err := planbuilder.BuildStreaming(sql, qe.tables)
+	splan, err := planbuilder.BuildStreaming(sql, qe.tables, isReservedConn)
 	if err != nil {
 		return nil, err
 	}
diff --git a/go/vt/vttablet/tabletserver/tabletserver.go b/go/vt/vttablet/tabletserver/tabletserver.go
index ae1dfa4c17..d773fca1eb 100644
--- a/go/vt/vttablet/tabletserver/tabletserver.go
+++ b/go/vt/vttablet/tabletserver/tabletserver.go
@@ -629,7 +629,7 @@ func (tsv *TabletServer) Execute(ctx context.Context, target *querypb.Target, sq
 				bindVariables = make(map[string]*querypb.BindVariable)
 			}
 			query, comments := sqlparser.SplitMarginComments(sql)
-			plan, err := tsv.qe.GetPlan(ctx, logStats, query, skipQueryPlanCache(options))
+			plan, err := tsv.qe.GetPlan(ctx, logStats, query, skipQueryPlanCache(options), reservedID != 0)
 			if err != nil {
 				return err
 			}
@@ -676,7 +676,8 @@ func (tsv *TabletServer) StreamExecute(ctx context.Context, target *querypb.Targ
 				bindVariables = make(map[string]*querypb.BindVariable)
 			}
 			query, comments := sqlparser.SplitMarginComments(sql)
-			plan, err := tsv.qe.GetStreamPlan(query)
+			// TODO: update the isReservedConn logic when StreamExecute supports reserved connections.
+			plan, err := tsv.qe.GetStreamPlan(query, false /* isReservedConn */)
 			if err != nil {
 				return err
 			}
@@ -857,7 +858,7 @@ func (tsv *TabletServer) beginWaitForSameRangeTransactions(ctx context.Context,
 func (tsv *TabletServer) computeTxSerializerKey(ctx context.Context, logStats *tabletenv.LogStats, sql string, bindVariables map[string]*querypb.BindVariable) (string, string) {
 	// Strip trailing comments so we don't pollute the query cache.
 	sql, _ = sqlparser.SplitMarginComments(sql)
-	plan, err := tsv.qe.GetPlan(ctx, logStats, sql, false /* skipQueryPlanCache */)
+	plan, err := tsv.qe.GetPlan(ctx, logStats, sql, false /* skipQueryPlanCache */, false /* isReservedConn */)
 	if err != nil {
 		logComputeRowSerializerKey.Errorf("failed to get plan for query: %v err: %v", sql, err)
 		return "", ""


Test patch:
diff --git a/go/test/endtoend/cellalias/cell_alias_test.go b/go/test/endtoend/cellalias/cell_alias_test.go
index b4bf8804f9..0d905b3d72 100644
--- a/go/test/endtoend/cellalias/cell_alias_test.go
+++ b/go/test/endtoend/cellalias/cell_alias_test.go
@@ -322,6 +322,7 @@ func TestAddAliasWhileVtgateUp(t *testing.T) {
 	vtgateInstance := localCluster.NewVtgateInstance()
 	vtgateInstance.CellsToWatch = allCells
 	vtgateInstance.TabletTypesToWait = "MASTER,REPLICA,RDONLY"
+	vtgateInstance.GatewayImplementation = "discoverygateway"
 	err = vtgateInstance.Setup()
 	require.Nil(t, err)
 	defer vtgateInstance.TearDown()
diff --git a/go/test/endtoend/cluster/vtgate_process.go b/go/test/endtoend/cluster/vtgate_process.go
index bd1eb7bee8..3c412eacc4 100644
--- a/go/test/endtoend/cluster/vtgate_process.go
+++ b/go/test/endtoend/cluster/vtgate_process.go
@@ -222,7 +222,7 @@ func VtgateProcessInstance(port int, grpcPort int, mySQLServerPort int, cell str
 		Cell:                  cell,
 		CellsToWatch:          cellsToWatch,
 		TabletTypesToWait:     tabletTypesToWait,
-		GatewayImplementation: "discoverygateway",
+		GatewayImplementation: "tabletgateway",
 		CommonArg:             *vtctl,
 		PidFile:               path.Join(tmpDirectory, "/vtgate.pid"),
 		MySQLAuthServerImpl:   "none",
diff --git a/go/test/endtoend/vtgate/get_lock_test.go b/go/test/endtoend/vtgate/get_lock_test.go
new file mode 100644
index 0000000000..75368e02ff
--- /dev/null
+++ b/go/test/endtoend/vtgate/get_lock_test.go
@@ -0,0 +1,230 @@
+/*
+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 vtgate
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"vitess.io/vitess/go/sync2"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"vitess.io/vitess/go/mysql"
+)
+
+func TestLockUnlock(t *testing.T) {
+	conn, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	assertMatches(t, conn, `select release_lock('lock name')`, `[[NULL]]`)
+	assertMatches(t, conn, `select get_lock('lock name', 2)`, `[[INT64(1)]]`)
+	assertMatches(t, conn, `select get_lock('lock name', 2)`, `[[INT64(1)]]`)
+	assertMatches(t, conn, `select is_free_lock('lock name')`, `[[INT64(0)]]`)
+	assert.NotEmpty(t,
+		exec(t, conn, `select is_used_lock('lock name')`))
+	assertMatches(t, conn, `select release_lock('lock name')`, `[[INT64(1)]]`)
+	assertMatches(t, conn, `select release_all_locks()`, `[[UINT64(1)]]`)
+	assertMatches(t, conn, `select release_lock('lock name')`, `[[NULL]]`)
+}
+
+func TestLocksDontIntersect(t *testing.T) {
+	conn1, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn1.Close()
+	conn2, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn2.Close()
+
+	assertMatches(t, conn1, `select get_lock('lock1', 2)`, `[[INT64(1)]]`)
+	assertMatches(t, conn2, `select get_lock('lock2', 2)`, `[[INT64(1)]]`)
+	assertMatches(t, conn1, `select release_lock('lock1')`, `[[INT64(1)]]`)
+	assertMatches(t, conn2, `select release_lock('lock2')`, `[[INT64(1)]]`)
+}
+
+func TestLocksIntersect(t *testing.T) {
+	conn1, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn1.Close()
+	conn2, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn2.Close()
+
+	assertMatches(t, conn1, `select get_lock('lock1', 100)`, `[[INT64(1)]]`)
+	assertMatches(t, conn2, `select get_lock('lock2', 100)`, `[[INT64(1)]]`)
+
+	// Locks will not succeed.
+	assertMatches(t, conn1, `select get_lock('lock2', 1)`, `[[INT64(0)]]`)
+	assertMatches(t, conn2, `select get_lock('lock1', 1)`, `[[INT64(0)]]`)
+	assertMatches(t, conn1, `select release_lock('lock2')`, `[[INT64(0)]]`)
+	assertMatches(t, conn2, `select release_lock('lock1')`, `[[INT64(0)]]`)
+
+	assertMatches(t, conn1, `select release_lock('lock1')`, `[[INT64(1)]]`)
+	assertMatches(t, conn2, `select release_lock('lock2')`, `[[INT64(1)]]`)
+}
+
+func TestLocksAreExplicitlyReleaseAndRegrab(t *testing.T) {
+	conn1, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn1.Close()
+	conn2, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn2.Close()
+
+	assertMatches(t, conn1, `select get_lock('lock', 2)`, `[[INT64(1)]]`)
+	assertMatches(t, conn1, `select release_lock('lock')`, `[[INT64(1)]]`)
+	assertMatches(t, conn2, `select get_lock('lock', 2)`, `[[INT64(1)]]`)
+}
+
+func TestLocksAreReleasedWhenConnectionIsClosed(t *testing.T) {
+	conn1, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn1.Close()
+	conn2, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn2.Close()
+
+	assertMatches(t, conn1, `select get_lock('lock', 2)`, `[[INT64(1)]]`)
+	conn1.Close()
+
+	assertMatches(t, conn2, `select get_lock('lock', 2)`, `[[INT64(1)]]`)
+}
+
+func TestLocksBlockEachOther(t *testing.T) {
+	conn1, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn1.Close()
+
+	// in the first connection, grab a lock
+	assertMatches(t, conn1, `select get_lock('lock', 2)`, `[[INT64(1)]]`)
+
+	released := sync2.NewAtomicBool(false)
+
+	go func() {
+		conn2, err := mysql.Connect(context.Background(), &vtParams)
+		require.NoError(t, err)
+		defer conn2.Close()
+
+		// in the second connection, we try to grab a lock, and should get blocked
+		assertMatches(t, conn2, `select get_lock('lock', 2)`, `[[INT64(1)]]`)
+		assert.True(t, released.Get(), "was not blocked by get_lock")
+		assertMatches(t, conn2, `select release_lock('lock')`, `[[INT64(1)]]`)
+	}()
+
+	time.Sleep(1 * time.Second)
+
+	released.Set(true)
+	assertMatches(t, conn1, `select release_lock('lock')`, `[[INT64(1)]]`)
+}
+
+func TestLocksBlocksWithTx(t *testing.T) {
+	conn1, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn1.Close()
+
+	// in the first connection, grab a lock
+	assertMatches(t, conn1, `select get_lock('lock', 2)`, `[[INT64(1)]]`)
+	exec(t, conn1, "begin")
+	exec(t, conn1, "insert into t1(id1, id2) values(1,1)") // -80
+	exec(t, conn1, "commit")
+
+	released := sync2.NewAtomicBool(false)
+
+	go func() {
+		conn2, err := mysql.Connect(context.Background(), &vtParams)
+		require.NoError(t, err)
+		defer conn2.Close()
+
+		// in the second connection, we try to grab a lock, and should get blocked
+		assertMatches(t, conn2, `select get_lock('lock', 2)`, `[[INT64(1)]]`)
+		assert.True(t, released.Get(), "was not blocked by get_lock")
+		assertMatches(t, conn2, `select release_lock('lock')`, `[[INT64(1)]]`)
+	}()
+
+	time.Sleep(1 * time.Second)
+
+	released.Set(true)
+	assertMatches(t, conn1, `select release_lock('lock')`, `[[INT64(1)]]`)
+	exec(t, conn1, "delete from t1")
+}
+
+func TestLocksWithTxFailure(t *testing.T) {
+	conn1, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn1.Close()
+
+	conn2, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn2.Close()
+
+	// in the first connection, grab a lock for infinite time
+	assertMatches(t, conn1, `select get_lock('lock', -1)`, `[[INT64(1)]]`)
+
+	exec(t, conn1, "use `ks:80-`")
+	exec(t, conn1, "begin")
+	qr := exec(t, conn1, "select connection_id()")
+	exec(t, conn1, "use ks")
+	// kill the mysql connection shard which has transaction open.
+	vttablet1 := clusterInstance.Keyspaces[0].Shards[1].MasterTablet() // 80-
+	vttablet1.VttabletProcess.QueryTablet(fmt.Sprintf("kill %s", qr.Rows[0][0].ToString()), KeyspaceName, false)
+
+	// transaction fails on commit.
+	_, err = conn1.ExecuteFetch("commit", 1, true)
+	require.Error(t, err)
+
+	// in the second connection, lock acquisition succeeds.
+	assertMatches(t, conn2, `select get_lock('lock', 2)`, `[[INT64(1)]]`)
+	assertMatches(t, conn2, `select release_lock('lock')`, `[[INT64(1)]]`)
+}
+
+func TestLocksWithTxOngoingAndReleaseLock(t *testing.T) {
+	conn1, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn1.Close()
+
+	assertMatches(t, conn1, `select get_lock('lock', -1)`, `[[INT64(1)]]`)
+	exec(t, conn1, "begin")
+	exec(t, conn1, "insert into t1(id1, id2) values(1,1)")
+	assertMatches(t, conn1, `select release_lock('lock')`, `[[INT64(1)]]`)
+	assertMatches(t, conn1, `select id1, id2 from t1 where id1 = 1`, `[[INT64(1) INT64(1)]]`)
+	exec(t, conn1, "rollback")
+	assertIsEmpty(t, conn1, `select id1, id2 from t1 where id1 = 1`)
+}
+
+func TestLocksWithTxOngoingAndLockFails(t *testing.T) {
+	conn1, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn1.Close()
+
+	conn2, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn2.Close()
+
+	assertMatches(t, conn2, `select get_lock('lock', -1)`, `[[INT64(1)]]`)
+
+	exec(t, conn1, "begin")
+	exec(t, conn1, "insert into t1(id1, id2) values(1,1)")
+	assertMatches(t, conn1, `select get_lock('lock', 1)`, `[[INT64(0)]]`)
+	assertMatches(t, conn1, `select id1, id2 from t1 where id1 = 1`, `[[INT64(1) INT64(1)]]`)
+	exec(t, conn1, "rollback")
+	assertIsEmpty(t, conn1, `select id1, id2 from t1 where id1 = 1`)
+
+	assertMatches(t, conn2, `select get_lock('lock', -1)`, `[[INT64(1)]]`)
+}
diff --git a/go/vt/vtgate/autocommit_test.go b/go/vt/vtgate/autocommit_test.go
index 9f00225a05..f0bed579fa 100644
--- a/go/vt/vtgate/autocommit_test.go
+++ b/go/vt/vtgate/autocommit_test.go
@@ -36,7 +36,7 @@ import (
 
 // TestAutocommitUpdateSharded: instant-commit.
 func TestAutocommitUpdateSharded(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	_, err := autocommitExec(executor, "update user set a=2 where id = 1")
 	require.NoError(t, err)
@@ -53,7 +53,7 @@ func TestAutocommitUpdateSharded(t *testing.T) {
 
 // TestAutocommitUpdateLookup: transaction: select before update.
 func TestAutocommitUpdateLookup(t *testing.T) {
-	executor, sbc1, _, sbclookup := createExecutorEnv()
+	executor, sbc1, _, sbclookup := createLegacyExecutorEnv()
 	sbclookup.SetResults([]*sqltypes.Result{sqltypes.MakeTestResult(
 		sqltypes.MakeTestFields("b|a", "int64|varbinary"),
 		"2|1",
@@ -82,7 +82,7 @@ func TestAutocommitUpdateLookup(t *testing.T) {
 
 // TestAutocommitUpdateVindexChange: transaction: select & update before final update.
 func TestAutocommitUpdateVindexChange(t *testing.T) {
-	executor, sbc, _, sbclookup := createExecutorEnv()
+	executor, sbc, _, sbclookup := createLegacyExecutorEnv()
 	sbc.SetResults([]*sqltypes.Result{sqltypes.MakeTestResult(
 		sqltypes.MakeTestFields("id|name|lastname", "int64|int32|varchar"),
 		"1|1|foo",
@@ -121,7 +121,7 @@ func TestAutocommitUpdateVindexChange(t *testing.T) {
 
 // TestAutocommitDeleteSharded: instant-commit.
 func TestAutocommitDeleteSharded(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	_, err := autocommitExec(executor, "delete from user_extra where user_id = 1")
 	require.NoError(t, err)
@@ -138,7 +138,7 @@ func TestAutocommitDeleteSharded(t *testing.T) {
 
 // TestAutocommitDeleteLookup: transaction: select before update.
 func TestAutocommitDeleteLookup(t *testing.T) {
-	executor, sbc1, _, sbclookup := createExecutorEnv()
+	executor, sbc1, _, sbclookup := createLegacyExecutorEnv()
 	sbc1.SetResults([]*sqltypes.Result{sqltypes.MakeTestResult(
 		sqltypes.MakeTestFields("id|name|lastname", "int64|int32|varchar"),
 		"1|1|foo",
@@ -180,7 +180,7 @@ func TestAutocommitDeleteLookup(t *testing.T) {
 
 // TestAutocommitDeleteIn: instant-commit.
 func TestAutocommitDeleteIn(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	_, err := autocommitExec(executor, "delete from user_extra where user_id in (1, 2)")
 	require.NoError(t, err)
@@ -197,7 +197,7 @@ func TestAutocommitDeleteIn(t *testing.T) {
 
 // TestAutocommitDeleteMultiShard: instant-commit.
 func TestAutocommitDeleteMultiShard(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	_, err := autocommitExec(executor, "delete from user_extra where user_id = user_id + 1")
 	require.NoError(t, err)
@@ -217,7 +217,7 @@ func TestAutocommitDeleteMultiShard(t *testing.T) {
 
 // TestAutocommitDeleteMultiShardAutoCommit: instant-commit.
 func TestAutocommitDeleteMultiShardAutoCommit(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	_, err := autocommitExec(executor, "delete /*vt+ MULTI_SHARD_AUTOCOMMIT=1 */ from user_extra where user_id = user_id + 1")
 	require.NoError(t, err)
@@ -237,7 +237,7 @@ func TestAutocommitDeleteMultiShardAutoCommit(t *testing.T) {
 
 // TestAutocommitInsertSharded: instant-commit.
 func TestAutocommitInsertSharded(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	_, err := autocommitExec(executor, "insert into user_extra(user_id, v) values (1, 2)")
 	require.NoError(t, err)
@@ -256,7 +256,7 @@ func TestAutocommitInsertSharded(t *testing.T) {
 
 // TestAutocommitInsertLookup: transaction: select before update.
 func TestAutocommitInsertLookup(t *testing.T) {
-	executor, sbc1, _, sbclookup := createExecutorEnv()
+	executor, sbc1, _, sbclookup := createLegacyExecutorEnv()
 
 	_, err := autocommitExec(executor, "insert into user(id, v, name) values (1, 2, 'myname')")
 	require.NoError(t, err)
@@ -283,7 +283,7 @@ func TestAutocommitInsertLookup(t *testing.T) {
 
 // TestAutocommitInsertShardAutoCommit: instant-commit.
 func TestAutocommitInsertMultishardAutoCommit(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	_, err := autocommitExec(executor, "insert /*vt+ MULTI_SHARD_AUTOCOMMIT=1 */ into user_extra(user_id, v) values (1, 2), (3, 4)")
 	require.NoError(t, err)
@@ -306,7 +306,7 @@ func TestAutocommitInsertMultishardAutoCommit(t *testing.T) {
 	}})
 	testCommitCount(t, "sbc2", sbc2, 0)
 
-	executor, sbc1, sbc2, _ = createExecutorEnv()
+	executor, sbc1, sbc2, _ = createLegacyExecutorEnv()
 	// Make the first shard fail - the second completes anyway
 	sbc1.MustFailCodes[vtrpcpb.Code_INVALID_ARGUMENT] = 1
 	_, err = autocommitExec(executor, "insert /*vt+ MULTI_SHARD_AUTOCOMMIT=1 */ into user_extra(user_id, v) values (1, 2), (3, 4)")
@@ -327,7 +327,7 @@ func TestAutocommitInsertMultishardAutoCommit(t *testing.T) {
 }
 
 func TestAutocommitInsertMultishard(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	_, err := autocommitExec(executor, "insert into user_extra(user_id, v) values (1, 2), (3, 4)")
 	require.NoError(t, err)
@@ -353,7 +353,7 @@ func TestAutocommitInsertMultishard(t *testing.T) {
 
 // TestAutocommitInsertAutoinc: instant-commit: sequence fetch is not transactional.
 func TestAutocommitInsertAutoinc(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 
 	_, err := autocommitExec(executor, "insert into main1(id, name) values (null, 'myname')")
 	require.NoError(t, err)
@@ -372,7 +372,7 @@ func TestAutocommitInsertAutoinc(t *testing.T) {
 
 // TestAutocommitTransactionStarted: no instant-commit.
 func TestAutocommitTransactionStarted(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 
 	session := &vtgatepb.Session{
 		TargetString:    "@master",
@@ -394,7 +394,7 @@ func TestAutocommitTransactionStarted(t *testing.T) {
 
 // TestAutocommitDirectTarget: instant-commit.
 func TestAutocommitDirectTarget(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 
 	session := &vtgatepb.Session{
 		TargetString:    "TestUnsharded/0@master",
@@ -415,7 +415,7 @@ func TestAutocommitDirectTarget(t *testing.T) {
 
 // TestAutocommitDirectRangeTarget: no instant-commit.
 func TestAutocommitDirectRangeTarget(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 
 	session := &vtgatepb.Session{
 		TargetString:    "TestExecutor[-]@master",
diff --git a/go/vt/vtgate/engine/fake_vcursor_test.go b/go/vt/vtgate/engine/fake_vcursor_test.go
index 2683863501..1a07f32691 100644
--- a/go/vt/vtgate/engine/fake_vcursor_test.go
+++ b/go/vt/vtgate/engine/fake_vcursor_test.go
@@ -54,6 +54,9 @@ type noopVCursor struct {
 	ctx context.Context
 }
 
+func (t noopVCursor) NeedsReservedConn() {
+}
+
 func (t noopVCursor) SetUDV(key string, value interface{}) error {
 	panic("implement me")
 }
@@ -165,6 +168,9 @@ func (f *loggingVCursor) SetSysVar(name string, expr string) {
 	f.log = append(f.log, fmt.Sprintf("SysVar set with (%s,%v)", name, expr))
 }
 
+func (f *loggingVCursor) NeedsReservedConn() {
+}
+
 func (f *loggingVCursor) ExecuteVSchema(keyspace string, vschemaDDL *sqlparser.DDL) error {
 	panic("implement me")
 }
diff --git a/go/vt/vtgate/executor_dml_test.go b/go/vt/vtgate/executor_dml_test.go
index dd4c471adc..942cede020 100644
--- a/go/vt/vtgate/executor_dml_test.go
+++ b/go/vt/vtgate/executor_dml_test.go
@@ -38,7 +38,7 @@ import (
 )
 
 func TestUpdateEqual(t *testing.T) {
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
@@ -269,7 +269,7 @@ func TestUpdateMultiOwned(t *testing.T) {
 }
 
 func TestUpdateComments(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	_, err := executorExec(executor, "update user set a=2 where id = 1 /* trailing */", nil)
 	require.NoError(t, err)
@@ -286,7 +286,7 @@ func TestUpdateComments(t *testing.T) {
 }
 
 func TestUpdateNormalize(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	executor.normalize = true
 	_, err := executorExec(executor, "/* leading */ update user set a=2 where id = 1 /* trailing */", nil)
@@ -324,7 +324,7 @@ func TestUpdateNormalize(t *testing.T) {
 }
 
 func TestDeleteEqual(t *testing.T) {
-	executor, sbc, _, sbclookup := createExecutorEnv()
+	executor, sbc, _, sbclookup := createLegacyExecutorEnv()
 
 	sbc.SetResults([]*sqltypes.Result{{
 		Fields: []*querypb.Field{
@@ -457,7 +457,7 @@ func TestDeleteEqual(t *testing.T) {
 }
 
 func TestUpdateScatter(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	_, err := executorExec(executor, "update user_extra set col = 2", nil)
 	require.NoError(t, err)
 	// Queries get annotatted.
@@ -474,7 +474,7 @@ func TestUpdateScatter(t *testing.T) {
 }
 
 func TestDeleteScatter(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	_, err := executorExec(executor, "delete from user_extra", nil)
 	require.NoError(t, err)
 	// Queries get annotatted.
@@ -491,7 +491,7 @@ func TestDeleteScatter(t *testing.T) {
 }
 
 func TestDeleteByDestination(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	// This query is not supported in v3, so we know for sure is taking the DeleteByDestination route
 	_, err := executorExec(executor, "delete from `TestExecutor[-]`.user_extra limit 10", nil)
 	require.NoError(t, err)
@@ -509,7 +509,7 @@ func TestDeleteByDestination(t *testing.T) {
 }
 
 func TestDeleteComments(t *testing.T) {
-	executor, sbc, _, sbclookup := createExecutorEnv()
+	executor, sbc, _, sbclookup := createLegacyExecutorEnv()
 
 	sbc.SetResults([]*sqltypes.Result{{
 		Fields: []*querypb.Field{
@@ -549,7 +549,7 @@ func TestDeleteComments(t *testing.T) {
 }
 
 func TestInsertSharded(t *testing.T) {
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
@@ -630,7 +630,7 @@ func TestInsertSharded(t *testing.T) {
 }
 
 func TestInsertShardedKeyrange(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 
 	// If a unique vindex returns a keyrange, we fail the insert
 	_, err := executorExec(executor, "insert into keyrange_table(krcol_unique, krcol) values(1, 1)", nil)
@@ -718,7 +718,7 @@ func TestInsertShardedAutocommitLookup(t *testing.T) {
 }
 
 func TestInsertShardedIgnore(t *testing.T) {
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 
 	// Build the sequence of responses for sbclookup. This should
 	// match the sequence of queries we validate below.
@@ -882,7 +882,7 @@ func TestInsertShardedIgnore(t *testing.T) {
 func TestInsertOnDupKey(t *testing.T) {
 	// This test just sanity checks that the statement is getting passed through
 	// correctly. The full set of use cases are covered by TestInsertShardedIgnore.
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 	sbclookup.SetResults([]*sqltypes.Result{sqltypes.MakeTestResult(
 		sqltypes.MakeTestFields("b|a", "int64|varbinary"),
 		"1|1",
@@ -930,7 +930,7 @@ func TestInsertOnDupKey(t *testing.T) {
 }
 
 func TestAutocommitFail(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 
 	query := "insert into user (id) values (1)"
 	sbc1.MustFailCodes[vtrpcpb.Code_ALREADY_EXISTS] = 1
@@ -947,7 +947,7 @@ func TestAutocommitFail(t *testing.T) {
 }
 
 func TestInsertComments(t *testing.T) {
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 
 	_, err := executorExec(executor, "insert into user(id, v, name) values (1, 2, 'myname') /* trailing */", nil)
 	require.NoError(t, err)
@@ -978,7 +978,7 @@ func TestInsertComments(t *testing.T) {
 }
 
 func TestInsertGeneratorSharded(t *testing.T) {
-	executor, sbc, _, sbclookup := createExecutorEnv()
+	executor, sbc, _, sbclookup := createLegacyExecutorEnv()
 
 	sbclookup.SetResults([]*sqltypes.Result{{
 		Rows: [][]sqltypes.Value{{
@@ -1019,7 +1019,7 @@ func TestInsertGeneratorSharded(t *testing.T) {
 }
 
 func TestInsertAutoincSharded(t *testing.T) {
-	router, sbc, _, _ := createExecutorEnv()
+	router, sbc, _, _ := createLegacyExecutorEnv()
 
 	// Fake a mysql auto-inc response.
 	wantResult := &sqltypes.Result{
@@ -1048,7 +1048,7 @@ func TestInsertAutoincSharded(t *testing.T) {
 }
 
 func TestInsertGeneratorUnsharded(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 	result, err := executorExec(executor, "insert into main1(id, name) values (null, 'myname')", nil)
 	require.NoError(t, err)
 	wantQueries := []*querypb.BoundQuery{{
@@ -1071,7 +1071,7 @@ func TestInsertGeneratorUnsharded(t *testing.T) {
 }
 
 func TestInsertAutoincUnsharded(t *testing.T) {
-	router, _, _, sbclookup := createExecutorEnv()
+	router, _, _, sbclookup := createLegacyExecutorEnv()
 
 	// Fake a mysql auto-inc response.
 	query := "insert into simple(val) values ('val')"
@@ -1099,7 +1099,7 @@ func TestInsertAutoincUnsharded(t *testing.T) {
 }
 
 func TestInsertLookupOwned(t *testing.T) {
-	executor, sbc, _, sbclookup := createExecutorEnv()
+	executor, sbc, _, sbclookup := createLegacyExecutorEnv()
 
 	_, err := executorExec(executor, "insert into music(user_id, id) values (2, 3)", nil)
 	require.NoError(t, err)
@@ -1125,7 +1125,7 @@ func TestInsertLookupOwned(t *testing.T) {
 }
 
 func TestInsertLookupOwnedGenerator(t *testing.T) {
-	executor, sbc, _, sbclookup := createExecutorEnv()
+	executor, sbc, _, sbclookup := createLegacyExecutorEnv()
 
 	sbclookup.SetResults([]*sqltypes.Result{{
 		Rows: [][]sqltypes.Value{{
@@ -1168,7 +1168,7 @@ func TestInsertLookupOwnedGenerator(t *testing.T) {
 }
 
 func TestInsertLookupUnowned(t *testing.T) {
-	executor, sbc, _, sbclookup := createExecutorEnv()
+	executor, sbc, _, sbclookup := createLegacyExecutorEnv()
 
 	_, err := executorExec(executor, "insert into music_extra(user_id, music_id) values (2, 3)", nil)
 	require.NoError(t, err)
@@ -1195,7 +1195,7 @@ func TestInsertLookupUnowned(t *testing.T) {
 }
 
 func TestInsertLookupUnownedUnsupplied(t *testing.T) {
-	executor, sbc, _, sbclookup := createExecutorEnv()
+	executor, sbc, _, sbclookup := createLegacyExecutorEnv()
 	sbclookup.SetResults([]*sqltypes.Result{sqltypes.MakeTestResult(
 		sqltypes.MakeTestFields("b|a", "int64|varbinary"),
 		"3|1",
@@ -1228,7 +1228,7 @@ func TestInsertLookupUnownedUnsupplied(t *testing.T) {
 // If a statement gets broken up into two, and the first one fails,
 // then an error should be returned normally.
 func TestInsertPartialFail1(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 
 	// Make the first DML fail, there should be no rollback.
 	sbclookup.MustFailCodes[vtrpcpb.Code_INVALID_ARGUMENT] = 1
@@ -1250,7 +1250,7 @@ func TestInsertPartialFail1(t *testing.T) {
 // after successful execution of the first, then the transaction must
 // be rolled back due to partial execution.
 func TestInsertPartialFail2(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 
 	// Make the second DML fail, it should result in a rollback.
 	sbc1.MustFailCodes[vtrpcpb.Code_INVALID_ARGUMENT] = 1
@@ -1269,7 +1269,7 @@ func TestInsertPartialFail2(t *testing.T) {
 }
 
 func TestMultiInsertSharded(t *testing.T) {
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 
 	_, err := executorExec(executor, "insert into user(id, v, name) values (1, 1, 'myname1'),(3, 3, 'myname3')", nil)
 	require.NoError(t, err)
@@ -1390,7 +1390,7 @@ func TestMultiInsertSharded(t *testing.T) {
 }
 
 func TestMultiInsertGenerator(t *testing.T) {
-	executor, sbc, _, sbclookup := createExecutorEnv()
+	executor, sbc, _, sbclookup := createLegacyExecutorEnv()
 
 	sbclookup.SetResults([]*sqltypes.Result{{
 		Rows: [][]sqltypes.Value{{
@@ -1437,7 +1437,7 @@ func TestMultiInsertGenerator(t *testing.T) {
 }
 
 func TestMultiInsertGeneratorSparse(t *testing.T) {
-	executor, sbc, _, sbclookup := createExecutorEnv()
+	executor, sbc, _, sbclookup := createLegacyExecutorEnv()
 
 	sbclookup.SetResults([]*sqltypes.Result{{
 		Rows: [][]sqltypes.Value{{
@@ -1583,7 +1583,7 @@ func TestKeyDestRangeQuery(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.targetString+" - "+tc.inputQuery, func(t *testing.T) {
-			executor, sbc1, sbc2, _ := createExecutorEnv()
+			executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 			masterSession.TargetString = tc.targetString
 			_, err := executorExec(executor, tc.inputQuery, nil)
@@ -1604,7 +1604,7 @@ func TestKeyDestRangeQuery(t *testing.T) {
 	}
 
 	// it does not work for inserts
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	masterSession.TargetString = "TestExecutor[-]"
 	_, err := executorExec(executor, insertInput, nil)
 
@@ -1624,7 +1624,7 @@ func assertQueriesContain(t *testing.T, sql, sbcName string, sbc *sandboxconn.Sa
 
 // Prepared statement tests
 func TestUpdateEqualWithPrepare(t *testing.T) {
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
@@ -1648,7 +1648,7 @@ func TestUpdateEqualWithPrepare(t *testing.T) {
 	}
 }
 func TestInsertShardedWithPrepare(t *testing.T) {
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
@@ -1675,7 +1675,7 @@ func TestInsertShardedWithPrepare(t *testing.T) {
 }
 
 func TestDeleteEqualWithPrepare(t *testing.T) {
-	executor, sbc, _, sbclookup := createExecutorEnv()
+	executor, sbc, _, sbclookup := createLegacyExecutorEnv()
 	_, err := executorPrepare(executor, "delete from user where id = :id0", map[string]*querypb.BindVariable{
 		"id0": sqltypes.Int64BindVariable(1),
 	})
@@ -1693,7 +1693,7 @@ func TestDeleteEqualWithPrepare(t *testing.T) {
 }
 
 func TestUpdateLastInsertID(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 	executor.normalize = true
 
 	sql := "update user set a = last_insert_id() where id = 1"
diff --git a/go/vt/vtgate/executor_framework_test.go b/go/vt/vtgate/executor_framework_test.go
index 5e01f14341..c986c14805 100644
--- a/go/vt/vtgate/executor_framework_test.go
+++ b/go/vt/vtgate/executor_framework_test.go
@@ -336,7 +336,7 @@ func init() {
 	*GatewayImplementation = GatewayImplementationDiscovery
 }
 
-func createExecutorEnv() (executor *Executor, sbc1, sbc2, sbclookup *sandboxconn.SandboxConn) {
+func createLegacyExecutorEnv() (executor *Executor, sbc1, sbc2, sbclookup *sandboxconn.SandboxConn) {
 	// Use legacy gateway until we can rewrite these tests to use new tabletgateway
 	*GatewayImplementation = GatewayImplementationDiscovery
 	cell := "aa"
@@ -370,6 +370,40 @@ func createExecutorEnv() (executor *Executor, sbc1, sbc2, sbclookup *sandboxconn
 	return executor, sbc1, sbc2, sbclookup
 }
 
+func createExecutorEnv() (executor *Executor, sbc1, sbc2, sbclookup *sandboxconn.SandboxConn) {
+	// Use legacy gateway until we can rewrite these tests to use new tabletgateway
+	*GatewayImplementation = GatewayImplementationDiscovery
+	cell := "aa"
+	hc := discovery.NewFakeHealthCheck()
+	s := createSandbox("TestExecutor")
+	s.VSchema = executorVSchema
+	serv := newSandboxForCells([]string{cell})
+	resolver := newTestResolver(hc, serv, cell)
+	sbc1 = hc.AddTestTablet(cell, "-20", 1, "TestExecutor", "-20", topodatapb.TabletType_MASTER, true, 1, nil)
+	sbc2 = hc.AddTestTablet(cell, "40-60", 1, "TestExecutor", "40-60", topodatapb.TabletType_MASTER, true, 1, nil)
+	// Create these connections so scatter queries don't fail.
+	_ = hc.AddTestTablet(cell, "20-40", 1, "TestExecutor", "20-40", topodatapb.TabletType_MASTER, true, 1, nil)
+	_ = hc.AddTestTablet(cell, "60-60", 1, "TestExecutor", "60-80", topodatapb.TabletType_MASTER, true, 1, nil)
+	_ = hc.AddTestTablet(cell, "80-a0", 1, "TestExecutor", "80-a0", topodatapb.TabletType_MASTER, true, 1, nil)
+	_ = hc.AddTestTablet(cell, "a0-c0", 1, "TestExecutor", "a0-c0", topodatapb.TabletType_MASTER, true, 1, nil)
+	_ = hc.AddTestTablet(cell, "c0-e0", 1, "TestExecutor", "c0-e0", topodatapb.TabletType_MASTER, true, 1, nil)
+	_ = hc.AddTestTablet(cell, "e0-", 1, "TestExecutor", "e0-", topodatapb.TabletType_MASTER, true, 1, nil)
+
+	createSandbox(KsTestUnsharded)
+	sbclookup = hc.AddTestTablet(cell, "0", 1, KsTestUnsharded, "0", topodatapb.TabletType_MASTER, true, 1, nil)
+
+	// Ues the 'X' in the name to ensure it's not alphabetically first.
+	// Otherwise, it would become the default keyspace for the dual table.
+	bad := createSandbox("TestXBadSharding")
+	bad.VSchema = badVSchema
+
+	getSandbox(KsTestUnsharded).VSchema = unshardedVSchema
+	executor = NewExecutor(context.Background(), serv, cell, resolver, false, testBufferSize, testCacheSize)
+
+	key.AnyShardPicker = DestinationAnyShardPickerFirstShard{}
+	return executor, sbc1, sbc2, sbclookup
+}
+
 func createCustomExecutor(vschema string) (executor *Executor, sbc1, sbc2, sbclookup *sandboxconn.SandboxConn) {
 	cell := "aa"
 	hc := discovery.NewFakeLegacyHealthCheck()
@@ -548,3 +582,8 @@ func newTestLegacyResolver(hc discovery.LegacyHealthCheck, serv srvtopo.Server,
 	srvResolver := srvtopo.NewResolver(serv, sc.gateway, cell)
 	return NewResolver(srvResolver, serv, cell, sc)
 }
+func newTestResolver(hc discovery.HealthCheck, serv srvtopo.Server, cell string) *Resolver {
+	sc := newTestScatterConn(hc, serv, cell)
+	srvResolver := srvtopo.NewResolver(serv, sc.gateway, cell)
+	return NewResolver(srvResolver, serv, cell, sc)
+}
diff --git a/go/vt/vtgate/executor_scatter_stats_test.go b/go/vt/vtgate/executor_scatter_stats_test.go
index 9183df9a13..f1b18a6c08 100644
--- a/go/vt/vtgate/executor_scatter_stats_test.go
+++ b/go/vt/vtgate/executor_scatter_stats_test.go
@@ -28,7 +28,7 @@ import (
 )
 
 func TestScatterStatsWithNoScatterQuery(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master"})
 
 	_, err := executor.Execute(context.Background(), "TestExecutorResultsExceeded", session, "select * from main1", nil)
@@ -40,7 +40,7 @@ func TestScatterStatsWithNoScatterQuery(t *testing.T) {
 }
 
 func TestScatterStatsWithSingleScatterQuery(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master"})
 
 	_, err := executor.Execute(context.Background(), "TestExecutorResultsExceeded", session, "select * from user", nil)
@@ -52,7 +52,7 @@ func TestScatterStatsWithSingleScatterQuery(t *testing.T) {
 }
 
 func TestScatterStatsHttpWriting(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master"})
 
 	_, err := executor.Execute(context.Background(), "TestExecutorResultsExceeded", session, "select * from user", nil)
diff --git a/go/vt/vtgate/executor_select_test.go b/go/vt/vtgate/executor_select_test.go
index d2a087208a..f59cb0a2ca 100644
--- a/go/vt/vtgate/executor_select_test.go
+++ b/go/vt/vtgate/executor_select_test.go
@@ -42,7 +42,7 @@ import (
 )
 
 func TestSelectNext(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 
 	query := "select next :n values from user_seq"
 	bv := map[string]*querypb.BindVariable{"n": sqltypes.Int64BindVariable(2)}
@@ -59,7 +59,7 @@ func TestSelectNext(t *testing.T) {
 }
 
 func TestSelectDBA(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 
 	query := "select * from INFORMATION_SCHEMA.foo"
 	_, err := executor.Execute(
@@ -80,7 +80,7 @@ func TestSelectDBA(t *testing.T) {
 }
 
 func TestUnsharded(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 
 	_, err := executorExec(executor, "select id from music_user_map where id = 1", nil)
 	require.NoError(t, err)
@@ -94,7 +94,7 @@ func TestUnsharded(t *testing.T) {
 }
 
 func TestUnshardedComments(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 
 	_, err := executorExec(executor, "/* leading */ select id from music_user_map where id = 1 /* trailing */", nil)
 	require.NoError(t, err)
@@ -143,7 +143,7 @@ func TestUnshardedComments(t *testing.T) {
 }
 
 func TestStreamUnsharded(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
 
@@ -158,7 +158,7 @@ func TestStreamUnsharded(t *testing.T) {
 }
 
 func TestStreamBuffering(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 
 	// This test is similar to TestStreamUnsharded except that it returns a Result > 10 bytes,
 	// such that the splitting of the Result into multiple Result responses gets tested.
@@ -223,7 +223,7 @@ func TestStreamBuffering(t *testing.T) {
 
 func TestSelectLastInsertId(t *testing.T) {
 	masterSession.LastInsertId = 52
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	executor.normalize = true
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
@@ -243,7 +243,7 @@ func TestSelectLastInsertId(t *testing.T) {
 }
 
 func TestSelectUserDefindVariable(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	executor.normalize = true
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
@@ -276,7 +276,7 @@ func TestSelectUserDefindVariable(t *testing.T) {
 }
 
 func TestFoundRows(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	executor.normalize = true
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
@@ -300,7 +300,7 @@ func TestFoundRows(t *testing.T) {
 }
 
 func TestRowCount(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	executor.normalize = true
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
@@ -329,7 +329,7 @@ func testRowCount(t *testing.T, executor *Executor, wantRowCount int64) {
 }
 
 func TestSelectLastInsertIdInUnion(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 	executor.normalize = true
 	sql := "select last_insert_id() as id union select id from user"
 	_, err := executorExec(executor, sql, map[string]*querypb.BindVariable{})
@@ -343,7 +343,7 @@ func TestSelectLastInsertIdInUnion(t *testing.T) {
 }
 
 func TestSelectLastInsertIdInWhere(t *testing.T) {
-	executor, _, _, lookup := createExecutorEnv()
+	executor, _, _, lookup := createLegacyExecutorEnv()
 	executor.normalize = true
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
@@ -360,7 +360,7 @@ func TestSelectLastInsertIdInWhere(t *testing.T) {
 }
 
 func TestLastInsertIDInVirtualTable(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 	executor.normalize = true
 	result1 := []*sqltypes.Result{{
 		Fields: []*querypb.Field{
@@ -386,7 +386,7 @@ func TestLastInsertIDInVirtualTable(t *testing.T) {
 }
 
 func TestLastInsertIDInSubQueryExpression(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 	executor.normalize = true
 	result1 := []*sqltypes.Result{{
 		Fields: []*querypb.Field{
@@ -412,7 +412,7 @@ func TestLastInsertIDInSubQueryExpression(t *testing.T) {
 }
 
 func TestSelectDatabase(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	executor.normalize = true
 	sql := "select database()"
 	newSession := *masterSession
@@ -438,7 +438,7 @@ func TestSelectDatabase(t *testing.T) {
 }
 
 func TestSelectBindvars(t *testing.T) {
-	executor, sbc1, sbc2, lookup := createExecutorEnv()
+	executor, sbc1, sbc2, lookup := createLegacyExecutorEnv()
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
 
@@ -561,7 +561,7 @@ func TestSelectBindvars(t *testing.T) {
 }
 
 func TestSelectEqual(t *testing.T) {
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 
 	_, err := executorExec(executor, "select id from user where id = 1", nil)
 	require.NoError(t, err)
@@ -638,7 +638,7 @@ func TestSelectEqual(t *testing.T) {
 }
 
 func TestSelectDual(t *testing.T) {
-	executor, sbc1, _, lookup := createExecutorEnv()
+	executor, sbc1, _, lookup := createLegacyExecutorEnv()
 
 	_, err := executorExec(executor, "select @@aa.bb from dual", nil)
 	require.NoError(t, err)
@@ -658,7 +658,7 @@ func TestSelectDual(t *testing.T) {
 }
 
 func TestSelectComments(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	_, err := executorExec(executor, "/* leading */ select id from user where id = 1 /* trailing */", nil)
 	require.NoError(t, err)
@@ -676,7 +676,7 @@ func TestSelectComments(t *testing.T) {
 }
 
 func TestSelectNormalize(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	executor.normalize = true
 
 	_, err := executorExec(executor, "/* leading */ select id from user where id = 1 /* trailing */", nil)
@@ -712,7 +712,7 @@ func TestSelectNormalize(t *testing.T) {
 }
 
 func TestSelectCaseSensitivity(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	_, err := executorExec(executor, "select Id from user where iD = 1", nil)
 	require.NoError(t, err)
@@ -730,7 +730,7 @@ func TestSelectCaseSensitivity(t *testing.T) {
 }
 
 func TestStreamSelectEqual(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 
 	sql := "select id from user where id = 1"
 	result, err := executorStream(executor, sql)
@@ -742,7 +742,7 @@ func TestStreamSelectEqual(t *testing.T) {
 }
 
 func TestSelectKeyRange(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	_, err := executorExec(executor, "select krcol_unique, krcol from keyrange_table where krcol = 1", nil)
 	require.NoError(t, err)
@@ -760,7 +760,7 @@ func TestSelectKeyRange(t *testing.T) {
 }
 
 func TestSelectKeyRangeUnique(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 
 	_, err := executorExec(executor, "select krcol_unique, krcol from keyrange_table where krcol_unique = 1", nil)
 	require.NoError(t, err)
@@ -778,7 +778,7 @@ func TestSelectKeyRangeUnique(t *testing.T) {
 }
 
 func TestSelectIN(t *testing.T) {
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 
 	// Constant in IN clause is just a number, not a bind variable.
 	_, err := executorExec(executor, "select id from user where id in (1)", nil)
@@ -880,7 +880,7 @@ func TestSelectIN(t *testing.T) {
 }
 
 func TestStreamSelectIN(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 
 	sql := "select id from user where id in (1)"
 	result, err := executorStream(executor, sql)
@@ -927,7 +927,7 @@ func TestStreamSelectIN(t *testing.T) {
 }
 
 func TestSelectScatter(t *testing.T) {
-	// Special setup: Don't use createExecutorEnv.
+	// Special setup: Don't use createLegacyExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeLegacyHealthCheck()
 	s := createSandbox("TestExecutor")
@@ -960,7 +960,7 @@ func TestSelectScatter(t *testing.T) {
 }
 
 func TestSelectScatterPartial(t *testing.T) {
-	// Special setup: Don't use createExecutorEnv.
+	// Special setup: Don't use createLegacyExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeLegacyHealthCheck()
 	s := createSandbox("TestExecutor")
@@ -1020,7 +1020,7 @@ func TestSelectScatterPartial(t *testing.T) {
 }
 
 func TestStreamSelectScatter(t *testing.T) {
-	// Special setup: Don't use createExecutorEnv.
+	// Special setup: Don't use createLegacyExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeLegacyHealthCheck()
 	s := createSandbox("TestExecutor")
@@ -1057,7 +1057,7 @@ func TestStreamSelectScatter(t *testing.T) {
 
 // TestSelectScatterOrderBy will run an ORDER BY query that will scatter out to 8 shards and return the 8 rows (one per shard) sorted.
 func TestSelectScatterOrderBy(t *testing.T) {
-	// Special setup: Don't use createExecutorEnv.
+	// Special setup: Don't use createLegacyExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeLegacyHealthCheck()
 	s := createSandbox("TestExecutor")
@@ -1127,7 +1127,7 @@ func TestSelectScatterOrderBy(t *testing.T) {
 
 // TestSelectScatterOrderByVarChar will run an ORDER BY query that will scatter out to 8 shards and return the 8 rows (one per shard) sorted.
 func TestSelectScatterOrderByVarChar(t *testing.T) {
-	// Special setup: Don't use createExecutorEnv.
+	// Special setup: Don't use createLegacyExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeLegacyHealthCheck()
 	s := createSandbox("TestExecutor")
@@ -1197,7 +1197,7 @@ func TestSelectScatterOrderByVarChar(t *testing.T) {
 }
 
 func TestStreamSelectScatterOrderBy(t *testing.T) {
-	// Special setup: Don't use createExecutorEnv.
+	// Special setup: Don't use createLegacyExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeLegacyHealthCheck()
 	s := createSandbox("TestExecutor")
@@ -1258,7 +1258,7 @@ func TestStreamSelectScatterOrderBy(t *testing.T) {
 }
 
 func TestStreamSelectScatterOrderByVarChar(t *testing.T) {
-	// Special setup: Don't use createExecutorEnv.
+	// Special setup: Don't use createLegacyExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeLegacyHealthCheck()
 	s := createSandbox("TestExecutor")
@@ -1321,7 +1321,7 @@ func TestStreamSelectScatterOrderByVarChar(t *testing.T) {
 
 // TestSelectScatterAggregate will run an aggregate query that will scatter out to 8 shards and return 4 aggregated rows.
 func TestSelectScatterAggregate(t *testing.T) {
-	// Special setup: Don't use createExecutorEnv.
+	// Special setup: Don't use createLegacyExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeLegacyHealthCheck()
 	s := createSandbox("TestExecutor")
@@ -1384,7 +1384,7 @@ func TestSelectScatterAggregate(t *testing.T) {
 }
 
 func TestStreamSelectScatterAggregate(t *testing.T) {
-	// Special setup: Don't use createExecutorEnv.
+	// Special setup: Don't use createLegacyExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeLegacyHealthCheck()
 	s := createSandbox("TestExecutor")
@@ -1447,7 +1447,7 @@ func TestStreamSelectScatterAggregate(t *testing.T) {
 // TestSelectScatterLimit will run a limit query (ordered for consistency) against
 // a scatter route and verify that the limit primitive works as intended.
 func TestSelectScatterLimit(t *testing.T) {
-	// Special setup: Don't use createExecutorEnv.
+	// Special setup: Don't use createLegacyExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeLegacyHealthCheck()
 	s := createSandbox("TestExecutor")
@@ -1519,7 +1519,7 @@ func TestSelectScatterLimit(t *testing.T) {
 // TestStreamSelectScatterLimit will run a streaming limit query (ordered for consistency) against
 // a scatter route and verify that the limit primitive works as intended.
 func TestStreamSelectScatterLimit(t *testing.T) {
-	// Special setup: Don't use createExecutorEnv.
+	// Special setup: Don't use createLegacyExecutorEnv.
 	cell := "aa"
 	hc := discovery.NewFakeLegacyHealthCheck()
 	s := createSandbox("TestExecutor")
@@ -1589,7 +1589,7 @@ func TestStreamSelectScatterLimit(t *testing.T) {
 // TODO(sougou): stream and non-stream testing are very similar.
 // Could reuse code,
 func TestSimpleJoin(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
 
@@ -1631,7 +1631,7 @@ func TestSimpleJoin(t *testing.T) {
 }
 
 func TestJoinComments(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
 
@@ -1657,7 +1657,7 @@ func TestJoinComments(t *testing.T) {
 }
 
 func TestSimpleJoinStream(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
 
@@ -1699,7 +1699,7 @@ func TestSimpleJoinStream(t *testing.T) {
 }
 
 func TestVarJoin(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
 
@@ -1737,7 +1737,7 @@ func TestVarJoin(t *testing.T) {
 }
 
 func TestVarJoinStream(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
 
@@ -1775,7 +1775,7 @@ func TestVarJoinStream(t *testing.T) {
 }
 
 func TestLeftJoin(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
 	result1 := []*sqltypes.Result{{
@@ -1822,7 +1822,7 @@ func TestLeftJoin(t *testing.T) {
 }
 
 func TestLeftJoinStream(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	result1 := []*sqltypes.Result{{
 		Fields: []*querypb.Field{
 			{Name: "id", Type: sqltypes.Int32},
@@ -1863,7 +1863,7 @@ func TestLeftJoinStream(t *testing.T) {
 }
 
 func TestEmptyJoin(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 	// Empty result requires a field query for the second part of join,
 	// which is sent to shard 0.
 	sbc1.SetResults([]*sqltypes.Result{{
@@ -1901,7 +1901,7 @@ func TestEmptyJoin(t *testing.T) {
 }
 
 func TestEmptyJoinStream(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 	// Empty result requires a field query for the second part of join,
 	// which is sent to shard 0.
 	sbc1.SetResults([]*sqltypes.Result{{
@@ -1939,7 +1939,7 @@ func TestEmptyJoinStream(t *testing.T) {
 }
 
 func TestEmptyJoinRecursive(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 	// Make sure it also works recursively.
 	sbc1.SetResults([]*sqltypes.Result{{
 		Fields: []*querypb.Field{
@@ -1985,7 +1985,7 @@ func TestEmptyJoinRecursive(t *testing.T) {
 }
 
 func TestEmptyJoinRecursiveStream(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 	// Make sure it also works recursively.
 	sbc1.SetResults([]*sqltypes.Result{{
 		Fields: []*querypb.Field{
@@ -2031,7 +2031,7 @@ func TestEmptyJoinRecursiveStream(t *testing.T) {
 }
 
 func TestCrossShardSubquery(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	result1 := []*sqltypes.Result{{
 		Fields: []*querypb.Field{
 			{Name: "id", Type: sqltypes.Int32},
@@ -2076,7 +2076,7 @@ func TestCrossShardSubquery(t *testing.T) {
 }
 
 func TestCrossShardSubqueryStream(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	result1 := []*sqltypes.Result{{
 		Fields: []*querypb.Field{
 			{Name: "id", Type: sqltypes.Int32},
@@ -2120,7 +2120,7 @@ func TestCrossShardSubqueryStream(t *testing.T) {
 }
 
 func TestCrossShardSubqueryGetFields(t *testing.T) {
-	executor, sbc1, _, sbclookup := createExecutorEnv()
+	executor, sbc1, _, sbclookup := createLegacyExecutorEnv()
 	sbclookup.SetResults([]*sqltypes.Result{{
 		Fields: []*querypb.Field{
 			{Name: "col", Type: sqltypes.Int32},
@@ -2160,7 +2160,7 @@ func TestCrossShardSubqueryGetFields(t *testing.T) {
 }
 
 func TestSelectBindvarswithPrepare(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
 
@@ -2183,7 +2183,7 @@ func TestSelectBindvarswithPrepare(t *testing.T) {
 }
 
 func TestSelectWithUnionAll(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	executor.normalize = true
 	sql := "select id from user where id in (1, 2, 3) union all select id from user where id in (1, 2, 3)"
 	bv, _ := sqltypes.BuildBindVariable([]int64{1, 2, 3})
@@ -2233,3 +2233,50 @@ func TestSelectWithUnionAll(t *testing.T) {
 	utils.MustMatch(t, sbc1WantQueries, sbc1.Queries, "sbc1")
 	utils.MustMatch(t, sbc2WantQueries, sbc2.Queries, "sbc2")
 }
+
+func TestSelectLock(t *testing.T) {
+	executor, sbc1, _, _ := createExecutorEnv()
+	session := NewSafeSession(nil)
+	session.Session.InTransaction = true
+	session.ShardSessions = []*vtgatepb.Session_ShardSession{{
+		Target: &querypb.Target{
+			Keyspace:   "TestExecutor",
+			Shard:      "-20",
+			TabletType: topodatapb.TabletType_MASTER,
+		},
+		TransactionId: 12345,
+		TabletAlias:   sbc1.Tablet().Alias,
+	}}
+
+	wantQueries := []*querypb.BoundQuery{{
+		Sql:           "select get_lock('lock name', 10) from dual",
+		BindVariables: map[string]*querypb.BindVariable{},
+	}}
+	wantSession := &vtgatepb.Session{
+		InTransaction:  true,
+		InReservedConn: true,
+		ShardSessions: []*vtgatepb.Session_ShardSession{
+			{
+				Target:        &querypb.Target{Keyspace: "TestExecutor", Shard: "-20", TabletType: topodatapb.TabletType_MASTER},
+				TabletAlias:   sbc1.Tablet().Alias,
+				TransactionId: 12345,
+				ReservedId:    12345,
+			},
+		},
+		FoundRows: 1,
+		RowCount:  -1,
+	}
+
+	_, err := exec(executor, session, "select get_lock('lock name', 10) from dual")
+	require.NoError(t, err)
+	utils.MustMatch(t, wantSession, session.Session, "")
+	utils.MustMatch(t, wantQueries, sbc1.Queries, "")
+
+	wantQueries = append(wantQueries, &querypb.BoundQuery{
+		Sql:           "select release_lock('lock name') from dual",
+		BindVariables: map[string]*querypb.BindVariable{},
+	})
+	exec(executor, session, "select release_lock('lock name') from dual")
+	utils.MustMatch(t, wantQueries, sbc1.Queries, "")
+	utils.MustMatch(t, wantSession, session.Session, "")
+}
diff --git a/go/vt/vtgate/executor_set_test.go b/go/vt/vtgate/executor_set_test.go
index 2954c22363..e56678fda6 100644
--- a/go/vt/vtgate/executor_set_test.go
+++ b/go/vt/vtgate/executor_set_test.go
@@ -39,7 +39,7 @@ import (
 )
 
 func TestExecutorSet(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 
 	testcases := []struct {
 		in  string
@@ -300,7 +300,7 @@ func TestExecutorSet(t *testing.T) {
 }
 
 func TestExecutorSetOp(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 
 	testcases := []struct {
 		in      string
@@ -335,7 +335,7 @@ func TestExecutorSetOp(t *testing.T) {
 }
 
 func TestExecutorSetMetadata(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master", Autocommit: true})
 
 	set := "set @@vitess_metadata.app_keyspace_v1= '1'"
@@ -347,7 +347,7 @@ func TestExecutorSetMetadata(t *testing.T) {
 		*vschemaacl.AuthorizedDDLUsers = ""
 	}()
 
-	executor, _, _, _ = createExecutorEnv()
+	executor, _, _, _ = createLegacyExecutorEnv()
 	session = NewSafeSession(&vtgatepb.Session{TargetString: "@master", Autocommit: true})
 
 	set = "set @@vitess_metadata.app_keyspace_v1= '1'"
@@ -392,7 +392,7 @@ func TestExecutorSetMetadata(t *testing.T) {
 }
 
 func TestPlanExecutorSetUDV(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 
 	testcases := []struct {
 		in  string
@@ -422,7 +422,7 @@ func TestPlanExecutorSetUDV(t *testing.T) {
 }
 
 func TestSetUDVFromTabletInput(t *testing.T) {
-	executor, sbc1, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createLegacyExecutorEnv()
 
 	fields := sqltypes.MakeTestFields("some", "VARBINARY")
 	sbc1.SetResults([]*sqltypes.Result{
diff --git a/go/vt/vtgate/executor_stream_test.go b/go/vt/vtgate/executor_stream_test.go
index c934b9cffd..5b1eec7d5f 100644
--- a/go/vt/vtgate/executor_stream_test.go
+++ b/go/vt/vtgate/executor_stream_test.go
@@ -31,7 +31,7 @@ import (
 )
 
 func TestStreamSQLUnsharded(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
 
diff --git a/go/vt/vtgate/executor_test.go b/go/vt/vtgate/executor_test.go
index f147927736..e713ec6965 100644
--- a/go/vt/vtgate/executor_test.go
+++ b/go/vt/vtgate/executor_test.go
@@ -57,7 +57,7 @@ func TestExecutorResultsExceeded(t *testing.T) {
 	*warnMemoryRows = 3
 	defer func() { *warnMemoryRows = save }()
 
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master"})
 
 	initial := warnings.Counts()["ResultsExceeded"]
@@ -114,7 +114,7 @@ func TestExecutorMaxMemoryRowsExceeded(t *testing.T) {
 }
 
 func TestLegacyExecutorTransactionsNoAutoCommit(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master"})
 
 	logChan := QueryLogger.Subscribe("Test")
@@ -199,7 +199,7 @@ func TestLegacyExecutorTransactionsNoAutoCommit(t *testing.T) {
 }
 
 func TestDirectTargetRewrites(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 	executor.normalize = true
 
 	session := &vtgatepb.Session{
@@ -218,7 +218,7 @@ func TestDirectTargetRewrites(t *testing.T) {
 }
 
 func TestExecutorTransactionsAutoCommit(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master", Autocommit: true})
 
 	logChan := QueryLogger.Subscribe("Test")
@@ -268,7 +268,7 @@ func TestExecutorDeleteMetadata(t *testing.T) {
 		*vschemaacl.AuthorizedDDLUsers = ""
 	}()
 
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master", Autocommit: true})
 
 	set := "set @@vitess_metadata.app_v1= '1'"
@@ -295,7 +295,7 @@ func TestExecutorDeleteMetadata(t *testing.T) {
 }
 
 func TestExecutorAutocommit(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master"})
 
 	logChan := QueryLogger.Subscribe("Test")
@@ -395,7 +395,7 @@ func TestExecutorAutocommit(t *testing.T) {
 }
 
 func TestExecutorShowColumns(t *testing.T) {
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: ""})
 
 	queries := []string{
@@ -433,7 +433,7 @@ func TestExecutorShowColumns(t *testing.T) {
 }
 
 func TestExecutorShow(t *testing.T) {
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master"})
 
 	for _, query := range []string{"show databases", "show vitess_keyspaces", "show keyspaces", "show DATABASES"} {
@@ -995,7 +995,7 @@ func TestExecutorShow(t *testing.T) {
 }
 
 func TestExecutorUse(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{Autocommit: true, TargetString: "@master"})
 
 	stmts := []string{
@@ -1029,7 +1029,7 @@ func TestExecutorUse(t *testing.T) {
 }
 
 func TestExecutorComment(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 
 	stmts := []string{
 		"/*! SET autocommit=1*/",
@@ -1049,7 +1049,7 @@ func TestExecutorComment(t *testing.T) {
 }
 
 func TestExecutorOther(t *testing.T) {
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 
 	type cnts struct {
 		Sbc1Cnt      int64
@@ -1146,7 +1146,7 @@ func TestExecutorDDL(t *testing.T) {
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
 
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 
 	type cnts struct {
 		Sbc1Cnt      int64
@@ -1243,7 +1243,7 @@ func TestExecutorAlterVSchemaKeyspace(t *testing.T) {
 	defer func() {
 		*vschemaacl.AuthorizedDDLUsers = ""
 	}()
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master", Autocommit: true})
 
 	vschemaUpdates := make(chan *vschemapb.SrvVSchema, 2)
@@ -1270,7 +1270,7 @@ func TestExecutorCreateVindexDDL(t *testing.T) {
 	defer func() {
 		*vschemaacl.AuthorizedDDLUsers = ""
 	}()
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 	ks := "TestExecutor"
 
 	vschemaUpdates := make(chan *vschemapb.SrvVSchema, 4)
@@ -1341,7 +1341,7 @@ func TestExecutorAddDropVschemaTableDDL(t *testing.T) {
 	defer func() {
 		*vschemaacl.AuthorizedDDLUsers = ""
 	}()
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 	ks := KsTestUnsharded
 
 	vschemaUpdates := make(chan *vschemapb.SrvVSchema, 4)
@@ -1388,7 +1388,7 @@ func TestExecutorAddDropVschemaTableDDL(t *testing.T) {
 }
 
 func TestExecutorVindexDDLACL(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	ks := "TestExecutor"
 	session := NewSafeSession(&vtgatepb.Session{TargetString: ks})
 
@@ -1439,7 +1439,7 @@ func TestExecutorVindexDDLACL(t *testing.T) {
 }
 
 func TestExecutorUnrecognized(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	_, err := executor.Execute(ctx, "TestExecute", NewSafeSession(&vtgatepb.Session{}), "invalid statement", nil)
 	require.Error(t, err, "unrecognized statement: invalid statement'")
 }
@@ -1447,7 +1447,7 @@ func TestExecutorUnrecognized(t *testing.T) {
 // TestVSchemaStats makes sure the building and displaying of the
 // VSchemaStats works.
 func TestVSchemaStats(t *testing.T) {
-	r, _, _, _ := createExecutorEnv()
+	r, _, _, _ := createLegacyExecutorEnv()
 
 	stats := r.VSchemaStats()
 
@@ -1468,7 +1468,7 @@ func TestVSchemaStats(t *testing.T) {
 }
 
 func TestGetPlanUnnormalized(t *testing.T) {
-	r, _, _, _ := createExecutorEnv()
+	r, _, _, _ := createLegacyExecutorEnv()
 	emptyvc, _ := newVCursorImpl(ctx, NewSafeSession(&vtgatepb.Session{TargetString: "@unknown"}), makeComments(""), r, nil, r.vm, r.VSchema(), r.resolver.resolver)
 	unshardedvc, _ := newVCursorImpl(ctx, NewSafeSession(&vtgatepb.Session{TargetString: KsTestUnsharded + "@unknown"}), makeComments(""), r, nil, r.vm, r.VSchema(), r.resolver.resolver)
 
@@ -1527,7 +1527,7 @@ func TestGetPlanUnnormalized(t *testing.T) {
 }
 
 func TestGetPlanCacheUnnormalized(t *testing.T) {
-	r, _, _, _ := createExecutorEnv()
+	r, _, _, _ := createLegacyExecutorEnv()
 	emptyvc, _ := newVCursorImpl(ctx, NewSafeSession(&vtgatepb.Session{TargetString: "@unknown"}), makeComments(""), r, nil, r.vm, r.VSchema(), r.resolver.resolver)
 	query1 := "select * from music_user_map where id = 1"
 	logStats1 := NewLogStats(ctx, "Test", "", nil)
@@ -1552,7 +1552,7 @@ func TestGetPlanCacheUnnormalized(t *testing.T) {
 	}
 
 	// Skip cache using directive
-	r, _, _, _ = createExecutorEnv()
+	r, _, _, _ = createLegacyExecutorEnv()
 	unshardedvc, _ := newVCursorImpl(ctx, NewSafeSession(&vtgatepb.Session{TargetString: KsTestUnsharded + "@unknown"}), makeComments(""), r, nil, r.vm, r.VSchema(), r.resolver.resolver)
 
 	query1 = "insert /*vt+ SKIP_QUERY_PLAN_CACHE=1 */ into user(id) values (1), (2)"
@@ -1588,7 +1588,7 @@ func TestGetPlanCacheUnnormalized(t *testing.T) {
 }
 
 func TestGetPlanCacheNormalized(t *testing.T) {
-	r, _, _, _ := createExecutorEnv()
+	r, _, _, _ := createLegacyExecutorEnv()
 	r.normalize = true
 	emptyvc, _ := newVCursorImpl(ctx, NewSafeSession(&vtgatepb.Session{TargetString: "@unknown"}), makeComments(""), r, nil, r.vm, r.VSchema(), r.resolver.resolver)
 	query1 := "select * from music_user_map where id = 1"
@@ -1613,7 +1613,7 @@ func TestGetPlanCacheNormalized(t *testing.T) {
 	}
 
 	// Skip cache using directive
-	r, _, _, _ = createExecutorEnv()
+	r, _, _, _ = createLegacyExecutorEnv()
 	r.normalize = true
 	unshardedvc, _ := newVCursorImpl(ctx, NewSafeSession(&vtgatepb.Session{TargetString: KsTestUnsharded + "@unknown"}), makeComments(""), r, nil, r.vm, r.VSchema(), r.resolver.resolver)
 
@@ -1650,7 +1650,7 @@ func TestGetPlanCacheNormalized(t *testing.T) {
 }
 
 func TestGetPlanNormalized(t *testing.T) {
-	r, _, _, _ := createExecutorEnv()
+	r, _, _, _ := createLegacyExecutorEnv()
 	r.normalize = true
 	emptyvc, _ := newVCursorImpl(ctx, NewSafeSession(&vtgatepb.Session{TargetString: "@unknown"}), makeComments(""), r, nil, r.vm, r.VSchema(), r.resolver.resolver)
 	unshardedvc, _ := newVCursorImpl(ctx, NewSafeSession(&vtgatepb.Session{TargetString: KsTestUnsharded + "@unknown"}), makeComments(""), r, nil, r.vm, r.VSchema(), r.resolver.resolver)
@@ -1743,7 +1743,7 @@ func TestGetPlanNormalized(t *testing.T) {
 }
 
 func TestPassthroughDDL(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	masterSession.TargetString = "TestExecutor"
 
 	alterDDL := "/* leading */ alter table passthrough_ddl add columne col bigint default 123 /* trailing */"
@@ -1792,7 +1792,7 @@ func TestPassthroughDDL(t *testing.T) {
 }
 
 func TestParseEmptyTargetSingleKeyspace(t *testing.T) {
-	r, _, _, _ := createExecutorEnv()
+	r, _, _, _ := createLegacyExecutorEnv()
 	altVSchema := &vindexes.VSchema{
 		Keyspaces: map[string]*vindexes.KeyspaceSchema{
 			KsTestUnsharded: r.vschema.Keyspaces[KsTestUnsharded],
@@ -1814,7 +1814,7 @@ func TestParseEmptyTargetSingleKeyspace(t *testing.T) {
 }
 
 func TestParseEmptyTargetMultiKeyspace(t *testing.T) {
-	r, _, _, _ := createExecutorEnv()
+	r, _, _, _ := createLegacyExecutorEnv()
 	altVSchema := &vindexes.VSchema{
 		Keyspaces: map[string]*vindexes.KeyspaceSchema{
 			KsTestUnsharded: r.vschema.Keyspaces[KsTestUnsharded],
@@ -1837,7 +1837,7 @@ func TestParseEmptyTargetMultiKeyspace(t *testing.T) {
 }
 
 func TestParseTargetSingleKeyspace(t *testing.T) {
-	r, _, _, _ := createExecutorEnv()
+	r, _, _, _ := createLegacyExecutorEnv()
 	altVSchema := &vindexes.VSchema{
 		Keyspaces: map[string]*vindexes.KeyspaceSchema{
 			KsTestUnsharded: r.vschema.Keyspaces[KsTestUnsharded],
@@ -1862,7 +1862,7 @@ func TestDebugVSchema(t *testing.T) {
 	resp := httptest.NewRecorder()
 	req, _ := http.NewRequest("GET", "/debug/vschema", nil)
 
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	executor.ServeHTTP(resp, req)
 	v := make(map[string]interface{})
 	if err := json.Unmarshal(resp.Body.Bytes(), &v); err != nil {
@@ -1943,7 +1943,7 @@ func TestExecutorMaxPayloadSizeExceeded(t *testing.T) {
 		*warnPayloadSize = saveWarn
 	}()
 
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master"})
 	warningCount := warnings.Counts()["WarnPayloadSizeExceeded"]
 	testMaxPayloadSizeExceeded := []string{
@@ -1980,7 +1980,7 @@ func TestExecutorMaxPayloadSizeExceeded(t *testing.T) {
 }
 
 func TestOlapSelectDatabase(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	executor.normalize = true
 
 	session := &vtgatepb.Session{Autocommit: true}
@@ -1998,7 +1998,7 @@ func TestOlapSelectDatabase(t *testing.T) {
 }
 
 func TestExecutorClearsWarnings(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{
 		Warnings: []*querypb.QueryWarning{{Code: 234, Message: "oh noes"}},
 	})
@@ -2008,7 +2008,7 @@ func TestExecutorClearsWarnings(t *testing.T) {
 }
 
 func TestExecutorOtherRead(t *testing.T) {
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 
 	type cnts struct {
 		Sbc1Cnt      int64
@@ -2081,7 +2081,7 @@ func TestExecutorOtherRead(t *testing.T) {
 }
 
 func TestExecutorExplain(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	executor.normalize = true
 	logChan := QueryLogger.Subscribe("Test")
 	defer QueryLogger.Unsubscribe(logChan)
@@ -2107,7 +2107,7 @@ func TestExecutorExplain(t *testing.T) {
 }
 
 func TestExecutorOtherAdmin(t *testing.T) {
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 
 	type cnts struct {
 		Sbc1Cnt      int64
@@ -2181,7 +2181,7 @@ func TestExecutorOtherAdmin(t *testing.T) {
 }
 
 func TestExecutorSavepointInTx(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	logChan := QueryLogger.Subscribe("TestExecutorSavepoint")
 	defer QueryLogger.Unsubscribe(logChan)
 
@@ -2263,7 +2263,7 @@ func TestExecutorSavepointInTx(t *testing.T) {
 }
 
 func TestExecutorSavepointWithoutTx(t *testing.T) {
-	executor, sbc1, sbc2, _ := createExecutorEnv()
+	executor, sbc1, sbc2, _ := createLegacyExecutorEnv()
 	logChan := QueryLogger.Subscribe("TestExecutorSavepoint")
 	defer QueryLogger.Unsubscribe(logChan)
 
diff --git a/go/vt/vtgate/executor_vschema_ddl_test.go b/go/vt/vtgate/executor_vschema_ddl_test.go
index 20e99a1fa7..931a23eeee 100644
--- a/go/vt/vtgate/executor_vschema_ddl_test.go
+++ b/go/vt/vtgate/executor_vschema_ddl_test.go
@@ -133,7 +133,7 @@ func TestPlanExecutorAlterVSchemaKeyspace(t *testing.T) {
 	defer func() {
 		*vschemaacl.AuthorizedDDLUsers = ""
 	}()
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master", Autocommit: true})
 
 	vschemaUpdates := make(chan *vschemapb.SrvVSchema, 2)
@@ -160,7 +160,7 @@ func TestPlanExecutorCreateVindexDDL(t *testing.T) {
 	defer func() {
 		*vschemaacl.AuthorizedDDLUsers = ""
 	}()
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	ks := "TestExecutor"
 
 	vschemaUpdates := make(chan *vschemapb.SrvVSchema, 4)
@@ -201,7 +201,7 @@ func TestPlanExecutorDropVindexDDL(t *testing.T) {
 	defer func() {
 		*vschemaacl.AuthorizedDDLUsers = ""
 	}()
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	ks := "TestExecutor"
 
 	vschemaUpdates := make(chan *vschemapb.SrvVSchema, 4)
@@ -269,7 +269,7 @@ func TestPlanExecutorAddDropVschemaTableDDL(t *testing.T) {
 	defer func() {
 		*vschemaacl.AuthorizedDDLUsers = ""
 	}()
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv()
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv()
 	ks := KsTestUnsharded
 
 	vschemaUpdates := make(chan *vschemapb.SrvVSchema, 4)
@@ -325,7 +325,7 @@ func TestExecutorAddSequenceDDL(t *testing.T) {
 	defer func() {
 		*vschemaacl.AuthorizedDDLUsers = ""
 	}()
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	ks := KsTestUnsharded
 
 	vschema := executor.vm.GetCurrentSrvVschema()
@@ -385,7 +385,7 @@ func TestExecutorAddDropVindexDDL(t *testing.T) {
 	defer func() {
 		*vschemaacl.AuthorizedDDLUsers = ""
 	}()
-	executor, sbc1, sbc2, sbclookup := createExecutorEnv() //nolint
+	executor, sbc1, sbc2, sbclookup := createLegacyExecutorEnv() //nolint
 	ks := "TestExecutor"
 	session := NewSafeSession(&vtgatepb.Session{TargetString: ks})
 	vschemaUpdates := make(chan *vschemapb.SrvVSchema, 4)
@@ -715,7 +715,7 @@ func TestExecutorAddDropVindexDDL(t *testing.T) {
 
 func TestPlanExecutorVindexDDLACL(t *testing.T) {
 	//t.Skip("not yet planned")
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 	ks := "TestExecutor"
 	session := NewSafeSession(&vtgatepb.Session{TargetString: ks})
 
diff --git a/go/vt/vtgate/legacy_scatter_conn_test.go b/go/vt/vtgate/legacy_scatter_conn_test.go
index 381a8a60e5..790e6ffa3a 100644
--- a/go/vt/vtgate/legacy_scatter_conn_test.go
+++ b/go/vt/vtgate/legacy_scatter_conn_test.go
@@ -331,187 +331,30 @@ func TestMaxMemoryRows(t *testing.T) {
 	}
 }
 
-func TestReservedBeginTableDriven(t *testing.T) {
-	type testAction struct {
-		transaction, reserved    bool
-		shards                   []string
-		sbc0Reserve, sbc1Reserve int64
-		sbc0Begin, sbc1Begin     int64
-	}
-	type testCase struct {
-		name    string
-		actions []testAction
-	}
-
-	tests := []testCase{{
-		name: "begin",
-		actions: []testAction{
-			{
-				shards:      []string{"0"},
-				transaction: true,
-				sbc0Begin:   1,
-			}, {
-				shards:      []string{"0", "1"},
-				transaction: true,
-				sbc1Begin:   1,
-			}, {
-				shards:      []string{"0", "1"},
-				transaction: true,
-				// nothing needs to be done
-			}},
-	}, {
-		name: "reserve",
-		actions: []testAction{
-			{
-				shards:      []string{"1"},
-				reserved:    true,
-				sbc1Reserve: 1,
-			}, {
-				shards:      []string{"0", "1"},
-				reserved:    true,
-				sbc0Reserve: 1,
-			}, {
-				shards:   []string{"0", "1"},
-				reserved: true,
-				// nothing needs to be done
-			}},
-	}, {
-		name: "reserve everywhere",
-		actions: []testAction{
-			{
-				shards:      []string{"0", "1"},
-				reserved:    true,
-				sbc0Reserve: 1,
-				sbc1Reserve: 1,
-			}},
-	}, {
-		name: "begin then reserve",
-		actions: []testAction{
-			{
-				shards:      []string{"0"},
-				transaction: true,
-				sbc0Begin:   1,
-			}, {
-				shards:      []string{"0", "1"},
-				transaction: true,
-				reserved:    true,
-				sbc0Reserve: 1,
-				sbc1Reserve: 1,
-				sbc1Begin:   1,
-			}},
-	}, {
-		name: "reserve then begin",
-		actions: []testAction{
-			{
-				shards:      []string{"1"},
-				reserved:    true,
-				sbc1Reserve: 1,
-			}, {
-				shards:      []string{"0"},
-				transaction: true,
-				reserved:    true,
-				sbc0Reserve: 1,
-				sbc0Begin:   1,
-			}, {
-				shards:      []string{"0", "1"},
-				transaction: true,
-				reserved:    true,
-				sbc1Begin:   1,
-			}},
-	}, {
-		name: "reserveBegin",
-		actions: []testAction{
-			{
-				shards:      []string{"1"},
-				transaction: true,
-				reserved:    true,
-				sbc1Reserve: 1,
-				sbc1Begin:   1,
-			}, {
-				shards:      []string{"0"},
-				transaction: true,
-				reserved:    true,
-				sbc0Reserve: 1,
-				sbc0Begin:   1,
-			}, {
-				shards:      []string{"0", "1"},
-				transaction: true,
-				reserved:    true,
-				// nothing needs to be done
-			}},
-	}, {
-		name: "reserveBegin everywhere",
-		actions: []testAction{
-			{
-				shards:      []string{"0", "1"},
-				transaction: true,
-				reserved:    true,
-				sbc0Reserve: 1,
-				sbc0Begin:   1,
-				sbc1Reserve: 1,
-				sbc1Begin:   1,
-			}},
-	}}
-	for _, test := range tests {
-		keyspace := "keyspace"
-		createSandbox(keyspace)
-		hc := discovery.NewFakeLegacyHealthCheck()
-		sc := newTestLegacyScatterConn(hc, new(sandboxTopo), "aa")
-		sbc0 := hc.AddTestTablet("aa", "0", 1, keyspace, "0", topodatapb.TabletType_REPLICA, true, 1, nil)
-		sbc1 := hc.AddTestTablet("aa", "1", 1, keyspace, "1", topodatapb.TabletType_REPLICA, true, 1, nil)
-
-		// empty results
-		sbc0.SetResults([]*sqltypes.Result{{}})
-		sbc1.SetResults([]*sqltypes.Result{{}})
-
-		res := srvtopo.NewResolver(&sandboxTopo{}, sc.gateway, "aa")
-
-		t.Run(test.name, func(t *testing.T) {
-			session := NewSafeSession(&vtgatepb.Session{})
-			for _, action := range test.actions {
-				session.Session.InTransaction = action.transaction
-				session.Session.InReservedConn = action.reserved
-				var destinations []key.Destination
-				for _, shard := range action.shards {
-					destinations = append(destinations, key.DestinationShard(shard))
-				}
-				executeOnShards(t, res, keyspace, sc, session, destinations)
-				assert.EqualValues(t, action.sbc0Reserve, sbc0.ReserveCount.Get(), "sbc0 reserve count")
-				assert.EqualValues(t, action.sbc0Begin, sbc0.BeginCount.Get(), "sbc0 begin count")
-				assert.EqualValues(t, action.sbc1Reserve, sbc1.ReserveCount.Get(), "sbc1 reserve count")
-				assert.EqualValues(t, action.sbc1Begin, sbc1.BeginCount.Get(), "sbc1 begin count")
-				sbc0.BeginCount.Set(0)
-				sbc0.ReserveCount.Set(0)
-				sbc1.BeginCount.Set(0)
-				sbc1.ReserveCount.Set(0)
-			}
-		})
-	}
-}
-
-// TODO (harshit): This test should actual fail.
-func TestReservedOnMultiReplica(t *testing.T) {
+func TestLegaceHealthCheckFailsOnReservedConnections(t *testing.T) {
 	keyspace := "keyspace"
 	createSandbox(keyspace)
 	hc := discovery.NewFakeLegacyHealthCheck()
 	sc := newTestLegacyScatterConn(hc, new(sandboxTopo), "aa")
-	sbc0_1 := hc.AddTestTablet("aa", "0", 1, keyspace, "0", topodatapb.TabletType_REPLICA, true, 1, nil)
-	sbc0_2 := hc.AddTestTablet("aa", "2", 1, keyspace, "0", topodatapb.TabletType_REPLICA, true, 1, nil)
-	//	sbc1 := hc.AddTestTablet("aa", "1", 1, keyspace, "1", topodatapb.TabletType_REPLICA, true, 1, nil)
-
-	// empty results
-	sbc0_1.SetResults([]*sqltypes.Result{{}})
-	sbc0_2.SetResults([]*sqltypes.Result{{}})
 
 	res := srvtopo.NewResolver(&sandboxTopo{}, sc.gateway, "aa")
 
 	session := NewSafeSession(&vtgatepb.Session{InTransaction: false, InReservedConn: true})
 	destinations := []key.Destination{key.DestinationShard("0")}
-	for i := 0; i < 10; i++ {
-		executeOnShards(t, res, keyspace, sc, session, destinations)
-		assert.EqualValues(t, 1, sbc0_1.ReserveCount.Get()+sbc0_2.ReserveCount.Get(), "sbc0 reserve count")
-		assert.EqualValues(t, 0, sbc0_1.BeginCount.Get()+sbc0_2.BeginCount.Get(), "sbc0 begin count")
+	rss, _, err := res.ResolveDestinations(ctx, keyspace, topodatapb.TabletType_REPLICA, nil, destinations)
+	require.NoError(t, err)
+
+	var queries []*querypb.BoundQuery
+
+	for range rss {
+		queries = append(queries, &querypb.BoundQuery{
+			Sql:           "query1",
+			BindVariables: map[string]*querypb.BindVariable{},
+		})
 	}
+
+	_, errs := sc.ExecuteMultiShard(ctx, rss, queries, session, false, false)
+	require.Error(t, vterrors.Aggregate(errs))
 }
 
 func executeOnShards(t *testing.T, res *srvtopo.Resolver, keyspace string, sc *ScatterConn, session *SafeSession, destinations []key.Destination) {
diff --git a/go/vt/vtgate/planbuilder/plan_test.go b/go/vt/vtgate/planbuilder/plan_test.go
index 35e64234b7..215b7d3bf9 100644
--- a/go/vt/vtgate/planbuilder/plan_test.go
+++ b/go/vt/vtgate/planbuilder/plan_test.go
@@ -322,6 +322,10 @@ func (vw *vschemaWrapper) AnyKeyspace() (*vindexes.Keyspace, error) {
 	return vw.DefaultKeyspace()
 }
 
+func (vw *vschemaWrapper) FirstSortedKeyspace() (*vindexes.Keyspace, error) {
+	panic("implement me")
+}
+
 func (vw *vschemaWrapper) TargetString() string {
 	return "targetString"
 }
diff --git a/go/vt/vtgate/queryz_test.go b/go/vt/vtgate/queryz_test.go
index 62ea30e9d1..0dc85b33d8 100644
--- a/go/vt/vtgate/queryz_test.go
+++ b/go/vt/vtgate/queryz_test.go
@@ -36,7 +36,7 @@ func TestQueryzHandler(t *testing.T) {
 	resp := httptest.NewRecorder()
 	req, _ := http.NewRequest("GET", "/schemaz", nil)
 
-	executor, _, _, _ := createExecutorEnv()
+	executor, _, _, _ := createLegacyExecutorEnv()
 
 	// single shard query
 	sql := "select id from user where id = 1"
diff --git a/go/vt/vtgate/scatter_conn_test.go b/go/vt/vtgate/scatter_conn_test.go
index da415f90d0..0f0445d105 100644
--- a/go/vt/vtgate/scatter_conn_test.go
+++ b/go/vt/vtgate/scatter_conn_test.go
@@ -19,6 +19,9 @@ package vtgate
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+	"vitess.io/vitess/go/vt/key"
+
 	"vitess.io/vitess/go/test/utils"
 
 	"github.com/stretchr/testify/require"
@@ -95,3 +98,185 @@ func TestExecuteFailOnAutocommit(t *testing.T) {
 	utils.MustMatch(t, 0, len(sbc0.Queries), "")
 	utils.MustMatch(t, []*querypb.BoundQuery{queries[1]}, sbc1.Queries, "")
 }
+
+func TestReservedOnMultiReplica(t *testing.T) {
+	keyspace := "keyspace"
+	createSandbox(keyspace)
+	hc := discovery.NewFakeHealthCheck()
+	sc := newTestScatterConn(hc, new(sandboxTopo), "aa")
+	sbc0_1 := hc.AddTestTablet("aa", "0", 1, keyspace, "0", topodatapb.TabletType_REPLICA, true, 1, nil)
+	sbc0_2 := hc.AddTestTablet("aa", "2", 1, keyspace, "0", topodatapb.TabletType_REPLICA, true, 1, nil)
+	//	sbc1 := hc.AddTestTablet("aa", "1", 1, keyspace, "1", topodatapb.TabletType_REPLICA, true, 1, nil)
+
+	// empty results
+	sbc0_1.SetResults([]*sqltypes.Result{{}})
+	sbc0_2.SetResults([]*sqltypes.Result{{}})
+
+	res := srvtopo.NewResolver(&sandboxTopo{}, sc.gateway, "aa")
+
+	session := NewSafeSession(&vtgatepb.Session{InTransaction: false, InReservedConn: true})
+	destinations := []key.Destination{key.DestinationShard("0")}
+	for i := 0; i < 10; i++ {
+		executeOnShards(t, res, keyspace, sc, session, destinations)
+		assert.EqualValues(t, 1, sbc0_1.ReserveCount.Get()+sbc0_2.ReserveCount.Get(), "sbc0 reserve count")
+		assert.EqualValues(t, 0, sbc0_1.BeginCount.Get()+sbc0_2.BeginCount.Get(), "sbc0 begin count")
+	}
+}
+
+func TestReservedBeginTableDriven(t *testing.T) {
+	type testAction struct {
+		transaction, reserved    bool
+		shards                   []string
+		sbc0Reserve, sbc1Reserve int64
+		sbc0Begin, sbc1Begin     int64
+	}
+	type testCase struct {
+		name    string
+		actions []testAction
+	}
+
+	tests := []testCase{{
+		name: "begin",
+		actions: []testAction{
+			{
+				shards:      []string{"0"},
+				transaction: true,
+				sbc0Begin:   1,
+			}, {
+				shards:      []string{"0", "1"},
+				transaction: true,
+				sbc1Begin:   1,
+			}, {
+				shards:      []string{"0", "1"},
+				transaction: true,
+				// nothing needs to be done
+			}},
+	}, {
+		name: "reserve",
+		actions: []testAction{
+			{
+				shards:      []string{"1"},
+				reserved:    true,
+				sbc1Reserve: 1,
+			}, {
+				shards:      []string{"0", "1"},
+				reserved:    true,
+				sbc0Reserve: 1,
+			}, {
+				shards:   []string{"0", "1"},
+				reserved: true,
+				// nothing needs to be done
+			}},
+	}, {
+		name: "reserve everywhere",
+		actions: []testAction{
+			{
+				shards:      []string{"0", "1"},
+				reserved:    true,
+				sbc0Reserve: 1,
+				sbc1Reserve: 1,
+			}},
+	}, {
+		name: "begin then reserve",
+		actions: []testAction{
+			{
+				shards:      []string{"0"},
+				transaction: true,
+				sbc0Begin:   1,
+			}, {
+				shards:      []string{"0", "1"},
+				transaction: true,
+				reserved:    true,
+				sbc0Reserve: 1,
+				sbc1Reserve: 1,
+				sbc1Begin:   1,
+			}},
+	}, {
+		name: "reserve then begin",
+		actions: []testAction{
+			{
+				shards:      []string{"1"},
+				reserved:    true,
+				sbc1Reserve: 1,
+			}, {
+				shards:      []string{"0"},
+				transaction: true,
+				reserved:    true,
+				sbc0Reserve: 1,
+				sbc0Begin:   1,
+			}, {
+				shards:      []string{"0", "1"},
+				transaction: true,
+				reserved:    true,
+				sbc1Begin:   1,
+			}},
+	}, {
+		name: "reserveBegin",
+		actions: []testAction{
+			{
+				shards:      []string{"1"},
+				transaction: true,
+				reserved:    true,
+				sbc1Reserve: 1,
+				sbc1Begin:   1,
+			}, {
+				shards:      []string{"0"},
+				transaction: true,
+				reserved:    true,
+				sbc0Reserve: 1,
+				sbc0Begin:   1,
+			}, {
+				shards:      []string{"0", "1"},
+				transaction: true,
+				reserved:    true,
+				// nothing needs to be done
+			}},
+	}, {
+		name: "reserveBegin everywhere",
+		actions: []testAction{
+			{
+				shards:      []string{"0", "1"},
+				transaction: true,
+				reserved:    true,
+				sbc0Reserve: 1,
+				sbc0Begin:   1,
+				sbc1Reserve: 1,
+				sbc1Begin:   1,
+			}},
+	}}
+	for _, test := range tests {
+		keyspace := "keyspace"
+		createSandbox(keyspace)
+		hc := discovery.NewFakeHealthCheck()
+		sc := newTestScatterConn(hc, new(sandboxTopo), "aa")
+		sbc0 := hc.AddTestTablet("aa", "0", 1, keyspace, "0", topodatapb.TabletType_REPLICA, true, 1, nil)
+		sbc1 := hc.AddTestTablet("aa", "1", 1, keyspace, "1", topodatapb.TabletType_REPLICA, true, 1, nil)
+
+		// empty results
+		sbc0.SetResults([]*sqltypes.Result{{}})
+		sbc1.SetResults([]*sqltypes.Result{{}})
+
+		res := srvtopo.NewResolver(&sandboxTopo{}, sc.gateway, "aa")
+
+		t.Run(test.name, func(t *testing.T) {
+			session := NewSafeSession(&vtgatepb.Session{})
+			for _, action := range test.actions {
+				session.Session.InTransaction = action.transaction
+				session.Session.InReservedConn = action.reserved
+				var destinations []key.Destination
+				for _, shard := range action.shards {
+					destinations = append(destinations, key.DestinationShard(shard))
+				}
+				executeOnShards(t, res, keyspace, sc, session, destinations)
+				assert.EqualValues(t, action.sbc0Reserve, sbc0.ReserveCount.Get(), "sbc0 reserve count")
+				assert.EqualValues(t, action.sbc0Begin, sbc0.BeginCount.Get(), "sbc0 begin count")
+				assert.EqualValues(t, action.sbc1Reserve, sbc1.ReserveCount.Get(), "sbc1 reserve count")
+				assert.EqualValues(t, action.sbc1Begin, sbc1.BeginCount.Get(), "sbc1 begin count")
+				sbc0.BeginCount.Set(0)
+				sbc0.ReserveCount.Set(0)
+				sbc1.BeginCount.Set(0)
+				sbc1.ReserveCount.Set(0)
+			}
+		})
+	}
+}
diff --git a/go/vt/vtgate/tx_conn_test.go b/go/vt/vtgate/tx_conn_test.go
index 89d820ea31..8bfe0c4ffb 100644
--- a/go/vt/vtgate/tx_conn_test.go
+++ b/go/vt/vtgate/tx_conn_test.go
@@ -41,7 +41,7 @@ var queries = []*querypb.BoundQuery{{Sql: "query1"}}
 var twoQueries = []*querypb.BoundQuery{{Sql: "query1"}, {Sql: "query1"}}
 
 func TestTxConnBegin(t *testing.T) {
-	sc, sbc0, _, rss0, _, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, _, rss0, _, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 	session := &vtgatepb.Session{}
 
 	// begin
@@ -61,7 +61,7 @@ func TestTxConnBegin(t *testing.T) {
 }
 
 func TestTxConnCommitSuccess(t *testing.T) {
-	sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, rss0, _, rss01 := newLegacyTestTxConnEnv(t, "TestTxConn")
 	sc.txConn.mode = vtgatepb.TransactionMode_MULTI
 
 	// Sequence the executes to ensure commit order
@@ -112,7 +112,7 @@ func TestTxConnCommitSuccess(t *testing.T) {
 }
 
 func TestTxConnReservedCommitSuccess(t *testing.T) {
-	sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, rss0, _, rss01 := newLegacyTestTxConnEnv(t, "TestTxConn")
 	sc.txConn.mode = vtgatepb.TransactionMode_MULTI
 
 	// Sequence the executes to ensure commit order
@@ -216,7 +216,7 @@ func TestTxConnReservedOn2ShardTxOn1ShardAndCommit(t *testing.T) {
 				TabletType: topodatapb.TabletType_MASTER,
 			},
 			ReservedId:  1,
-			TabletAlias: sbc0.Tablet().Alias,
+			TabletAlias: sbc1.Tablet().Alias,
 		}, {
 			Target: &querypb.Target{
 				Keyspace:   keyspace,
@@ -224,7 +224,7 @@ func TestTxConnReservedOn2ShardTxOn1ShardAndCommit(t *testing.T) {
 				TabletType: topodatapb.TabletType_MASTER,
 			},
 			ReservedId:  1,
-			TabletAlias: sbc1.Tablet().Alias,
+			TabletAlias: sbc0.Tablet().Alias,
 		}},
 	}
 	utils.MustMatch(t, &wantSession, session.Session, "Session")
@@ -309,7 +309,7 @@ func TestTxConnReservedOn2ShardTxOn1ShardAndRollback(t *testing.T) {
 				TabletType: topodatapb.TabletType_MASTER,
 			},
 			ReservedId:  1,
-			TabletAlias: sbc0.Tablet().Alias,
+			TabletAlias: sbc1.Tablet().Alias,
 		}, {
 			Target: &querypb.Target{
 				Keyspace:   keyspace,
@@ -317,7 +317,7 @@ func TestTxConnReservedOn2ShardTxOn1ShardAndRollback(t *testing.T) {
 				TabletType: topodatapb.TabletType_MASTER,
 			},
 			ReservedId:  1,
-			TabletAlias: sbc1.Tablet().Alias,
+			TabletAlias: sbc0.Tablet().Alias,
 		}},
 	}
 	utils.MustMatch(t, &wantSession, session.Session, "Session")
@@ -380,7 +380,7 @@ func TestTxConnReservedOn2ShardTxOn1ShardAndRollback(t *testing.T) {
 }
 
 func TestTxConnCommitOrderFailure1(t *testing.T) {
-	sc, sbc0, sbc1, rss0, rss1, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, rss0, rss1, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 	sc.txConn.mode = vtgatepb.TransactionMode_MULTI
 
 	queries := []*querypb.BoundQuery{{Sql: "query1"}}
@@ -411,7 +411,7 @@ func TestTxConnCommitOrderFailure1(t *testing.T) {
 }
 
 func TestTxConnCommitOrderFailure2(t *testing.T) {
-	sc, sbc0, sbc1, rss0, rss1, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, rss0, rss1, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 	sc.txConn.mode = vtgatepb.TransactionMode_MULTI
 
 	queries := []*querypb.BoundQuery{{
@@ -443,7 +443,7 @@ func TestTxConnCommitOrderFailure2(t *testing.T) {
 }
 
 func TestTxConnCommitOrderFailure3(t *testing.T) {
-	sc, sbc0, sbc1, rss0, rss1, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, rss0, rss1, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 	sc.txConn.mode = vtgatepb.TransactionMode_MULTI
 
 	queries := []*querypb.BoundQuery{{
@@ -478,7 +478,7 @@ func TestTxConnCommitOrderFailure3(t *testing.T) {
 }
 
 func TestTxConnCommitOrderSuccess(t *testing.T) {
-	sc, sbc0, sbc1, rss0, rss1, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, rss0, rss1, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 	sc.txConn.mode = vtgatepb.TransactionMode_MULTI
 
 	queries := []*querypb.BoundQuery{{
@@ -574,7 +574,7 @@ func TestTxConnCommitOrderSuccess(t *testing.T) {
 }
 
 func TestTxConnReservedCommitOrderSuccess(t *testing.T) {
-	sc, sbc0, sbc1, rss0, rss1, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, rss0, rss1, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 	sc.txConn.mode = vtgatepb.TransactionMode_MULTI
 
 	queries := []*querypb.BoundQuery{{
@@ -715,7 +715,7 @@ func TestTxConnReservedCommitOrderSuccess(t *testing.T) {
 }
 
 func TestTxConnCommit2PC(t *testing.T) {
-	sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, "TestTxConnCommit2PC")
+	sc, sbc0, sbc1, rss0, _, rss01 := newLegacyTestTxConnEnv(t, "TestTxConnCommit2PC")
 
 	session := NewSafeSession(&vtgatepb.Session{InTransaction: true})
 	sc.ExecuteMultiShard(ctx, rss0, queries, session, false, false)
@@ -731,7 +731,7 @@ func TestTxConnCommit2PC(t *testing.T) {
 }
 
 func TestTxConnCommit2PCOneParticipant(t *testing.T) {
-	sc, sbc0, _, rss0, _, _ := newTestTxConnEnv(t, "TestTxConnCommit2PCOneParticipant")
+	sc, sbc0, _, rss0, _, _ := newLegacyTestTxConnEnv(t, "TestTxConnCommit2PCOneParticipant")
 	session := NewSafeSession(&vtgatepb.Session{InTransaction: true})
 	sc.ExecuteMultiShard(ctx, rss0, queries, session, false, false)
 	session.TransactionMode = vtgatepb.TransactionMode_TWOPC
@@ -741,7 +741,7 @@ func TestTxConnCommit2PCOneParticipant(t *testing.T) {
 }
 
 func TestTxConnCommit2PCCreateTransactionFail(t *testing.T) {
-	sc, sbc0, sbc1, rss0, rss1, _ := newTestTxConnEnv(t, "TestTxConnCommit2PCCreateTransactionFail")
+	sc, sbc0, sbc1, rss0, rss1, _ := newLegacyTestTxConnEnv(t, "TestTxConnCommit2PCCreateTransactionFail")
 
 	session := NewSafeSession(&vtgatepb.Session{InTransaction: true})
 	sc.ExecuteMultiShard(ctx, rss0, queries, session, false, false)
@@ -763,7 +763,7 @@ func TestTxConnCommit2PCCreateTransactionFail(t *testing.T) {
 }
 
 func TestTxConnCommit2PCPrepareFail(t *testing.T) {
-	sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, "TestTxConnCommit2PCPrepareFail")
+	sc, sbc0, sbc1, rss0, _, rss01 := newLegacyTestTxConnEnv(t, "TestTxConnCommit2PCPrepareFail")
 
 	session := NewSafeSession(&vtgatepb.Session{InTransaction: true})
 	sc.ExecuteMultiShard(ctx, rss0, queries, session, false, false)
@@ -783,7 +783,7 @@ func TestTxConnCommit2PCPrepareFail(t *testing.T) {
 }
 
 func TestTxConnCommit2PCStartCommitFail(t *testing.T) {
-	sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, "TestTxConnCommit2PCStartCommitFail")
+	sc, sbc0, sbc1, rss0, _, rss01 := newLegacyTestTxConnEnv(t, "TestTxConnCommit2PCStartCommitFail")
 
 	session := NewSafeSession(&vtgatepb.Session{InTransaction: true})
 	sc.ExecuteMultiShard(ctx, rss0, queries, session, false, false)
@@ -803,7 +803,7 @@ func TestTxConnCommit2PCStartCommitFail(t *testing.T) {
 }
 
 func TestTxConnCommit2PCCommitPreparedFail(t *testing.T) {
-	sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, "TestTxConnCommit2PCCommitPreparedFail")
+	sc, sbc0, sbc1, rss0, _, rss01 := newLegacyTestTxConnEnv(t, "TestTxConnCommit2PCCommitPreparedFail")
 
 	session := NewSafeSession(&vtgatepb.Session{InTransaction: true})
 	sc.ExecuteMultiShard(ctx, rss0, queries, session, false, false)
@@ -823,7 +823,7 @@ func TestTxConnCommit2PCCommitPreparedFail(t *testing.T) {
 }
 
 func TestTxConnCommit2PCConcludeTransactionFail(t *testing.T) {
-	sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, "TestTxConnCommit2PCConcludeTransactionFail")
+	sc, sbc0, sbc1, rss0, _, rss01 := newLegacyTestTxConnEnv(t, "TestTxConnCommit2PCConcludeTransactionFail")
 
 	session := NewSafeSession(&vtgatepb.Session{InTransaction: true})
 	sc.ExecuteMultiShard(ctx, rss0, queries, session, false, false)
@@ -843,7 +843,7 @@ func TestTxConnCommit2PCConcludeTransactionFail(t *testing.T) {
 }
 
 func TestTxConnRollback(t *testing.T) {
-	sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, "TxConnRollback")
+	sc, sbc0, sbc1, rss0, _, rss01 := newLegacyTestTxConnEnv(t, "TxConnRollback")
 
 	session := NewSafeSession(&vtgatepb.Session{InTransaction: true})
 	sc.ExecuteMultiShard(ctx, rss0, queries, session, false, false)
@@ -857,7 +857,7 @@ func TestTxConnRollback(t *testing.T) {
 }
 
 func TestTxConnReservedRollback(t *testing.T) {
-	sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, "TxConnReservedRollback")
+	sc, sbc0, sbc1, rss0, _, rss01 := newLegacyTestTxConnEnv(t, "TxConnReservedRollback")
 
 	session := NewSafeSession(&vtgatepb.Session{InTransaction: true, InReservedConn: true})
 	sc.ExecuteMultiShard(ctx, rss0, queries, session, false, false)
@@ -892,7 +892,7 @@ func TestTxConnReservedRollback(t *testing.T) {
 }
 
 func TestTxConnReservedRollbackFailure(t *testing.T) {
-	sc, sbc0, sbc1, rss0, _, rss01 := newTestTxConnEnv(t, "TxConnReservedRollback")
+	sc, sbc0, sbc1, rss0, _, rss01 := newLegacyTestTxConnEnv(t, "TxConnReservedRollback")
 
 	session := NewSafeSession(&vtgatepb.Session{InTransaction: true, InReservedConn: true})
 	sc.ExecuteMultiShard(ctx, rss0, queries, session, false, false)
@@ -914,7 +914,7 @@ func TestTxConnReservedRollbackFailure(t *testing.T) {
 }
 
 func TestTxConnResolveOnPrepare(t *testing.T) {
-	sc, sbc0, sbc1, _, _, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, _, _, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 
 	dtid := "TestTxConn:0:1234"
 	sbc0.ReadTransactionResults = []*querypb.TransactionMetadata{{
@@ -935,7 +935,7 @@ func TestTxConnResolveOnPrepare(t *testing.T) {
 }
 
 func TestTxConnResolveOnRollback(t *testing.T) {
-	sc, sbc0, sbc1, _, _, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, _, _, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 
 	dtid := "TestTxConn:0:1234"
 	sbc0.ReadTransactionResults = []*querypb.TransactionMetadata{{
@@ -956,7 +956,7 @@ func TestTxConnResolveOnRollback(t *testing.T) {
 }
 
 func TestTxConnResolveOnCommit(t *testing.T) {
-	sc, sbc0, sbc1, _, _, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, _, _, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 
 	dtid := "TestTxConn:0:1234"
 	sbc0.ReadTransactionResults = []*querypb.TransactionMetadata{{
@@ -977,7 +977,7 @@ func TestTxConnResolveOnCommit(t *testing.T) {
 }
 
 func TestTxConnResolveInvalidDTID(t *testing.T) {
-	sc, _, _, _, _, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, _, _, _, _, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 
 	err := sc.txConn.Resolve(ctx, "abcd")
 	want := "invalid parts in dtid: abcd"
@@ -985,7 +985,7 @@ func TestTxConnResolveInvalidDTID(t *testing.T) {
 }
 
 func TestTxConnResolveReadTransactionFail(t *testing.T) {
-	sc, sbc0, _, _, _, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, _, _, _, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 
 	dtid := "TestTxConn:0:1234"
 	sbc0.MustFailCodes[vtrpcpb.Code_INVALID_ARGUMENT] = 1
@@ -996,7 +996,7 @@ func TestTxConnResolveReadTransactionFail(t *testing.T) {
 }
 
 func TestTxConnResolveInternalError(t *testing.T) {
-	sc, sbc0, _, _, _, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, _, _, _, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 
 	dtid := "TestTxConn:0:1234"
 	sbc0.ReadTransactionResults = []*querypb.TransactionMetadata{{
@@ -1015,7 +1015,7 @@ func TestTxConnResolveInternalError(t *testing.T) {
 }
 
 func TestTxConnResolveSetRollbackFail(t *testing.T) {
-	sc, sbc0, sbc1, _, _, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, _, _, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 
 	dtid := "TestTxConn:0:1234"
 	sbc0.ReadTransactionResults = []*querypb.TransactionMetadata{{
@@ -1039,7 +1039,7 @@ func TestTxConnResolveSetRollbackFail(t *testing.T) {
 }
 
 func TestTxConnResolveRollbackPreparedFail(t *testing.T) {
-	sc, sbc0, sbc1, _, _, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, _, _, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 
 	dtid := "TestTxConn:0:1234"
 	sbc0.ReadTransactionResults = []*querypb.TransactionMetadata{{
@@ -1063,7 +1063,7 @@ func TestTxConnResolveRollbackPreparedFail(t *testing.T) {
 }
 
 func TestTxConnResolveCommitPreparedFail(t *testing.T) {
-	sc, sbc0, sbc1, _, _, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, _, _, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 
 	dtid := "TestTxConn:0:1234"
 	sbc0.ReadTransactionResults = []*querypb.TransactionMetadata{{
@@ -1087,7 +1087,7 @@ func TestTxConnResolveCommitPreparedFail(t *testing.T) {
 }
 
 func TestTxConnResolveConcludeTransactionFail(t *testing.T) {
-	sc, sbc0, sbc1, _, _, _ := newTestTxConnEnv(t, "TestTxConn")
+	sc, sbc0, sbc1, _, _, _ := newLegacyTestTxConnEnv(t, "TestTxConn")
 
 	dtid := "TestTxConn:0:1234"
 	sbc0.ReadTransactionResults = []*querypb.TransactionMetadata{{
@@ -1177,7 +1177,7 @@ func TestTxConnMultiGoTargets(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func newTestTxConnEnv(t *testing.T, name string) (sc *ScatterConn, sbc0, sbc1 *sandboxconn.SandboxConn, rss0, rss1, rss01 []*srvtopo.ResolvedShard) {
+func newLegacyTestTxConnEnv(t *testing.T, name string) (sc *ScatterConn, sbc0, sbc1 *sandboxconn.SandboxConn, rss0, rss1, rss01 []*srvtopo.ResolvedShard) {
 	t.Helper()
 	createSandbox(name)
 	hc := discovery.NewFakeLegacyHealthCheck()
@@ -1194,3 +1194,21 @@ func newTestTxConnEnv(t *testing.T, name string) (sc *ScatterConn, sbc0, sbc1 *s
 	require.NoError(t, err)
 	return sc, sbc0, sbc1, rss0, rss1, rss01
 }
+
+func newTestTxConnEnv(t *testing.T, name string) (sc *ScatterConn, sbc0, sbc1 *sandboxconn.SandboxConn, rss0, rss1, rss01 []*srvtopo.ResolvedShard) {
+	t.Helper()
+	createSandbox(name)
+	hc := discovery.NewFakeHealthCheck()
+	sc = newTestScatterConn(hc, new(sandboxTopo), "aa")
+	sbc0 = hc.AddTestTablet("aa", "0", 1, name, "0", topodatapb.TabletType_MASTER, true, 1, nil)
+	sbc1 = hc.AddTestTablet("aa", "1", 1, name, "1", topodatapb.TabletType_MASTER, true, 1, nil)
+	res := srvtopo.NewResolver(&sandboxTopo{}, sc.gateway, "aa")
+	var err error
+	rss0, err = res.ResolveDestination(ctx, name, topodatapb.TabletType_MASTER, key.DestinationShard("0"))
+	require.NoError(t, err)
+	rss1, err = res.ResolveDestination(ctx, name, topodatapb.TabletType_MASTER, key.DestinationShard("1"))
+	require.NoError(t, err)
+	rss01, err = res.ResolveDestination(ctx, name, topodatapb.TabletType_MASTER, key.DestinationShards([]string{"0", "1"}))
+	require.NoError(t, err)
+	return sc, sbc0, sbc1, rss0, rss1, rss01
+}
diff --git a/go/vt/vtgate/vcursor_impl_test.go b/go/vt/vtgate/vcursor_impl_test.go
index 823f8e77eb..dd956f6b51 100644
--- a/go/vt/vtgate/vcursor_impl_test.go
+++ b/go/vt/vtgate/vcursor_impl_test.go
@@ -195,33 +195,22 @@ func TestDestinationKeyspace(t *testing.T) {
 	}
 }
 
-func TestSetTarget(t *testing.T) {
-	ks1 := &vindexes.Keyspace{
-		Name:    "ks1",
-		Sharded: false,
-	}
-	ks1Schema := &vindexes.KeyspaceSchema{
-		Keyspace: ks1,
-		Tables:   nil,
-		Vindexes: nil,
-		Error:    nil,
-	}
-	ks2 := &vindexes.Keyspace{
-		Name:    "ks2",
-		Sharded: false,
-	}
-	ks2Schema := &vindexes.KeyspaceSchema{
-		Keyspace: ks2,
-		Tables:   nil,
-		Vindexes: nil,
-		Error:    nil,
-	}
-	vschemaWith2KS := &vindexes.VSchema{
-		Keyspaces: map[string]*vindexes.KeyspaceSchema{
-			ks1.Name: ks1Schema,
-			ks2.Name: ks2Schema,
-		}}
+var ks1 = &vindexes.Keyspace{Name: "ks1"}
+var ks1Schema = &vindexes.KeyspaceSchema{Keyspace: ks1}
+var ks2 = &vindexes.Keyspace{Name: "ks2"}
+var ks2Schema = &vindexes.KeyspaceSchema{Keyspace: ks2}
+var vschemaWith1KS = &vindexes.VSchema{
+	Keyspaces: map[string]*vindexes.KeyspaceSchema{
+		ks1.Name: ks1Schema,
+	},
+}
+var vschemaWith2KS = &vindexes.VSchema{
+	Keyspaces: map[string]*vindexes.KeyspaceSchema{
+		ks1.Name: ks1Schema,
+		ks2.Name: ks2Schema,
+	}}
 
+func TestSetTarget(t *testing.T) {
 	type testCase struct {
 		vschema       *vindexes.VSchema
 		targetString  string
@@ -263,22 +252,6 @@ func TestSetTarget(t *testing.T) {
 }
 
 func TestPlanPrefixKey(t *testing.T) {
-	ks1 := &vindexes.Keyspace{
-		Name:    "ks1",
-		Sharded: false,
-	}
-	ks1Schema := &vindexes.KeyspaceSchema{
-		Keyspace: ks1,
-		Tables:   nil,
-		Vindexes: nil,
-		Error:    nil,
-	}
-	vschemaWith1KS := &vindexes.VSchema{
-		Keyspaces: map[string]*vindexes.KeyspaceSchema{
-			ks1.Name: ks1Schema,
-		},
-	}
-
 	type testCase struct {
 		vschema               *vindexes.VSchema
 		targetString          string
@@ -314,3 +287,21 @@ func TestPlanPrefixKey(t *testing.T) {
 		})
 	}
 }
+
+func TestFirstSortedKeyspace(t *testing.T) {
+	ks1Schema := &vindexes.KeyspaceSchema{Keyspace: &vindexes.Keyspace{Name: "xks1"}}
+	ks2Schema := &vindexes.KeyspaceSchema{Keyspace: &vindexes.Keyspace{Name: "aks2"}}
+	ks3Schema := &vindexes.KeyspaceSchema{Keyspace: &vindexes.Keyspace{Name: "aks1"}}
+	vschemaWith2KS := &vindexes.VSchema{
+		Keyspaces: map[string]*vindexes.KeyspaceSchema{
+			ks1Schema.Keyspace.Name: ks1Schema,
+			ks2Schema.Keyspace.Name: ks2Schema,
+			ks3Schema.Keyspace.Name: ks3Schema,
+		}}
+
+	vc, err := newVCursorImpl(context.Background(), NewSafeSession(nil), sqlparser.MarginComments{}, nil, nil, &fakeVSchemaOperator{vschema: vschemaWith2KS}, vschemaWith2KS, srvtopo.NewResolver(&fakeTopoServer{}, nil, ""))
+	require.NoError(t, err)
+	ks, err := vc.FirstSortedKeyspace()
+	require.NoError(t, err)
+	require.Equal(t, ks3Schema.Keyspace, ks)
+}
diff --git a/go/vt/vttablet/tabletserver/planbuilder/plan_test.go b/go/vt/vttablet/tabletserver/planbuilder/plan_test.go
index 8f5f030c57..f1314592c1 100644
--- a/go/vt/vttablet/tabletserver/planbuilder/plan_test.go
+++ b/go/vt/vttablet/tabletserver/planbuilder/plan_test.go
@@ -72,7 +72,76 @@ func TestPlan(t *testing.T) {
 			var err error
 			statement, err := sqlparser.Parse(tcase.input)
 			if err == nil {
-				plan, err = Build(statement, testSchema)
+				plan, err = Build(statement, testSchema, false)
+			}
+			PassthroughDMLs = false
+
+			var out string
+			if err != nil {
+				out = err.Error()
+			} else {
+				bout, err := json.Marshal(plan)
+				if err != nil {
+					t.Fatalf("Error marshalling %v: %v", plan, err)
+				}
+				out = string(bout)
+			}
+			if out != tcase.output {
+				t.Errorf("Line:%v\ngot  = %s\nwant = %s", tcase.lineno, out, tcase.output)
+				if err != nil {
+					out = fmt.Sprintf("\"%s\"", out)
+				} else {
+					bout, _ := json.MarshalIndent(plan, "", "  ")
+					out = string(bout)
+				}
+				fmt.Printf("\"%s\"\n%s\n\n", tcase.input, out)
+			}
+		})
+	}
+}
+
+func TestPlanPoolUnsafe(t *testing.T) {
+	testSchema := loadSchema("schema_test.json")
+	for tcase := range iterateExecFile("pool_unsafe_cases.txt") {
+		t.Run(tcase.input, func(t *testing.T) {
+			var plan *Plan
+			var err error
+			statement, err := sqlparser.Parse(tcase.input)
+			require.NoError(t, err)
+			// In Pooled Connection, plan building will fail.
+			plan, err = Build(statement, testSchema, false /* isReservedConn */)
+			require.Error(t, err)
+			out := err.Error()
+			if out != tcase.output {
+				t.Errorf("Line:%v\ngot  = %s\nwant = %s", tcase.lineno, out, tcase.output)
+				if err != nil {
+					out = fmt.Sprintf("\"%s\"", out)
+				} else {
+					bout, _ := json.MarshalIndent(plan, "", "  ")
+					out = string(bout)
+				}
+				fmt.Printf("\"%s\"\n%s\n\n", tcase.input, out)
+			}
+			// In Reserved Connection, plan will be built.
+			plan, err = Build(statement, testSchema, true /* isReservedConn */)
+			require.NoError(t, err)
+			require.NotEmpty(t, plan)
+		})
+	}
+}
+
+func TestPlanInReservedConn(t *testing.T) {
+	testSchema := loadSchema("schema_test.json")
+	for tcase := range iterateExecFile("exec_cases.txt") {
+		t.Run(tcase.input, func(t *testing.T) {
+			if strings.Contains(tcase.options, "PassthroughDMLs") {
+				PassthroughDMLs = true
+			}
+			var plan *Plan
+			var err error
+			statement, err := sqlparser.Parse(tcase.input)
+			if err == nil {
+				plan, err = Build(statement, testSchema, true)
 			}
 			PassthroughDMLs = false
 
@@ -123,7 +192,7 @@ func TestCustom(t *testing.T) {
 				if err != nil {
 					t.Fatalf("Got error: %v, parsing sql: %v", err.Error(), tcase.input)
 				}
-				plan, err := Build(statement, schem)
+				plan, err := Build(statement, schem, false)
 				var out string
 				if err != nil {
 					out = err.Error()
@@ -145,7 +214,7 @@ func TestCustom(t *testing.T) {
 func TestStreamPlan(t *testing.T) {
 	testSchema := loadSchema("schema_test.json")
 	for tcase := range iterateExecFile("stream_cases.txt") {
-		plan, err := BuildStreaming(tcase.input, testSchema)
+		plan, err := BuildStreaming(tcase.input, testSchema, false)
 		var out string
 		if err != nil {
 			out = err.Error()
diff --git a/go/vt/vttablet/tabletserver/planbuilder/testdata/exec_cases.txt b/go/vt/vttablet/tabletserver/planbuilder/testdata/exec_cases.txt
index a38170b92d..185afbafd8 100644
--- a/go/vt/vttablet/tabletserver/planbuilder/testdata/exec_cases.txt
+++ b/go/vt/vttablet/tabletserver/planbuilder/testdata/exec_cases.txt
@@ -787,7 +787,3 @@ options:PassthroughDMLs
 # syntax error
 "syntax error"
 "syntax error at position 7 near 'syntax'"
-
-# named locks are unsafe with server-side connection pooling
-"select get_lock('foo') from dual"
-"get_lock() not allowed"
diff --git a/go/vt/vttablet/tabletserver/planbuilder/testdata/pool_unsafe_cases.txt b/go/vt/vttablet/tabletserver/planbuilder/testdata/pool_unsafe_cases.txt
new file mode 100644
index 0000000000..e7e6b9d3e4
--- /dev/null
+++ b/go/vt/vttablet/tabletserver/planbuilder/testdata/pool_unsafe_cases.txt
@@ -0,0 +1,19 @@
+# get_lock named locks are unsafe with server-side connection pooling
+"select get_lock('foo', 10) from dual"
+"get_lock() not allowed"
+
+# is_free_lock named locks are unsafe with server-side connection pooling
+"select is_free_lock('foo') from dual"
+"is_free_lock() not allowed"
+
+# is_used_lock named locks are unsafe with server-side connection pooling
+"select is_used_lock('foo') from dual"
+"is_used_lock() not allowed"
+
+# release_all_locks named locks are unsafe with server-side connection pooling
+"select release_all_locks() from dual"
+"release_all_locks() not allowed"
+
+# release_lock named locks are unsafe with server-side connection pooling
+"select release_lock('foo') from dual"
+"release_lock() not allowed"
diff --git a/go/vt/vttablet/tabletserver/query_engine_test.go b/go/vt/vttablet/tabletserver/query_engine_test.go
index 902ce85820..e89ab569ee 100644
--- a/go/vt/vttablet/tabletserver/query_engine_test.go
+++ b/go/vt/vttablet/tabletserver/query_engine_test.go
@@ -98,7 +98,7 @@ func TestGetPlanPanicDuetoEmptyQuery(t *testing.T) {
 
 	ctx := context.Background()
 	logStats := tabletenv.NewLogStats(ctx, "GetPlanStats")
-	_, err := qe.GetPlan(ctx, logStats, "", false)
+	_, err := qe.GetPlan(ctx, logStats, "", false, false /* inReservedConn */)
 	want := "empty statement"
 	if err == nil || !strings.Contains(err.Error(), want) {
 		t.Errorf("qe.GetPlan: %v, want %s", err, want)
@@ -156,14 +156,14 @@ func TestQueryPlanCache(t *testing.T) {
 	ctx := context.Background()
 	logStats := tabletenv.NewLogStats(ctx, "GetPlanStats")
 	qe.SetQueryPlanCacheCap(1)
-	firstPlan, err := qe.GetPlan(ctx, logStats, firstQuery, false)
+	firstPlan, err := qe.GetPlan(ctx, logStats, firstQuery, false, false /* inReservedConn */)
 	if err != nil {
 		t.Fatal(err)
 	}
 	if firstPlan == nil {
 		t.Fatalf("plan should not be nil")
 	}
-	secondPlan, err := qe.GetPlan(ctx, logStats, secondQuery, false)
+	secondPlan, err := qe.GetPlan(ctx, logStats, secondQuery, false, false /* inReservedConn */)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -198,7 +198,7 @@ func TestNoQueryPlanCache(t *testing.T) {
 	ctx := context.Background()
 	logStats := tabletenv.NewLogStats(ctx, "GetPlanStats")
 	qe.SetQueryPlanCacheCap(1)
-	firstPlan, err := qe.GetPlan(ctx, logStats, firstQuery, true)
+	firstPlan, err := qe.GetPlan(ctx, logStats, firstQuery, true, false /* inReservedConn */)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -230,7 +230,7 @@ func TestNoQueryPlanCacheDirective(t *testing.T) {
 	ctx := context.Background()
 	logStats := tabletenv.NewLogStats(ctx, "GetPlanStats")
 	qe.SetQueryPlanCacheCap(1)
-	firstPlan, err := qe.GetPlan(ctx, logStats, firstQuery, false)
+	firstPlan, err := qe.GetPlan(ctx, logStats, firstQuery, false, false /* inReservedConn */)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -258,7 +258,7 @@ func TestStatsURL(t *testing.T) {
 	// warm up cache
 	ctx := context.Background()
 	logStats := tabletenv.NewLogStats(ctx, "GetPlanStats")
-	qe.GetPlan(ctx, logStats, query, false)
+	qe.GetPlan(ctx, logStats, query, false, false /* inReservedConn */)
 
 	request, _ := http.NewRequest("GET", "/debug/tablet_plans", nil)
 	response := httptest.NewRecorder()
diff --git a/go/vt/vttablet/tabletserver/query_executor_test.go b/go/vt/vttablet/tabletserver/query_executor_test.go
index 922cf7d276..df98ad74c4 100644
--- a/go/vt/vttablet/tabletserver/query_executor_test.go
+++ b/go/vt/vttablet/tabletserver/query_executor_test.go
@@ -1127,7 +1127,7 @@ func newTransaction(tsv *TabletServer, options *querypb.ExecuteOptions) int64 {
 
 func newTestQueryExecutor(ctx context.Context, tsv *TabletServer, sql string, txID int64) *QueryExecutor {
 	logStats := tabletenv.NewLogStats(ctx, "TestQueryExecutor")
-	plan, err := tsv.qe.GetPlan(ctx, logStats, sql, false)
+	plan, err := tsv.qe.GetPlan(ctx, logStats, sql, false, false /* inReservedConn */)
 	if err != nil {
 		panic(err)
 	}