sessionctx: add sysvar to control low resolution tso update interval (p…

…ingcap#51437)

close pingcap#51081

pkg/domain/domain_sysvars.go

@@ -39,6 +39,7 @@ func (do *Domain) initDomainSysVars() {
 
 	setGlobalResourceControlFunc := do.setGlobalResourceControl
 	variable.SetGlobalResourceControl.Store(&setGlobalResourceControlFunc)
+	variable.SetLowResolutionTSOUpdateInterval = do.setLowResolutionTSOUpdateInterval
 }
 
 // setStatsCacheCapacity sets statsCache cap
@@ -90,6 +91,10 @@ func (*Domain) setGlobalResourceControl(enable bool) {
 	}
 }
 
+func (do *Domain) setLowResolutionTSOUpdateInterval(interval time.Duration) error {
+	return do.store.GetOracle().SetLowResolutionTimestampUpdateInterval(interval)
+}
+
 // updatePDClient is used to set the dynamic option into the PD client.
 func (do *Domain) updatePDClient(option pd.DynamicOption, val any) error {
 	store, ok := do.store.(interface{ GetPDClient() pd.Client })

pkg/sessionctx/variable/sysvar.go

@@ -1385,6 +1385,16 @@ var defaultSysVars = []*SysVar{
 			HistoricalStatsDuration.Store(d)
 			return nil
 		}},
+	{Scope: ScopeGlobal, Name: TiDBLowResolutionTSOUpdateInterval, Value: strconv.Itoa(DefTiDBLowResolutionTSOUpdateInterval), Type: TypeInt, MinValue: 10, MaxValue: 60000,
+		SetGlobal: func(_ context.Context, s *SessionVars, val string) error {
+			LowResolutionTSOUpdateInterval.Store(uint32(TidbOptInt64(val, DefTiDBLowResolutionTSOUpdateInterval)))
+			if SetLowResolutionTSOUpdateInterval != nil {
+				interval := time.Duration(LowResolutionTSOUpdateInterval.Load()) * time.Millisecond
+				return SetLowResolutionTSOUpdateInterval(interval)
+			}
+			return nil
+		},
+	},
 
 	/* The system variables below have GLOBAL and SESSION scope  */
 	{Scope: ScopeGlobal | ScopeSession, Name: TiDBEnablePlanReplayerContinuousCapture, Value: BoolToOnOff(false), Type: TypeBool,

pkg/sessionctx/variable/sysvar_test.go

@@ -1526,3 +1526,27 @@ func TestTiDBOptTxnAutoRetry(t *testing.T) {
 		require.Equal(t, "[variable:1287]'OFF' is deprecated and will be removed in a future release. Please use ON instead", warn.Error())
 	}
 }
+
+func TestTiDBLowResTSOUpdateInterval(t *testing.T) {
+	sv := GetSysVar(TiDBLowResolutionTSOUpdateInterval)
+	vars := NewSessionVars(nil)
+
+	// Too low, will get raised to the min value
+	val, err := sv.Validate(vars, "0", ScopeGlobal)
+	require.NoError(t, err)
+	require.Equal(t, strconv.FormatInt(GetSysVar(TiDBLowResolutionTSOUpdateInterval).MinValue, 10), val)
+	warn := vars.StmtCtx.GetWarnings()[0].Err
+	require.Equal(t, "[variable:1292]Truncated incorrect tidb_low_resolution_tso_update_interval value: '0'", warn.Error())
+
+	// Too high, will get lowered to the max value
+	val, err = sv.Validate(vars, "100000", ScopeGlobal)
+	require.NoError(t, err)
+	require.Equal(t, strconv.FormatUint(GetSysVar(TiDBLowResolutionTSOUpdateInterval).MaxValue, 10), val)
+	warn = vars.StmtCtx.GetWarnings()[1].Err
+	require.Equal(t, "[variable:1292]Truncated incorrect tidb_low_resolution_tso_update_interval value: '100000'", warn.Error())
+
+	// valid
+	val, err = sv.Validate(vars, "1000", ScopeGlobal)
+	require.NoError(t, err)
+	require.Equal(t, "1000", val)
+}

pkg/sessionctx/variable/tidb_vars.go

@@ -1152,6 +1152,8 @@ const (
 	TiDBIdleTransactionTimeout = "tidb_idle_transaction_timeout"
 	// TiDBEnableParallelSort indicates if parallel sort is enabled.
 	TiDBEnableParallelSort = "enable_parallel_sort"
+	// TiDBLowResolutionTSOUpdateInterval defines how often to refresh low resolution timestamps.
+	TiDBLowResolutionTSOUpdateInterval = "tidb_low_resolution_tso_update_interval"
 )
 
 // TiDB intentional limits
@@ -1477,6 +1479,7 @@ const (
 	DefEnableParallelSort                             = false
 	DefTiDBTxnEntrySizeLimit                          = 0
 	DefTiDBSchemaCacheSize                            = 0
+	DefTiDBLowResolutionTSOUpdateInterval             = 2000
 )
 
 // Process global variables.
@@ -1535,7 +1538,8 @@ var (
 	EnableForeignKey    = atomic.NewBool(true)
 	EnableRCReadCheckTS = atomic.NewBool(false)
 	// EnableRowLevelChecksum indicates whether to append checksum to row values.
-	EnableRowLevelChecksum = atomic.NewBool(DefTiDBEnableRowLevelChecksum)
+	EnableRowLevelChecksum         = atomic.NewBool(DefTiDBEnableRowLevelChecksum)
+	LowResolutionTSOUpdateInterval = atomic.NewUint32(DefTiDBLowResolutionTSOUpdateInterval)
 
 	// DefTiDBServerMemoryLimit indicates the default value of TiDBServerMemoryLimit(TotalMem * 80%).
 	// It should be a const and shouldn't be modified after tidb is started.
@@ -1610,6 +1614,8 @@ var (
 	SetGlobalResourceControl atomic.Pointer[func(bool)]
 	// ValidateCloudStorageURI validates the cloud storage URI.
 	ValidateCloudStorageURI func(ctx context.Context, uri string) error
+	// SetLowResolutionTSOUpdateInterval is the func registered by domain to set slow resolution tso update interval.
+	SetLowResolutionTSOUpdateInterval func(interval time.Duration) error = nil
 )
 
 // Hooks functions for Cluster Resource Control.