scheduler: Speed ​​up region scanning before table splitting (#12222)

close #12221

Author: hongyunyan

cdc/scheduler/internal/v3/keyspan/mock.go

@@ -15,7 +15,9 @@ package keyspan
 
 import (
 	"bytes"
+	"unsafe"
 
+	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -26,12 +28,10 @@ import (
 // RegionCache is a simplified interface of tikv.RegionCache.
 // It is useful to restrict RegionCache usage and mocking in tests.
 type RegionCache interface {
-	// ListRegionIDsInKeyRange lists ids of regions in [startKey,endKey].
-	ListRegionIDsInKeyRange(
+	// LoadRegionsInKeyRange loads regions in [startKey,endKey].
+	LoadRegionsInKeyRange(
 		bo *tikv.Backoffer, startKey, endKey []byte,
-	) (regionIDs []uint64, err error)
-	// LocateRegionByID searches for the region with ID.
-	LocateRegionByID(bo *tikv.Backoffer, regionID uint64) (*tikv.KeyLocation, error)
+	) (regions []*tikv.Region, err error)
 }
 
 // mockCache mocks tikv.RegionCache.
@@ -42,34 +42,29 @@ type mockCache struct {
 // NewMockRegionCache returns a new MockCache.
 func NewMockRegionCache() *mockCache { return &mockCache{regions: spanz.NewBtreeMap[uint64]()} }
 
-// ListRegionIDsInKeyRange lists ids of regions in [startKey,endKey].
-func (m *mockCache) ListRegionIDsInKeyRange(
+func (m *mockCache) LoadRegionsInKeyRange(
 	bo *tikv.Backoffer, startKey, endKey []byte,
-) (regionIDs []uint64, err error) {
+) (regions []*tikv.Region, err error) {
 	m.regions.Ascend(func(loc tablepb.Span, id uint64) bool {
 		if bytes.Compare(loc.StartKey, endKey) >= 0 ||
 			bytes.Compare(loc.EndKey, startKey) <= 0 {
 			return true
 		}
-		regionIDs = append(regionIDs, id)
-		return true
-	})
-	return
-}
-
-// LocateRegionByID searches for the region with ID.
-func (m *mockCache) LocateRegionByID(
-	bo *tikv.Backoffer, regionID uint64,
-) (loc *tikv.KeyLocation, err error) {
-	m.regions.Ascend(func(span tablepb.Span, id uint64) bool {
-		if id != regionID {
-			return true
-		}
-		loc = &tikv.KeyLocation{
-			StartKey: span.StartKey,
-			EndKey:   span.EndKey,
+		region := &tikv.Region{}
+		meta := &metapb.Region{
+			Id:       id,
+			StartKey: loc.StartKey,
+			EndKey:   loc.EndKey,
 		}
-		return false
+
+		// region.meta is not exported, so we use unsafe to set it for testing.
+		regionPtr := (*struct {
+			meta *metapb.Region
+		})(unsafe.Pointer(region))
+		regionPtr.meta = meta
+
+		regions = append(regions, region)
+		return true
 	})
 	return
 }

cdc/scheduler/internal/v3/keyspan/splitter_region_count.go

@@ -16,7 +16,9 @@ package keyspan
 import (
 	"bytes"
 	"context"
+	"encoding/hex"
 	"math"
+	"time"
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
@@ -44,8 +46,9 @@ func newRegionCountSplitter(
 func (m *regionCountSplitter) split(
 	ctx context.Context, span tablepb.Span, captureNum int,
 ) []tablepb.Span {
+	startTimestamp := time.Now()
 	bo := tikv.NewBackoffer(ctx, 500)
-	regions, err := m.regionCache.ListRegionIDsInKeyRange(bo, span.StartKey, span.EndKey)
+	regions, err := m.regionCache.LoadRegionsInKeyRange(bo, span.StartKey, span.EndKey)
 	if err != nil {
 		log.Warn("schedulerv3: list regions failed, skip split span",
 			zap.String("namespace", m.changefeedID.Namespace),
@@ -72,38 +75,23 @@ func (m *regionCountSplitter) split(
 	spans := make([]tablepb.Span, 0, stepper.SpanCount())
 	start, end := 0, stepper.Step()
 	for {
-		startRegion, err := m.regionCache.LocateRegionByID(bo, regions[start])
-		if err != nil {
-			log.Warn("schedulerv3: get regions failed, skip split span",
-				zap.String("namespace", m.changefeedID.Namespace),
-				zap.String("changefeed", m.changefeedID.ID),
-				zap.String("span", span.String()),
-				zap.Error(err))
-			return []tablepb.Span{span}
-		}
-		endRegion, err := m.regionCache.LocateRegionByID(bo, regions[end-1])
-		if err != nil {
-			log.Warn("schedulerv3: get regions failed, skip split span",
-				zap.String("namespace", m.changefeedID.Namespace),
-				zap.String("changefeed", m.changefeedID.ID),
-				zap.String("span", span.String()),
-				zap.Error(err))
-			return []tablepb.Span{span}
-		}
+		startKey := regions[start].StartKey()
+		endKey := regions[end-1].EndKey()
 		if len(spans) > 0 &&
-			bytes.Compare(spans[len(spans)-1].EndKey, startRegion.StartKey) > 0 {
+			bytes.Compare(spans[len(spans)-1].EndKey, startKey) > 0 {
 			log.Warn("schedulerv3: list region out of order detected",
 				zap.String("namespace", m.changefeedID.Namespace),
 				zap.String("changefeed", m.changefeedID.ID),
 				zap.String("span", span.String()),
 				zap.Stringer("lastSpan", &spans[len(spans)-1]),
-				zap.Stringer("region", startRegion))
+				zap.String("startKey", hex.EncodeToString(startKey)),
+				zap.String("endKey", hex.EncodeToString(endKey)))
 			return []tablepb.Span{span}
 		}
 		spans = append(spans, tablepb.Span{
 			TableID:  span.TableID,
-			StartKey: startRegion.StartKey,
-			EndKey:   endRegion.EndKey,
+			StartKey: startKey,
+			EndKey:   endKey,
 		})
 
 		if end == len(regions) {
@@ -128,7 +116,8 @@ func (m *regionCountSplitter) split(
 		zap.Int("totalCaptures", captureNum),
 		zap.Int("regionCount", len(regions)),
 		zap.Int("regionThreshold", m.regionThreshold),
-		zap.Int("spanRegionLimit", spanRegionLimit))
+		zap.Int("spanRegionLimit", spanRegionLimit),
+		zap.Duration("time", time.Since(startTimestamp)))
 	return spans
 }