pkg/lightning : remove get_regions call in physical backend (#46202)

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

@@ -1333,6 +1333,7 @@ func (local *local) writeAndIngestByRange(
 	ctx, cancel := context.WithCancel(ctxt)
 	defer cancel()
 
+<<<<<<< HEAD
 WriteAndIngest:
 	for retry := 0; retry < maxRetryTimes; {
 		if retry != 0 {
@@ -1376,6 +1377,11 @@ WriteAndIngest:
 					logutil.Key("endKey", end), log.ShortError(err), zap.Int("retry", retry))
 				continue WriteAndIngest
 			}
+=======
+		err = local.SplitAndScatterRegionInBatches(ctx, initialSplitRanges, needSplit, maxBatchSplitRanges)
+		if err == nil || common.IsContextCanceledError(err) {
+			break
+>>>>>>> f15ba117bc2 (pkg/lightning : remove get_regions call in physical backend (#46202))
 		}
 
 		return err

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

@@ -67,17 +67,15 @@ var (
 func (local *local) SplitAndScatterRegionInBatches(
 	ctx context.Context,
 	ranges []Range,
-	tableInfo *checkpoints.TidbTableInfo,
 	needSplit bool,
-	regionSplitSize int64,
 	batchCnt int,
 ) error {
 	for i := 0; i < len(ranges); i += batchCnt {
 		batch := ranges[i:]
 		if len(batch) > batchCnt {
 			batch = batch[:batchCnt]
 		}
-		if err := local.SplitAndScatterRegionByRanges(ctx, batch, tableInfo, needSplit, regionSplitSize); err != nil {
+		if err := local.SplitAndScatterRegionByRanges(ctx, batch, needSplit); err != nil {
 			return errors.Trace(err)
 		}
 	}
@@ -91,10 +89,13 @@ func (local *local) SplitAndScatterRegionInBatches(
 func (local *local) SplitAndScatterRegionByRanges(
 	ctx context.Context,
 	ranges []Range,
-	tableInfo *checkpoints.TidbTableInfo,
 	needSplit bool,
+<<<<<<< HEAD
 	regionSplitSize int64,
 ) error {
+=======
+) (err error) {
+>>>>>>> f15ba117bc2 (pkg/lightning : remove get_regions call in physical backend (#46202))
 	if len(ranges) == 0 {
 		return nil
 	}
@@ -110,7 +111,6 @@ func (local *local) SplitAndScatterRegionByRanges(
 	scatterRegions := make([]*split.RegionInfo, 0)
 	var retryKeys [][]byte
 	waitTime := splitRegionBaseBackOffTime
-	skippedKeys := 0
 	for i := 0; i < splitRetryTimes; i++ {
 		log.FromContext(ctx).Info("split and scatter region",
 			logutil.Key("minKey", minKey),
@@ -172,6 +172,7 @@ func (local *local) SplitAndScatterRegionByRanges(
 			return nil
 		}
 
+<<<<<<< HEAD
 		var tableRegionStats map[uint64]int64
 		if tableInfo != nil {
 			tableRegionStats, err = fetchTableRegionSizeStats(ctx, db, tableInfo.ID)
@@ -182,6 +183,8 @@ func (local *local) SplitAndScatterRegionByRanges(
 			}
 		}
 
+=======
+>>>>>>> f15ba117bc2 (pkg/lightning : remove get_regions call in physical backend (#46202))
 		regionMap := make(map[uint64]*split.RegionInfo)
 		for _, region := range regions {
 			regionMap[region.Region.GetId()] = region
@@ -291,15 +294,6 @@ func (local *local) SplitAndScatterRegionByRanges(
 		}
 	sendLoop:
 		for regionID, keys := range splitKeyMap {
-			// if region not in tableRegionStats, that means this region is newly split, so
-			// we can skip split it again.
-			regionSize, ok := tableRegionStats[regionID]
-			if !ok {
-				log.FromContext(ctx).Warn("region stats not found", zap.Uint64("region", regionID))
-			}
-			if len(keys) == 1 && regionSize < regionSplitSize {
-				skippedKeys++
-			}
 			select {
 			case ch <- &splitInfo{region: regionMap[regionID], keys: keys}:
 			case <-ctx.Done():
@@ -335,11 +329,9 @@ func (local *local) SplitAndScatterRegionByRanges(
 	scatterCount, err := local.waitForScatterRegions(ctx, scatterRegions)
 	if scatterCount == len(scatterRegions) {
 		log.FromContext(ctx).Info("waiting for scattering regions done",
-			zap.Int("skipped_keys", skippedKeys),
 			zap.Int("regions", len(scatterRegions)), zap.Duration("take", time.Since(startTime)))
 	} else {
 		log.FromContext(ctx).Info("waiting for scattering regions timeout",
-			zap.Int("skipped_keys", skippedKeys),
 			zap.Int("scatterCount", scatterCount),
 			zap.Int("regions", len(scatterRegions)),
 			zap.Duration("take", time.Since(startTime)),

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

@@ -480,6 +480,7 @@ func doTestBatchSplitRegionByRanges(ctx context.Context, t *testing.T, hook clie
 		start = end
 	}
 
+<<<<<<< HEAD
 	if len(errPat) > 0 {
 		backup := split.ScanRegionAttemptTimes
 		split.ScanRegionAttemptTimes = 3
@@ -491,6 +492,10 @@ func doTestBatchSplitRegionByRanges(ctx context.Context, t *testing.T, hook clie
 	if len(errPat) == 0 {
 		require.NoError(t, err)
 	} else {
+=======
+	err = local.SplitAndScatterRegionByRanges(ctx, ranges, true)
+	if len(errPat) != 0 {
+>>>>>>> f15ba117bc2 (pkg/lightning : remove get_regions call in physical backend (#46202))
 		require.Error(t, err)
 		require.Regexp(t, errPat, err.Error())
 		return
@@ -515,6 +520,94 @@ func TestBatchSplitRegionByRanges(t *testing.T) {
 	doTestBatchSplitRegionByRanges(context.Background(), t, nil, "", nil)
 }
 
+<<<<<<< HEAD
+=======
+type checkScatterClient struct {
+	*testSplitClient
+
+	mu                sync.Mutex
+	notFoundFirstTime map[uint64]struct{}
+	scatterCounter    atomic.Int32
+}
+
+func newCheckScatterClient(inner *testSplitClient) *checkScatterClient {
+	return &checkScatterClient{
+		testSplitClient:   inner,
+		notFoundFirstTime: map[uint64]struct{}{},
+		scatterCounter:    atomic.Int32{},
+	}
+}
+
+func (c *checkScatterClient) ScatterRegion(ctx context.Context, regionInfo *split.RegionInfo) error {
+	c.scatterCounter.Add(1)
+	return nil
+}
+
+func (c *checkScatterClient) GetRegionByID(ctx context.Context, regionID uint64) (*split.RegionInfo, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if _, ok := c.notFoundFirstTime[regionID]; !ok {
+		c.notFoundFirstTime[regionID] = struct{}{}
+		return nil, nil
+	}
+	return c.testSplitClient.GetRegionByID(ctx, regionID)
+}
+
+func TestMissingScatter(t *testing.T) {
+	ctx := context.Background()
+	splitHook := defaultHook{}
+	deferFunc := splitHook.setup(t)
+	defer deferFunc()
+
+	keys := [][]byte{[]byte(""), []byte("aay"), []byte("bba"), []byte("bbh"), []byte("cca"), []byte("")}
+	client := initTestSplitClient(keys, nil)
+	checkClient := newCheckScatterClient(client)
+	local := &Backend{
+		splitCli: checkClient,
+		logger:   log.L(),
+	}
+	local.RegionSplitBatchSize = 4
+	local.RegionSplitConcurrency = 4
+
+	// current region ranges: [, aay), [aay, bba), [bba, bbh), [bbh, cca), [cca, )
+	rangeStart := codec.EncodeBytes([]byte{}, []byte("b"))
+	rangeEnd := codec.EncodeBytes([]byte{}, []byte("c"))
+	regions, err := split.PaginateScanRegion(ctx, client, rangeStart, rangeEnd, 5)
+	require.NoError(t, err)
+	// regions is: [aay, bba), [bba, bbh), [bbh, cca)
+	checkRegionRanges(t, regions, [][]byte{[]byte("aay"), []byte("bba"), []byte("bbh"), []byte("cca")})
+
+	// generate:  ranges [b, ba), [ba, bb), [bb, bc), ... [by, bz)
+	ranges := make([]Range, 0)
+	start := []byte{'b'}
+	for i := byte('a'); i <= 'z'; i++ {
+		end := []byte{'b', i}
+		ranges = append(ranges, Range{start: start, end: end})
+		start = end
+	}
+
+	err = local.SplitAndScatterRegionByRanges(ctx, ranges, true)
+	require.NoError(t, err)
+
+	splitHook.check(t, client)
+
+	// check split ranges
+	regions, err = split.PaginateScanRegion(ctx, client, rangeStart, rangeEnd, 5)
+	require.NoError(t, err)
+	result := [][]byte{
+		[]byte("b"), []byte("ba"), []byte("bb"), []byte("bba"), []byte("bbh"), []byte("bc"),
+		[]byte("bd"), []byte("be"), []byte("bf"), []byte("bg"), []byte("bh"), []byte("bi"), []byte("bj"),
+		[]byte("bk"), []byte("bl"), []byte("bm"), []byte("bn"), []byte("bo"), []byte("bp"), []byte("bq"),
+		[]byte("br"), []byte("bs"), []byte("bt"), []byte("bu"), []byte("bv"), []byte("bw"), []byte("bx"),
+		[]byte("by"), []byte("bz"), []byte("cca"),
+	}
+	checkRegionRanges(t, regions, result)
+
+	// the old regions will not be scattered. They are [..., bba), [bba, bbh), [..., cca)
+	require.Equal(t, len(result)-3, int(checkClient.scatterCounter.Load()))
+}
+
+>>>>>>> f15ba117bc2 (pkg/lightning : remove get_regions call in physical backend (#46202))
 type batchSizeHook struct{}
 
 func (h batchSizeHook) setup(t *testing.T) func() {
@@ -650,7 +743,7 @@ func TestSplitAndScatterRegionInBatches(t *testing.T) {
 		})
 	}
 
-	err := local.SplitAndScatterRegionInBatches(ctx, ranges, nil, true, 1000, 4)
+	err := local.SplitAndScatterRegionInBatches(ctx, ranges, true, 4)
 	require.NoError(t, err)
 
 	rangeStart := codec.EncodeBytes([]byte{}, []byte("a"))
@@ -747,7 +840,7 @@ func doTestBatchSplitByRangesWithClusteredIndex(t *testing.T, hook clientHook) {
 		start = e
 	}
 
-	err := local.SplitAndScatterRegionByRanges(ctx, ranges, nil, true, 1000)
+	err := local.SplitAndScatterRegionByRanges(ctx, ranges, true)
 	require.NoError(t, err)
 
 	startKey := codec.EncodeBytes([]byte{}, rangeKeys[0])