pkg/dxf/importinto, pkg/executor: show conflict progress in SHOW IMPORT JOB (#67551)

ref #66019

Author: D3Hunter

pkg/ddl/backfilling_clean_s3.go

@@ -109,7 +109,7 @@ func sendMeterOnCleanUp(ctx context.Context, task *proto.Task, logger *zap.Logge
 			return errors.Trace(err)
 		}
 		rowCount += summary.RowCnt.Load()
-		indexKVSize += summary.Bytes.Load()
+		indexKVSize += summary.Processed.Load()
 	}
 	return handle.SendRowAndSizeMeterData(ctx, task, rowCount, 0, indexKVSize, logger)
 }

pkg/ddl/backfilling_read_index.go

@@ -490,7 +490,7 @@ func (d *distTaskRowCntCollector) Accepted(bytes int64) {
 }
 
 func (d *distTaskRowCntCollector) Processed(bytes, rowCnt int64) {
-	d.summary.Bytes.Add(bytes)
+	d.summary.Processed.Add(bytes)
 	d.summary.RowCnt.Add(rowCnt)
 	d.counter.Add(float64(rowCnt))
 }

pkg/dxf/framework/storage/table_test.go

@@ -333,8 +333,8 @@ func TestGetSubtaskSummaries(t *testing.T) {
 	require.NoError(t, tm.SwitchTaskStepInBatch(ctx, task, proto.TaskStateRunning, proto.StepOne, subtasks))
 
 	summary := &execute.SubtaskSummary{
-		RowCnt: *atomic.NewInt64(100),
-		Bytes:  *atomic.NewInt64(200),
+		RowCnt:    *atomic.NewInt64(100),
+		Processed: *atomic.NewInt64(200),
 	}
 	bytes, err := json.Marshal(summary)
 	require.NoError(t, err)
@@ -345,7 +345,7 @@ func TestGetSubtaskSummaries(t *testing.T) {
 	require.Len(t, summaries, len(subtasks))
 	for _, summary := range summaries {
 		require.EqualValues(t, 100, summary.RowCnt.Load())
-		require.EqualValues(t, 200, summary.Bytes.Load())
+		require.EqualValues(t, 200, summary.Processed.Load())
 	}
 
 	// If the JSON value is wrong, we still get an empty summary.
@@ -356,7 +356,7 @@ func TestGetSubtaskSummaries(t *testing.T) {
 	require.NoError(t, err)
 	for _, summary := range summaries {
 		require.EqualValues(t, 0, summary.RowCnt.Load())
-		require.EqualValues(t, 0, summary.Bytes.Load())
+		require.EqualValues(t, 0, summary.Processed.Load())
 	}
 }
 

pkg/dxf/framework/taskexecutor/execute/interface.go

@@ -86,7 +86,9 @@ type Progress struct {
 	// For now, RowCnt is not used, but as it's collected by the collector,
 	// we still keep it here for future possible usage.
 	RowCnt int64 `json:"row_count,omitempty"`
-	Bytes  int64 `json:"bytes,omitempty"`
+	// Processed stores generic progress units, such as bytes processed or rows processed.
+	// Keep the JSON tag as "bytes" for backward compatibility with persisted summaries.
+	Processed int64 `json:"bytes,omitempty"`
 
 	// UpdateTime is the time when this progress is stored.
 	UpdateTime time.Time `json:"update_time,omitempty"`
@@ -95,10 +97,12 @@ type Progress struct {
 // SubtaskSummary contains the summary of a subtask.
 // It tracks the runtime summary of the subtask.
 type SubtaskSummary struct {
-	// RowCnt and Bytes are updated by the collector.
+	// RowCnt and Processed are updated by the collector.
 	RowCnt atomic.Int64 `json:"row_count,omitempty"`
-	// Bytes is the number of bytes to process.
-	Bytes atomic.Int64 `json:"bytes,omitempty"`
+	// Processed is the number of processed units reported by the collector.
+	// Its unit may be bytes processed or rows processed, depending on the step.
+	// Keep the JSON tag as "bytes" for backward compatibility with persisted summaries.
+	Processed atomic.Int64 `json:"bytes,omitempty"`
 	// ReadBytes is the number of bytes that read from the source.
 	ReadBytes atomic.Int64 `json:"read_bytes,omitempty"`
 	// GetReqCnt is the number of get requests to the external storage.
@@ -124,7 +128,7 @@ func (s *SubtaskSummary) MergeObjStoreRequests(reqs *recording.Requests) {
 func (s *SubtaskSummary) Update() {
 	s.Progresses = append(s.Progresses, Progress{
 		RowCnt:     s.RowCnt.Load(),
-		Bytes:      s.Bytes.Load(),
+		Processed:  s.Processed.Load(),
 		UpdateTime: time.Now(),
 	})
 
@@ -145,15 +149,15 @@ func (s *SubtaskSummary) GetSpeedInTimeRange(endTime time.Time, duration time.Du
 	}
 
 	// The number of point is small, so we can afford to iterate through all points.
-	var totalBytes float64
+	var totalProcessed float64
 	for i := range len(s.Progresses) - 1 {
 		rangeStart := s.Progresses[i].UpdateTime
 		rangeEnd := s.Progresses[i+1].UpdateTime
-		rangeBytes := float64(s.Progresses[i+1].Bytes - s.Progresses[i].Bytes)
+		rangeProcessed := float64(s.Progresses[i+1].Processed - s.Progresses[i].Processed)
 		if endTime.Before(rangeStart) || startTime.After(rangeEnd) {
 			continue
 		} else if startTime.Before(rangeStart) && endTime.After(rangeEnd) {
-			totalBytes += rangeBytes
+			totalProcessed += rangeProcessed
 			continue
 		}
 
@@ -167,10 +171,10 @@ func (s *SubtaskSummary) GetSpeedInTimeRange(endTime time.Time, duration time.Du
 			iEnd = endTime
 		}
 
-		totalBytes += rangeBytes * float64(iEnd.Sub(iStart)) / float64(rangeEnd.Sub(rangeStart))
+		totalProcessed += rangeProcessed * float64(iEnd.Sub(iStart)) / float64(rangeEnd.Sub(rangeStart))
 	}
 
-	return int64(totalBytes / duration.Seconds())
+	return int64(totalProcessed / duration.Seconds())
 }
 
 // UpdateTime returns the last update time of the summary.
@@ -184,7 +188,7 @@ func (s *SubtaskSummary) UpdateTime() time.Time {
 // Reset resets the summary to zero values and clears history data.
 func (s *SubtaskSummary) Reset() {
 	s.RowCnt.Store(0)
-	s.Bytes.Store(0)
+	s.Processed.Store(0)
 	s.ReadBytes.Store(0)
 	s.PutReqCnt.Store(0)
 	s.GetReqCnt.Store(0)
@@ -198,12 +202,14 @@ type Collector interface {
 	// The difference between Accepted and Processed is that Accepted is called
 	// when the data is accepted to be processed.
 	Accepted(bytes int64)
-	// Processed is used collects metrics.
-	// `bytes` is the number of bytes processed, and `rows` is the number of rows processed.
-	// The meaning of `bytes` may vary by scenario, for example:
-	//   - During encoding, it represents the number of bytes read from the source data file.
-	//   - During merge sort, it represents the number of bytes merged.
-	Processed(bytes, rows int64)
+	// Processed collects progress metrics.
+	// `processedUnits` is the number of processed units, and `rows` is the number
+	// of rows processed.
+	// The meaning of `processedUnits` may vary by scenario, for example:
+	//   - During encoding, it represents bytes read from source data files.
+	//   - During merge sort, it represents bytes merged.
+	//   - During conflict handling, it represents processed conflict KV pairs.
+	Processed(processedUnits, rows int64)
 }
 
 // NoopCollector is a no-op implementation of Collector.
@@ -218,9 +224,9 @@ func (*NoopCollector) Processed(_, _ int64) {}
 // TestCollector is an implementation used for test.
 type TestCollector struct {
 	NoopCollector
-	ReadBytes atomic.Int64
-	Bytes     atomic.Int64
-	Rows      atomic.Int64
+	ReadBytes    atomic.Int64
+	ProcessedCnt atomic.Int64
+	Rows         atomic.Int64
 }
 
 // Accepted implements Collector.Accepted
@@ -229,8 +235,8 @@ func (c *TestCollector) Accepted(bytes int64) {
 }
 
 // Processed implements Collector.Processed
-func (c *TestCollector) Processed(bytes, rows int64) {
-	c.Bytes.Add(bytes)
+func (c *TestCollector) Processed(processedUnits, rows int64) {
+	c.ProcessedCnt.Add(processedUnits)
 	c.Rows.Add(rows)
 }
 

pkg/dxf/framework/taskexecutor/execute/interface_test.go

@@ -34,7 +34,7 @@ func TestSubtaskSummaryGetSpeed(t *testing.T) {
 			name: "insufficient data points",
 			setup: func(s *SubtaskSummary) {
 				s.Progresses = []Progress{
-					{Bytes: 100, UpdateTime: time.Unix(1000, 0)},
+					{Processed: 100, UpdateTime: time.Unix(1000, 0)},
 				}
 			},
 			endTime:     time.Unix(1010, 0),
@@ -47,8 +47,8 @@ func TestSubtaskSummaryGetSpeed(t *testing.T) {
 			setup: func(s *SubtaskSummary) {
 				baseTime := time.Unix(1000, 0)
 				s.Progresses = []Progress{
-					{Bytes: 0, UpdateTime: baseTime},
-					{Bytes: 100, UpdateTime: baseTime.Add(1 * time.Second)},
+					{Processed: 0, UpdateTime: baseTime},
+					{Processed: 100, UpdateTime: baseTime.Add(1 * time.Second)},
 				}
 			},
 			endTime:     time.Unix(1010, 0),
@@ -61,10 +61,10 @@ func TestSubtaskSummaryGetSpeed(t *testing.T) {
 			setup: func(s *SubtaskSummary) {
 				baseTime := time.Unix(1000, 0)
 				s.Progresses = []Progress{
-					{Bytes: 0, UpdateTime: baseTime},
-					{Bytes: 50, UpdateTime: baseTime.Add(1 * time.Second)},
-					{Bytes: 100, UpdateTime: baseTime.Add(2 * time.Second)},
-					{Bytes: 150, UpdateTime: baseTime.Add(3 * time.Second)},
+					{Processed: 0, UpdateTime: baseTime},
+					{Processed: 50, UpdateTime: baseTime.Add(1 * time.Second)},
+					{Processed: 100, UpdateTime: baseTime.Add(2 * time.Second)},
+					{Processed: 150, UpdateTime: baseTime.Add(3 * time.Second)},
 				}
 			},
 			endTime:     time.Unix(1002, 500000000),
@@ -77,10 +77,10 @@ func TestSubtaskSummaryGetSpeed(t *testing.T) {
 			setup: func(s *SubtaskSummary) {
 				baseTime := time.Unix(1000, 0)
 				s.Progresses = []Progress{
-					{Bytes: 0, UpdateTime: baseTime},
-					{Bytes: 30, UpdateTime: baseTime.Add(1 * time.Second)},
-					{Bytes: 60, UpdateTime: baseTime.Add(2 * time.Second)},
-					{Bytes: 90, UpdateTime: baseTime.Add(3 * time.Second)},
+					{Processed: 0, UpdateTime: baseTime},
+					{Processed: 30, UpdateTime: baseTime.Add(1 * time.Second)},
+					{Processed: 60, UpdateTime: baseTime.Add(2 * time.Second)},
+					{Processed: 90, UpdateTime: baseTime.Add(3 * time.Second)},
 				}
 			},
 			endTime:     time.Unix(1004, 0),
@@ -93,11 +93,11 @@ func TestSubtaskSummaryGetSpeed(t *testing.T) {
 			setup: func(s *SubtaskSummary) {
 				baseTime := time.Unix(1000, 0)
 				s.Progresses = []Progress{
-					{Bytes: 0, UpdateTime: baseTime},
-					{Bytes: 60, UpdateTime: baseTime.Add(1 * time.Second)},
-					{Bytes: 120, UpdateTime: baseTime.Add(2 * time.Second)},
-					{Bytes: 180, UpdateTime: baseTime.Add(3 * time.Second)},
-					{Bytes: 240, UpdateTime: baseTime.Add(4 * time.Second)},
+					{Processed: 0, UpdateTime: baseTime},
+					{Processed: 60, UpdateTime: baseTime.Add(1 * time.Second)},
+					{Processed: 120, UpdateTime: baseTime.Add(2 * time.Second)},
+					{Processed: 180, UpdateTime: baseTime.Add(3 * time.Second)},
+					{Processed: 240, UpdateTime: baseTime.Add(4 * time.Second)},
 				}
 			},
 			endTime:     time.Unix(1004, 500000000),
@@ -110,11 +110,11 @@ func TestSubtaskSummaryGetSpeed(t *testing.T) {
 			setup: func(s *SubtaskSummary) {
 				baseTime := time.Unix(1000, 0)
 				s.Progresses = []Progress{
-					{Bytes: 0, UpdateTime: baseTime},
-					{Bytes: 60, UpdateTime: baseTime.Add(1 * time.Second)},
-					{Bytes: 120, UpdateTime: baseTime.Add(2 * time.Second)},
-					{Bytes: 180, UpdateTime: baseTime.Add(3 * time.Second)},
-					{Bytes: 240, UpdateTime: baseTime.Add(4 * time.Second)},
+					{Processed: 0, UpdateTime: baseTime},
+					{Processed: 60, UpdateTime: baseTime.Add(1 * time.Second)},
+					{Processed: 120, UpdateTime: baseTime.Add(2 * time.Second)},
+					{Processed: 180, UpdateTime: baseTime.Add(3 * time.Second)},
+					{Processed: 240, UpdateTime: baseTime.Add(4 * time.Second)},
 				}
 			},
 			endTime:     time.Unix(1004, 0),
@@ -127,11 +127,11 @@ func TestSubtaskSummaryGetSpeed(t *testing.T) {
 			setup: func(s *SubtaskSummary) {
 				baseTime := time.Unix(1001, 0)
 				s.Progresses = []Progress{
-					{Bytes: 0, UpdateTime: baseTime},
-					{Bytes: 60, UpdateTime: baseTime.Add(1 * time.Second)},
-					{Bytes: 120, UpdateTime: baseTime.Add(2 * time.Second)},
-					{Bytes: 180, UpdateTime: baseTime.Add(3 * time.Second)},
-					{Bytes: 240, UpdateTime: baseTime.Add(4 * time.Second)},
+					{Processed: 0, UpdateTime: baseTime},
+					{Processed: 60, UpdateTime: baseTime.Add(1 * time.Second)},
+					{Processed: 120, UpdateTime: baseTime.Add(2 * time.Second)},
+					{Processed: 180, UpdateTime: baseTime.Add(3 * time.Second)},
+					{Processed: 240, UpdateTime: baseTime.Add(4 * time.Second)},
 				}
 			},
 			endTime:     time.Unix(1006, 500000000),

pkg/dxf/importinto/collect_conflicts.go

@@ -68,6 +68,7 @@ type collectConflictsStepExecutor struct {
 }
 
 var _ execute.StepExecutor = &collectConflictsStepExecutor{}
+var _ execute.Collector = &collectConflictsStepExecutor{}
 
 // NewCollectConflictsStepExecutor creates a new collectConflictsStepExecutor.
 // exported for test.
@@ -200,6 +201,7 @@ func (e *collectConflictsStepExecutor) collectConflictsOfKVGroup(
 			e.sharedHandleSet,
 			localSet,
 			&e.sizeOfConflictRowFiles,
+			e,
 			e.GetMeterRecorder(),
 		)
 		eg.Go(func() (err error) {
@@ -251,6 +253,14 @@ func (e *collectConflictsStepExecutor) ResetSummary() {
 	e.summary.Reset()
 }
 
+// Accepted implements Collector.Accepted interface.
+func (*collectConflictsStepExecutor) Accepted(_ int64) {}
+
+// Processed implements Collector.Processed interface.
+func (e *collectConflictsStepExecutor) Processed(processedConflictKVs, _ int64) {
+	e.summary.Processed.Add(processedConflictKVs)
+}
+
 // getConflictRowFilenamePrefix returns the file name prefix to store the conflict
 // rows for the given task and subtask.
 func getConflictRowFilenamePrefix(taskID, subtaskID int64, uuid string) string {

pkg/dxf/importinto/conflict_resolution.go

@@ -48,6 +48,7 @@ type conflictResolutionStepExecutor struct {
 }
 
 var _ execute.StepExecutor = &conflictResolutionStepExecutor{}
+var _ execute.Collector = &conflictResolutionStepExecutor{}
 
 // NewConflictResolutionStepExecutor creates a new StepExecutor for conflict
 // resolution step, exported for test.
@@ -139,7 +140,7 @@ func (e *conflictResolutionStepExecutor) resolveConflictsOfKVGroup(
 	pairCh := external.ReadKVFilesAsync(egCtx, eg, objStore, ci.Files)
 	for i := range concurrency {
 		encoder := encoders[i]
-		deleter := conflictedkv.NewDeleter(e.tableImporter.Table, e.logger, e.store, kvGroup, encoder, e.GetMeterRecorder())
+		deleter := conflictedkv.NewDeleter(e.tableImporter.Table, e.logger, e.store, kvGroup, encoder, e, e.GetMeterRecorder())
 		eg.Go(func() error {
 			return deleter.Run(egCtx, pairCh)
 		})
@@ -162,6 +163,14 @@ func (e *conflictResolutionStepExecutor) ResetSummary() {
 	e.summary.Reset()
 }
 
+// Accepted implements Collector.Accepted interface.
+func (*conflictResolutionStepExecutor) Accepted(_ int64) {}
+
+// Processed implements Collector.Processed interface.
+func (e *conflictResolutionStepExecutor) Processed(processedConflictKVs, _ int64) {
+	e.summary.Processed.Add(processedConflictKVs)
+}
+
 // when create encoder, if the table have generated column, when calling
 // backend/kv.CollectGeneratedColumns(), buildSimpleExpr will rewrite the AST node,
 // and data race. and the data race might happen during encoding, in

pkg/dxf/importinto/conflictedkv/BUILD.bazel

@@ -14,6 +14,7 @@ go_library(
     deps = [
         "//pkg/config/kerneltype",
         "//pkg/dxf/framework/handle",
+        "//pkg/dxf/framework/taskexecutor/execute",
         "//pkg/executor/importer",
         "//pkg/kv",
         "//pkg/lightning/backend/external",
@@ -53,6 +54,7 @@ go_test(
     shard_count = 8,
     deps = [
         "//pkg/config/kerneltype",
+        "//pkg/dxf/framework/taskexecutor/execute",
         "//pkg/executor/importer",
         "//pkg/kv",
         "//pkg/lightning/backend/encode",

pkg/dxf/importinto/conflictedkv/collector.go

@@ -23,6 +23,7 @@ import (
 	"github.com/docker/go-units"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
+	"github.com/pingcap/tidb/pkg/dxf/framework/taskexecutor/execute"
 	"github.com/pingcap/tidb/pkg/executor/importer"
 	tidbkv "github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/lightning/backend/external"
@@ -113,6 +114,7 @@ func NewCollector(
 	encoder *importer.TableKVEncoder,
 	globalSet, localSet *BoundedHandleSet,
 	sharedTotalFileSize *atomic.Int64,
+	progressCollector execute.Collector,
 	trafficRec TrafficRecorder,
 ) *Collector {
 	// Safety check: if caller doesn't pass the shared counter, allocate one to
@@ -129,7 +131,7 @@ func NewCollector(
 		hdlSet:              localSet,
 		sharedTotalFileSize: sharedTotalFileSize,
 	}
-	base := NewBaseHandler(targetTbl, kvGroup, encoder, collector, logger)
+	base := NewBaseHandler(targetTbl, kvGroup, encoder, collector, progressCollector, logger)
 	var h Handler
 	if kvGroup == external.DataKVGroup {
 		h = NewDataKVHandler(base)