Merge remote-tracking branch 'upstream/master' into compress_16m

.gitignore

@@ -37,3 +37,4 @@ bazel-testlogs
 bazel-tidb
 .ijwb/
 /oom_record/
+*.log.json

Makefile

@@ -191,9 +191,9 @@ else
 endif
 
 enterprise-server:
-	@make init-submodule
-	@make enterprise-prepare
-	@make enterprise-server-build
+	$(MAKE) init-submodule
+	$(MAKE) enterprise-prepare
+	$(MAKE) enterprise-server-build
 
 server_check:
 ifeq ($(TARGET), "")
@@ -568,6 +568,30 @@ bazel_addindextest: failpoint-enable bazel_ci_simple_prepare
 		-- //tests/realtikvtest/addindextest/...
 	./build/jenkins_collect_coverage.sh
 
+bazel_addindextest1: failpoint-enable bazel_ci_simple_prepare
+	bazel $(BAZEL_GLOBAL_CONFIG) coverage $(BAZEL_CMD_CONFIG) --test_arg=-with-real-tikv --define gotags=deadlock,intest \
+	--@io_bazel_rules_go//go/config:cover_format=go_cover \
+		-- //tests/realtikvtest/addindextest1/...
+	./build/jenkins_collect_coverage.sh
+
+bazel_addindextest2: failpoint-enable bazel_ci_simple_prepare
+	bazel $(BAZEL_GLOBAL_CONFIG) coverage $(BAZEL_CMD_CONFIG) --test_arg=-with-real-tikv --define gotags=deadlock,intest \
+	--@io_bazel_rules_go//go/config:cover_format=go_cover \
+		-- //tests/realtikvtest/addindextest2/...
+	./build/jenkins_collect_coverage.sh
+
+bazel_addindextest3: failpoint-enable bazel_ci_simple_prepare
+	bazel $(BAZEL_GLOBAL_CONFIG) coverage $(BAZEL_CMD_CONFIG) --test_arg=-with-real-tikv --define gotags=deadlock,intest \
+	--@io_bazel_rules_go//go/config:cover_format=go_cover \
+		-- //tests/realtikvtest/addindextest3/...
+	./build/jenkins_collect_coverage.sh
+
+bazel_addindextest4: failpoint-enable bazel_ci_simple_prepare
+	bazel $(BAZEL_GLOBAL_CONFIG) coverage $(BAZEL_CMD_CONFIG) --test_arg=-with-real-tikv --define gotags=deadlock,intest \
+	--@io_bazel_rules_go//go/config:cover_format=go_cover \
+		-- //tests/realtikvtest/addindextest4/...
+	./build/jenkins_collect_coverage.sh
+
 # on timeout, bazel won't print log sometimes, so we use --test_output=all to print log always
 bazel_importintotest: failpoint-enable bazel_ci_simple_prepare
 	bazel $(BAZEL_GLOBAL_CONFIG) coverage $(BAZEL_CMD_CONFIG) --test_output=all --test_arg=-with-real-tikv --define gotags=deadlock,intest \

OWNERS

@@ -123,6 +123,7 @@ reviewers:
   - fixdb
   - fzzf678
   - iamxy
+  - jiyfhust
   - JmPotato
   - js00070
   - lamxTyler

OWNERS_ALIASES

@@ -26,6 +26,10 @@ aliases:
     - windtalker
     - XuHuaiyu
     - zanmato1984
+  sig-approvers-executor-import: # approvers for importer part of executor pkg.
+    - D3Hunter
+    - gmhdbjd
+    - lance6716    
   sig-approvers-expression: # approvers for expression pkg
     - windtalker
     - XuHuaiyu
@@ -44,3 +48,70 @@ aliases:
     - lance6716
     - tangenta
     - wjhuang2016
+  sig-approvers-ddl: # approvers for ddl pkg
+    - Benjamin2037
+    - tangenta
+    - wjhuang2016
+    - ywqzzy
+    - zimulala
+  sig-approvers-disttask: # approvers for disttask pkg
+    - Benjamin2037
+    - D3Hunter
+    - gmhdbjd
+    - lance6716
+    - tangenta
+    - wjhuang2016
+    - ywqzzy
+  sig-approvers-dumpling: # approvers for dumpling module
+    - Benjamin2037
+    - gmhdbjd
+    - lichunzhu
+    - okJiang
+  sig-approvers-infoschema: # approvers for infoschema pkg
+    - zimulala
+    - wjhuang2016
+    - tangenta
+    - ywqzzy
+    - D3Hunter
+    - Benjamin2037
+    - gmhdbjd
+  sig-approvers-meta: # approvers for meta pkg
+    - Benjamin2037
+    - gmhdbjd
+    - tangenta
+    - wjhuang2016
+    - ywqzzy
+    - zimulala    
+  sig-approvers-owner: # approvers for `owner` pkg
+    - Benjamin2037
+    - lichunzhu
+    - tangenta
+    - wjhuang2016
+    - ywqzzy
+    - zimulala
+  sig-approvers-resourcemanager: # approvers for resourcemanager pkg
+    - Benjamin2037
+    - D3Hunter
+    - gmhdbjd
+    - lance6716
+    - tangenta
+    - wjhuang2016
+    - ywqzzy
+  sig-approvers-table: # approvers for table packages.
+    - Benjamin2037
+    - cfzjywxk
+    - gmhdbjd
+    - tangenta
+    - wjhuang2016
+    - ywqzzy
+    - zimulala
+  sig-approvers-lock: # approvers for lock pkg
+    - Benjamin2037
+    - lance6716
+    - tangenta
+    - wjhuang2016
+    - zimulala
+  sig-approvers-tidb-binlog: # approvers for tidb-binlog module
+    - Benjamin2037
+    - gmhdbjd
+    - lichunzhu

br/OWNERS

@@ -2,7 +2,7 @@
 options:
   no_parent_owners: true
 filters:
-  "(tidb-lightning\\.tom)$":
+  "(tidb-lightning\\.toml)$":
     approvers:
       - sig-critical-approvers-tidb-lightning
   ".*":

br/pkg/aws/ebs.go

@@ -281,6 +281,152 @@ func (e *EC2Session) DeleteSnapshots(snapIDMap map[string]string) {
 	log.Info("delete snapshot end", zap.Int("need-to-del", len(snapIDMap)), zap.Int32("deleted", deletedCnt.Load()))
 }
 
+// EnableDataFSR enables FSR for data volume snapshots
+func (e *EC2Session) EnableDataFSR(meta *config.EBSBasedBRMeta, targetAZ string) (map[string][]*string, error) {
+	snapshotsIDsMap := fetchTargetSnapshots(meta, targetAZ)
+
+	if len(snapshotsIDsMap) == 0 {
+		return snapshotsIDsMap, errors.Errorf("empty backup meta")
+	}
+
+	eg, _ := errgroup.WithContext(context.Background())
+
+	for availableZone := range snapshotsIDsMap {
+		targetAZ := availableZone
+		eg.Go(func() error {
+			log.Info("enable fsr for snapshots", zap.String("available zone", targetAZ))
+			resp, err := e.ec2.EnableFastSnapshotRestores(&ec2.EnableFastSnapshotRestoresInput{
+				AvailabilityZones: []*string{&targetAZ},
+				SourceSnapshotIds: snapshotsIDsMap[targetAZ],
+			})
+
+			if err != nil {
+				return errors.Trace(err)
+			}
+
+			if len(resp.Unsuccessful) > 0 {
+				log.Warn("not all snapshots enabled FSR")
+				return errors.Errorf("Some snapshot fails to enable FSR for available zone %s, such as %s, error code is %v", targetAZ, *resp.Unsuccessful[0].SnapshotId, resp.Unsuccessful[0].FastSnapshotRestoreStateErrors)
+			}
+
+			return e.waitDataFSREnabled(snapshotsIDsMap[targetAZ], targetAZ)
+		})
+	}
+	return snapshotsIDsMap, eg.Wait()
+}
+
+// waitDataFSREnabled waits FSR for data volume snapshots are all enabled
+func (e *EC2Session) waitDataFSREnabled(snapShotIDs []*string, targetAZ string) error {
+	// Create a map to store the strings as keys
+	pendingSnapshots := make(map[string]struct{})
+
+	// Populate the map with the strings from the array
+	for _, str := range snapShotIDs {
+		pendingSnapshots[*str] = struct{}{}
+	}
+
+	log.Info("starts check fsr pending snapshots", zap.Any("snapshots", pendingSnapshots), zap.String("available zone", targetAZ))
+	for {
+		if len(pendingSnapshots) == 0 {
+			log.Info("all snapshots fsr enablement is finished", zap.String("available zone", targetAZ))
+			return nil
+		}
+
+		// check pending snapshots every 1 minute
+		time.Sleep(1 * time.Minute)
+		log.Info("check snapshots not fsr enabled", zap.Int("count", len(pendingSnapshots)))
+		input := &ec2.DescribeFastSnapshotRestoresInput{
+			Filters: []*ec2.Filter{
+				{
+					Name:   aws.String("state"),
+					Values: []*string{aws.String("disabled"), aws.String("disabling"), aws.String("enabling"), aws.String("optimizing")},
+				},
+				{
+					Name:   aws.String("availability-zone"),
+					Values: []*string{aws.String(targetAZ)},
+				},
+			},
+		}
+
+		result, err := e.ec2.DescribeFastSnapshotRestores(input)
+		if err != nil {
+			return errors.Trace(err)
+		}
+
+		uncompletedSnapshots := make(map[string]struct{})
+		for _, fastRestore := range result.FastSnapshotRestores {
+			_, found := pendingSnapshots[*fastRestore.SnapshotId]
+			if found {
+				// Detect some conflict states
+				if strings.EqualFold(*fastRestore.State, "disabled") || strings.EqualFold(*fastRestore.State, "disabling") {
+					log.Error("detect conflict status", zap.String("snapshot", *fastRestore.SnapshotId), zap.String("status", *fastRestore.State))
+					return errors.Errorf("status of snapshot %s is %s ", *fastRestore.SnapshotId, *fastRestore.State)
+				}
+				uncompletedSnapshots[*fastRestore.SnapshotId] = struct{}{}
+			}
+		}
+		pendingSnapshots = uncompletedSnapshots
+	}
+}
+
+// DisableDataFSR disables FSR for data volume snapshots
+func (e *EC2Session) DisableDataFSR(snapshotsIDsMap map[string][]*string) error {
+	if len(snapshotsIDsMap) == 0 {
+		return nil
+	}
+
+	eg, _ := errgroup.WithContext(context.Background())
+
+	for availableZone := range snapshotsIDsMap {
+		targetAZ := availableZone
+		eg.Go(func() error {
+			resp, err := e.ec2.DisableFastSnapshotRestores(&ec2.DisableFastSnapshotRestoresInput{
+				AvailabilityZones: []*string{&targetAZ},
+				SourceSnapshotIds: snapshotsIDsMap[targetAZ],
+			})
+
+			if err != nil {
+				return errors.Trace(err)
+			}
+
+			if len(resp.Unsuccessful) > 0 {
+				log.Warn("not all snapshots disabled FSR", zap.String("available zone", targetAZ))
+				return errors.Errorf("Some snapshot fails to disable FSR for available zone %s, such as %s, error code is %v", targetAZ, *resp.Unsuccessful[0].SnapshotId, resp.Unsuccessful[0].FastSnapshotRestoreStateErrors)
+			}
+
+			log.Info("Disable FSR issued", zap.String("available zone", targetAZ))
+
+			return nil
+		})
+	}
+	return eg.Wait()
+}
+
+func fetchTargetSnapshots(meta *config.EBSBasedBRMeta, specifiedAZ string) map[string][]*string {
+	var sourceSnapshotIDs = make(map[string][]*string)
+
+	if len(meta.TiKVComponent.Stores) == 0 {
+		return sourceSnapshotIDs
+	}
+
+	for i := range meta.TiKVComponent.Stores {
+		store := meta.TiKVComponent.Stores[i]
+		for j := range store.Volumes {
+			oldVol := store.Volumes[j]
+			// Handle data volume snapshots only
+			if strings.Compare(oldVol.Type, "storage.data-dir") == 0 {
+				if specifiedAZ != "" {
+					sourceSnapshotIDs[specifiedAZ] = append(sourceSnapshotIDs[specifiedAZ], &oldVol.SnapshotID)
+				} else {
+					sourceSnapshotIDs[oldVol.VolumeAZ] = append(sourceSnapshotIDs[oldVol.VolumeAZ], &oldVol.SnapshotID)
+				}
+			}
+		}
+	}
+
+	return sourceSnapshotIDs
+}
+
 // CreateVolumes create volumes from snapshots
 // if err happens in the middle, return half-done result
 // returned map: store id -> old volume id -> new volume id
@@ -377,7 +523,7 @@ func (e *EC2Session) WaitVolumesCreated(volumeIDMap map[string]string, progress
 	for len(pendingVolumes) > 0 {
 		// check every 5 seconds
 		time.Sleep(5 * time.Second)
-		log.Info("check pending snapshots", zap.Int("count", len(pendingVolumes)))
+		log.Info("check pending volumes", zap.Int("count", len(pendingVolumes)))
 		resp, err := e.ec2.DescribeVolumes(&ec2.DescribeVolumesInput{
 			VolumeIds: pendingVolumes,
 		})

br/pkg/config/ebs.go

@@ -100,6 +100,14 @@ func (c *EBSBasedBRMeta) GetStoreCount() uint64 {
 	return uint64(len(c.TiKVComponent.Stores))
 }
 
+func (c *EBSBasedBRMeta) GetTiKVVolumeCount() uint64 {
+	if c.TiKVComponent == nil || len(c.TiKVComponent.Stores) == 0 {
+		return 0
+	}
+	// Assume TiKV nodes are symmetric
+	return uint64(len(c.TiKVComponent.Stores[0].Volumes))
+}
+
 func (c *EBSBasedBRMeta) String() string {
 	cfg, err := json.Marshal(c)
 	if err != nil {

br/pkg/lightning/backend/backend.go

@@ -106,14 +106,15 @@ type ExternalEngineConfig struct {
 	StorageURI      string
 	DataFiles       []string
 	StatFiles       []string
-	MinKey          []byte
-	MaxKey          []byte
+	StartKey        []byte
+	EndKey          []byte
 	SplitKeys       [][]byte
 	RegionSplitSize int64
 	// TotalFileSize can be an estimated value.
 	TotalFileSize int64
 	// TotalKVCount can be an estimated value.
 	TotalKVCount int64
+	CheckHotspot bool
 }
 
 // CheckCtx contains all parameters used in CheckRequirements

br/pkg/lightning/backend/external/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "engine.go",
         "file.go",
         "iter.go",
+        "kv_buf.go",
         "kv_reader.go",
         "merge.go",
         "split.go",
@@ -32,14 +33,17 @@ go_library(
         "//pkg/sessionctx/variable",
         "//pkg/util/hack",
         "//pkg/util/logutil",
+        "//pkg/util/memory",
         "//pkg/util/size",
         "@com_github_cockroachdb_pebble//:pebble",
         "@com_github_docker_go_units//:go-units",
+        "@com_github_google_uuid//:uuid",
         "@com_github_jfcg_sorty_v2//:sorty",
         "@com_github_pingcap_errors//:errors",
         "@org_golang_x_sync//errgroup",
         "@org_uber_go_atomic//:atomic",
         "@org_uber_go_zap//:zap",
+        "@org_uber_go_zap//zapcore",
     ],
 )
 
@@ -53,19 +57,22 @@ go_test(
         "engine_test.go",
         "file_test.go",
         "iter_test.go",
+        "kv_buf_test.go",
         "split_test.go",
         "util_test.go",
         "writer_test.go",
     ],
     embed = [":external"],
     flaky = True,
-    shard_count = 40,
+    shard_count = 43,
     deps = [
         "//br/pkg/lightning/backend/kv",
         "//br/pkg/lightning/common",
+        "//br/pkg/membuf",
         "//br/pkg/storage",
         "//pkg/kv",
         "//pkg/util/codec",
+        "//pkg/util/logutil",
         "//pkg/util/size",
         "@com_github_aws_aws_sdk_go//aws",
         "@com_github_aws_aws_sdk_go//aws/credentials",