ttl, domain: setup a customized session pool with stats collector

domain/domain.go

@@ -1062,10 +1062,6 @@ func (do *Domain) Init(
  return err
  }
 
- do.wg.Run(func() {
- do.runTTLJobManager(ctx)
- })
-
  return nil
 }
 
@@ -2457,18 +2453,21 @@ func (do *Domain) serverIDKeeper() {
  }
 }
 
-func (do *Domain) runTTLJobManager(ctx context.Context) {
- ttlJobManager := ttlworker.NewJobManager(do.ddl.GetID(), do.sysSessionPool, do.store)
- ttlJobManager.Start()
- do.ttlJobManager = ttlJobManager
+// StartTTLJobManager creates and starts the ttl job manager
+func (do *Domain) StartTTLJobManager() {
+ do.wg.Run(func() {
+ ttlJobManager := ttlworker.NewJobManager(do.ddl.GetID(), do.sysSessionPool, do.store)
+ do.ttlJobManager = ttlJobManager
+ ttlJobManager.Start()
 
- <-do.exit
+  <-do.exit
 
- ttlJobManager.Stop()
- err := ttlJobManager.WaitStopped(ctx, 30*time.Second)
- if err != nil {
- logutil.BgLogger().Warn("fail to wait until the ttl job manager stop", zap.Error(err))
- }
+ ttlJobManager.Stop()
+ err := ttlJobManager.WaitStopped(context.Background(), 30*time.Second)
+ if err != nil {
+ logutil.BgLogger().Warn("fail to wait until the ttl job manager stop", zap.Error(err))
+ }
+ })
 }
 
 // TTLJobManager returns the ttl job manager on this domain

session/BUILD.bazel

@@ -65,6 +65,7 @@ go_library(
  "//table/temptable",
  "//tablecodec",
  "//telemetry",
+ "//ttl/ttlworker",
  "//types",
  "//types/parser_driver",
  "//util",

session/session.go

@@ -86,6 +86,7 @@ import (
  "github.com/pingcap/tidb/table/temptable"
  "github.com/pingcap/tidb/tablecodec"
  "github.com/pingcap/tidb/telemetry"
+ "github.com/pingcap/tidb/ttl/ttlworker"
  "github.com/pingcap/tidb/util"
  "github.com/pingcap/tidb/util/chunk"
  "github.com/pingcap/tidb/util/collate"
@@ -3418,6 +3419,22 @@ func BootstrapSession(store kv.Storage) (*domain.Domain, error) {
  return nil, err
  }
 
+ // start TTL job manager after setup stats collector
+ // because TTL could modify a lot of columns, and need to trigger auto analyze
+ ttlworker.AttachStatsCollector = func(s sqlexec.SQLExecutor) sqlexec.SQLExecutor {
+ if s, ok := s.(*session); ok {
+ return attachStatsCollector(s, dom)
+ }
+ return s
+ }
+ ttlworker.DetachStatsCollector = func(s sqlexec.SQLExecutor) sqlexec.SQLExecutor {
+ if s, ok := s.(*session); ok {
+ return detachStatsCollector(s)
+ }
+ return s
+ }
+ dom.StartTTLJobManager()
+
  analyzeCtxs, err := createSessions(store, analyzeConcurrencyQuota)
  if err != nil {
  return nil, err
@@ -3523,6 +3540,26 @@ func createSessionWithOpt(store kv.Storage, opt *Opt) (*session, error) {
  return s, nil
 }
 
+// attachStatsCollector attaches the stats collector in the dom for the session
+func attachStatsCollector(s *session, dom *domain.Domain) *session {
+ if dom.StatsHandle() != nil && dom.StatsUpdating() {
+ s.statsCollector = dom.StatsHandle().NewSessionStatsCollector()
+ if GetIndexUsageSyncLease() > 0 {
+ s.idxUsageCollector = dom.StatsHandle().NewSessionIndexUsageCollector()
+ }
+ }
+
+ return s
+}
+
+// detachStatsCollector removes the stats collector in the session
+func detachStatsCollector(s *session) *session {
+ s.statsCollector = nil
+ s.idxUsageCollector = nil
+
+ return s
+}
+
 // CreateSessionWithDomain creates a new Session and binds it with a Domain.
 // We need this because when we start DDL in Domain, the DDL need a session
 // to change some system tables. But at that time, we have been already in

ttl/ttlworker/BUILD.bazel

@@ -31,6 +31,7 @@ go_library(
  "//util/timeutil",
  "@com_github_ngaut_pools//:pools",
  "@com_github_pingcap_errors//:errors",
+ "@com_github_pingcap_failpoint//:failpoint",
  "@org_golang_x_time//rate",
  "@org_uber_go_multierr//:multierr",
  "@org_uber_go_zap//:zap",
@@ -58,6 +59,7 @@ go_test(
  "//session",
  "//sessionctx",
  "//sessionctx/variable",
+ "//statistics/handle",
  "//testkit",
  "//ttl/cache",
  "//ttl/session",
@@ -66,6 +68,7 @@ go_test(
  "//util/logutil",
  "@com_github_ngaut_pools//:pools",
  "@com_github_pingcap_errors//:errors",
+ "@com_github_pingcap_failpoint//:failpoint",
  "@com_github_stretchr_testify//assert",
  "@com_github_stretchr_testify//require",
  "@org_golang_x_time//rate",

ttl/ttlworker/config.go

@@ -16,6 +16,8 @@ package ttlworker
 
 import (
  "time"
+
+ "github.com/pingcap/failpoint"
 )
 
 const jobManagerLoopTickerInterval = 10 * time.Second
@@ -27,3 +29,24 @@ const ttlInternalSQLTimeout = 30 * time.Second
 const resizeWorkersInterval = 30 * time.Second
 const splitScanCount = 64
 const ttlJobTimeout = 6 * time.Hour
+
+func getUpdateInfoSchemaCacheInterval() time.Duration {
+ failpoint.Inject("update-info-schema-cache-interval", func(val failpoint.Value) time.Duration {
+ return time.Duration(val.(int))
+ })
+ return updateInfoSchemaCacheInterval
+}
+
+func getUpdateTTLTableStatusCacheInterval() time.Duration {
+ failpoint.Inject("update-status-table-cache-interval", func(val failpoint.Value) time.Duration {
+ return time.Duration(val.(int))
+ })
+ return updateTTLTableStatusCacheInterval
+}
+
+func getResizeWorkersInterval() time.Duration {
+ failpoint.Inject("resize-workers-interval", func(val failpoint.Value) time.Duration {
+ return time.Duration(val.(int))
+ })
+ return resizeWorkersInterval
+}

ttl/ttlworker/job_manager.go

@@ -102,8 +102,8 @@ func NewJobManager(id string, sessPool sessionPool, store kv.Storage) (manager *
  manager.init(manager.jobLoop)
  manager.ctx = logutil.WithKeyValue(manager.ctx, "ttl-worker", "manager")
 
- manager.infoSchemaCache = cache.NewInfoSchemaCache(updateInfoSchemaCacheInterval)
- manager.tableStatusCache = cache.NewTableStatusCache(updateTTLTableStatusCacheInterval)
+ manager.infoSchemaCache = cache.NewInfoSchemaCache(getUpdateInfoSchemaCacheInterval())
+ manager.tableStatusCache = cache.NewTableStatusCache(getUpdateTTLTableStatusCacheInterval())
 
  return
 }
@@ -125,7 +125,7 @@ func (m *JobManager) jobLoop() error {
  updateScanTaskStateTicker := time.Tick(jobManagerLoopTickerInterval)
  infoSchemaCacheUpdateTicker := time.Tick(m.infoSchemaCache.GetInterval())
  tableStatusCacheUpdateTicker := time.Tick(m.tableStatusCache.GetInterval())
- resizeWorkersTicker := time.Tick(resizeWorkersInterval)
+ resizeWorkersTicker := time.Tick(getResizeWorkersInterval())
  for {
  m.reportMetrics()
  now := se.Now()
@@ -487,7 +487,7 @@ func (m *JobManager) couldTrySchedule(table *cache.TableStatus, now time.Time) b
  hbTime := table.CurrentJobOwnerHBTime
  // a more concrete value is `2 * max(updateTTLTableStatusCacheInterval, jobManagerLoopTickerInterval)`, but the
  // `updateTTLTableStatusCacheInterval` is greater than `jobManagerLoopTickerInterval` in most cases.
- if hbTime.Add(2 * updateTTLTableStatusCacheInterval).Before(now) {
+ if hbTime.Add(2 * getUpdateTTLTableStatusCacheInterval()).Before(now) {
  logutil.Logger(m.ctx).Info("task heartbeat has stopped", zap.Int64("tableID", table.TableID), zap.Time("hbTime", hbTime), zap.Time("now", now))
  return true
  }

ttl/ttlworker/job_manager_integration_test.go

@@ -22,11 +22,13 @@ import (
  "time"
 
  "github.com/pingcap/errors"
+ "github.com/pingcap/failpoint"
  "github.com/pingcap/tidb/kv"
  "github.com/pingcap/tidb/parser/ast"
  "github.com/pingcap/tidb/parser/model"
  dbsession "github.com/pingcap/tidb/session"
  "github.com/pingcap/tidb/sessionctx/variable"
+ "github.com/pingcap/tidb/statistics/handle"
  "github.com/pingcap/tidb/testkit"
  "github.com/pingcap/tidb/ttl/cache"
  "github.com/pingcap/tidb/ttl/session"
@@ -123,3 +125,58 @@ 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'\\\"\"}"))
 }
+
+func TestTTLAutoAnalyze(t *testing.T) {
+ failpoint.Enable("github.com/pingcap/tidb/ttl/ttlworker/update-info-schema-cache-interval", fmt.Sprintf("return(%d)", time.Second))
+ failpoint.Enable("github.com/pingcap/tidb/ttl/ttlworker/update-status-table-cache-interval", fmt.Sprintf("return(%d)", time.Second))
+ failpoint.Enable("github.com/pingcap/tidb/ttl/ttlworker/resize-workers-interval", fmt.Sprintf("return(%d)", time.Second))
+ 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))
+}

ttl/ttlworker/job_manager_test.go

@@ -167,7 +167,8 @@ func newMockTTLJob(tbl *cache.PhysicalTable, status cache.JobStatus) *ttlJob {
 
 func TestReadyForNewJobTables(t *testing.T) {
  tbl := newMockTTLTbl(t, "t1")
- m := NewJobManager("test-id", newMockSessionPool(t, tbl), nil)
+ m := NewJobManager("test-id", nil, nil)
+ m.sessPool = newMockSessionPool(t, tbl)
  se := newMockSession(t, tbl)
 
  cases := []struct {
@@ -295,7 +296,8 @@ func TestLockNewTable(t *testing.T) {
  t.Run(c.name, func(t *testing.T) {
  tbl := newMockTTLTbl(t, "t1")
 
- m := NewJobManager("test-id", newMockSessionPool(t, tbl), nil)
+ m := NewJobManager("test-id", nil, nil)
+ m.sessPool = newMockSessionPool(t, tbl)
  sqlCounter := 0
  se := newMockSession(t, tbl)
  se.executeSQL = func(ctx context.Context, sql string, args ...interface{}) (rows []chunk.Row, err error) {
@@ -333,7 +335,8 @@ func TestResizeWorkers(t *testing.T) {
  scanWorker1.Start()
  scanWorker2 := newMockScanWorker(t)
 
- m := NewJobManager("test-id", newMockSessionPool(t, tbl), nil)
+ m := NewJobManager("test-id", nil, nil)
+ m.sessPool = newMockSessionPool(t, tbl)
  m.SetScanWorkers4Test([]worker{
  scanWorker1,
  })
@@ -351,7 +354,8 @@ func TestResizeWorkers(t *testing.T) {
  scanWorker2 = newMockScanWorker(t)
  scanWorker2.Start()
 
- m = NewJobManager("test-id", newMockSessionPool(t, tbl), nil)
+ m = NewJobManager("test-id", nil, nil)
+ m.sessPool = newMockSessionPool(t, tbl)
  m.SetScanWorkers4Test([]worker{
  scanWorker1,
  scanWorker2,
@@ -366,7 +370,8 @@ func TestResizeWorkers(t *testing.T) {
  scanWorker2 = newMockScanWorker(t)
  scanWorker2.Start()
 
- m = NewJobManager("test-id", newMockSessionPool(t, tbl), nil)
+ m = NewJobManager("test-id", nil, nil)
+ m.sessPool = newMockSessionPool(t, tbl)
  m.SetScanWorkers4Test([]worker{
  scanWorker1,
  scanWorker2,
@@ -384,7 +389,8 @@ func TestLocalJobs(t *testing.T) {
  tbl1.ID = 1
  tbl2 := newMockTTLTbl(t, "t2")
  tbl2.ID = 2
- m := NewJobManager("test-id", newMockSessionPool(t, tbl1, tbl2), nil)
+ m := NewJobManager("test-id", nil, nil)
+ m.sessPool = newMockSessionPool(t, tbl1, tbl2)
 
  m.runningJobs = []*ttlJob{{tbl: tbl1, id: "1", ctx: context.Background()}, {tbl: tbl2, id: "2", ctx: context.Background()}}
  m.tableStatusCache.Tables = map[int64]*cache.TableStatus{
@@ -410,7 +416,8 @@ func TestRescheduleJobs(t *testing.T) {
  scanWorker2.Start()
  scanWorker2.setOneRowResult(tbl, 2022)
 
- m := NewJobManager("test-id", newMockSessionPool(t, tbl), nil)
+ m := NewJobManager("test-id", nil, nil)
+ m.sessPool = newMockSessionPool(t, tbl)
  m.SetScanWorkers4Test([]worker{
  scanWorker1,
  scanWorker2,
@@ -463,7 +470,8 @@ func TestRescheduleJobsOutOfWindow(t *testing.T) {
  scanWorker2.Start()
  scanWorker2.setOneRowResult(tbl, 2022)
 
- m := NewJobManager("test-id", newMockSessionPool(t, tbl), nil)
+ m := NewJobManager("test-id", nil, nil)
+ m.sessPool = newMockSessionPool(t, tbl)
  m.SetScanWorkers4Test([]worker{
  scanWorker1,
  scanWorker2,
@@ -508,7 +516,8 @@ func TestCheckFinishedJob(t *testing.T) {
  se := newMockSession(t, tbl)
 
  // cancelled job will be regarded as finished
- m := NewJobManager("test-id", newMockSessionPool(t, tbl), nil)
+ m := NewJobManager("test-id", nil, nil)
+ m.sessPool = newMockSessionPool(t, tbl)
  m.runningJobs = []*ttlJob{newMockTTLJob(tbl, cache.JobStatusCancelled)}
  m.checkFinishedJob(se, se.Now())
  assert.Len(t, m.runningJobs, 0)
@@ -517,7 +526,8 @@ func TestCheckFinishedJob(t *testing.T) {
  finishedStatistics := &ttlStatistics{}
  finishedStatistics.TotalRows.Store(1)
  finishedStatistics.SuccessRows.Store(1)
- m = NewJobManager("test-id", newMockSessionPool(t, tbl), nil)
+ m = NewJobManager("test-id", nil, nil)
+ m.sessPool = newMockSessionPool(t, tbl)
  m.runningJobs = []*ttlJob{newMockTTLJob(tbl, cache.JobStatusRunning)}
  m.runningJobs[0].statistics = finishedStatistics
  m.runningJobs[0].tasks[0].statistics = finishedStatistics
@@ -545,7 +555,8 @@ func TestCheckFinishedJob(t *testing.T) {
  // check timeout job
  now = se.Now()
  createTime := now.Add(-20 * time.Hour)
- m = NewJobManager("test-id", newMockSessionPool(t, tbl), nil)
+ m = NewJobManager("test-id", nil, nil)
+ m.sessPool = newMockSessionPool(t, tbl)
  m.runningJobs = []*ttlJob{
  {
  ctx: context.Background(),