Skip to content

Commit

Permalink
cherry pick pingcap#420 to release-3.1
Browse files Browse the repository at this point in the history
Signed-off-by: ti-srebot <ti-srebot@pingcap.com>
  • Loading branch information
YuJuncen authored and ti-srebot committed Jan 26, 2021
1 parent c2d8e2d commit 46b58af
Show file tree
Hide file tree
Showing 3 changed files with 84 additions and 0 deletions.
12 changes: 12 additions & 0 deletions pkg/backup/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -237,7 +237,19 @@ func BuildBackupRangeAndSchema(
// Skip tables other than the given table.
continue
}
<<<<<<< HEAD
globalAutoID, err := idAlloc.NextGlobalAutoID(tableInfo.ID)
=======
var globalAutoID int64
switch {
case tableInfo.IsSequence():
globalAutoID, err = seqAlloc.NextGlobalAutoID(tableInfo.ID)
case tableInfo.IsView() || !utils.NeedAutoID(tableInfo):
// no auto ID for views or table without either rowID nor auto_increment ID.
default:
globalAutoID, err = idAlloc.NextGlobalAutoID(tableInfo.ID)
}
>>>>>>> 6390453... restore: don't restore auto id if the table doesn't has it. (#420)
if err != nil {
return nil, nil, errors.Trace(err)
}
Expand Down
65 changes: 65 additions & 0 deletions pkg/restore/db.go
Original file line number Diff line number Diff line change
Expand Up @@ -112,7 +112,72 @@ func (db *DB) CreateTable(ctx context.Context, table *utils.Table) error {
utils.EncloseName(table.Info.Name.O),
table.Info.AutoIncID)

<<<<<<< HEAD
err = db.se.Execute(ctx, alterAutoIncIDSQL)
=======
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))
var setValSQL string
if increment < 0 {
setValSQL = fmt.Sprintf(setValFormat, table.Info.Sequence.MinValue)
} else {
setValSQL = fmt.Sprintf(setValFormat, table.Info.Sequence.MaxValue)
}
err = db.se.Execute(ctx, setValSQL)
if err != nil {
log.Error("restore meta sql failed",
zap.String("query", setValSQL),
zap.Stringer("db", table.Db.Name),
zap.Stringer("table", table.Info.Name),
zap.Error(err))
return errors.Trace(err)
}

// trigger cycle round > 0
err = db.se.Execute(ctx, nextSeqSQL)
if err != nil {
log.Error("restore meta sql failed",
zap.String("query", nextSeqSQL),
zap.Stringer("db", table.Db.Name),
zap.Stringer("table", table.Info.Name),
zap.Error(err))
return errors.Trace(err)
}
}
restoreMetaSQL = fmt.Sprintf(setValFormat, table.Info.AutoIncID)
err = db.se.Execute(ctx, restoreMetaSQL)
} 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)
if utils.NeedAutoID(table.Info) {
err = db.se.Execute(ctx, restoreMetaSQL)
}
}

>>>>>>> 6390453... restore: don't restore auto id if the table doesn't has it. (#420)
if err != nil {
log.Error("alter AutoIncID failed",
zap.String("query", alterAutoIncIDSQL),
Expand Down
7 changes: 7 additions & 0 deletions pkg/utils/schema.go
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,13 @@ func (tbl *Table) NoChecksum() bool {
return tbl.Crc64Xor == 0 && tbl.TotalKvs == 0 && tbl.TotalBytes == 0
}

// NeedAutoID checks whether the table needs backing up with an autoid.
func NeedAutoID(tblInfo *model.TableInfo) bool {
hasRowID := !tblInfo.PKIsHandle && !tblInfo.IsCommonHandle
hasAutoIncID := tblInfo.GetAutoIncrementColInfo() != nil
return hasRowID || hasAutoIncID
}

// Database wraps the schema and tables of a database.
type Database struct {
Info *model.DBInfo
Expand Down

0 comments on commit 46b58af

Please sign in to comment.