*: adjust delta schema count and add metrics

domain/schema_validator.go

@@ -17,6 +17,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/pingcap/tidb/metrics"
+	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/store/tikv/oracle"
 	"github.com/pingcap/tidb/util/logutil"
 	"go.uber.org/zap"
@@ -65,14 +67,16 @@ type schemaValidator struct {
 	latestSchemaExpire time.Time
 	// deltaSchemaInfos is a queue that maintain the history of changes.
 	deltaSchemaInfos []deltaSchemaInfo
+	// notMergeCnt is used to record the number of deltaSchemaInfo that have not been merged.
+	notMergeCnt int
 }
 
 // NewSchemaValidator returns a SchemaValidator structure.
 func NewSchemaValidator(lease time.Duration) SchemaValidator {
 	return &schemaValidator{
 		isStarted:        true,
 		lease:            lease,
-		deltaSchemaInfos: make([]deltaSchemaInfo, 0, maxNumberOfDiffsToLoad),
+		deltaSchemaInfos: make([]deltaSchemaInfo, 0, variable.DefTiDBMaxDeltaSchemaCount),
 	}
 }
 
@@ -85,26 +89,31 @@ func (s *schemaValidator) IsStarted() bool {
 
 func (s *schemaValidator) Stop() {
 	logutil.BgLogger().Info("the schema validator stops")
+	metrics.LoadSchemaCounter.WithLabelValues(metrics.SchemaValidatorStop).Inc()
 	s.mux.Lock()
 	defer s.mux.Unlock()
 	s.isStarted = false
 	s.latestSchemaVer = 0
-	s.deltaSchemaInfos = make([]deltaSchemaInfo, 0, maxNumberOfDiffsToLoad)
+	s.notMergeCnt = 0
+	s.deltaSchemaInfos = s.deltaSchemaInfos[:0]
 }
 
 func (s *schemaValidator) Restart() {
+	metrics.LoadSchemaCounter.WithLabelValues(metrics.SchemaValidatorRestart).Inc()
 	logutil.BgLogger().Info("the schema validator restarts")
 	s.mux.Lock()
 	defer s.mux.Unlock()
 	s.isStarted = true
 }
 
 func (s *schemaValidator) Reset() {
+	metrics.LoadSchemaCounter.WithLabelValues(metrics.SchemaValidatorReset).Inc()
 	s.mux.Lock()
 	defer s.mux.Unlock()
 	s.isStarted = true
 	s.latestSchemaVer = 0
-	s.deltaSchemaInfos = make([]deltaSchemaInfo, 0, maxNumberOfDiffsToLoad)
+	s.notMergeCnt = 0
+	s.deltaSchemaInfos = s.deltaSchemaInfos[:0]
 }
 
 func (s *schemaValidator) Update(leaseGrantTS uint64, oldVer, currVer int64, changedTableIDs []int64) {
@@ -146,13 +155,15 @@ func hasRelatedTableID(relatedTableIDs, updateTableIDs []int64) bool {
 // NOTE, this function should be called under lock!
 func (s *schemaValidator) isRelatedTablesChanged(currVer int64, tableIDs []int64) bool {
 	if len(s.deltaSchemaInfos) == 0 {
+		metrics.LoadSchemaCounter.WithLabelValues(metrics.SchemaValidatorCacheEmpty).Inc()
 		logutil.BgLogger().Info("schema change history is empty", zap.Int64("currVer", currVer))
 		return true
 	}
 	newerDeltas := s.findNewerDeltas(currVer)
 	if len(newerDeltas) == len(s.deltaSchemaInfos) {
+		metrics.LoadSchemaCounter.WithLabelValues(metrics.SchemaValidatorCacheMiss).Inc()
 		logutil.BgLogger().Info("the schema version is much older than the latest version", zap.Int64("currVer", currVer),
-			zap.Int64("latestSchemaVer", s.latestSchemaVer))
+			zap.Int64("latestSchemaVer", s.latestSchemaVer), zap.Reflect("deltas", newerDeltas))
 		return true
 	}
 	for _, item := range newerDeltas {
@@ -209,7 +220,35 @@ func (s *schemaValidator) Check(txnTS uint64, schemaVer int64, relatedTableIDs [
 
 func (s *schemaValidator) enqueue(schemaVersion int64, relatedTableIDs []int64) {
 	s.deltaSchemaInfos = append(s.deltaSchemaInfos, deltaSchemaInfo{schemaVersion, relatedTableIDs})
-	if len(s.deltaSchemaInfos) > maxNumberOfDiffsToLoad {
+	s.notMergeCnt++
+
+	maxCnt := int(variable.GetMaxDetalSchemaCount())
+	if len(s.deltaSchemaInfos) > maxCnt && s.notMergeCnt > maxCnt/2 {
+		s.merge()
+		s.notMergeCnt = 1
+	}
+	if len(s.deltaSchemaInfos) > maxCnt {
 		s.deltaSchemaInfos = s.deltaSchemaInfos[1:]
 	}
 }
+
+func equal(a, b []int64) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := 0; i < len(a); i++ {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+func (s *schemaValidator) merge() {
+	// The first item we needn't to merge, because we hope to cover more versions.
+	for i := len(s.deltaSchemaInfos) - 1; i > 1; i-- {
+		if equal(s.deltaSchemaInfos[i].relatedTableIDs, s.deltaSchemaInfos[i-1].relatedTableIDs) {
+			s.deltaSchemaInfos = append(s.deltaSchemaInfos[:i-1], s.deltaSchemaInfos[i:]...)
+		}
+	}
+}

domain/schema_validator_test.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	. "github.com/pingcap/check"
+	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/store/tikv/oracle"
 	"github.com/pingcap/tidb/util/testleak"
 )
@@ -143,3 +144,89 @@ func serverFunc(lease time.Duration, requireLease chan leaseGrantItem, oracleCh
 		}
 	}
 }
+
+func (*testSuite) TestMerge(c *C) {
+	lease := 10 * time.Millisecond
+	originalCnt := variable.GetMaxDetalSchemaCount()
+	variable.SetMaxDetalSchemaCount(10)
+	defer variable.SetMaxDetalSchemaCount(originalCnt)
+
+	validator := NewSchemaValidator(lease).(*schemaValidator)
+	c.Assert(validator.IsStarted(), IsTrue)
+	ds := []deltaSchemaInfo{
+		{0, []int64{1}},
+		{1, []int64{1}},
+		{2, []int64{1}},
+		{3, []int64{1, 2}},
+		{4, []int64{1}},
+		{5, []int64{1, 3}},
+		{6, []int64{1, 3}},
+		{7, []int64{1, 3}},
+		{8, []int64{1, 2, 3}},
+		{9, []int64{1, 2, 3}},
+	}
+	for _, d := range ds {
+		validator.enqueue(d.schemaVersion, d.relatedTableIDs)
+	}
+	validator.enqueue(10, []int64{1})
+	ret := []deltaSchemaInfo{
+		{0, []int64{1}},
+		{2, []int64{1}},
+		{3, []int64{1, 2}},
+		{4, []int64{1}},
+		{7, []int64{1, 3}},
+		{9, []int64{1, 2, 3}},
+		{10, []int64{1}},
+	}
+	c.Assert(validator.deltaSchemaInfos, DeepEquals, ret)
+	// notMergeCnt <= max deltal schema count
+	validator.enqueue(11, []int64{1})
+	ret = append(ret, deltaSchemaInfo{11, []int64{1}})
+	validator.enqueue(12, []int64{1})
+	ret = append(ret, deltaSchemaInfo{12, []int64{1}})
+	validator.enqueue(13, []int64{1})
+	ret = append(ret, deltaSchemaInfo{13, []int64{1}})
+	c.Assert(validator.deltaSchemaInfos, DeepEquals, ret)
+	validator.enqueue(14, []int64{1})
+	ret = append(ret, deltaSchemaInfo{14, []int64{1}})
+	c.Assert(validator.deltaSchemaInfos, DeepEquals, ret[1:])
+	// notMergeCnt > max deltal schema count
+	validator.enqueue(15, []int64{1})
+	ret = []deltaSchemaInfo{
+		{2, []int64{1}},
+		{3, []int64{1, 2}},
+		{4, []int64{1}},
+		{7, []int64{1, 3}},
+		{9, []int64{1, 2, 3}},
+		{15, []int64{1}},
+	}
+	c.Assert(validator.deltaSchemaInfos, DeepEquals, ret)
+
+	validator.notMergeCnt = 0
+	validator.deltaSchemaInfos = []deltaSchemaInfo{
+		{0, []int64{1, 2, 3}},
+		{1, []int64{2}},
+		{2, []int64{3}},
+		{3, []int64{1, 2, 3}},
+		{4, []int64{1, 4}},
+		{5, []int64{1, 5}},
+		{6, []int64{1, 6}},
+		{7, []int64{1, 7}},
+		{8, []int64{1, 2, 3}},
+		{9, []int64{1, 2}},
+	}
+	validator.enqueue(10, []int64{1})
+	ret = []deltaSchemaInfo{
+		{1, []int64{2}},
+		{2, []int64{3}},
+		{3, []int64{1, 2, 3}},
+		{4, []int64{1, 4}},
+		{5, []int64{1, 5}},
+		{6, []int64{1, 6}},
+		{7, []int64{1, 7}},
+		{8, []int64{1, 2, 3}},
+		{9, []int64{1, 2}},
+		{10, []int64{1}},
+	}
+	c.Assert(validator.deltaSchemaInfos, DeepEquals, ret)
+}

executor/seqtest/seq_executor_test.go

@@ -36,6 +36,7 @@ import (
 	pb "github.com/pingcap/kvproto/pkg/kvrpcpb"
 	"github.com/pingcap/parser"
 	"github.com/pingcap/parser/model"
+	"github.com/pingcap/parser/terror"
 	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/domain"
 	"github.com/pingcap/tidb/executor"
@@ -1053,3 +1054,31 @@ func (s *seqTestSuite) TestAutoIDInRetry(c *C) {
 	tk.MustExec("insert into t values ()")
 	tk.MustQuery(`select * from t`).Check(testkit.Rows("1", "2", "3", "4", "5"))
 }
+
+func (s *seqTestSuite) TestMaxDetalSchemaCount(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	c.Assert(variable.GetMaxDetalSchemaCount(), Equals, int64(variable.DefTiDBMaxDeltaSchemaCount))
+	gvc := domain.GetDomain(tk.Se).GetGlobalVarsCache()
+	gvc.Disable()
+
+	tk.MustExec("set @@global.tidb_max_delta_schema_count= -1")
+	tk.MustQuery("show warnings;").Check(testkit.Rows("Warning 1292 Truncated incorrect tidb_max_delta_schema_count value: '-1'"))
+	// Make sure a new session will load global variables.
+	tk.Se = nil
+	tk.MustExec("use test")
+	c.Assert(variable.GetMaxDetalSchemaCount(), Equals, int64(100))
+	tk.MustExec(fmt.Sprintf("set @@global.tidb_max_delta_schema_count= %v", uint64(math.MaxInt64)))
+	tk.MustQuery("show warnings;").Check(testkit.Rows(fmt.Sprintf("Warning 1292 Truncated incorrect tidb_max_delta_schema_count value: '%d'", uint64(math.MaxInt64))))
+	tk.Se = nil
+	tk.MustExec("use test")
+	c.Assert(variable.GetMaxDetalSchemaCount(), Equals, int64(16384))
+	_, err := tk.Exec("set @@global.tidb_max_delta_schema_count= invalid_val")
+	c.Assert(terror.ErrorEqual(err, variable.ErrWrongTypeForVar), IsTrue, Commentf("err %v", err))
+
+	tk.MustExec("set @@global.tidb_max_delta_schema_count= 2048")
+	tk.Se = nil
+	tk.MustExec("use test")
+	c.Assert(variable.GetMaxDetalSchemaCount(), Equals, int64(2048))
+	tk.MustQuery("select @@global.tidb_max_delta_schema_count").Check(testkit.Rows("2048"))
+}

metrics/domain.go

@@ -17,6 +17,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
+// Metrics for the domain package.
 var (
 	// LoadSchemaCounter records the counter of load schema.
 	LoadSchemaCounter = prometheus.NewCounterVec(
@@ -45,4 +46,18 @@ var (
 			Name:      "load_privilege_total",
 			Help:      "Counter of load privilege",
 		}, []string{LblType})
+
+	SchemaValidatorStop       = "stop"
+	SchemaValidatorRestart    = "restart"
+	SchemaValidatorReset      = "reset"
+	SchemaValidatorCacheEmpty = "cache_empty"
+	SchemaValidatorCacheMiss  = "cache_miss"
+	// HandleSchemaValidate records the counter of handling schema validate.
+	HandleSchemaValidate = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "tidb",
+			Subsystem: "domain",
+			Name:      "handle_schema_validate",
+			Help:      "Counter of handle schema validate",
+		}, []string{LblType})
 )

metrics/metrics.go

@@ -152,4 +152,5 @@ func RegisterMetrics() {
 	prometheus.MustRegister(TiKVBatchClientUnavailable)
 	prometheus.MustRegister(TiKVRangeTaskStats)
 	prometheus.MustRegister(TiKVRangeTaskPushDuration)
+	prometheus.MustRegister(HandleSchemaValidate)
 }

session/session.go

@@ -1770,6 +1770,7 @@ var builtinGlobalVariable = []string{
 	variable.TiDBExpensiveQueryTimeThreshold,
 	variable.TiDBEnableNoopFuncs,
 	variable.TiDBEnableIndexMerge,
+	variable.TiDBMaxDeltaSchemaCount,
 }
 
 var (

sessionctx/variable/session.go

@@ -841,6 +841,9 @@ func (s *SessionVars) SetSystemVar(name string, val string) error {
 		} else if strings.EqualFold(val, "leader") || len(val) == 0 {
 			s.ReplicaRead = kv.ReplicaReadLeader
 		}
+	// It's a global variable, but it also wants to be cached in server.
+	case TiDBMaxDeltaSchemaCount:
+		SetMaxDetalSchemaCount(tidbOptInt64(val, DefTiDBMaxDeltaSchemaCount))
 	}
 	s.systems[name] = val
 	return nil

sessionctx/variable/sysvar.go

@@ -693,6 +693,7 @@ var defaultSysVars = []*SysVar{
 	{ScopeGlobal, TiDBDDLReorgBatchSize, strconv.Itoa(DefTiDBDDLReorgBatchSize)},
 	{ScopeGlobal, TiDBDDLErrorCountLimit, strconv.Itoa(DefTiDBDDLErrorCountLimit)},
 	{ScopeSession, TiDBDDLReorgPriority, "PRIORITY_LOW"},
+	{ScopeGlobal, TiDBMaxDeltaSchemaCount, strconv.Itoa(DefTiDBMaxDeltaSchemaCount)},
 	{ScopeSession, TiDBForcePriority, mysql.Priority2Str[DefTiDBForcePriority]},
 	{ScopeSession, TiDBEnableRadixJoin, BoolToIntStr(DefTiDBUseRadixJoin)},
 	{ScopeGlobal | ScopeSession, TiDBOptJoinReorderThreshold, strconv.Itoa(DefTiDBOptJoinReorderThreshold)},

sessionctx/variable/tidb_vars.go

@@ -247,6 +247,10 @@ const (
 	// It can be: PRIORITY_LOW, PRIORITY_NORMAL, PRIORITY_HIGH
 	TiDBDDLReorgPriority = "tidb_ddl_reorg_priority"
 
+	// tidb_max_delta_schema_count defines the max length of deltaSchemaInfos.
+	// deltaSchemaInfos is a queue that maintains the history of schema changes.
+	TiDBMaxDeltaSchemaCount = "tidb_max_delta_schema_count"
+
 	// tidb_scatter_region will scatter the regions for DDLs when it is ON.
 	TiDBScatterRegion = "tidb_scatter_region"
 
@@ -339,6 +343,7 @@ const (
 	DefTiDBDDLReorgWorkerCount         = 16
 	DefTiDBDDLReorgBatchSize           = 1024
 	DefTiDBDDLErrorCountLimit          = 512
+	DefTiDBMaxDeltaSchemaCount         = 1024
 	DefTiDBHashAggPartialConcurrency   = 4
 	DefTiDBHashAggFinalConcurrency     = 4
 	DefTiDBForcePriority               = mysql.NoPriority
@@ -362,6 +367,7 @@ var (
 	maxDDLReorgWorkerCount int32 = 128
 	ddlReorgBatchSize      int32 = DefTiDBDDLReorgBatchSize
 	ddlErrorCountlimit     int64 = DefTiDBDDLErrorCountLimit
+	maxDetalSchemaCount    int64 = DefTiDBMaxDeltaSchemaCount
 	// Export for testing.
 	MaxDDLReorgBatchSize int32 = 10240
 	MinDDLReorgBatchSize int32 = 32

sessionctx/variable/varsutil.go

@@ -73,6 +73,16 @@ func GetDDLErrorCountLimit() int64 {
 	return atomic.LoadInt64(&ddlErrorCountlimit)
 }
 
+// SetMaxDetalSchemaCount sets maxDetalSchemaCount size.
+func SetMaxDetalSchemaCount(cnt int64) {
+	atomic.StoreInt64(&maxDetalSchemaCount, cnt)
+}
+
+// GetMaxDetalSchemaCount gets maxDetalSchemaCount size.
+func GetMaxDetalSchemaCount() int64 {
+	return atomic.LoadInt64(&maxDetalSchemaCount)
+}
+
 // GetSessionSystemVar gets a system variable.
 // If it is a session only variable, use the default value defined in code.
 // Returns error if there is no such variable.
@@ -320,6 +330,8 @@ func ValidateSetSystemVar(vars *SessionVars, name string, value string) (string,
 		return checkUInt64SystemVar(name, value, 0, 4294967295, vars)
 	case OldPasswords:
 		return checkUInt64SystemVar(name, value, 0, 2, vars)
+	case TiDBMaxDeltaSchemaCount:
+		return checkInt64SystemVar(name, value, 100, 16384, vars)
 	case SessionTrackGtids:
 		if strings.EqualFold(value, "OFF") || value == "0" {
 			return "OFF", nil