ddl: query since min job id to mitigate slowness due to un-compacted deleted kv (#54122)

ref #53246

Author: D3Hunter

Makefile

@@ -496,6 +496,7 @@ gen_mock: mockgen
 	tools/bin/mockgen -package mock github.com/pingcap/tidb/pkg/util/sqlexec RestrictedSQLExecutor > pkg/util/sqlexec/mock/restricted_sql_executor_mock.go
 	tools/bin/mockgen -package mockstorage github.com/pingcap/tidb/br/pkg/storage ExternalStorage > br/pkg/mock/storage/storage.go
 	tools/bin/mockgen -package mock github.com/pingcap/tidb/pkg/ddl SchemaLoader > pkg/ddl/mock/schema_loader_mock.go
+	tools/bin/mockgen -package mock github.com/pingcap/tidb/pkg/ddl/systable Manager > pkg/ddl/mock/systable_manager_mock.go
 
 # There is no FreeBSD environment for GitHub actions. So cross-compile on Linux
 # but that doesn't work with CGO_ENABLED=1, so disable cgo. The reason to have

pkg/ddl/job_scheduler_test.go

@@ -61,3 +61,39 @@ func TestMustReloadSchemas(t *testing.T) {
 	sch.mustReloadSchemas()
 	require.True(t, ctrl.Satisfied())
 }
+
+func TestRefreshMinJobID(t *testing.T) {
+	reduceIntervals(t)
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	mgr := mock.NewMockManager(ctrl)
+
+	sch := &jobScheduler{
+		schCtx:    context.Background(),
+		sysTblMgr: mgr,
+	}
+	// success
+	start := time.Now()
+	mgr.EXPECT().GetMinJobID(gomock.Any(), int64(0)).Return(int64(1), nil)
+	sch.refreshMinJobID()
+	require.EqualValues(t, 1, sch.currMinJobID)
+	require.GreaterOrEqual(t, sch.lastRefreshMinIDTime, start)
+	require.True(t, ctrl.Satisfied())
+	// not refresh too fast
+	schedulerLoopRetryInterval = time.Hour
+	sch.refreshMinJobID()
+	require.True(t, ctrl.Satisfied())
+	// ignore refresh error
+	sch.lastRefreshMinIDTime = time.Time{}
+	mgr.EXPECT().GetMinJobID(gomock.Any(), int64(1)).Return(int64(0), errors.New("mock err"))
+	sch.refreshMinJobID()
+	require.EqualValues(t, 1, sch.currMinJobID)
+	require.True(t, ctrl.Satisfied())
+	// success again
+	sch.lastRefreshMinIDTime = time.Time{}
+	mgr.EXPECT().GetMinJobID(gomock.Any(), int64(1)).Return(int64(100), nil)
+	sch.refreshMinJobID()
+	require.EqualValues(t, 100, sch.currMinJobID)
+	require.GreaterOrEqual(t, sch.lastRefreshMinIDTime, start)
+	require.True(t, ctrl.Satisfied())
+}

pkg/ddl/job_table.go

@@ -128,6 +128,10 @@ type jobScheduler struct {
 	runningJobs  *runningJobs
 	sysTblMgr    systable.Manager
 	schemaLoader SchemaLoader
+	// currMinJobID is the minimal job ID in tidb_ddl_job table, we use it to mitigate
+	// this issue https://github.com/pingcap/tidb/issues/52905
+	currMinJobID         int64
+	lastRefreshMinIDTime time.Time
 
 	// those fields are created on start
 	reorgWorkerPool      *workerPool
@@ -191,14 +195,15 @@ func (s *jobScheduler) getJob(se *sess.Session, tp jobType) (*model.Job, error)
 		not = ""
 		label = "get_job_reorg"
 	}
+	// TODO replace this sub-query with memory implementation.
 	const getJobSQL = `select job_meta, processing from mysql.tidb_ddl_job where job_id in
-		(select min(job_id) from mysql.tidb_ddl_job group by schema_ids, table_ids, processing)
+		(select min(job_id) from mysql.tidb_ddl_job where job_id >= %d group by schema_ids, table_ids, processing)
 		and %s reorg %s order by processing desc, job_id`
 	var excludedJobIDs string
 	if ids := s.runningJobs.allIDs(); len(ids) > 0 {
 		excludedJobIDs = fmt.Sprintf("and job_id not in (%s)", ids)
 	}
-	sql := fmt.Sprintf(getJobSQL, not, excludedJobIDs)
+	sql := fmt.Sprintf(getJobSQL, s.currMinJobID, not, excludedJobIDs)
 	rows, err := se.Execute(context.Background(), sql, label)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -387,6 +392,7 @@ func (s *jobScheduler) startDispatch() error {
 		if err := s.checkAndUpdateClusterState(false); err != nil {
 			continue
 		}
+		s.refreshMinJobID()
 		failpoint.InjectCall("beforeAllLoadDDLJobAndRun")
 		s.loadDDLJobAndRun(se, s.generalDDLWorkerPool, jobTypeGeneral)
 		s.loadDDLJobAndRun(se, s.reorgWorkerPool, jobTypeReorg)
@@ -645,6 +651,21 @@ func (*jobScheduler) markJobProcessing(se *sess.Session, job *model.Job) error {
 	return errors.Trace(err)
 }
 
+func (s *jobScheduler) refreshMinJobID() {
+	now := time.Now()
+	if now.Sub(s.lastRefreshMinIDTime) < dispatchLoopWaitingDuration {
+		return
+	}
+	s.lastRefreshMinIDTime = now
+	minID, err := s.sysTblMgr.GetMinJobID(s.schCtx, s.currMinJobID)
+	if err != nil {
+		logutil.DDLLogger().Info("get min job ID failed", zap.Error(err))
+		return
+	}
+	// use max, in case all job are finished to avoid the currMinJobID go back.
+	s.currMinJobID = max(s.currMinJobID, minID)
+}
+
 func (d *ddl) getTableByTxn(r autoid.Requirement, schemaID, tableID int64) (*model.DBInfo, table.Table, error) {
 	var tbl table.Table
 	var dbInfo *model.DBInfo

pkg/ddl/mock/BUILD.bazel

@@ -2,8 +2,14 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library")
 
 go_library(
     name = "mock",
-    srcs = ["schema_loader_mock.go"],
+    srcs = [
+        "schema_loader_mock.go",
+        "systable_manager_mock.go",
+    ],
     importpath = "github.com/pingcap/tidb/pkg/ddl/mock",
     visibility = ["//visibility:public"],
-    deps = ["@org_uber_go_mock//gomock"],
+    deps = [
+        "//pkg/parser/model",
+        "@org_uber_go_mock//gomock",
+    ],
 )

pkg/ddl/mock/systable_manager_mock.go

@@ -38,6 +38,10 @@ type Manager interface {
 	GetJobByID(ctx context.Context, jobID int64) (*model.Job, error)
 	// GetMDLVer gets the MDL version by job ID, returns ErrNotFound if the MDL info does not exist.
 	GetMDLVer(ctx context.Context, jobID int64) (int64, error)
+	// GetMinJobID gets current minimum job ID in the job table for job_id >= prevMinJobID,
+	// if no jobs, returns 0. prevMinJobID is used to avoid full table scan, see
+	// https://github.com/pingcap/tidb/issues/52905
+	GetMinJobID(ctx context.Context, prevMinJobID int64) (int64, error)
 }
 
 type manager struct {
@@ -105,3 +109,22 @@ func (mgr *manager) GetMDLVer(ctx context.Context, jobID int64) (int64, error) {
 	}
 	return ver, nil
 }
+
+func (mgr *manager) GetMinJobID(ctx context.Context, prevMinJobID int64) (int64, error) {
+	var minID int64
+	if err := mgr.withNewSession(func(se *session.Session) error {
+		sql := fmt.Sprintf(`select min(job_id) from mysql.tidb_ddl_job where job_id >= %d`, prevMinJobID)
+		rows, err := se.Execute(ctx, sql, "get-min-job-id")
+		if err != nil {
+			return errors.Trace(err)
+		}
+		if len(rows) == 0 {
+			return nil
+		}
+		minID = rows[0].GetInt64(0)
+		return nil
+	}); err != nil {
+		return 0, err
+	}
+	return minID, nil
+}

pkg/ddl/systable/manager.go

@@ -64,4 +64,22 @@ func TestManager(t *testing.T) {
 		require.NoError(t, err)
 		require.EqualValues(t, 123, ver)
 	})
+
+	t.Run("GetMinJobID", func(t *testing.T) {
+		tk.MustExec("delete from mysql.tidb_ddl_job")
+		id, err := mgr.GetMinJobID(ctx, 0)
+		require.NoError(t, err)
+		require.EqualValues(t, 0, id)
+		tk.MustExec(`insert into mysql.tidb_ddl_job(job_id, reorg, schema_ids, table_ids, job_meta, type, processing)
+					values(123456, 0, '1', '1', '{"id":9998}', 1, 0)`)
+		id, err = mgr.GetMinJobID(ctx, 0)
+		require.NoError(t, err)
+		require.EqualValues(t, 123456, id)
+		id, err = mgr.GetMinJobID(ctx, 123456)
+		require.NoError(t, err)
+		require.EqualValues(t, 123456, id)
+		id, err = mgr.GetMinJobID(ctx, 123457)
+		require.NoError(t, err)
+		require.EqualValues(t, 0, id)
+	})
 }