Repository: vitessio/vitess
Pull request #6459 (2020-07-20 22:47:27)

System Variables support using Reserved Connections
Closes: #6107

This PR has some framework dependency on #6370 


Patch:
diff --git a/go/mysql/constants.go b/go/mysql/constants.go
index a2d6ec3c0a..7236a7e742 100644
--- a/go/mysql/constants.go
+++ b/go/mysql/constants.go
@@ -560,11 +560,10 @@ var CharacterSetMap = map[string]uint8{
 
 // IsNum returns true if a MySQL type is a numeric value.
 // It is the same as IS_NUM defined in mysql.h.
-//
-// FIXME(alainjobart) This needs to use the constants in
-// replication/constants.go, so we are using numerical values here.
 func IsNum(typ uint8) bool {
-	return ((typ <= 9 /* MYSQL_TYPE_INT24 */ && typ != 7 /* MYSQL_TYPE_TIMESTAMP */) || typ == 13 /* MYSQL_TYPE_YEAR */ || typ == 246 /* MYSQL_TYPE_NEWDECIMAL */)
+	return (typ <= TypeInt24 && typ != TypeTimestamp) ||
+		typ == TypeYear ||
+		typ == TypeNewDecimal
 }
 
 // IsConnErr returns true if the error is a connection error.
diff --git a/go/vt/sqlparser/normalizer.go b/go/vt/sqlparser/normalizer.go
index 36f2c08c25..23660fbec9 100644
--- a/go/vt/sqlparser/normalizer.go
+++ b/go/vt/sqlparser/normalizer.go
@@ -61,6 +61,9 @@ func newNormalizer(stmt Statement, bindVars map[string]*querypb.BindVariable, pr
 // where variables are deduped.
 func (nz *normalizer) WalkStatement(node SQLNode) (bool, error) {
 	switch node := node.(type) {
+	// no need to normalize the statement types
+	case *Set, *Show, *Begin, *Commit, *Rollback, *Savepoint, *SetTransaction, *DDL, *SRollback, *Release, *OtherAdmin, *OtherRead:
+		return false, nil
 	case *Select:
 		_ = Walk(nz.WalkSelect, node)
 		// Don't continue
diff --git a/go/vt/vtgate/engine/primitive.go b/go/vt/vtgate/engine/primitive.go
index d094ce6ed0..db31ea39b1 100644
--- a/go/vt/vtgate/engine/primitive.go
+++ b/go/vt/vtgate/engine/primitive.go
@@ -97,8 +97,14 @@ type (
 
 		SetSysVar(name string, expr string)
 
-		// NeedsReservedConn marks this session as needing a dedicated connection to mysql
+		// NeedsReservedConn marks this session as needing a dedicated connection to underlying database
 		NeedsReservedConn()
+
+		// InReservedConn provides whether this session is using reserved connection
+		InReservedConn() bool
+
+		// ShardSession returns shard info about open connections
+		ShardSession() []*srvtopo.ResolvedShard
 	}
 
 	// Plan represents the execution strategy for a given query.
diff --git a/go/vt/vtgate/engine/set.go b/go/vt/vtgate/engine/set.go
index 10ad7deeb4..042d34528f 100644
--- a/go/vt/vtgate/engine/set.go
+++ b/go/vt/vtgate/engine/set.go
@@ -17,12 +17,15 @@ limitations under the License.
 package engine
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
 
-	"vitess.io/vitess/go/vt/vtgate/evalengine"
+	"vitess.io/vitess/go/vt/log"
+
+	"vitess.io/vitess/go/vt/srvtopo"
 
-	"vitess.io/vitess/go/mysql"
+	"vitess.io/vitess/go/vt/vtgate/evalengine"
 
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/key"
@@ -63,7 +66,7 @@ type (
 	SysVarCheckAndIgnore struct {
 		Name              string
 		Keyspace          *vindexes.Keyspace
-		TargetDestination key.Destination
+		TargetDestination key.Destination `json:",omitempty"`
 		Expr              string
 	}
 
@@ -71,7 +74,16 @@ type (
 	SysVarSet struct {
 		Name              string
 		Keyspace          *vindexes.Keyspace
-		TargetDestination key.Destination
+		TargetDestination key.Destination `json:",omitempty"`
+		Expr              string
+	}
+
+	// SysVarSetSpecial implements the SetOp interface and will write the changes variable into the session
+	// The special part is that these settings change the sessions behaviour in different ways
+	SysVarSetSpecial struct {
+		Name              string
+		Keyspace          *vindexes.Keyspace
+		TargetDestination key.Destination `json:",omitempty"`
 		Expr              string
 	}
 )
@@ -140,7 +152,6 @@ func (s *Set) description() PrimitiveDescription {
 	}
 	return PrimitiveDescription{
 		OperatorType: "Set",
-		Variant:      "",
 		Other:        other,
 	}
 }
@@ -195,8 +206,8 @@ func (svi *SysVarIgnore) VariableName() string {
 }
 
 //Execute implements the SetOp interface method.
-func (svi *SysVarIgnore) Execute(vcursor VCursor, _ evalengine.ExpressionEnv) error {
-	vcursor.Session().RecordWarning(&querypb.QueryWarning{Code: mysql.ERNotSupportedYet, Message: fmt.Sprintf("Ignored inapplicable SET %v = %v", svi.Name, svi.Expr)})
+func (svi *SysVarIgnore) Execute(VCursor, evalengine.ExpressionEnv) error {
+	log.Infof("Ignored inapplicable SET %v = %v", svi.Name, svi.Expr)
 	return nil
 }
 
@@ -234,13 +245,9 @@ func (svci *SysVarCheckAndIgnore) Execute(vcursor VCursor, env evalengine.Expres
 	if err != nil {
 		return err
 	}
-	var warning *querypb.QueryWarning
 	if result.RowsAffected == 0 {
-		warning = &querypb.QueryWarning{Code: mysql.ERNotSupportedYet, Message: fmt.Sprintf("Modification not allowed using set construct for: %s", svci.Name)}
-	} else {
-		warning = &querypb.QueryWarning{Code: mysql.ERNotSupportedYet, Message: fmt.Sprintf("Ignored inapplicable SET %v = %v", svci.Name, svci.Expr)}
+		log.Infof("Ignored inapplicable SET %v = %v", svci.Name, svci.Expr)
 	}
-	vcursor.Session().RecordWarning(warning)
 	return nil
 }
 
@@ -264,20 +271,70 @@ func (svs *SysVarSet) VariableName() string {
 }
 
 //Execute implements the SetOp interface method
-func (svs *SysVarSet) Execute(vcursor VCursor, res evalengine.ExpressionEnv) error {
-	rss, _, err := vcursor.ResolveDestinations(svs.Keyspace.Name, nil, []key.Destination{svs.TargetDestination})
+func (svs *SysVarSet) Execute(vcursor VCursor, env evalengine.ExpressionEnv) error {
+	// For those running on advanced vitess settings.
+	if svs.TargetDestination != nil {
+		rss, _, err := vcursor.ResolveDestinations(svs.Keyspace.Name, nil, []key.Destination{svs.TargetDestination})
+		if err != nil {
+			return vterrors.Wrap(err, "SysVarSet")
+		}
+		vcursor.Session().NeedsReservedConn()
+		return svs.execSetStatement(vcursor, rss, env)
+	}
+	isSysVarModified, err := svs.checkAndUpdateSysVar(vcursor, env)
 	if err != nil {
-		return vterrors.Wrap(err, "SysVarSet")
+		return err
+	}
+	if !isSysVarModified {
+		// setting ignored, same as underlying datastore
+		return nil
 	}
+	// Update existing shard session with new system variable settings.
+	rss := vcursor.Session().ShardSession()
+	if len(rss) == 0 {
+		return nil
+	}
+	queries := make([]*querypb.BoundQuery, len(rss))
+	for i := 0; i < len(rss); i++ {
+		queries[i] = &querypb.BoundQuery{
+			Sql:           fmt.Sprintf("set @@%s = %s", svs.Name, svs.Expr),
+			BindVariables: env.BindVars,
+		}
+	}
+	_, errs := vcursor.ExecuteMultiShard(rss, queries, false /* rollbackOnError */, false /* canAutocommit */)
+	return vterrors.Aggregate(errs)
+}
 
-	if len(rss) != 1 {
-		return vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "Unexpected error, DestinationKeyspaceID mapping to multiple shards: %v", svs.TargetDestination)
+func (svs *SysVarSet) execSetStatement(vcursor VCursor, rss []*srvtopo.ResolvedShard, env evalengine.ExpressionEnv) error {
+	queries := make([]*querypb.BoundQuery, len(rss))
+	for i := 0; i < len(rss); i++ {
+		queries[i] = &querypb.BoundQuery{
+			Sql:           fmt.Sprintf("set @@%s = %s", svs.Name, svs.Expr),
+			BindVariables: env.BindVars,
+		}
 	}
-	sysVarExprValidationQuery := fmt.Sprintf("select %s from dual where false", svs.Expr)
-	_, err = execShard(vcursor, sysVarExprValidationQuery, res.BindVars, rss[0], false /* rollbackOnError */, false /* canAutocommit */)
+	_, errs := vcursor.ExecuteMultiShard(rss, queries, false /* rollbackOnError */, false /* canAutocommit */)
+	return vterrors.Aggregate(errs)
+}
+
+func (svs *SysVarSet) checkAndUpdateSysVar(vcursor VCursor, res evalengine.ExpressionEnv) (bool, error) {
+	sysVarExprValidationQuery := fmt.Sprintf("select %s from dual where @@%s != %s", svs.Expr, svs.Name, svs.Expr)
+	rss, _, err := vcursor.ResolveDestinations(svs.Keyspace.Name, nil, []key.Destination{key.DestinationKeyspaceID{0}})
 	if err != nil {
-		return err
+		return false, vterrors.Wrap(err, "SysVarSet")
 	}
-	vcursor.Session().SetSysVar(svs.Name, svs.Expr)
-	return nil
+	qr, err := execShard(vcursor, sysVarExprValidationQuery, res.BindVars, rss[0], false /* rollbackOnError */, false /* canAutocommit */)
+	if err != nil {
+		return false, err
+	}
+	if len(qr.Rows) == 0 {
+		return false, nil
+	}
+	// TODO : validate how value needs to be stored.
+	value := qr.Rows[0][0]
+	buf := new(bytes.Buffer)
+	value.EncodeSQL(buf)
+	vcursor.Session().SetSysVar(svs.Name, buf.String())
+	vcursor.Session().NeedsReservedConn()
+	return true, nil
 }
diff --git a/go/vt/vtgate/planbuilder/set.go b/go/vt/vtgate/planbuilder/set.go
index 7bab90dc40..8bfd5ddf06 100644
--- a/go/vt/vtgate/planbuilder/set.go
+++ b/go/vt/vtgate/planbuilder/set.go
@@ -31,12 +31,132 @@ import (
 	"vitess.io/vitess/go/vt/vtgate/evalengine"
 )
 
-var sysVarPlanningFunc = map[string]func(expr *sqlparser.SetExpr, vschema ContextVSchema) (engine.SetOp, error){}
+type planFunc = func(expr *sqlparser.SetExpr, vschema ContextVSchema) (engine.SetOp, error)
+
+var sysVarPlanningFunc = map[string]planFunc{}
+
+var notSupported = []string{
+	"auto_increment_increment",
+	"auto_increment_offset",
+	"binlog_direct_non_transactional_updates",
+	"innodb_ft_enable_stopword",
+	"innodb_ft_user_stopword_table",
+	"max_points_in_geometry",
+	"max_sp_recursion_depth",
+	"myisam_repair_threads",
+	"myisam_sort_buffer_size",
+	"myisam_stats_method",
+	"ndb_allow_copying_alter_table",
+	"ndb_autoincrement_prefetch_sz",
+	"ndb_blob_read_batch_bytes",
+	"ndb_blob_write_batch_bytes",
+	"ndb_deferred_constraints",
+	"ndb_force_send",
+	"ndb_fully_replicated",
+	"ndb_index_stat_enable",
+	"ndb_index_stat_option",
+	"ndb_join_pushdown",
+	"ndb_log_bin",
+	"ndb_log_exclusive_reads",
+	"ndb_row_checksum",
+	"ndb_use_exact_count",
+	"ndb_use_transactions",
+	"ndbinfo_max_bytes",
+	"ndbinfo_max_rows",
+	"ndbinfo_show_hidden",
+	"ndbinfo_table_prefix",
+	"old_alter_table",
+	"preload_buffer_size",
+	"rbr_exec_mode",
+	"sql_log_off",
+	"transaction_write_set_extraction",
+	"audit_log_read_buffer_size",
+}
+
+var ignoreThese = []string{
+	"big_tables",
+	"bulk_insert_buffer_size",
+	"debug",
+	"default_storage_engine",
+	"default_tmp_storage_engine",
+	"innodb_strict_mode",
+	"innodb_support_xa",
+	"innodb_table_locks",
+	"innodb_tmpdir",
+	"join_buffer_size",
+	"keep_files_on_create",
+	"lc_messages",
+	"long_query_time",
+	"low_priority_updates",
+	"max_delayed_threads",
+	"max_insert_delayed_threads",
+	"multi_range_count",
+	"net_buffer_length",
+	"new",
+	"query_cache_type",
+	"query_cache_wlock_invalidate",
+	"query_prealloc_size",
+	"sql_buffer_result",
+	"transaction_alloc_block_size",
+	"wait_timeout",
+}
+
+var saveSettingsToSession = []string{
+	"sql_mode",
+	"sql_safe_updates",
+}
+
+var allowSetIfValueAlreadySet = []string{}
+
+var vitessShouldBeAwareOf = []string{
+	"block_encryption_mode",
+	"character_set_client",
+	"character_set_connection",
+	"character_set_database",
+	"character_set_filesystem",
+	"character_set_server",
+	"collation_connection",
+	"collation_database",
+	"collation_server",
+	"completion_type",
+	"div_precision_increment",
+	"innodb_lock_wait_timeout",
+	"interactive_timeout",
+	"lc_time_names",
+	"lock_wait_timeout",
+	"max_allowed_packet",
+	"max_error_count",
+	"max_execution_time",
+	"max_join_size",
+	"max_length_for_sort_data",
+	"max_sort_length",
+	"max_user_connections",
+	"session_track_gtids",
+	"session_track_schema",
+	"session_track_state_change",
+	"session_track_system_variables",
+	"session_track_transaction_info",
+	"time_zone",
+	"transaction_isolation",
+	"version_tokens_session",
+	"sql_auto_is_null",
+}
 
 func init() {
-	sysVarPlanningFunc["default_storage_engine"] = buildSetOpIgnore
-	sysVarPlanningFunc["sql_mode"] = buildSetOpCheckAndIgnore
-	sysVarPlanningFunc["sql_safe_updates"] = buildSetOpVarSet
+	forSettings(ignoreThese, buildSetOpIgnore)
+	forSettings(saveSettingsToSession, buildSetOpVarSet)
+	forSettings(allowSetIfValueAlreadySet, buildSetOpCheckAndIgnore)
+	forSettings(vitessShouldBeAwareOf, buildSetOpCheckAndIgnore)
+	forSettings(notSupported, buildNotSupported)
+}
+
+func forSettings(settings []string, f planFunc) {
+	for _, setting := range settings {
+		if _, alreadyExists := sysVarPlanningFunc[setting]; alreadyExists {
+			panic("bug in set plan init - " + setting + " aleady configured")
+		}
+		sysVarPlanningFunc[setting] = f
+	}
 }
 
 func buildSetPlan(stmt *sqlparser.Set, vschema ContextVSchema) (engine.Primitive, error) {
@@ -124,6 +244,10 @@ func planTabletInput(vschema ContextVSchema, tabletExpressions []*sqlparser.SetE
 	return primitive, nil
 }
 
+func buildNotSupported(e *sqlparser.SetExpr, _ ContextVSchema) (engine.SetOp, error) {
+	return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "%s: system setting is not supported", e.Name)
+}
+
 func buildSetOpIgnore(expr *sqlparser.SetExpr, _ ContextVSchema) (engine.SetOp, error) {
 	buf := sqlparser.NewTrackedBuffer(nil)
 	buf.Myprintf("%v", expr.Expr)
@@ -166,7 +290,7 @@ func expressionOkToDelegateToTablet(e sqlparser.Expr) bool {
 }
 
 func buildSetOpVarSet(expr *sqlparser.SetExpr, vschema ContextVSchema) (engine.SetOp, error) {
-	ks, dest, err := resolveDestination(vschema)
+	ks, err := vschema.AnyKeyspace()
 	if err != nil {
 		return nil, err
 	}
@@ -174,7 +298,7 @@ func buildSetOpVarSet(expr *sqlparser.SetExpr, vschema ContextVSchema) (engine.S
 	return &engine.SysVarSet{
 		Name:              expr.Name.Lowered(),
 		Keyspace:          ks,
-		TargetDestination: dest,
+		TargetDestination: vschema.Destination(),
 		Expr:              sqlparser.String(expr.Expr),
 	}, nil
 }
diff --git a/go/vt/vtgate/safe_session.go b/go/vt/vtgate/safe_session.go
index 69821a5b98..0d6aeae990 100644
--- a/go/vt/vtgate/safe_session.go
+++ b/go/vt/vtgate/safe_session.go
@@ -17,6 +17,7 @@ limitations under the License.
 package vtgate
 
 import (
+	"fmt"
 	"sync"
 
 	"github.com/golang/protobuf/proto"
@@ -113,7 +114,6 @@ func (session *SafeSession) Reset() {
 	session.ShardSessions = nil
 	session.PreSessions = nil
 	session.PostSessions = nil
-	session.Session.InReservedConn = false
 }
 
 // SetAutocommittable sets the state to autocommitable if true.
@@ -347,3 +347,16 @@ func (session *SafeSession) SetReservedConn(reservedConn bool) {
 	defer session.mu.Unlock()
 	session.Session.InReservedConn = reservedConn
 }
+
+//SetPreQueries returns the prequeries that need to be run when reserving a connection
+func (session *SafeSession) SetPreQueries() []string {
+	session.mu.Lock()
+	defer session.mu.Unlock()
+	result := make([]string, len(session.SystemVariables))
+	idx := 0
+	for k, v := range session.SystemVariables {
+		result[idx] = fmt.Sprintf("set @@%s = %s", k, v)
+		idx++
+	}
+	return result
+}
diff --git a/go/vt/vtgate/scatter_conn.go b/go/vt/vtgate/scatter_conn.go
index b01043871e..343e73122d 100644
--- a/go/vt/vtgate/scatter_conn.go
+++ b/go/vt/vtgate/scatter_conn.go
@@ -166,6 +166,7 @@ func (stc *ScatterConn) ExecuteMultiShard(
 		"Execute",
 		rss,
 		session,
+		autocommit,
 		func(rs *srvtopo.ResolvedShard, i int, info *shardActionInfo) (*shardActionInfo, error) {
 			var (
 				innerqr *sqltypes.Result
@@ -174,51 +175,49 @@ func (stc *ScatterConn) ExecuteMultiShard(
 				alias   *topodatapb.TabletAlias
 			)
 			transactionID := info.transactionID
-			reservedID := info.reserveID
+			reservedID := info.reservedID
 
 			if session != nil && session.Session != nil {
 				opts = session.Session.Options
 			}
 
-			switch {
-			case autocommit:
+			if autocommit {
 				// As this is auto-commit, the transactionID is supposed to be zero.
 				if info.transactionID != int64(0) {
 					return nil, vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "in autocommit mode, transactionID should be zero but was: %d", info.transactionID)
 				}
-				innerqr, err = rs.Gateway.Execute(ctx, rs.Target, queries[i].Sql, queries[i].BindVariables, 0, 0, opts)
-				if err != nil {
-					return nil, err
-				}
-			case nothing == info.actionNeeded:
+			}
+
+			switch info.actionNeeded {
+			case nothing:
 				qs, err := getQueryService(rs, info)
 				if err != nil {
 					return nil, err
 				}
-				innerqr, err = qs.Execute(ctx, rs.Target, queries[i].Sql, queries[i].BindVariables, info.transactionID, info.reserveID, opts)
+				innerqr, err = qs.Execute(ctx, rs.Target, queries[i].Sql, queries[i].BindVariables, info.transactionID, info.reservedID, opts)
 				if err != nil {
 					return nil, err
 				}
-			case begin == info.actionNeeded:
+			case begin:
 				qs, err := getQueryService(rs, info)
 				if err != nil {
 					return nil, err
 				}
-				innerqr, transactionID, alias, err = qs.BeginExecute(ctx, rs.Target, session.Savepoints, queries[i].Sql, queries[i].BindVariables, info.reserveID, opts)
+				innerqr, transactionID, alias, err = qs.BeginExecute(ctx, rs.Target, session.Savepoints, queries[i].Sql, queries[i].BindVariables, info.reservedID, opts)
 				if err != nil {
 					return info.updateTransactionID(transactionID, alias), err
 				}
-			case reserve == info.actionNeeded:
+			case reserve:
 				qs, err := getQueryService(rs, info)
 				if err != nil {
 					return nil, err
 				}
-				innerqr, reservedID, alias, err = qs.ReserveExecute(ctx, rs.Target, nil, queries[i].Sql, queries[i].BindVariables, info.transactionID, opts)
+				innerqr, reservedID, alias, err = qs.ReserveExecute(ctx, rs.Target, session.SetPreQueries(), queries[i].Sql, queries[i].BindVariables, info.transactionID, opts)
 				if err != nil {
 					return info.updateReservedID(reservedID, alias), err
 				}
-			case reserveBegin == info.actionNeeded:
-				innerqr, transactionID, reservedID, alias, err = rs.Gateway.ReserveBeginExecute(ctx, rs.Target, nil, queries[i].Sql, queries[i].BindVariables, opts)
+			case reserveBegin:
+				innerqr, transactionID, reservedID, alias, err = rs.Gateway.ReserveBeginExecute(ctx, rs.Target, session.SetPreQueries(), queries[i].Sql, queries[i].BindVariables, opts)
 				if err != nil {
 					return info.updateTransactionAndReservedID(transactionID, reservedID, alias), err
 				}
@@ -495,6 +494,7 @@ func (stc *ScatterConn) multiGoTransaction(
 	name string,
 	rss []*srvtopo.ResolvedShard,
 	session *SafeSession,
+	autocommit bool,
 	action shardActionTransactionFunc,
 ) (allErrors *concurrency.AllErrorRecorder) {
 
@@ -509,16 +509,16 @@ func (stc *ScatterConn) multiGoTransaction(
 		startTime, statsKey := stc.startAction(name, rs.Target)
 		defer stc.endAction(startTime, allErrors, statsKey, &err, session)
 
-		shardActionInfo := transactionInfo(rs.Target, session)
+		shardActionInfo := transactionInfo(rs.Target, session, autocommit)
 		updated, err := action(rs, i, shardActionInfo)
 		if updated == nil {
 			return
 		}
-		if updated.actionNeeded != nothing && (updated.transactionID != 0 || updated.reserveID != 0) {
+		if updated.actionNeeded != nothing && (updated.transactionID != 0 || updated.reservedID != 0) {
 			appendErr := session.AppendOrUpdate(&vtgatepb.Session_ShardSession{
 				Target:        rs.Target,
 				TransactionId: updated.transactionID,
-				ReservedId:    updated.reserveID,
+				ReservedId:    updated.reservedID,
 				TabletAlias:   updated.alias,
 			}, stc.txConn.mode)
 			if appendErr != nil {
@@ -551,7 +551,7 @@ func (stc *ScatterConn) multiGoTransaction(
 }
 
 // transactionInfo looks at the current session, and returns information about what needs to be done for this tablet
-func transactionInfo(target *querypb.Target, session *SafeSession) *shardActionInfo {
+func transactionInfo(target *querypb.Target, session *SafeSession, autocommit bool) *shardActionInfo {
 	if !(session.InTransaction() || session.InReservedConn()) {
 		return &shardActionInfo{}
 	}
@@ -559,10 +559,10 @@ func transactionInfo(target *querypb.Target, session *SafeSession) *shardActionI
 	// Find and AppendOrUpdate. The higher level functions ensure that no
 	// duplicate (target) tuples can execute
 	// this at the same time.
-	transactionID, reserveID, alias := session.Find(target.Keyspace, target.Shard, target.TabletType)
+	transactionID, reservedID, alias := session.Find(target.Keyspace, target.Shard, target.TabletType)
 
-	shouldReserve := session.InReservedConn() && reserveID == 0
-	shouldBegin := session.InTransaction() && transactionID == 0
+	shouldReserve := session.InReservedConn() && reservedID == 0
+	shouldBegin := session.InTransaction() && transactionID == 0 && !autocommit
 
 	var act = nothing
 	switch {
@@ -577,19 +577,19 @@ func transactionInfo(target *querypb.Target, session *SafeSession) *shardActionI
 	return &shardActionInfo{
 		actionNeeded:  act,
 		transactionID: transactionID,
-		reserveID:     reserveID,
+		reservedID:    reservedID,
 		alias:         alias,
 	}
 }
 
 type shardActionInfo struct {
-	actionNeeded             actionNeeded
-	reserveID, transactionID int64
-	alias                    *topodatapb.TabletAlias
+	actionNeeded              actionNeeded
+	reservedID, transactionID int64
+	alias                     *topodatapb.TabletAlias
 }
 
 func (sai *shardActionInfo) updateTransactionID(txID int64, alias *topodatapb.TabletAlias) *shardActionInfo {
-	return sai.updateTransactionAndReservedID(txID, sai.reserveID, alias)
+	return sai.updateTransactionAndReservedID(txID, sai.reservedID, alias)
 }
 
 func (sai *shardActionInfo) updateReservedID(rID int64, alias *topodatapb.TabletAlias) *shardActionInfo {
@@ -598,7 +598,7 @@ func (sai *shardActionInfo) updateReservedID(rID int64, alias *topodatapb.Tablet
 
 func (sai *shardActionInfo) updateTransactionAndReservedID(txID int64, rID int64, alias *topodatapb.TabletAlias) *shardActionInfo {
 	newInfo := *sai
-	newInfo.reserveID = rID
+	newInfo.reservedID = rID
 	newInfo.transactionID = txID
 	newInfo.alias = alias
 	return &newInfo
diff --git a/go/vt/vtgate/vcursor_impl.go b/go/vt/vtgate/vcursor_impl.go
index 344b58eb73..c90b3b913d 100644
--- a/go/vt/vtgate/vcursor_impl.go
+++ b/go/vt/vtgate/vcursor_impl.go
@@ -408,6 +408,25 @@ func (vc *vcursorImpl) NeedsReservedConn() {
 	vc.safeSession.SetReservedConn(true)
 }
 
+func (vc *vcursorImpl) InReservedConn() bool {
+	return vc.safeSession.InReservedConn()
+}
+
+func (vc *vcursorImpl) ShardSession() []*srvtopo.ResolvedShard {
+	ss := vc.safeSession.GetShardSessions()
+	if len(ss) == 0 {
+		return nil
+	}
+	rss := make([]*srvtopo.ResolvedShard, len(ss))
+	for i, shardSession := range ss {
+		rss[i] = &srvtopo.ResolvedShard{
+			Target:  shardSession.Target,
+			Gateway: vc.resolver.GetGateway(),
+		}
+	}
+	return rss
+}
+
 // Destination implements the ContextVSchema interface
 func (vc *vcursorImpl) Destination() key.Destination {
 	return vc.destination
diff --git a/go/vt/vttablet/sandboxconn/sandboxconn.go b/go/vt/vttablet/sandboxconn/sandboxconn.go
index 15a2b4d7df..2437e78255 100644
--- a/go/vt/vttablet/sandboxconn/sandboxconn.go
+++ b/go/vt/vttablet/sandboxconn/sandboxconn.go
@@ -491,6 +491,15 @@ func (sbc *SandboxConn) getNextResult() *sqltypes.Result {
 	return SingleRowResult
 }
 
+//StringQueries returns the queries executed as a slice of strings
+func (sbc *SandboxConn) StringQueries() []string {
+	result := make([]string, len(sbc.Queries))
+	for i, query := range sbc.Queries {
+		result[i] = query.Sql
+	}
+	return result
+}
+
 // SingleRowResult is returned when there is no pre-stored result.
 var SingleRowResult = &sqltypes.Result{
 	Fields: []*querypb.Field{
diff --git a/go/vt/vttablet/tabletserver/planbuilder/plan.go b/go/vt/vttablet/tabletserver/planbuilder/plan.go
index fda8de64ab..694ec49ae0 100644
--- a/go/vt/vttablet/tabletserver/planbuilder/plan.go
+++ b/go/vt/vttablet/tabletserver/planbuilder/plan.go
@@ -260,10 +260,22 @@ func BuildMessageStreaming(name string, tables map[string]*schema.Table) (*Plan,
 // a call to GET_LOCK(), which is unsafe with server-side connection pooling.
 // For more background, see https://github.com/vitessio/vitess/issues/3631.
 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 sqlparser.IsLockingFunc(f) {
-				return false, vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "%s() not allowed", f.Name.String())
+
+	genError := func(node sqlparser.SQLNode) error {
+		return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "%s not allowed without a reserved connections", sqlparser.String(node))
+	}
+
+	return sqlparser.Walk(func(in sqlparser.SQLNode) (kontinue bool, err error) {
+		switch node := in.(type) {
+		case *sqlparser.Set:
+			for _, setExpr := range node.Exprs {
+				if setExpr.Name.AtCount() > 0 {
+					return false, genError(node)
+				}
+			}
+		case *sqlparser.FuncExpr:
+			if sqlparser.IsLockingFunc(node) {
+				return false, genError(node)
 			}
 		}
 
diff --git a/go/vt/vttablet/tabletserver/stateful_connection.go b/go/vt/vttablet/tabletserver/stateful_connection.go
index df4f7f286c..17b353094a 100644
--- a/go/vt/vttablet/tabletserver/stateful_connection.go
+++ b/go/vt/vttablet/tabletserver/stateful_connection.go
@@ -96,6 +96,7 @@ func (sc *StatefulConnection) Exec(ctx context.Context, query string, maxrows in
 			default:
 				sc.env.CheckMySQL()
 			}
+			return nil, err
 		}
 		return nil, err
 	}


Test patch:
diff --git a/go/test/endtoend/vtgate/setstatement/main_test.go b/go/test/endtoend/vtgate/setstatement/main_test.go
index b8747055ab..8cf8d798ab 100644
--- a/go/test/endtoend/vtgate/setstatement/main_test.go
+++ b/go/test/endtoend/vtgate/setstatement/main_test.go
@@ -21,6 +21,8 @@ import (
 	"os"
 	"testing"
 
+	"github.com/stretchr/testify/require"
+
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/test/endtoend/cluster"
@@ -131,3 +133,10 @@ func exec(t *testing.T, conn *mysql.Conn, query string) (*sqltypes.Result, error
 	t.Helper()
 	return conn.ExecuteFetch(query, 1000, true)
 }
+
+func checkedExec(t *testing.T, conn *mysql.Conn, query string) *sqltypes.Result {
+	t.Helper()
+	qr, err := conn.ExecuteFetch(query, 1000, true)
+	require.NoError(t, err)
+	return qr
+}
diff --git a/go/test/endtoend/vtgate/setstatement/sysvar_test.go b/go/test/endtoend/vtgate/setstatement/sysvar_test.go
index 71ac50e4a3..afc0f7e7e8 100644
--- a/go/test/endtoend/vtgate/setstatement/sysvar_test.go
+++ b/go/test/endtoend/vtgate/setstatement/sysvar_test.go
@@ -39,16 +39,11 @@ func TestCharsetIntro(t *testing.T) {
 	require.NoError(t, err)
 	defer conn.Close()
 
-	_, err = exec(t, conn, "delete from test")
-	require.NoError(t, err)
-	_, err = exec(t, conn, "insert into test (id,val1) values (666, _binary'abc')")
-	require.NoError(t, err)
-	_, err = exec(t, conn, "update test set val1 = _latin1'xyz' where id = 666")
-	require.NoError(t, err)
-	_, err = exec(t, conn, "delete from test where val1 = _utf8'xyz'")
-	require.NoError(t, err)
-	qr, err := exec(t, conn, "select id from test where val1 = _utf8mb4'xyz'")
-	require.NoError(t, err)
+	checkedExec(t, conn, "delete from test")
+	checkedExec(t, conn, "insert into test (id,val1) values (666, _binary'abc')")
+	checkedExec(t, conn, "update test set val1 = _latin1'xyz' where id = 666")
+	checkedExec(t, conn, "delete from test where val1 = _utf8'xyz'")
+	qr := checkedExec(t, conn, "select id from test where val1 = _utf8mb4'xyz'")
 	require.EqualValues(t, 0, qr.RowsAffected)
 }
 
@@ -60,29 +55,29 @@ func TestSetSysVar(t *testing.T) {
 		Port: clusterInstance.VtgateMySQLPort,
 	}
 	type queriesWithExpectations struct {
-		query           string
-		expectedRows    string
-		rowsAffected    int
-		errMsg          string
-		expectedWarning string
+		name, expr, expected string
 	}
 
 	queries := []queriesWithExpectations{{
-		query:        `set @@default_storage_engine = INNODB`,
-		expectedRows: ``, rowsAffected: 0,
-		expectedWarning: "[[VARCHAR(\"Warning\") UINT16(1235) VARCHAR(\"Ignored inapplicable SET default_storage_engine = INNODB\")]]",
+		name:     "default_storage_engine",
+		expr:     "INNODB",
+		expected: `[[VARCHAR("InnoDB")]]`,
 	}, {
-		query:        `set @@sql_mode = @@sql_mode`,
-		expectedRows: ``, rowsAffected: 0,
+		name:     "sql_mode",
+		expr:     "''",
+		expected: `[[VARCHAR("")]]`,
 	}, {
-		query:        `set @@sql_mode = concat(@@sql_mode,"")`,
-		expectedRows: ``, rowsAffected: 0,
+		name:     "sql_mode",
+		expr:     `concat(@@sql_mode,"NO_ZERO_DATE")`,
+		expected: `[[VARCHAR("NO_ZERO_DATE")]]`,
 	}, {
-		query:           `set @@sql_mode = concat(@@sql_mode,"ALLOW_INVALID_DATES")`,
-		expectedWarning: "[[VARCHAR(\"Warning\") UINT16(1235) VARCHAR(\"Modification not allowed using set construct for: sql_mode\")]]",
+		name:     "sql_mode",
+		expr:     "@@sql_mode",
+		expected: `[[VARCHAR("NO_ZERO_DATE")]]`,
 	}, {
-		query:        `set @@SQL_SAFE_UPDATES = 1`,
-		expectedRows: ``, rowsAffected: 0,
+		name:     "SQL_SAFE_UPDATES",
+		expr:     "1",
+		expected: "[[INT64(1)]]",
 	}}
 
 	conn, err := mysql.Connect(ctx, &vtParams)
@@ -90,27 +85,139 @@ func TestSetSysVar(t *testing.T) {
 	defer conn.Close()
 
 	for i, q := range queries {
-		t.Run(fmt.Sprintf("%d-%s", i, q.query), func(t *testing.T) {
-			qr, err := exec(t, conn, q.query)
-			if q.errMsg != "" {
-				require.Contains(t, err.Error(), q.errMsg)
-			} else {
-				require.NoError(t, err)
-				require.Equal(t, uint64(q.rowsAffected), qr.RowsAffected, "rows affected wrong for query: %s", q.query)
-				if q.expectedRows != "" {
-					result := fmt.Sprintf("%v", qr.Rows)
-					if diff := cmp.Diff(q.expectedRows, result); diff != "" {
-						t.Errorf("%s\nfor query: %s", diff, q.query)
-					}
-				}
-				if q.expectedWarning != "" {
-					qr, err := exec(t, conn, "show warnings")
-					require.NoError(t, err)
-					if got, want := fmt.Sprintf("%v", qr.Rows), q.expectedWarning; got != want {
-						t.Errorf("select:\n%v want\n%v", got, want)
-					}
-				}
-			}
+		query := fmt.Sprintf("set %s = %s", q.name, q.expr)
+		t.Run(fmt.Sprintf("%d-%s", i, query), func(t *testing.T) {
+			_, err := exec(t, conn, query)
+			require.NoError(t, err)
+			assertMatches(t, conn, fmt.Sprintf("select @@%s", q.name), q.expected)
 		})
 	}
 }
+
+func TestSetSystemVariable(t *testing.T) {
+	vtParams := mysql.ConnParams{
+		Host: "localhost",
+		Port: clusterInstance.VtgateMySQLPort,
+	}
+	conn, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	checkedExec(t, conn, "set @@sql_mode = 'NO_ZERO_DATE'")
+	q := `select str_to_date('00/00/0000', '%m/%d/%Y')`
+	assertMatches(t, conn, q, `[[NULL]]`)
+
+	assertMatches(t, conn, "select @@sql_mode", `[[VARCHAR("NO_ZERO_DATE")]]`)
+	checkedExec(t, conn, "set @@sql_mode = ''")
+
+	assertMatches(t, conn, q, `[[DATE("0000-00-00")]]`)
+
+	checkedExec(t, conn, "SET @@SESSION.sql_mode = CONCAT(CONCAT(@@sql_mode, ',STRICT_ALL_TABLES'), ',NO_AUTO_VALUE_ON_ZERO'),  @@SESSION.sql_auto_is_null = 0, @@SESSION.wait_timeout = 2147483")
+	assertMatches(t, conn, "select @@sql_mode", `[[VARCHAR("NO_AUTO_VALUE_ON_ZERO,STRICT_ALL_TABLES")]]`)
+}
+
+func TestSetSystemVarWithTxFailure(t *testing.T) {
+	vtParams := mysql.ConnParams{
+		Host: "localhost",
+		Port: clusterInstance.VtgateMySQLPort,
+	}
+
+	conn, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	checkedExec(t, conn, "insert into test (id, val1) values (80, null)")
+
+	// before changing any settings, let's confirm sql_safe_updates value
+	assertMatches(t, conn, `select @@sql_safe_updates from test where id = 80`, `[[INT64(0)]]`)
+
+	checkedExec(t, conn, "set sql_safe_updates = 1")
+	checkedExec(t, conn, "begin")
+
+	qr := checkedExec(t, conn, "select connection_id() from test where id = 80")
+
+	// kill the mysql connection shard which has transaction open.
+	vttablet1 := clusterInstance.Keyspaces[0].Shards[0].MasterTablet() // 80-
+	vttablet1.VttabletProcess.QueryTablet(fmt.Sprintf("kill %s", qr.Rows[0][0].ToString()), keyspaceName, false)
+
+	// transaction fails on commit - we should no longer be in a transaction
+	_, err = conn.ExecuteFetch("commit", 1, true)
+	require.Error(t, err)
+
+	// we still want to have our system setting applied
+	assertMatches(t, conn, `select @@sql_safe_updates`, `[[INT64(1)]]`)
+}
+
+func TestSetSystemVarWithConnectionFailure(t *testing.T) {
+	t.Skip("failing at the moment")
+	vtParams := mysql.ConnParams{
+		Host: "localhost",
+		Port: clusterInstance.VtgateMySQLPort,
+	}
+
+	conn, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn.Close()
+	checkedExec(t, conn, "delete from test")
+
+	checkedExec(t, conn, "insert into test (id, val1) values (80, null)")
+	checkedExec(t, conn, "set sql_safe_updates = 1")
+	qr := checkedExec(t, conn, "select connection_id() from test where id = 80")
+
+	// kill the mysql connection shard which has transaction open.
+	vttablet1 := clusterInstance.Keyspaces[0].Shards[0].MasterTablet() // 80-
+	vttablet1.VttabletProcess.QueryTablet(fmt.Sprintf("kill %s", qr.Rows[0][0].ToString()), keyspaceName, false)
+
+	// we still want to have our system setting applied
+	_, err = exec(t, conn, `select @@sql_safe_updates from test where id = 80`)
+	require.NoError(t, err)
+}
+
+func TestSetSystemVariableAndThenSuccessfulTx(t *testing.T) {
+	vtParams := mysql.ConnParams{
+		Host: "localhost",
+		Port: clusterInstance.VtgateMySQLPort,
+	}
+
+	conn, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn.Close()
+	checkedExec(t, conn, "delete from test")
+
+	checkedExec(t, conn, "set sql_safe_updates = 1")
+	checkedExec(t, conn, "begin")
+	checkedExec(t, conn, "insert into test (id, val1) values (80, null)")
+	checkedExec(t, conn, "commit")
+	assertMatches(t, conn, "select id, val1 from test", "[[INT64(80) NULL]]")
+	assertMatches(t, conn, "select @@sql_safe_updates", "[[INT64(1)]]")
+}
+
+func TestStartTxAndSetSystemVariableAndThenSuccessfulCommit(t *testing.T) {
+	vtParams := mysql.ConnParams{
+		Host: "localhost",
+		Port: clusterInstance.VtgateMySQLPort,
+	}
+
+	conn, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn.Close()
+	checkedExec(t, conn, "delete from test")
+
+	checkedExec(t, conn, "begin")
+	checkedExec(t, conn, "set sql_safe_updates = 1")
+	checkedExec(t, conn, "insert into test (id, val1) values (54, null)")
+	checkedExec(t, conn, "commit")
+	assertMatches(t, conn, "select id, val1 from test", "[[INT64(54) NULL]]")
+	assertMatches(t, conn, "select @@sql_safe_updates", "[[INT64(1)]]")
+}
+
+func assertMatches(t *testing.T, conn *mysql.Conn, query, expected string) {
+	t.Helper()
+	qr, err := exec(t, conn, query)
+	require.NoError(t, err)
+	got := fmt.Sprintf("%v", qr.Rows)
+	diff := cmp.Diff(expected, got)
+	if diff != "" {
+		t.Errorf("Query: %s (-want +got):\n%s", query, diff)
+	}
+}
diff --git a/go/vt/vtgate/engine/fake_vcursor_test.go b/go/vt/vtgate/engine/fake_vcursor_test.go
index 1a07f32691..9c886354fb 100644
--- a/go/vt/vtgate/engine/fake_vcursor_test.go
+++ b/go/vt/vtgate/engine/fake_vcursor_test.go
@@ -65,6 +65,14 @@ func (t noopVCursor) SetSysVar(name string, expr string) {
 	//panic("implement me")
 }
 
+func (t noopVCursor) InReservedConn() bool {
+	panic("implement me")
+}
+
+func (t noopVCursor) ShardSession() []*srvtopo.ResolvedShard {
+	panic("implement me")
+}
+
 func (t noopVCursor) ExecuteVSchema(keyspace string, vschemaDDL *sqlparser.DDL) error {
 	panic("implement me")
 }
@@ -76,6 +84,7 @@ func (t noopVCursor) Session() SessionActions {
 func (t noopVCursor) SetTarget(target string) error {
 	panic("implement me")
 }
+
 func (t noopVCursor) Context() context.Context {
 	if t.ctx == nil {
 		return context.Background()
@@ -132,6 +141,7 @@ func (t noopVCursor) ResolveDestinations(keyspace string, ids []*querypb.Value,
 }
 
 var _ VCursor = (*loggingVCursor)(nil)
+
 var _ SessionActions = (*loggingVCursor)(nil)
 
 // loggingVCursor logs requests and allows you to verify
@@ -171,6 +181,14 @@ func (f *loggingVCursor) SetSysVar(name string, expr string) {
 func (f *loggingVCursor) NeedsReservedConn() {
 }
 
+func (f *loggingVCursor) InReservedConn() bool {
+	panic("implement me")
+}
+
+func (f *loggingVCursor) ShardSession() []*srvtopo.ResolvedShard {
+	return nil
+}
+
 func (f *loggingVCursor) ExecuteVSchema(keyspace string, vschemaDDL *sqlparser.DDL) error {
 	panic("implement me")
 }
diff --git a/go/vt/vtgate/engine/set_test.go b/go/vt/vtgate/engine/set_test.go
index 34a949edb6..da4b7ae3ec 100644
--- a/go/vt/vtgate/engine/set_test.go
+++ b/go/vt/vtgate/engine/set_test.go
@@ -31,6 +31,40 @@ import (
 	querypb "vitess.io/vitess/go/vt/proto/query"
 )
 
+func TestSetSystemVariableAsString(t *testing.T) {
+	setOp := SysVarSet{
+		Name: "x",
+		Keyspace: &vindexes.Keyspace{
+			Name:    "ks",
+			Sharded: true,
+		},
+		Expr: "dummy_expr",
+	}
+
+	set := &Set{
+		Ops:   []SetOp{&setOp},
+		Input: &SingleRow{},
+	}
+	vc := &loggingVCursor{
+		shards: []string{"-20", "20-"},
+		results: []*sqltypes.Result{sqltypes.MakeTestResult(
+			sqltypes.MakeTestFields(
+				"id",
+				"varchar",
+			),
+			"foobar",
+		)},
+	}
+	_, err := set.Execute(vc, map[string]*querypb.BindVariable{}, false)
+	require.NoError(t, err)
+
+	vc.ExpectLog(t, []string{
+		"ResolveDestinations ks [] Destinations:DestinationKeyspaceID(00)",
+		"ExecuteMultiShard ks.-20: select dummy_expr from dual where @@x != dummy_expr {} false false",
+		"SysVar set with (x,'foobar')",
+	})
+}
+
 func TestSetTable(t *testing.T) {
 	type testCase struct {
 		testName         string
@@ -72,9 +106,6 @@ func TestSetTable(t *testing.T) {
 					Expr: "42",
 				},
 			},
-			expectedWarning: []*querypb.QueryWarning{
-				{Code: 1235, Message: "Ignored inapplicable SET x = 42"},
-			},
 		},
 		{
 			testName: "sysvar check and ignore",
@@ -100,9 +131,6 @@ func TestSetTable(t *testing.T) {
 				`ResolveDestinations ks [] Destinations:DestinationAnyShard()`,
 				`ExecuteMultiShard ks.-20: select 1 from dual where @@x = dummy_expr {} false false`,
 			},
-			expectedWarning: []*querypb.QueryWarning{
-				{Code: 1235, Message: "Ignored inapplicable SET x = dummy_expr"},
-			},
 		},
 		{
 			testName: "sysvar check and error",
@@ -121,9 +149,6 @@ func TestSetTable(t *testing.T) {
 				`ResolveDestinations ks [] Destinations:DestinationAnyShard()`,
 				`ExecuteMultiShard ks.-20: select 1 from dual where @@x = dummy_expr {} false false`,
 			},
-			expectedWarning: []*querypb.QueryWarning{
-				{Code: 1235, Message: "Modification not allowed using set construct for: x"},
-			},
 		},
 		{
 			testName: "sysvar checkAndIgnore multi destination error",
@@ -144,7 +169,7 @@ func TestSetTable(t *testing.T) {
 			expectedError: "Unexpected error, DestinationKeyspaceID mapping to multiple shards: DestinationAllShards()",
 		},
 		{
-			testName: "udv_ignr_chignr",
+			testName: "udv ignore checkAndIgnore ",
 			setOps: []SetOp{
 				&UserDefinedVariable{
 					Name: "x",
@@ -169,10 +194,6 @@ func TestSetTable(t *testing.T) {
 				`ResolveDestinations ks [] Destinations:DestinationAnyShard()`,
 				`ExecuteMultiShard ks.-20: select 1 from dual where @@z = dummy_expr {} false false`,
 			},
-			expectedWarning: []*querypb.QueryWarning{
-				{Code: 1235, Message: "Ignored inapplicable SET y = 2"},
-				{Code: 1235, Message: "Ignored inapplicable SET z = dummy_expr"},
-			},
 			qr: []*sqltypes.Result{sqltypes.MakeTestResult(
 				sqltypes.MakeTestFields(
 					"id",
@@ -182,7 +203,7 @@ func TestSetTable(t *testing.T) {
 			)},
 		},
 		{
-			testName: "sysvar set",
+			testName: "sysvar set without destination",
 			setOps: []SetOp{
 				&SysVarSet{
 					Name: "x",
@@ -196,10 +217,51 @@ func TestSetTable(t *testing.T) {
 			},
 			expectedQueryLog: []string{
 				`ResolveDestinations ks [] Destinations:DestinationAnyShard()`,
-				`ExecuteMultiShard ks.-20: select dummy_expr from dual where false {} false false`,
-				`SysVar set with (x,dummy_expr)`,
+				`ExecuteMultiShard ks.-20: set @@x = dummy_expr {} false false`,
 			},
 		},
+		{
+			testName: "sysvar set not modifying setting",
+			setOps: []SetOp{
+				&SysVarSet{
+					Name: "x",
+					Keyspace: &vindexes.Keyspace{
+						Name:    "ks",
+						Sharded: true,
+					},
+					Expr: "dummy_expr",
+				},
+			},
+			expectedQueryLog: []string{
+				`ResolveDestinations ks [] Destinations:DestinationKeyspaceID(00)`,
+				`ExecuteMultiShard ks.-20: select dummy_expr from dual where @@x != dummy_expr {} false false`,
+			},
+		},
+		{
+			testName: "sysvar set modifying setting",
+			setOps: []SetOp{
+				&SysVarSet{
+					Name: "x",
+					Keyspace: &vindexes.Keyspace{
+						Name:    "ks",
+						Sharded: true,
+					},
+					Expr: "dummy_expr",
+				},
+			},
+			expectedQueryLog: []string{
+				`ResolveDestinations ks [] Destinations:DestinationKeyspaceID(00)`,
+				`ExecuteMultiShard ks.-20: select dummy_expr from dual where @@x != dummy_expr {} false false`,
+				`SysVar set with (x,123456)`,
+			},
+			qr: []*sqltypes.Result{sqltypes.MakeTestResult(
+				sqltypes.MakeTestFields(
+					"id",
+					"int64",
+				),
+				"123456",
+			)},
+		},
 	}
 
 	for _, tc := range tests {
@@ -226,7 +288,6 @@ func TestSetTable(t *testing.T) {
 }
 
 func TestSysVarSetErr(t *testing.T) {
-
 	setOps := []SetOp{
 		&SysVarSet{
 			Name: "x",
@@ -241,7 +302,7 @@ func TestSysVarSetErr(t *testing.T) {
 
 	expectedQueryLog := []string{
 		`ResolveDestinations ks [] Destinations:DestinationAnyShard()`,
-		`ExecuteMultiShard ks.-20: select dummy_expr from dual where false {} false false`,
+		`ExecuteMultiShard ks.-20: set @@x = dummy_expr {} false false`,
 	}
 
 	set := &Set{
diff --git a/go/vt/vtgate/executor_set_test.go b/go/vt/vtgate/executor_set_test.go
index e56678fda6..8e509e5019 100644
--- a/go/vt/vtgate/executor_set_test.go
+++ b/go/vt/vtgate/executor_set_test.go
@@ -19,7 +19,6 @@ package vtgate
 import (
 	"testing"
 
-	"vitess.io/vitess/go/mysql"
 	querypb "vitess.io/vitess/go/vt/proto/query"
 
 	"vitess.io/vitess/go/test/utils"
@@ -39,12 +38,13 @@ import (
 )
 
 func TestExecutorSet(t *testing.T) {
-	executor, _, _, _ := createLegacyExecutorEnv()
+	executorEnv, _, _, _ := createExecutorEnv()
 
 	testcases := []struct {
-		in  string
-		out *vtgatepb.Session
-		err string
+		in     string
+		out    *vtgatepb.Session
+		err    string
+		target string
 	}{{
 		in:  "set autocommit = 1",
 		out: &vtgatepb.Session{Autocommit: true},
@@ -238,11 +238,13 @@ func TestExecutorSet(t *testing.T) {
 		in:  "set skip_query_plan_cache = 0",
 		out: &vtgatepb.Session{Autocommit: true, Options: &querypb.ExecuteOptions{}},
 	}, {
-		in:  "set sql_auto_is_null = 0",
-		out: &vtgatepb.Session{Autocommit: true}, // no effect
+		in:     "set sql_auto_is_null = 0",
+		target: "TestExecutor",
+		out:    &vtgatepb.Session{Autocommit: true, TargetString: "TestExecutor"},
 	}, {
-		in:  "set sql_auto_is_null = 1",
-		err: "sql_auto_is_null is not currently supported",
+		in:     "set sql_auto_is_null = 1",
+		target: "TestExecutor",
+		out:    &vtgatepb.Session{Autocommit: true, TargetString: "TestExecutor"},
 	}, {
 		in:  "set tx_read_only = 2",
 		err: "unexpected value for tx_read_only: 2",
@@ -288,37 +290,40 @@ func TestExecutorSet(t *testing.T) {
 	}}
 	for _, tcase := range testcases {
 		t.Run(tcase.in, func(t *testing.T) {
-			session := NewSafeSession(&vtgatepb.Session{Autocommit: true})
-			_, err := executor.Execute(context.Background(), "TestExecute", session, tcase.in, nil)
-			if err != nil {
-				require.EqualError(t, err, tcase.err)
+			session := NewSafeSession(&vtgatepb.Session{Autocommit: true, TargetString: tcase.target})
+			_, err := executorEnv.Execute(context.Background(), "TestExecute", session, tcase.in, nil)
+			if tcase.err == "" {
+				utils.MustMatch(t, tcase.out, session.Session, "new executor")
 			} else {
-				utils.MustMatch(t, tcase.out, session.Session, "session output was not as expected")
+				require.EqualError(t, err, tcase.err)
 			}
 		})
 	}
 }
 
 func TestExecutorSetOp(t *testing.T) {
-	executor, sbc1, _, _ := createLegacyExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
+
+	sbclookup.SetResults([]*sqltypes.Result{
+		sqltypes.MakeTestResult(sqltypes.MakeTestFields("'STRICT_ALL_TABLES,NO_AUTO_UPDATES'", "varchar"), "STRICT_ALL_TABLES,NO_AUTO_UPDATES"),
+		sqltypes.MakeTestResult(sqltypes.MakeTestFields("1", "int64"), "1"),
+	})
 
 	testcases := []struct {
 		in      string
 		warning []*querypb.QueryWarning
 		sysVars map[string]string
 	}{{
-		in: "set sql_mode = 'STRICT_ALL_TABLES'",
-		warning: []*querypb.QueryWarning{{
-			Code:    mysql.ERNotSupportedYet,
-			Message: "Ignored inapplicable SET sql_mode = 'STRICT_ALL_TABLES'",
-		}},
-	}, {
-		in:      "set sql_safe_updates = 2",
-		sysVars: map[string]string{"sql_safe_updates": "2"},
+		in: "set big_tables = 1",
+	}, {
+		in:      "set sql_mode = 'STRICT_ALL_TABLES,NO_AUTO_UPDATES'",
+		sysVars: map[string]string{"sql_mode": "'STRICT_ALL_TABLES,NO_AUTO_UPDATES'"},
+	}, {
+		in:      "set sql_safe_updates = 1",
+		sysVars: map[string]string{"sql_safe_updates": "1"},
 	}}
 	for _, tcase := range testcases {
 		t.Run(tcase.in, func(t *testing.T) {
-			sbc1.SetResults([]*sqltypes.Result{{}})
 			session := NewAutocommitSession(masterSession)
 			session.TargetString = KsTestUnsharded
 			_, err := executor.Execute(
diff --git a/go/vt/vtgate/executor_test.go b/go/vt/vtgate/executor_test.go
index e713ec6965..7deb4e1534 100644
--- a/go/vt/vtgate/executor_test.go
+++ b/go/vt/vtgate/executor_test.go
@@ -80,7 +80,7 @@ func TestExecutorMaxMemoryRowsExceeded(t *testing.T) {
 	*maxMemoryRows = 3
 	defer func() { *maxMemoryRows = save }()
 
-	executor, _, _, sbclookup := createExecutorEnv()
+	executor, _, _, sbclookup := createLegacyExecutorEnv()
 	session := NewSafeSession(&vtgatepb.Session{TargetString: "@master"})
 	result := sqltypes.MakeTestResult(sqltypes.MakeTestFields("col", "int64"), "1", "2", "3", "4")
 	target := querypb.Target{}
diff --git a/go/vt/vtgate/legacy_scatter_conn_test.go b/go/vt/vtgate/legacy_scatter_conn_test.go
index 790e6ffa3a..09bcfbfb96 100644
--- a/go/vt/vtgate/legacy_scatter_conn_test.go
+++ b/go/vt/vtgate/legacy_scatter_conn_test.go
@@ -580,6 +580,34 @@ func TestAppendResult(t *testing.T) {
 	}
 }
 
+func TestReservePrequeries(t *testing.T) {
+	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")
+
+	session := NewSafeSession(&vtgatepb.Session{
+		InTransaction:  false,
+		InReservedConn: true,
+		SystemVariables: map[string]string{
+			"s1": "'value'",
+			"s2": "42",
+		},
+	})
+	destinations := []key.Destination{key.DestinationShard("0")}
+
+	executeOnShards(t, res, keyspace, sc, session, destinations)
+	assert.Equal(t, 2+1, len(sbc0.StringQueries()))
+}
+
 func newTestLegacyScatterConn(hc discovery.LegacyHealthCheck, serv srvtopo.Server, cell string) *ScatterConn {
 	// The topo.Server is used to start watching the cells described
 	// in '-cells_to_watch' command line parameter, which is
diff --git a/go/vt/vtgate/planbuilder/testdata/set_cases.txt b/go/vt/vtgate/planbuilder/testdata/set_cases.txt
index 934dd769b1..4b2372b5bc 100644
--- a/go/vt/vtgate/planbuilder/testdata/set_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/set_cases.txt
@@ -107,3 +107,64 @@
 # only allow whitelisted functions
 "set @foo = BAD_FUNC()"
 "expression not supported for SET: BAD_FUNC()"
+
+# single sysvar cases
+"SET sql_mode = 'STRICT_ALL_TABLES,NO_AUTO_VALUE_ON_ZERO'"
+{
+  "QueryType": "SET",
+  "Original": "SET sql_mode = 'STRICT_ALL_TABLES,NO_AUTO_VALUE_ON_ZERO'",
+  "Instructions": {
+    "OperatorType": "Set",
+    "Ops": [
+      {
+        "Type": "SysVarSet",
+        "Name": "sql_mode",
+        "Keyspace": {
+          "Name": "main",
+          "Sharded": false
+        },
+        "Expr": "'STRICT_ALL_TABLES,NO_AUTO_VALUE_ON_ZERO'"
+      }
+    ],
+    "Inputs": [
+      {
+        "OperatorType": "SingleRow"
+      }
+    ]
+  }
+}
+
+# multiple sysvar cases
+"SET @@SESSION.sql_mode = CONCAT(CONCAT(@@sql_mode, ',STRICT_ALL_TABLES'), ',NO_AUTO_VALUE_ON_ZERO'), @@SESSION.sql_safe_updates = 0"
+{
+  "QueryType": "SET",
+  "Original": "SET @@SESSION.sql_mode = CONCAT(CONCAT(@@sql_mode, ',STRICT_ALL_TABLES'), ',NO_AUTO_VALUE_ON_ZERO'), @@SESSION.sql_safe_updates = 0",
+  "Instructions": {
+    "OperatorType": "Set",
+    "Ops": [
+      {
+        "Type": "SysVarSet",
+        "Name": "sql_mode",
+        "Keyspace": {
+          "Name": "main",
+          "Sharded": false
+        },
+        "Expr": "CONCAT(CONCAT(@@sql_mode, ',STRICT_ALL_TABLES'), ',NO_AUTO_VALUE_ON_ZERO')"
+      },
+      {
+        "Type": "SysVarSet",
+        "Name": "sql_safe_updates",
+        "Keyspace": {
+          "Name": "main",
+          "Sharded": false
+        },
+        "Expr": "0"
+      }
+    ],
+    "Inputs": [
+      {
+        "OperatorType": "SingleRow"
+      }
+    ]
+  }
+}
diff --git a/go/vt/vtgate/planbuilder/testdata/set_sysvar_cases.txt b/go/vt/vtgate/planbuilder/testdata/set_sysvar_cases.txt
index 857bb6fa18..a2aa4ddcc7 100644
--- a/go/vt/vtgate/planbuilder/testdata/set_sysvar_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/set_sysvar_cases.txt
@@ -29,13 +29,12 @@
     "OperatorType": "Set",
     "Ops": [
       {
-        "Type": "SysVarCheckAndIgnore",
+        "Type": "SysVarSet",
         "Name": "sql_mode",
         "Keyspace": {
           "Name": "main",
           "Sharded": false
         },
-        "TargetDestination": {},
         "Expr": "concat(@@sql_mode, ',NO_AUTO_CREATE_USER')"
       }
     ],
@@ -47,7 +46,7 @@
   }
 }
 
-# set check and ignore plan
+# set system settings
 "set @@sql_safe_updates = 1"
 {
   "QueryType": "SET",
@@ -62,7 +61,6 @@
           "Name": "main",
           "Sharded": false
         },
-        "TargetDestination": {},
         "Expr": "1"
       }
     ],
@@ -76,4 +74,26 @@
 
 # set plan building not supported
 "set @@innodb_strict_mode = OFF"
-"plan building not supported"
+{
+  "QueryType": "SET",
+  "Original": "set @@innodb_strict_mode = OFF",
+  "Instructions": {
+    "OperatorType": "Set",
+    "Ops": [
+      {
+        "Type": "SysVarIgnore",
+        "Name": "innodb_strict_mode",
+        "Expr": "'off'"
+      }
+    ],
+    "Inputs": [
+      {
+        "OperatorType": "SingleRow"
+      }
+    ]
+  }
+}
+
+# set system settings
+"set @@ndbinfo_max_bytes = 192"
+"ndbinfo_max_bytes: system setting is not supported"
diff --git a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
index 2d3cabdded..bae077d6c2 100644
--- a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.txt
@@ -7,7 +7,7 @@
 "plan building not supported"
 
 # set user defined and system variable
-"set @foo = 42, @bar = @foo, @@wait_timeout = 28800"
+"set @foo = 42, @bar = @foo, @@xyz = 28800"
 "plan building not supported"
 
 # SHOW
diff --git a/go/vt/vtgate/safe_session_test.go b/go/vt/vtgate/safe_session_test.go
index 66131132fc..37ee0972f3 100644
--- a/go/vt/vtgate/safe_session_test.go
+++ b/go/vt/vtgate/safe_session_test.go
@@ -17,6 +17,7 @@ limitations under the License.
 package vtgate
 
 import (
+	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -46,3 +47,22 @@ func TestFailToMultiShardWhenSetToSingleDb(t *testing.T) {
 	err = session.AppendOrUpdate(sess1, vtgatepb.TransactionMode_SINGLE)
 	require.Error(t, err)
 }
+
+func TestPrequeries(t *testing.T) {
+	session := NewSafeSession(&vtgatepb.Session{
+		SystemVariables: map[string]string{
+			"s1": "'apa'",
+			"s2": "42",
+		},
+	})
+
+	q1 := "set @@s1 = 'apa'"
+	q2 := "set @@s2 = 42"
+	want := []string{q1, q2}
+	wantReversed := []string{q2, q1}
+	preQueries := session.SetPreQueries()
+
+	if !reflect.DeepEqual(want, preQueries) && !reflect.DeepEqual(wantReversed, preQueries) {
+		t.Errorf("got %v but wanted %v", preQueries, want)
+	}
+}
diff --git a/go/vt/vtgate/tx_conn_test.go b/go/vt/vtgate/tx_conn_test.go
index 8bfe0c4ffb..f1edd17aa7 100644
--- a/go/vt/vtgate/tx_conn_test.go
+++ b/go/vt/vtgate/tx_conn_test.go
@@ -187,7 +187,7 @@ func TestTxConnReservedCommitSuccess(t *testing.T) {
 
 	require.NoError(t,
 		sc.txConn.Release(ctx, session))
-	wantSession = vtgatepb.Session{}
+	wantSession = vtgatepb.Session{InReservedConn: true}
 	utils.MustMatch(t, &wantSession, session.Session, "Session")
 	assert.EqualValues(t, 1, sbc0.ReleaseCount.Get(), "sbc0.ReleaseCount")
 	assert.EqualValues(t, 1, sbc1.ReleaseCount.Get(), "sbc1.ReleaseCount")
@@ -708,7 +708,7 @@ func TestTxConnReservedCommitOrderSuccess(t *testing.T) {
 
 	require.NoError(t,
 		sc.txConn.Release(ctx, session))
-	wantSession = vtgatepb.Session{}
+	wantSession = vtgatepb.Session{InReservedConn: true}
 	utils.MustMatch(t, &wantSession, session.Session, "Session")
 	assert.EqualValues(t, 2, sbc0.ReleaseCount.Get(), "sbc0.ReleaseCount")
 	assert.EqualValues(t, 1, sbc1.ReleaseCount.Get(), "sbc1.ReleaseCount")
@@ -902,6 +902,7 @@ func TestTxConnReservedRollbackFailure(t *testing.T) {
 	require.Error(t,
 		sc.txConn.Rollback(ctx, session))
 	wantSession := vtgatepb.Session{
+		InReservedConn: true,
 		Warnings: []*querypb.QueryWarning{{
 			Message: "rollback encountered an error and connection to all shard for this session is released: Code: INVALID_ARGUMENT\nINVALID_ARGUMENT error\n\ntarget: TxConnReservedRollback.1.master, used tablet: aa-0 (1)",
 		}},
diff --git a/go/vt/vttablet/tabletserver/planbuilder/testdata/pool_unsafe_cases.txt b/go/vt/vttablet/tabletserver/planbuilder/testdata/pool_unsafe_cases.txt
index e7e6b9d3e4..2691c4928b 100644
--- a/go/vt/vttablet/tabletserver/planbuilder/testdata/pool_unsafe_cases.txt
+++ b/go/vt/vttablet/tabletserver/planbuilder/testdata/pool_unsafe_cases.txt
@@ -1,19 +1,23 @@
 # get_lock named locks are unsafe with server-side connection pooling
 "select get_lock('foo', 10) from dual"
-"get_lock() not allowed"
+"get_lock('foo', 10) not allowed without a reserved connections"
 
 # 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_free_lock('foo') not allowed without a reserved connections"
 
 # is_used_lock named locks are unsafe with server-side connection pooling
 "select is_used_lock('foo') from dual"
-"is_used_lock() not allowed"
+"is_used_lock('foo') not allowed without a reserved connections"
 
 # release_all_locks named locks are unsafe with server-side connection pooling
 "select release_all_locks() from dual"
-"release_all_locks() not allowed"
+"release_all_locks() not allowed without a reserved connections"
 
 # release_lock named locks are unsafe with server-side connection pooling
 "select release_lock('foo') from dual"
-"release_lock() not allowed"
+"release_lock('foo') not allowed without a reserved connections"
+
+# setting system variables must happen inside reserved connections
+"set @sql_safe_updates = false"
+"set @sql_safe_updates = false not allowed without a reserved connections"
diff --git a/go/vt/vttablet/tabletserver/planbuilder/testdata/stream_cases.txt b/go/vt/vttablet/tabletserver/planbuilder/testdata/stream_cases.txt
index 1a5cce0146..28ed3798ac 100644
--- a/go/vt/vttablet/tabletserver/planbuilder/testdata/stream_cases.txt
+++ b/go/vt/vttablet/tabletserver/planbuilder/testdata/stream_cases.txt
@@ -55,4 +55,4 @@
 
 # named locks are unsafe with server-side connection pooling
 "select get_lock('foo') from dual"
-"get_lock() not allowed"
+"get_lock('foo') not allowed without a reserved connections"