ddl: check the key existence on original index

ddl/index.go

@@ -875,6 +875,12 @@ func doReorgWorkForCreateIndex(w *worker, d *ddlCtx, t *meta.Meta, job *model.Jo
 		ver, err = updateVersionAndTableInfo(d, t, job, tbl.Meta(), true)
 		return false, ver, errors.Trace(err)
 	case model.BackfillStateMerging:
+		failpoint.Inject("mockDMLExecutionMerging", func(val failpoint.Value) {
+			//nolint:forcetypeassert
+			if val.(bool) && MockDMLExecutionMerging != nil {
+				MockDMLExecutionMerging()
+			}
+		})
 		done, ver, err = runReorgJobAndHandleErr(w, d, t, job, tbl, indexInfo, true)
 		if !done {
 			return false, ver, err
@@ -1713,6 +1719,9 @@ func (w *addIndexWorker) BackfillDataInTxn(handleRange reorgBackfillTask) (taskC
 // MockDMLExecution is only used for test.
 var MockDMLExecution func()
 
+// MockDMLExecutionMerging is only used for test.
+var MockDMLExecutionMerging func()
+
 func (w *worker) addPhysicalTableIndex(t table.PhysicalTable, reorgInfo *reorgInfo) error {
 	if reorgInfo.mergingTmpIdx {
 		logutil.BgLogger().Info("[ddl] start to merge temp index", zap.String("job", reorgInfo.Job.String()), zap.String("reorgInfo", reorgInfo.String()))

ddl/indexmergetest/merge_test.go

@@ -521,3 +521,122 @@ func TestAddIndexMergeConflictWithPessimistic(t *testing.T) {
 	tk.MustExec("admin check table t;")
 	tk.MustQuery("select * from t;").Check(testkit.Rows("1 2"))
 }
+
+func TestAddIndexMergeInsertOnMerging(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t(a int default 0, b int default 0)")
+
+	tk1 := testkit.NewTestKit(t, store)
+	tk1.MustExec("use test")
+
+	d := dom.DDL()
+	originalCallback := d.GetHook()
+	defer d.SetHook(originalCallback)
+	callback := &callback.TestDDLCallback{}
+	onJobUpdatedExportedFunc := func(job *model.Job) {
+		if t.Failed() {
+			return
+		}
+		var err error
+		switch job.SchemaState {
+		case model.StateDeleteOnly:
+			_, err = tk1.Exec("insert into t values (5, 5)")
+			assert.NoError(t, err)
+		case model.StateWriteOnly:
+			_, err = tk1.Exec("insert into t values (5, 7)")
+			assert.NoError(t, err)
+			_, err = tk1.Exec("delete from t where b = 7")
+			assert.NoError(t, err)
+		}
+	}
+	callback.OnJobUpdatedExported.Store(&onJobUpdatedExportedFunc)
+	d.SetHook(callback)
+
+	ddl.MockDMLExecutionMerging = func() {
+		_, err := tk1.Exec("insert into t values (5, 8);")
+		assert.Error(t, err) // [kv:1062]Duplicate entry '5' for key 't.idx'
+		_, err = tk1.Exec("insert into t values (5, 8) on duplicate key update a = 6;")
+		assert.NoError(t, err) // The row should be normally updated to (6, 5).
+	}
+	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/mockDMLExecutionMerging", "1*return(true)->return(false)"))
+	tk.MustExec("alter table t add unique index idx(a);")
+	tk.MustExec("admin check table t;")
+	tk.MustQuery("select * from t;").Check(testkit.Rows("6 5"))
+	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockDMLExecutionMerging"))
+}
+
+func TestAddIndexMergeReplaceOnMerging(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t(a int default 0, b int default 0);")
+	tk.MustExec("insert into t values (5, 5);")
+
+	tk1 := testkit.NewTestKit(t, store)
+	tk1.MustExec("use test")
+
+	ddl.MockDMLExecution = func() {
+		_, err := tk1.Exec("delete from t where b = 5;")
+		assert.NoError(t, err)
+	}
+
+	ddl.MockDMLExecutionMerging = func() {
+		_, err := tk1.Exec("replace into t values (5, 8);")
+		assert.NoError(t, err)
+	}
+	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/mockDMLExecution", "1*return(true)->return(false)"))
+	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/ddl/mockDMLExecutionMerging", "1*return(true)->return(false)"))
+	tk.MustExec("alter table t add unique index idx(a);")
+	tk.MustExec("admin check table t;")
+	tk.MustQuery("select * from t;").Check(testkit.Rows("5 8"))
+	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockDMLExecution"))
+	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockDMLExecutionMerging"))
+}
+
+func TestAddIndexMergeInsertToDeletedTempIndex(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t(a int default 0, b int default 0)")
+	tk.MustExec("insert into t values (5, 5);")
+
+	tk1 := testkit.NewTestKit(t, store)
+	tk1.MustExec("use test")
+
+	d := dom.DDL()
+	originalCallback := d.GetHook()
+	defer d.SetHook(originalCallback)
+	callback := &callback.TestDDLCallback{}
+	onJobUpdatedExportedFunc := func(job *model.Job) {
+		if t.Failed() {
+			return
+		}
+		var err error
+		switch job.SchemaState {
+		case model.StateWriteOnly:
+			_, err = tk1.Exec("delete from t where b = 5")
+			assert.NoError(t, err)
+			_, err := tk1.Exec("set @@tidb_constraint_check_in_place = true;")
+			assert.NoError(t, err)
+			_, err = tk1.Exec("insert into t values (5, 8);")
+			assert.NoError(t, err)
+			_, err = tk1.Exec("insert into t values (5, 8);")
+			assert.Error(t, err)
+			_, err = tk1.Exec("set @@tidb_constraint_check_in_place = false;")
+			assert.NoError(t, err)
+			_, err = tk1.Exec("insert into t values (5, 8);")
+			assert.Error(t, err)
+		}
+	}
+	callback.OnJobUpdatedExported.Store(&onJobUpdatedExportedFunc)
+	d.SetHook(callback)
+
+	tk.MustExec("alter table t add unique index idx(a);")
+	tk.MustExec("admin check table t;")
+	tk.MustQuery("select * from t;").Check(testkit.Rows("5 8"))
+}

executor/insert.go

@@ -31,6 +31,7 @@ import (
 	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/parser/terror"
 	"github.com/pingcap/tidb/table"
+	"github.com/pingcap/tidb/table/tables"
 	"github.com/pingcap/tidb/tablecodec"
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/chunk"
@@ -158,11 +159,8 @@ func prefetchConflictedOldRows(ctx context.Context, txn kv.Transaction, rows []t
 	for _, r := range rows {
 		for _, uk := range r.uniqueKeys {
 			if val, found := values[string(uk.newKey)]; found {
-				if tablecodec.IsTempIndexKey(uk.newKey) {
-					if tablecodec.CheckTempIndexValueIsDelete(val) {
-						continue
-					}
-					val = tablecodec.DecodeTempIndexOriginValue(val)
+				if isTemp, _ := tablecodec.IsTempIndexKey(uk.newKey); isTemp {
+					continue
 				}
 				handle, err := tablecodec.DecodeHandleInUniqueIndexValue(val, uk.commonHandle)
 				if err != nil {
@@ -261,26 +259,13 @@ func (e *InsertExec) batchUpdateDupRows(ctx context.Context, newRows [][]types.D
 		}
 
 		for _, uk := range r.uniqueKeys {
-			val, err := txn.Get(ctx, uk.newKey)
+			_, handle, err := tables.FetchDuplicatedHandle(ctx, uk.newKey, true, txn, e.Table.Meta().ID, uk.commonHandle)
 			if err != nil {
-				if kv.IsErrNotFound(err) {
-					continue
-				}
 				return err
 			}
-			// Since the temp index stores deleted key with marked 'deleteu' for unique key at the end
-			// of value, So if return a key we check and skip deleted key.
-			if tablecodec.IsTempIndexKey(uk.newKey) {
-				if tablecodec.CheckTempIndexValueIsDelete(val) {
-					continue
-				}
-				val = tablecodec.DecodeTempIndexOriginValue(val)
-			}
-			handle, err := tablecodec.DecodeHandleInUniqueIndexValue(val, uk.commonHandle)
-			if err != nil {
-				return err
+			if handle == nil {
+				continue
 			}
-
 			err = e.updateDupRow(ctx, i, txn, r, handle, e.OnDuplicate)
 			if err != nil {
 				if kv.IsErrNotFound(err) {

executor/replace.go

@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/tidb/meta/autoid"
 	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/sessionctx/stmtctx"
+	"github.com/pingcap/tidb/table/tables"
 	"github.com/pingcap/tidb/tablecodec"
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/chunk"
@@ -176,22 +177,12 @@ func (e *ReplaceExec) replaceRow(ctx context.Context, r toBeCheckedRow) error {
 //  3. error: the error.
 func (e *ReplaceExec) removeIndexRow(ctx context.Context, txn kv.Transaction, r toBeCheckedRow) (bool, bool, error) {
 	for _, uk := range r.uniqueKeys {
-		val, err := txn.Get(ctx, uk.newKey)
+		_, handle, err := tables.FetchDuplicatedHandle(ctx, uk.newKey, true, txn, e.Table.Meta().ID, uk.commonHandle)
 		if err != nil {
-			if kv.IsErrNotFound(err) {
-				continue
-			}
 			return false, false, err
 		}
-		if tablecodec.IsTempIndexKey(uk.newKey) {
-			if tablecodec.CheckTempIndexValueIsDelete(val) {
-				continue
-			}
-			val = tablecodec.DecodeTempIndexOriginValue(val)
-		}
-		handle, err := tablecodec.DecodeHandleInUniqueIndexValue(val, uk.commonHandle)
-		if err != nil {
-			return false, true, err
+		if handle == nil {
+			continue
 		}
 		rowUnchanged, err := e.removeRow(ctx, txn, handle, r)
 		if err != nil {