Merge branch 'master' into fix-37412

Makefile

@@ -363,15 +363,16 @@ endif
 #	make bench-daily TO=/path/to/file.json
 .PHONY: bench-daily
 bench-daily:
-	go test github.com/pingcap/tidb/pkg/session -run TestBenchDaily -bench Ignore --outfile bench_daily.json
+	go test github.com/pingcap/tidb/pkg/distsql -run TestBenchDaily -bench Ignore --outfile bench_daily.json
 	go test github.com/pingcap/tidb/pkg/executor -run TestBenchDaily -bench Ignore --outfile bench_daily.json
 	go test github.com/pingcap/tidb/pkg/executor/test/splittest -run TestBenchDaily -bench Ignore --outfile bench_daily.json
-	go test github.com/pingcap/tidb/pkg/tablecodec -run TestBenchDaily -bench Ignore --outfile bench_daily.json
 	go test github.com/pingcap/tidb/pkg/expression -run TestBenchDaily -bench Ignore --outfile bench_daily.json
-	go test github.com/pingcap/tidb/pkg/util/rowcodec -run TestBenchDaily -bench Ignore --outfile bench_daily.json
-	go test github.com/pingcap/tidb/pkg/util/codec -run TestBenchDaily -bench Ignore --outfile bench_daily.json
-	go test github.com/pingcap/tidb/pkg/distsql -run TestBenchDaily -bench Ignore --outfile bench_daily.json
+	go test github.com/pingcap/tidb/pkg/planner/core/tests/partition -run TestBenchDaily -bench Ignore --outfile bench_daily.json
+	go test github.com/pingcap/tidb/pkg/session -run TestBenchDaily -bench Ignore --outfile bench_daily.json
 	go test github.com/pingcap/tidb/pkg/statistics -run TestBenchDaily -bench Ignore --outfile bench_daily.json
+	go test github.com/pingcap/tidb/pkg/tablecodec -run TestBenchDaily -bench Ignore --outfile bench_daily.json
+	go test github.com/pingcap/tidb/pkg/util/codec -run TestBenchDaily -bench Ignore --outfile bench_daily.json
+	go test github.com/pingcap/tidb/pkg/util/rowcodec -run TestBenchDaily -bench Ignore --outfile bench_daily.json
 	go test github.com/pingcap/tidb/pkg/util/benchdaily -run TestBenchDaily -bench Ignore \
 		-date `git log -n1 --date=unix --pretty=format:%cd` \
 		-commit `git log -n1 --pretty=format:%h` \
@@ -565,7 +566,15 @@ dumpling_bins:
 
 .PHONY: generate_grafana_scripts
 generate_grafana_scripts:
-	@cd metrics/grafana && mv tidb_summary.json tidb_summary.json.committed && ./generate_json.sh && diff -u tidb_summary.json.committed tidb_summary.json && rm tidb_summary.json.committed
+	@cd pkg/metrics/grafana && \
+	  mv tidb_summary.json tidb_summary.json.committed && \
+	  mv tidb_resource_control.json tidb_resource_control.json.committed && \
+	  ./generate_json.sh && \
+	  diff -u tidb_summary.json.committed tidb_summary.json && \
+	  diff -u tidb_resource_control.json.committed tidb_resource_control.json && \
+	  rm tidb_summary.json.committed && \
+	  rm tidb_resource_control.json.committed
+
 
 .PHONY: bazel_ci_prepare
 bazel_ci_prepare:
@@ -623,6 +632,13 @@ bazel_coverage_test: failpoint-enable bazel_ci_simple_prepare
 		-- //... -//cmd/... -//tests/graceshutdown/... \
 		-//tests/globalkilltest/... -//tests/readonlytest/... -//br/pkg/task:task_test -//tests/realtikvtest/...
 
+.PHONY: bazel_coverage_test_ddlargsv1
+bazel_coverage_test_ddlargsv1: failpoint-enable bazel_ci_simple_prepare
+	bazel $(BAZEL_GLOBAL_CONFIG) --nohome_rc coverage $(BAZEL_CMD_CONFIG) $(BAZEL_INSTRUMENTATION_FILTER) --jobs=35 --build_tests_only --test_keep_going=false \
+		--@io_bazel_rules_go//go/config:cover_format=go_cover --define gotags=deadlock,intest,ddlargsv1 \
+		-- //... -//cmd/... -//tests/graceshutdown/... \
+		-//tests/globalkilltest/... -//tests/readonlytest/... -//br/pkg/task:task_test -//tests/realtikvtest/...
+
 .PHONY: bazel_build
 bazel_build:
 	mkdir -p bin

OWNERS

@@ -98,6 +98,7 @@ approvers:
   - winoros
   - WizardXiao
   - wjhuang2016
+  - wk989898
   - wshwsh12
   - xhebox
   - xiongjiwei
@@ -127,6 +128,7 @@ reviewers:
   - dhysum
   - fengou1
   - fzzf678
+  - ghazalfamilyusa
   - iamxy
   - JmPotato
   - js00070
@@ -145,6 +147,7 @@ reviewers:
   - shihongzhi
   - spongedu
   - tangwz
+  - terry1purcell
   - Tjianke
   - TonsnakeLin
   - tsthght

OWNERS_ALIASES

@@ -2,13 +2,13 @@
 aliases:
   sig-critical-approvers-tidb-server:
     - yudongusa
-    - easonn7
+    - BenMeadowcroft
   sig-critical-approvers-tidb-lightning:
     - yudongusa
-    - easonn7
+    - BenMeadowcroft
   sig-critical-approvers-parser:
     - yudongusa
-    - easonn7
+    - BenMeadowcroft
   sig-approvers-autoid-service: # approvers for auto-id service
     - bb7133
     - tiancaiamao
@@ -76,6 +76,8 @@ aliases:
     - gmhdbjd
     - tangenta
     - wjhuang2016
+    - D3Hunter
+    - lance6716
   sig-approvers-owner: # approvers for `owner` pkg
     - Benjamin2037
     - tangenta

WORKSPACE

@@ -37,12 +37,12 @@ versions.check(minimum_bazel_version = "6.0.0")
 
 http_archive(
     name = "io_bazel_rules_go",
-    sha256 = "67b4d1f517ba73e0a92eb2f57d821f2ddc21f5bc2bd7a231573f11bd8758192e",
+    sha256 = "f4a9314518ca6acfa16cc4ab43b0b8ce1e4ea64b81c38d8a3772883f153346b8",
     urls = [
-        "http://bazel-cache.pingcap.net:8080/bazelbuild/rules_go/releases/download/v0.50.0/rules_go-v0.50.0.zip",
-        "http://ats.apps.svc/bazelbuild/rules_go/releases/download/v0.50.0/rules_go-v0.50.0.zip",
-        "https://mirror.bazel.build/github.com/bazelbuild/rules_go/releases/download/v0.50.0/rules_go-v0.50.0.zip",
-        "https://github.com/bazelbuild/rules_go/releases/download/v0.50.0/rules_go-v0.50.0.zip",
+        "http://bazel-cache.pingcap.net:8080/bazelbuild/rules_go/releases/download/v0.50.1/rules_go-v0.50.1.zip",
+        "http://ats.apps.svc/bazelbuild/rules_go/releases/download/v0.50.1/rules_go-v0.50.1.zip",
+        "https://mirror.bazel.build/github.com/bazelbuild/rules_go/releases/download/v0.50.1/rules_go-v0.50.1.zip",
+        "https://github.com/bazelbuild/rules_go/releases/download/v0.50.1/rules_go-v0.50.1.zip",
     ],
 )
 

br/cmd/br/restore.go

@@ -74,14 +74,14 @@ func runRestoreCommand(command *cobra.Command, cmdName string) error {
 
 	if err := task.RunRestore(GetDefaultContext(), tidbGlue, cmdName, &cfg); err != nil {
 		log.Error("failed to restore", zap.Error(err))
-		printWorkaroundOnFullRestoreError(command, err)
+		printWorkaroundOnFullRestoreError(err)
 		return errors.Trace(err)
 	}
 	return nil
 }
 
 // print workaround when we met not fresh or incompatible cluster error on full cluster restore
-func printWorkaroundOnFullRestoreError(command *cobra.Command, err error) {
+func printWorkaroundOnFullRestoreError(err error) {
 	if !errors.ErrorEqual(err, berrors.ErrRestoreNotFreshCluster) &&
 		!errors.ErrorEqual(err, berrors.ErrRestoreIncompatibleSys) {
 		return

br/pkg/aws/BUILD.bazel

@@ -8,10 +8,13 @@ go_library(
     deps = [
         "//br/pkg/config",
         "//br/pkg/glue",
+        "//br/pkg/logutil",
         "//br/pkg/utils",
         "//pkg/util",
         "@com_github_aws_aws_sdk_go//aws",
         "@com_github_aws_aws_sdk_go//aws/awserr",
+        "@com_github_aws_aws_sdk_go//aws/client",
+        "@com_github_aws_aws_sdk_go//aws/request",
         "@com_github_aws_aws_sdk_go//aws/session",
         "@com_github_aws_aws_sdk_go//service/cloudwatch",
         "@com_github_aws_aws_sdk_go//service/ec2",

br/pkg/aws/ebs.go

@@ -11,6 +11,8 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/aws/client"
+	"github.com/aws/aws-sdk-go/aws/request"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/cloudwatch"
 	"github.com/aws/aws-sdk-go/service/ec2"
@@ -19,6 +21,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/br/pkg/config"
 	"github.com/pingcap/tidb/br/pkg/glue"
+	"github.com/pingcap/tidb/br/pkg/logutil"
 	"github.com/pingcap/tidb/br/pkg/utils"
 	"github.com/pingcap/tidb/pkg/util"
 	"go.uber.org/atomic"
@@ -41,11 +44,53 @@ type EC2Session struct {
 
 type VolumeAZs map[string]string
 
+type ebsBackupRetryer struct {
+	delegate request.Retryer
+}
+
+func (e *ebsBackupRetryer) MaxRetries() int {
+	return e.delegate.MaxRetries()
+}
+
+var backOffTimeOverride = map[string]time.Duration{
+	// From the SDK:
+	// Sadly it seems there isn't an exported operation name...
+	// const opCreateSnapshots = "CreateSnapshots"
+	// The quota for create snapshots is 5 per minute.
+	// Back off for a longer time so we won't excced it.
+	"CreateSnapshots": 20 * time.Second,
+	// const opCreateVolume = "CreateVolume"
+	"CreateVolume": 20 * time.Second,
+}
+
+func (e *ebsBackupRetryer) RetryRules(r *request.Request) time.Duration {
+	backOff := e.delegate.RetryRules(r)
+	if override, ok := backOffTimeOverride[r.Operation.Name]; ok {
+		backOff = max(override, backOff)
+	}
+	log.Warn(
+		"Retrying an operation.",
+		logutil.ShortError(r.Error),
+		zap.Duration("backoff", backOff),
+		zap.StackSkip("stack", 1),
+	)
+	return backOff
+}
+
+func (e *ebsBackupRetryer) ShouldRetry(r *request.Request) bool {
+	return e.delegate.ShouldRetry(r)
+}
+
 func NewEC2Session(concurrency uint, region string) (*EC2Session, error) {
 	// aws-sdk has builtin exponential backoff retry mechanism, see:
 	// https://github.com/aws/aws-sdk-go/blob/db4388e8b9b19d34dcde76c492b17607cd5651e2/aws/client/default_retryer.go#L12-L16
 	// with default retryer & max-retry=9, we will wait for at least 30s in total
 	awsConfig := aws.NewConfig().WithMaxRetries(9).WithRegion(region)
+	defRetry := new(client.DefaultRetryer)
+	ourRetry := ebsBackupRetryer{
+		delegate: defRetry,
+	}
+	awsConfig.Retryer = ourRetry
 	// TiDB Operator need make sure we have the correct permission to call aws api(through aws env variables)
 	// we may change this behaviour in the future.
 	sessionOptions := session.Options{Config: *awsConfig}

br/pkg/backup/client.go

@@ -725,7 +725,7 @@ func BuildBackupRangeAndInitSchema(
 	buildRange bool,
 ) ([]rtree.Range, *Schemas, []*backuppb.PlacementPolicy, error) {
 	snapshot := storage.GetSnapshot(kv.NewVersion(backupTS))
-	m := meta.NewSnapshotMeta(snapshot)
+	m := meta.NewReader(snapshot)
 
 	var policies []*backuppb.PlacementPolicy
 	if isFullBackup {
@@ -821,7 +821,7 @@ func BuildBackupSchemas(
 	fn func(dbInfo *model.DBInfo, tableInfo *model.TableInfo),
 ) error {
 	snapshot := storage.GetSnapshot(kv.NewVersion(backupTS))
-	m := meta.NewSnapshotMeta(snapshot)
+	m := meta.NewReader(snapshot)
 
 	dbs, err := m.ListDatabases()
 	if err != nil {
@@ -936,37 +936,6 @@ func BuildBackupSchemas(
 	return nil
 }
 
-// BuildFullSchema builds a full backup schemas for databases and tables.
-func BuildFullSchema(storage kv.Storage, backupTS uint64, fn func(dbInfo *model.DBInfo, tableInfo *model.TableInfo)) error {
-	snapshot := storage.GetSnapshot(kv.NewVersion(backupTS))
-	m := meta.NewSnapshotMeta(snapshot)
-
-	dbs, err := m.ListDatabases()
-	if err != nil {
-		return errors.Trace(err)
-	}
-
-	for _, db := range dbs {
-		hasTable := false
-		err = m.IterTables(db.ID, func(table *model.TableInfo) error {
-			// add table
-			fn(db, table)
-			hasTable = true
-			return nil
-		})
-		if err != nil {
-			return errors.Trace(err)
-		}
-
-		// backup this empty db if this schema is empty.
-		if !hasTable {
-			fn(db, nil)
-		}
-	}
-
-	return nil
-}
-
 func skipUnsupportedDDLJob(job *model.Job) bool {
 	switch job.Type {
 	// TiDB V5.3.0 supports TableAttributes and TablePartitionAttributes.
@@ -988,9 +957,9 @@ func skipUnsupportedDDLJob(job *model.Job) bool {
 // WriteBackupDDLJobs sends the ddl jobs are done in (lastBackupTS, backupTS] to metaWriter.
 func WriteBackupDDLJobs(metaWriter *metautil.MetaWriter, g glue.Glue, store kv.Storage, lastBackupTS, backupTS uint64, needDomain bool) error {
 	snapshot := store.GetSnapshot(kv.NewVersion(backupTS))
-	snapMeta := meta.NewSnapshotMeta(snapshot)
+	snapMeta := meta.NewReader(snapshot)
 	lastSnapshot := store.GetSnapshot(kv.NewVersion(lastBackupTS))
-	lastSnapMeta := meta.NewSnapshotMeta(lastSnapshot)
+	lastSnapMeta := meta.NewReader(lastSnapshot)
 	lastSchemaVersion, err := lastSnapMeta.GetSchemaVersionWithNonEmptyDiff()
 	if err != nil {
 		return errors.Trace(err)
@@ -1033,10 +1002,10 @@ func WriteBackupDDLJobs(metaWriter *metautil.MetaWriter, g glue.Glue, store kv.S
 		return appendJobs, false
 	}
 
-	newestMeta := meta.NewSnapshotMeta(store.GetSnapshot(kv.NewVersion(version.Ver)))
+	newestMeta := meta.NewReader(store.GetSnapshot(kv.NewVersion(version.Ver)))
 	var allJobs []*model.Job
 	err = g.UseOneShotSession(store, !needDomain, func(se glue.Session) error {
-		allJobs, err = ddl.GetAllDDLJobs(se.GetSessionCtx())
+		allJobs, err = ddl.GetAllDDLJobs(context.Background(), se.GetSessionCtx())
 		if err != nil {
 			return errors.Trace(err)
 		}

br/pkg/checkpoint/BUILD.bazel

@@ -5,23 +5,30 @@ go_library(
     srcs = [
         "backup.go",
         "checkpoint.go",
+        "external_storage.go",
         "log_restore.go",
         "restore.go",
+        "storage.go",
         "ticker.go",
     ],
     importpath = "github.com/pingcap/tidb/br/pkg/checkpoint",
     visibility = ["//visibility:public"],
     deps = [
+        "//br/pkg/glue",
         "//br/pkg/logutil",
         "//br/pkg/metautil",
         "//br/pkg/pdutil",
         "//br/pkg/rtree",
         "//br/pkg/storage",
         "//br/pkg/summary",
         "//br/pkg/utils",
+        "//pkg/domain",
+        "//pkg/kv",
         "//pkg/meta/model",
         "//pkg/parser/model",
         "//pkg/util",
+        "//pkg/util/sqlexec",
+        "@com_github_google_uuid//:uuid",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_pingcap_kvproto//pkg/brpb",
@@ -38,11 +45,13 @@ go_test(
     srcs = ["checkpoint_test.go"],
     flaky = True,
     race = "on",
-    shard_count = 5,
+    shard_count = 6,
     deps = [
         ":checkpoint",
+        "//br/pkg/gluetidb",
         "//br/pkg/pdutil",
         "//br/pkg/storage",
+        "//br/pkg/utiltest",
         "//pkg/meta/model",
         "//pkg/parser/model",
         "@com_github_pingcap_kvproto//pkg/brpb",

br/pkg/checkpoint/backup.go

@@ -57,13 +57,13 @@ func StartCheckpointBackupRunnerForTest(
 	tick time.Duration,
 	timer GlobalTimer,
 ) (*CheckpointRunner[BackupKeyType, BackupValueType], error) {
-	runner := newCheckpointRunner[BackupKeyType, BackupValueType](
-		ctx, storage, cipher, timer, flushPositionForBackup(), valueMarshalerForBackup)
-
-	err := runner.initialLock(ctx)
+	checkpointStorage, err := newExternalCheckpointStorage(ctx, storage, timer)
 	if err != nil {
-		return nil, errors.Annotate(err, "Failed to initialize checkpoint lock.")
+		return nil, errors.Trace(err)
 	}
+	runner := newCheckpointRunner[BackupKeyType, BackupValueType](
+		checkpointStorage, cipher, valueMarshalerForBackup)
+
 	runner.startCheckpointMainLoop(ctx, tick, tick, tick)
 	return runner, nil
 }
@@ -74,13 +74,13 @@ func StartCheckpointRunnerForBackup(
 	cipher *backuppb.CipherInfo,
 	timer GlobalTimer,
 ) (*CheckpointRunner[BackupKeyType, BackupValueType], error) {
-	runner := newCheckpointRunner[BackupKeyType, BackupValueType](
-		ctx, storage, cipher, timer, flushPositionForBackup(), valueMarshalerForBackup)
-
-	err := runner.initialLock(ctx)
+	checkpointStorage, err := newExternalCheckpointStorage(ctx, storage, timer)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
+	runner := newCheckpointRunner[BackupKeyType, BackupValueType](
+		checkpointStorage, cipher, valueMarshalerForBackup)
+
 	runner.startCheckpointMainLoop(
 		ctx,
 		defaultTickDurationForFlush,