infoschema, util: refresh statement summary table periodically (#13680)

domain/global_vars_cache.go

@@ -76,6 +76,12 @@ func checkEnableServerGlobalVar(rows []chunk.Row) {
 				sVal = row.GetString(1)
 			}
 			stmtsummary.StmtSummaryByDigestMap.SetEnabled(sVal, false)
+		case variable.TiDBStmtSummaryRefreshInterval:
+			sVal := ""
+			if !row.IsNull(1) {
+				sVal = row.GetString(1)
+			}
+			stmtsummary.StmtSummaryByDigestMap.SetRefreshInterval(sVal, false)
 		case variable.TiDBCapturePlanBaseline:
 			sVal := ""
 			if !row.IsNull(1) {

executor/set.go

@@ -197,9 +197,12 @@ func (e *SetExecutor) setSysVariable(name string, v *expression.VarAssignment) e
 		}
 	}
 
-	if name == variable.TiDBEnableStmtSummary {
+	switch name {
+	case variable.TiDBEnableStmtSummary:
 		stmtsummary.StmtSummaryByDigestMap.SetEnabled(valStr, !v.IsGlobal)
-	} else if name == variable.TiDBCapturePlanBaseline {
+	case variable.TiDBStmtSummaryRefreshInterval:
+		stmtsummary.StmtSummaryByDigestMap.SetRefreshInterval(valStr, !v.IsGlobal)
+	case variable.TiDBCapturePlanBaseline:
 		variable.CapturePlanBaseline.Set(valStr, !v.IsGlobal)
 	}
 

infoschema/perfschema/const.go

@@ -386,6 +386,7 @@ const tableStagesHistoryLong = "CREATE TABLE if not exists performance_schema.ev
 // tableEventsStatementsSummaryByDigest contains the column name definitions for table
 // events_statements_summary_by_digest, same as MySQL.
 const tableEventsStatementsSummaryByDigest = "CREATE TABLE if not exists events_statements_summary_by_digest (" +
+	"SUMMARY_BEGIN_TIME TIMESTAMP(6) NOT NULL," +
 	"STMT_TYPE VARCHAR(64) NOT NULL," +
 	"SCHEMA_NAME VARCHAR(64) DEFAULT NULL," +
 	"DIGEST VARCHAR(64) NOT NULL," +

infoschema/perfschema/tables_test.go

@@ -74,14 +74,14 @@ func (s *testTableSuite) TestPerfSchemaTables(c *C) {
 	tk.MustQuery("select * from events_stages_history_long").Check(testkit.Rows())
 }
 
-// Test events_statements_summary_by_digest
+// Test events_statements_summary_by_digest.
 func (s *testTableSuite) TestStmtSummaryTable(c *C) {
 	tk := testkit.NewTestKitWithInit(c, s.store)
 
 	tk.MustExec("drop table if exists t")
 	tk.MustExec("create table t(a int, b varchar(10), key k(a))")
 
-	// Statement summary is disabled by default
+	// Statement summary is disabled by default.
 	tk.MustQuery("select @@global.tidb_enable_stmt_summary").Check(testkit.Rows("0"))
 	tk.MustExec("insert into t values(1, 'a')")
 	tk.MustQuery("select * from performance_schema.events_statements_summary_by_digest").Check(testkit.Rows())
@@ -91,8 +91,11 @@ func (s *testTableSuite) TestStmtSummaryTable(c *C) {
 
 	// Invalidate the cache manually so that tidb_enable_stmt_summary works immediately.
 	s.dom.GetGlobalVarsCache().Disable()
+	// Disable refreshing summary.
+	tk.MustExec("set global tidb_stmt_summary_refresh_interval = 999999999")
+	tk.MustQuery("select @@global.tidb_stmt_summary_refresh_interval").Check(testkit.Rows("999999999"))
 
-	// Create a new session to test
+	// Create a new session to test.
 	tk = testkit.NewTestKitWithInit(c, s.store)
 
 	// Test INSERT
@@ -107,7 +110,7 @@ func (s *testTableSuite) TestStmtSummaryTable(c *C) {
 		where digest_text like 'insert into t%'`,
 	).Check(testkit.Rows("insert test test.t <nil> 4 0 0 0 0 0 2 2 1 1 1 /**/insert into t values(4, 'd')"))
 
-	// Test SELECT
+	// Test SELECT.
 	tk.MustQuery("select * from t where a=2")
 	tk.MustQuery(`select stmt_type, schema_name, table_names, index_names, exec_count, cop_task_num, avg_total_keys, 
 		max_total_keys, avg_processed_keys, max_processed_keys, avg_write_keys, max_write_keys, avg_prewrite_regions, 
@@ -124,26 +127,26 @@ func (s *testTableSuite) TestStmtSummaryTable(c *C) {
 		order by exec_count desc limit 1`,
 	).Check(testkit.Rows("insert test test.t <nil> 4 0 0 0 0 0 2 2 1 1 1 /**/insert into t values(4, 'd')"))
 
-	// Disable it again
+	// Disable it again.
 	tk.MustExec("set global tidb_enable_stmt_summary = false")
 	tk.MustQuery("select @@global.tidb_enable_stmt_summary").Check(testkit.Rows("0"))
 
 	// Create a new session to test
 	tk = testkit.NewTestKitWithInit(c, s.store)
 
-	// This statement shouldn't be summarized
+	// This statement shouldn't be summarized.
 	tk.MustQuery("select * from t where a=2")
 
-	// The table should be cleared
+	// The table should be cleared.
 	tk.MustQuery(`select stmt_type, schema_name, table_names, index_names, exec_count, cop_task_num, avg_total_keys, 
 		max_total_keys, avg_processed_keys, max_processed_keys, avg_write_keys, max_write_keys, avg_prewrite_regions, 
 		max_prewrite_regions, avg_affected_rows, query_sample_text 
 		from performance_schema.events_statements_summary_by_digest`,
 	).Check(testkit.Rows())
 
-	// Enable it in session scope
+	// Enable it in session scope.
 	tk.MustExec("set session tidb_enable_stmt_summary = on")
-	// It should work immediately
+	// It should work immediately.
 	tk.MustExec("begin")
 	tk.MustExec("insert into t values(1, 'a')")
 	tk.MustExec("commit")
@@ -168,27 +171,27 @@ func (s *testTableSuite) TestStmtSummaryTable(c *C) {
 		where digest_text like 'select * from t%'`,
 	).Check(testkit.Rows("select test test.t t:k 1 2 0 0 0 0 0 0 0 0 0 select * from t where a=2"))
 
-	// Disable it in global scope
+	// Disable it in global scope.
 	tk.MustExec("set global tidb_enable_stmt_summary = off")
 
-	// Create a new session to test
+	// Create a new session to test.
 	tk = testkit.NewTestKitWithInit(c, s.store)
 
 	tk.MustQuery("select * from t where a=2")
 
-	// Statement summary is still enabled
+	// Statement summary is still enabled.
 	tk.MustQuery(`select stmt_type, schema_name, table_names, index_names, exec_count, cop_task_num, avg_total_keys, 
 		max_total_keys, avg_processed_keys, max_processed_keys, avg_write_keys, max_write_keys, avg_prewrite_regions, 
 		max_prewrite_regions, avg_affected_rows, query_sample_text 
 		from performance_schema.events_statements_summary_by_digest
 		where digest_text like 'select * from t%'`,
 	).Check(testkit.Rows("select test test.t t:k 2 4 0 0 0 0 0 0 0 0 0 select * from t where a=2"))
 
-	// Unset session variable
+	// Unset session variable.
 	tk.MustExec("set session tidb_enable_stmt_summary = ''")
 	tk.MustQuery("select * from t where a=2")
 
-	// Statement summary is disabled
+	// Statement summary is disabled.
 	tk.MustQuery(`select stmt_type, schema_name, table_names, index_names, exec_count, cop_task_num, avg_total_keys, 
 		max_total_keys, avg_processed_keys, max_processed_keys, avg_write_keys, max_write_keys, avg_prewrite_regions, 
 		max_prewrite_regions, avg_affected_rows, query_sample_text 

session/session.go

@@ -1851,6 +1851,7 @@ var builtinGlobalVariable = []string{
 	variable.TiDBEnableIndexMerge,
 	variable.TiDBTxnMode,
 	variable.TiDBEnableStmtSummary,
+	variable.TiDBStmtSummaryRefreshInterval,
 	variable.TiDBMaxDeltaSchemaCount,
 	variable.TiDBCapturePlanBaseline,
 	variable.TiDBUsePlanBaselines,

sessionctx/variable/sysvar.go

@@ -717,6 +717,7 @@ var defaultSysVars = []*SysVar{
 	{ScopeSession, TiDBReplicaRead, "leader"},
 	{ScopeSession, TiDBAllowRemoveAutoInc, BoolToIntStr(DefTiDBAllowRemoveAutoInc)},
 	{ScopeGlobal | ScopeSession, TiDBEnableStmtSummary, "0"},
+	{ScopeGlobal | ScopeSession, TiDBStmtSummaryRefreshInterval, strconv.Itoa(DefTiDBStmtSummaryRefreshInterval)},
 	{ScopeGlobal | ScopeSession, TiDBCapturePlanBaseline, "0"},
 	{ScopeGlobal | ScopeSession, TiDBUsePlanBaselines, BoolToIntStr(DefTiDBUsePlanBaselines)},
 	{ScopeGlobal | ScopeSession, TiDBEvolvePlanBaselines, BoolToIntStr(DefTiDBEvolvePlanBaselines)},

sessionctx/variable/tidb_vars.go

@@ -336,6 +336,9 @@ const (
 	// TiDBEnableStmtSummary indicates whether the statement summary is enabled.
 	TiDBEnableStmtSummary = "tidb_enable_stmt_summary"
 
+	// TiDBStmtSummaryRefreshInterval indicates the refresh interval in seconds for each statement summary.
+	TiDBStmtSummaryRefreshInterval = "tidb_stmt_summary_refresh_interval"
+
 	// TiDBCapturePlanBaseline indicates whether the capture of plan baselines is enabled.
 	TiDBCapturePlanBaseline = "tidb_capture_plan_baselines"
 
@@ -426,6 +429,7 @@ const (
 	DefWaitSplitRegionTimeout          = 300 // 300s
 	DefTiDBEnableNoopFuncs             = false
 	DefTiDBAllowRemoveAutoInc          = false
+	DefTiDBStmtSummaryRefreshInterval  = 1800 // 1800s
 	DefTiDBUsePlanBaselines            = true
 	DefTiDBEvolvePlanBaselines         = false
 	DefTiDBEvolvePlanTaskMaxTime       = 600 // 600s

sessionctx/variable/varsutil.go

@@ -629,6 +629,11 @@ func ValidateSetSystemVar(vars *SessionVars, name string, value string) (string,
 			return "", nil
 		}
 		return value, ErrWrongValueForVar.GenWithStackByArgs(name, value)
+	case TiDBStmtSummaryRefreshInterval:
+		if value == "" {
+			return "", nil
+		}
+		return checkUInt64SystemVar(name, value, 1, math.MaxUint32, vars)
 	case TiDBIsolationReadEngines:
 		engines := strings.Split(value, ",")
 		var formatVal string

sessionctx/variable/varsutil_test.go

@@ -370,6 +370,16 @@ func (s *testVarsutilSuite) TestVarsutil(c *C) {
 	c.Assert(err, IsNil)
 	c.Assert(val, Equals, "leader")
 	c.Assert(v.GetReplicaRead(), Equals, kv.ReplicaReadLeader)
+
+	SetSessionSystemVar(v, TiDBEnableStmtSummary, types.NewStringDatum("on"))
+	val, err = GetSessionSystemVar(v, TiDBEnableStmtSummary)
+	c.Assert(err, IsNil)
+	c.Assert(val, Equals, "1")
+
+	SetSessionSystemVar(v, TiDBStmtSummaryRefreshInterval, types.NewStringDatum("10"))
+	val, err = GetSessionSystemVar(v, TiDBStmtSummaryRefreshInterval)
+	c.Assert(err, IsNil)
+	c.Assert(val, Equals, "10")
 }
 
 func (s *testVarsutilSuite) TestSetOverflowBehave(c *C) {
@@ -463,6 +473,11 @@ func (s *testVarsutilSuite) TestValidate(c *C) {
 		{TiDBIsolationReadEngines, "tikv", false},
 		{TiDBIsolationReadEngines, "TiKV,tiflash", false},
 		{TiDBIsolationReadEngines, "   tikv,   tiflash  ", false},
+		{TiDBEnableStmtSummary, "a", true},
+		{TiDBEnableStmtSummary, "-1", true},
+		{TiDBEnableStmtSummary, "", false},
+		{TiDBStmtSummaryRefreshInterval, "a", true},
+		{TiDBStmtSummaryRefreshInterval, "", false},
 	}
 
 	for _, t := range tests {