Restore: implement restorer for compacted SST/Snapshot/log files

br/pkg/restore/BUILD.bazel

@@ -5,25 +5,33 @@ go_library(
     srcs = [
         "import_mode_switcher.go",
         "misc.go",
+        "restorer.go",
     ],
     importpath = "github.com/pingcap/tidb/br/pkg/restore",
     visibility = ["//visibility:public"],
     deps = [
+        "//br/pkg/checkpoint",
         "//br/pkg/conn",
         "//br/pkg/conn/util",
         "//br/pkg/errors",
         "//br/pkg/logutil",
         "//br/pkg/pdutil",
+        "//br/pkg/restore/split",
+        "//br/pkg/restore/utils",
+        "//br/pkg/summary",
         "//br/pkg/utils",
+        "//br/pkg/utils/iter",
         "//pkg/domain",
         "//pkg/kv",
         "//pkg/meta",
         "//pkg/meta/model",
         "//pkg/parser/model",
         "//pkg/util",
         "@com_github_go_sql_driver_mysql//:mysql",
+        "@com_github_opentracing_opentracing_go//:opentracing-go",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
+        "@com_github_pingcap_kvproto//pkg/brpb",
         "@com_github_pingcap_kvproto//pkg/import_sstpb",
         "@com_github_pingcap_log//:log",
         "@com_github_tikv_client_go_v2//oracle",
@@ -34,6 +42,7 @@ go_library(
         "@org_golang_google_grpc//credentials/insecure",
         "@org_golang_x_sync//errgroup",
         "@org_uber_go_zap//:zap",
+        "@org_uber_go_zap//zapcore",
     ],
 )
 
@@ -43,21 +52,28 @@ go_test(
     srcs = [
         "import_mode_switcher_test.go",
         "misc_test.go",
+        "restorer_test.go",
     ],
     flaky = True,
-    race = "off",
-    shard_count = 6,
+    shard_count = 13,
     deps = [
         ":restore",
         "//br/pkg/conn",
         "//br/pkg/mock",
         "//br/pkg/pdutil",
         "//br/pkg/restore/split",
+        "//br/pkg/restore/utils",
+        "//br/pkg/utils/iter",
         "//pkg/kv",
         "//pkg/parser/model",
         "//pkg/session",
+        "//pkg/tablecodec",
+        "//pkg/util",
+        "//pkg/util/codec",
         "@com_github_coreos_go_semver//semver",
+        "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
+        "@com_github_pingcap_kvproto//pkg/brpb",
         "@com_github_pingcap_kvproto//pkg/import_sstpb",
         "@com_github_pingcap_kvproto//pkg/metapb",
         "@com_github_stretchr_testify//require",

br/pkg/restore/import_mode_switcher.go

@@ -5,6 +5,7 @@ package restore
 import (
 	"context"
 	"crypto/tls"
+	"sync"
 	"time"
 
 	_ "github.com/go-sql-driver/mysql" // mysql driver
@@ -46,9 +47,11 @@ func NewImportModeSwitcher(
 	}
 }
 
+var closeOnce sync.Once
+
 // switchToNormalMode switch tikv cluster to normal mode.
-func (switcher *ImportModeSwitcher) switchToNormalMode(ctx context.Context) error {
-	close(switcher.switchCh)
+func (switcher *ImportModeSwitcher) SwitchToNormalMode(ctx context.Context) error {
+	closeOnce.Do(func() { close(switcher.switchCh) })
 	return switcher.switchTiKVMode(ctx, import_sstpb.SwitchMode_Normal)
 }
 
@@ -113,8 +116,8 @@ func (switcher *ImportModeSwitcher) switchTiKVMode(
 	return nil
 }
 
-// switchToImportMode switch tikv cluster to import mode.
-func (switcher *ImportModeSwitcher) switchToImportMode(
+// SwitchToImportMode switch tikv cluster to import mode.
+func (switcher *ImportModeSwitcher) SwitchToImportMode(
 	ctx context.Context,
 ) {
 	// tikv automatically switch to normal mode in every 10 minutes
@@ -163,7 +166,7 @@ func RestorePreWork(
 
 	if switchToImport {
 		// Switch TiKV cluster to import mode (adjust rocksdb configuration).
-		switcher.switchToImportMode(ctx)
+		switcher.SwitchToImportMode(ctx)
 	}
 
 	return mgr.RemoveSchedulersWithConfig(ctx)
@@ -186,7 +189,7 @@ func RestorePostWork(
 		ctx = context.Background()
 	}
 
-	if err := switcher.switchToNormalMode(ctx); err != nil {
+	if err := switcher.SwitchToNormalMode(ctx); err != nil {
 		log.Warn("fail to switch to normal mode", zap.Error(err))
 	}
 	if err := restoreSchedulers(ctx); err != nil {

br/pkg/restore/log_client/BUILD.bazel

@@ -4,10 +4,12 @@ go_library(
     name = "log_client",
     srcs = [
         "client.go",
+        "compacted_file_strategy.go",
         "import.go",
         "import_retry.go",
         "log_file_manager.go",
         "log_file_map.go",
+        "log_split_strategy.go",
         "migration.go",
     ],
     importpath = "github.com/pingcap/tidb/br/pkg/restore/log_client",
@@ -26,6 +28,7 @@ go_library(
         "//br/pkg/restore/ingestrec",
         "//br/pkg/restore/internal/import_client",
         "//br/pkg/restore/internal/rawkv",
+        "//br/pkg/restore/snap_client",
         "//br/pkg/restore/split",
         "//br/pkg/restore/tiflashrec",
         "//br/pkg/restore/utils",
@@ -43,6 +46,7 @@ go_library(
         "//pkg/util",
         "//pkg/util/codec",
         "//pkg/util/redact",
+        "//pkg/util/sqlexec",
         "//pkg/util/table-filter",
         "@com_github_fatih_color//:color",
         "@com_github_gogo_protobuf//proto",
@@ -86,12 +90,13 @@ go_test(
     ],
     embed = [":log_client"],
     flaky = True,
-    shard_count = 42,
+    shard_count = 45,
     deps = [
         "//br/pkg/errors",
         "//br/pkg/glue",
         "//br/pkg/gluetidb",
         "//br/pkg/mock",
+        "//br/pkg/restore",
         "//br/pkg/restore/internal/import_client",
         "//br/pkg/restore/split",
         "//br/pkg/restore/utils",
@@ -113,17 +118,16 @@ go_test(
         "//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",
         "@com_github_pingcap_kvproto//pkg/brpb",
         "@com_github_pingcap_kvproto//pkg/encryptionpb",
         "@com_github_pingcap_kvproto//pkg/errorpb",
         "@com_github_pingcap_kvproto//pkg/import_sstpb",
         "@com_github_pingcap_kvproto//pkg/metapb",
-        "@com_github_pingcap_kvproto//pkg/pdpb",
         "@com_github_pingcap_log//:log",
         "@com_github_stretchr_testify//require",
-        "@com_github_tikv_pd_client//:client",
         "@org_golang_google_grpc//codes",
         "@org_golang_google_grpc//keepalive",
         "@org_golang_google_grpc//status",