Port over changes from vitessio#13238

Signed-off-by: Matt Lord <mattalord@gmail.com>
planetscale · Jul 5, 2023 · ad4f20d · ad4f20d
1 parent a849a54
commit ad4f20d
Show file tree

Hide file tree

Showing 5 changed files with 73 additions and 0 deletions.
diff --git a/go/vt/vtctl/workflow/server.go b/go/vt/vtctl/workflow/server.go
@@ -2473,6 +2473,13 @@ func (s *Server) switchWrites(ctx context.Context, req *vtctldatapb.WorkflowSwit
 			return 0, nil, err
 		}
 
+		ts.Logger().Infof("Resetting sequences")
+		if err := sw.resetSequences(ctx); err != nil {
+			ts.Logger().Errorf("resetSequences failed: %v", err)
+			sw.cancelMigration(ctx, sm)
+			return 0, nil, err
+		}
+
 		ts.Logger().Infof("Creating reverse streams")
 		if err := sw.createReverseVReplication(ctx); err != nil {
 			ts.Logger().Errorf("createReverseVReplication failed: %v", err)

diff --git a/go/vt/vtctl/workflow/switcher.go b/go/vt/vtctl/workflow/switcher.go
@@ -137,3 +137,7 @@ func (r *switcher) dropTargetShards(ctx context.Context) error {
 func (r *switcher) logs() *[]string {
 	return nil
 }
+
+func (r *switcher) resetSequences(ctx context.Context) error {
+	return r.ts.resetSequences(ctx)
+}
diff --git a/go/vt/vtctl/workflow/switcher_dry_run.go b/go/vt/vtctl/workflow/switcher_dry_run.go
@@ -369,3 +369,16 @@ func (dr *switcherDryRun) dropTargetShards(ctx context.Context) error {
 
 	return nil
 }
+
+func (dr *switcherDryRun) resetSequences(ctx context.Context) error {
+	var err error
+	mustReset := false
+	if mustReset, err = dr.ts.mustResetSequences(ctx); err != nil {
+		return err
+	}
+	if !mustReset {
+		return nil
+	}
+	dr.drLog.Log("The sequence caches will be reset on the source since sequence tables are being moved")
+	return nil
+}
diff --git a/go/vt/vtctl/workflow/switcher_interface.go b/go/vt/vtctl/workflow/switcher_interface.go
@@ -51,4 +51,5 @@ type iswitcher interface {
 	deleteShardRoutingRules(ctx context.Context) error
 	addParticipatingTablesToKeyspace(ctx context.Context, keyspace, tableSpecs string) error
 	logs() *[]string
+	resetSequences(ctx context.Context) error
 }
diff --git a/go/vt/vtctl/workflow/traffic_switcher.go b/go/vt/vtctl/workflow/traffic_switcher.go
@@ -1169,3 +1169,51 @@ func (ts *trafficSwitcher) gatherPositions(ctx context.Context) error {
 		return err
 	})
 }
+
+func (ts *trafficSwitcher) isSequenceParticipating(ctx context.Context) (bool, error) {
+	vschema, err := ts.TopoServer().GetVSchema(ctx, ts.targetKeyspace)
+	if err != nil {
+		return false, err
+	}
+	if vschema == nil || vschema.Tables == nil {
+		return false, nil
+	}
+	sequenceFound := false
+	for _, table := range ts.Tables() {
+		vs, ok := vschema.Tables[table]
+		if !ok || vs == nil {
+			continue
+		}
+		if vs.Type == vindexes.TypeSequence {
+			sequenceFound = true
+			break
+		}
+	}
+	return sequenceFound, nil
+}
+
+func (ts *trafficSwitcher) mustResetSequences(ctx context.Context) (bool, error) {
+	switch ts.workflowType {
+	case binlogdatapb.VReplicationWorkflowType_Migrate,
+		binlogdatapb.VReplicationWorkflowType_MoveTables:
+		return ts.isSequenceParticipating(ctx)
+	default:
+		return false, nil
+	}
+}
+
+func (ts *trafficSwitcher) resetSequences(ctx context.Context) error {
+	var err error
+	mustReset := false
+	if mustReset, err = ts.mustResetSequences(ctx); err != nil {
+		return err
+	}
+	if !mustReset {
+		return nil
+	}
+	return ts.ForAllSources(func(source *MigrationSource) error {
+		ts.Logger().Infof("Resetting sequences for source shard %s.%s on tablet %s",
+			source.GetShard().Keyspace(), source.GetShard().ShardName(), source.GetPrimary().String())
+		return ts.TabletManagerClient().ResetSequences(ctx, source.GetPrimary().Tablet, ts.Tables())
+	})
+}