ddl: fix runnable ingest job checking

pkg/ddl/ddl.go

@@ -831,6 +831,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.delRangeMgr = d.newDeleteRangeManager(ctxPool == nil)
@@ -869,13 +870,6 @@ func (d *ddl) Start(ctxPool *pools.ResourcePool) error {
 	defer d.sessPool.Put(ctx)
 
 	ingest.InitGlobalLightningEnv()
-	d.ownerManager.SetRetireOwnerHook(func() {
-		// Since this instance is not DDL owner anymore, we clean up the processing job info.
-		if ingest.LitBackCtxMgr != nil {
-			ingest.LitBackCtxMgr.MarkJobFinish()
-		}
-		d.runningJobs.clear()
-	})
 
 	return nil
 }

pkg/ddl/ddl_running_jobs.go

@@ -22,8 +22,11 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/pingcap/tidb/pkg/parser/model"
+	"github.com/pingcap/tidb/pkg/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 {
@@ -47,6 +55,8 @@ 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{})
 }
@@ -56,6 +66,9 @@ func (j *runningJobs) add(job *model.Job) {
 	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.
@@ -75,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)
@@ -115,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
@@ -136,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 &&
+		job.ReorgMeta.IsFastReorg
+}

pkg/ddl/ddl_worker.go

@@ -26,7 +26,6 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/kvproto/pkg/kvrpcpb"
-	"github.com/pingcap/tidb/pkg/ddl/ingest"
 	sess "github.com/pingcap/tidb/pkg/ddl/internal/session"
 	"github.com/pingcap/tidb/pkg/ddl/util"
 	"github.com/pingcap/tidb/pkg/kv"
@@ -693,7 +692,6 @@ func (w *worker) finishDDLJob(t *meta.Meta, job *model.Job) (err error) {
 	startTime := time.Now()
 	defer func() {
 		metrics.DDLWorkerHistogram.WithLabelValues(metrics.WorkerFinishDDLJob, job.Type.String(), metrics.RetLabel(err)).Observe(time.Since(startTime).Seconds())
-		markJobFinish(job)
 	}()
 
 	if JobNeedGC(job) {
@@ -743,15 +741,6 @@ func (w *worker) finishDDLJob(t *meta.Meta, job *model.Job) (err error) {
 	return errors.Trace(err)
 }
 
-func markJobFinish(job *model.Job) {
-	if (job.Type == model.ActionAddIndex || job.Type == model.ActionAddPrimaryKey) &&
-		job.ReorgMeta != nil &&
-		job.ReorgMeta.IsFastReorg &&
-		ingest.LitBackCtxMgr != nil {
-		ingest.LitBackCtxMgr.MarkJobFinish()
-	}
-}
-
 func (w *worker) writeDDLSeqNum(job *model.Job) {
 	w.ddlSeqNumMu.Lock()
 	w.ddlSeqNumMu.seqNum++

pkg/ddl/ingest/backend_mgr.go

@@ -19,16 +19,17 @@ import (
 	"fmt"
 	"math"
 	"strconv"
-	"sync"
 	"time"
 
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/lightning/backend/local"
 	"github.com/pingcap/tidb/pkg/lightning/config"
 	"github.com/pingcap/tidb/pkg/util/generic"
 	"github.com/pingcap/tidb/pkg/util/logutil"
 	kvutil "github.com/tikv/client-go/v2/util"
 	pd "github.com/tikv/pd/client"
 	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.uber.org/atomic"
 	"go.uber.org/zap"
 )
 
@@ -48,18 +49,12 @@ type BackendCtxMgr interface {
 	) (BackendCtx, error)
 	Unregister(jobID int64)
 	Load(jobID int64) (BackendCtx, bool)
-
-	MarkJobProcessing(jobID int64) (ok bool)
-	MarkJobFinish()
 }
 
 type litBackendCtxMgr struct {
 	generic.SyncMap[int64, *litBackendCtx]
-	memRoot         MemRoot
-	diskRoot        DiskRoot
-	processingJobID int64
-	lastLoggingTime time.Time
-	mu              sync.Mutex
+	memRoot  MemRoot
+	diskRoot DiskRoot
 }
 
 func newLitBackendCtxMgr(path string, memQuota uint64) BackendCtxMgr {
@@ -80,30 +75,6 @@ func newLitBackendCtxMgr(path string, memQuota uint64) BackendCtxMgr {
 	return mgr
 }
 
-// MarkJobProcessing marks ingest backfill is processing.
-func (m *litBackendCtxMgr) MarkJobProcessing(jobID int64) bool {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if m.processingJobID == 0 || m.processingJobID == jobID {
-		m.processingJobID = jobID
-		return true
-	}
-	if time.Since(m.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", m.processingJobID))
-		m.lastLoggingTime = time.Now()
-	}
-	return false
-}
-
-// MarkJobFinish marks ingest backfill is finished.
-func (m *litBackendCtxMgr) MarkJobFinish() {
-	m.mu.Lock()
-	m.processingJobID = 0
-	m.mu.Unlock()
-}
-
 // CheckAvailable checks if the ingest backfill is available.
 func (m *litBackendCtxMgr) CheckAvailable() (bool, error) {
 	if err := m.diskRoot.PreCheckUsage(); err != nil {
@@ -113,6 +84,8 @@ func (m *litBackendCtxMgr) CheckAvailable() (bool, error) {
 	return true, nil
 }
 
+var ResignOwnerForTest = atomic.NewBool(false)
+
 // Register creates a new backend and registers it to the backend context.
 func (m *litBackendCtxMgr) Register(
 	ctx context.Context,
@@ -137,6 +110,9 @@ func (m *litBackendCtxMgr) Register(
 		logutil.Logger(ctx).Warn(LitWarnConfigError, zap.Int64("job ID", jobID), zap.Error(err))
 		return nil, err
 	}
+	failpoint.Inject("beforeCreateLocalBackend", func() {
+		ResignOwnerForTest.Store(true)
+	})
 	bd, err := createLocalBackend(ctx, cfg, pdSvcDiscovery)
 	if err != nil {
 		logutil.Logger(ctx).Error(LitErrCreateBackendFail, zap.Int64("job ID", jobID), zap.Error(err))

pkg/ddl/ingest/mock.go

@@ -43,15 +43,6 @@ func NewMockBackendCtxMgr(sessCtxProvider func() sessionctx.Context) *MockBacken
 	}
 }
 
-// MarkJobProcessing implements BackendCtxMgr.MarkJobProcessing interface.
-func (*MockBackendCtxMgr) MarkJobProcessing(_ int64) bool {
-	return true
-}
-
-// MarkJobFinish implements BackendCtxMgr.MarkJobFinish interface.
-func (*MockBackendCtxMgr) MarkJobFinish() {
-}
-
 // CheckAvailable implements BackendCtxMgr.Available interface.
 func (m *MockBackendCtxMgr) CheckAvailable() (bool, error) {
 	return len(m.runningJobs) == 0, nil

pkg/ddl/job_table.go

@@ -228,17 +228,6 @@ func (d *ddl) getReorgJob(sess *sess.Session) (*model.Job, error) {
 		if !d.runningJobs.checkRunnable(job) {
 			return false, nil
 		}
-		if (job.Type == model.ActionAddIndex || job.Type == model.ActionAddPrimaryKey) &&
-			job.State == model.JobStateQueueing &&
-			job.ReorgMeta != nil &&
-			job.ReorgMeta.IsFastReorg &&
-			ingest.LitBackCtxMgr != nil {
-			succeed := ingest.LitBackCtxMgr.MarkJobProcessing(job.ID)
-			if !succeed {
-				// We only allow one task to use ingest at the same time in order to limit the CPU/memory usage.
-				return false, nil
-			}
-		}
 		// Check if there is any block ddl running, like drop schema and flashback cluster.
 		sql := fmt.Sprintf("select job_id from mysql.tidb_ddl_job where "+
 			"(CONCAT(',', schema_ids, ',') REGEXP CONCAT(',', %s, ',') != 0 and type = %d and processing) "+
@@ -292,6 +281,16 @@ func (d *ddl) startDispatchLoop() {
 			time.Sleep(dispatchLoopWaitingDuration)
 			continue
 		}
+		failpoint.Inject("ownerResignAfterDispatchLoopCheck", func() {
+			if ingest.ResignOwnerForTest.Load() {
+				err2 := d.ownerManager.ResignOwner(context.Background())
+				if err2 != nil {
+					logutil.BgLogger().Info("resign meet error", zap.Error(err2))
+				}
+				time.Sleep(500 * time.Millisecond)
+				ingest.ResignOwnerForTest.Store(false)
+			}
+		})
 		select {
 		case <-d.ddlJobCh:
 		case <-ticker.C:

pkg/lightning/backend/local/engine_mgr.go

@@ -19,9 +19,12 @@ import (
 	"math"
 	"os"
 	"path/filepath"
+	"strings"
 	"sync"
+	"time"
 
 	"github.com/cockroachdb/pebble"
+	"github.com/cockroachdb/pebble/vfs"
 	"github.com/docker/go-units"
 	"github.com/google/uuid"
 	"github.com/pingcap/errors"
@@ -581,6 +584,19 @@ func (em *engineManager) getBufferPool() *membuf.Pool {
 	return em.bufferPool
 }
 
+// only used in tests
+type slowCreateFS struct {
+	vfs.FS
+}
+
+func (s slowCreateFS) Create(name string) (vfs.File, error) {
+	if strings.Contains(name, "temporary") {
+		// print stack
+		time.Sleep(time.Second)
+	}
+	return s.FS.Create(name)
+}
+
 func openDuplicateDB(storeDir string) (*pebble.DB, error) {
 	dbPath := filepath.Join(storeDir, duplicateDBName)
 	// TODO: Optimize the opts for better write.
@@ -589,6 +605,9 @@ func openDuplicateDB(storeDir string) (*pebble.DB, error) {
 			newRangePropertiesCollector,
 		},
 	}
+	failpoint.Inject("slowCreateFS", func() {
+		opts.FS = slowCreateFS{vfs.Default}
+	})
 	return pebble.Open(dbPath, opts)
 }
 

pkg/owner/manager.go

@@ -65,8 +65,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 (
@@ -118,8 +116,7 @@ type ownerManager struct {
 	wg             sync.WaitGroup
 	campaignCancel context.CancelFunc
 
-	beOwnerHook     func()
-	retireOwnerHook func()
+	beOwnerHook func()
 }
 
 // NewOwnerManager creates a new Manager.
@@ -164,10 +161,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
 
@@ -230,9 +223,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)
 }
 

pkg/owner/mock.go

@@ -174,11 +174,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() {
 	m.campaignDone <- struct{}{}