func (m *gcTTLManager) Len() int {
	return len(m.tableGCSafeTS)
}
following function are for implement `heap.Interface`
package restore
import (
	"container/heap"
	"context"
	"database/sql"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/google/uuid"
	"github.com/pingcap/br/pkg/checksum"
	"github.com/pingcap/br/pkg/utils"
	"github.com/pingcap/errors"
	"github.com/pingcap/failpoint"
	tidbcfg "github.com/pingcap/tidb/config"
	"github.com/pingcap/tidb/kv"
	"github.com/pingcap/tidb/store/tikv"
	"github.com/pingcap/tidb/store/tikv/oracle"
	"github.com/pingcap/tipb/go-tipb"
	pd "github.com/tikv/pd/client"
	"go.uber.org/zap"
	. "github.com/pingcap/tidb-lightning/lightning/checkpoints"
	"github.com/pingcap/tidb-lightning/lightning/common"
	"github.com/pingcap/tidb-lightning/lightning/config"
	"github.com/pingcap/tidb-lightning/lightning/log"
	"github.com/pingcap/tidb-lightning/lightning/metric"
)
const (
	preUpdateServiceSafePointFactor = 3
	maxErrorRetryCount = 3
)
var (
	serviceSafePointTTL int64 = 10 * 60  
	minDistSQLScanConcurrency = 4
)
type RemoteChecksum struct {
	Schema     string
	Table      string
	Checksum   uint64
	TotalKVs   uint64
	TotalBytes uint64
}
type ChecksumManager interface {
	Checksum(ctx context.Context, tableInfo *TidbTableInfo) (*RemoteChecksum, error)
}
func newChecksumManager(ctx context.Context, rc *RestoreController) (ChecksumManager, error) {
	if rc.cfg.TikvImporter.Backend == config.BackendTiDB || rc.cfg.PostRestore.Checksum == config.OpLevelOff {
		return nil, nil
	}
	pdAddr := rc.cfg.TiDB.PdAddr
	pdVersion, err := common.FetchPDVersion(ctx, rc.tls, pdAddr)
	if err != nil {
		return nil, errors.Trace(err)
	}
	var manager ChecksumManager
	if pdVersion.Major >= 4 {
		tlsOpt := rc.tls.ToPDSecurityOption()
		pdCli, err := pd.NewClientWithContext(ctx, []string{pdAddr}, tlsOpt)
		if err != nil {
			return nil, errors.Trace(err)
		}
		if tlsOpt.CAPath != "" {
			conf := tidbcfg.GetGlobalConfig()
			conf.Security.ClusterSSLCA = tlsOpt.CAPath
			conf.Security.ClusterSSLCert = tlsOpt.CertPath
			conf.Security.ClusterSSLKey = tlsOpt.KeyPath
			tidbcfg.StoreGlobalConfig(conf)
		}
		store, err := tikv.Driver{}.Open(fmt.Sprintf("tikv://%s?disableGC=true", pdAddr))
		if err != nil {
			return nil, errors.Trace(err)
		}
		manager = newTiKVChecksumManager(store.(tikv.Storage).GetClient(), pdCli, uint(rc.cfg.TiDB.DistSQLScanConcurrency))
	} else {
		db, err := rc.tidbGlue.GetDB()
		if err != nil {
			return nil, errors.Trace(err)
		}
		manager = newTiDBChecksumExecutor(db)
	}
	return manager, nil
}
type tidbChecksumExecutor struct {
	db      *sql.DB
	manager *gcLifeTimeManager
}
func newTiDBChecksumExecutor(db *sql.DB) *tidbChecksumExecutor {
	return &tidbChecksumExecutor{
		db:      db,
		manager: newGCLifeTimeManager(),
	}
}
func (e *tidbChecksumExecutor) Checksum(ctx context.Context, tableInfo *TidbTableInfo) (*RemoteChecksum, error) {
	var err error
	if err = e.manager.addOneJob(ctx, e.db); err != nil {
		return nil, err
	}
	defer e.manager.removeOneJob(ctx, e.db)
	tableName := common.UniqueTable(tableInfo.DB, tableInfo.Name)
	task := log.With(zap.String("table", tableName)).Begin(zap.InfoLevel, "remote checksum")
	cs := RemoteChecksum{}
	err = common.SQLWithRetry{DB: e.db, Logger: task.Logger}.QueryRow(ctx, "compute remote checksum",
		"ADMIN CHECKSUM TABLE "+tableName, &cs.Schema, &cs.Table, &cs.Checksum, &cs.TotalKVs, &cs.TotalBytes,
	)
	dur := task.End(zap.ErrorLevel, err)
	metric.ChecksumSecondsHistogram.Observe(dur.Seconds())
	if err != nil {
		return nil, errors.Trace(err)
	}
	return &cs, nil
}
func DoChecksum(ctx context.Context, table *TidbTableInfo) (*RemoteChecksum, error) {
	var err error
	manager, ok := ctx.Value(&checksumManagerKey).(ChecksumManager)
	if !ok {
		return nil, errors.New("No gcLifeTimeManager found in context, check context initialization")
	}
	task := log.With(zap.String("table", table.Name)).Begin(zap.InfoLevel, "remote checksum")
	cs, err := manager.Checksum(ctx, table)
	dur := task.End(zap.ErrorLevel, err)
	metric.ChecksumSecondsHistogram.Observe(dur.Seconds())
	return cs, err
}
type gcLifeTimeManager struct {
	runningJobsLock sync.Mutex
	runningJobs     int
	oriGCLifeTime   string
}
func newGCLifeTimeManager() *gcLifeTimeManager {
	return &gcLifeTimeManager{}
}
func (m *gcLifeTimeManager) addOneJob(ctx context.Context, db *sql.DB) error {
	m.runningJobsLock.Lock()
	defer m.runningJobsLock.Unlock()
	if m.runningJobs == 0 {
		oriGCLifeTime, err := ObtainGCLifeTime(ctx, db)
		if err != nil {
			return err
		}
		m.oriGCLifeTime = oriGCLifeTime
		err = increaseGCLifeTime(ctx, m, db)
		if err != nil {
			return err
		}
	}
	m.runningJobs += 1
	return nil
}
func (m *gcLifeTimeManager) removeOneJob(ctx context.Context, db *sql.DB) {
	m.runningJobsLock.Lock()
	defer m.runningJobsLock.Unlock()
	m.runningJobs -= 1
	if m.runningJobs == 0 {
		err := UpdateGCLifeTime(ctx, db, m.oriGCLifeTime)
		if err != nil {
			query := fmt.Sprintf(
				"UPDATE mysql.tidb SET VARIABLE_VALUE = '%s' WHERE VARIABLE_NAME = 'tikv_gc_life_time'",
				m.oriGCLifeTime,
			)
			log.L().Warn("revert GC lifetime failed, please reset the GC lifetime manually after Lightning completed",
				zap.String("query", query),
				log.ShortError(err),
			)
		}
	}
}
func increaseGCLifeTime(ctx context.Context, manager *gcLifeTimeManager, db *sql.DB) (err error) {
	var increaseGCLifeTime bool
	if manager.oriGCLifeTime != "" {
		ori, err := time.ParseDuration(manager.oriGCLifeTime)
		if err != nil {
			return errors.Trace(err)
		}
		if ori < defaultGCLifeTime {
			increaseGCLifeTime = true
		}
	} else {
		increaseGCLifeTime = true
	}
	if increaseGCLifeTime {
		err = UpdateGCLifeTime(ctx, db, defaultGCLifeTime.String())
		if err != nil {
			return err
		}
	}
	failpoint.Inject("IncreaseGCUpdateDuration", nil)
	return nil
}
type tikvChecksumManager struct {
	client                 kv.Client
	manager                gcTTLManager
	distSQLScanConcurrency uint
}
func newTiKVChecksumManager(client kv.Client, pdClient pd.Client, distSQLScanConcurrency uint) *tikvChecksumManager {
	return &tikvChecksumManager{
		client:                 client,
		manager:                newGCTTLManager(pdClient),
		distSQLScanConcurrency: distSQLScanConcurrency,
	}
}
func (e *tikvChecksumManager) checksumDB(ctx context.Context, tableInfo *TidbTableInfo) (*RemoteChecksum, error) {
	executor, err := checksum.NewExecutorBuilder(tableInfo.Core, oracle.ComposeTS(time.Now().Unix()*1000, 0)).
		SetConcurrency(e.distSQLScanConcurrency).
		Build()
	if err != nil {
		return nil, errors.Trace(err)
	}
	distSQLScanConcurrency := int(e.distSQLScanConcurrency)
	for i := 0; i < maxErrorRetryCount; i++ {
		_ = executor.Each(func(request *kv.Request) error {
			request.Concurrency = distSQLScanConcurrency
			return nil
		})
		var execRes *tipb.ChecksumResponse
		execRes, err = executor.Execute(ctx, e.client, func() {})
		if err == nil {
			return &RemoteChecksum{
				Schema:     tableInfo.DB,
				Table:      tableInfo.Name,
				Checksum:   execRes.Checksum,
				TotalBytes: execRes.TotalBytes,
				TotalKVs:   execRes.TotalKvs,
			}, nil
		}
		log.L().Warn("remote checksum failed", zap.String("db", tableInfo.DB),
			zap.String("table", tableInfo.Name), zap.Error(err),
			zap.Int("concurrency", distSQLScanConcurrency), zap.Int("retry", i))
		if !common.IsRetryableError(err) {
			break
		}
		if distSQLScanConcurrency > minDistSQLScanConcurrency {
			distSQLScanConcurrency = utils.MaxInt(distSQLScanConcurrency/2, minDistSQLScanConcurrency)
		}
	}
	return nil, err
}
func (e *tikvChecksumManager) Checksum(ctx context.Context, tableInfo *TidbTableInfo) (*RemoteChecksum, error) {
	tbl := common.UniqueTable(tableInfo.DB, tableInfo.Name)
	err := e.manager.addOneJob(ctx, tbl, oracle.ComposeTS(time.Now().Unix()*1000, 0))
	if err != nil {
		return nil, errors.Trace(err)
	}
	return e.checksumDB(ctx, tableInfo)
}
type tableChecksumTS struct {
	table    string
	gcSafeTS uint64
}