ddl: fix runnable ingest job checking (#52503) (#59761)

close #52475

Author: ti-chi-bot

ddl/ddl.go

@@ -740,6 +740,7 @@ func (d *ddl) Start(ctxPool *pools.ResourcePool) error {
 		if err != nil {
 			logutil.BgLogger().Error("error when getting the ddl history count", zap.Error(err))
 		}
+		d.runningJobs.clear()
 		d.ddlCtx.setOwnerTS(time.Now().Unix())
 	})
 
@@ -769,9 +770,6 @@ func (d *ddl) Start(ctxPool *pools.ResourcePool) error {
 	d.wg.Run(d.PollTiFlashRoutine)
 
 	ingest.InitGlobalLightningEnv()
-	d.ownerManager.SetRetireOwnerHook(func() {
-		d.runningJobs = newRunningJobs()
-	})
 
 	return nil
 }

ddl/ddl_running_jobs.go

@@ -22,8 +22,11 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/pingcap/tidb/parser/model"
+	"github.com/pingcap/tidb/util/logutil"
+	"go.uber.org/zap"
 )
 
 type runningJobs struct {
@@ -36,6 +39,11 @@ type runningJobs struct {
 	// It is not necessarily being processed by a worker.
 	unfinishedIDs    map[int64]struct{}
 	unfinishedSchema map[string]map[string]struct{} // database -> table -> struct{}
+
+	// processingReorgJobID records the ID of the ingest job that is being processed by a worker.
+	// TODO(tangenta): remove this when we support running multiple concurrent ingest jobs.
+	processingIngestJobID int64
+	lastLoggingTime       time.Time
 }
 
 func newRunningJobs() *runningJobs {
@@ -46,11 +54,21 @@ func newRunningJobs() *runningJobs {
 	}
 }
 
+func (j *runningJobs) clear() {
+	j.Lock()
+	defer j.Unlock()
+	j.unfinishedIDs = make(map[int64]struct{})
+	j.unfinishedSchema = make(map[string]map[string]struct{})
+}
+
 func (j *runningJobs) add(job *model.Job) {
 	j.Lock()
 	defer j.Unlock()
 	j.processingIDs[job.ID] = struct{}{}
 	j.updateInternalRunningJobIDs()
+	if isIngestJob(job) {
+		j.processingIngestJobID = job.ID
+	}
 
 	if _, ok := j.unfinishedIDs[job.ID]; ok {
 		// Already exists, no need to add it again.
@@ -70,6 +88,9 @@ func (j *runningJobs) remove(job *model.Job) {
 	defer j.Unlock()
 	delete(j.processingIDs, job.ID)
 	j.updateInternalRunningJobIDs()
+	if isIngestJob(job) && job.ID == j.processingIngestJobID {
+		j.processingIngestJobID = 0
+	}
 
 	if job.IsFinished() || job.IsSynced() {
 		delete(j.unfinishedIDs, job.ID)
@@ -110,6 +131,16 @@ func (j *runningJobs) checkRunnable(job *model.Job) bool {
 		// Already processing by a worker. Skip running it again.
 		return false
 	}
+	if isIngestJob(job) && j.processingIngestJobID != 0 {
+		// We only allow one task to use ingest at the same time in order to limit the CPU/memory usage.
+		if time.Since(j.lastLoggingTime) > 1*time.Minute {
+			logutil.BgLogger().Info("ingest backfill worker is already in used by another DDL job",
+				zap.String("category", "ddl-ingest"),
+				zap.Int64("processing job ID", j.processingIngestJobID))
+			j.lastLoggingTime = time.Now()
+		}
+		return false
+	}
 	for _, info := range job.GetInvolvingSchemaInfo() {
 		if _, ok := j.unfinishedSchema[model.InvolvingAll]; ok {
 			return false
@@ -131,3 +162,9 @@ func (j *runningJobs) checkRunnable(job *model.Job) bool {
 	}
 	return true
 }
+
+func isIngestJob(job *model.Job) bool {
+	return (job.Type == model.ActionAddIndex || job.Type == model.ActionAddPrimaryKey) &&
+		job.ReorgMeta != nil &&
+		IsEnableFastReorg()
+}

owner/manager.go

@@ -59,8 +59,6 @@ type Manager interface {
 
 	// SetBeOwnerHook sets a hook. The hook is called before becoming an owner.
 	SetBeOwnerHook(hook func())
-	// SetRetireOwnerHook will be called after retiring the owner.
-	SetRetireOwnerHook(hook func())
 }
 
 const (
@@ -87,8 +85,7 @@ type ownerManager struct {
 	wg             sync.WaitGroup
 	campaignCancel context.CancelFunc
 
-	beOwnerHook     func()
-	retireOwnerHook func()
+	beOwnerHook func()
 }
 
 // NewOwnerManager creates a new Manager.
@@ -132,10 +129,6 @@ func (m *ownerManager) SetBeOwnerHook(hook func()) {
 	m.beOwnerHook = hook
 }
 
-func (m *ownerManager) SetRetireOwnerHook(hook func()) {
-	m.retireOwnerHook = hook
-}
-
 // ManagerSessionTTL is the etcd session's TTL in seconds. It's exported for testing.
 var ManagerSessionTTL = 60
 
@@ -195,9 +188,6 @@ func (m *ownerManager) toBeOwner(elec *concurrency.Election) {
 
 // RetireOwner make the manager to be a not owner.
 func (m *ownerManager) RetireOwner() {
-	if m.retireOwnerHook != nil {
-		m.retireOwnerHook()
-	}
 	atomic.StorePointer(&m.elec, nil)
 }
 

owner/mock.go

@@ -102,11 +102,6 @@ func (m *mockManager) SetBeOwnerHook(hook func()) {
 	m.beOwnerHook = hook
 }
 
-// SetRetireOwnerHook implements Manager.SetRetireOwnerHook interface.
-func (m *mockManager) SetRetireOwnerHook(hook func()) {
-	m.retireHook = hook
-}
-
 // CampaignCancel implements Manager.CampaignCancel interface
 func (m *mockManager) CampaignCancel() {
 	// do nothing