[Enhancement] add support for restore to ccr

pkg/ccr/job.go

@@ -2609,6 +2609,46 @@ func (j *Job) handleRecoverInfoRecord(commitSeq int64, recoverInfo *record.Recov
 	return j.newPartialSnapshot(recoverInfo.TableId, recoverInfo.TableName, nil, true)
 }
 
+func (j *Job) handleRestoreInfo(binlog *festruct.TBinlog) error {
+	log.Infof("handle restore info binlog, prevCommitSeq: %d, commitSeq: %d",
+		j.progress.PrevCommitSeq, j.progress.CommitSeq)
+
+	data := binlog.GetData()
+	restoreInfo, err := record.NewRestoreInfoFromJson(data)
+	if err != nil {
+		return err
+	}
+	return j.handleRestoreInfoRecord(binlog.GetCommitSeq(), restoreInfo)
+}
+
+func (j *Job) handleRestoreInfoRecord(commitSeq int64, restoreInfo *record.RestoreInfo) error {
+	if len(restoreInfo.TableInfo) != 1 {
+		// for both table and db sync take a full snapshot.
+		log.Warnf("Lets do new snapshot")
+		return j.newSnapshot(commitSeq)
+	}
+
+	if len(restoreInfo.TableInfo) == 1 {
+		for tableId, tableName := range restoreInfo.TableInfo {
+			switch j.SyncType {
+			case TableSync:
+				log.Warnf("full snapshot, table:%d and name:%s",
+					tableId, tableName)
+				return j.newSnapshot(commitSeq)
+			case DBSync:
+				log.Warnf("new partial snapshot, table:%d and name:%s",
+					tableId, tableName)
+				replace := true // replace the old data to avoid blocking reading
+				return j.newPartialSnapshot(tableId, tableName, nil, replace)
+			default:
+				break
+			}
+		}
+	}
+	//This is unreachable.
+	return nil
+}
+
 func (j *Job) handleBarrier(binlog *festruct.TBinlog) error {
 	data := binlog.GetData()
 	barrierLog, err := record.NewBarrierLogFromJson(data)
@@ -2693,6 +2733,12 @@ func (j *Job) handleBarrier(binlog *festruct.TBinlog) error {
 			return err
 		}
 		return j.handleRecoverInfoRecord(commitSeq, recoverInfo)
+	case festruct.TBinlogType_RESTORE_INFO:
+		restoreInfo, err := record.NewRestoreInfoFromJson(barrierLog.Binlog)
+		if err != nil {
+			return err
+		}
+		return j.handleRestoreInfoRecord(commitSeq, restoreInfo)
 	case festruct.TBinlogType_BARRIER:
 		log.Info("handle barrier binlog, ignore it")
 	default:
@@ -2807,6 +2853,8 @@ func (j *Job) handleBinlog(binlog *festruct.TBinlog) error {
 		return j.handleDropRollup(binlog)
 	case festruct.TBinlogType_RECOVER_INFO:
 		return j.handleRecoverInfo(binlog)
+	case festruct.TBinlogType_RESTORE_INFO:
+		return j.handleRestoreInfo(binlog)
 	default:
 		return xerror.Errorf(xerror.Normal, "unknown binlog type: %v", binlog.GetType())
 	}

pkg/ccr/record/restore_info.go

@@ -0,0 +1,26 @@
+package record
+
+import (
+	"encoding/json"
+
+	"github.com/selectdb/ccr_syncer/pkg/xerror"
+)
+
+type RestoreInfo struct {
+	DbId      int64            `json:"dbId"`
+	DbName    string           `json:"dbName"`
+	TableInfo map[int64]string `json:"tableInfo"`
+}
+
+func NewRestoreInfoFromJson(data string) (*RestoreInfo, error) {
+	var restoreInfo RestoreInfo
+	err := json.Unmarshal([]byte(data), &restoreInfo)
+	if err != nil {
+		return nil, xerror.Wrap(err, xerror.Normal, "unmarshal create table error")
+	}
+
+	if restoreInfo.DbId == 0 {
+		return nil, xerror.Errorf(xerror.Normal, "db id not found")
+	}
+	return &restoreInfo, nil
+}

pkg/rpc/thrift/FrontendService.thrift

@@ -1198,6 +1198,7 @@ enum TBinlogType {
   RENAME_PARTITION = 22,
   DROP_ROLLUP = 23,
   RECOVER_INFO = 24,
+  RESTORE_INFO = 25,
   // Keep some IDs for allocation so that when new binlog types are added in the
   // future, the changes can be picked back to the old versions without breaking
   // compatibility.
@@ -1213,8 +1214,7 @@ enum TBinlogType {
   //    MODIFY_XXX = 17,
   //    MIN_UNKNOWN = 18,
   //    UNKNOWN_3 = 19,
-  MIN_UNKNOWN = 25,
-  UNKNOWN_10 = 26,
+  MIN_UNKNOWN = 26,
   UNKNOWN_11 = 27,
   UNKNOWN_12 = 28,
   UNKNOWN_13 = 29,

regression-test/data/db_sync/restore/test_db_sync_table_restore.out

@@ -0,0 +1,25 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql_source_content --
+0	0
+0	1
+0	2
+0	3
+0	4
+0	5
+0	6
+0	7
+0	8
+0	9
+
+-- !target_sql_content --
+0	0
+0	1
+0	2
+0	3
+0	4
+0	5
+0	6
+0	7
+0	8
+0	9
+

regression-test/data/db_sync/restore_1/test_db_sync_table_restore1.out

@@ -0,0 +1,37 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql_source_content_backup --
+0	0
+0	1
+0	2
+
+-- !target_sql_content_backup --
+0	0
+0	1
+0	2
+
+-- !sql_source_content_new --
+0	0
+0	1
+0	2
+9	0
+9	1
+9	2
+
+-- !target_sql_content_new --
+0	0
+0	1
+0	2
+9	0
+9	1
+9	2
+
+-- !sql_source_content_restore --
+0	0
+0	1
+0	2
+
+-- !target_sql_content_restore --
+0	0
+0	1
+0	2
+

regression-test/data/db_sync/restore_multi/test_db_sync_table_restore_multi.out

@@ -0,0 +1,49 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql_source_content --
+0	0
+0	1
+0	2
+0	3
+0	4
+0	5
+0	6
+0	7
+0	8
+0	9
+
+-- !target_sql_content --
+0	0
+0	1
+0	2
+0	3
+0	4
+0	5
+0	6
+0	7
+0	8
+0	9
+
+-- !sql_source_content --
+0	0
+0	1
+0	2
+0	3
+0	4
+0	5
+0	6
+0	7
+0	8
+0	9
+
+-- !target_sql_content --
+0	0
+0	1
+0	2
+0	3
+0	4
+0	5
+0	6
+0	7
+0	8
+0	9
+

regression-test/data/table_sync/restore/test_tbl_restore.out

@@ -0,0 +1,69 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql_source_content --
+0	0
+0	1
+0	2
+0	3
+0	4
+0	5
+0	6
+0	7
+0	8
+0	9
+1	0
+1	1
+1	2
+1	3
+1	4
+1	5
+1	6
+1	7
+1	8
+1	9
+
+-- !target_sql_content --
+0	0
+0	1
+0	2
+0	3
+0	4
+0	5
+0	6
+0	7
+0	8
+0	9
+1	0
+1	1
+1	2
+1	3
+1	4
+1	5
+1	6
+1	7
+1	8
+1	9
+
+-- !sql_source_content --
+0	0
+0	1
+0	2
+0	3
+0	4
+0	5
+0	6
+0	7
+0	8
+0	9
+
+-- !target_sql_content --
+0	0
+0	1
+0	2
+0	3
+0	4
+0	5
+0	6
+0	7
+0	8
+0	9
+