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

This reverts commit a33a868.

Author: D3Hunter

Makefile

@@ -496,7 +496,6 @@ 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,39 +61,3 @@ 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,10 +128,6 @@ 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
@@ -198,15 +194,14 @@ 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 where job_id >= %d group by schema_ids, table_ids, processing)
+		(select min(job_id) from mysql.tidb_ddl_job 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, s.currMinJobID, not, excludedJobIDs)
+	sql := fmt.Sprintf(getJobSQL, not, excludedJobIDs)
 	rows, err := se.Execute(context.Background(), sql, label)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -398,7 +393,6 @@ 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)
@@ -657,21 +651,6 @@ 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,14 +2,8 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library")
 
 go_library(
     name = "mock",
-    srcs = [
-        "schema_loader_mock.go",
-        "systable_manager_mock.go",
-    ],
+    srcs = ["schema_loader_mock.go"],
     importpath = "github.com/pingcap/tidb/pkg/ddl/mock",
     visibility = ["//visibility:public"],
-    deps = [
-        "//pkg/parser/model",
-        "@org_uber_go_mock//gomock",
-    ],
+    deps = ["@org_uber_go_mock//gomock"],
 )

pkg/ddl/mock/systable_manager_mock.go

@@ -38,10 +38,6 @@ 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 {
@@ -109,22 +105,3 @@ 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,22 +64,4 @@ 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)
-	})
 }