ddl: fix tidb_ddl_enable_fast_reorg default on ut failed problem

ddl/backfilling.go

@@ -725,7 +725,7 @@ func (b *backfillScheduler) adjustWorkerSize() error {
 	if b.copReqSenderPool != nil {
 		b.copReqSenderPool.adjustSize(len(b.workers))
 	}
-	return injectCheckBackfillWorkerNum(len(b.workers))
+	return injectCheckBackfillWorkerNum(len(b.workers), b.tp == typeAddIndexMergeTmpWorker)
 }
 
 func (b *backfillScheduler) initCopReqSenderPool() {
@@ -871,7 +871,10 @@ func (dc *ddlCtx) writePhysicalTableRecord(sessPool *sessionPool, t table.Physic
 	return nil
 }
 
-func injectCheckBackfillWorkerNum(curWorkerSize int) error {
+func injectCheckBackfillWorkerNum(curWorkerSize int, isMergeWorker bool) error {
+	if isMergeWorker {
+		return nil
+	}
 	failpoint.Inject("checkBackfillWorkerNum", func(val failpoint.Value) {
 		//nolint:forcetypeassert
 		if val.(bool) {

ddl/cancel_test.go

@@ -232,8 +232,6 @@ func TestCancel(t *testing.T) {
 
 	// Prepare schema.
 	tk.MustExec("use test")
-	// TODO: Will check why tidb_ddl_enable_fast_reorg could not default be on in another PR.
-	tk.MustExec("set global tidb_ddl_enable_fast_reorg = 0;")
 	tk.MustExec("drop table if exists t_partition;")
 	tk.MustExec(`create table t_partition (
 		c1 int, c2 int, c3 int

ddl/db_change_test.go

@@ -1719,8 +1719,6 @@ func TestCreateUniqueExpressionIndex(t *testing.T) {
 
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
-	// TODO: will check why tidb_ddl_enable_fast_reorg could not default be on in another pr.pr.
-	tk.MustExec("set global tidb_ddl_enable_fast_reorg = 0;")
 	tk.MustExec("create table t(a int default 0, b int default 0)")
 	tk.MustExec("insert into t values (1, 1), (2, 2), (3, 3), (4, 4)")
 

ddl/failtest/fail_db_test.go

@@ -320,8 +320,6 @@ func TestRunDDLJobPanicEnableClusteredIndex(t *testing.T) {
 	s := createFailDBSuite(t)
 	testAddIndexWorkerNum(t, s, func(tk *testkit.TestKit) {
 		tk.Session().GetSessionVars().EnableClusteredIndex = variable.ClusteredIndexDefModeOn
-		// TODO: will check why tidb_ddl_enable_fast_reorg could not default be on in another pr.
-		tk.MustExec("set global tidb_ddl_enable_fast_reorg = 0;")
 		tk.MustExec("create table test_add_index (c1 bigint, c2 bigint, c3 bigint, primary key(c1, c3))")
 	})
 }
@@ -330,8 +328,6 @@ func TestRunDDLJobPanicEnableClusteredIndex(t *testing.T) {
 func TestRunDDLJobPanicDisableClusteredIndex(t *testing.T) {
 	s := createFailDBSuite(t)
 	testAddIndexWorkerNum(t, s, func(tk *testkit.TestKit) {
-		// TODO: will check why tidb_ddl_enable_fast_reorg could not default be on in another pr.
-		tk.MustExec("set global tidb_ddl_enable_fast_reorg = 0;")
 		tk.MustExec("create table test_add_index (c1 bigint, c2 bigint, c3 bigint, primary key(c1))")
 	})
 }
@@ -424,7 +420,6 @@ func TestPartitionAddIndexGC(t *testing.T) {
 	s := createFailDBSuite(t)
 	tk := testkit.NewTestKit(t, s.store)
 	tk.MustExec("use test")
-	tk.MustExec("set global tidb_ddl_enable_fast_reorg = 0;")
 	tk.MustExec(`create table partition_add_idx (
 	id int not null,
 	hired date not null

ddl/index.go

@@ -802,13 +802,13 @@ func doReorgWorkForCreateIndex(w *worker, d *ddlCtx, t *meta.Meta, job *model.Jo
 			bc, err = ingest.LitBackCtxMgr.Register(w.ctx, indexInfo.Unique, job.ID, job.ReorgMeta.SQLMode)
 			if err != nil {
 				tryFallbackToTxnMerge(job, err)
-				return false, ver, errors.Trace(err)
+				return false, ver, nil
 			}
 			done, ver, err = runReorgJobAndHandleErr(w, d, t, job, tbl, indexInfo, false)
 			if err != nil {
 				ingest.LitBackCtxMgr.Unregister(job.ID)
 				tryFallbackToTxnMerge(job, err)
-				return false, ver, errors.Trace(err)
+				return false, ver, nil
 			}
 			if !done {
 				return false, ver, nil

ddl/index_change_test.go

@@ -37,8 +37,6 @@ func TestIndexChange(t *testing.T) {
 	ddl.SetWaitTimeWhenErrorOccurred(1 * time.Microsecond)
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
-	// TODO: Will check why tidb_ddl_enable_fast_reorg could not default be on in another PR.
-	tk.MustExec("set global tidb_ddl_enable_fast_reorg = 0;")
 	tk.MustExec("create table t (c1 int primary key, c2 int)")
 	tk.MustExec("insert t values (1, 1), (2, 2), (3, 3);")
 
@@ -188,7 +186,11 @@ func checkAddWriteOnlyForAddIndex(ctx sessionctx.Context, delOnlyTbl, writeOnlyT
 		return errors.Trace(err)
 	}
 	// old value index not exists.
-	err = checkIndexExists(ctx, writeOnlyTbl, 1, 4, false)
+	if ddl.IsEnableFastReorg() {
+		err = checkIndexExists(ctx, writeOnlyTbl, 1, 4, true)
+	} else {
+		err = checkIndexExists(ctx, writeOnlyTbl, 1, 4, false)
+	}
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -203,7 +205,11 @@ func checkAddWriteOnlyForAddIndex(ctx sessionctx.Context, delOnlyTbl, writeOnlyT
 	if err != nil {
 		return errors.Trace(err)
 	}
-	err = checkIndexExists(ctx, writeOnlyTbl, 3, 4, false)
+	if ddl.IsEnableFastReorg() {
+		err = checkIndexExists(ctx, writeOnlyTbl, 3, 4, true)
+	} else {
+		err = checkIndexExists(ctx, writeOnlyTbl, 3, 4, false)
+	}
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -213,14 +219,19 @@ func checkAddWriteOnlyForAddIndex(ctx sessionctx.Context, delOnlyTbl, writeOnlyT
 	if err != nil {
 		return errors.Trace(err)
 	}
-	err = checkIndexExists(ctx, writeOnlyTbl, 5, 5, false)
+	if ddl.IsEnableFastReorg() {
+		err = checkIndexExists(ctx, writeOnlyTbl, 5, 5, true)
+	} else {
+		err = checkIndexExists(ctx, writeOnlyTbl, 5, 5, false)
+	}
 	if err != nil {
 		return errors.Trace(err)
 	}
 	return nil
 }
 
 func checkAddPublicForAddIndex(ctx sessionctx.Context, writeTbl, publicTbl table.Table) error {
+	var err1 error
 	// WriteOnlyTable: insert t values (6, 6)
 	err := sessiontxn.NewTxn(context.Background(), ctx)
 	if err != nil {
@@ -231,7 +242,11 @@ func checkAddPublicForAddIndex(ctx sessionctx.Context, writeTbl, publicTbl table
 		return errors.Trace(err)
 	}
 	err = checkIndexExists(ctx, publicTbl, 6, 6, true)
-	if err != nil {
+	if ddl.IsEnableFastReorg() {
+		// Need check temp index also.
+		err1 = checkIndexExists(ctx, writeTbl, 6, 6, true)
+	}
+	if err != nil && err1 != nil {
 		return errors.Trace(err)
 	}
 	// PublicTable: insert t values (7, 7)
@@ -250,10 +265,18 @@ func checkAddPublicForAddIndex(ctx sessionctx.Context, writeTbl, publicTbl table
 		return errors.Trace(err)
 	}
 	err = checkIndexExists(ctx, publicTbl, 5, 7, true)
-	if err != nil {
+	if ddl.IsEnableFastReorg() {
+		// Need check temp index also.
+		err1 = checkIndexExists(ctx, writeTbl, 5, 7, true)
+	}
+	if err != nil && err1 != nil {
 		return errors.Trace(err)
 	}
-	err = checkIndexExists(ctx, publicTbl, 7, 7, false)
+	if ddl.IsEnableFastReorg() {
+		err = checkIndexExists(ctx, publicTbl, 7, 7, true)
+	} else {
+		err = checkIndexExists(ctx, publicTbl, 7, 7, false)
+	}
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -284,6 +307,10 @@ func checkAddPublicForAddIndex(ctx sessionctx.Context, writeTbl, publicTbl table
 		handle := row[0].GetInt64()
 		err = checkIndexExists(ctx, publicTbl, idxVal, handle, true)
 		if err != nil {
+			// Need check temp index also.
+			err1 = checkIndexExists(ctx, writeTbl, idxVal, handle, true)
+		}
+		if err != nil && err1 != nil {
 			return errors.Trace(err)
 		}
 	}

ddl/ingest/env.go

@@ -19,6 +19,7 @@ import (
 	"path/filepath"
 	"strconv"
 
+	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/br/pkg/lightning/log"
 	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/util"
@@ -47,6 +48,13 @@ const maxMemoryQuota = 2 * size.GB
 // InitGlobalLightningEnv initialize Lightning backfill environment.
 func InitGlobalLightningEnv() {
 	log.SetAppLogger(logutil.BgLogger())
+	globalCfg := config.GetGlobalConfig()
+	if globalCfg.Store != "tikv" {
+		logutil.BgLogger().Warn(LitWarnEnvInitFail,
+			zap.Error(errors.New("only support TiKV storage")),
+			zap.String("current storage", globalCfg.Store),
+			zap.Bool("lightning is initialized", LitInitialized))
+	}
 	sPath, err := genLightningDataDir()
 	if err != nil {
 		logutil.BgLogger().Warn(LitWarnEnvInitFail, zap.Error(err),

ddl/multi_schema_change_test.go

@@ -568,8 +568,6 @@ func TestMultiSchemaChangeAddIndexesCancelled(t *testing.T) {
 	store, dom := testkit.CreateMockStoreAndDomain(t)
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
-	// TODO: will check why tidb_ddl_enable_fast_reorg could not default be on in another pr.
-	tk.MustExec("set global tidb_ddl_enable_fast_reorg = 0;")
 	originHook := dom.DDL().GetHook()
 
 	// Test cancel successfully.

ddl/serial_test.go

@@ -432,8 +432,6 @@ func TestCancelAddIndexPanic(t *testing.T) {
 		require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/errorMockPanic"))
 	}()
 	tk.MustExec("use test")
-	// TODO: will check why tidb_ddl_enable_fast_reorg could not default be on in another pr.
-	tk.MustExec("set global tidb_ddl_enable_fast_reorg = 0;")
 	tk.MustExec("drop table if exists t")
 	tk.MustExec("create table t(c1 int, c2 int)")
 

ddl/stat_test.go

@@ -52,8 +52,6 @@ func TestDDLStatsInfo(t *testing.T) {
 	tblInfo, err := testTableInfo(store, "t", 2)
 	require.NoError(t, err)
 	testCreateTable(t, ctx, d, dbInfo, tblInfo)
-	// TODO: will check why tidb_ddl_enable_fast_reorg could not default be on in another pr.
-	tk.MustExec("set global tidb_ddl_enable_fast_reorg = 0;")
 	err = sessiontxn.NewTxn(context.Background(), ctx)
 	require.NoError(t, err)
 

executor/batch_checker.go

@@ -180,6 +180,9 @@ func getKeysNeedCheckOneRow(ctx sessionctx.Context, t table.Table, row []types.D
 		if !distinct {
 			continue
 		}
+		if v.Meta().BackfillState == model.BackfillStateRunning {
+			_, key, _ = tables.GenTempIdxKeyByState(v.Meta(), key)
+		}
 		colValStr, err1 := formatDataForDupError(colVals)
 		if err1 != nil {
 			return nil, err1

executor/insert.go

@@ -15,9 +15,11 @@
 package executor
 
 import (
+	"bytes"
 	"context"
 	"encoding/hex"
 	"fmt"
+	"github.com/pingcap/tidb/table/tables"
 	"runtime/trace"
 	"time"
 
@@ -264,6 +266,10 @@ func (e *InsertExec) batchUpdateDupRows(ctx context.Context, newRows [][]types.D
 				}
 				return err
 			}
+			rowVal := val[:len(val)-1]
+			if bytes.Equal(rowVal, tables.DeleteMarkerUnique) {
+				continue
+			}
 			handle, err := tablecodec.DecodeHandleInUniqueIndexValue(val, uk.commonHandle)
 			if err != nil {
 				return err

table/tables/index.go

@@ -15,6 +15,7 @@
 package tables
 
 import (
+	"bytes"
 	"context"
 	"sync"
 
@@ -127,7 +128,7 @@ func (c *index) Create(sctx sessionctx.Context, txn kv.Transaction, indexedValue
 		keyIsTempIdxKey bool
 	)
 	if !opt.FromBackFill {
-		key, tempKey, keyVer = genTempIdxKeyByState(c.idxInfo, key)
+		key, tempKey, keyVer = GenTempIdxKeyByState(c.idxInfo, key)
 		if keyVer == TempIndexKeyTypeBackfill {
 			key, tempKey = tempKey, nil
 			keyIsTempIdxKey = true
@@ -230,7 +231,19 @@ func (c *index) Create(sctx sessionctx.Context, txn kv.Transaction, indexedValue
 			idxVal = append(idxVal, keyVer)
 		}
 		if lazyCheck {
-			flags := []kv.FlagsOp{kv.SetPresumeKeyNotExists}
+			var (
+				needPresumeKey bool
+				flags          []kv.FlagsOp
+			)
+			if !opt.FromBackFill {
+				needPresumeKey, err = KeyExistInTempIndex(txn, key, h, c.tblInfo.IsCommonHandle)
+				if err != nil {
+					return nil, err
+				}
+			}
+			if !needPresumeKey {
+				flags = []kv.FlagsOp{kv.SetPresumeKeyNotExists}
+			}
 			if !vars.ConstraintCheckInPlacePessimistic && vars.TxnCtx.IsPessimistic && vars.InTxn() &&
 				!vars.InRestrictedSQL && vars.ConnectionID > 0 {
 				flags = append(flags, kv.SetNeedConstraintCheckInPrewrite)
@@ -285,7 +298,7 @@ func (c *index) Delete(sc *stmtctx.StatementContext, txn kv.Transaction, indexed
 		return err
 	}
 
-	key, tempKey, tempKeyVer := genTempIdxKeyByState(c.idxInfo, key)
+	key, tempKey, tempKeyVer := GenTempIdxKeyByState(c.idxInfo, key)
 
 	if distinct {
 		if len(key) > 0 {
@@ -336,9 +349,9 @@ const (
 	TempIndexKeyTypeMerge byte = 'm'
 )
 
-// genTempIdxKeyByState is used to get the key version and the temporary key.
+// GenTempIdxKeyByState is used to get the key version and the temporary key.
 // The tempKeyVer means the temp index key/value version.
-func genTempIdxKeyByState(indexInfo *model.IndexInfo, indexKey kv.Key) (key, tempKey kv.Key, tempKeyVer byte) {
+func GenTempIdxKeyByState(indexInfo *model.IndexInfo, indexKey kv.Key) (key, tempKey kv.Key, tempKeyVer byte) {
 	if indexInfo.State != model.StatePublic {
 		switch indexInfo.BackfillState {
 		case model.BackfillStateInapplicable:
@@ -364,6 +377,14 @@ func (c *index) Exist(sc *stmtctx.StatementContext, txn kv.Transaction, indexedV
 		return false, nil, err
 	}
 
+	var (
+		tempKey kv.Key
+		keyVer  byte
+	)
+	key, tempKey, keyVer = GenTempIdxKeyByState(c.idxInfo, key)
+	if keyVer == TempIndexKeyTypeBackfill {
+		key, tempKey = tempKey, nil
+	}
 	value, err := txn.Get(context.TODO(), key)
 	if kv.IsErrNotFound(err) {
 		return false, nil, nil
@@ -463,3 +484,31 @@ func TryAppendCommonHandleRowcodecColInfos(colInfo []rowcodec.ColInfo, tblInfo *
 	}
 	return colInfo
 }
+
+// KeyExistInTempIndex is used to check if there is unique key is marked delete in temp index.
+func KeyExistInTempIndex(txn kv.Transaction, key kv.Key, h kv.Handle, IsCommonHandle bool) (bool, error) {
+	value, err := txn.Get(context.TODO(), key)
+	if kv.IsErrNotFound(err) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+
+	length := len(value)
+	value = value[:length-1]
+	if bytes.Equal(value, DeleteMarkerUnique) {
+		return true, nil
+	}
+	// Check if handle equal?
+	var handle kv.Handle
+	handle, err = tablecodec.DecodeHandleInUniqueIndexValue(value, IsCommonHandle)
+	if err != nil {
+		return false, err
+	}
+	if !handle.Equal(h) {
+		return false, kv.ErrKeyExists
+	}
+
+	return true, nil
+}