This is an automated cherry-pick of pingcap#57813

Signed-off-by: ti-chi-bot <[email protected]>

Author: tangenta

pkg/ddl/backfilling.go

@@ -278,7 +278,7 @@ func (w *backfillWorker) handleBackfillTask(d *ddlCtx, task *reorgBackfillTask,
 		// we will never cancel the job once there is panic in bf.BackfillData.
 		// Because reorgRecordTask may run a long time,
 		// we should check whether this ddl job is still runnable.
-		err := d.isReorgRunnable(jobID, false)
+		err := d.isReorgRunnable(d.ctx, false)
 		if err != nil {
 			result.err = err
 			return result
@@ -588,11 +588,73 @@ func makeupDecodeColMap(sessCtx sessionctx.Context, dbName model.CIStr, t table.
 	return decodeColMap, nil
 }
 
+<<<<<<< HEAD
 func setSessCtxLocation(sctx sessionctx.Context, tzLocation *model.TimeZoneLocation) error {
 	// It is set to SystemLocation to be compatible with nil LocationInfo.
 	tz := *timeutil.SystemLocation()
 	if sctx.GetSessionVars().TimeZone == nil {
 		sctx.GetSessionVars().TimeZone = &tz
+=======
+const backfillTaskChanSize = 128
+
+func (dc *ddlCtx) runAddIndexInLocalIngestMode(
+	ctx context.Context,
+	sessPool *sess.Pool,
+	t table.PhysicalTable,
+	reorgInfo *reorgInfo,
+) error {
+	if err := dc.isReorgRunnable(ctx, false); err != nil {
+		return errors.Trace(err)
+	}
+	job := reorgInfo.Job
+	opCtx, cancel := NewLocalOperatorCtx(ctx, job.ID)
+	defer cancel()
+
+	idxCnt := len(reorgInfo.elements)
+	indexIDs := make([]int64, 0, idxCnt)
+	indexInfos := make([]*model.IndexInfo, 0, idxCnt)
+	uniques := make([]bool, 0, idxCnt)
+	hasUnique := false
+	for _, e := range reorgInfo.elements {
+		indexIDs = append(indexIDs, e.ID)
+		indexInfo := model.FindIndexInfoByID(t.Meta().Indices, e.ID)
+		if indexInfo == nil {
+			logutil.DDLIngestLogger().Warn("index info not found",
+				zap.Int64("jobID", job.ID),
+				zap.Int64("tableID", t.Meta().ID),
+				zap.Int64("indexID", e.ID))
+			return errors.Errorf("index info not found: %d", e.ID)
+		}
+		indexInfos = append(indexInfos, indexInfo)
+		uniques = append(uniques, indexInfo.Unique)
+		hasUnique = hasUnique || indexInfo.Unique
+	}
+
+	//nolint: forcetypeassert
+	discovery := dc.store.(tikv.Storage).GetRegionCache().PDClient().GetServiceDiscovery()
+	importConc := job.ReorgMeta.GetConcurrencyOrDefault(int(variable.GetDDLReorgWorkerCounter()))
+	maxWriteSpeed := job.ReorgMeta.GetMaxWriteSpeedOrDefault()
+	bcCtx, err := ingest.LitBackCtxMgr.Register(
+		ctx, job.ID, hasUnique, nil, discovery, job.ReorgMeta.ResourceGroupName, importConc, maxWriteSpeed, job.RealStartTS)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	defer ingest.LitBackCtxMgr.Unregister(job.ID)
+
+	cpMgr, err := ingest.NewCheckpointManager(
+		ctx,
+		sessPool,
+		reorgInfo.PhysicalTableID,
+		job.ID,
+		indexIDs,
+		ingest.LitBackCtxMgr.EncodeJobSortPath(job.ID),
+		dc.store.(kv.StorageWithPD).GetPDClient(),
+	)
+	if err != nil {
+		logutil.DDLIngestLogger().Warn("create checkpoint manager failed",
+			zap.Int64("jobID", job.ID),
+			zap.Error(err))
+>>>>>>> 575310677da (ddl: check context done in isReorgRunnable function (#57813))
 	} else {
 		*sctx.GetSessionVars().TimeZone = tz
 	}
@@ -636,7 +698,7 @@ func (dc *ddlCtx) writePhysicalTableRecord(
 ) error {
 	startKey, endKey := reorgInfo.StartKey, reorgInfo.EndKey
 
-	if err := dc.isReorgRunnable(reorgInfo.Job.ID, false); err != nil {
+	if err := dc.isReorgRunnable(ctx, false); err != nil {
 		return errors.Trace(err)
 	}
 

pkg/ddl/ddl_test.go

@@ -45,6 +45,7 @@ type DDLForTest interface {
 	RemoveReorgCtx(id int64)
 }
 
+<<<<<<< HEAD
 // SetInterceptor implements DDL.SetInterceptor interface.
 func (d *ddl) SetInterceptor(i Interceptor) {
 	d.mu.Lock()
@@ -58,6 +59,8 @@ func (rc *reorgCtx) IsReorgCanceled() bool {
 	return rc.isReorgCanceled()
 }
 
+=======
+>>>>>>> 575310677da (ddl: check context done in isReorgRunnable function (#57813))
 // NewReorgCtx exports for testing.
 func (d *ddl) NewReorgCtx(jobID int64, rowCount int64) *reorgCtx {
 	return d.newReorgCtx(jobID, rowCount)

pkg/ddl/ddl_worker_test.go

@@ -304,27 +304,3 @@ func TestJobNeedGC(t *testing.T) {
 		}}}
 	require.True(t, ddl.JobNeedGCForTest(job))
 }
-
-func TestUsingReorgCtx(t *testing.T) {
-	_, domain := testkit.CreateMockStoreAndDomainWithSchemaLease(t, testLease)
-	d := domain.DDL()
-
-	wg := util.WaitGroupWrapper{}
-	wg.Run(func() {
-		jobID := int64(1)
-		for i := 0; i < 500; i++ {
-			d.(ddl.DDLForTest).NewReorgCtx(jobID, 0)
-			d.(ddl.DDLForTest).GetReorgCtx(jobID).IsReorgCanceled()
-			d.(ddl.DDLForTest).RemoveReorgCtx(jobID)
-		}
-	})
-	wg.Run(func() {
-		jobID := int64(1)
-		for i := 0; i < 500; i++ {
-			d.(ddl.DDLForTest).NewReorgCtx(jobID, 0)
-			d.(ddl.DDLForTest).GetReorgCtx(jobID).IsReorgCanceled()
-			d.(ddl.DDLForTest).RemoveReorgCtx(jobID)
-		}
-	})
-	wg.Wait()
-}

pkg/ddl/index.go

@@ -2009,7 +2009,11 @@ func (w *worker) addTableIndex(t table.Table, reorgInfo *reorgInfo) error {
 	// TODO: Support typeAddIndexMergeTmpWorker.
 	if reorgInfo.ReorgMeta.IsDistReorg && !reorgInfo.mergingTmpIdx {
 		if reorgInfo.ReorgMeta.ReorgTp == model.ReorgTypeLitMerge {
+<<<<<<< HEAD
 			err := w.executeDistGlobalTask(reorgInfo)
+=======
+			err := w.executeDistTask(ctx, t, reorgInfo)
+>>>>>>> 575310677da (ddl: check context done in isReorgRunnable function (#57813))
 			if err != nil {
 				return err
 			}
@@ -2082,6 +2086,7 @@ func checkDuplicateForUniqueIndex(ctx context.Context, t table.Table, reorgInfo
 	return nil
 }
 
+<<<<<<< HEAD
 // MockDMLExecutionOnTaskFinished is used to mock DML execution when tasks finished.
 var MockDMLExecutionOnTaskFinished func()
 
@@ -2092,6 +2097,9 @@ var MockDMLExecutionOnDDLPaused func()
 var TestSyncChan = make(chan struct{})
 
 func (w *worker) executeDistGlobalTask(reorgInfo *reorgInfo) error {
+=======
+func (w *worker) executeDistTask(stepCtx context.Context, t table.Table, reorgInfo *reorgInfo) error {
+>>>>>>> 575310677da (ddl: check context done in isReorgRunnable function (#57813))
 	if reorgInfo.mergingTmpIdx {
 		return errors.New("do not support merge index")
 	}
@@ -2142,8 +2150,13 @@ func (w *worker) executeDistGlobalTask(reorgInfo *reorgInfo) error {
 			if err != nil {
 				return err
 			}
+<<<<<<< HEAD
 			err = handle.WaitGlobalTask(ctx, task.ID)
 			if err := w.isReorgRunnable(reorgInfo.Job.ID, true); err != nil {
+=======
+			err = handle.WaitTaskDoneOrPaused(ctx, task.ID)
+			if err := w.isReorgRunnable(stepCtx, true); err != nil {
+>>>>>>> 575310677da (ddl: check context done in isReorgRunnable function (#57813))
 				if dbterror.ErrPausedDDLJob.Equal(err) {
 					logutil.BgLogger().Warn("job paused by user", zap.String("category", "ddl"), zap.Error(err))
 					return dbterror.ErrPausedDDLJob.GenWithStackByArgs(reorgInfo.Job.ID)
@@ -2167,11 +2180,17 @@ func (w *worker) executeDistGlobalTask(reorgInfo *reorgInfo) error {
 
 		g.Go(func() error {
 			defer close(done)
+<<<<<<< HEAD
 			err := handle.SubmitAndRunGlobalTask(ctx, taskKey, taskType, distPhysicalTableConcurrency, metaData)
 			failpoint.Inject("pauseAfterDistTaskFinished", func() {
 				MockDMLExecutionOnTaskFinished()
 			})
 			if err := w.isReorgRunnable(reorgInfo.Job.ID, true); err != nil {
+=======
+			err := submitAndWaitTask(ctx, taskKey, taskType, concurrency, reorgInfo.ReorgMeta.TargetScope, metaData)
+			failpoint.InjectCall("pauseAfterDistTaskFinished")
+			if err := w.isReorgRunnable(stepCtx, true); err != nil {
+>>>>>>> 575310677da (ddl: check context done in isReorgRunnable function (#57813))
 				if dbterror.ErrPausedDDLJob.Equal(err) {
 					logutil.BgLogger().Warn("job paused by user", zap.String("category", "ddl"), zap.Error(err))
 					return dbterror.ErrPausedDDLJob.GenWithStackByArgs(reorgInfo.Job.ID)
@@ -2192,7 +2211,7 @@ func (w *worker) executeDistGlobalTask(reorgInfo *reorgInfo) error {
 				w.updateJobRowCount(taskKey, reorgInfo.Job.ID)
 				return nil
 			case <-checkFinishTk.C:
-				if err = w.isReorgRunnable(reorgInfo.Job.ID, true); err != nil {
+				if err = w.isReorgRunnable(stepCtx, true); err != nil {
 					if dbterror.ErrPausedDDLJob.Equal(err) {
 						if err = handle.PauseTask(w.ctx, taskKey); err != nil {
 							logutil.BgLogger().Error("pause global task error", zap.String("category", "ddl"), zap.String("task_key", taskKey), zap.Error(err))