update demo

domain/domain.go

@@ -27,8 +27,6 @@ import (
 	"time"
 	"unsafe"
 
-	"github.com/pingcap/tidb/ttl"
-
 	"github.com/ngaut/pools"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
@@ -61,6 +59,7 @@ import (
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/statistics/handle"
 	"github.com/pingcap/tidb/telemetry"
+	"github.com/pingcap/tidb/ttl/ttlworker"
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util"
 	"github.com/pingcap/tidb/util/dbterror"
@@ -1059,7 +1058,7 @@ func (do *Domain) Init(
 		return err
 	}
 
-	ttl.NewJobManager(do.sysSessionPool).Start()
+	ttlworker.NewJobManager(do.sysSessionPool).Start()
 	return nil
 }
 

ttl/util.go → ttl/session.go

@@ -17,15 +17,40 @@ package ttl
 import (
 	"context"
 
+	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/parser/terror"
+	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/util/chunk"
 	"github.com/pingcap/tidb/util/sqlexec"
 )
 
-func executeSQL(ctx context.Context, se *session, sql string, args ...interface{}) ([]chunk.Row, error) {
+type Session struct {
+	Sctx     sessionctx.Context
+	Executor sqlexec.SQLExecutor
+	CloseFn  func()
+}
+
+func (s *Session) GetDomainInfoSchema() infoschema.InfoSchema {
+	is, ok := s.Sctx.GetDomainInfoSchema().(infoschema.InfoSchema)
+	if !ok {
+		return nil
+	}
+
+	if ext, ok := is.(*infoschema.SessionExtendedInfoSchema); ok {
+		return ext.InfoSchema
+	}
+
+	return is
+}
+
+func (s *Session) ExecuteSQL(ctx context.Context, sql string, args ...interface{}) ([]chunk.Row, error) {
 	ctx = kv.WithInternalSourceType(ctx, kv.InternalTxnTTL)
-	rs, err := se.ExecuteInternal(ctx, sql, args...)
+	rs, err := s.Executor.ExecuteInternal(ctx, sql, args...)
+	if err != nil {
+		return nil, err
+	}
+
 	if err != nil {
 		return nil, err
 	}
@@ -40,3 +65,9 @@ func executeSQL(ctx context.Context, se *session, sql string, args ...interface{
 
 	return sqlexec.DrainRecordSet(ctx, rs, 8)
 }
+
+func (s *Session) Close() {
+	if s.CloseFn != nil {
+		s.CloseFn()
+	}
+}

ttl/states.go

@@ -0,0 +1,76 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ttl
+
+import (
+	"context"
+	"time"
+)
+
+type TableStatus string
+
+const (
+	TableStatusWaiting    TableStatus = "waiting"
+	TableStatusRunning                = "running"
+	TableStatusCancelling             = "cancelling"
+	TableStatusCancelled              = "cancelled"
+	TableStatusDone                   = "done"
+	TableStatusFailed                 = "failed"
+)
+
+type JobInfo struct {
+	ID        string
+	TableID   int64
+	StartTime time.Time
+}
+
+type TableStateInfo struct {
+	ID     int64
+	Status string
+}
+
+type TableStatesStore interface {
+	CreateNewJob(ctx context.Context, tblID int64, owner string) error
+}
+
+func NewTableStatesStore() TableStatesStore {
+	// TODO:
+	return nil
+}
+
+type TableStatesCache struct {
+	tables map[int64]*TableStateInfo
+}
+
+func NewTableStatesCache() *TableStatesCache {
+	return &TableStatesCache{
+		tables: make(map[int64]*TableStateInfo),
+	}
+}
+
+func (c *TableStatesCache) Update(ctx context.Context, se *Session) error {
+	// TODO:
+	return nil
+}
+
+func (c *TableStatesCache) UpdateTable(ctx context.Context, se *Session, tblID int) error {
+	// TODO:
+	return nil
+}
+
+func (c *TableStatesCache) GetTableState(id int64) (info *TableStateInfo, ok bool) {
+	info, ok = c.tables[id]
+	return
+}