start adjust go routine after pipeline execute

pkg/ddl/backfilling.go

@@ -777,47 +777,7 @@ func (dc *ddlCtx) runAddIndexInLocalIngestMode(
 		return err
 	}
 
-	// Adjust worker pool size dynamically.
-	go func() {
-		opR, opW := pipe.GetLocalIngestModeReaderAndWriter()
-		if opR == nil || opW == nil {
-			logutil.DDLIngestLogger().Error("failed to get local ingest mode reader or writer", zap.Int64("jobID", job.ID))
-			return
-		}
-		reader, readerOk := opR.(*TableScanOperator)
-		writer, writerOk := opW.(*IndexIngestOperator)
-		if !readerOk || !writerOk {
-			logutil.DDLIngestLogger().Error(
-				"unexpected operator types, config can't be adjusted",
-				zap.Int64("jobID", job.ID),
-				zap.Bool("isReaderValid", readerOk),
-				zap.Bool("isWriterValid", writerOk),
-			)
-			return
-		}
-		ticker := time.NewTicker(UpdateDDLJobReorgCfgInterval)
-		defer ticker.Stop()
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-ticker.C:
-				targetReaderCnt, targetWriterCnt := expectedIngestWorkerCnt(job.ReorgMeta.GetConcurrency(), avgRowSize)
-				currentReaderCnt, currentWriterCnt := reader.GetWorkerPoolSize(), writer.GetWorkerPoolSize()
-				if int32(targetReaderCnt) == currentReaderCnt && int32(targetWriterCnt) == currentWriterCnt {
-					continue
-				}
-				reader.TuneWorkerPoolSize(int32(targetReaderCnt))
-				writer.TuneWorkerPoolSize(int32(targetWriterCnt))
-				logutil.DDLIngestLogger().Info("adjust ddl job config success",
-					zap.Int64("jobID", job.ID),
-					zap.Int32("table scan operator count", reader.GetWorkerPoolSize()),
-					zap.Int32("index ingest operator count", writer.GetWorkerPoolSize()))
-			}
-		}
-	}()
-
-	err = executeAndClosePipeline(opCtx, pipe)
+	err = executeAndClosePipeline(opCtx, pipe, reorgInfo.Job, avgRowSize)
 	if err != nil {
 		err1 := bcCtx.FinishAndUnregisterEngines(ingest.OptCloseEngines)
 		if err1 != nil {
@@ -834,11 +794,58 @@ func (dc *ddlCtx) runAddIndexInLocalIngestMode(
 	return bcCtx.FinishAndUnregisterEngines(ingest.OptCleanData | ingest.OptCheckDup)
 }
 
-func executeAndClosePipeline(ctx *OperatorCtx, pipe *operator.AsyncPipeline) error {
+func adjustWorkerPoolSize(ctx context.Context, pipe *operator.AsyncPipeline, job *model.Job, avgRowSize int) {
+	opR, opW := pipe.GetLocalIngestModeReaderAndWriter()
+	if opR == nil || opW == nil {
+		logutil.DDLIngestLogger().Error("failed to get local ingest mode reader or writer", zap.Int64("jobID", job.ID))
+		return
+	}
+	reader, readerOk := opR.(*TableScanOperator)
+	writer, writerOk := opW.(*IndexIngestOperator)
+	if !readerOk || !writerOk {
+		logutil.DDLIngestLogger().Error(
+			"unexpected operator types, config can't be adjusted",
+			zap.Int64("jobID", job.ID),
+			zap.Bool("isReaderValid", readerOk),
+			zap.Bool("isWriterValid", writerOk),
+		)
+		return
+	}
+	ticker := time.NewTicker(UpdateDDLJobReorgCfgInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			targetReaderCnt, targetWriterCnt := expectedIngestWorkerCnt(job.ReorgMeta.GetConcurrency(), avgRowSize)
+			currentReaderCnt, currentWriterCnt := reader.GetWorkerPoolSize(), writer.GetWorkerPoolSize()
+			if int32(targetReaderCnt) == currentReaderCnt && int32(targetWriterCnt) == currentWriterCnt {
+				continue
+			}
+			reader.TuneWorkerPoolSize(int32(targetReaderCnt))
+			writer.TuneWorkerPoolSize(int32(targetWriterCnt))
+			logutil.DDLIngestLogger().Info("adjust ddl job config success",
+				zap.Int64("jobID", job.ID),
+				zap.Int32("table scan operator count", reader.GetWorkerPoolSize()),
+				zap.Int32("index ingest operator count", writer.GetWorkerPoolSize()))
+		}
+	}
+}
+
+func executeAndClosePipeline(ctx *OperatorCtx, pipe *operator.AsyncPipeline, job *model.Job, avgRowSize int) error {
 	err := pipe.Execute()
 	if err != nil {
 		return err
 	}
+
+	// Adjust worker pool size dynamically.
+	if job != nil && avgRowSize != 0 {
+		go func() {
+			adjustWorkerPoolSize(ctx, pipe, job, avgRowSize)
+		}()
+	}
+
 	err = pipe.Close()
 	if opErr := ctx.OperatorErr(); opErr != nil {
 		return opErr

pkg/ddl/backfilling_read_index.go

@@ -114,14 +114,14 @@ func (r *readIndexExecutor) RunSubtask(ctx context.Context, subtask *proto.Subta
 		if err != nil {
 			return err
 		}
-		return executeAndClosePipeline(opCtx, pipe)
+		return executeAndClosePipeline(opCtx, pipe, nil, 0)
 	}
 
 	pipe, err := r.buildLocalStorePipeline(opCtx, sm, subtask.Concurrency)
 	if err != nil {
 		return err
 	}
-	err = executeAndClosePipeline(opCtx, pipe)
+	err = executeAndClosePipeline(opCtx, pipe, nil, 0)
 	if err != nil {
 		// For dist task local based ingest, checkpoint is unsupported.
 		// If there is an error we should keep local sort dir clean.