Repository: vitessio/vitess
Pull request #6787 (2020-09-25 10:03:34)

vrepl metrics: add metrics around vstreamer and vreplication metrics
Signed-off-by: Rohit Nayak rohit@planetscale.com


Patch:
diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
index 314ce18532..5c74a5ef21 100644
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@@ -7,14 +7,10 @@
 /go/test/endtoend/vtgate @harshit-gangal @systay
 /go/vt/vtgate @harshit-gangal @systay
 /go/vt/vttablet/tabletserver @harshit-gangal @systay
-/go/vt/vttablet/tabletmanager/vstreamer @rohit-nayak-ps
 /go/vt/sqlparser @harshit-gangal @systay
 /go/vt/discovery @deepthi
 /go/vt/vttablet/tabletmanager @deepthi
-/go/vt/vttablet/tabletmanager/vreplication @rohit-nayak-ps
 /go/vt/orchestrator @deepthi
 /go/vt/vtctl @deepthi
-/go/vt/wrangler @deepthi @rohit-nayak-ps
+/go/vt/wrangler @deepthi
 /go/vt/mysqlctl @deepthi
-/web/vtctld2 @rohit-nayak-ps
-/examples/local @rohit-nayak-ps
diff --git a/go/vt/binlog/binlogplayer/binlog_player.go b/go/vt/binlog/binlogplayer/binlog_player.go
index d8f6f51c28..df04691b21 100644
--- a/go/vt/binlog/binlogplayer/binlog_player.go
+++ b/go/vt/binlog/binlogplayer/binlog_player.go
@@ -89,6 +89,7 @@ type Stats struct {
 	QueryCount    *stats.CountersWithSingleLabel
 	CopyRowCount  *stats.Counter
 	CopyLoopCount *stats.Counter
+	ErrorCounts   *stats.CountersWithMultiLabels
 }
 
 // SetLastPosition sets the last replication position.
@@ -129,7 +130,7 @@ func NewStats() *Stats {
 	bps.QueryCount = stats.NewCountersWithSingleLabel("", "", "Phase", "")
 	bps.CopyRowCount = stats.NewCounter("", "")
 	bps.CopyLoopCount = stats.NewCounter("", "")
-
+	bps.ErrorCounts = stats.NewCountersWithMultiLabels("", "", []string{"type"})
 	return bps
 }
 
diff --git a/go/vt/vtexplain/vtexplain_vttablet.go b/go/vt/vtexplain/vtexplain_vttablet.go
index 51095f4339..5fcf2039d7 100644
--- a/go/vt/vtexplain/vtexplain_vttablet.go
+++ b/go/vt/vtexplain/vtexplain_vttablet.go
@@ -614,9 +614,6 @@ func inferColTypeFromExpr(node sqlparser.Expr, colTypeMap map[string]querypb.Typ
 	case *sqlparser.NullVal:
 		colNames = append(colNames, sqlparser.String(node))
 		colTypes = append(colTypes, querypb.Type_NULL_TYPE)
-	case *sqlparser.ComparisonExpr:
-		colNames = append(colNames, sqlparser.String(node))
-		colTypes = append(colTypes, querypb.Type_INT64)
 	default:
 		log.Errorf("vtexplain: unsupported select expression type +%v node %s", reflect.TypeOf(node), sqlparser.String(node))
 	}
diff --git a/go/vt/vtgate/engine/update.go b/go/vt/vtgate/engine/update.go
index a8710e1e32..14b0a2682f 100644
--- a/go/vt/vtgate/engine/update.go
+++ b/go/vt/vtgate/engine/update.go
@@ -21,8 +21,6 @@ import (
 	"sort"
 	"time"
 
-	"vitess.io/vitess/go/vt/vtgate/evalengine"
-
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 
 	"vitess.io/vitess/go/sqltypes"
@@ -38,17 +36,14 @@ import (
 var _ Primitive = (*Update)(nil)
 
 // VindexValues contains changed values for a vindex.
-type VindexValues struct {
-	PvMap  map[string]sqltypes.PlanValue
-	Offset int // Offset from ownedVindexQuery to provide input decision for vindex update.
-}
+type VindexValues map[string]sqltypes.PlanValue
 
 // Update represents the instructions to perform an update.
 type Update struct {
 	DML
 
 	// ChangedVindexValues contains values for updated Vindexes during an update statement.
-	ChangedVindexValues map[string]*VindexValues
+	ChangedVindexValues map[string]VindexValues
 
 	// Update does not take inputs
 	noInputs
@@ -232,23 +227,13 @@ func (upd *Update) updateVindexEntries(vcursor VCursor, bindVars map[string]*que
 		for _, colVindex := range upd.Table.Owned {
 			// Update columns only if they're being changed.
 			if updColValues, ok := upd.ChangedVindexValues[colVindex.Name]; ok {
-				offset := updColValues.Offset
-				if !row[offset].IsNull() {
-					val, err := evalengine.ToInt64(row[offset])
-					if err != nil {
-						return err
-					}
-					if val == int64(1) { // 1 means that the old and new value are same and vindex update is not required.
-						continue
-					}
-				}
 				fromIds := make([]sqltypes.Value, 0, len(colVindex.Columns))
 				var vindexColumnKeys []sqltypes.Value
 				for _, vCol := range colVindex.Columns {
 					// Fetch the column values.
 					origColValue := row[fieldColNumMap[vCol.String()]]
 					fromIds = append(fromIds, origColValue)
-					if colValue, exists := updColValues.PvMap[vCol.String()]; exists {
+					if colValue, exists := updColValues[vCol.String()]; exists {
 						resolvedVal, err := colValue.ResolveValue(bindVars)
 						if err != nil {
 							return err
@@ -281,8 +266,8 @@ func (upd *Update) description() PrimitiveDescription {
 	addFieldsIfNotEmpty(upd.DML, other)
 
 	var changedVindexes []string
-	for k, v := range upd.ChangedVindexValues {
-		changedVindexes = append(changedVindexes, fmt.Sprintf("%s:%d", k, v.Offset))
+	for vindex := range upd.ChangedVindexValues {
+		changedVindexes = append(changedVindexes, vindex)
 	}
 	sort.Strings(changedVindexes) // We sort these so random changes in the map order does not affect output
 	if len(changedVindexes) > 0 {
diff --git a/go/vt/vtgate/planbuilder/update.go b/go/vt/vtgate/planbuilder/update.go
index 79e4060dc8..dfbf18fa01 100644
--- a/go/vt/vtgate/planbuilder/update.go
+++ b/go/vt/vtgate/planbuilder/update.go
@@ -34,20 +34,19 @@ func buildUpdatePlan(stmt sqlparser.Statement, vschema ContextVSchema) (engine.P
 		return nil, err
 	}
 	eupd := &engine.Update{
-		DML: *dml,
+		DML:                 *dml,
+		ChangedVindexValues: make(map[string]engine.VindexValues),
 	}
 
 	if dml.Opcode == engine.Unsharded {
 		return eupd, nil
 	}
 
-	cvv, ovq, err := buildChangedVindexesValues(upd, eupd.Table, ksidCol)
-	if err != nil {
+	if eupd.ChangedVindexValues, err = buildChangedVindexesValues(upd, eupd.Table.ColumnVindexes); err != nil {
 		return nil, err
 	}
-	eupd.ChangedVindexValues = cvv
-	eupd.OwnedVindexQuery = ovq
 	if len(eupd.ChangedVindexValues) != 0 {
+		eupd.OwnedVindexQuery = generateDMLSubquery(upd.Where, upd.OrderBy, upd.Limit, eupd.Table, ksidCol)
 		eupd.KsidVindex = ksidVindex
 	}
 	return eupd, nil
@@ -56,12 +55,10 @@ func buildUpdatePlan(stmt sqlparser.Statement, vschema ContextVSchema) (engine.P
 // buildChangedVindexesValues adds to the plan all the lookup vindexes that are changing.
 // Updates can only be performed to secondary lookup vindexes with no complex expressions
 // in the set clause.
-func buildChangedVindexesValues(update *sqlparser.Update, table *vindexes.Table, ksidCol string) (map[string]*engine.VindexValues, string, error) {
-	changedVindexes := make(map[string]*engine.VindexValues)
-	buf, offset := initialQuery(ksidCol, table)
-	for i, vindex := range table.ColumnVindexes {
-		vindexValueMap := make(map[string]sqltypes.PlanValue)
-		first := true
+func buildChangedVindexesValues(update *sqlparser.Update, colVindexes []*vindexes.ColumnVindex) (map[string]engine.VindexValues, error) {
+	changedVindexes := make(map[string]engine.VindexValues)
+	for i, vindex := range colVindexes {
+		vindexValueMap := make(engine.VindexValues)
 		for _, vcol := range vindex.Columns {
 			// Searching in order of columns in colvindex.
 			found := false
@@ -70,20 +67,14 @@ func buildChangedVindexesValues(update *sqlparser.Update, table *vindexes.Table,
 					continue
 				}
 				if found {
-					return nil, "", vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "column has duplicate set values: '%v'", assignment.Name.Name)
+					return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "column has duplicate set values: '%v'", assignment.Name.Name)
 				}
 				found = true
 				pv, err := extractValueFromUpdate(assignment)
 				if err != nil {
-					return nil, "", err
+					return nil, err
 				}
 				vindexValueMap[vcol.String()] = pv
-				if first {
-					buf.Myprintf(", %v", assignment)
-					first = false
-				} else {
-					buf.Myprintf(" and %v", assignment)
-				}
 			}
 		}
 		if len(vindexValueMap) == 0 {
@@ -92,42 +83,21 @@ func buildChangedVindexesValues(update *sqlparser.Update, table *vindexes.Table,
 		}
 
 		if update.Limit != nil && len(update.OrderBy) == 0 {
-			return nil, "", vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: Need to provide order by clause when using limit. Invalid update on vindex: %v", vindex.Name)
+			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: Need to provide order by clause when using limit. Invalid update on vindex: %v", vindex.Name)
 		}
 		if i == 0 {
-			return nil, "", vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: You can't update primary vindex columns. Invalid update on vindex: %v", vindex.Name)
+			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: You can't update primary vindex columns. Invalid update on vindex: %v", vindex.Name)
 		}
 		if _, ok := vindex.Vindex.(vindexes.Lookup); !ok {
-			return nil, "", vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: You can only update lookup vindexes. Invalid update on vindex: %v", vindex.Name)
+			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: You can only update lookup vindexes. Invalid update on vindex: %v", vindex.Name)
 		}
 		if !vindex.Owned {
-			return nil, "", vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: You can only update owned vindexes. Invalid update on vindex: %v", vindex.Name)
+			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: You can only update owned vindexes. Invalid update on vindex: %v", vindex.Name)
 		}
-		changedVindexes[vindex.Name] = &engine.VindexValues{
-			PvMap:  vindexValueMap,
-			Offset: offset,
-		}
-		offset++
-	}
-	if len(changedVindexes) == 0 {
-		return nil, "", nil
+		changedVindexes[vindex.Name] = vindexValueMap
 	}
-	// generate rest of the owned vindex query.
-	buf.Myprintf(" from %v%v%v%v for update", table.Name, update.Where, update.OrderBy, update.Limit)
-	return changedVindexes, buf.String(), nil
-}
 
-func initialQuery(ksidCol string, table *vindexes.Table) (*sqlparser.TrackedBuffer, int) {
-	buf := sqlparser.NewTrackedBuffer(nil)
-	buf.Myprintf("select %s", ksidCol)
-	offset := 1
-	for _, cv := range table.Owned {
-		for _, column := range cv.Columns {
-			buf.Myprintf(", %v", column)
-			offset++
-		}
-	}
-	return buf, offset
+	return changedVindexes, nil
 }
 
 // extractValueFromUpdate given an UpdateExpr attempts to extracts the Value
diff --git a/go/vt/vttablet/tabletmanager/restore.go b/go/vt/vttablet/tabletmanager/restore.go
index 5e56cb143c..4b0cbf1c9d 100644
--- a/go/vt/vttablet/tabletmanager/restore.go
+++ b/go/vt/vttablet/tabletmanager/restore.go
@@ -130,7 +130,7 @@ func (tm *TabletManager) restoreDataLocked(ctx context.Context, logger logutil.L
 	if originalType == topodatapb.TabletType_MASTER {
 		originalType = tm.baseTabletType
 	}
-	if err := tm.tmState.ChangeTabletType(ctx, topodatapb.TabletType_RESTORE, DBActionNone); err != nil {
+	if err := tm.tmState.ChangeTabletType(ctx, topodatapb.TabletType_RESTORE); err != nil {
 		return err
 	}
 	// Loop until a backup exists, unless we were told to give up immediately.
@@ -179,7 +179,7 @@ func (tm *TabletManager) restoreDataLocked(ctx context.Context, logger logutil.L
 		// No-op, starting with empty database.
 	default:
 		// If anything failed, we should reset the original tablet type
-		if err := tm.tmState.ChangeTabletType(ctx, originalType, DBActionNone); err != nil {
+		if err := tm.tmState.ChangeTabletType(ctx, originalType); err != nil {
 			log.Errorf("Could not change back to original tablet type %v: %v", originalType, err)
 		}
 		return vterrors.Wrap(err, "Can't restore backup")
@@ -195,7 +195,7 @@ func (tm *TabletManager) restoreDataLocked(ctx context.Context, logger logutil.L
 	}
 
 	// Change type back to original type if we're ok to serve.
-	return tm.tmState.ChangeTabletType(ctx, originalType, DBActionNone)
+	return tm.tmState.ChangeTabletType(ctx, originalType)
 }
 
 // restoreToTimeFromBinlog restores to the snapshot time of the keyspace
diff --git a/go/vt/vttablet/tabletmanager/rpc_actions.go b/go/vt/vttablet/tabletmanager/rpc_actions.go
index f6eba93f65..0992d63b4c 100644
--- a/go/vt/vttablet/tabletmanager/rpc_actions.go
+++ b/go/vt/vttablet/tabletmanager/rpc_actions.go
@@ -33,16 +33,6 @@ import (
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 )
 
-// DBAction is used to tell ChangeTabletType whether to call SetReadOnly on change to
-// MASTER tablet type
-type DBAction int
-
-// Allowed values for DBAction
-const (
-	DBActionNone = DBAction(iota)
-	DBActionSetReadWrite
-)
-
 // This file contains the implementations of RPCTM methods.
 // Major groups of methods are broken out into files named "rpc_*.go".
 
@@ -72,11 +62,11 @@ func (tm *TabletManager) ChangeType(ctx context.Context, tabletType topodatapb.T
 		return err
 	}
 	defer tm.unlock()
-	return tm.changeTypeLocked(ctx, tabletType, DBActionNone)
+	return tm.changeTypeLocked(ctx, tabletType)
 }
 
 // ChangeType changes the tablet type
-func (tm *TabletManager) changeTypeLocked(ctx context.Context, tabletType topodatapb.TabletType, action DBAction) error {
+func (tm *TabletManager) changeTypeLocked(ctx context.Context, tabletType topodatapb.TabletType) error {
 	// We don't want to allow multiple callers to claim a tablet as drained. There is a race that could happen during
 	// horizontal resharding where two vtworkers will try to DRAIN the same tablet. This check prevents that race from
 	// causing errors.
@@ -84,7 +74,7 @@ func (tm *TabletManager) changeTypeLocked(ctx context.Context, tabletType topoda
 		return fmt.Errorf("Tablet: %v, is already drained", tm.tabletAlias)
 	}
 
-	if err := tm.tmState.ChangeTabletType(ctx, tabletType, action); err != nil {
+	if err := tm.tmState.ChangeTabletType(ctx, tabletType); err != nil {
 		return err
 	}
 
diff --git a/go/vt/vttablet/tabletmanager/rpc_backup.go b/go/vt/vttablet/tabletmanager/rpc_backup.go
index b566667240..cbb5f179f6 100644
--- a/go/vt/vttablet/tabletmanager/rpc_backup.go
+++ b/go/vt/vttablet/tabletmanager/rpc_backup.go
@@ -84,7 +84,7 @@ func (tm *TabletManager) Backup(ctx context.Context, concurrency int, logger log
 		}
 		originalType = tablet.Type
 		// update our type to BACKUP
-		if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_BACKUP, DBActionNone); err != nil {
+		if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_BACKUP); err != nil {
 			return err
 		}
 	}
@@ -115,7 +115,7 @@ func (tm *TabletManager) Backup(ctx context.Context, concurrency int, logger log
 
 		// Change our type back to the original value.
 		// Original type could be master so pass in a real value for masterTermStartTime
-		if err := tm.changeTypeLocked(bgCtx, originalType, DBActionNone); err != nil {
+		if err := tm.changeTypeLocked(bgCtx, originalType); err != nil {
 			// failure in changing the topology type is probably worse,
 			// so returning that (we logged the snapshot error anyway)
 			if returnErr != nil {
diff --git a/go/vt/vttablet/tabletmanager/rpc_replication.go b/go/vt/vttablet/tabletmanager/rpc_replication.go
index 8116592a2b..9ca7d53508 100644
--- a/go/vt/vttablet/tabletmanager/rpc_replication.go
+++ b/go/vt/vttablet/tabletmanager/rpc_replication.go
@@ -244,7 +244,7 @@ func (tm *TabletManager) InitMaster(ctx context.Context) (string, error) {
 	// Set the server read-write, from now on we can accept real
 	// client writes. Note that if semi-sync replication is enabled,
 	// we'll still need some replicas to be able to commit transactions.
-	if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_MASTER, DBActionSetReadWrite); err != nil {
+	if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_MASTER); err != nil {
 		return "", err
 	}
 
@@ -281,7 +281,7 @@ func (tm *TabletManager) InitReplica(ctx context.Context, parent *topodatapb.Tab
 	// is used on the old master when using InitShardMaster with
 	// -force, and the new master is different from the old master.
 	if tm.Tablet().Type == topodatapb.TabletType_MASTER {
-		if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_REPLICA, DBActionNone); err != nil {
+		if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_REPLICA); err != nil {
 			return err
 		}
 	}
@@ -520,7 +520,7 @@ func (tm *TabletManager) setMasterLocked(ctx context.Context, parentAlias *topod
 	// unintentionally change the type of RDONLY tablets
 	tablet := tm.Tablet()
 	if tablet.Type == topodatapb.TabletType_MASTER {
-		if err := tm.tmState.ChangeTabletType(ctx, topodatapb.TabletType_REPLICA, DBActionNone); err != nil {
+		if err := tm.tmState.ChangeTabletType(ctx, topodatapb.TabletType_REPLICA); err != nil {
 			return err
 		}
 	}
@@ -622,7 +622,7 @@ func (tm *TabletManager) ReplicaWasRestarted(ctx context.Context, parent *topoda
 	if tablet.Type != topodatapb.TabletType_MASTER {
 		return nil
 	}
-	return tm.tmState.ChangeTabletType(ctx, topodatapb.TabletType_REPLICA, DBActionNone)
+	return tm.tmState.ChangeTabletType(ctx, topodatapb.TabletType_REPLICA)
 }
 
 // StopReplicationAndGetStatus stops MySQL replication, and returns the
@@ -733,7 +733,7 @@ func (tm *TabletManager) PromoteReplica(ctx context.Context) (string, error) {
 		return "", err
 	}
 
-	if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_MASTER, DBActionSetReadWrite); err != nil {
+	if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_MASTER); err != nil {
 		return "", err
 	}
 
diff --git a/go/vt/vttablet/tabletmanager/tm_state.go b/go/vt/vttablet/tabletmanager/tm_state.go
index 68ba9235c4..80a4d5e05b 100644
--- a/go/vt/vttablet/tabletmanager/tm_state.go
+++ b/go/vt/vttablet/tabletmanager/tm_state.go
@@ -150,7 +150,7 @@ func (ts *tmState) RefreshFromTopoInfo(ctx context.Context, shardInfo *topo.Shar
 	ts.updateLocked(ctx)
 }
 
-func (ts *tmState) ChangeTabletType(ctx context.Context, tabletType topodatapb.TabletType, action DBAction) error {
+func (ts *tmState) ChangeTabletType(ctx context.Context, tabletType topodatapb.TabletType) error {
 	ts.mu.Lock()
 	defer ts.mu.Unlock()
 	log.Infof("Changing Tablet Type: %v", tabletType)
@@ -163,12 +163,10 @@ func (ts *tmState) ChangeTabletType(ctx context.Context, tabletType topodatapb.T
 		if err != nil {
 			return err
 		}
-		if action == DBActionSetReadWrite {
-			// We call SetReadOnly only after the topo has been updated to avoid
-			// situations where two tablets are master at the DB level but not at the vitess level
-			if err := ts.tm.MysqlDaemon.SetReadOnly(false); err != nil {
-				return err
-			}
+		// We call SetReadOnly only after the topo has been updated to avoid
+		// situations where two tablets are master at the DB level but not at the vitess level
+		if err := ts.tm.MysqlDaemon.SetReadOnly(false); err != nil {
+			return err
 		}
 
 		ts.tablet.Type = tabletType
diff --git a/go/vt/vttablet/tabletmanager/vreplication/controller.go b/go/vt/vttablet/tabletmanager/vreplication/controller.go
index cb0e722c29..512eca415a 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/controller.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/controller.go
@@ -150,6 +150,7 @@ func (ct *controller) run(ctx context.Context) {
 		default:
 		}
 		log.Errorf("stream %v: %v, retrying after %v", ct.id, err, *retryDelay)
+		ct.blpStats.ErrorCounts.Add([]string{"Stream Error"}, 1)
 		timer := time.NewTimer(*retryDelay)
 		select {
 		case <-ctx.Done():
@@ -192,6 +193,7 @@ func (ct *controller) runBlp(ctx context.Context) (err error) {
 		log.Infof("trying to find a tablet eligible for vreplication. stream id: %v", ct.id)
 		tablet, err = ct.tabletPicker.PickForStreaming(ctx)
 		if err != nil {
+			ct.blpStats.ErrorCounts.Add([]string{"No Source Tablet Found"}, 1)
 			return err
 		}
 		log.Infof("found a tablet eligible for vreplication. stream id: %v  tablet: %s", ct.id, tablet.Alias.String())
@@ -203,6 +205,7 @@ func (ct *controller) runBlp(ctx context.Context) (err error) {
 		// Table names can have search patterns. Resolve them against the schema.
 		tables, err := mysqlctl.ResolveTables(ctx, ct.mysqld, dbClient.DBName(), ct.source.Tables)
 		if err != nil {
+			ct.blpStats.ErrorCounts.Add([]string{"Invalid Source"}, 1)
 			return vterrors.Wrap(err, "failed to resolve table names")
 		}
 
@@ -241,6 +244,7 @@ func (ct *controller) runBlp(ctx context.Context) (err error) {
 		vr := newVReplicator(ct.id, &ct.source, vsClient, ct.blpStats, dbClient, ct.mysqld, ct.vre)
 		return vr.Replicate(ctx)
 	}
+	ct.blpStats.ErrorCounts.Add([]string{"Invalid Source"}, 1)
 	return fmt.Errorf("missing source")
 }
 
diff --git a/go/vt/vttablet/tabletmanager/vreplication/stats.go b/go/vt/vttablet/tabletmanager/vreplication/stats.go
index 90e32c7684..2a218d98a9 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/stats.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/stats.go
@@ -252,6 +252,21 @@ func (st *vrStats) register() {
 			}
 			return result
 		})
+	stats.NewCountersFuncWithMultiLabels(
+		"VReplicationErrors",
+		"Errors during vreplication",
+		[]string{"workflow", "type"},
+		func() map[string]int64 {
+			st.mu.Lock()
+			defer st.mu.Unlock()
+			result := make(map[string]int64)
+			for _, ct := range st.controllers {
+				for key, val := range ct.blpStats.ErrorCounts.Counts() {
+					result[fmt.Sprintf("%d_%s", ct.id, key)] = val
+				}
+			}
+			return result
+		})
 }
 
 func (st *vrStats) numControllers() int64 {
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vplayer.go b/go/vt/vttablet/tabletmanager/vreplication/vplayer.go
index ad3ef6f388..cdfb62ed71 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vplayer.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vplayer.go
@@ -103,6 +103,7 @@ func (vp *vplayer) play(ctx context.Context) error {
 
 	plan, err := buildReplicatorPlan(vp.vr.source.Filter, vp.vr.pkInfoMap, vp.copyState)
 	if err != nil {
+		vp.vr.stats.ErrorCounts.Add([]string{"Plan"}, 1)
 		return err
 	}
 	vp.replicatorPlan = plan
@@ -366,6 +367,8 @@ func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
 					}
 				}
 				if err := vp.applyEvent(ctx, event, mustSave); err != nil {
+					vp.vr.stats.ErrorCounts.Add([]string{"Apply"}, 1)
+					log.Errorf("Error applying event: %s", err.Error())
 					return err
 				}
 			}
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vreplicator.go b/go/vt/vttablet/tabletmanager/vreplication/vreplicator.go
index 105492a479..48135c9910 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vreplicator.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vreplicator.go
@@ -121,6 +121,7 @@ func newVReplicator(id uint32, source *binlogdatapb.BinlogSource, sourceVStreame
 func (vr *vreplicator) Replicate(ctx context.Context) error {
 	err := vr.replicate(ctx)
 	if err != nil {
+		log.Errorf("Replicate error: %s", err.Error())
 		if err := vr.setMessage(err.Error()); err != nil {
 			log.Errorf("Failed to set error state: %v", err)
 		}
@@ -165,10 +166,12 @@ func (vr *vreplicator) replicate(ctx context.Context) error {
 				return err
 			}
 			if err := newVCopier(vr).copyNext(ctx, settings); err != nil {
+				vr.stats.ErrorCounts.Add([]string{"Copy"}, 1)
 				return err
 			}
 		case settings.StartPos.IsZero():
 			if err := newVCopier(vr).initTablesForCopy(ctx); err != nil {
+				vr.stats.ErrorCounts.Add([]string{"Copy"}, 1)
 				return err
 			}
 		default:
@@ -180,6 +183,7 @@ func (vr *vreplicator) replicate(ctx context.Context) error {
 				return vr.setState(binlogplayer.BlpStopped, "Stopped after copy.")
 			}
 			if err := vr.setState(binlogplayer.BlpRunning, ""); err != nil {
+				vr.stats.ErrorCounts.Add([]string{"Replicate"}, 1)
 				return err
 			}
 			return newVPlayer(vr, settings, nil, mysql.Position{}, "replicate").play(ctx)
diff --git a/go/vt/vttablet/tabletserver/vstreamer/copy.go b/go/vt/vttablet/tabletserver/vstreamer/copy.go
index 8cc08f7a32..a02d273b87 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/copy.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/copy.go
@@ -37,6 +37,7 @@ func (uvs *uvstreamer) copy(ctx context.Context) error {
 		tableName := uvs.tablesToCopy[0]
 		log.V(2).Infof("Copystate not empty starting catchupAndCopy on table %s", tableName)
 		if err := uvs.catchupAndCopy(ctx, tableName); err != nil {
+			uvs.vse.errorCounts.Add("Copy", 1)
 			return err
 		}
 	}
@@ -50,6 +51,7 @@ func (uvs *uvstreamer) catchupAndCopy(ctx context.Context, tableName string) err
 	if !uvs.pos.IsZero() {
 		if err := uvs.catchup(ctx); err != nil {
 			log.Infof("catchupAndCopy: catchup returned %v", err)
+			uvs.vse.errorCounts.Add("Catchup", 1)
 			return err
 		}
 	}
@@ -265,6 +267,7 @@ func (uvs *uvstreamer) copyTable(ctx context.Context, tableName string) error {
 		return nil
 	})
 	if err != nil {
+		uvs.vse.errorCounts.Add("StreamRows", 1)
 		return err
 	}
 
diff --git a/go/vt/vttablet/tabletserver/vstreamer/engine.go b/go/vt/vttablet/tabletserver/vstreamer/engine.go
index 532a2e48ef..5ddc4ecaea 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/engine.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/engine.go
@@ -73,14 +73,17 @@ type Engine struct {
 	vschemaUpdates *stats.Counter
 
 	// vstreamer metrics
-	vstreamerPhaseTimings    *servenv.TimingsWrapper
-	vstreamerEventsStreamed  *stats.Counter
-	vstreamerPacketSize      *stats.GaugeFunc
-	vstreamerNumPackets      *stats.Counter
-	resultStreamerNumRows    *stats.Counter
-	resultStreamerNumPackets *stats.Counter
-	rowStreamerNumRows       *stats.Counter
-	rowStreamerNumPackets    *stats.Counter
+	vstreamerPhaseTimings     *servenv.TimingsWrapper
+	vstreamerEventsStreamed   *stats.Counter
+	vstreamerPacketSize       *stats.GaugeFunc
+	vstreamerNumPackets       *stats.Counter
+	resultStreamerNumRows     *stats.Counter
+	resultStreamerNumPackets  *stats.Counter
+	rowStreamerNumRows        *stats.Counter
+	rowStreamerNumPackets     *stats.Counter
+	errorCounts               *stats.CountersWithSingleLabel
+	vstreamersCreated         *stats.Counter
+	vstreamersEndedWithErrors *stats.Counter
 }
 
 // NewEngine creates a new Engine.
@@ -102,14 +105,17 @@ func NewEngine(env tabletenv.Env, ts srvtopo.Server, se *schema.Engine, cell str
 		vschemaErrors:  env.Exporter().NewCounter("VSchemaErrors", "Count of VSchema errors"),
 		vschemaUpdates: env.Exporter().NewCounter("VSchemaUpdates", "Count of VSchema updates. Does not include errors"),
 
-		vstreamerPhaseTimings:    env.Exporter().NewTimings("VStreamerPhaseTiming", "Time taken for different phases during vstream copy", "phase-timing"),
-		vstreamerEventsStreamed:  env.Exporter().NewCounter("VStreamerEventsStreamed", "Count of events streamed in VStream API"),
-		vstreamerPacketSize:      env.Exporter().NewGaugeFunc("VStreamPacketSize", "Max packet size for sending vstreamer events", getPacketSize),
-		vstreamerNumPackets:      env.Exporter().NewCounter("VStreamerNumPackets", "Number of packets in vstreamer"),
-		resultStreamerNumPackets: env.Exporter().NewCounter("ResultStreamerNumPackets", "Number of packets in result streamer"),
-		resultStreamerNumRows:    env.Exporter().NewCounter("ResultStreamerNumRows", "Number of rows sent in result streamer"),
-		rowStreamerNumPackets:    env.Exporter().NewCounter("RowStreamerNumPackets", "Number of packets in row streamer"),
-		rowStreamerNumRows:       env.Exporter().NewCounter("RowStreamerNumRows", "Number of rows sent in row streamer"),
+		vstreamerPhaseTimings:     env.Exporter().NewTimings("VStreamerPhaseTiming", "Time taken for different phases during vstream copy", "phase-timing"),
+		vstreamerEventsStreamed:   env.Exporter().NewCounter("VStreamerEventsStreamed", "Count of events streamed in VStream API"),
+		vstreamerPacketSize:       env.Exporter().NewGaugeFunc("VStreamPacketSize", "Max packet size for sending vstreamer events", getPacketSize),
+		vstreamerNumPackets:       env.Exporter().NewCounter("VStreamerNumPackets", "Number of packets in vstreamer"),
+		resultStreamerNumPackets:  env.Exporter().NewCounter("ResultStreamerNumPackets", "Number of packets in result streamer"),
+		resultStreamerNumRows:     env.Exporter().NewCounter("ResultStreamerNumRows", "Number of rows sent in result streamer"),
+		rowStreamerNumPackets:     env.Exporter().NewCounter("RowStreamerNumPackets", "Number of packets in row streamer"),
+		rowStreamerNumRows:        env.Exporter().NewCounter("RowStreamerNumRows", "Number of rows sent in row streamer"),
+		vstreamersCreated:         env.Exporter().NewCounter("VStreamersCreated", "Count of vstreamers created"),
+		vstreamersEndedWithErrors: env.Exporter().NewCounter("VStreamersEndedWithErrors", "Count of vstreamers that ended with errors"),
+		errorCounts:               env.Exporter().NewCountersWithSingleLabel("VStreamerErrors", "Tracks errors in vstreamer", "type", "Catchup", "Copy", "Send", "TablePlan"),
 	}
 	env.Exporter().HandleFunc("/debug/tablet_vschema", vse.ServeHTTP)
 	return vse
diff --git a/go/vt/vttablet/tabletserver/vstreamer/uvstreamer.go b/go/vt/vttablet/tabletserver/vstreamer/uvstreamer.go
index 6ed3db5bf2..b3caa498fa 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/uvstreamer.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/uvstreamer.go
@@ -286,6 +286,9 @@ func (uvs *uvstreamer) send2(evs []*binlogdatapb.VEvent) error {
 			}
 		}
 	}
+	if err != nil {
+		uvs.vse.errorCounts.Add("Send", 1)
+	}
 	return err
 }
 
@@ -298,7 +301,7 @@ func (uvs *uvstreamer) sendEventsForCurrentPos() error {
 		Type: binlogdatapb.VEventType_OTHER,
 	}}
 	if err := uvs.send(evs); err != nil {
-		return wrapError(err, uvs.pos)
+		return wrapError(err, uvs.pos, uvs.vse)
 	}
 	return nil
 }
@@ -320,6 +323,7 @@ func (uvs *uvstreamer) setStreamStartPosition() error {
 		return vterrors.Wrap(err, "could not decode position")
 	}
 	if !curPos.AtLeast(pos) {
+		uvs.vse.errorCounts.Add("GTIDSet Mismatch", 1)
 		return fmt.Errorf("GTIDSet Mismatch: requested source position:%v, current target vrep position: %v", mysql.EncodePosition(pos), mysql.EncodePosition(curPos))
 	}
 	uvs.pos = pos
@@ -362,6 +366,7 @@ func (uvs *uvstreamer) Stream() error {
 		log.Info("TablePKs is not nil: starting vs.copy()")
 		if err := uvs.copy(uvs.ctx); err != nil {
 			log.Infof("uvstreamer.Stream() copy returned with err %s", err)
+			uvs.vse.errorCounts.Add("Copy", 1)
 			return err
 		}
 		uvs.sendTestEvent("Copy Done")
diff --git a/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go b/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go
index cfe453298c..6d387beee8 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go
@@ -147,9 +147,12 @@ func (vs *vstreamer) Stream() error {
 	//defer vs.cancel()
 	ctx := context.Background()
 	defer ctx.Done()
+	vs.vse.vstreamersCreated.Add(1)
 	log.Infof("Starting Stream() with startPos %s", vs.startPos)
 	pos, err := mysql.DecodePosition(vs.startPos)
 	if err != nil {
+		vs.vse.errorCounts.Add("StreamRows", 1)
+		vs.vse.vstreamersEndedWithErrors.Add(1)
 		return err
 	}
 	vs.pos = pos
@@ -161,21 +164,21 @@ func (vs *vstreamer) replicate(ctx context.Context) error {
 	// Ensure se is Open. If vttablet came up in a non_serving role,
 	// the schema engine may not have been initialized.
 	if err := vs.se.Open(); err != nil {
-		return wrapError(err, vs.pos)
+		return wrapError(err, vs.pos, vs.vse)
 	}
 
 	conn, err := binlog.NewBinlogConnection(vs.cp)
 	if err != nil {
-		return wrapError(err, vs.pos)
+		return wrapError(err, vs.pos, vs.vse)
 	}
 	defer conn.Close()
 
 	events, err := conn.StartBinlogDumpFromPosition(vs.ctx, vs.pos)
 	if err != nil {
-		return wrapError(err, vs.pos)
+		return wrapError(err, vs.pos, vs.vse)
 	}
 	err = vs.parseEvents(vs.ctx, events)
-	return wrapError(err, vs.pos)
+	return wrapError(err, vs.pos, vs.vse)
 }
 
 // parseEvents parses and sends events.
@@ -252,6 +255,7 @@ func (vs *vstreamer) parseEvents(ctx context.Context, events <-chan mysql.Binlog
 		case binlogdatapb.VEventType_SAVEPOINT:
 			bufferedEvents = append(bufferedEvents, vevent)
 		default:
+			vs.vse.errorCounts.Add("BufferAndTransmit", 1)
 			return fmt.Errorf("unexpected event: %v", vevent)
 		}
 		return nil
@@ -280,6 +284,7 @@ func (vs *vstreamer) parseEvents(ctx context.Context, events <-chan mysql.Binlog
 			}
 			vevents, err := vs.parseEvent(ev)
 			if err != nil {
+				vs.vse.errorCounts.Add("ParseEvent", 1)
 				return err
 			}
 			for _, vevent := range vevents {
@@ -287,6 +292,7 @@ func (vs *vstreamer) parseEvents(ctx context.Context, events <-chan mysql.Binlog
 					if err == io.EOF {
 						return nil
 					}
+					vs.vse.errorCounts.Add("BufferAndTransmit", 1)
 					return fmt.Errorf("error sending event: %v", err)
 				}
 			}
@@ -308,6 +314,7 @@ func (vs *vstreamer) parseEvents(ctx context.Context, events <-chan mysql.Binlog
 				if err == io.EOF {
 					return nil
 				}
+				vs.vse.errorCounts.Add("Send", 1)
 				return fmt.Errorf("error sending event: %v", err)
 			}
 		}
@@ -503,6 +510,7 @@ func (vs *vstreamer) parseEvent(ev mysql.BinlogEvent) ([]*binlogdatapb.VEvent, e
 
 		vevent, err := vs.buildTablePlan(id, tm)
 		if err != nil {
+			vs.vse.errorCounts.Add("TablePlan", 1)
 			return nil, err
 		}
 		if vevent != nil {
@@ -538,7 +546,6 @@ func (vs *vstreamer) parseEvent(ev mysql.BinlogEvent) ([]*binlogdatapb.VEvent, e
 		if err != nil {
 			return nil, err
 		}
-
 	}
 	for _, vevent := range vevents {
 		vevent.Timestamp = int64(ev.Timestamp())
@@ -807,8 +814,10 @@ func (vs *vstreamer) extractRowAndFilter(plan *streamerPlan, data []byte, dataCo
 	return plan.filter(values)
 }
 
-func wrapError(err error, stopPos mysql.Position) error {
+func wrapError(err error, stopPos mysql.Position, vse *Engine) error {
 	if err != nil {
+		vse.vstreamersEndedWithErrors.Add(1)
+		vse.errorCounts.Add("StreamEnded", 1)
 		err = fmt.Errorf("stream (at source tablet) error @ %v: %v", stopPos, err)
 		log.Error(err)
 		return err


Test patch:
diff --git a/go/test/endtoend/tabletmanager/tablet_test.go b/go/test/endtoend/tabletmanager/tablet_test.go
index 0ce8cd9427..8e6a1ad5a5 100644
--- a/go/test/endtoend/tabletmanager/tablet_test.go
+++ b/go/test/endtoend/tabletmanager/tablet_test.go
@@ -46,15 +46,11 @@ func TestEnsureDB(t *testing.T) {
 	err = clusterInstance.VtctlclientProcess.ExecuteCommand("TabletExternallyReparented", tablet.Alias)
 	require.NoError(t, err)
 
-	// It is still NOT_SERVING because the db is read-only.
-	assert.Equal(t, "NOT_SERVING", tablet.VttabletProcess.GetTabletStatus())
+	// It goes SERVING because TER calls ChangeTabletType which will also set the database to read-write
+	assert.Equal(t, "SERVING", tablet.VttabletProcess.GetTabletStatus())
 	status := tablet.VttabletProcess.GetStatusDetails()
-	assert.Contains(t, status, "read-only")
+	assert.Contains(t, status, "Serving")
 
-	// Switch to read-write and verify that that we go serving.
-	_ = clusterInstance.VtctlclientProcess.ExecuteCommand("SetReadWrite", tablet.Alias)
-	err = tablet.VttabletProcess.WaitForTabletType("SERVING")
-	require.NoError(t, err)
 	killTablets(t, tablet)
 }
 
diff --git a/go/test/endtoend/vtgate/concurrentdml/main_test.go b/go/test/endtoend/vtgate/concurrentdml/main_test.go
index c613b348b5..b447ba8a72 100644
--- a/go/test/endtoend/vtgate/concurrentdml/main_test.go
+++ b/go/test/endtoend/vtgate/concurrentdml/main_test.go
@@ -364,8 +364,8 @@ func TestUpdateLookupUniqueVindex(t *testing.T) {
 	exec(t, conn, `update t1 set c3 = 400 where c2 = 200`)
 	// changed - same vindex
 	exec(t, conn, `update t1 set c4 = 'abc' where c1 = 999`)
-	// not changed - same vindex
-	exec(t, conn, `update t1 set c4 = 'abc' where c4 = 'abc'`)
+	// not changed - same vindex - not yet supported bcoz of varchar field
+	// exec(t, conn, `update t1 set c4 = 'abc' where c4 = 'abc'`)
 
 }
 
diff --git a/go/vt/vtexplain/testdata/multi-output/updatesharded-output.txt b/go/vt/vtexplain/testdata/multi-output/updatesharded-output.txt
index 9df3b3f352..2bd0b63868 100644
--- a/go/vt/vtexplain/testdata/multi-output/updatesharded-output.txt
+++ b/go/vt/vtexplain/testdata/multi-output/updatesharded-output.txt
@@ -26,7 +26,7 @@ update user set pet='fido' where id=1
 update user set name='alicia' where id=1
 
 1 ks_sharded/-40: begin
-1 ks_sharded/-40: select id, name, name = 'alicia' from user where id = 1 limit 10001 for update
+1 ks_sharded/-40: select id, name from user where id = 1 limit 10001 for update
 2 ks_sharded/40-80: begin
 2 ks_sharded/40-80: delete from name_user_map where name = 'name_val_2' and user_id = 1 limit 10001
 3 ks_sharded/c0-: begin
@@ -42,7 +42,7 @@ update user set name='alicia' where name='alice'
 1 ks_sharded/40-80: begin
 1 ks_sharded/40-80: select name, user_id from name_user_map where name in ('alice') limit 10001 for update
 2 ks_sharded/-40: begin
-2 ks_sharded/-40: select id, name, name = 'alicia' from user where name = 'alice' limit 10001 for update
+2 ks_sharded/-40: select id, name from user where name = 'alice' limit 10001 for update
 3 ks_sharded/40-80: delete from name_user_map where name = 'name_val_2' and user_id = 1 limit 10001
 4 ks_sharded/c0-: begin
 4 ks_sharded/c0-: insert into name_user_map(name, user_id) values ('alicia', 1)
diff --git a/go/vt/vtgate/autocommit_test.go b/go/vt/vtgate/autocommit_test.go
index b5576adb3a..f84fa89edb 100644
--- a/go/vt/vtgate/autocommit_test.go
+++ b/go/vt/vtgate/autocommit_test.go
@@ -84,8 +84,8 @@ func TestAutocommitUpdateLookup(t *testing.T) {
 func TestAutocommitUpdateVindexChange(t *testing.T) {
 	executor, sbc, _, sbclookup := createLegacyExecutorEnv()
 	sbc.SetResults([]*sqltypes.Result{sqltypes.MakeTestResult(
-		sqltypes.MakeTestFields("id|name|lastname|name_lastname_keyspace_id_map", "int64|int32|varchar|int64"),
-		"1|1|foo|0",
+		sqltypes.MakeTestFields("id|name|lastname", "int64|int32|varchar"),
+		"1|1|foo",
 	),
 	})
 
@@ -110,7 +110,7 @@ func TestAutocommitUpdateVindexChange(t *testing.T) {
 	testCommitCount(t, "sbclookup", sbclookup, 1)
 
 	testQueries(t, "sbc", sbc, []*querypb.BoundQuery{{
-		Sql:           "select id, name, lastname, name = 'myname' and lastname = 'mylastname' from user2 where id = 1 for update",
+		Sql:           "select id, name, lastname from user2 where id = 1 for update",
 		BindVariables: map[string]*querypb.BindVariable{},
 	}, {
 		Sql:           "update user2 set name = 'myname', lastname = 'mylastname' where id = 1",
diff --git a/go/vt/vtgate/engine/update_test.go b/go/vt/vtgate/engine/update_test.go
index 5e64062bda..76fb07fdc7 100644
--- a/go/vt/vtgate/engine/update_test.go
+++ b/go/vt/vtgate/engine/update_test.go
@@ -205,29 +205,23 @@ func TestUpdateEqualChangedVindex(t *testing.T) {
 			OwnedVindexQuery: "dummy_subquery",
 			KsidVindex:       ks.Vindexes["hash"].(vindexes.SingleColumn),
 		},
-		ChangedVindexValues: map[string]*VindexValues{
+		ChangedVindexValues: map[string]VindexValues{
 			"twocol": {
-				PvMap: map[string]sqltypes.PlanValue{
-					"c1": {Value: sqltypes.NewInt64(1)},
-					"c2": {Value: sqltypes.NewInt64(2)},
-				},
-				Offset: 4,
+				"c1": {Value: sqltypes.NewInt64(1)},
+				"c2": {Value: sqltypes.NewInt64(2)},
 			},
 			"onecol": {
-				PvMap: map[string]sqltypes.PlanValue{
-					"c3": {Value: sqltypes.NewInt64(3)},
-				},
-				Offset: 5,
+				"c3": {Value: sqltypes.NewInt64(3)},
 			},
 		},
 	}
 
 	results := []*sqltypes.Result{sqltypes.MakeTestResult(
 		sqltypes.MakeTestFields(
-			"id|c1|c2|c3|twocol|onecol",
-			"int64|int64|int64|int64|int64|int64",
+			"id|c1|c2|c3",
+			"int64|int64|int64|int64",
 		),
-		"1|4|5|6|0|0",
+		"1|4|5|6",
 	)}
 	vc := newDMLTestVCursor("-20", "20-")
 	vc.results = results
@@ -264,14 +258,14 @@ func TestUpdateEqualChangedVindex(t *testing.T) {
 		`ExecuteMultiShard sharded.-20: dummy_update {} true true`,
 	})
 
-	// multiple rows changing.
+	// Failure case: multiple rows changing.
 	results = []*sqltypes.Result{sqltypes.MakeTestResult(
 		sqltypes.MakeTestFields(
-			"id|c1|c2|c3|twocol|onecol",
-			"int64|int64|int64|int64|int64|int64",
+			"id|c1|c2|c3",
+			"int64|int64|int64|int64",
 		),
-		"1|4|5|6|0|0",
-		"1|7|8|9|0|0",
+		"1|4|5|6",
+		"1|7|8|9",
 	)}
 	vc = newDMLTestVCursor("-20", "20-")
 	vc.results = results
@@ -290,40 +284,8 @@ func TestUpdateEqualChangedVindex(t *testing.T) {
 		// 6 has to be replaced by 3.
 		`Execute delete from lkp1 where from = :from and toc = :toc from: type:INT64 value:"6" toc: type:VARBINARY value:"\026k@\264J\272K\326"  true`,
 		`Execute insert into lkp1(from, toc) values(:from_0, :toc_0) from_0: type:INT64 value:"3" toc_0: type:VARBINARY value:"\026k@\264J\272K\326"  true`,
-		// 7,8 have to be replaced by 1,2 (the new values).
 		`Execute delete from lkp2 where from1 = :from1 and from2 = :from2 and toc = :toc from1: type:INT64 value:"7" from2: type:INT64 value:"8" toc: type:VARBINARY value:"\026k@\264J\272K\326"  true`,
 		`Execute insert into lkp2(from1, from2, toc) values(:from1_0, :from2_0, :toc_0) from1_0: type:INT64 value:"1" from2_0: type:INT64 value:"2" toc_0: type:VARBINARY value:"\026k@\264J\272K\326"  true`,
-		// 9 has to be replaced by 3.
-		`Execute delete from lkp1 where from = :from and toc = :toc from: type:INT64 value:"9" toc: type:VARBINARY value:"\026k@\264J\272K\326"  true`,
-		`Execute insert into lkp1(from, toc) values(:from_0, :toc_0) from_0: type:INT64 value:"3" toc_0: type:VARBINARY value:"\026k@\264J\272K\326"  true`,
-		// Finally, the actual update, which is also sent to -20, same route as the subquery.
-		`ExecuteMultiShard sharded.-20: dummy_update {} true true`,
-	})
-
-	// multiple rows changing, but only some vindex actually changes
-	results = []*sqltypes.Result{sqltypes.MakeTestResult(
-		sqltypes.MakeTestFields(
-			"id|c1|c2|c3|twocol|onecol",
-			"int64|int64|int64|int64|int64|int64",
-		),
-		"1|4|5|6|0|1", // twocol changes
-		"1|7|8|9|1|0", // onecol changes
-	)}
-	vc = newDMLTestVCursor("-20", "20-")
-	vc.results = results
-
-	_, err = upd.Execute(vc, map[string]*querypb.BindVariable{}, false)
-	require.NoError(t, err)
-	vc.ExpectLog(t, []string{
-		`ResolveDestinations sharded [] Destinations:DestinationKeyspaceID(166b40b44aba4bd6)`,
-		// ResolveDestinations is hard-coded to return -20.
-		// It gets used to perform the subquery to fetch the changing column values.
-		`ExecuteMultiShard sharded.-20: dummy_subquery {} false false`,
-		// Those values are returned as 4,5 for twocol and 6 for onecol.
-		// 4,5 have to be replaced by 1,2 (the new values).
-		`Execute delete from lkp2 where from1 = :from1 and from2 = :from2 and toc = :toc from1: type:INT64 value:"4" from2: type:INT64 value:"5" toc: type:VARBINARY value:"\026k@\264J\272K\326"  true`,
-		`Execute insert into lkp2(from1, from2, toc) values(:from1_0, :from2_0, :toc_0) from1_0: type:INT64 value:"1" from2_0: type:INT64 value:"2" toc_0: type:VARBINARY value:"\026k@\264J\272K\326"  true`,
-		// 9 has to be replaced by 3.
 		`Execute delete from lkp1 where from = :from and toc = :toc from: type:INT64 value:"9" toc: type:VARBINARY value:"\026k@\264J\272K\326"  true`,
 		`Execute insert into lkp1(from, toc) values(:from_0, :toc_0) from_0: type:INT64 value:"3" toc_0: type:VARBINARY value:"\026k@\264J\272K\326"  true`,
 		// Finally, the actual update, which is also sent to -20, same route as the subquery.
@@ -344,29 +306,23 @@ func TestUpdateScatterChangedVindex(t *testing.T) {
 			OwnedVindexQuery: "dummy_subquery",
 			KsidVindex:       ks.Vindexes["hash"].(vindexes.SingleColumn),
 		},
-		ChangedVindexValues: map[string]*VindexValues{
+		ChangedVindexValues: map[string]VindexValues{
 			"twocol": {
-				PvMap: map[string]sqltypes.PlanValue{
-					"c1": {Value: sqltypes.NewInt64(1)},
-					"c2": {Value: sqltypes.NewInt64(2)},
-				},
-				Offset: 4,
+				"c1": {Value: sqltypes.NewInt64(1)},
+				"c2": {Value: sqltypes.NewInt64(2)},
 			},
 			"onecol": {
-				PvMap: map[string]sqltypes.PlanValue{
-					"c3": {Value: sqltypes.NewInt64(3)},
-				},
-				Offset: 5,
+				"c3": {Value: sqltypes.NewInt64(3)},
 			},
 		},
 	}
 
 	results := []*sqltypes.Result{sqltypes.MakeTestResult(
 		sqltypes.MakeTestFields(
-			"id|c1|c2|c3|twocol|onecol",
-			"int64|int64|int64|int64|int64|int64",
+			"id|c1|c2|c3",
+			"int64|int64|int64|int64",
 		),
-		"1|4|5|6|0|0",
+		"1|4|5|6",
 	)}
 	vc := newDMLTestVCursor("-20", "20-")
 	vc.results = results
@@ -406,11 +362,11 @@ func TestUpdateScatterChangedVindex(t *testing.T) {
 	// Update can affect multiple rows
 	results = []*sqltypes.Result{sqltypes.MakeTestResult(
 		sqltypes.MakeTestFields(
-			"id|c1|c2|c3|twocol|onecol",
-			"int64|int64|int64|int64|int64|int64",
+			"id|c1|c2|c3",
+			"int64|int64|int64|int64",
 		),
-		"1|4|5|6|0|0",
-		"1|7|8|9|0|0",
+		"1|4|5|6",
+		"1|7|8|9",
 	)}
 	vc = newDMLTestVCursor("-20", "20-")
 	vc.results = results
@@ -485,30 +441,24 @@ func TestUpdateInChangedVindex(t *testing.T) {
 			OwnedVindexQuery: "dummy_subquery",
 			KsidVindex:       ks.Vindexes["hash"].(vindexes.SingleColumn),
 		},
-		ChangedVindexValues: map[string]*VindexValues{
+		ChangedVindexValues: map[string]VindexValues{
 			"twocol": {
-				PvMap: map[string]sqltypes.PlanValue{
-					"c1": {Value: sqltypes.NewInt64(1)},
-					"c2": {Value: sqltypes.NewInt64(2)},
-				},
-				Offset: 4,
+				"c1": {Value: sqltypes.NewInt64(1)},
+				"c2": {Value: sqltypes.NewInt64(2)},
 			},
 			"onecol": {
-				PvMap: map[string]sqltypes.PlanValue{
-					"c3": {Value: sqltypes.NewInt64(3)},
-				},
-				Offset: 5,
+				"c3": {Value: sqltypes.NewInt64(3)},
 			},
 		},
 	}
 
 	results := []*sqltypes.Result{sqltypes.MakeTestResult(
 		sqltypes.MakeTestFields(
-			"id|c1|c2|c3|twocol|onecol",
-			"int64|int64|int64|int64|int64|int64",
+			"id|c1|c2|c3",
+			"int64|int64|int64|int64",
 		),
-		"1|4|5|6|0|0",
-		"2|21|22|23|0|0",
+		"1|4|5|6",
+		"2|21|22|23",
 	)}
 	vc := newDMLTestVCursor("-20", "20-")
 	vc.results = results
@@ -551,15 +501,15 @@ func TestUpdateInChangedVindex(t *testing.T) {
 		`ExecuteMultiShard sharded.-20: dummy_update {} true true`,
 	})
 
-	// multiple rows changing.
+	// Failure case: multiple rows changing.
 	results = []*sqltypes.Result{sqltypes.MakeTestResult(
 		sqltypes.MakeTestFields(
-			"id|c1|c2|c3|twocol|onecol",
-			"int64|int64|int64|int64|int64|int64",
+			"id|c1|c2|c3",
+			"int64|int64|int64|int64",
 		),
-		"1|4|5|6|0|0",
-		"1|7|8|9|0|0",
-		"2|21|22|23|0|0",
+		"1|4|5|6",
+		"1|7|8|9",
+		"2|21|22|23",
 	)}
 	vc = newDMLTestVCursor("-20", "20-")
 	vc.results = results
diff --git a/go/vt/vtgate/executor_dml_test.go b/go/vt/vtgate/executor_dml_test.go
index 6505411802..b5ea0d2964 100644
--- a/go/vt/vtgate/executor_dml_test.go
+++ b/go/vt/vtgate/executor_dml_test.go
@@ -101,8 +101,8 @@ func TestUpdateEqual(t *testing.T) {
 	sbc2.Queries = nil
 	sbclookup.Queries = nil
 	sbc1.SetResults([]*sqltypes.Result{sqltypes.MakeTestResult(
-		sqltypes.MakeTestFields("id|name|lastname|name_lastname_keyspace_id_map", "int64|int32|varchar|int64"),
-		"1|1|foo|0",
+		sqltypes.MakeTestFields("id|name|lastname", "int64|int32|varchar"),
+		"1|1|foo",
 	),
 	})
 
@@ -110,7 +110,7 @@ func TestUpdateEqual(t *testing.T) {
 	require.NoError(t, err)
 	wantQueries = []*querypb.BoundQuery{
 		{
-			Sql:           "select id, name, lastname, name = 'myname' and lastname = 'mylastname' from user2 where id = 1 for update",
+			Sql:           "select id, name, lastname from user2 where id = 1 for update",
 			BindVariables: map[string]*querypb.BindVariable{},
 		},
 		{
@@ -213,8 +213,8 @@ func TestUpdateMultiOwned(t *testing.T) {
 
 	sbc1.SetResults([]*sqltypes.Result{
 		sqltypes.MakeTestResult(
-			sqltypes.MakeTestFields("id|a|b|c|d|e|f|lookup1|lookup3", "int64|int64|int64|int64|int64|int64|int64|int64|int64"),
-			"1|10|20|30|40|50|60|0|0",
+			sqltypes.MakeTestFields("id|a|b|c|d|e|f", "int64|int64|int64|int64|int64|int64|int64"),
+			"1|10|20|30|40|50|60",
 		),
 	})
 	_, err := executorExec(executor, "update user set a=1, b=2, f=4, e=3 where id=1", nil)
@@ -222,7 +222,7 @@ func TestUpdateMultiOwned(t *testing.T) {
 		t.Fatal(err)
 	}
 	wantQueries := []*querypb.BoundQuery{{
-		Sql:           "select id, a, b, c, d, e, f, a = 1 and b = 2, e = 3 and f = 4 from user where id = 1 for update",
+		Sql:           "select id, a, b, c, d, e, f from user where id = 1 for update",
 		BindVariables: map[string]*querypb.BindVariable{},
 	}, {
 		Sql:           "update user set a = 1, b = 2, f = 4, e = 3 where id = 1",
diff --git a/go/vt/vtgate/executor_select_test.go b/go/vt/vtgate/executor_select_test.go
index 40eab7b621..88a2f808c6 100644
--- a/go/vt/vtgate/executor_select_test.go
+++ b/go/vt/vtgate/executor_select_test.go
@@ -418,7 +418,7 @@ func TestRowCount(t *testing.T) {
 	require.NoError(t, err)
 	testRowCount(t, executor, -1)
 
-	_, err = executorExec(executor, "delete from user where id in (42, 24)", map[string]*querypb.BindVariable{})
+	_, err = executorExec(executor, "update user set name = 'abc' where id in (42, 24)", map[string]*querypb.BindVariable{})
 	require.NoError(t, err)
 	testRowCount(t, executor, 2)
 }
diff --git a/go/vt/vtgate/planbuilder/testdata/dml_cases.txt b/go/vt/vtgate/planbuilder/testdata/dml_cases.txt
index 67a5bc0b26..4531dc693c 100644
--- a/go/vt/vtgate/planbuilder/testdata/dml_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/dml_cases.txt
@@ -279,11 +279,11 @@
     },
     "TargetTabletType": "MASTER",
     "ChangedVindexValues": [
-      "email_user_map:3"
+      "email_user_map"
     ],
     "KsidVindex": "user_index",
     "MultiShardAutocommit": false,
-    "OwnedVindexQuery": "select user_id, email, address, email = 'juan@vitess.io' from user_metadata where user_id = 1 for update",
+    "OwnedVindexQuery": "select user_id, email, address from user_metadata where user_id = 1 for update",
     "Query": "update user_metadata set email = 'juan@vitess.io' where user_id = 1",
     "Table": "user_metadata",
     "Values": [
@@ -311,12 +311,12 @@
     },
     "TargetTabletType": "MASTER",
     "ChangedVindexValues": [
-      "address_user_map:4",
-      "email_user_map:3"
+      "address_user_map",
+      "email_user_map"
     ],
     "KsidVindex": "user_index",
     "MultiShardAutocommit": false,
-    "OwnedVindexQuery": "select user_id, email, address, email = 'juan@vitess.io', address = '155 5th street' from user_metadata where user_id = 1 for update",
+    "OwnedVindexQuery": "select user_id, email, address from user_metadata where user_id = 1 for update",
     "Query": "update user_metadata set email = 'juan@vitess.io', address = '155 5th street' where user_id = 1",
     "Table": "user_metadata",
     "Values": [
@@ -340,11 +340,11 @@
     },
     "TargetTabletType": "MASTER",
     "ChangedVindexValues": [
-      "email_user_map:3"
+      "email_user_map"
     ],
     "KsidVindex": "user_index",
     "MultiShardAutocommit": false,
-    "OwnedVindexQuery": "select user_id, email, address, email = 'juan@vitess.io' from user_metadata where user_id = 1 order by user_id asc limit 10 for update",
+    "OwnedVindexQuery": "select user_id, email, address from user_metadata where user_id = 1 order by user_id asc limit 10 for update",
     "Query": "update user_metadata set email = 'juan@vitess.io' where user_id = 1 order by user_id asc limit 10",
     "Table": "user_metadata",
     "Values": [
@@ -1412,11 +1412,11 @@
     },
     "TargetTabletType": "MASTER",
     "ChangedVindexValues": [
-      "colb_colc_map:4"
+      "colb_colc_map"
     ],
     "KsidVindex": "kid_index",
     "MultiShardAutocommit": false,
-    "OwnedVindexQuery": "select kid, column_a, column_b, column_c, column_b = 1 and column_c = 2 from multicolvin where kid = 1 for update",
+    "OwnedVindexQuery": "select kid, column_a, column_b, column_c from multicolvin where kid = 1 for update",
     "Query": "update multicolvin set column_b = 1, column_c = 2 where kid = 1",
     "Table": "multicolvin",
     "Values": [
@@ -1440,12 +1440,12 @@
     },
     "TargetTabletType": "MASTER",
     "ChangedVindexValues": [
-      "cola_map:4",
-      "colb_colc_map:5"
+      "cola_map",
+      "colb_colc_map"
     ],
     "KsidVindex": "kid_index",
     "MultiShardAutocommit": false,
-    "OwnedVindexQuery": "select kid, column_a, column_b, column_c, column_a = 0, column_b = 1 and column_c = 2 from multicolvin where kid = 1 for update",
+    "OwnedVindexQuery": "select kid, column_a, column_b, column_c from multicolvin where kid = 1 for update",
     "Query": "update multicolvin set column_a = 0, column_b = 1, column_c = 2 where kid = 1",
     "Table": "multicolvin",
     "Values": [
@@ -1825,11 +1825,11 @@
     },
     "TargetTabletType": "MASTER",
     "ChangedVindexValues": [
-      "name_user_map:3"
+      "name_user_map"
     ],
     "KsidVindex": "user_index",
     "MultiShardAutocommit": false,
-    "OwnedVindexQuery": "select Id, Name, Costly, name = null from user where id = 1 for update",
+    "OwnedVindexQuery": "select Id, Name, Costly from user where id = 1 for update",
     "Query": "update user set name = null where id = 1",
     "Table": "user",
     "Values": [
@@ -1872,11 +1872,11 @@
     },
     "TargetTabletType": "MASTER",
     "ChangedVindexValues": [
-      "name_user_map:3"
+      "name_user_map"
     ],
     "KsidVindex": "user_index",
     "MultiShardAutocommit": false,
-    "OwnedVindexQuery": "select Id, Name, Costly, name = null from user where id in (1, 2, 3) for update",
+    "OwnedVindexQuery": "select Id, Name, Costly from user where id in (1, 2, 3) for update",
     "Query": "update user set name = null where id in (1, 2, 3)",
     "Table": "user",
     "Values": [
@@ -1904,11 +1904,11 @@
     },
     "TargetTabletType": "MASTER",
     "ChangedVindexValues": [
-      "name_user_map:3"
+      "name_user_map"
     ],
     "KsidVindex": "user_index",
     "MultiShardAutocommit": false,
-    "OwnedVindexQuery": "select Id, Name, Costly, name = null from user for update",
+    "OwnedVindexQuery": "select Id, Name, Costly from user for update",
     "Query": "update user set name = null",
     "Table": "user"
   }
@@ -1928,11 +1928,11 @@
     },
     "TargetTabletType": "MASTER",
     "ChangedVindexValues": [
-      "name_user_map:3"
+      "name_user_map"
     ],
     "KsidVindex": "user_index",
     "MultiShardAutocommit": false,
-    "OwnedVindexQuery": "select Id, Name, Costly, name = null from user where id + 1 = 2 for update",
+    "OwnedVindexQuery": "select Id, Name, Costly from user where id + 1 = 2 for update",
     "Query": "update user set name = null where id + 1 = 2",
     "Table": "user"
   }
@@ -2088,11 +2088,11 @@
     },
     "TargetTabletType": "MASTER",
     "ChangedVindexValues": [
-      "colb_colc_map:4"
+      "colb_colc_map"
     ],
     "KsidVindex": "kid_index",
     "MultiShardAutocommit": false,
-    "OwnedVindexQuery": "select kid, column_a, column_b, column_c, column_c = 2 from multicolvin where kid = 1 for update",
+    "OwnedVindexQuery": "select kid, column_a, column_b, column_c from multicolvin where kid = 1 for update",
     "Query": "update multicolvin set column_c = 2 where kid = 1",
     "Table": "multicolvin",
     "Values": [
@@ -2116,11 +2116,11 @@
     },
     "TargetTabletType": "MASTER",
     "ChangedVindexValues": [
-      "name_user_map:3"
+      "name_user_map"
     ],
     "KsidVindex": "user_index",
     "MultiShardAutocommit": false,
-    "OwnedVindexQuery": "select Id, Name, Costly, name = _binary 'abc' from user where id = 1 for update",
+    "OwnedVindexQuery": "select Id, Name, Costly from user where id = 1 for update",
     "Query": "update user set name = _binary 'abc' where id = 1",
     "Table": "user",
     "Values": [
@@ -2186,11 +2186,11 @@
     },
     "TargetTabletType": "MASTER",
     "ChangedVindexValues": [
-      "name_user_map:3"
+      "name_user_map"
     ],
     "KsidVindex": "user_index",
     "MultiShardAutocommit": false,
-    "OwnedVindexQuery": "select Id, Name, Costly, name = 'myname' from user for update",
+    "OwnedVindexQuery": "select Id, Name, Costly from user for update",
     "Query": "update user set name = 'myname'",
     "Table": "user"
   }
diff --git a/go/vt/vttablet/tabletmanager/tm_state_test.go b/go/vt/vttablet/tabletmanager/tm_state_test.go
index 4de8c87d55..680fa41e08 100644
--- a/go/vt/vttablet/tabletmanager/tm_state_test.go
+++ b/go/vt/vttablet/tabletmanager/tm_state_test.go
@@ -176,14 +176,14 @@ func TestStateChangeTabletType(t *testing.T) {
 		Uid:  2,
 	}
 
-	err := tm.tmState.ChangeTabletType(ctx, topodatapb.TabletType_MASTER, DBActionSetReadWrite)
+	err := tm.tmState.ChangeTabletType(ctx, topodatapb.TabletType_MASTER)
 	require.NoError(t, err)
 	ti, err := ts.GetTablet(ctx, alias)
 	require.NoError(t, err)
 	assert.Equal(t, topodatapb.TabletType_MASTER, ti.Type)
 	assert.NotNil(t, ti.MasterTermStartTime)
 
-	err = tm.tmState.ChangeTabletType(ctx, topodatapb.TabletType_REPLICA, DBActionNone)
+	err = tm.tmState.ChangeTabletType(ctx, topodatapb.TabletType_REPLICA)
 	require.NoError(t, err)
 	ti, err = ts.GetTablet(ctx, alias)
 	require.NoError(t, err)
diff --git a/go/vt/vttablet/tabletserver/vstreamer/uvstreamer_test.go b/go/vt/vttablet/tabletserver/vstreamer/uvstreamer_test.go
index c9cf44a6c5..1d3f3e7e04 100644
--- a/go/vt/vttablet/tabletserver/vstreamer/uvstreamer_test.go
+++ b/go/vt/vttablet/tabletserver/vstreamer/uvstreamer_test.go
@@ -308,8 +308,6 @@ func resetMetrics(t *testing.T) {
 	engine.resultStreamerNumRows.Reset()
 	engine.rowStreamerNumRows.Reset()
 	engine.vstreamerPhaseTimings.Reset()
-	engine.vstreamerPhaseTimings.Reset()
-	engine.vstreamerPhaseTimings.Reset()
 }
 
 func validateMetrics(t *testing.T) {