br: add log restore ts into snapshot checkpoint metadata (#59678)

ref #57613

br/pkg/checkpoint/restore.go

@@ -137,6 +137,7 @@ func LoadCheckpointChecksumForRestore(
 type CheckpointMetadataForSnapshotRestore struct {
 	UpstreamClusterID uint64                `json:"upstream-cluster-id"`
 	RestoredTS        uint64                `json:"restored-ts"`
+	LogRestoredTS     uint64                `json:"log-restored-ts"`
 	SchedulersConfig  *pdutil.ClusterConfig `json:"schedulers-config"`
 
 	RestoreUUID uuid.UUID `json:"restore-uuid"`

br/pkg/restore/snap_client/client.go

@@ -321,6 +321,7 @@ func (rc *SnapClient) InitCheckpoint(
 	ctx context.Context,
 	g glue.Glue, store kv.Storage,
 	config *pdutil.ClusterConfig,
+	logRestoredTS uint64,
 	checkpointFirstRun bool,
 ) (checkpointSetWithTableID map[int64]map[string]struct{}, checkpointClusterConfig *pdutil.ClusterConfig, err error) {
 	// checkpoint sets distinguished by range key
@@ -352,6 +353,19 @@ func (rc *SnapClient) InitCheckpoint(
 			)
 		}
 
+		// The filter feature is determined by the PITR restored ts, so the snapshot
+		// restore checkpoint should check whether the PITR restored ts is changed.
+		// Notice that if log restore checkpoint metadata is not stored, BR always enters
+		// snapshot restore.
+		if meta.LogRestoredTS != logRestoredTS {
+			return checkpointSetWithTableID, nil, errors.Errorf(
+				"The current PITR want to restore cluster to the log restored ts[%d], which is different from that[%d] recorded in checkpoint. "+
+					"Perhaps you shoud specify the log restored ts instead, "+
+					"or just clean the checkpoint database[%s] if the cluster has been cleaned up.",
+				logRestoredTS, meta.LogRestoredTS, checkpoint.LogRestoreCheckpointDatabaseName,
+			)
+		}
+
 		// The schedulers config is nil, so the restore-schedulers operation is just nil.
 		// Then the undo function would use the result undo of `remove schedulers` operation,
 		// instead of that in checkpoint meta.
@@ -389,6 +403,7 @@ func (rc *SnapClient) InitCheckpoint(
 		meta := &checkpoint.CheckpointMetadataForSnapshotRestore{
 			UpstreamClusterID: rc.backupMeta.ClusterId,
 			RestoredTS:        rc.backupMeta.EndVersion,
+			LogRestoredTS:     logRestoredTS,
 			RestoreUUID:       restoreID,
 		}
 		rc.restoreUUID = restoreID

br/pkg/task/restore.go

@@ -970,7 +970,12 @@ func runSnapshotRestore(c context.Context, mgr *conn.Mgr, g glue.Glue, cmdName s
 	// reload or register the checkpoint
 	var checkpointSetWithTableID map[int64]map[string]struct{}
 	if cfg.UseCheckpoint {
-		sets, restoreSchedulersConfigFromCheckpoint, err := client.InitCheckpoint(ctx, g, mgr.GetStorage(), schedulersConfig, checkpointFirstRun)
+		logRestoredTS := uint64(0)
+		if cfg.piTRTaskInfo != nil {
+			logRestoredTS = cfg.piTRTaskInfo.RestoreTS
+		}
+		sets, restoreSchedulersConfigFromCheckpoint, err := client.InitCheckpoint(
+			ctx, g, mgr.GetStorage(), schedulersConfig, logRestoredTS, checkpointFirstRun)
 		if err != nil {
 			return errors.Trace(err)
 		}

br/pkg/task/stream.go

@@ -1987,6 +1987,7 @@ func checkPiTRRequirements(mgr *conn.Mgr) error {
 
 type PiTRTaskInfo struct {
 	CheckpointInfo      *checkpoint.CheckpointTaskInfoForLogRestore
+	RestoreTS           uint64
 	NeedFullRestore     bool
 	FullRestoreCheckErr error
 }
@@ -2049,6 +2050,7 @@ func generatePiTRTaskInfo(
 	}
 	checkInfo.CheckpointInfo = curTaskInfo
 	checkInfo.NeedFullRestore = doFullRestore
+	checkInfo.RestoreTS = cfg.RestoreTS
 	// restore full snapshot precheck.
 	if doFullRestore {
 		if !(cfg.UseCheckpoint && (curTaskInfo.Metadata != nil || curTaskInfo.HasSnapshotMetadata)) {