merge master

Signed-off-by: Cabinfever_B <cabinfeveroier@gmail.com>

DEPS.bzl

@@ -82,6 +82,14 @@ def go_deps():
         sum = "h1:wVe6/Ea46ZMeNkQjjBW6xcqyQA/j5e0D6GytH95g0gQ=",
         version = "v0.0.0-20180226025133-644b8db467af",
     )
+    go_repository(
+        name = "com_github_alecthomas_kingpin_v2",
+        build_file_proto_mode = "disable",
+        importpath = "github.com/alecthomas/kingpin/v2",
+        sum = "h1:ANLJcKmQm4nIaog7xdr/id6FM6zm5hHnfZrvtKPxqGg=",
+        version = "v2.3.1",
+    )
+
     go_repository(
         name = "com_github_alecthomas_template",
         build_file_proto_mode = "disable_global",
@@ -93,8 +101,8 @@ def go_deps():
         name = "com_github_alecthomas_units",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/alecthomas/units",
-        sum = "h1:UQZhZ2O0vMHr2cI+DC1Mbh0TJxzA3RcLoMsFw+aXw7E=",
-        version = "v0.0.0-20190924025748-f65c72e2690d",
+        sum = "h1:s6gZFSlWYmbqAuRjVTiNNhvNRfY2Wxp9nhfyel4rklc=",
+        version = "v0.0.0-20211218093645-b94a6e3cc137",
     )
     go_repository(
         name = "com_github_aleksi_gocov_xml",
@@ -2545,8 +2553,8 @@ def go_deps():
         name = "com_github_kr_pretty",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/kr/pretty",
-        sum = "h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=",
-        version = "v0.3.0",
+        sum = "h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=",
+        version = "v0.3.1",
     )
     go_repository(
         name = "com_github_kr_pty",
@@ -3412,6 +3420,14 @@ def go_deps():
         sum = "h1:49lOXmGaUpV9Fz3gd7TFZY106KVlPVa5jcYD1gaQf98=",
         version = "v0.0.0-20180916011732-0a3d74bf9ce4",
     )
+    go_repository(
+        name = "com_github_pkg_diff",
+        build_file_proto_mode = "disable",
+        importpath = "github.com/pkg/diff",
+        sum = "h1:aoZm08cpOy4WuID//EZDgcC4zIxODThtZNPirFr42+A=",
+        version = "v0.0.0-20210226163009-20ebb0f2a09e",
+    )
+
     go_repository(
         name = "com_github_pkg_errors",
         build_file_proto_mode = "disable_global",
@@ -3467,8 +3483,8 @@ def go_deps():
         name = "com_github_prometheus_client_golang",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/prometheus/client_golang",
-        sum = "h1:nJdhIvne2eSX/XRAFV9PcvFFRbrjbcTUj0VP62TMhnw=",
-        version = "v1.14.0",
+        sum = "h1:5fCgGYogn0hFdhyhLbw7hEsWxufKtY9klyvdNfFlFhM=",
+        version = "v1.15.0",
     )
     go_repository(
         name = "com_github_prometheus_client_model",
@@ -3481,8 +3497,8 @@ def go_deps():
         name = "com_github_prometheus_common",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/prometheus/common",
-        sum = "h1:oOyhkDq05hPZKItWVBkJ6g6AtGxi+fy7F4JvUV8uhsI=",
-        version = "v0.39.0",
+        sum = "h1:EKsfXEYo4JpWMHH5cg+KOUWeuJSov1Id8zGR8eeI1YM=",
+        version = "v0.42.0",
     )
     go_repository(
         name = "com_github_prometheus_procfs",
@@ -3605,8 +3621,8 @@ def go_deps():
         name = "com_github_rogpeppe_go_internal",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/rogpeppe/go-internal",
-        sum = "h1:/FiVV8dS/e+YqF2JvO3yXRFbBLTIuSDkuC7aBOAvL+k=",
-        version = "v1.6.1",
+        sum = "h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=",
+        version = "v1.9.0",
     )
     go_repository(
         name = "com_github_rs_cors",
@@ -4394,6 +4410,13 @@ def go_deps():
         sum = "h1:L8IbaI/W6h5Cwgh0n4zGeZpVK78r/jBf9ASurHo9+/o=",
         version = "v0.0.0-20200623134604-12b17a7ff502",
     )
+    go_repository(
+        name = "com_github_xhit_go_str2duration",
+        build_file_proto_mode = "disable",
+        importpath = "github.com/xhit/go-str2duration",
+        sum = "h1:BcV5u025cITWxEQKGWr1URRzrcXtu7uk8+luz3Yuhwc=",
+        version = "v1.2.0",
+    )
 
     go_repository(
         name = "com_github_xiang90_probing",

br/cmd/br/backup.go

@@ -77,6 +77,26 @@ func runBackupRawCommand(command *cobra.Command, cmdName string) error {
 	return nil
 }
 
+func runBackupTxnCommand(command *cobra.Command, cmdName string) error {
+	cfg := task.TxnKvConfig{Config: task.Config{LogProgress: HasLogFile()}}
+	if err := cfg.ParseBackupConfigFromFlags(command.Flags()); err != nil {
+		command.SilenceUsage = false
+		return errors.Trace(err)
+	}
+
+	ctx := GetDefaultContext()
+	if cfg.EnableOpenTracing {
+		var store *appdash.MemoryStore
+		ctx, store = trace.TracerStartSpan(ctx)
+		defer trace.TracerFinishSpan(ctx, store)
+	}
+	if err := task.RunBackupTxn(ctx, gluetikv.Glue{}, cmdName, &cfg); err != nil {
+		log.Error("failed to backup txn kv", zap.Error(err))
+		return errors.Trace(err)
+	}
+	return nil
+}
+
 // NewBackupCommand return a full backup subcommand.
 func NewBackupCommand() *cobra.Command {
 	command := &cobra.Command{
@@ -103,6 +123,7 @@ func NewBackupCommand() *cobra.Command {
 		newDBBackupCommand(),
 		newTableBackupCommand(),
 		newRawBackupCommand(),
+		newTxnBackupCommand(),
 	)
 
 	task.DefineBackupFlags(command.PersistentFlags())
@@ -170,3 +191,18 @@ func newRawBackupCommand() *cobra.Command {
 	task.DefineRawBackupFlags(command)
 	return command
 }
+
+// newTxnBackupCommand return a txn kv range backup subcommand.
+func newTxnBackupCommand() *cobra.Command {
+	command := &cobra.Command{
+		Use:   "txn",
+		Short: "(experimental) backup a txn kv range from TiKV cluster",
+		Args:  cobra.NoArgs,
+		RunE: func(command *cobra.Command, _ []string) error {
+			return runBackupTxnCommand(command, task.TxnBackupCmd)
+		},
+	}
+
+	task.DefineTxnBackupFlags(command)
+	return command
+}

br/cmd/br/restore.go

@@ -108,6 +108,26 @@ func runRestoreRawCommand(command *cobra.Command, cmdName string) error {
 	return nil
 }
 
+func runRestoreTxnCommand(command *cobra.Command, cmdName string) error {
+	cfg := task.Config{LogProgress: HasLogFile()}
+	if err := cfg.ParseFromFlags(command.Flags()); err != nil {
+		command.SilenceUsage = false
+		return errors.Trace(err)
+	}
+
+	ctx := GetDefaultContext()
+	if cfg.EnableOpenTracing {
+		var store *appdash.MemoryStore
+		ctx, store = trace.TracerStartSpan(ctx)
+		defer trace.TracerFinishSpan(ctx, store)
+	}
+	if err := task.RunRestoreTxn(GetDefaultContext(), gluetikv.Glue{}, cmdName, &cfg); err != nil {
+		log.Error("failed to restore txn kv", zap.Error(err))
+		return errors.Trace(err)
+	}
+	return nil
+}
+
 // NewRestoreCommand returns a restore subcommand.
 func NewRestoreCommand() *cobra.Command {
 	command := &cobra.Command{
@@ -132,6 +152,7 @@ func NewRestoreCommand() *cobra.Command {
 		newDBRestoreCommand(),
 		newTableRestoreCommand(),
 		newRawRestoreCommand(),
+		newTxnRestoreCommand(),
 		newStreamRestoreCommand(),
 	)
 	task.DefineRestoreFlags(command.PersistentFlags())
@@ -193,6 +214,20 @@ func newRawRestoreCommand() *cobra.Command {
 	return command
 }
 
+func newTxnRestoreCommand() *cobra.Command {
+	command := &cobra.Command{
+		Use:   "txn",
+		Short: "(experimental) restore txn kv to TiKV cluster",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			return runRestoreTxnCommand(cmd, task.TxnRestoreCmd)
+		},
+	}
+
+	task.DefineRawRestoreFlags(command)
+	return command
+}
+
 func newStreamRestoreCommand() *cobra.Command {
 	command := &cobra.Command{
 		Use:   "point",

br/pkg/backup/client.go

@@ -120,8 +120,8 @@ func (bc *Client) SetCipher(cipher *backuppb.CipherInfo) {
 	bc.cipher = cipher
 }
 
-// GetTS gets a new timestamp from PD.
-func (bc *Client) GetCurerntTS(ctx context.Context) (uint64, error) {
+// GetCurrentTS gets a new timestamp from PD.
+func (bc *Client) GetCurrentTS(ctx context.Context) (uint64, error) {
 	p, l, err := bc.mgr.GetPDClient().GetTS(ctx)
 	if err != nil {
 		return 0, errors.Trace(err)

br/pkg/lightning/backend/kv/base.go

@@ -116,6 +116,9 @@ type BaseKVEncoder struct {
 
 	logger      *zap.Logger
 	recordCache []types.Datum
+	// the first auto-generated ID in the current encoder.
+	// if there's no auto-generated id column or the column value is not auto-generated, it will be 0.
+	LastInsertID uint64
 }
 
 // NewBaseKVEncoder creates a new BaseKVEncoder.
@@ -241,6 +244,9 @@ func (e *BaseKVEncoder) getActualDatum(col *table.Column, rowID int64, inputDatu
 		// we still need a conversion, e.g. to catch overflow with a TINYINT column.
 		value, err = table.CastValue(e.SessionCtx,
 			types.NewIntDatum(rowID), col.ToInfo(), false, false)
+		if err == nil && e.LastInsertID == 0 {
+			e.LastInsertID = value.GetUint64()
+		}
 	case e.IsAutoRandomCol(col.ToInfo()):
 		var val types.Datum
 		realRowID := e.AutoIDFn(rowID)
@@ -250,6 +256,9 @@ func (e *BaseKVEncoder) getActualDatum(col *table.Column, rowID int64, inputDatu
 			val = types.NewIntDatum(realRowID)
 		}
 		value, err = table.CastValue(e.SessionCtx, val, col.ToInfo(), false, false)
+		if err == nil && e.LastInsertID == 0 {
+			e.LastInsertID = value.GetUint64()
+		}
 	case col.IsGenerated():
 		// inject some dummy value for gen col so that MutRowFromDatums below sees a real value instead of nil.
 		// if MutRowFromDatums sees a nil it won't initialize the underlying storage and cause SetDatum to panic.

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

@@ -34,6 +34,7 @@ go_library(
         "//br/pkg/lightning/metric",
         "//br/pkg/lightning/mydump",
         "//br/pkg/lightning/tikv",
+        "//br/pkg/lightning/verification",
         "//br/pkg/logutil",
         "//br/pkg/membuf",
         "//br/pkg/pdutil",

br/pkg/lightning/backend/local/checksum.go

@@ -30,6 +30,7 @@ import (
 	"github.com/pingcap/tidb/br/pkg/lightning/common"
 	"github.com/pingcap/tidb/br/pkg/lightning/log"
 	"github.com/pingcap/tidb/br/pkg/lightning/metric"
+	"github.com/pingcap/tidb/br/pkg/lightning/verification"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/util/mathutil"
 	"github.com/pingcap/tipb/go-tipb"
@@ -60,6 +61,13 @@ type RemoteChecksum struct {
 	TotalBytes uint64
 }
 
+// IsEqual checks whether the checksum is equal to the other.
+func (rc *RemoteChecksum) IsEqual(other *verification.KVChecksum) bool {
+	return rc.Checksum == other.Sum() &&
+		rc.TotalKVs == other.SumKVS() &&
+		rc.TotalBytes == other.SumSize()
+}
+
 // ChecksumManager is a manager that manages checksums.
 type ChecksumManager interface {
 	Checksum(ctx context.Context, tableInfo *checkpoints.TidbTableInfo) (*RemoteChecksum, error)

br/pkg/lightning/backend/local/local.go

@@ -1726,6 +1726,11 @@ func (local *Backend) EngineFileSizes() (res []backend.EngineFileSize) {
 	return
 }
 
+// GetPDClient returns the PD client.
+func (local *Backend) GetPDClient() pd.Client {
+	return local.pdCtl.GetPDClient()
+}
+
 var getSplitConfFromStoreFunc = getSplitConfFromStore
 
 // return region split size, region split keys, error

br/pkg/lightning/importer/BUILD.bazel

@@ -32,6 +32,7 @@ go_library(
         "//br/pkg/lightning/log",
         "//br/pkg/lightning/metric",
         "//br/pkg/lightning/mydump",
+        "//br/pkg/lightning/precheck",
         "//br/pkg/lightning/tikv",
         "//br/pkg/lightning/verification",
         "//br/pkg/lightning/web",
@@ -122,6 +123,7 @@ go_test(
         "//br/pkg/lightning/log",
         "//br/pkg/lightning/metric",
         "//br/pkg/lightning/mydump",
+        "//br/pkg/lightning/precheck",
         "//br/pkg/lightning/verification",
         "//br/pkg/lightning/web",
         "//br/pkg/lightning/worker",