Merge pull request #1 from 3pointer/sequence_cycle

trigger sequence cycle round
kennytm · May 25, 2020 · 07778c7 · 07778c7
2 parents c75e712 + ca4b3ff
commit 07778c7
Showing 1 changed file with 42 additions and 16 deletions.
diff --git a/pkg/restore/db.go b/pkg/restore/db.go
@@ -107,25 +107,51 @@ func (db *DB) CreateTable(ctx context.Context, table *utils.Table) error {
 		return errors.Trace(err)
 	}
 
-	var alterAutoIncIDFormat string
-	switch {
-	case table.Info.IsSequence():
-		alterAutoIncIDFormat = "do setval(%s.%s, %d);"
-	case table.Info.IsView():
-		return nil
-	default:
-		alterAutoIncIDFormat = "alter table %s.%s auto_increment = %d;"
+	var restoreMetaSQL string
+	if table.Info.IsSequence() {
+		setValFormat := fmt.Sprintf("do setval(%s.%s, %%d);",
+			utils.EncloseName(table.Db.Name.O),
+			utils.EncloseName(table.Info.Name.O))
+		if table.Info.Sequence.Cycle {
+			increment := table.Info.Sequence.Increment
+			// TiDB sequence's behaviour is designed to keep the same pace
+			// among all nodes within the same cluster. so we need restore round.
+			// Here is a hack way to trigger sequence cycle round > 0 according to
+			// https://github.com/pingcap/br/pull/242#issuecomment-631307978
+			// TODO use sql to set cycle round
+			nextSeqSQL := fmt.Sprintf("do nextval(%s.%s);",
+				utils.EncloseName(table.Db.Name.O),
+				utils.EncloseName(table.Info.Name.O))
+			if increment < 0 {
+				restoreMetaSQL += fmt.Sprintf(setValFormat, table.Info.Sequence.MinValue)
+			} else {
+				restoreMetaSQL += fmt.Sprintf(setValFormat, table.Info.Sequence.MaxValue)
+			}
+			// trigger cycle round > 0
+			restoreMetaSQL += nextSeqSQL
+			restoreMetaSQL += fmt.Sprintf(setValFormat, table.Info.AutoIncID)
+		} else {
+			restoreMetaSQL = fmt.Sprintf(setValFormat, table.Info.AutoIncID)
+		}
+	} else {
+		var alterAutoIncIDFormat string
+		switch {
+		case table.Info.IsView():
+			return nil
+		default:
+			alterAutoIncIDFormat = "alter table %s.%s auto_increment = %d;"
+		}
+		restoreMetaSQL = fmt.Sprintf(
+			alterAutoIncIDFormat,
+			utils.EncloseName(table.Db.Name.O),
+			utils.EncloseName(table.Info.Name.O),
+			table.Info.AutoIncID)
 	}
-	alterAutoIncIDSQL := fmt.Sprintf(
-		alterAutoIncIDFormat,
-		utils.EncloseName(table.Db.Name.O),
-		utils.EncloseName(table.Info.Name.O),
-		table.Info.AutoIncID)
 
-	err = db.se.Execute(ctx, alterAutoIncIDSQL)
+	err = db.se.Execute(ctx, restoreMetaSQL)
 	if err != nil {
-		log.Error("alter AutoIncID failed",
-			zap.String("query", alterAutoIncIDSQL),
+		log.Error("restore meta sql failed",
+			zap.String("query", restoreMetaSQL),
 			zap.Stringer("db", table.Db.Name),
 			zap.Stringer("table", table.Info.Name),
 			zap.Error(err))