From ddaf3e737fb1b9aed7404f37a834f1dc322f6770 Mon Sep 17 00:00:00 2001
From: Chao Wang <cclcwangchao@hotmail.com>
Date: Tue, 17 Jan 2023 16:27:32 +0800
Subject: [PATCH 1/5] ttl: add table `mysql.ttl_job_history` to record
histories
---
session/bootstrap.go | 25 +++++++-
ttl/cache/ttlstatus.go | 10 ++--
ttl/ttlworker/job.go | 57 ++++++++++++++++++-
ttl/ttlworker/job_manager.go | 3 +-
ttl/ttlworker/job_manager_integration_test.go | 26 +++++++--
ttl/ttlworker/job_manager_test.go | 4 ++
6 files changed, 114 insertions(+), 11 deletions(-)
diff --git a/session/bootstrap.go b/session/bootstrap.go
index 74eecb28a68a4..9ac20fe55e491 100644
--- a/session/bootstrap.go
+++ b/session/bootstrap.go
@@ -516,6 +516,26 @@ const (
created_time timestamp NOT NULL,
primary key(job_id, scan_id),
key(created_time));`
+
+ // CreateTTLJobHistory is a table that stores ttl job's history
+ CreateTTLJobHistory = `CREATE TABLE IF NOT EXISTS mysql.tidb_ttl_job_history (
+ job_id varchar(64) PRIMARY KEY,
+ table_id bigint(64) NOT NULL,
+ parent_table_id bigint(64) NOT NULL,
+ table_schema varchar(64) NOT NULL,
+ table_name varchar(64) NOT NULL,
+ partition_name varchar(64) DEFAULT NULL,
+ create_time timestamp NOT NULL,
+ finish_time timestamp NOT NULL,
+ ttl_expire timestamp NOT NULL,
+ summary_text text,
+ expired_rows bigint(64) DEFAULT NULL,
+ deleted_rows bigint(64) DEFAULT NULL,
+ error_delete_rows bigint(64) DEFAULT NULL,
+ status varchar(64) NOT NULL,
+ key(create_time),
+ key(finish_time)
+ );`
)
// bootstrap initiates system DB for a store.
@@ -757,7 +777,7 @@ const (
version109 = 109
// version110 sets tidb_enable_gc_aware_memory_track to off when a cluster upgrades from some version lower than v6.5.0.
version110 = 110
- // version111 adds the table tidb_ttl_task
+ // version111 adds the table tidb_ttl_task and tidb_ttl_job_history
version111 = 111
)
@@ -2239,6 +2259,7 @@ func upgradeToVer111(s Session, ver int64) {
return
}
doReentrantDDL(s, CreateTTLTask)
+ doReentrantDDL(s, CreateTTLJobHistory)
}
func writeOOMAction(s Session) {
@@ -2349,6 +2370,8 @@ func doDDLWorks(s Session) {
mustExecute(s, CreateTTLTableStatus)
// Create tidb_ttl_task table
mustExecute(s, CreateTTLTask)
+ // Create tidb_ttl_job_history table
+ mustExecute(s, CreateTTLJobHistory)
}
// doBootstrapSQLFile executes SQL commands in a file as the last stage of bootstrap.
diff --git a/ttl/cache/ttlstatus.go b/ttl/cache/ttlstatus.go
index d28bafa5a76c8..b21a50a161f79 100644
--- a/ttl/cache/ttlstatus.go
+++ b/ttl/cache/ttlstatus.go
@@ -30,13 +30,15 @@ const (
// JobStatusWaiting means the job hasn't started
JobStatusWaiting JobStatus = "waiting"
// JobStatusRunning means this job is running
- JobStatusRunning = "running"
+ JobStatusRunning JobStatus = "running"
// JobStatusCancelling means this job is being canceled, but not canceled yet
- JobStatusCancelling = "cancelling"
+ JobStatusCancelling JobStatus = "cancelling"
// JobStatusCancelled means this job has been canceled successfully
- JobStatusCancelled = "cancelled"
+ JobStatusCancelled JobStatus = "cancelled"
// JobStatusTimeout means this job has timeout
- JobStatusTimeout = "timeout"
+ JobStatusTimeout JobStatus = "timeout"
+ // JobStatusFinished means job has been finished
+ JobStatusFinished JobStatus = "finished"
)
const selectFromTTLTableStatus = "SELECT LOW_PRIORITY table_id,parent_table_id,table_statistics,last_job_id,last_job_start_time,last_job_finish_time,last_job_ttl_expire,last_job_summary,current_job_id,current_job_owner_id,current_job_owner_addr,current_job_owner_hb_time,current_job_start_time,current_job_ttl_expire,current_job_state,current_job_status,current_job_status_update_time FROM mysql.tidb_ttl_table_status"
diff --git a/ttl/ttlworker/job.go b/ttl/ttlworker/job.go
index 4b5ffb147bf1d..b15f04f8b03a5 100644
--- a/ttl/ttlworker/job.go
+++ b/ttl/ttlworker/job.go
@@ -46,6 +46,25 @@ const finishJobTemplate = `UPDATE mysql.tidb_ttl_table_status
WHERE table_id = %? AND current_job_id = %?`
const updateJobStateTemplate = "UPDATE mysql.tidb_ttl_table_status SET current_job_state = %? WHERE table_id = %? AND current_job_id = %? AND current_job_owner_id = %?"
const removeTaskForJobTemplate = "DELETE FROM mysql.tidb_ttl_task WHERE job_id = %?"
+const addJobHistoryTemplate = `INSERT INTO
+ mysql.tidb_ttl_job_history (
+ job_id,
+ table_id,
+ parent_table_id,
+ table_schema,
+ table_name,
+ partition_name,
+ create_time,
+ finish_time,
+ ttl_expire,
+ summary_text,
+ expired_rows,
+ deleted_rows,
+ error_delete_rows,
+ status
+ )
+VALUES
+ (%?, %?, %?, %?, %?, %?, %?, %?, %?, %?, %?, %?, %?, %?)`
func updateJobCurrentStatusSQL(tableID int64, oldStatus cache.JobStatus, newStatus cache.JobStatus, jobID string) (string, []interface{}) {
return updateJobCurrentStatusTemplate, []interface{}{string(newStatus), tableID, string(oldStatus), jobID}
@@ -63,6 +82,35 @@ func removeTaskForJob(jobID string) (string, []interface{}) {
return removeTaskForJobTemplate, []interface{}{jobID}
}
+func addJobHistorySQL(job *ttlJob, finishTime time.Time, summaryText string) (string, []interface{}) {
+ status := cache.JobStatusFinished
+ if job.status == cache.JobStatusTimeout || job.status == cache.JobStatusCancelled {
+ status = job.status
+ }
+
+ var partitionName interface{}
+ if job.tbl.Partition.L != "" {
+ partitionName = job.tbl.Partition.O
+ }
+
+ return addJobHistoryTemplate, []interface{}{
+ job.id,
+ job.tbl.ID,
+ job.tbl.TableInfo.ID,
+ job.tbl.Schema.O,
+ job.tbl.Name.O,
+ partitionName,
+ job.createTime.Format(timeFormat),
+ finishTime.Format(timeFormat),
+ job.ttlExpireTime.Format(timeFormat),
+ summaryText,
+ job.statistics.TotalRows.Load(),
+ job.statistics.SuccessRows.Load(),
+ job.statistics.ErrorRows.Load(),
+ string(status),
+ }
+}
+
type ttlJob struct {
id string
ownerID string
@@ -70,7 +118,8 @@ type ttlJob struct {
ctx context.Context
cancel func()
- createTime time.Time
+ createTime time.Time
+ ttlExpireTime time.Time
tbl *cache.PhysicalTable
@@ -149,6 +198,12 @@ func (job *ttlJob) finish(se session.Session, now time.Time) {
return errors.Wrapf(err, "execute sql: %s", sql)
}
+ sql, args = addJobHistorySQL(job, now, summary)
+ _, err = se.ExecuteSQL(context.TODO(), sql, args...)
+ if err != nil {
+ return errors.Wrapf(err, "execute sql: %s", sql)
+ }
+
return nil
})
diff --git a/ttl/ttlworker/job_manager.go b/ttl/ttlworker/job_manager.go
index 910038666c1b6..25139cd7d6c3a 100644
--- a/ttl/ttlworker/job_manager.go
+++ b/ttl/ttlworker/job_manager.go
@@ -782,7 +782,8 @@ func (m *JobManager) createNewJob(expireTime time.Time, now time.Time, table *ca
ctx: jobCtx,
cancel: cancel,
- createTime: now,
+ createTime: now,
+ ttlExpireTime: expireTime,
// at least, the info schema cache and table status cache are consistent in table id, so it's safe to get table
// information from schema cache directly
tbl: table,
diff --git a/ttl/ttlworker/job_manager_integration_test.go b/ttl/ttlworker/job_manager_integration_test.go
index a1e284540a594..4495ce3668c16 100644
--- a/ttl/ttlworker/job_manager_integration_test.go
+++ b/ttl/ttlworker/job_manager_integration_test.go
@@ -18,6 +18,7 @@ import (
"context"
"fmt"
"strconv"
+ "strings"
"sync"
"testing"
"time"
@@ -112,7 +113,7 @@ func TestFinishJob(t *testing.T) {
sessionFactory := sessionFactory(t, store)
- testTable := &cache.PhysicalTable{ID: 2, TableInfo: &model.TableInfo{ID: 1, TTLInfo: &model.TTLInfo{IntervalExprStr: "1", IntervalTimeUnit: int(ast.TimeUnitDay)}}}
+ testTable := &cache.PhysicalTable{ID: 2, Schema: model.NewCIStr("db1"), TableInfo: &model.TableInfo{ID: 1, Name: model.NewCIStr("t1"), TTLInfo: &model.TTLInfo{IntervalExprStr: "1", IntervalTimeUnit: int(ast.TimeUnitDay)}}}
tk.MustExec("insert into mysql.tidb_ttl_table_status(table_id) values (2)")
@@ -120,13 +121,30 @@ func TestFinishJob(t *testing.T) {
m := ttlworker.NewJobManager("test-id", nil, store, nil)
m.InfoSchemaCache().Tables[testTable.ID] = testTable
se := sessionFactory()
- job, err := m.LockNewJob(context.Background(), se, testTable, time.Now(), false)
+ startTime := time.Now()
+ job, err := m.LockNewJob(context.Background(), se, testTable, startTime, false)
require.NoError(t, err)
job.SetScanErr(errors.New(`"'an error message contains both single and double quote'"`))
- job.Finish(se, time.Now())
+ job.Statistics().TotalRows.Add(128)
+ job.Statistics().SuccessRows.Add(120)
+ job.Statistics().ErrorRows.Add(8)
+ time.Sleep(time.Second)
+ endTime := time.Now()
+ job.Finish(se, endTime)
+
+ expireTime, err := testTable.EvalExpireTime(context.Background(), se, startTime)
+ require.NoError(t, err)
- 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'\\\"\"}"))
+ timeFormat := "2006-01-02 15:04:05"
+ lastJobSummary := "{\"total_rows\":128,\"success_rows\":120,\"error_rows\":8,\"total_scan_task\":1,\"scheduled_scan_task\":0,\"finished_scan_task\":0,\"scan_task_err\":\"\\\"'an error message contains both single and double quote'\\\"\"}"
+ tk.MustQuery("select table_id, last_job_summary from mysql.tidb_ttl_table_status").Check(testkit.Rows("2 " + lastJobSummary))
tk.MustQuery("select * from mysql.tidb_ttl_task").Check(testkit.Rows())
+ expectedRow := []string{
+ job.ID(), "2", "1", "db1", "t1", "<nil>",
+ startTime.Format(timeFormat), endTime.Format(timeFormat), expireTime.Format(timeFormat),
+ lastJobSummary, "128", "120", "8", "finished",
+ }
+ tk.MustQuery("select * from mysql.tidb_ttl_job_history").Check(testkit.Rows(strings.Join(expectedRow, " ")))
}
func TestTTLAutoAnalyze(t *testing.T) {
diff --git a/ttl/ttlworker/job_manager_test.go b/ttl/ttlworker/job_manager_test.go
index 97a84b3cb82a2..bbfa83fb1111d 100644
--- a/ttl/ttlworker/job_manager_test.go
+++ b/ttl/ttlworker/job_manager_test.go
@@ -164,6 +164,10 @@ func (j *ttlJob) Finish(se session.Session, now time.Time) {
j.finish(se, now)
}
+func (j *ttlJob) Statistics() *ttlStatistics {
+ return j.statistics
+}
+
func (j *ttlJob) ID() string {
return j.id
}
From a606d5822c502db350c0ab9aae2b64c49c538d85 Mon Sep 17 00:00:00 2001
From: Chao Wang <cclcwangchao@hotmail.com>
Date: Tue, 17 Jan 2023 17:24:49 +0800
Subject: [PATCH 2/5] fix ut
---
ttl/ttlworker/job_manager_test.go | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/ttl/ttlworker/job_manager_test.go b/ttl/ttlworker/job_manager_test.go
index bbfa83fb1111d..df67e957ce0b4 100644
--- a/ttl/ttlworker/job_manager_test.go
+++ b/ttl/ttlworker/job_manager_test.go
@@ -16,6 +16,7 @@ package ttlworker
import (
"context"
+ "strings"
"testing"
"time"
@@ -582,7 +583,7 @@ func TestCheckFinishedJob(t *testing.T) {
now := se.Now()
jobID := m.runningJobs[0].id
se.executeSQL = func(ctx context.Context, sql string, args ...interface{}) ([]chunk.Row, error) {
- if len(args) > 1 {
+ if strings.Contains(sql, "tidb_ttl_table_status") {
meetArg = true
expectedSQL, expectedArgs := finishJobSQL(tbl.ID, now, "{\"total_rows\":1,\"success_rows\":1,\"error_rows\":0,\"total_scan_task\":1,\"scheduled_scan_task\":1,\"finished_scan_task\":1}", jobID)
assert.Equal(t, expectedSQL, sql)
From da71974c1c4e5858324a6a9124f12e4e47e532cd Mon Sep 17 00:00:00 2001
From: Chao Wang <cclcwangchao@hotmail.com>
Date: Tue, 17 Jan 2023 18:19:54 +0800
Subject: [PATCH 3/5] update time
---
executor/infoschema_cluster_table_test.go | 2 +-
session/bootstrap.go | 4 ++--
2 files changed, 3 insertions(+), 3 deletions(-)
diff --git a/executor/infoschema_cluster_table_test.go b/executor/infoschema_cluster_table_test.go
index be2f04cb5c6ac..b1a6d4c57f4f8 100644
--- a/executor/infoschema_cluster_table_test.go
+++ b/executor/infoschema_cluster_table_test.go
@@ -290,7 +290,7 @@ func TestTableStorageStats(t *testing.T) {
"test 2",
))
rows := tk.MustQuery("select TABLE_NAME from information_schema.TABLE_STORAGE_STATS where TABLE_SCHEMA = 'mysql';").Rows()
- result := 45
+ result := 46
require.Len(t, rows, result)
// More tests about the privileges.
diff --git a/session/bootstrap.go b/session/bootstrap.go
index 9ac20fe55e491..af856acb268ad 100644
--- a/session/bootstrap.go
+++ b/session/bootstrap.go
@@ -533,8 +533,8 @@ const (
deleted_rows bigint(64) DEFAULT NULL,
error_delete_rows bigint(64) DEFAULT NULL,
status varchar(64) NOT NULL,
- key(create_time),
- key(finish_time)
+ key(parent_table_id, create_time),
+ key(create_time)
);`
)
From b8ff918be284f4315a69e26b892e6048ad8e2a72 Mon Sep 17 00:00:00 2001
From: Chao Wang <cclcwangchao@hotmail.com>
Date: Tue, 17 Jan 2023 18:28:27 +0800
Subject: [PATCH 4/5] update
---
session/bootstrap.go | 1 +
1 file changed, 1 insertion(+)
diff --git a/session/bootstrap.go b/session/bootstrap.go
index af856acb268ad..ed65bb0720cf0 100644
--- a/session/bootstrap.go
+++ b/session/bootstrap.go
@@ -533,6 +533,7 @@ const (
deleted_rows bigint(64) DEFAULT NULL,
error_delete_rows bigint(64) DEFAULT NULL,
status varchar(64) NOT NULL,
+ key(table_schema, table_name, create_time),
key(parent_table_id, create_time),
key(create_time)
);`
From 30e88b2367f573de0f8eb878282abfd688b45fa5 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E7=8E=8B=E8=B6=85?= <cclcwangchao@hotmail.com>
Date: Wed, 18 Jan 2023 10:04:23 +0800
Subject: [PATCH 5/5] Update ttl/ttlworker/job.go
Co-authored-by: bb7133 <bb7133@gmail.com>
---
ttl/ttlworker/job.go | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/ttl/ttlworker/job.go b/ttl/ttlworker/job.go
index b15f04f8b03a5..d3d919cb89d80 100644
--- a/ttl/ttlworker/job.go
+++ b/ttl/ttlworker/job.go
@@ -89,7 +89,7 @@ func addJobHistorySQL(job *ttlJob, finishTime time.Time, summaryText string) (st
}
var partitionName interface{}
- if job.tbl.Partition.L != "" {
+ if job.tbl.Partition.O != "" {
partitionName = job.tbl.Partition.O
}