ddl/ingest: set minCommitTS when detect remote duplicate keys

pkg/ddl/ingest/backend.go

@@ -102,6 +102,7 @@ type litBackendCtx struct {
 	updateInterval  time.Duration
 	checkpointMgr   *CheckpointManager
 	etcdClient      *clientv3.Client
+	initTS          uint64
 
 	// unregisterMu prevents concurrent calls of `FinishAndUnregisterEngines`.
 	// For details, see https://github.com/pingcap/tidb/issues/53843.
@@ -149,9 +150,10 @@ func (bc *litBackendCtx) CollectRemoteDuplicateRows(indexID int64, tbl table.Tab
 func (bc *litBackendCtx) collectRemoteDuplicateRows(indexID int64, tbl table.Table) error {
 	dupeController := bc.backend.GetDupeController(bc.cfg.WorkerConcurrency, nil)
 	hasDupe, err := dupeController.CollectRemoteDuplicateRows(bc.ctx, tbl, tbl.Meta().Name.L, &encode.SessionOptions{
-		SQLMode: mysql.ModeStrictAllTables,
-		SysVars: bc.sysVars,
-		IndexID: indexID,
+		SQLMode:     mysql.ModeStrictAllTables,
+		SysVars:     bc.sysVars,
+		IndexID:     indexID,
+		MinCommitTS: bc.initTS,
 	}, lightning.ErrorOnDup)
 	return bc.handleErrorAfterCollectRemoteDuplicateRows(err, indexID, tbl, hasDupe)
 }

pkg/ddl/ingest/backend_mgr.go

@@ -30,6 +30,7 @@ import (
 	"github.com/pingcap/tidb/pkg/lightning/backend/local"
 	"github.com/pingcap/tidb/pkg/lightning/common"
 	"github.com/pingcap/tidb/pkg/util/logutil"
+	"github.com/tikv/client-go/v2/oracle"
 	pd "github.com/tikv/pd/client"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/atomic"
@@ -153,7 +154,13 @@ func (m *litBackendCtxMgr) Register(
 		return nil, err
 	}
 
-	bcCtx := newBackendContext(ctx, jobID, bd, cfg, defaultImportantVariables, m.memRoot, m.diskRoot, etcdClient)
+	physical, logical, err := bd.GetTS(ctx)
+	if err != nil {
+		return nil, err
+	}
+	curTS := oracle.ComposeTS(physical, logical)
+
+	bcCtx := newBackendContext(ctx, jobID, bd, cfg, defaultImportantVariables, m.memRoot, m.diskRoot, etcdClient, curTS)
 	m.backends.m[jobID] = bcCtx
 	m.memRoot.Consume(structSizeBackendCtx)
 	m.backends.mu.Unlock()
@@ -205,6 +212,7 @@ func newBackendContext(
 	memRoot MemRoot,
 	diskRoot DiskRoot,
 	etcdClient *clientv3.Client,
+	initTS uint64,
 ) *litBackendCtx {
 	bCtx := &litBackendCtx{
 		engines:        make(map[int64]*engineInfo, 10),
@@ -217,6 +225,7 @@ func newBackendContext(
 		sysVars:        vars,
 		updateInterval: checkpointUpdateInterval,
 		etcdClient:     etcdClient,
+		initTS:         initTS,
 	}
 	bCtx.timeOfLastFlush.Store(time.Now())
 	return bCtx

pkg/lightning/backend/encode/encode.go

@@ -57,8 +57,10 @@ 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 is used by the dupeDetector. Only the key range with the specified index ID is scanned.
 	IndexID int64
+	// MinCommitTS is used by dupeDetector. Only records that larger than commit TS are considered.
+	MinCommitTS uint64
 }
 
 // Rows represents a collection of encoded rows.

pkg/lightning/backend/local/duplicate.go

@@ -310,6 +310,7 @@ func getDupDetectClient(
 	importClientFactory ImportClientFactory,
 	resourceGroupName string,
 	taskType string,
+	minCommitTS uint64,
 ) (import_sstpb.ImportSST_DuplicateDetectClient, error) {
 	leader := region.Leader
 	if leader == nil {
@@ -330,9 +331,10 @@ func getDupDetectClient(
 		RequestSource: kvutil.BuildRequestSource(true, tidbkv.InternalTxnLightning, taskType),
 	}
 	req := &import_sstpb.DuplicateDetectRequest{
-		Context:  reqCtx,
-		StartKey: keyRange.StartKey,
-		EndKey:   keyRange.EndKey,
+		Context:     reqCtx,
+		StartKey:    keyRange.StartKey,
+		EndKey:      keyRange.EndKey,
+		MinCommitTs: minCommitTS,
 	}
 	cli, err := importClient.DuplicateDetect(ctx, req)
 	if err != nil {
@@ -349,9 +351,10 @@ func NewRemoteDupKVStream(
 	importClientFactory ImportClientFactory,
 	resourceGroupName string,
 	taskType string,
+	minCommitTS uint64,
 ) (*RemoteDupKVStream, error) {
 	subCtx, cancel := context.WithCancel(ctx)
-	cli, err := getDupDetectClient(subCtx, region, keyRange, importClientFactory, resourceGroupName, taskType)
+	cli, err := getDupDetectClient(subCtx, region, keyRange, importClientFactory, resourceGroupName, taskType, minCommitTS)
 	if err != nil {
 		cancel()
 		return nil, errors.Trace(err)
@@ -422,6 +425,7 @@ type dupeDetector struct {
 	indexID           int64
 	resourceGroupName string
 	taskType          string
+	minCommitTS       uint64
 }
 
 // NewDupeDetector creates a new dupeDetector.
@@ -456,6 +460,7 @@ func NewDupeDetector(
 		indexID:           sessOpts.IndexID,
 		resourceGroupName: resourceGroupName,
 		taskType:          taskType,
+		minCommitTS:       sessOpts.MinCommitTS,
 	}, nil
 }
 
@@ -937,7 +942,7 @@ func (m *dupeDetector) processRemoteDupTaskOnce(
 				logutil.Key("dupDetectEndKey", kr.EndKey),
 			)
 			err := func() error {
-				stream, err := NewRemoteDupKVStream(ctx, region, kr, importClientFactory, m.resourceGroupName, m.taskType)
+				stream, err := NewRemoteDupKVStream(ctx, region, kr, importClientFactory, m.resourceGroupName, m.taskType, m.minCommitTS)
 				if err != nil {
 					return errors.Annotatef(err, "failed to create remote duplicate kv stream")
 				}

tests/realtikvtest/addindextest1/disttask_test.go

@@ -20,6 +20,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/config"
@@ -286,3 +287,26 @@ func TestAddUKErrorMessage(t *testing.T) {
 	err := tk.ExecToErr("alter table t add unique index uk(b);")
 	require.ErrorContains(t, err, "Duplicate entry '1' for key 't.uk'")
 }
+
+func TestAddUniqueIndexAfterFlashBack(t *testing.T) {
+	store := realtikvtest.CreateMockStoreAndSetup(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("drop database if exists addindexlit;")
+	tk.MustExec("create database addindexlit;")
+	tk.MustExec("use addindexlit;")
+	tk.MustExec(`set global tidb_ddl_enable_fast_reorg=on;`)
+
+	tk.MustExec("set @@global.tidb_gc_life_time = '100m';")
+	defer tk.MustExec("set @@global.tidb_gc_life_time = default;")
+	tk.MustExec("create table t (a int, b int);")
+	tk.MustExec("insert into t values (1, 1);")
+	time.Sleep(1 * time.Second)
+	tk.MustExec("begin;")
+	currentTime := tk.MustQuery("select @@tidb_current_ts;").Rows()[0][0].(string)
+	tk.MustExec("rollback;")
+	time.Sleep(1 * time.Second) // Fixme: flashback cluster use timestamp instead of tso to check if there is a DDL job.
+	tk.MustExec("alter table t add index idx(b);")
+	tk.MustExec(fmt.Sprintf("flashback cluster to tso %s", currentTime))
+	tk.MustExec("alter table t add unique index idx(b);")
+	tk.MustExec("admin check table t;")
+}

tests/realtikvtest/addindextest3/ingest_test.go

@@ -20,6 +20,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/config"