ttl: don't schedule ttl job when EnableTTLJob is off (#40336)

ttl/ttlworker/job_manager.go

@@ -353,6 +353,24 @@ func (m *JobManager) rescheduleJobs(se session.Session, now time.Time) {
 		// and keep the job in memory, it could start the left task in the next window.
 		return
 	}
+	if !variable.EnableTTLJob.Load() {
+		if len(m.runningJobs) > 0 {
+			ctx, cancel := context.WithTimeout(m.ctx, ttlInternalSQLTimeout)
+
+			for _, job := range m.runningJobs {
+				logutil.Logger(m.ctx).Info("cancel job because tidb_ttl_job_enable turned off", zap.String("jobID", job.id), zap.String("statistics", job.statistics.String()))
+				m.removeJob(job)
+				err := job.Cancel(ctx, se)
+				if err != nil {
+					logutil.Logger(m.ctx).Warn("fail to cancel job", zap.Error(err))
+				}
+				job.finish(se, se.Now())
+			}
+
+			cancel()
+		}
+		return
+	}
 
 	idleScanWorkers := m.idleScanWorkers()
 	if len(idleScanWorkers) == 0 {

ttl/ttlworker/job_manager_integration_test.go

@@ -17,6 +17,7 @@ package ttlworker_test
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"sync"
 	"testing"
 	"time"
@@ -123,3 +124,118 @@ func TestFinishJob(t *testing.T) {
 
 	tk.MustQuery("select table_id, last_job_summary from mysql.tidb_ttl_table_status").Check(testkit.Rows("2 {\"total_rows\":0,\"success_rows\":0,\"error_rows\":0,\"total_scan_task\":1,\"scheduled_scan_task\":0,\"finished_scan_task\":0,\"scan_task_err\":\"\\\"'an error message contains both single and double quote'\\\"\"}"))
 }
+<<<<<<< HEAD
+=======
+
+func TestTTLAutoAnalyze(t *testing.T) {
+	failpoint.Enable("github.com/pingcap/tidb/ttl/ttlworker/update-info-schema-cache-interval", fmt.Sprintf("return(%d)", time.Second))
+	defer failpoint.Disable("github.com/pingcap/tidb/ttl/ttlworker/update-info-schema-cache-interval")
+	failpoint.Enable("github.com/pingcap/tidb/ttl/ttlworker/update-status-table-cache-interval", fmt.Sprintf("return(%d)", time.Second))
+	defer failpoint.Disable("github.com/pingcap/tidb/ttl/ttlworker/update-status-table-cache-interval")
+	failpoint.Enable("github.com/pingcap/tidb/ttl/ttlworker/resize-workers-interval", fmt.Sprintf("return(%d)", time.Second))
+	defer failpoint.Disable("github.com/pingcap/tidb/ttl/ttlworker/resize-workers-interval")
+
+	originAutoAnalyzeMinCnt := handle.AutoAnalyzeMinCnt
+	handle.AutoAnalyzeMinCnt = 0
+	defer func() {
+		handle.AutoAnalyzeMinCnt = originAutoAnalyzeMinCnt
+	}()
+
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+
+	tk.MustExec("use test")
+	tk.MustExec("create table t (id int, created_at datetime) ttl = `created_at` + interval 1 day")
+
+	// insert ten rows, the 2,3,4,6,9,10 of them are expired
+	for i := 1; i <= 10; i++ {
+		t := time.Now()
+		if i%2 == 0 || i%3 == 0 {
+			t = t.Add(-time.Hour * 48)
+		}
+
+		tk.MustExec("insert into t values(?, ?)", i, t.Format(time.RFC3339))
+	}
+	// TODO: use a better way to pause and restart ttl worker after analyze the table to make it more stable
+	// but as the ttl worker takes several seconds to start, it's not too serious.
+	tk.MustExec("analyze table t")
+	rows := tk.MustQuery("show stats_meta").Rows()
+	require.Equal(t, rows[0][4], "0")
+	require.Equal(t, rows[0][5], "10")
+
+	retryTime := 15
+	retryInterval := time.Second * 2
+	deleted := false
+	for retryTime >= 0 {
+		retryTime--
+		time.Sleep(retryInterval)
+
+		rows := tk.MustQuery("select count(*) from t").Rows()
+		count := rows[0][0].(string)
+		if count == "3" {
+			deleted = true
+			break
+		}
+	}
+	require.True(t, deleted, "ttl should remove expired rows")
+
+	h := dom.StatsHandle()
+	is := dom.InfoSchema()
+	require.NoError(t, h.DumpStatsDeltaToKV(handle.DumpAll))
+	require.NoError(t, h.Update(is))
+	require.True(t, h.HandleAutoAnalyze(is))
+}
+
+func TestTTLJobDisable(t *testing.T) {
+	failpoint.Enable("github.com/pingcap/tidb/ttl/ttlworker/update-info-schema-cache-interval", fmt.Sprintf("return(%d)", time.Second))
+	defer failpoint.Disable("github.com/pingcap/tidb/ttl/ttlworker/update-info-schema-cache-interval")
+	failpoint.Enable("github.com/pingcap/tidb/ttl/ttlworker/update-status-table-cache-interval", fmt.Sprintf("return(%d)", time.Second))
+	defer failpoint.Disable("github.com/pingcap/tidb/ttl/ttlworker/update-status-table-cache-interval")
+	failpoint.Enable("github.com/pingcap/tidb/ttl/ttlworker/resize-workers-interval", fmt.Sprintf("return(%d)", time.Second))
+	defer failpoint.Disable("github.com/pingcap/tidb/ttl/ttlworker/resize-workers-interval")
+
+	originAutoAnalyzeMinCnt := handle.AutoAnalyzeMinCnt
+	handle.AutoAnalyzeMinCnt = 0
+	defer func() {
+		handle.AutoAnalyzeMinCnt = originAutoAnalyzeMinCnt
+	}()
+
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+
+	tk.MustExec("use test")
+	tk.MustExec("create table t (id int, created_at datetime) ttl = `created_at` + interval 1 day")
+
+	// insert ten rows, the 2,3,4,6,9,10 of them are expired
+	for i := 1; i <= 10; i++ {
+		t := time.Now()
+		if i%2 == 0 || i%3 == 0 {
+			t = t.Add(-time.Hour * 48)
+		}
+
+		tk.MustExec("insert into t values(?, ?)", i, t.Format(time.RFC3339))
+	}
+	// turn off the `tidb_ttl_job_enable`
+	tk.MustExec("set global tidb_ttl_job_enable = 'OFF'")
+	defer tk.MustExec("set global tidb_ttl_job_enable = 'ON'")
+
+	retryTime := 15
+	retryInterval := time.Second * 2
+	deleted := false
+	for retryTime >= 0 {
+		retryTime--
+		time.Sleep(retryInterval)
+
+		rows := tk.MustQuery("select count(*) from t").Rows()
+		count, err := strconv.Atoi(rows[0][0].(string))
+		require.NoError(t, err)
+		if count < 10 {
+			deleted = true
+			break
+		}
+
+		require.Len(t, dom.TTLJobManager().RunningJobs(), 0)
+	}
+	require.False(t, deleted)
+}
+>>>>>>> 7e64d04f73 (ttl: don't schedule ttl job when EnableTTLJob is off (#40336))

ttl/ttlworker/job_manager_test.go

@@ -148,6 +148,11 @@ func (m *JobManager) LockNewJob(ctx context.Context, se session.Session, table *
 	return m.lockNewJob(ctx, se, table, now)
 }
 
+// RunningJobs returns the running jobs inside ttl job manager
+func (m *JobManager) RunningJobs() []*TTLJob {
+	return m.runningJobs
+}
+
 func (j *ttlJob) Finish(se session.Session, now time.Time) {
 	j.finish(se, now)
 }