Skip to content

Commit

Permalink
This is an automated cherry-pick of pingcap#40336
Browse files Browse the repository at this point in the history
Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>
  • Loading branch information
YangKeao authored and ti-chi-bot committed Jan 5, 2023
1 parent f700b56 commit 2c056cf
Show file tree
Hide file tree
Showing 3 changed files with 139 additions and 0 deletions.
18 changes: 18 additions & 0 deletions ttl/ttlworker/job_manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
116 changes: 116 additions & 0 deletions ttl/ttlworker/job_manager_integration_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ package ttlworker_test
import (
"context"
"fmt"
"strconv"
"sync"
"testing"
"time"
Expand Down Expand Up @@ -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))
5 changes: 5 additions & 0 deletions ttl/ttlworker/job_manager_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down

0 comments on commit 2c056cf

Please sign in to comment.