Merge remote-tracking branch 'upstream/master' into improve-bootstrap…

…-code

README.md

@@ -14,33 +14,29 @@
 
 TiDB ("Ti" stands for Titanium) is an open-source NewSQL database that supports Hybrid Transactional and Analytical Processing (HTAP) workloads. It is MySQL compatible and features horizontal scalability, strong consistency, and high availability.
 
-- __Horizontal Scalability__
+- [Key features](https://docs.pingcap.com/tidb/stable/overview#key-features)
+- [Architecture](#architecture)
+- [MySQL Compatibility](https://docs.pingcap.com/tidb/stable/mysql-compatibility)
 
-    TiDB expands both SQL processing and storage by simply adding new nodes. This makes infrastructure capacity planning both easier and more cost-effective than traditional relational databases which only scale vertically.
-
-- __MySQL Compatible Syntax__
-
-    TiDB acts like it is a MySQL 5.7 server to your applications. You can continue to use all of the existing MySQL client libraries, and in many cases, you will not need to change a single line of code in your application. Because TiDB is built from scratch, not a MySQL fork, please check out the list of [known compatibility differences](https://docs.pingcap.com/tidb/stable/mysql-compatibility).
-
-- __Distributed Transactions__
+For more details and latest updates, see [TiDB docs](https://docs.pingcap.com/tidb/stable) and [release notes](https://docs.pingcap.com/tidb/dev/release-notes).
 
-    TiDB internally shards table into small range-based chunks that we refer to as "Regions". Each Region defaults to approximately 100 MiB in size, and TiDB uses an [optimized](https://pingcap.com/blog/async-commit-the-accelerator-for-transaction-commit-in-tidb-5.0) Two-phase commit to ensure that Regions are maintained in a transactionally consistent way.
+## Quick start
 
-- __Cloud Native__
+### Start with TiDB Cloud
 
-    TiDB is designed to work in the cloud -- public, private, or hybrid -- making deployment, provisioning, operations, and maintenance simple.
+TiDB Cloud is the fully-managed service of TiDB, currently available on AWS and GCP.
 
-    The storage layer of TiDB, called TiKV, is a [Cloud Native Computing Foundation (CNCF) Graduated](https://www.cncf.io/announcements/2020/09/02/cloud-native-computing-foundation-announces-tikv-graduation/) project. The architecture of the TiDB platform also allows SQL processing and storage to be scaled independently of each other in a very cloud-friendly manner.
+Quickly check out TiDB Cloud with [a free trial](https://tidbcloud.com/signup).
 
-- __Minimize ETL__
+See [TiDB Cloud Quick Start Guide](https://docs.pingcap.com/tidbcloud/tidb-cloud-quickstart).
 
-    TiDB is designed to support both transaction processing (OLTP) and analytical processing (OLAP) workloads. This means that while you may have traditionally transacted on MySQL and then Extracted, Transformed and Loaded (ETL) data into a column store for analytical processing, this step is no longer required.
+### Start with TiDB
 
-- __High Availability__
+See [TiDB Quick Start Guide](https://docs.pingcap.com/tidb/stable/quick-start-with-tidb).
 
-    TiDB uses the Raft consensus algorithm to ensure that data is highly available and safely replicated throughout storage in Raft groups. In the event of failure, a Raft group will automatically elect a new leader for the failed member, and self-heal the TiDB cluster without any required manual intervention. Failure and self-healing operations are also transparent to applications.
+### Start developing TiDB
 
-For more details and latest updates, see [TiDB docs](https://docs.pingcap.com/tidb/stable) and [release notes](https://docs.pingcap.com/tidb/dev/release-notes).
+See [Get Started](https://pingcap.github.io/tidb-dev-guide/get-started/introduction.html) chapter of [TiDB Dev Guide](https://pingcap.github.io/tidb-dev-guide/index.html).
 
 ## Community
 
@@ -58,22 +54,6 @@ In addition, you may enjoy following:
 
 For support, please contact [PingCAP](http://bit.ly/contact_us_via_github).
 
-## Quick start
-
-### To start using TiDB Cloud
-
-We provide TiDB Cloud - a fully-managed Database as a Service for you. You can [sign up](https://tidbcloud.com/signup) and get started with TiDB Cloud Developer Tier for free.
-
-See [TiDB Cloud Quick Start](https://docs.pingcap.com/tidbcloud/tidb-cloud-quickstart).
-
-### To start using TiDB
-
-See [Quick Start Guide](https://docs.pingcap.com/tidb/stable/quick-start-with-tidb).
-
-### To start developing TiDB
-
-See [Get Started](https://pingcap.github.io/tidb-dev-guide/get-started/introduction.html) chapter of [TiDB Dev Guide](https://pingcap.github.io/tidb-dev-guide/index.html).
-
 ## Contributing
 
 The [community repository](https://github.com/pingcap/community) hosts all information about the TiDB community, including how to contribute to TiDB, how TiDB community is governed, how special interest groups are organized, etc.

br/cmd/br/stream.go

@@ -150,7 +150,7 @@ func newStreamCheckCommand() *cobra.Command {
 		Short: "get the metadata of log dir.",
 		Args:  cobra.NoArgs,
 		RunE: func(cmd *cobra.Command, args []string) error {
-			return streamCommand(cmd, task.StreamCheck)
+			return streamCommand(cmd, task.StreamMetadata)
 		},
 	}
 	return command
@@ -171,7 +171,7 @@ func streamCommand(command *cobra.Command, cmdName string) error {
 	}
 
 	switch cmdName {
-	case task.StreamCheck:
+	case task.StreamMetadata:
 		{
 			// do nothing.
 		}

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

@@ -52,6 +52,7 @@ import (
 	split "github.com/pingcap/tidb/br/pkg/restore"
 	"github.com/pingcap/tidb/br/pkg/utils"
 	"github.com/pingcap/tidb/br/pkg/version"
+	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/table"
@@ -1846,6 +1847,9 @@ func (local *local) EngineFileSizes() (res []backend.EngineFileSize) {
 	return
 }
 
+var getSplitConfFromStoreFunc = getSplitConfFromStore
+
+// return region split size, region split keys, error
 func getSplitConfFromStore(ctx context.Context, host string, tls *common.TLS) (int64, int64, error) {
 	var (
 		nested struct {
@@ -1866,6 +1870,7 @@ func getSplitConfFromStore(ctx context.Context, host string, tls *common.TLS) (i
 	return splitSize, nested.Coprocessor.RegionSplitKeys, nil
 }
 
+// return region split size, region split keys, error
 func getRegionSplitSizeKeys(ctx context.Context, cli pd.Client, tls *common.TLS) (int64, int64, error) {
 	stores, err := cli.GetAllStores(ctx, pd.WithExcludeTombstone())
 	if err != nil {
@@ -1875,11 +1880,16 @@ func getRegionSplitSizeKeys(ctx context.Context, cli pd.Client, tls *common.TLS)
 		if store.StatusAddress == "" || version.IsTiFlash(store) {
 			continue
 		}
-		regionSplitSize, regionSplitKeys, err := getSplitConfFromStore(ctx, store.StatusAddress, tls)
+		serverInfo := infoschema.ServerInfo{
+			Address:    store.Address,
+			StatusAddr: store.StatusAddress,
+		}
+		serverInfo.ResolveLoopBackAddr()
+		regionSplitSize, regionSplitKeys, err := getSplitConfFromStoreFunc(ctx, serverInfo.StatusAddr, tls)
 		if err == nil {
 			return regionSplitSize, regionSplitKeys, nil
 		}
-		log.L().Warn("get region split size and keys failed", zap.Error(err), zap.String("store", store.StatusAddress))
+		log.L().Warn("get region split size and keys failed", zap.Error(err), zap.String("store", serverInfo.StatusAddr))
 	}
 	return 0, 0, errors.New("get region split size and keys failed")
 }

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

@@ -23,6 +23,7 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -45,6 +46,7 @@ import (
 	"github.com/pingcap/tidb/br/pkg/mock"
 	"github.com/pingcap/tidb/br/pkg/pdutil"
 	"github.com/pingcap/tidb/br/pkg/restore"
+	"github.com/pingcap/tidb/br/pkg/utils"
 	"github.com/pingcap/tidb/br/pkg/version"
 	tidbkv "github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/sessionctx/stmtctx"
@@ -1200,3 +1202,36 @@ func TestLocalWriteAndIngestPairsFailFast(t *testing.T) {
 	require.Error(t, err)
 	require.Regexp(t, "The available disk of TiKV.*", err.Error())
 }
+
+func TestGetRegionSplitSizeKeys(t *testing.T) {
+	allStores := []*metapb.Store{
+		{
+			Address:       "172.16.102.1:20160",
+			StatusAddress: "0.0.0.0:20180",
+		},
+		{
+			Address:       "172.16.102.2:20160",
+			StatusAddress: "0.0.0.0:20180",
+		},
+		{
+			Address:       "172.16.102.3:20160",
+			StatusAddress: "0.0.0.0:20180",
+		},
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	cli := utils.FakePDClient{Stores: allStores}
+	defer func() {
+		getSplitConfFromStoreFunc = getSplitConfFromStore
+	}()
+	getSplitConfFromStoreFunc = func(ctx context.Context, host string, tls *common.TLS) (int64, int64, error) {
+		if strings.Contains(host, "172.16.102.3:20180") {
+			return int64(1), int64(2), nil
+		}
+		return 0, 0, errors.New("invalid connection")
+	}
+	splitSize, splitKeys, err := getRegionSplitSizeKeys(ctx, cli, nil)
+	require.NoError(t, err)
+	require.Equal(t, int64(1), splitSize)
+	require.Equal(t, int64(2), splitKeys)
+}

br/pkg/restore/client.go

@@ -1452,14 +1452,16 @@ func (rc *Client) GetRebasedTables() map[UniqueTableName]bool {
 func (rc *Client) PreCheckTableTiFlashReplica(
 	ctx context.Context,
 	tables []*metautil.Table,
+	skipTiflash bool,
 ) error {
 	tiFlashStores, err := conn.GetAllTiKVStores(ctx, rc.pdClient, conn.TiFlashOnly)
 	if err != nil {
 		return errors.Trace(err)
 	}
 	tiFlashStoreCount := len(tiFlashStores)
 	for _, table := range tables {
-		if table.Info.TiFlashReplica != nil && table.Info.TiFlashReplica.Count > uint64(tiFlashStoreCount) {
+		if skipTiflash ||
+			(table.Info.TiFlashReplica != nil && table.Info.TiFlashReplica.Count > uint64(tiFlashStoreCount)) {
 			// we cannot satisfy TiFlash replica in restore cluster. so we should
 			// set TiFlashReplica to unavailable in tableInfo, to avoid TiDB cannot sense TiFlash and make plan to TiFlash
 			// see details at https://github.com/pingcap/br/issues/931
@@ -1681,7 +1683,11 @@ func (rc *Client) RestoreKVFiles(
 					summary.CollectInt("File", 1)
 					log.Info("import files done", zap.String("name", file.Path), zap.Duration("take", time.Since(fileStart)))
 				}()
-				return rc.fileImporter.ImportKVFiles(ectx, file, rule, rc.restoreTS)
+				startTS := rc.startTS
+				if file.Cf == stream.DefaultCF {
+					startTS = rc.shiftStartTS
+				}
+				return rc.fileImporter.ImportKVFiles(ectx, file, rule, startTS, rc.restoreTS)
 			})
 		}
 	}
@@ -1795,7 +1801,7 @@ func (rc *Client) RestoreMetaKVFiles(
 ) error {
 	filesInWriteCF := make([]*backuppb.DataFileInfo, 0, len(files))
 
-	// The k-v envets in default CF should be restored firstly. The reason is that:
+	// The k-v events in default CF should be restored firstly. The reason is that:
 	// The error of transactions of meta will happen,
 	// if restore default CF events successfully, but failed to restore write CF events.
 	for _, f := range files {
@@ -1804,6 +1810,13 @@ func (rc *Client) RestoreMetaKVFiles(
 			continue
 		}
 
+		if f.Type == backuppb.FileType_Delete {
+			// this should happen abnormally.
+			// only do some preventive checks here.
+			log.Warn("detected delete file of meta key, skip it", zap.Any("file", f))
+			continue
+		}
+
 		err := rc.RestoreMetaKVFile(ctx, f, schemasReplace)
 		if err != nil {
 			return errors.Trace(err)
@@ -1869,7 +1882,14 @@ func (rc *Client) RestoreMetaKVFile(
 		} else if file.Cf == stream.DefaultCF && ts < rc.shiftStartTS {
 			continue
 		}
-
+		if len(txnEntry.Value) == 0 {
+			// we might record duplicated prewrite keys in some conor cases.
+			// the first prewrite key has the value but the second don't.
+			// so we can ignore the empty value key.
+			// see details at https://github.com/pingcap/tiflow/issues/5468.
+			log.Warn("txn entry is null", zap.Uint64("key-ts", ts), zap.ByteString("tnxKey", txnEntry.Key))
+			continue
+		}
 		log.Debug("txn entry", zap.Uint64("key-ts", ts), zap.Int("txnKey-len", len(txnEntry.Key)),
 			zap.Int("txnValue-len", len(txnEntry.Value)), zap.ByteString("txnKey", txnEntry.Key))
 		newEntry, err := sr.RewriteKvEntry(&txnEntry, file.Cf)
@@ -1948,7 +1968,7 @@ func (rc *Client) UpdateSchemaVersion(ctx context.Context) error {
 		func(ctx context.Context, txn kv.Transaction) error {
 			t := meta.NewMeta(txn)
 			var e error
-			schemaVersion, e = t.GenSchemaVersion()
+			schemaVersion, e = t.GenSchemaVersions(128)
 			return e
 		},
 	); err != nil {

br/pkg/restore/client_test.go

@@ -223,7 +223,7 @@ func TestPreCheckTableTiFlashReplicas(t *testing.T) {
 		}
 	}
 	ctx := context.Background()
-	require.Nil(t, client.PreCheckTableTiFlashReplica(ctx, tables))
+	require.Nil(t, client.PreCheckTableTiFlashReplica(ctx, tables, false))
 
 	for i := 0; i < len(tables); i++ {
 		if i == 0 || i > 2 {
@@ -234,4 +234,9 @@ func TestPreCheckTableTiFlashReplicas(t *testing.T) {
 			require.Equal(t, i, obtainCount)
 		}
 	}
+
+	require.Nil(t, client.PreCheckTableTiFlashReplica(ctx, tables, true))
+	for i := 0; i < len(tables); i++ {
+		require.Nil(t, tables[i].Info.TiFlashReplica)
+	}
 }

br/pkg/restore/import.go

@@ -335,11 +335,12 @@ func (importer *FileImporter) ImportKVFileForRegion(
 	ctx context.Context,
 	file *backuppb.DataFileInfo,
 	rule *RewriteRules,
-	restoreTs uint64,
+	startTS uint64,
+	restoreTS uint64,
 	info *RegionInfo,
 ) RPCResult {
 	// Try to download file.
-	result := importer.downloadAndApplyKVFile(ctx, file, rule, info, restoreTs)
+	result := importer.downloadAndApplyKVFile(ctx, file, rule, info, startTS, restoreTS)
 	if !result.OK() {
 		errDownload := result.Err
 		for _, e := range multierr.Errors(errDownload) {
@@ -380,11 +381,13 @@ func (importer *FileImporter) ClearFiles(ctx context.Context, pdClient pd.Client
 	return nil
 }
 
+// ImportKVFiles restores the kv events.
 func (importer *FileImporter) ImportKVFiles(
 	ctx context.Context,
 	file *backuppb.DataFileInfo,
 	rule *RewriteRules,
-	restoreTs uint64,
+	startTS uint64,
+	restoreTS uint64,
 ) error {
 	startTime := time.Now()
 	log.Debug("import kv files", zap.String("file", file.Path))
@@ -401,7 +404,7 @@ func (importer *FileImporter) ImportKVFiles(
 	rs := utils.InitialRetryState(32, 100*time.Millisecond, 8*time.Second)
 	ctl := OverRegionsInRange(startKey, endKey, importer.metaClient, &rs)
 	err = ctl.Run(ctx, func(ctx context.Context, r *RegionInfo) RPCResult {
-		return importer.ImportKVFileForRegion(ctx, file, rule, restoreTs, r)
+		return importer.ImportKVFileForRegion(ctx, file, rule, startTS, restoreTS, r)
 	})
 
 	log.Debug("download and apply file done",
@@ -801,7 +804,8 @@ func (importer *FileImporter) downloadAndApplyKVFile(
 	file *backuppb.DataFileInfo,
 	rules *RewriteRules,
 	regionInfo *RegionInfo,
-	restoreTs uint64,
+	startTS uint64,
+	restoreTS uint64,
 ) RPCResult {
 	leader := regionInfo.Leader
 	if leader == nil {
@@ -823,12 +827,13 @@ func (importer *FileImporter) downloadAndApplyKVFile(
 		Name: file.Path,
 		Cf:   file.Cf,
 		// TODO fill the length
-		Length:    0,
-		IsDelete:  file.Type == backuppb.FileType_Delete,
-		RestoreTs: restoreTs,
-		StartKey:  regionInfo.Region.GetStartKey(),
-		EndKey:    regionInfo.Region.GetEndKey(),
-		Sha256:    file.GetSha256(),
+		Length:          0,
+		IsDelete:        file.Type == backuppb.FileType_Delete,
+		StartSnapshotTs: startTS,
+		RestoreTs:       restoreTS,
+		StartKey:        regionInfo.Region.GetStartKey(),
+		EndKey:          regionInfo.Region.GetEndKey(),
+		Sha256:          file.GetSha256(),
 	}
 
 	reqCtx := &kvrpcpb.Context{

br/pkg/restore/stream_metas.go

@@ -171,8 +171,6 @@ func swapAndOverrideFile(ctx context.Context, s storage.ExternalStorage, path st
 const (
 	// TruncateSafePointFileName is the filename that the ts(the log have been truncated) is saved into.
 	TruncateSafePointFileName = "v1_stream_trancate_safepoint.txt"
-	// GlobalCheckpointFileName is the filename that the ts(the global checkpoint) is saved into.
-	GlobalCheckpointFileName = "v1_stream_global_checkpoint.txt"
 )
 
 // GetTSFromFile gets the current truncate safepoint.