external: range job keys and region split keys are 2 concepts (part 1)

pkg/ddl/backfilling_dist_executor.go

@@ -57,6 +57,7 @@ type BackfillSubTaskMeta struct {
 	RowEnd   []byte `json:"row_end"`
 
 	// Used by global sort write & ingest step.
+	RangeJobKeys   [][]byte `json:"range_job_keys,omitempty"`
 	RangeSplitKeys [][]byte `json:"range_split_keys,omitempty"`
 	DataFiles      []string `json:"data-files,omitempty"`
 	StatFiles      []string `json:"stat-files,omitempty"`

pkg/ddl/backfilling_dist_scheduler.go

@@ -447,7 +447,7 @@ func splitSubtaskMetaForOneKVMetaGroup(
 	startKey := kvMeta.StartKey
 	var endKey kv.Key
 	for {
-		endKeyOfGroup, dataFiles, statFiles, rangeSplitKeys, err := splitter.SplitOneRangesGroup()
+		endKeyOfGroup, dataFiles, statFiles, rangeJobKeys, regionSplitKeys, err := splitter.SplitOneRangesGroup()
 		if err != nil {
 			return nil, err
 		}
@@ -472,7 +472,8 @@ func splitSubtaskMetaForOneKVMetaGroup(
 			}},
 			DataFiles:      dataFiles,
 			StatFiles:      statFiles,
-			RangeSplitKeys: rangeSplitKeys,
+			RangeJobKeys:   rangeJobKeys,
+			RangeSplitKeys: regionSplitKeys,
 			TS:             ts,
 		}
 		if eleID > 0 {
@@ -596,16 +597,16 @@ func getRangeSplitter(
 	rangeGroupSize := totalSize / instanceCnt
 	rangeGroupKeys := int64(math.MaxInt64)
 
-	var maxSizePerRange = int64(config.SplitRegionSize)
-	var maxKeysPerRange = int64(config.SplitRegionKeys)
+	var regionSplitSize = int64(config.SplitRegionSize)
+	var regionSplitKeys = int64(config.SplitRegionKeys)
 	if store != nil {
 		pdCli := store.GetPDClient()
 		tls, err := ingest.NewDDLTLS()
 		if err == nil {
 			size, keys, err := local.GetRegionSplitSizeKeys(ctx, pdCli, tls)
 			if err == nil {
-				maxSizePerRange = max(maxSizePerRange, size)
-				maxKeysPerRange = max(maxKeysPerRange, keys)
+				regionSplitSize = max(regionSplitSize, size)
+				regionSplitKeys = max(regionSplitKeys, keys)
 			} else {
 				logger.Warn("fail to get region split keys and size", zap.Error(err))
 			}
@@ -614,8 +615,11 @@ func getRangeSplitter(
 		}
 	}
 
+	// no matter region split size and keys, we always split range jobs by 96MB
 	return external.NewRangeSplitter(ctx, multiFileStat, extStore,
-		rangeGroupSize, rangeGroupKeys, maxSizePerRange, maxKeysPerRange)
+		rangeGroupSize, rangeGroupKeys,
+		int64(config.SplitRegionSize), int64(config.SplitRegionKeys),
+		regionSplitSize, regionSplitKeys)
 }
 
 func forEachBackfillSubtaskMeta(

pkg/ddl/backfilling_import_cloud.go

@@ -109,13 +109,19 @@ func (m *cloudImportExecutor) RunSubtask(ctx context.Context, subtask *proto.Sub
 		all.Merge(g)
 	}
 
+	// compatible with old version TiDB
+	jobKeys := sm.RangeJobKeys
+	if jobKeys == nil {
+		jobKeys = sm.RangeSplitKeys
+	}
 	err = local.CloseEngine(ctx, &backend.EngineConfig{
 		External: &backend.ExternalEngineConfig{
 			StorageURI:    m.cloudStoreURI,
 			DataFiles:     sm.DataFiles,
 			StatFiles:     sm.StatFiles,
 			StartKey:      all.StartKey,
 			EndKey:        all.EndKey,
+			JobKeys:       jobKeys,
 			SplitKeys:     sm.RangeSplitKeys,
 			TotalFileSize: int64(all.TotalKVSize),
 			TotalKVCount:  0,

pkg/disttask/importinto/planner.go

@@ -387,7 +387,7 @@ func generateWriteIngestSpecs(planCtx planner.PlanCtx, p *LogicalPlan) ([]planne
 			startKey := tidbkv.Key(kvMeta.StartKey)
 			var endKey tidbkv.Key
 			for {
-				endKeyOfGroup, dataFiles, statFiles, rangeSplitKeys, err2 := splitter.SplitOneRangesGroup()
+				endKeyOfGroup, dataFiles, statFiles, rangeJobKeys, regionSplitKeys, err2 := splitter.SplitOneRangesGroup()
 				if err2 != nil {
 					return err2
 				}
@@ -415,8 +415,8 @@ func generateWriteIngestSpecs(planCtx planner.PlanCtx, p *LogicalPlan) ([]planne
 					},
 					DataFiles:      dataFiles,
 					StatFiles:      statFiles,
-					RangeSplitKeys: rangeSplitKeys,
-					RangeSplitSize: splitter.GetRangeSplitSize(),
+					RangeJobKeys:   rangeJobKeys,
+					RangeSplitKeys: regionSplitKeys,
 					TS:             ts,
 				}
 				specs = append(specs, &WriteIngestSpec{m})
@@ -516,12 +516,15 @@ func getRangeSplitter(ctx context.Context, store storage.ExternalStorage, kvMeta
 		zap.Int64("region-split-size", regionSplitSize),
 		zap.Int64("region-split-keys", regionSplitKeys))
 
+	// no matter region split size and keys, we always split range jobs by 96MB
 	return external.NewRangeSplitter(
 		ctx,
 		kvMeta.MultipleFilesStats,
 		store,
 		int64(config.DefaultBatchSize),
 		int64(math.MaxInt64),
+		int64(config.SplitRegionSize),
+		int64(config.SplitRegionKeys),
 		regionSplitSize,
 		regionSplitKeys,
 	)

pkg/disttask/importinto/proto.go

@@ -87,8 +87,8 @@ type WriteIngestStepMeta struct {
 	external.SortedKVMeta `json:"sorted-kv-meta"`
 	DataFiles             []string `json:"data-files"`
 	StatFiles             []string `json:"stat-files"`
+	RangeJobKeys          [][]byte `json:"range-job-keys"`
 	RangeSplitKeys        [][]byte `json:"range-split-keys"`
-	RangeSplitSize        int64    `json:"range-split-size"`
 	TS                    uint64   `json:"ts"`
 
 	Result Result

pkg/disttask/importinto/task_executor.go

@@ -413,18 +413,23 @@ func (e *writeAndIngestStepExecutor) RunSubtask(ctx context.Context, subtask *pr
 	_, engineUUID := backend.MakeUUID("", subtask.ID)
 	localBackend := e.tableImporter.Backend()
 	localBackend.WorkerConcurrency = subtask.Concurrency * 2
+	// compatible with old version TiDB
+	jobKeys := sm.RangeJobKeys
+	if jobKeys == nil {
+		jobKeys = sm.RangeSplitKeys
+	}
 	err = localBackend.CloseEngine(ctx, &backend.EngineConfig{
 		External: &backend.ExternalEngineConfig{
-			StorageURI:      e.taskMeta.Plan.CloudStorageURI,
-			DataFiles:       sm.DataFiles,
-			StatFiles:       sm.StatFiles,
-			StartKey:        sm.StartKey,
-			EndKey:          sm.EndKey,
-			SplitKeys:       sm.RangeSplitKeys,
-			RegionSplitSize: sm.RangeSplitSize,
-			TotalFileSize:   int64(sm.TotalKVSize),
-			TotalKVCount:    0,
-			CheckHotspot:    false,
+			StorageURI:    e.taskMeta.Plan.CloudStorageURI,
+			DataFiles:     sm.DataFiles,
+			StatFiles:     sm.StatFiles,
+			StartKey:      sm.StartKey,
+			EndKey:        sm.EndKey,
+			JobKeys:       jobKeys,
+			SplitKeys:     sm.RangeSplitKeys,
+			TotalFileSize: int64(sm.TotalKVSize),
+			TotalKVCount:  0,
+			CheckHotspot:  false,
 		},
 		TS: sm.TS,
 	}, engineUUID)

pkg/lightning/backend/backend.go

@@ -117,13 +117,13 @@ type LocalEngineConfig struct {
 
 // ExternalEngineConfig is the configuration used for local backend external engine.
 type ExternalEngineConfig struct {
-	StorageURI      string
-	DataFiles       []string
-	StatFiles       []string
-	StartKey        []byte
-	EndKey          []byte
-	SplitKeys       [][]byte
-	RegionSplitSize int64
+	StorageURI string
+	DataFiles  []string
+	StatFiles  []string
+	StartKey   []byte
+	EndKey     []byte
+	JobKeys    [][]byte
+	SplitKeys  [][]byte
 	// TotalFileSize can be an estimated value.
 	TotalFileSize int64
 	// TotalKVCount can be an estimated value.

pkg/lightning/backend/external/engine.go

@@ -140,7 +140,6 @@ func NewExternalEngine(
 	startKey []byte,
 	endKey []byte,
 	splitKeys [][]byte,
-	regionSplitSize int64,
 	keyAdapter common.KeyAdapter,
 	duplicateDetection bool,
 	duplicateDB *pebble.DB,
@@ -153,13 +152,12 @@ func NewExternalEngine(
 ) common.Engine {
 	memLimiter := membuf.NewLimiter(memLimit)
 	return &Engine{
-		storage:         storage,
-		dataFiles:       dataFiles,
-		statsFiles:      statsFiles,
-		startKey:        startKey,
-		endKey:          endKey,
-		splitKeys:       splitKeys,
-		regionSplitSize: regionSplitSize,
+		storage:    storage,
+		dataFiles:  dataFiles,
+		statsFiles: statsFiles,
+		startKey:   startKey,
+		endKey:     endKey,
+		splitKeys:  splitKeys,
 		smallBlockBufPool: membuf.NewPool(
 			membuf.WithBlockNum(0),
 			membuf.WithPoolMemoryLimiter(memLimiter),

pkg/lightning/backend/external/merge_v2.go

@@ -79,6 +79,8 @@ func MergeOverlappingFilesV2(
 		math.MaxInt64,
 		int64(4*size.GB),
 		math.MaxInt64,
+		math.MaxInt64,
+		math.MaxInt64,
 	)
 	if err != nil {
 		return err
@@ -114,7 +116,7 @@ func MergeOverlappingFilesV2(
 	var curEnd kv.Key
 
 	for {
-		endKeyOfGroup, dataFilesOfGroup, statFilesOfGroup, _, err1 := splitter.SplitOneRangesGroup()
+		endKeyOfGroup, dataFilesOfGroup, statFilesOfGroup, _, _, err1 := splitter.SplitOneRangesGroup()
 		if err1 != nil {
 			logutil.Logger(ctx).Warn("split one ranges group failed", zap.Error(err1))
 			return