*: redirect the DML writes to a temp index during creating index

ddl/BUILD.bazel

@@ -27,6 +27,7 @@ go_library(
         "foreign_key.go",
         "generated_column.go",
         "index.go",
+        "index_merge_tmp.go",
         "job_table.go",
         "mock.go",
         "multi_schema_change.go",
@@ -164,6 +165,7 @@ go_test(
         "fail_test.go",
         "foreign_key_test.go",
         "index_change_test.go",
+        "index_merge_tmp_test.go",
         "index_modify_test.go",
         "integration_test.go",
         "job_table_test.go",
@@ -193,6 +195,7 @@ go_test(
     shard_count = 50,
     deps = [
         "//config",
+        "//ddl/ingest",
         "//ddl/placement",
         "//ddl/schematracker",
         "//ddl/testutil",

ddl/backfilling.go

@@ -51,9 +51,10 @@ import (
 type backfillWorkerType byte
 
 const (
-	typeAddIndexWorker     backfillWorkerType = 0
-	typeUpdateColumnWorker backfillWorkerType = 1
-	typeCleanUpIndexWorker backfillWorkerType = 2
+	typeAddIndexWorker         backfillWorkerType = 0
+	typeUpdateColumnWorker     backfillWorkerType = 1
+	typeCleanUpIndexWorker     backfillWorkerType = 2
+	typeAddIndexMergeTmpWorker backfillWorkerType = 3
 )
 
 // By now the DDL jobs that need backfilling include:
@@ -116,6 +117,8 @@ func (bWT backfillWorkerType) String() string {
 		return "update column"
 	case typeCleanUpIndexWorker:
 		return "clean up index"
+	case typeAddIndexMergeTmpWorker:
+		return "merge temporary index"
 	default:
 		return "unknown"
 	}
@@ -466,15 +469,21 @@ func (dc *ddlCtx) handleRangeTasks(sessPool *sessionPool, t table.Table, workers
 	batchTasks := make([]*reorgBackfillTask, 0, len(workers))
 	physicalTableID := reorgInfo.PhysicalTableID
 
+	var prefix kv.Key
+	if reorgInfo.mergingTmpIdx {
+		prefix = t.IndexPrefix()
+	} else {
+		prefix = t.RecordPrefix()
+	}
 	// Build reorg tasks.
 	for i, keyRange := range kvRanges {
 		endKey := keyRange.EndKey
-		endK, err := getRangeEndKey(reorgInfo.d.jobContext(reorgInfo.Job), workers[0].sessCtx.GetStore(), workers[0].priority, t.RecordPrefix(), keyRange.StartKey, endKey)
+		endK, err := getRangeEndKey(reorgInfo.d.jobContext(reorgInfo.Job), workers[0].sessCtx.GetStore(), workers[0].priority, prefix, keyRange.StartKey, endKey)
 		if err != nil {
 			logutil.BgLogger().Info("[ddl] send range task to workers, get reverse key failed", zap.Error(err))
 		} else {
 			logutil.BgLogger().Info("[ddl] send range task to workers, change end key",
-				zap.String("end key", tryDecodeToHandleString(endKey)), zap.String("current end key", tryDecodeToHandleString(endK)))
+				zap.String("end key", hex.EncodeToString(endKey)), zap.String("current end key", hex.EncodeToString(endK)))
 			endKey = endK
 		}
 
@@ -646,9 +655,16 @@ func (dc *ddlCtx) writePhysicalTableRecord(sessPool *sessionPool, t table.Physic
 
 			switch bfWorkerType {
 			case typeAddIndexWorker:
-				idxWorker := newAddIndexWorker(sessCtx, i, t, decodeColMap, reorgInfo, jc)
+				idxWorker, err := newAddIndexWorker(sessCtx, i, t, decodeColMap, reorgInfo, jc)
+				if err != nil {
+					return errors.Trace(err)
+				}
 				backfillWorkers = append(backfillWorkers, idxWorker.backfillWorker)
 				go idxWorker.backfillWorker.run(reorgInfo.d, idxWorker, job)
+			case typeAddIndexMergeTmpWorker:
+				tmpIdxWorker := newMergeTempIndexWorker(sessCtx, i, t, reorgInfo, jc)
+				backfillWorkers = append(backfillWorkers, tmpIdxWorker.backfillWorker)
+				go tmpIdxWorker.backfillWorker.run(reorgInfo.d, tmpIdxWorker, job)
 			case typeUpdateColumnWorker:
 				// Setting InCreateOrAlterStmt tells the difference between SELECT casting and ALTER COLUMN casting.
 				sessCtx.GetSessionVars().StmtCtx.InCreateOrAlterStmt = true

ddl/column.go

@@ -797,8 +797,9 @@ func doReorgWorkForModifyColumnMultiSchema(w *worker, d *ddlCtx, t *meta.Meta, j
 
 func doReorgWorkForModifyColumn(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job, tbl table.Table,
 	oldCol, changingCol *model.ColumnInfo, changingIdxs []*model.IndexInfo) (done bool, ver int64, err error) {
+	job.ReorgMeta.ReorgTp = model.ReorgTypeTxn
 	rh := newReorgHandler(t, w.sess, w.concurrentDDL)
-	reorgInfo, err := getReorgInfo(d.jobContext(job), d, rh, job, tbl, BuildElements(changingCol, changingIdxs))
+	reorgInfo, err := getReorgInfo(d.jobContext(job), d, rh, job, tbl, BuildElements(changingCol, changingIdxs), false)
 	if err != nil || reorgInfo.first {
 		// If we run reorg firstly, we should update the job snapshot version
 		// and then run the reorg next time.

ddl/ddl_worker.go

@@ -514,14 +514,9 @@ func jobNeedGC(job *model.Job) bool {
 			return false
 		}
 		switch job.Type {
-		case model.ActionAddIndex, model.ActionAddPrimaryKey:
-			if job.State != model.JobStateRollbackDone {
-				break
-			}
-			// After rolling back an AddIndex operation, we need to use delete-range to delete the half-done index data.
-			return true
 		case model.ActionDropSchema, model.ActionDropTable, model.ActionTruncateTable, model.ActionDropIndex, model.ActionDropPrimaryKey,
-			model.ActionDropTablePartition, model.ActionTruncateTablePartition, model.ActionDropColumn, model.ActionModifyColumn:
+			model.ActionDropTablePartition, model.ActionTruncateTablePartition, model.ActionDropColumn, model.ActionModifyColumn,
+			model.ActionAddIndex, model.ActionAddPrimaryKey:
 			return true
 		case model.ActionMultiSchemaChange:
 			for _, sub := range job.MultiSchemaInfo.SubJobs {

ddl/ddl_worker_test.go

@@ -269,22 +269,30 @@ func TestJobNeedGC(t *testing.T) {
 	job := &model.Job{Type: model.ActionAddIndex, State: model.JobStateCancelled}
 	require.False(t, ddl.JobNeedGCForTest(job))
 
-	job = &model.Job{Type: model.ActionAddIndex, State: model.JobStateDone}
+	job = &model.Job{Type: model.ActionAddColumn, State: model.JobStateDone}
 	require.False(t, ddl.JobNeedGCForTest(job))
+	job = &model.Job{Type: model.ActionAddIndex, State: model.JobStateDone}
+	require.True(t, ddl.JobNeedGCForTest(job))
 	job = &model.Job{Type: model.ActionAddPrimaryKey, State: model.JobStateDone}
-	require.False(t, ddl.JobNeedGCForTest(job))
+	require.True(t, ddl.JobNeedGCForTest(job))
 	job = &model.Job{Type: model.ActionAddIndex, State: model.JobStateRollbackDone}
 	require.True(t, ddl.JobNeedGCForTest(job))
 	job = &model.Job{Type: model.ActionAddPrimaryKey, State: model.JobStateRollbackDone}
 	require.True(t, ddl.JobNeedGCForTest(job))
 
 	job = &model.Job{Type: model.ActionMultiSchemaChange, State: model.JobStateDone, MultiSchemaInfo: &model.MultiSchemaInfo{
 		SubJobs: []*model.SubJob{
-			{Type: model.ActionAddIndex, State: model.JobStateDone},
 			{Type: model.ActionAddColumn, State: model.JobStateDone},
 			{Type: model.ActionRebaseAutoID, State: model.JobStateDone},
 		}}}
 	require.False(t, ddl.JobNeedGCForTest(job))
+	job = &model.Job{Type: model.ActionMultiSchemaChange, State: model.JobStateDone, MultiSchemaInfo: &model.MultiSchemaInfo{
+		SubJobs: []*model.SubJob{
+			{Type: model.ActionAddIndex, State: model.JobStateDone},
+			{Type: model.ActionAddColumn, State: model.JobStateDone},
+			{Type: model.ActionRebaseAutoID, State: model.JobStateDone},
+		}}}
+	require.True(t, ddl.JobNeedGCForTest(job))
 	job = &model.Job{Type: model.ActionMultiSchemaChange, State: model.JobStateDone, MultiSchemaInfo: &model.MultiSchemaInfo{
 		SubJobs: []*model.SubJob{
 			{Type: model.ActionAddIndex, State: model.JobStateDone},

ddl/delete_range.go

@@ -322,27 +322,34 @@ func insertJobIntoDeleteRangeTable(ctx context.Context, sctx sessionctx.Context,
 		}
 	// ActionAddIndex, ActionAddPrimaryKey needs do it, because it needs to be rolled back when it's canceled.
 	case model.ActionAddIndex, model.ActionAddPrimaryKey:
-		tableID := job.TableID
 		var indexID int64
 		var ifExists bool
 		var partitionIDs []int64
 		if err := job.DecodeArgs(&indexID, &ifExists, &partitionIDs); err != nil {
 			return errors.Trace(err)
 		}
+		// Determine the physicalIDs to be added.
+		physicalIDs := []int64{job.TableID}
 		if len(partitionIDs) > 0 {
-			for _, pid := range partitionIDs {
-				startKey := tablecodec.EncodeTableIndexPrefix(pid, indexID)
-				endKey := tablecodec.EncodeTableIndexPrefix(pid, indexID+1)
-				elemID := ea.allocForIndexID(pid, indexID)
-				if err := doInsert(ctx, s, job.ID, elemID, startKey, endKey, now, fmt.Sprintf("partition table ID is %d", pid)); err != nil {
+			physicalIDs = partitionIDs
+		}
+		// Determine the index IDs to be added.
+		tempIdxID := tablecodec.TempIndexPrefix | indexID
+		var indexIDs []int64
+		if job.State == model.JobStateRollbackDone {
+			indexIDs = []int64{indexID, tempIdxID}
+		} else {
+			indexIDs = []int64{tempIdxID}
+		}
+		for _, pid := range physicalIDs {
+			for _, iid := range indexIDs {
+				startKey := tablecodec.EncodeTableIndexPrefix(pid, iid)
+				endKey := tablecodec.EncodeTableIndexPrefix(pid, iid+1)
+				elemID := ea.allocForIndexID(pid, iid)
+				if err := doInsert(ctx, s, job.ID, elemID, startKey, endKey, now, fmt.Sprintf("physical ID is %d", pid)); err != nil {
 					return errors.Trace(err)
 				}
 			}
-		} else {
-			startKey := tablecodec.EncodeTableIndexPrefix(tableID, indexID)
-			endKey := tablecodec.EncodeTableIndexPrefix(tableID, indexID+1)
-			elemID := ea.allocForIndexID(tableID, indexID)
-			return doInsert(ctx, s, job.ID, elemID, startKey, endKey, now, fmt.Sprintf("table ID is %d", tableID))
 		}
 	case model.ActionDropIndex, model.ActionDropPrimaryKey:
 		tableID := job.TableID