Merge branch 'release-5.4' into release-5.4-381e870c5ca0

.github/licenserc.yml

@@ -28,4 +28,7 @@ header:
     - '.github/'
     - 'parser/'
     - 'dumpling/'
+    - '**/*.sql'
+    - "**/OWNERS"
+    - "OWNERS_ALIASES"
   comment: on-failure

OWNERS

@@ -0,0 +1,131 @@
+# See the OWNERS docs at https://go.k8s.io/owners
+approvers:
+  # tidb-maintainers:
+  - amyangfei
+  - bb7133
+  - breezewish
+  - c4pt0r
+  - cfzjywxk
+  - coocood
+  - disksing
+  - jackysp
+  - kennytm
+  - lance6716
+  - ngaut
+  - qiuyesuifeng
+  - qw4990
+  - shenli
+  - siddontang
+  - tangenta
+  - tiancaiamao
+  - winkyao
+  - winoros
+  - XuHuaiyu
+  - zimulala
+  - zz-jason
+  # tidb-committers:
+  - 3pointer
+  - AilinKid
+  - AndreMouche
+  - andylokandy
+  - b41sh
+  - Benjamin2037
+  - CbcWestwolf
+  - chrysan
+  - Deardrops
+  - Defined2014
+  - dveeden
+  - eurekaka
+  - glorv
+  - gozssky
+  - guo-shaoge
+  - hawkingrei
+  - lcwangchao
+  - Little-Wallace
+  - mjonss
+  - morgo
+  - MyonKeminta
+  - overvenus
+  - sticnarf
+  - windtalker
+  - wjhuang2016
+  - wshwsh12
+  - xhebox
+  - you06
+  - zanmato1984
+  - zyguan
+  # migration-maintainers:
+  - amyangfei
+  - kennytm
+  - lance6716
+  # migration-committers:
+  - 3AceShowHand
+  - 3pointer
+  - asddongmen
+  - CharlesCheung96
+  - csuzhangxc
+  - D3Hunter
+  - dsdashun
+  - Ehco1996
+  - glorv
+  - GMHDBJD
+  - gozssky
+  - hi-rustin
+  - hicqu
+  - IANTHEREAL
+  - Leavrth
+  - lichunzhu
+  - Little-Wallace
+  - niubell
+  - okJiang
+  - overvenus
+  - sdojjy
+  - suzaku
+  - YuJuncen
+reviewers:
+  # tidb-reviewers:
+  - aytrack
+  - Benjamin2037
+  - BornChanger
+  - CbcWestwolf
+  - Defined2014
+  - dveeden
+  - ekexium
+  - fixdb
+  - gozssky
+  - guo-shaoge
+  - hawkingrei
+  - hi-rustin
+  - JmPotato
+  - js00070
+  - mjonss
+  - mornyx
+  - nolouch
+  - pingyu
+  - rleungx
+  - SeaRise
+  - shihongzhi
+  - sylzd
+  - tangwz
+  - time-and-fate
+  - TonsnakeLin
+  - TszKitLo40
+  - xuyifangreeneyes
+  - YangKeao
+  - zhangjinpeng1987
+  - zhongzc
+  # migration-reviewers:
+  - 3AceShowHand
+  - asddongmen
+  - ben1009
+  - CharlesCheung96
+  - D3Hunter
+  - gozssky
+  - hicqu
+  - iamxy
+  - joccau
+  - niubell
+  - nongfushanquan
+  - okJiang
+  - sdojjy
+  - tiancaiamao

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

@@ -37,6 +37,8 @@ import (
 	"go.uber.org/zap"
 )
 
+const maxAvailableBufSize int = 20
+
 // invalidIterator is a trimmed down Iterator type which is invalid.
 type invalidIterator struct {
 	kv.Iterator
@@ -91,15 +93,33 @@ func (mb *kvMemBuf) Recycle(buf *bytesBuf) {
 	buf.idx = 0
 	buf.cap = len(buf.buf)
 	mb.Lock()
+	if len(mb.availableBufs) >= maxAvailableBufSize {
+		// too many byte buffers, evict one byte buffer and continue
+		evictedByteBuf := mb.availableBufs[0]
+		evictedByteBuf.destroy()
+		mb.availableBufs = mb.availableBufs[1:]
+	}
 	mb.availableBufs = append(mb.availableBufs, buf)
 	mb.Unlock()
 }
 
 func (mb *kvMemBuf) AllocateBuf(size int) {
 	mb.Lock()
 	size = utils.MaxInt(units.MiB, int(utils.NextPowerOfTwo(int64(size)))*2)
-	if len(mb.availableBufs) > 0 && mb.availableBufs[0].cap >= size {
-		mb.buf = mb.availableBufs[0]
+	var (
+		existingBuf    *bytesBuf
+		existingBufIdx int
+	)
+	for i, buf := range mb.availableBufs {
+		if buf.cap >= size {
+			existingBuf = buf
+			existingBufIdx = i
+			break
+		}
+	}
+	if existingBuf != nil {
+		mb.buf = existingBuf
+		mb.availableBufs[existingBufIdx] = mb.availableBufs[0]
 		mb.availableBufs = mb.availableBufs[1:]
 	} else {
 		mb.buf = newBytesBuf(size)

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

@@ -17,6 +17,7 @@ package kv
 import (
 	"testing"
 
+	"github.com/docker/go-units"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/tidb/parser/mysql"
 )
@@ -34,3 +35,103 @@ func (s *kvSuite) TestSession(c *C) {
 	_, err := session.Txn(true)
 	c.Assert(err, IsNil)
 }
+
+func (s *kvSuite) TestKVMemBufInterweaveAllocAndRecycle(c *C) {
+	type testCase struct {
+		AllocSizes                []int
+		FinalAvailableByteBufCaps []int
+	}
+	for _, tc := range []testCase{
+		{
+			AllocSizes: []int{
+				1 * units.MiB,
+				2 * units.MiB,
+				3 * units.MiB,
+				4 * units.MiB,
+				5 * units.MiB,
+			},
+			// [2] => [2,4] => [2,4,8] => [4,2,8] => [4,2,8,16]
+			FinalAvailableByteBufCaps: []int{
+				4 * units.MiB,
+				2 * units.MiB,
+				8 * units.MiB,
+				16 * units.MiB,
+			},
+		},
+		{
+			AllocSizes: []int{
+				5 * units.MiB,
+				4 * units.MiB,
+				3 * units.MiB,
+				2 * units.MiB,
+				1 * units.MiB,
+			},
+			// [16] => [16] => [16] => [16] => [16]
+			FinalAvailableByteBufCaps: []int{16 * units.MiB},
+		},
+		{
+			AllocSizes: []int{5, 4, 3, 2, 1},
+			// [1] => [1] => [1] => [1] => [1]
+			FinalAvailableByteBufCaps: []int{1 * units.MiB},
+		},
+		{
+			AllocSizes: []int{
+				1 * units.MiB,
+				2 * units.MiB,
+				3 * units.MiB,
+				2 * units.MiB,
+				1 * units.MiB,
+				5 * units.MiB,
+			},
+			// [2] => [2,4] => [2,4,8] => [2,8,4] => [8,4,2] => [8,4,2,16]
+			FinalAvailableByteBufCaps: []int{
+				8 * units.MiB,
+				4 * units.MiB,
+				2 * units.MiB,
+				16 * units.MiB,
+			},
+		},
+	} {
+		testKVMemBuf := &kvMemBuf{}
+		for _, allocSize := range tc.AllocSizes {
+			testKVMemBuf.AllocateBuf(allocSize)
+			testKVMemBuf.Recycle(testKVMemBuf.buf)
+		}
+		c.Assert(len(testKVMemBuf.availableBufs), Equals, len(tc.FinalAvailableByteBufCaps))
+		for i, bb := range testKVMemBuf.availableBufs {
+			c.Assert(bb.cap, Equals, tc.FinalAvailableByteBufCaps[i])
+		}
+	}
+}
+
+func (s *kvSuite) TestKVMemBufBatchAllocAndRecycle(c *C) {
+	testKVMemBuf := &kvMemBuf{}
+	bBufs := []*bytesBuf{}
+	for i := 0; i < maxAvailableBufSize; i++ {
+		testKVMemBuf.AllocateBuf(1 * units.MiB)
+		bBufs = append(bBufs, testKVMemBuf.buf)
+	}
+	for i := 0; i < maxAvailableBufSize; i++ {
+		testKVMemBuf.AllocateBuf(2 * units.MiB)
+		bBufs = append(bBufs, testKVMemBuf.buf)
+	}
+	for _, bb := range bBufs {
+		testKVMemBuf.Recycle(bb)
+	}
+	c.Assert(len(testKVMemBuf.availableBufs), Equals, maxAvailableBufSize)
+	for _, bb := range testKVMemBuf.availableBufs {
+		c.Assert(bb.cap, Equals, 4*units.MiB)
+	}
+	bBufs = bBufs[:0]
+	for i := 0; i < maxAvailableBufSize; i++ {
+		testKVMemBuf.AllocateBuf(1 * units.MiB)
+		bb := testKVMemBuf.buf
+		c.Assert(bb.cap, Equals, 4*units.MiB)
+		bBufs = append(bBufs, bb)
+		c.Assert(len(testKVMemBuf.availableBufs), Equals, maxAvailableBufSize-i-1)
+	}
+	for _, bb := range bBufs {
+		testKVMemBuf.Recycle(bb)
+	}
+	c.Assert(len(testKVMemBuf.availableBufs), Equals, maxAvailableBufSize)
+}

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

@@ -56,6 +56,7 @@ import (
 	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/table"
 	"github.com/pingcap/tidb/tablecodec"
+	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/codec"
 	tikverror "github.com/tikv/client-go/v2/error"
 	"github.com/tikv/client-go/v2/oracle"
@@ -87,6 +88,8 @@ const (
 	// lower the max-key-count to avoid tikv trigger region auto split
 	regionMaxKeyCount      = 1_280_000
 	defaultRegionSplitSize = 96 * units.MiB
+	// The max ranges count in a batch to split and scatter.
+	maxBatchSplitRanges = 4096
 
 	propRangeIndex = "tikv.range_index"
 
@@ -1347,7 +1350,7 @@ func (local *local) ImportEngine(ctx context.Context, engineUUID uuid.UUID, regi
 		needSplit := len(unfinishedRanges) > 1 || lfTotalSize > regionSplitSize || lfLength > regionSplitKeys
 		// split region by given ranges
 		for i := 0; i < maxRetryTimes; i++ {
-			err = local.SplitAndScatterRegionByRanges(ctx, unfinishedRanges, lf.tableInfo, needSplit, regionSplitSize)
+			err = local.SplitAndScatterRegionInBatches(ctx, unfinishedRanges, lf.tableInfo, needSplit, regionSplitSize, maxBatchSplitRanges)
 			if err == nil || common.IsContextCanceledError(err) {
 				break
 			}
@@ -1444,6 +1447,10 @@ func (local *local) ResolveDuplicateRows(ctx context.Context, tbl table.Table, t
 				if err == nil {
 					return nil
 				}
+				if types.ErrBadNumber.Equal(err) {
+					logger.Warn("delete duplicate rows encounter error", log.ShortError(err))
+					return errors.Errorf("resolve duplicate rows error on table '%s'", tableName)
+				}
 				if log.IsContextCanceledError(err) {
 					return err
 				}