disttask/ddl: fix pause succeed task then run a new task

pkg/ddl/index.go

@@ -2051,6 +2051,9 @@ var MockDMLExecutionOnTaskFinished func()
 // MockDMLExecutionOnDDLPaused is used to mock DML execution when ddl job paused.
 var MockDMLExecutionOnDDLPaused func()
 
+// TestSyncChan is used to sync the test.
+var TestSyncChan = make(chan struct{})
+
 func (w *worker) executeDistGlobalTask(reorgInfo *reorgInfo) error {
 	if reorgInfo.mergingTmpIdx {
 		return errors.New("do not support merge index")
@@ -2066,19 +2069,27 @@ func (w *worker) executeDistGlobalTask(reorgInfo *reorgInfo) error {
 		taskKey = fmt.Sprintf("%s/%d", taskKey, mInfo.Seq)
 	}
 
-	// for resuming add index task.
+	// For resuming add index task.
+	// Need to fetch global task by taskKey in tidb_global_task and tidb_global_task_history tables.
+	// When pausing the related ddl job, it is possible that the global task with taskKey is succeed and in tidb_global_task_history.
+	// As a result, when resuming the related ddl job,
+	// it is necessary to check task exits in tidb_global_task and tidb_global_task_history tables.
 	taskManager, err := storage.GetTaskManager()
 	if err != nil {
 		return err
 	}
-	task, err := taskManager.GetGlobalTaskByKey(taskKey)
+	task, err := taskManager.GetGlobalTaskByKeyWithHistory(taskKey)
 	if err != nil {
 		return err
 	}
 	if task != nil {
 		// It's possible that the task state is succeed but the ddl job is paused.
 		// When task in succeed state, we can skip the dist task execution/scheduing process.
 		if task.State == proto.TaskStateSucceed {
+			logutil.BgLogger().Info(
+				"global task succeed, start to resume the ddl job",
+				zap.String("category", "ddl"),
+				zap.String("task-key", taskKey))
 			return nil
 		}
 		g.Go(func() error {
@@ -2123,10 +2134,9 @@ func (w *worker) executeDistGlobalTask(reorgInfo *reorgInfo) error {
 		g.Go(func() error {
 			defer close(done)
 			err := handle.SubmitAndRunGlobalTask(ctx, taskKey, taskType, distPhysicalTableConcurrency, metaData)
-			failpoint.Inject("pauseAfterDistTaskSuccess", func() {
+			failpoint.Inject("pauseAfterDistTaskFinished", func() {
 				MockDMLExecutionOnTaskFinished()
 			})
-
 			if err := w.isReorgRunnable(reorgInfo.Job.ID, true); err != nil {
 				if dbterror.ErrPausedDDLJob.Equal(err) {
 					logutil.BgLogger().Warn("job paused by user", zap.String("category", "ddl"), zap.Error(err))
@@ -2154,6 +2164,10 @@ func (w *worker) executeDistGlobalTask(reorgInfo *reorgInfo) error {
 							logutil.BgLogger().Error("pause global task error", zap.String("category", "ddl"), zap.String("task_key", taskKey), zap.Error(err))
 							continue
 						}
+						failpoint.Inject("syncDDLTaskPause", func() {
+							// make sure the task is paused.
+							TestSyncChan <- struct{}{}
+						})
 					}
 					if !dbterror.ErrCancelledDDLJob.Equal(err) {
 						return errors.Trace(err)

tests/realtikvtest/addindextest/add_index_test.go

@@ -236,6 +236,7 @@ func TestAddIndexDistPauseAndResume(t *testing.T) {
 		require.Equal(t, 1, len(row))
 		jobID := row[0][0].(string)
 		tk1.MustExec("admin pause ddl jobs " + jobID)
+		<-ddl.TestSyncChan
 	}
 
 	dispatcher.MockDMLExecutionOnPausedState = func(task *proto.Task) {
@@ -254,11 +255,10 @@ func TestAddIndexDistPauseAndResume(t *testing.T) {
 
 	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/ddl/mockDMLExecutionAddIndexSubTaskFinish", "3*return(true)"))
 	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/disttask/framework/dispatcher/mockDMLExecutionOnPausedState", "return(true)"))
+	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/ddl/syncDDLTaskPause", "return()"))
 	tk.MustExec(`set global tidb_enable_dist_task=1;`)
 	tk.MustExec("alter table t add index idx1(a);")
 	tk.MustExec("admin check table t;")
-	tk.MustExec("alter table t add index idx2(a);")
-	tk.MustExec("admin check table t;")
 	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/ddl/mockDMLExecutionAddIndexSubTaskFinish"))
 	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/disttask/framework/dispatcher/mockDMLExecutionOnPausedState"))
 
@@ -274,9 +274,11 @@ func TestAddIndexDistPauseAndResume(t *testing.T) {
 	}
 	hook.OnJobUpdatedExported.Store(&resumeFunc)
 	dom.DDL().SetHook(hook.Clone())
-	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/ddl/pauseAfterDistTaskSuccess", "1*return(true)"))
+	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/ddl/pauseAfterDistTaskFinished", "1*return(true)"))
 	tk.MustExec("alter table t add index idx3(a);")
 	tk.MustExec("admin check table t;")
-	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/ddl/pauseAfterDistTaskSuccess"))
+	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/ddl/pauseAfterDistTaskFinished"))
+	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/ddl/syncDDLTaskPause"))
+
 	tk.MustExec(`set global tidb_enable_dist_task=0;`)
 }