copy over existing vreplication rows copied to local counter if resuming from another tablet

go/vt/vttablet/tabletmanager/vreplication/framework_test.go

@@ -539,6 +539,7 @@ func shouldIgnoreQuery(query string) bool {
 		", time_throttled=",      // update of last throttle time, can happen out-of-band, so can't test for it
 		", component_throttled=", // update of last throttle time, can happen out-of-band, so can't test for it
 		"context cancel",
+		"SELECT rows_copied FROM _vt.vreplication",
 	}
 	if sidecardb.MatchesInitQuery(query) {
 		return true

go/vt/vttablet/tabletmanager/vreplication/vreplicator.go

@@ -142,7 +142,7 @@ func newVReplicator(id int32, source *binlogdatapb.BinlogSource, sourceVStreamer
 		log.Warningf("The supplied value for vreplication_heartbeat_update_interval:%d seconds is larger than the maximum allowed:%d seconds, vreplication will fallback to %d",
 			vreplicationHeartbeatUpdateInterval, vreplicationMinimumHeartbeatUpdateInterval, vreplicationMinimumHeartbeatUpdateInterval)
 	}
-	return &vreplicator{
+	v := &vreplicator{
 		vre:             vre,
 		id:              id,
 		source:          source,
@@ -151,6 +151,8 @@ func newVReplicator(id int32, source *binlogdatapb.BinlogSource, sourceVStreamer
 		dbClient:        newVDBClient(dbClient, stats),
 		mysqld:          mysqld,
 	}
+	v.readExistingRowsCopiedAndUpdateSelfIfNeeded()
+	return v
 }
 
 // Replicate starts a vreplication stream. It can be in one of three phases:
@@ -1030,3 +1032,36 @@ func (vr *vreplicator) newClientConnection(ctx context.Context) (*vdbClient, err
 	}
 	return dbClient, nil
 }
+
+func (vr *vreplicator) readExistingRowsCopiedAndUpdateSelfIfNeeded() {
+	if vr.stats.CopyRowCount.Get() == 0 {
+		rowsCopiedExisting, err := vr.readExistingRowsCopied(vr.id)
+		if err != nil {
+			log.Warningf("failed to read existing rows copied: %v", err)
+		} else if rowsCopiedExisting != 0 {
+			log.Infof("resuming vreplication from another tablet, setting rows copied counter to %v", rowsCopiedExisting)
+			vr.stats.CopyRowCount.Set(rowsCopiedExisting)
+		}
+	}
+}
+
+func (vr *vreplicator) readExistingRowsCopied(uid int32) (int64, error) {
+	query, err := sqlparser.ParseAndBind(`SELECT rows_copied FROM _vt.vreplication WHERE id=%a`,
+		sqltypes.Int32BindVariable(uid),
+	)
+	if err != nil {
+		return 0, err
+	}
+	r, err := vr.dbClient.Execute(query)
+	if err != nil {
+		return 0, err
+	}
+	if len(r.Rows) == 0 {
+		return 0, nil
+	}
+	row := r.Named().Row()
+	if row == nil {
+		return 0, vterrors.Errorf(vtrpcpb.Code_UNKNOWN, "Cannot find unique workflow for UUID: %+v", uid)
+	}
+	return row.AsInt64("rows_copied", 0), nil
+}

go/vt/vttablet/tabletmanager/vreplication/vreplicator_test.go

@@ -628,6 +628,53 @@ func TestCancelledDeferSecondaryKeys(t *testing.T) {
 	require.Equal(t, 1, len(res.Rows))
 }
 
+// TestResumingFromPreviousWorkflowKeepingRowsCopied tests that when you
+// resume a workflow started by another tablet (eg. a reparent occurred), the
+// rows_copied does not reset to zero but continues along from where it left off
+func TestResumingFromPreviousWorkflowKeepingRowsCopied(t *testing.T) {
+	_, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	tablet := addTablet(100)
+	defer deleteTablet(tablet)
+	filter := &binlogdatapb.Filter{
+		Rules: []*binlogdatapb.Rule{{
+			Match: "t1",
+		}},
+	}
+	bls := &binlogdatapb.BinlogSource{
+		Keyspace: env.KeyspaceName,
+		Shard:    env.ShardName,
+		Filter:   filter,
+	}
+	// The test env uses the same factory for both dba and
+	// filtered connections.
+	dbconfigs.GlobalDBConfigs.Filtered.User = "vt_dba"
+	id := int32(1)
+	vsclient := newTabletConnector(tablet)
+	stats := binlogplayer.NewStats()
+	defer stats.Stop()
+	dbaconn := playerEngine.dbClientFactoryDba()
+	err := dbaconn.Connect()
+	require.NoError(t, err)
+	defer dbaconn.Close()
+	dbClient := playerEngine.dbClientFactoryFiltered()
+	err = dbClient.Connect()
+	require.NoError(t, err)
+	defer dbClient.Close()
+	dbName := dbClient.DBName()
+	rowsCopied := int64(500000)
+	// Ensure there's an existing vreplication workflow
+	_, err = dbClient.ExecuteFetch(fmt.Sprintf("insert into _vt.vreplication (id, workflow, source, pos, max_tps, max_replication_lag, time_updated, transaction_timestamp, state, db_name, rows_copied) values (%d, 'test', '', '', 99999, 99999, 0, 0, 'Running', '%s', %v) on duplicate key update workflow='test', source='', pos='', max_tps=99999, max_replication_lag=99999, time_updated=0, transaction_timestamp=0, state='Running', db_name='%s', rows_copied=%v",
+		id, dbName, rowsCopied, dbName, rowsCopied), 1)
+	require.NoError(t, err)
+	defer func() {
+		_, err = dbClient.ExecuteFetch(fmt.Sprintf("delete from _vt.vreplication where id = %d", id), 1)
+		require.NoError(t, err)
+	}()
+	vr := newVReplicator(id, bls, vsclient, stats, dbClient, env.Mysqld, playerEngine)
+	assert.Equal(t, rowsCopied, vr.stats.CopyRowCount.Get())
+}
+
 // stripCruft removes all whitespace unicode chars and backticks.
 func stripCruft(in string) string {
 	out := strings.Builder{}