run prepare

Signed-off-by: Wenqi Mou <[email protected]>
pingcap · Nov 25, 2024 · 6ebc531 · 6ebc531
1 parent c29b578
commit 6ebc531
Show file tree

Hide file tree

Showing 12 changed files with 186 additions and 234 deletions.
diff --git a/br/pkg/checkpoint/log_restore.go b/br/pkg/checkpoint/log_restore.go
@@ -264,7 +264,8 @@ func ExistsCheckpointProgress(
 		TableExists(pmodel.NewCIStr(LogRestoreCheckpointDatabaseName), pmodel.NewCIStr(checkpointProgressTableName))
 }
 
-// CheckpointTaskInfoForLogRestore is tied to a specific cluster. It represents the last restore task executed  this cluster.
+// CheckpointTaskInfoForLogRestore is tied to a specific cluster.
+// It represents the last restore task executed in this cluster.
 type CheckpointTaskInfoForLogRestore struct {
 	Metadata            *CheckpointMetadataForLogRestore
 	HasSnapshotMetadata bool

diff --git a/br/pkg/restore/log_client/BUILD.bazel b/br/pkg/restore/log_client/BUILD.bazel
@@ -105,6 +105,7 @@ go_test(
         "//br/pkg/storage",
         "//br/pkg/stream",
         "//br/pkg/utils",
+        "//br/pkg/utils/consts",
         "//br/pkg/utils/iter",
         "//br/pkg/utiltest",
         "//pkg/domain",
@@ -119,7 +120,6 @@ go_test(
         "//pkg/util/chunk",
         "//pkg/util/codec",
         "//pkg/util/sqlexec",
-        "//pkg/util/table-filter",
         "@com_github_docker_go_units//:go-units",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",

diff --git a/br/pkg/restore/log_client/batch_file_processor.go b/br/pkg/restore/log_client/batch_file_processor.go
@@ -29,8 +29,8 @@ import (
 
 // BatchFileProcessor defines how to process a batch of files
 type BatchFileProcessor interface {
-	// process a batch of files and with a filterTS and return what's not processed for next iteration
-	processBatch(
+	// ProcessBatch processes a batch of files and with a filterTS and return what's not processed for next iteration
+	ProcessBatch(
 		ctx context.Context,
 		files []*backuppb.DataFileInfo,
 		entries []*KvEntryWithTS,
@@ -47,7 +47,7 @@ type RestoreProcessor struct {
 	progressInc    func()
 }
 
-func (rp *RestoreProcessor) processBatch(
+func (rp *RestoreProcessor) ProcessBatch(
 	ctx context.Context,
 	files []*backuppb.DataFileInfo,
 	entries []*KvEntryWithTS,
@@ -71,7 +71,7 @@ type DDLCollector struct {
 	tableRenameInfo *stream.LogBackupTableHistory
 }
 
-func (dc *DDLCollector) processBatch(
+func (dc *DDLCollector) ProcessBatch(
 	ctx context.Context,
 	files []*backuppb.DataFileInfo,
 	entries []*KvEntryWithTS,

diff --git a/br/pkg/restore/log_client/client.go b/br/pkg/restore/log_client/client.go
@@ -1016,15 +1016,17 @@ func (rc *LogClient) InitSchemasReplaceForDDL(
 			dbReplaces = stream.FromSchemaMaps(dbMaps)
 			if len(dbReplaces) <= 0 {
 				envVal, ok := os.LookupEnv(UnsafePITRLogRestoreStartBeforeAnyUpstreamUserDDL)
-				if ok && len(envVal) > 0 {
-					log.Info(fmt.Sprintf("the environment variable %s is active, skip loading the base schemas.", UnsafePITRLogRestoreStartBeforeAnyUpstreamUserDDL))
-				} else {
+				if !ok || len(envVal) <= 0 {
 					return nil, errors.Errorf("miss upstream table information at `start-ts`(%d) but the full backup path is not specified", rc.startTS)
 				}
+				log.Info(fmt.Sprintf("the environment variable %s is active, skip loading the base schemas.", UnsafePITRLogRestoreStartBeforeAnyUpstreamUserDDL))
 			}
 		} else {
 			log.Info("building table replaces from full backup storage")
 			dbReplaces, err = rc.generateDBReplacesFromFullBackupStorage(ctx, cfg, cipherInfo)
+			if err != nil {
+				return nil, errors.Trace(err)
+			}
 		}
 	}
 
@@ -1272,7 +1274,7 @@ func LoadAndProcessMetaKVFilesInBatch(
 			} else {
 				// Either f.MinTS > rangeMax or f.MinTs is the filterTs we need.
 				// So it is ok to pass f.MinTs as filterTs.
-				defaultKvEntries, err = processor.processBatch(ctx, defaultFiles[defaultIdx:i], defaultKvEntries, f.MinTs, consts.DefaultCF)
+				defaultKvEntries, err = processor.ProcessBatch(ctx, defaultFiles[defaultIdx:i], defaultKvEntries, f.MinTs, consts.DefaultCF)
 				if err != nil {
 					return errors.Trace(err)
 				}
@@ -1289,7 +1291,7 @@ func LoadAndProcessMetaKVFilesInBatch(
 						break
 					}
 				}
-				writeKvEntries, err = processor.processBatch(ctx, writeFiles[writeIdx:toWriteIdx], writeKvEntries, f.MinTs, consts.WriteCF)
+				writeKvEntries, err = processor.ProcessBatch(ctx, writeFiles[writeIdx:toWriteIdx], writeKvEntries, f.MinTs, consts.WriteCF)
 				if err != nil {
 					return errors.Trace(err)
 				}
@@ -1301,11 +1303,11 @@ func LoadAndProcessMetaKVFilesInBatch(
 	// restore the left meta kv files and entries
 	// Notice: restoreBatch needs to realize the parameter `files` and `kvEntries` might be empty
 	// Assert: defaultIdx <= len(defaultFiles) && writeIdx <= len(writeFiles)
-	_, err = processor.processBatch(ctx, defaultFiles[defaultIdx:], defaultKvEntries, math.MaxUint64, consts.DefaultCF)
+	_, err = processor.ProcessBatch(ctx, defaultFiles[defaultIdx:], defaultKvEntries, math.MaxUint64, consts.DefaultCF)
 	if err != nil {
 		return errors.Trace(err)
 	}
-	_, err = processor.processBatch(ctx, writeFiles[writeIdx:], writeKvEntries, math.MaxUint64, consts.WriteCF)
+	_, err = processor.ProcessBatch(ctx, writeFiles[writeIdx:], writeKvEntries, math.MaxUint64, consts.WriteCF)
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -1327,6 +1329,9 @@ func (rc *LogClient) RestoreBatchMetaKVFiles(
 	cf string,
 ) ([]*KvEntryWithTS, error) {
 	curSortedKvEntries, filteredOutKvEntries, err := rc.filterAndSortKvEntriesFromFiles(ctx, files, kvEntries, filterTS)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
 
 	// restore and rewrite these entries to TiKV with rawPut() method.
 	kvCount, size, err := rc.restoreAndRewriteMetaKvEntries(ctx, schemasReplace, curSortedKvEntries, cf)