VReplication: Make Target Sequence Initialization More Robust

go/sqltypes/testing.go

@@ -76,7 +76,7 @@ func MakeTestResult(fields []*querypb.Field, rows ...string) *Result {
 	for i, row := range rows {
 		result.Rows[i] = make([]Value, len(fields))
 		for j, col := range split(row) {
-			if col == "null" {
+			if strings.ToLower(col) == "null" {
 				result.Rows[i][j] = NULL
 				continue
 			}

go/vt/vtctl/workflow/server.go

@@ -3301,6 +3301,20 @@
 			sw.cancelMigration(ctx, sm)
 			return handleError("failed to create the reverse vreplication streams", err)
 		}
+
+		// Initialize any target sequences, if there are any, before allowing new writes.
+		if req.InitializeTargetSequences && len(sequenceMetadata) > 0 {
+			ts.Logger().Infof("Initializing target sequences")
+			// Writes are blocked so we can safely initialize the sequence tables but
+			// we also want to use a shorter timeout than the parent context.
+			// We use up at most half of the overall timeout.
+			initSeqCtx, cancel := context.WithTimeout(ctx, timeout/2)
+			defer cancel()
+			if err := sw.initializeTargetSequences(initSeqCtx, sequenceMetadata); err != nil {
+				sw.cancelMigration(ctx, sm)
+				return handleError(fmt.Sprintf("failed to initialize the sequences used in the %s keyspace", ts.TargetKeyspaceName()), err)
+			}
+		}
 	} else {
 		if cancel {
 			return handleError("invalid cancel", vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "traffic switching has reached the point of no return, cannot cancel"))
@@ -3317,17 +3331,6 @@
 	if err := sw.createJournals(ctx, sourceWorkflows); err != nil {
 		return handleError("failed to create the journal", err)
 	}
-	// Initialize any target sequences, if there are any, before allowing new writes.
-	if req.InitializeTargetSequences && len(sequenceMetadata) > 0 {
-		// Writes are blocked so we can safely initialize the sequence tables but
-		// we also want to use a shorter timeout than the parent context.
-		// We use up at most half of the overall timeout.
-		initSeqCtx, cancel := context.WithTimeout(ctx, timeout/2)
-		defer cancel()
-		if err := sw.initializeTargetSequences(initSeqCtx, sequenceMetadata); err != nil {
-			return handleError(fmt.Sprintf("failed to initialize the sequences used in the %s keyspace", ts.TargetKeyspaceName()), err)
-		}
-	}
 	if err := sw.allowTargetWrites(ctx); err != nil {
 		return handleError(fmt.Sprintf("failed to allow writes in the %s keyspace", ts.TargetKeyspaceName()), err)
 	}

go/vt/vtctl/workflow/traffic_switcher.go

@@ -1432,13 +1432,17 @@
 				MaxRows: 1,
 			})
 			if terr != nil || len(qr.Rows) != 1 {
-				return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "failed to get the max used sequence value for target table %s.%s in order to initialize the backing sequence table: %v",
-					ts.targetKeyspace, sequenceMetadata.usingTableName, terr)
+				return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "failed to get the max used sequence value for target table %s.%s on tablet %v in order to initialize the backing sequence table: %v",
+					ts.targetKeyspace, sequenceMetadata.usingTableName, primary.Alias, terr)
 			}
-			maxID, terr := sqltypes.Proto3ToResult(qr).Rows[0][0].ToInt64()
-			if terr != nil {
-				return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "failed to get the max used sequence value for target table %s.%s in order to initialize the backing sequence table: %v",
-					ts.targetKeyspace, sequenceMetadata.usingTableName, terr)
+			rawVal := sqltypes.Proto3ToResult(qr).Rows[0][0]
+			maxID := int64(0)
+			if !rawVal.IsNull() { // If it's NULL then there are no rows and 0 remains the max
+				maxID, terr = rawVal.ToInt64()
+				if terr != nil {
+					return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "failed to get the max used sequence value for target table %s.%s on tablet %v in order to initialize the backing sequence table: %v",
+						ts.targetKeyspace, sequenceMetadata.usingTableName, primary.Alias, terr)
+				}
 			}
 			srMu.Lock()
 			defer srMu.Unlock()

go/vt/wrangler/traffic_switcher.go

@@ -619,6 +619,20 @@
 			sw.cancelMigration(ctx, sm)
 			return handleError("failed to create the reverse vreplication streams", err)
 		}
+
+		// Initialize any target sequences, if there are any, before allowing new writes.
+		if initializeTargetSequences && len(sequenceMetadata) > 0 {
+			ts.Logger().Infof("Initializing target sequences")
+			// Writes are blocked so we can safely initialize the sequence tables but
+			// we also want to use a shorter timeout than the parent context.
+			// We use up at most half of the overall timeout.
+			initSeqCtx, cancel := context.WithTimeout(ctx, timeout/2)
+			defer cancel()
+			if err := sw.initializeTargetSequences(initSeqCtx, sequenceMetadata); err != nil {
+				sw.cancelMigration(ctx, sm)
+				return handleError(fmt.Sprintf("failed to initialize the sequences used in the %s keyspace", ts.TargetKeyspaceName()), err)
+			}
+		}
 	} else {
 		if cancel {
 			return handleError("invalid cancel", fmt.Errorf("traffic switching has reached the point of no return, cannot cancel"))
@@ -635,17 +649,6 @@
 	if err := sw.createJournals(ctx, sourceWorkflows); err != nil {
 		return handleError("failed to create the journal", err)
 	}
-	// Initialize any target sequences, if there are any, before allowing new writes.
-	if initializeTargetSequences && len(sequenceMetadata) > 0 {
-		// Writes are blocked so we can safely initialize the sequence tables but
-		// we also want to use a shorter timeout than the parent context.
-		// We use up at most half of the overall timeout.
-		initSeqCtx, cancel := context.WithTimeout(ctx, timeout/2)
-		defer cancel()
-		if err := sw.initializeTargetSequences(initSeqCtx, sequenceMetadata); err != nil {
-			return handleError(fmt.Sprintf("failed to initialize the sequences used in the %s keyspace", ts.TargetKeyspaceName()), err)
-		}
-	}
 	if err := sw.allowTargetWrites(ctx); err != nil {
 		return handleError(fmt.Sprintf("failed to allow writes in the %s keyspace", ts.TargetKeyspaceName()), err)
 	}
@@ -2197,13 +2200,17 @@
 			)
 			qr, terr := ts.wr.ExecuteFetchAsApp(ictx, primary.GetAlias(), true, query.Query, 1)
 			if terr != nil || len(qr.Rows) != 1 {
-				return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "failed to get the max used sequence value for target table %s.%s in order to initialize the backing sequence table: %v",
-					ts.targetKeyspace, sequenceMetadata.usingTableName, terr)
+				return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "failed to get the max used sequence value for target table %s.%s on tablet %v in order to initialize the backing sequence table: %v",
+					ts.targetKeyspace, sequenceMetadata.usingTableName, primary.Alias, terr)
 			}
-			maxID, terr := sqltypes.Proto3ToResult(qr).Rows[0][0].ToInt64()
-			if terr != nil {
-				return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "failed to get the max used sequence value for target table %s.%s in order to initialize the backing sequence table: %v",
-					ts.targetKeyspace, sequenceMetadata.usingTableName, terr)
+			rawVal := sqltypes.Proto3ToResult(qr).Rows[0][0]
+			maxID := int64(0)
+			if !rawVal.IsNull() { // If it's NULL then there are no rows and 0 remains the max
+				maxID, terr = rawVal.ToInt64()
+				if terr != nil {
+					return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "failed to get the max used sequence value for target table %s.%s on tablet %v in order to initialize the backing sequence table: %v",
+						ts.targetKeyspace, sequenceMetadata.usingTableName, primary.Alias, terr)
+				}
 			}
 			srMu.Lock()
 			defer srMu.Unlock()

go/vt/wrangler/traffic_switcher_env_test.go

@@ -258,7 +258,7 @@ func newTestTableMigraterCustom(ctx context.Context, t *testing.T, sourceShards,
 					"maxval",
 					"int64",
 				),
-				"5",
+				"NULL",
 			),
 		)
 		tme.tmeDB.AddQuery(fmt.Sprintf(maxValForSequence, "ks2", "t2"),
@@ -274,7 +274,7 @@ func newTestTableMigraterCustom(ctx context.Context, t *testing.T, sourceShards,
 		// Now tell the fakesqldb used by the global keyspace tablets to expect
 		// the sequence management related queries against the target keyspace.
 		gfdb.AddQuery(
-			sqlparser.BuildParsedQuery(sqlInitSequenceTable, sqlescape.EscapeID("vt_global"), sqlescape.EscapeID("t1_seq"), 6, 6, 6).Query,
+			sqlparser.BuildParsedQuery(sqlInitSequenceTable, sqlescape.EscapeID("vt_global"), sqlescape.EscapeID("t1_seq"), 1, 1, 1).Query,
 			&sqltypes.Result{RowsAffected: 0},
 		)
 		gfdb.AddQuery(

go/vt/wrangler/traffic_switcher_test.go

@@ -1011,8 +1011,8 @@ func TestTableMigrateOneToManyDryRun(t *testing.T) {
 		"\tKeyspace ks1, Shard 0 at Position MariaDB/5-456-892",
 		"Wait for VReplication on stopped streams to catchup for up to 1s",
 		"Create reverse replication workflow test_reverse",
-		"Create journal entries on source databases",
 		"The following sequence backing tables used by tables being moved will be initialized: t1_seq,t2_seq",
+		"Create journal entries on source databases",
 		"Enable writes on keyspace ks2 tables [t1,t2]",
 		"Switch routing from keyspace ks1 to keyspace ks2",
 		"Routing rules for tables [t1,t2] will be updated",