lightning: support detect duplicate keys on specific index ID

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

@@ -243,6 +243,8 @@ type SessionOptions struct {
 	SysVars   map[string]string
 	// a seed used for tableKvEncoder's auto random bits value
 	AutoRandomSeed int64
+	// IndexID is used by the DuplicateManager. Only the key range with the specified index ID is scanned.
+	IndexID int64
 }
 
 // NewSession creates a new trimmed down Session matching the options.

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

@@ -401,6 +401,7 @@ type DuplicateManager struct {
 	logger      log.Logger
 	concurrency int
 	hasDupe     *atomic.Bool
+	indexID     int64
 }
 
 // NewDuplicateManager creates a new DuplicateManager.
@@ -430,6 +431,7 @@ func NewDuplicateManager(
 		logger:      logger,
 		concurrency: concurrency,
 		hasDupe:     hasDupe,
+		indexID:     sessOpts.IndexID,
 	}, nil
 }
 
@@ -547,6 +549,9 @@ type dupTask struct {
 }
 
 func (m *DuplicateManager) buildDupTasks() ([]dupTask, error) {
+	if m.indexID != 0 {
+		return m.buildIndexDupTasks()
+	}
 	keyRanges, err := tableHandleKeyRanges(m.tbl.Meta())
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -579,6 +584,29 @@ func (m *DuplicateManager) buildDupTasks() ([]dupTask, error) {
 	return tasks, nil
 }
 
+func (m *DuplicateManager) buildIndexDupTasks() ([]dupTask, error) {
+	for _, indexInfo := range m.tbl.Meta().Indices {
+		if m.indexID != indexInfo.ID {
+			continue
+		}
+		keyRanges, err := tableIndexKeyRanges(m.tbl.Meta(), indexInfo)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+		tasks := make([]dupTask, 0, len(keyRanges))
+		for _, kr := range keyRanges {
+			tableID := tablecodec.DecodeTableID(kr.StartKey)
+			tasks = append(tasks, dupTask{
+				KeyRange:  kr,
+				tableID:   tableID,
+				indexInfo: indexInfo,
+			})
+		}
+		return tasks, nil
+	}
+	return nil, nil
+}
+
 func (m *DuplicateManager) splitLocalDupTaskByKeys(
 	task dupTask,
 	dupDB *pebble.DB,