parse metadata name flags

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

Author: Leavrth

br/pkg/restore/log_client/log_file_manager.go

@@ -236,11 +236,11 @@ func (lm *LogFileManager) loadShiftTS(ctx context.Context) error {
 }
 
 func (lm *LogFileManager) streamingMeta(ctx context.Context) (MetaNameIter, error) {
-	return lm.streamingMetaByTS(ctx)
+	return lm.streamingMetaByTS(ctx, func(_ string) bool { return true })
 }
 
-func (lm *LogFileManager) streamingMetaByTS(ctx context.Context) (MetaNameIter, error) {
-	it, err := lm.createMetaIterOver(ctx, lm.storage)
+func (lm *LogFileManager) streamingMetaByTS(ctx context.Context, cond func(path string) bool) (MetaNameIter, error) {
+	it, err := lm.createMetaIterOver(ctx, lm.storage, cond)
 	if err != nil {
 		return nil, err
 	}
@@ -250,15 +250,15 @@ func (lm *LogFileManager) streamingMetaByTS(ctx context.Context) (MetaNameIter,
 	return filtered, nil
 }
 
-func (lm *LogFileManager) createMetaIterOver(ctx context.Context, s storeapi.Storage) (MetaNameIter, error) {
+func (lm *LogFileManager) createMetaIterOver(ctx context.Context, s storeapi.Storage, cond func(path string) bool) (MetaNameIter, error) {
 	opt := &storeapi.WalkOption{SubDir: stream.GetStreamBackupMetaPrefix()}
 	names := []string{}
 	err := s.WalkDir(ctx, opt, func(path string, size int64) error {
 		if !strings.HasSuffix(path, ".meta") {
 			return nil
 		}
 		newPath := stream.FilterPathByTs(path, lm.shiftStartTS, lm.restoreTS)
-		if len(newPath) > 0 {
+		if len(newPath) > 0 && cond(newPath) {
 			names = append(names, newPath)
 		}
 		return nil
@@ -347,7 +347,13 @@ func (lm *LogFileManager) collectDDLFilesAndPrepareCache(
 // LoadDDLFiles loads all DDL files needs to be restored in the restoration.
 // This function returns all DDL files needing directly because we need sort all of them.
 func (lm *LogFileManager) LoadDDLFiles(ctx context.Context) ([]Log, error) {
-	m, err := lm.streamingMeta(ctx)
+	m, err := lm.streamingMetaByTS(ctx, func(path string) bool {
+		parsedName, err := stream.TryParseTaggedBackupMetaFileNameWrapper(path)
+		if err != nil {
+			return true
+		}
+		return parsedName.HasDDLFiles()
+	})
 	if err != nil {
 		return nil, err
 	}

br/pkg/stream/backupmetas/parser.go

@@ -32,6 +32,11 @@ const (
 	NameMinBeginTsInDefaultCfTag byte = 'd'
 	NameMinTSTag                 byte = 'l'
 	NameMaxTSTag                 byte = 'u'
+	NameFlagsTag                 byte = 'p'
+)
+
+const (
+	flagHasDDLFiles = 1 << iota
 )
 
 var (
@@ -46,6 +51,9 @@ type ParsedName struct {
 	MinBeginTsInDefaultCf uint64
 	MinTS                 uint64
 	MaxTS                 uint64
+	Flags                 uint64
+
+	HasFlags bool
 }
 
 type ShiftTSStatus uint8
@@ -77,6 +85,13 @@ func (parsedName *ParsedName) CalculateShiftTS(startTS uint64, restoreTS uint64)
 	return parsedName.MinBeginTsInDefaultCf, ShiftTSFound
 }
 
+func (parsedName *ParsedName) HasDDLFiles() bool {
+	if !parsedName.HasFlags {
+		return true
+	}
+	return parsedName.Flags&flagHasDDLFiles != 0
+}
+
 // TryParseTaggedBackupMetaFileName parses the tagged backupmeta file-name format.
 func TryParseTaggedBackupMetaFileName(fileName string) (ParsedName, error) {
 	if !taggedBackupMetaPattern.MatchString(fileName) {
@@ -138,6 +153,8 @@ func parseTaggedBackupMetaFileName(fileName string) (ParsedName, error) {
 		minBeginTsInDefaultCf uint64
 		minTs                 uint64
 		maxTs                 uint64
+		flags                 uint64
+		hasFlags              bool
 		seenTags              [256]bool
 	)
 	for pos := 0; pos < len(suffix); {
@@ -165,6 +182,9 @@ func parseTaggedBackupMetaFileName(fileName string) (ParsedName, error) {
 			minTs = value
 		case NameMaxTSTag:
 			maxTs = value
+		case NameFlagsTag:
+			hasFlags = true
+			flags = value
 		}
 		pos += taggedMetaTagValueLen
 	}
@@ -182,6 +202,8 @@ func parseTaggedBackupMetaFileName(fileName string) (ParsedName, error) {
 		MinBeginTsInDefaultCf: minBeginTsInDefaultCf,
 		MinTS:                 minTs,
 		MaxTS:                 maxTs,
+		Flags:                 flags,
+		HasFlags:              hasFlags,
 	}, nil
 }
 

br/pkg/stream/stream_mgr_fuzz_test.go

@@ -30,7 +30,8 @@ func FuzzParseBackupMetaFileNameRoundTrip(f *testing.F) {
 		if !isASCIIAlphanumeric(extraTag) ||
 			extraTag == backupmetas.NameMinBeginTsInDefaultCfTag ||
 			extraTag == backupmetas.NameMinTSTag ||
-			extraTag == backupmetas.NameMaxTSTag {
+			extraTag == backupmetas.NameMaxTSTag ||
+			extraTag == backupmetas.NameFlagsTag {
 			extraTag = 'x'
 		}
 
@@ -57,6 +58,32 @@ func FuzzParseBackupMetaFileNameRoundTrip(f *testing.F) {
 			MinTS:                 minTs,
 			MaxTS:                 maxTs,
 		}, taggedParsed)
+		require.True(t, taggedParsed.HasDDLFiles())
+
+		taggedWithFlagsFileName := fmt.Sprintf(
+			"%016X%016X-d%016Xu%016Xl%016Xp%016X",
+			flushTs, storeID, minBeginTsInDefaultCf, maxTs, minTs, uint64(1),
+		)
+		taggedWithFlagsParsed, err := backupmetas.ParseName(taggedWithFlagsFileName)
+		require.NoError(t, err)
+		require.Equal(t, backupmetas.ParsedName{
+			FlushTS:               flushTs,
+			StoreID:               storeID,
+			MinBeginTsInDefaultCf: minBeginTsInDefaultCf,
+			MinTS:                 minTs,
+			MaxTS:                 maxTs,
+			Flags:                 1,
+			HasFlags:              true,
+		}, taggedWithFlagsParsed)
+		require.True(t, taggedWithFlagsParsed.HasDDLFiles())
+
+		taggedWithoutDDLFileName := fmt.Sprintf(
+			"%016X%016X-d%016Xu%016Xl%016Xp%016X",
+			flushTs, storeID, minBeginTsInDefaultCf, maxTs, minTs, uint64(0),
+		)
+		taggedWithoutDDLParsed, err := backupmetas.ParseName(taggedWithoutDDLFileName)
+		require.NoError(t, err)
+		require.False(t, taggedWithoutDDLParsed.HasDDLFiles())
 
 		tagValues := map[byte]uint64{
 			backupmetas.NameMinBeginTsInDefaultCfTag: minBeginTsInDefaultCf,