try to skip downloading metadata when calculate shift ts

Signed-off-by: Jianjun Liao <jianjun.liao@outlook.com>

Author: Leavrth

br/pkg/restore/log_client/BUILD.bazel

@@ -37,6 +37,7 @@ go_library(
         "//br/pkg/restore/tiflashrec",
         "//br/pkg/restore/utils",
         "//br/pkg/stream",
+        "//br/pkg/stream/backupmetas",
         "//br/pkg/summary",
         "//br/pkg/utils",
         "//br/pkg/utils/consts",

br/pkg/restore/log_client/log_file_manager.go

@@ -19,6 +19,7 @@ import (
 	"github.com/pingcap/tidb/br/pkg/encryption"
 	berrors "github.com/pingcap/tidb/br/pkg/errors"
 	"github.com/pingcap/tidb/br/pkg/stream"
+	"github.com/pingcap/tidb/br/pkg/stream/backupmetas"
 	"github.com/pingcap/tidb/br/pkg/utils"
 	"github.com/pingcap/tidb/br/pkg/utils/consts"
 	"github.com/pingcap/tidb/br/pkg/utils/iter"
@@ -180,15 +181,37 @@ func (lm *LogFileManager) loadShiftTS(ctx context.Context) error {
 		// use start ts to calculate shift start ts
 		lm.startTS,
 		lm.restoreTS,
-		lm.metadataDownloadBatchSize, func(path string, raw []byte) error {
+		lm.metadataDownloadBatchSize,
+		func(filename string) bool {
+			parsedName, err := stream.TryParseTaggedBackupMetaFileNameWrapper(filename)
+			if err != nil {
+				return false
+			}
+			ts, status := parsedName.CalculateShiftTS(lm.startTS, lm.restoreTS)
+			switch status {
+			case backupmetas.ShiftTSFound:
+			case backupmetas.ShiftTSNotFound:
+				return true
+			default:
+				return false
+			}
+			shiftTS.Lock()
+			if !shiftTS.exists || shiftTS.value > ts {
+				shiftTS.value = ts
+				shiftTS.exists = true
+			}
+			shiftTS.Unlock()
+			return true
+		},
+		func(filename string, raw []byte) error {
 			m, err := lm.helper.ParseToMetadata(raw)
 			if err != nil {
 				return err
 			}
-			log.Info("read meta from storage and parse", zap.String("path", path), zap.Uint64("min-ts", m.MinTs),
+			log.Info("read meta from storage and parse", zap.String("path", filename), zap.Uint64("min-ts", m.MinTs),
 				zap.Uint64("max-ts", m.MaxTs), zap.Int32("meta-version", int32(m.MetaVersion)))
 
-			ts, ok := stream.UpdateShiftTS(m, lm.startTS, lm.restoreTS)
+			ts, ok := stream.UpdateShiftTSFromMetadata(m, lm.startTS, lm.restoreTS)
 			shiftTS.Lock()
 			if ok && (!shiftTS.exists || shiftTS.value > ts) {
 				shiftTS.value = ts
@@ -206,7 +229,7 @@ func (lm *LogFileManager) loadShiftTS(ctx context.Context) error {
 		lm.withMigrationBuilder.SetShiftStartTS(lm.shiftStartTS)
 		return nil
 	}
-	lm.shiftStartTS = shiftTS.value
+	lm.shiftStartTS = min(lm.startTS, shiftTS.value)
 	lm.withMigrationBuilder.SetShiftStartTS(lm.shiftStartTS)
 	return nil
 }
@@ -332,18 +355,6 @@ func (lm *LogFileManager) LoadDDLFiles(ctx context.Context) ([]Log, error) {
 	return lm.collectDDLFilesAndPrepareCache(ctx, mg)
 }
 
-type loadDMLFilesConfig struct {
-	Statistic *logFilesStatistic
-}
-
-type loadDMLFilesOption func(*loadDMLFilesConfig)
-
-func lDOptWithStatistics(s *logFilesStatistic) loadDMLFilesOption {
-	return func(c *loadDMLFilesConfig) {
-		c.Statistic = s
-	}
-}
-
 // LoadDMLFiles loads all DML files needs to be restored in the restoration.
 // This function returns a stream, because there are usually many DML files need to be restored.
 func (lm *LogFileManager) LoadDMLFiles(ctx context.Context) (LogIter, error) {

br/pkg/stream/backupmetas/parser.go

@@ -48,6 +48,14 @@ type ParsedName struct {
 	MaxTS                 uint64
 }
 
+type ShiftTSStatus uint8
+
+const (
+	ShiftTSFound ShiftTSStatus = iota
+	ShiftTSNotFound
+	ShiftTSInvalidStats
+)
+
 func ParseName(fileName string) (ParsedName, error) {
 	switch {
 	case taggedBackupMetaPattern.MatchString(fileName):
@@ -59,6 +67,24 @@ func ParseName(fileName string) (ParsedName, error) {
 	}
 }
 
+func (parsedName *ParsedName) CalculateShiftTS(startTS uint64, restoreTS uint64) (uint64, ShiftTSStatus) {
+	if parsedName.MinTS > restoreTS || parsedName.MaxTS < startTS {
+		return 0, ShiftTSNotFound
+	}
+	if parsedName.MinBeginTsInDefaultCf == 0 || parsedName.MinBeginTsInDefaultCf > parsedName.MinTS {
+		return 0, ShiftTSInvalidStats
+	}
+	return parsedName.MinBeginTsInDefaultCf, ShiftTSFound
+}
+
+// TryParseTaggedBackupMetaFileName parses the tagged backupmeta file-name format.
+func TryParseTaggedBackupMetaFileName(fileName string) (ParsedName, error) {
+	if !taggedBackupMetaPattern.MatchString(fileName) {
+		return ParsedName{}, errors.Errorf("invalid latest backupmeta file name format: %s", fileName)
+	}
+	return parseTaggedBackupMetaFileName(fileName)
+}
+
 func parseLegacyBackupMetaFileName(fileName string) (ParsedName, error) {
 	parts := strings.Split(fileName, "-")
 	if len(parts) != legacyBackupMetaPartCount {

br/pkg/stream/export_test.go

@@ -16,8 +16,10 @@ package stream
 
 import (
 	"context"
+	"math"
 
 	backuppb "github.com/pingcap/kvproto/pkg/brpb"
+	"github.com/pingcap/tidb/pkg/objstore/storeapi"
 )
 
 func (s *StreamBackupSearch) SearchFromDataFileForTest(
@@ -33,3 +35,9 @@ func (s *StreamBackupSearch) MergeCFEntriesForTest(
 ) []*StreamKVInfo {
 	return s.mergeCFEntries(defaultCFEntries, writeCFEntries)
 }
+
+// LoadFrom loads data from an external storage into the stream metadata set. (Now only for test)
+func (ms *StreamMetadataSet) LoadFrom(ctx context.Context, s storeapi.Storage) error {
+	_, err := ms.LoadUntilAndCalculateShiftTS(ctx, s, math.MaxUint64)
+	return err
+}

br/pkg/stream/stream_metas.go

@@ -28,6 +28,7 @@ import (
 	berrors "github.com/pingcap/tidb/br/pkg/errors"
 	"github.com/pingcap/tidb/br/pkg/glue"
 	"github.com/pingcap/tidb/br/pkg/logutil"
+	"github.com/pingcap/tidb/br/pkg/stream/backupmetas"
 	"github.com/pingcap/tidb/br/pkg/utils/consts"
 	"github.com/pingcap/tidb/br/pkg/utils/iter"
 	"github.com/pingcap/tidb/pkg/objstore"
@@ -112,7 +113,8 @@ func (ms *StreamMetadataSet) LoadUntilAndCalculateShiftTS(
 	err := FastUnmarshalMetaData(ctx, s,
 		0,
 		until,
-		ms.MetadataDownloadBatchSize, func(path string, raw []byte) error {
+		ms.MetadataDownloadBatchSize,
+		func(string) bool { return false }, func(filename string, raw []byte) error {
 			m, err := ms.Helper.ParseToMetadataHard(raw)
 			if err != nil {
 				return err
@@ -136,15 +138,15 @@ func (ms *StreamMetadataSet) LoadUntilAndCalculateShiftTS(
 					})
 				}
 				metadataMap.Lock()
-				metadataMap.metas[path] = &MetadataInfo{
+				metadataMap.metas[filename] = &MetadataInfo{
 					MinTS:          m.MinTs,
 					FileGroupInfos: fileGroupInfos,
 				}
 				metadataMap.Unlock()
 			}
 			// filter out the metadatas whose ts-range is overlap with [until, +inf)
 			// and calculate their minimum begin-default-ts
-			ts, ok := UpdateShiftTS(m, until, mathutil.MaxUint)
+			ts, ok := UpdateShiftTS(filename, m, until, mathutil.MaxUint)
 			if ok {
 				metadataMap.Lock()
 				if ts < metadataMap.shiftUntilTS {
@@ -164,12 +166,6 @@ func (ms *StreamMetadataSet) LoadUntilAndCalculateShiftTS(
 	return metadataMap.shiftUntilTS, nil
 }
 
-// LoadFrom loads data from an external storage into the stream metadata set. (Now only for test)
-func (ms *StreamMetadataSet) LoadFrom(ctx context.Context, s storeapi.Storage) error {
-	_, err := ms.LoadUntilAndCalculateShiftTS(ctx, s, math.MaxUint64)
-	return err
-}
-
 func (ms *StreamMetadataSet) iterateDataFiles(f func(d *FileGroupInfo) (shouldBreak bool)) {
 	for _, m := range ms.metadataInfos {
 		if slices.ContainsFunc(m.FileGroupInfos, f) {
@@ -295,7 +291,26 @@ func SetTSToFile(
 	return truncateAndWrite(ctx, s, filename, []byte(content))
 }
 
-func UpdateShiftTS(m *pb.Metadata, startTS uint64, restoreTS uint64) (uint64, bool) {
+func TryParseTaggedBackupMetaFileNameWrapper(filename string) (backupmetas.ParsedName, error) {
+	baseName := strings.TrimSuffix(path.Base(filename), metaSuffix)
+	return backupmetas.TryParseTaggedBackupMetaFileName(baseName)
+}
+
+func UpdateShiftTS(filename string, m *pb.Metadata, startTS, restoreTS uint64) (uint64, bool) {
+	parsedName, err := TryParseTaggedBackupMetaFileNameWrapper(filename)
+	if err == nil {
+		ts, status := parsedName.CalculateShiftTS(startTS, restoreTS)
+		switch status {
+		case backupmetas.ShiftTSFound:
+			return ts, true
+		case backupmetas.ShiftTSNotFound:
+			return 0, false
+		}
+	}
+	return UpdateShiftTSFromMetadata(m, startTS, restoreTS)
+}
+
+func UpdateShiftTSFromMetadata(m *pb.Metadata, startTS uint64, restoreTS uint64) (uint64, bool) {
 	var (
 		minBeginTS uint64
 		isExist    bool

br/pkg/stream/stream_mgr.go

@@ -404,6 +404,7 @@ func FastUnmarshalMetaData(
 	startTS uint64,
 	endTS uint64,
 	metaDataWorkerPoolSize uint,
+	skipCondition func(filename string) bool,
 	fn func(path string, rawMetaData []byte) error,
 ) error {
 	log.Info("use workers to speed up reading metadata files", zap.Uint("workers", metaDataWorkerPoolSize))
@@ -423,6 +424,9 @@ func FastUnmarshalMetaData(
 			)
 			return nil
 		}
+		if skipCondition(path) {
+			return nil
+		}
 		pool.ApplyOnErrorGroup(eg, func() error {
 			b, err := s.ReadFile(ectx, readPath)
 			if err != nil {

br/pkg/utils/iter/BUILD.bazel

@@ -15,7 +15,6 @@ go_library(
         "//pkg/util",
         "@com_github_prometheus_client_golang//prometheus",
         "@org_golang_x_exp//constraints",
-        "@org_golang_x_sync//errgroup",
     ],
 )