diff --git a/pkg/executor/select.go b/pkg/executor/select.go index 2b5de8c9901df..e5e69d64c8647 100644 --- a/pkg/executor/select.go +++ b/pkg/executor/select.go @@ -1230,7 +1230,9 @@ func ResetContextOfStmt(ctx sessionctx.Context, s ast.StmtNode) (err error) { } sc.LastInsertIDSet = false sc.PrevAffectedRows = 0 - if vars.StmtCtx.InUpdateStmt || vars.StmtCtx.InDeleteStmt || vars.StmtCtx.InInsertStmt || vars.StmtCtx.InSetSessionStatesStmt { + if vars.StmtCtx.InSetSessionStatesStmt { + sc.PrevAffectedRows = vars.StmtCtx.PrevAffectedRows + } else if vars.StmtCtx.InUpdateStmt || vars.StmtCtx.InDeleteStmt || vars.StmtCtx.InInsertStmt { sc.PrevAffectedRows = int64(vars.StmtCtx.AffectedRows()) } else if vars.StmtCtx.InSelectStmt { sc.PrevAffectedRows = -1 diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 547d2f5bba0b7..66eb0cdfbd0db 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -394,6 +394,7 @@ func RegisterMetrics() { // StmtSummary prometheus.MustRegister(StmtSummaryWindowRecordCount) prometheus.MustRegister(StmtSummaryWindowEvictedCount) + prometheus.MustRegister(StmtSummaryEvictedLogCounter) } // Register registers custom collectors. diff --git a/pkg/metrics/metrics_internal_test.go b/pkg/metrics/metrics_internal_test.go index 2c5d5a4203d59..13f8ba0f2cf9c 100644 --- a/pkg/metrics/metrics_internal_test.go +++ b/pkg/metrics/metrics_internal_test.go @@ -35,6 +35,13 @@ func readGaugeValue(t *testing.T, gauge prometheus.Gauge) float64 { return m.GetGauge().GetValue() } +func readCounterValue(t *testing.T, counter prometheus.Counter) float64 { + t.Helper() + m := &dto.Metric{} + require.NoError(t, counter.Write(m)) + return m.GetCounter().GetValue() +} + func countCollectedMetrics(collector prometheus.Collector) int { ch := make(chan prometheus.Metric, 16) collector.Collect(ch) @@ -51,6 +58,7 @@ func TestStmtSummaryMetricLabels(t *testing.T) { InitStmtSummaryMetrics() require.Equal(t, 0, countCollectedMetrics(StmtSummaryWindowRecordCount)) require.Equal(t, 0, countCollectedMetrics(StmtSummaryWindowEvictedCount)) + require.Equal(t, 0, countCollectedMetrics(StmtSummaryEvictedLogCounter)) SetStmtSummaryWindowMetrics(StmtSummaryTypeV1, 3, 1) require.Equal(t, 1, countCollectedMetrics(StmtSummaryWindowRecordCount)) @@ -63,4 +71,10 @@ func TestStmtSummaryMetricLabels(t *testing.T) { require.Equal(t, 2, countCollectedMetrics(StmtSummaryWindowEvictedCount)) require.Equal(t, 5.0, readGaugeValue(t, StmtSummaryWindowRecordCount.WithLabelValues(StmtSummaryTypeV2))) require.Equal(t, 2.0, readGaugeValue(t, StmtSummaryWindowEvictedCount.WithLabelValues(StmtSummaryTypeV2))) + + StmtSummaryEvictedLogCounter.WithLabelValues(StmtSummaryTypeV2, StmtSummaryEvictedLogResultPersisted).Add(3) + StmtSummaryEvictedLogCounter.WithLabelValues(StmtSummaryTypeV2, StmtSummaryEvictedLogResultDropped).Inc() + require.Equal(t, 2, countCollectedMetrics(StmtSummaryEvictedLogCounter)) + require.Equal(t, 3.0, readCounterValue(t, StmtSummaryEvictedLogCounter.WithLabelValues(StmtSummaryTypeV2, StmtSummaryEvictedLogResultPersisted))) + require.Equal(t, 1.0, readCounterValue(t, StmtSummaryEvictedLogCounter.WithLabelValues(StmtSummaryTypeV2, StmtSummaryEvictedLogResultDropped))) } diff --git a/pkg/metrics/stmtsummary.go b/pkg/metrics/stmtsummary.go index cd5306dac1760..28edda4c91041 100644 --- a/pkg/metrics/stmtsummary.go +++ b/pkg/metrics/stmtsummary.go @@ -27,6 +27,11 @@ const ( StmtSummaryTypeV1 = "v1" // StmtSummaryTypeV2 marks metrics reported by the persistent statement summary implementation. StmtSummaryTypeV2 = "v2" + + // StmtSummaryEvictedLogResultPersisted marks evicted records submitted to the stmt log. + StmtSummaryEvictedLogResultPersisted = "persisted" + // StmtSummaryEvictedLogResultDropped marks evicted records dropped before reaching the stmt log. + StmtSummaryEvictedLogResultDropped = "dropped" ) var ( @@ -39,6 +44,9 @@ var ( // This value resets to 0 when the window rotates. StmtSummaryWindowEvictedCount *prometheus.GaugeVec + // StmtSummaryEvictedLogCounter counts v2 evicted-log persistence outcomes. + StmtSummaryEvictedLogCounter *prometheus.CounterVec + stmtSummaryWindowRecordCountV1 prometheus.Gauge stmtSummaryWindowRecordCountV2 prometheus.Gauge stmtSummaryWindowEvictedCountV1 prometheus.Gauge @@ -65,6 +73,14 @@ func InitStmtSummaryMetrics() { Help: "The number of LRU evictions in the current statement summary window.", }, []string{LblType}) + StmtSummaryEvictedLogCounter = metricscommon.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "tidb", + Subsystem: "stmt_summary", + Name: "evicted_log_total", + Help: "The number of v2 statement summary evicted-log records by result.", + }, []string{LblType, LblResult}) + stmtSummaryWindowMetricsMu.Lock() stmtSummaryWindowRecordCountV1 = nil stmtSummaryWindowRecordCountV2 = nil diff --git a/pkg/sessionctx/sessionstates/session_states_test.go b/pkg/sessionctx/sessionstates/session_states_test.go index 82deb9bf3f77d..c7e9dd9c92f20 100644 --- a/pkg/sessionctx/sessionstates/session_states_test.go +++ b/pkg/sessionctx/sessionstates/session_states_test.go @@ -688,6 +688,7 @@ func TestStatementCtx(t *testing.T) { return nil }, checkFunc: func(tk *testkit.TestKit, param any) { + require.Equal(t, uint64(0), tk.Session().AffectedRows()) tk.MustQuery("select row_count()").Check(testkit.Rows("-1")) }, }, diff --git a/pkg/sessionctx/vardef/tidb_vars.go b/pkg/sessionctx/vardef/tidb_vars.go index fbd7e5ba29995..e6cdef4ce0d02 100644 --- a/pkg/sessionctx/vardef/tidb_vars.go +++ b/pkg/sessionctx/vardef/tidb_vars.go @@ -709,6 +709,16 @@ const ( // TiDBStmtSummaryMaxSQLLength indicates the max length of displayed normalized sql and sample sql. TiDBStmtSummaryMaxSQLLength = "tidb_stmt_summary_max_sql_length" + // TiDBStmtSummaryPersistEvicted controls whether per-record LRU evictions + // in the v2 (persistent) statement summary are persisted to the stmt log. + // Off by default because it adds log volume proportional to eviction rate. + TiDBStmtSummaryPersistEvicted = "tidb_stmt_summary_persist_evicted" + + // TiDBStmtSummaryGroupByUser, when enabled, adds the executing user to the + // statement summary grouping key so the same digest run by different users + // produces separate rows. Off by default to avoid cardinality growth. + TiDBStmtSummaryGroupByUser = "tidb_stmt_summary_group_by_user" + // TiDBIgnoreInlistPlanDigest enables TiDB to generate the same plan digest with SQL using different in-list arguments. TiDBIgnoreInlistPlanDigest = "tidb_ignore_inlist_plan_digest" @@ -1598,6 +1608,8 @@ const ( DefTiDBStmtSummaryHistorySize = 24 DefTiDBStmtSummaryMaxStmtCount = 3000 DefTiDBStmtSummaryMaxSQLLength = 32768 + DefTiDBStmtSummaryPersistEvicted = false + DefTiDBStmtSummaryGroupByUser = false DefTiDBCapturePlanBaseline = Off DefTiDBIgnoreInlistPlanDigest = true DefTiDBEnableIndexMerge = true diff --git a/pkg/sessionctx/variable/session.go b/pkg/sessionctx/variable/session.go index 90107e5d227e6..0e66bf8718ebf 100644 --- a/pkg/sessionctx/variable/session.go +++ b/pkg/sessionctx/variable/session.go @@ -3169,7 +3169,7 @@ func (s *SessionVars) DecodeSessionStates(_ context.Context, sessionStates *sess s.HypoTiFlashReplicas = sessionStates.HypoTiFlashReplicas // Decode StatementContext. - s.StmtCtx.SetAffectedRows(uint64(sessionStates.LastAffectedRows)) + s.StmtCtx.PrevAffectedRows = sessionStates.LastAffectedRows s.StmtCtx.PrevLastInsertID = sessionStates.LastInsertID s.StmtCtx.SetWarnings(sessionStates.Warnings) return diff --git a/pkg/sessionctx/variable/sysvar.go b/pkg/sessionctx/variable/sysvar.go index a1fe045712577..972174676eeba 100644 --- a/pkg/sessionctx/variable/sysvar.go +++ b/pkg/sessionctx/variable/sysvar.go @@ -952,6 +952,14 @@ var defaultSysVars = []*SysVar{ SetGlobal: func(_ context.Context, s *SessionVars, val string) error { return stmtsummaryv2.SetMaxSQLLength(TidbOptInt(val, vardef.DefTiDBStmtSummaryMaxSQLLength)) }}, + {Scope: vardef.ScopeGlobal, Name: vardef.TiDBStmtSummaryPersistEvicted, Value: BoolToOnOff(vardef.DefTiDBStmtSummaryPersistEvicted), Type: vardef.TypeBool, AllowEmpty: true, + SetGlobal: func(_ context.Context, s *SessionVars, val string) error { + return stmtsummaryv2.SetPersistEvicted(TiDBOptOn(val)) + }}, + {Scope: vardef.ScopeGlobal, Name: vardef.TiDBStmtSummaryGroupByUser, Value: BoolToOnOff(vardef.DefTiDBStmtSummaryGroupByUser), Type: vardef.TypeBool, AllowEmpty: true, + SetGlobal: func(_ context.Context, s *SessionVars, val string) error { + return stmtsummaryv2.SetGroupByUser(TiDBOptOn(val)) + }}, {Scope: vardef.ScopeGlobal, Name: vardef.TiDBCapturePlanBaseline, Value: vardef.DefTiDBCapturePlanBaseline, Type: vardef.TypeBool, AllowEmptyAll: true}, {Scope: vardef.ScopeGlobal, Name: vardef.TiDBEvolvePlanTaskMaxTime, Value: strconv.Itoa(vardef.DefTiDBEvolvePlanTaskMaxTime), Type: vardef.TypeInt, MinValue: -1, MaxValue: math.MaxInt64}, {Scope: vardef.ScopeGlobal, Name: vardef.TiDBEvolvePlanTaskStartTime, Value: vardef.DefTiDBEvolvePlanTaskStartTime, Type: vardef.TypeTime}, diff --git a/pkg/util/stmtsummary/BUILD.bazel b/pkg/util/stmtsummary/BUILD.bazel index 16d48c2c65ec6..f98a374797061 100644 --- a/pkg/util/stmtsummary/BUILD.bazel +++ b/pkg/util/stmtsummary/BUILD.bazel @@ -40,7 +40,7 @@ go_test( ], embed = [":stmtsummary"], flaky = True, - shard_count = 26, + shard_count = 28, deps = [ "//pkg/meta/model", "//pkg/metrics", diff --git a/pkg/util/stmtsummary/evicted_test.go b/pkg/util/stmtsummary/evicted_test.go index 54e572efb4911..73c183efcb7c8 100644 --- a/pkg/util/stmtsummary/evicted_test.go +++ b/pkg/util/stmtsummary/evicted_test.go @@ -52,7 +52,7 @@ func newInduceSsbde(beginTime int64, endTime int64) *stmtSummaryByDigestElement // generate new StmtDigestKey and stmtSummaryByDigest func generateStmtSummaryByDigestKeyValue(schema string, beginTime int64, endTime int64) (*StmtDigestKey, *stmtSummaryByDigest) { key := &StmtDigestKey{} - key.Init(schema, "", "", "", "") + key.Init(schema, "", "", "", "", "") value := newInduceSsbd(beginTime, endTime) return key, value } @@ -191,7 +191,7 @@ func TestSimpleStmtSummaryByDigestEvicted(t *testing.T) { require.Equal(t, "{begin: 8, end: 9, count: 1}, {begin: 5, end: 6, count: 1}, {begin: 2, end: 3, count: 1}", getAllEvicted(ssbde)) evictedKey = &StmtDigestKey{} - evictedKey.Init("b", "", "", "", "") + evictedKey.Init("b", "", "", "", "", "") ssbde.AddEvicted(evictedKey, evictedValue, 4) require.Equal(t, "{begin: 8, end: 9, count: 2}, {begin: 5, end: 6, count: 2}, {begin: 2, end: 3, count: 2}, {begin: 1, end: 2, count: 1}", getAllEvicted(ssbde)) diff --git a/pkg/util/stmtsummary/statement_summary.go b/pkg/util/stmtsummary/statement_summary.go index 6dfb2e310dbb5..d81e1a72c2275 100644 --- a/pkg/util/stmtsummary/statement_summary.go +++ b/pkg/util/stmtsummary/statement_summary.go @@ -18,6 +18,7 @@ import ( "bytes" "cmp" "container/list" + "encoding/binary" "fmt" "math" "slices" @@ -53,8 +54,16 @@ type StmtDigestKey struct { } // Init initialize the hash key. -func (key *StmtDigestKey) Init(schemaName, digest, prevDigest, planDigest, resourceGroupName string) { - length := len(schemaName) + len(digest) + len(prevDigest) + len(planDigest) + len(resourceGroupName) +// When user is empty (group_by_user disabled), the hash is byte-identical to +// the pre-user-dimension layout. When user is non-empty, the hash appends a +// length-prefixed user segment after resourceGroupName so the boundary is +// unambiguous and pairs like ("rg", "alice") and ("rga", "lice") cannot +// collide. +func (key *StmtDigestKey) Init(schemaName, digest, prevDigest, planDigest, resourceGroupName, user string) { + length := len(schemaName) + len(digest) + len(prevDigest) + len(planDigest) + len(resourceGroupName) + len(user) + if len(user) > 0 { + length += 4 + } if cap(key.hash) < length { key.hash = make([]byte, 0, length) } else { @@ -65,6 +74,12 @@ func (key *StmtDigestKey) Init(schemaName, digest, prevDigest, planDigest, resou key.hash = append(key.hash, hack.Slice(prevDigest)...) key.hash = append(key.hash, hack.Slice(planDigest)...) key.hash = append(key.hash, hack.Slice(resourceGroupName)...) + if len(user) > 0 { + var buf [4]byte + binary.BigEndian.PutUint32(buf[:], uint32(len(user))) + key.hash = append(key.hash, buf[:]...) + key.hash = append(key.hash, hack.Slice(user)...) + } } // Hash implements SimpleLRUCache.Key. @@ -90,6 +105,7 @@ type stmtSummaryByDigestMap struct { optRefreshInterval *atomic2.Int64 optHistorySize *atomic2.Int32 optMaxSQLLength *atomic2.Int32 + optGroupByUser *atomic2.Bool // other stores summary of evicted data. other *stmtSummaryByDigestEvicted @@ -324,6 +340,7 @@ func newStmtSummaryByDigestMap() *stmtSummaryByDigestMap { optRefreshInterval: atomic2.NewInt64(1800), optHistorySize: atomic2.NewInt32(24), optMaxSQLLength: atomic2.NewInt32(32768), + optGroupByUser: atomic2.NewBool(false), other: ssbde, } newSsMap.summaryMap.SetOnEvict(func(k kvcache.Key, v kvcache.Value) { @@ -357,8 +374,6 @@ func (ssMap *stmtSummaryByDigestMap) AddStatement(sei *StmtExecInfo) { } key := StmtDigestKeyPool.Get().(*StmtDigestKey) - // Init hash value in advance, to reduce the time holding the lock. - key.Init(sei.SchemaName, sei.Digest, sei.PrevSQLDigest, sei.PlanDigest, sei.ResourceGroupName) var exist bool @@ -372,6 +387,15 @@ func (ssMap *stmtSummaryByDigestMap) AddStatement(sei *StmtExecInfo) { ssMap.Lock() defer ssMap.Unlock() + // Decide userForKey under the lock so SetGroupByUser's flag flip + Clear + // is atomic w.r.t. AddStatement; otherwise a post-clear insert could land + // under the wrong grouping mode. + userForKey := "" + if ssMap.optGroupByUser.Load() { + userForKey = sei.User + } + key.Init(sei.SchemaName, sei.Digest, sei.PrevSQLDigest, sei.PlanDigest, sei.ResourceGroupName, userForKey) + // Check again. Statements could be added before disabling the flag and after Clear(). if !ssMap.Enabled() { return @@ -413,6 +437,11 @@ func (ssMap *stmtSummaryByDigestMap) Clear() { ssMap.Lock() defer ssMap.Unlock() + ssMap.clearLocked() +} + +// clearLocked removes all statement summaries. ssMap.Lock must be held. +func (ssMap *stmtSummaryByDigestMap) clearLocked() { ssMap.summaryMap.DeleteAll() ssMap.other.Clear() ssMap.beginTimeForCurInterval = 0 @@ -520,6 +549,28 @@ func (ssMap *stmtSummaryByDigestMap) historySize() int { return int(ssMap.optHistorySize.Load()) } +// SetGroupByUser enables or disables grouping statement summaries by the +// executing user. Switching the flag clears existing data because existing +// rows were aggregated under a different grouping key. +func (ssMap *stmtSummaryByDigestMap) SetGroupByUser(value bool) error { + // Hold ssMap.Lock across the flag flip and clear so AddStatement (which + // reads the flag under the same lock) cannot insert a record with the + // old grouping mode after Clear() completes. + ssMap.Lock() + defer ssMap.Unlock() + if ssMap.optGroupByUser.Load() == value { + return nil + } + ssMap.optGroupByUser.Store(value) + ssMap.clearLocked() + return nil +} + +// GroupByUser reports whether statement summaries are grouped by user. +func (ssMap *stmtSummaryByDigestMap) GroupByUser() bool { + return ssMap.optGroupByUser.Load() +} + // SetHistorySize sets the history size for all summaries. func (ssMap *stmtSummaryByDigestMap) SetMaxStmtCount(value uint) error { // `optMaxStmtCount` and `ssMap` don't need to be strictly atomically updated. diff --git a/pkg/util/stmtsummary/statement_summary_test.go b/pkg/util/stmtsummary/statement_summary_test.go index 626de1bd74019..bdf4fac921c78 100644 --- a/pkg/util/stmtsummary/statement_summary_test.go +++ b/pkg/util/stmtsummary/statement_summary_test.go @@ -85,7 +85,7 @@ func TestAddStatement(t *testing.T) { stmtExecInfo1 := generateAnyExecInfo() stmtExecInfo1.ExecDetail.CommitDetail.Mu.PrewriteBackoffTypes = make([]string, 0) key := &StmtDigestKey{} - key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, "", stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName) + key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, "", stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName, "") samplePlan, _, _ := stmtExecInfo1.LazyInfo.GetEncodedPlan() stmtExecInfo1.ExecDetail.CommitDetail.Mu.Lock() expectedSummaryElement := stmtSummaryByDigestElement{ @@ -502,7 +502,7 @@ func TestAddStatement(t *testing.T) { stmtExecInfo4.SchemaName = "schema2" stmtExecInfo4.ExecDetail.CommitDetail = nil key = &StmtDigestKey{} - key.Init(stmtExecInfo4.SchemaName, stmtExecInfo4.Digest, "", stmtExecInfo4.PlanDigest, stmtExecInfo4.ResourceGroupName) + key.Init(stmtExecInfo4.SchemaName, stmtExecInfo4.Digest, "", stmtExecInfo4.PlanDigest, stmtExecInfo4.ResourceGroupName, "") ssMap.AddStatement(stmtExecInfo4) require.Equal(t, 2, ssMap.summaryMap.Size()) _, ok = ssMap.summaryMap.Get(key) @@ -512,7 +512,7 @@ func TestAddStatement(t *testing.T) { stmtExecInfo5 := stmtExecInfo1 stmtExecInfo5.Digest = "digest2" key = &StmtDigestKey{} - key.Init(stmtExecInfo5.SchemaName, stmtExecInfo5.Digest, "", stmtExecInfo5.PlanDigest, stmtExecInfo5.ResourceGroupName) + key.Init(stmtExecInfo5.SchemaName, stmtExecInfo5.Digest, "", stmtExecInfo5.PlanDigest, stmtExecInfo5.ResourceGroupName, "") ssMap.AddStatement(stmtExecInfo5) require.Equal(t, 3, ssMap.summaryMap.Size()) _, ok = ssMap.summaryMap.Get(key) @@ -522,7 +522,7 @@ func TestAddStatement(t *testing.T) { stmtExecInfo6 := stmtExecInfo1 stmtExecInfo6.PlanDigest = "plan_digest2" key = &StmtDigestKey{} - key.Init(stmtExecInfo6.SchemaName, stmtExecInfo6.Digest, "", stmtExecInfo6.PlanDigest, stmtExecInfo6.ResourceGroupName) + key.Init(stmtExecInfo6.SchemaName, stmtExecInfo6.Digest, "", stmtExecInfo6.PlanDigest, stmtExecInfo6.ResourceGroupName, "") ssMap.AddStatement(stmtExecInfo6) require.Equal(t, 4, ssMap.summaryMap.Size()) _, ok = ssMap.summaryMap.Get(key) @@ -545,7 +545,7 @@ func TestAddStatement(t *testing.T) { bindingSQL: originalSQL, } key = &StmtDigestKey{} - key.Init(stmtExecInfo7.SchemaName, stmtExecInfo7.Digest, "", stmtExecInfo7.PlanDigest, stmtExecInfo7.ResourceGroupName) + key.Init(stmtExecInfo7.SchemaName, stmtExecInfo7.Digest, "", stmtExecInfo7.PlanDigest, stmtExecInfo7.ResourceGroupName, "") ssMap.AddStatement(stmtExecInfo7) require.Equal(t, 5, ssMap.summaryMap.Size()) v, ok := ssMap.summaryMap.Get(key) @@ -1122,7 +1122,7 @@ func TestMaxStmtCount(t *testing.T) { // LRU cache should work. for i := loops - 10; i < loops; i++ { key := &StmtDigestKey{} - key.Init(stmtExecInfo1.SchemaName, fmt.Sprintf("digest%d", i), "", stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName) + key.Init(stmtExecInfo1.SchemaName, fmt.Sprintf("digest%d", i), "", stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName, "") key.Hash() _, ok := sm.Get(key) require.True(t, ok) @@ -1166,7 +1166,7 @@ func TestMaxSQLLength(t *testing.T) { ssMap.AddStatement(stmtExecInfo1) key := &StmtDigestKey{} - key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, "", stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName) + key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, "", stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName, "") value, ok := ssMap.summaryMap.Get(key) require.True(t, ok) @@ -1418,7 +1418,7 @@ func TestRefreshCurrentSummary(t *testing.T) { ssMap.beginTimeForCurInterval = now + 10 stmtExecInfo1 := generateAnyExecInfo() key := &StmtDigestKey{} - key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, "", stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName) + key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, "", stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName, "") ssMap.AddStatement(stmtExecInfo1) require.Equal(t, 1, ssMap.summaryMap.Size()) value, ok := ssMap.summaryMap.Get(key) @@ -1465,7 +1465,7 @@ func TestSummaryHistory(t *testing.T) { stmtExecInfo1 := generateAnyExecInfo() key := &StmtDigestKey{} - key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, "", stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName) + key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, "", stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName, "") for i := range 11 { ssMap.beginTimeForCurInterval = now + int64(i+1)*10 ssMap.AddStatement(stmtExecInfo1) @@ -1534,7 +1534,7 @@ func TestPrevSQL(t *testing.T) { stmtExecInfo1.PrevSQLDigest = "prevSQLDigest" ssMap.AddStatement(stmtExecInfo1) key := &StmtDigestKey{} - key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, stmtExecInfo1.PrevSQLDigest, stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName) + key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, stmtExecInfo1.PrevSQLDigest, stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName, "") require.Equal(t, 1, ssMap.summaryMap.Size()) _, ok := ssMap.summaryMap.Get(key) require.True(t, ok) @@ -1549,7 +1549,7 @@ func TestPrevSQL(t *testing.T) { stmtExecInfo2.PrevSQLDigest = "prevSQLDigest1" ssMap.AddStatement(stmtExecInfo2) require.Equal(t, 2, ssMap.summaryMap.Size()) - key.Init(stmtExecInfo2.SchemaName, stmtExecInfo2.Digest, stmtExecInfo2.PrevSQLDigest, stmtExecInfo2.PlanDigest, stmtExecInfo2.ResourceGroupName) + key.Init(stmtExecInfo2.SchemaName, stmtExecInfo2.Digest, stmtExecInfo2.PrevSQLDigest, stmtExecInfo2.PlanDigest, stmtExecInfo2.ResourceGroupName, "") _, ok = ssMap.summaryMap.Get(key) require.True(t, ok) } @@ -1562,7 +1562,7 @@ func TestEndTime(t *testing.T) { stmtExecInfo1 := generateAnyExecInfo() ssMap.AddStatement(stmtExecInfo1) key := &StmtDigestKey{} - key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, "", stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName) + key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, "", stmtExecInfo1.PlanDigest, stmtExecInfo1.ResourceGroupName, "") require.Equal(t, 1, ssMap.summaryMap.Size()) value, ok := ssMap.summaryMap.Get(key) require.True(t, ok) @@ -1608,7 +1608,7 @@ func TestPointGet(t *testing.T) { stmtExecInfo1.LazyInfo.(*mockLazyInfo).plan = fakePlanDigestGenerator() ssMap.AddStatement(stmtExecInfo1) key := &StmtDigestKey{} - key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, "", "", stmtExecInfo1.ResourceGroupName) + key.Init(stmtExecInfo1.SchemaName, stmtExecInfo1.Digest, "", "", stmtExecInfo1.ResourceGroupName, "") require.Equal(t, 1, ssMap.summaryMap.Size()) value, ok := ssMap.summaryMap.Get(key) require.True(t, ok) @@ -1687,3 +1687,81 @@ func TestAccessPrivilege(t *testing.T) { datums = reader.GetStmtSummaryCurrentRows() require.Len(t, datums, loops) } + +// TestAddStatementGroupByUser verifies that flipping the group-by-user flag +// splits the same digest into per-user rows and fills ssbd.user. The default +// (flag OFF) keeps legacy behavior: one row per digest regardless of user. +func TestAddStatementGroupByUser(t *testing.T) { + ssMap := newStmtSummaryByDigestMap() + + info1 := generateAnyExecInfo() + info1.User = "alice" + info2 := generateAnyExecInfo() + info2.User = "bob" + + // Flag off: both statements collapse into one record. + ssMap.AddStatement(info1) + ssMap.AddStatement(info2) + require.Equal(t, 1, ssMap.summaryMap.Size()) + + // Flipping the flag clears prior data (different grouping key). + require.NoError(t, ssMap.SetGroupByUser(true)) + require.Equal(t, 0, ssMap.summaryMap.Size()) + + ssMap.AddStatement(info1) + ssMap.AddStatement(info2) + ssMap.AddStatement(info1) + require.Equal(t, 2, ssMap.summaryMap.Size()) + + // With grouping ON, each record's authUsers must hold exactly one user — + // the one that groups it — so SAMPLE_USER naturally reflects the grouping + // dimension without a dedicated column. + seen := map[string]bool{} + for _, v := range ssMap.summaryMap.Values() { + ssbd := v.(*stmtSummaryByDigest) + elem := ssbd.history.Front().Value.(*stmtSummaryByDigestElement) + require.Len(t, elem.authUsers, 1) + for u := range elem.authUsers { + seen[u] = true + } + } + require.True(t, seen["alice"]) + require.True(t, seen["bob"]) + + // Flipping back off clears again, and re-emitted records merge users. + require.NoError(t, ssMap.SetGroupByUser(false)) + require.Equal(t, 0, ssMap.summaryMap.Size()) + ssMap.AddStatement(info1) + ssMap.AddStatement(info2) + require.Equal(t, 1, ssMap.summaryMap.Size()) + for _, v := range ssMap.summaryMap.Values() { + ssbd := v.(*stmtSummaryByDigest) + elem := ssbd.history.Front().Value.(*stmtSummaryByDigestElement) + require.Len(t, elem.authUsers, 2) + } +} + +// TestStmtDigestKeyBoundary guards against two regressions: +// 1. Adjacent string fields must not collide across boundary, e.g. +// (resourceGroupName, user) = ("rg", "alice") vs ("rga", "lice"); without +// a boundary marker on user, both produce the same hash. +// 2. With user empty (group_by_user OFF), the hash must stay byte-identical +// to the pre-user-dimension encoding so persisted/in-memory rows from +// older versions match. +func TestStmtDigestKeyBoundary(t *testing.T) { + k1 := &StmtDigestKey{} + k1.Init("schema", "digest", "prev", "plan", "rg", "alice") + k2 := &StmtDigestKey{} + k2.Init("schema", "digest", "prev", "plan", "rga", "lice") + require.NotEqual(t, k1.Hash(), k2.Hash(), "user segment must have an unambiguous boundary") + + // user="" leaves the hash equal to the legacy 5-field layout. + off := &StmtDigestKey{} + off.Init("schema", "digest", "prev", "plan", "rg", "") + legacy := append([]byte{}, hack.Slice("digest")...) + legacy = append(legacy, hack.Slice("schema")...) + legacy = append(legacy, hack.Slice("prev")...) + legacy = append(legacy, hack.Slice("plan")...) + legacy = append(legacy, hack.Slice("rg")...) + require.Equal(t, legacy, off.Hash()) +} diff --git a/pkg/util/stmtsummary/v2/BUILD.bazel b/pkg/util/stmtsummary/v2/BUILD.bazel index 1abfaf8d31663..4267fbd3f77bb 100644 --- a/pkg/util/stmtsummary/v2/BUILD.bazel +++ b/pkg/util/stmtsummary/v2/BUILD.bazel @@ -49,7 +49,7 @@ go_test( ], embed = [":stmtsummary"], flaky = True, - shard_count = 15, + shard_count = 18, deps = [ "//pkg/meta/model", "//pkg/metrics", @@ -64,5 +64,7 @@ go_test( "@com_github_prometheus_client_model//go", "@com_github_stretchr_testify//require", "@org_uber_go_goleak//:goleak", + "@org_uber_go_zap//:zap", + "@org_uber_go_zap//zapcore", ], ) diff --git a/pkg/util/stmtsummary/v2/logger.go b/pkg/util/stmtsummary/v2/logger.go index 64c3499c6ba28..cd8abd9b88512 100644 --- a/pkg/util/stmtsummary/v2/logger.go +++ b/pkg/util/stmtsummary/v2/logger.go @@ -17,9 +17,11 @@ package stmtsummary import ( "encoding/json" "fmt" + "strings" "time" "github.com/pingcap/log" + "github.com/pingcap/tidb/pkg/metrics" "github.com/pingcap/tidb/pkg/util/logutil" "go.uber.org/zap" "go.uber.org/zap/buffer" @@ -58,10 +60,10 @@ func (s *stmtLogStorage) persist(w *stmtWindow, end time.Time) { r.Unlock() } w.evicted.Lock() - if w.evicted.other.ExecCount > 0 { - w.evicted.other.Begin = begin - w.evicted.other.End = end.Unix() - s.log(w.evicted.other) + if w.evicted.otherForPersist.ExecCount > 0 { + w.evicted.otherForPersist.Begin = begin + w.evicted.otherForPersist.End = end.Unix() + s.log(w.evicted.otherForPersist) } w.evicted.Unlock() } @@ -70,6 +72,42 @@ func (s *stmtLogStorage) sync() error { return s.logger.Sync() } +// logEvicted writes evicted records to the stmt log with an `"evicted":true` +// marker so downstream consumers can distinguish per-record eviction events +// from rotated-window records. +func (s *stmtLogStorage) logEvicted(records []*StmtRecord) { + var builder strings.Builder + persisted := 0 + for _, r := range records { + b, err := marshalEvictedStmtRecord(r) + if err != nil { + logutil.BgLogger().Warn("failed to marshal evicted statement summary", zap.Error(err)) + continue + } + if builder.Len() > 0 { + builder.WriteByte('\n') + } + _, _ = builder.Write(b) + persisted++ + } + if builder.Len() == 0 { + return + } + s.logger.Info(builder.String()) + metrics.StmtSummaryEvictedLogCounter.WithLabelValues( + metrics.StmtSummaryTypeV2, + metrics.StmtSummaryEvictedLogResultPersisted, + ).Add(float64(persisted)) +} + +// evictedStmtRecord embeds *StmtRecord and adds an "evicted" JSON tag. +// Keeping the embedded pointer means the JSON field order matches StmtRecord +// and parsers tolerant of the extra field work unchanged. +type evictedStmtRecord struct { + *StmtRecord + Evicted bool `json:"evicted"` +} + func (s *stmtLogStorage) log(r *StmtRecord) { b, err := json.Marshal(r) if err != nil { @@ -79,6 +117,10 @@ func (s *stmtLogStorage) log(r *StmtRecord) { s.logger.Info(string(b)) } +func marshalEvictedStmtRecord(r *StmtRecord) ([]byte, error) { + return json.Marshal(evictedStmtRecord{StmtRecord: r, Evicted: true}) +} + type stmtLogEncoder struct{} func (*stmtLogEncoder) EncodeEntry(entry zapcore.Entry, _ []zapcore.Field) (*buffer.Buffer, error) { diff --git a/pkg/util/stmtsummary/v2/reader.go b/pkg/util/stmtsummary/v2/reader.go index 498c9c7437238..589fc4aa0a59f 100644 --- a/pkg/util/stmtsummary/v2/reader.go +++ b/pkg/util/stmtsummary/v2/reader.go @@ -448,6 +448,11 @@ type stmtTinyRecord struct { End int64 `json:"end"` } +type stmtPersistedRecord struct { + StmtRecord + Evicted bool `json:"evicted"` +} + type stmtFile struct { file *os.File begin int64 @@ -757,11 +762,14 @@ func (w *stmtParseWorker) handleLines( rows := make([][]types.Datum, 0, len(lines)) for _, line := range lines { - record, err := w.parse(line) + record, skipped, err := w.parse(line) if err != nil { // ignore invalid lines continue } + if skipped { + continue + } if w.needStop(record) { break @@ -790,12 +798,15 @@ func (w *stmtParseWorker) putRows( } } -func (*stmtParseWorker) parse(raw []byte) (*StmtRecord, error) { - var record StmtRecord +func (*stmtParseWorker) parse(raw []byte) (*StmtRecord, bool, error) { + var record stmtPersistedRecord if err := json.Unmarshal(raw, &record); err != nil { - return nil, err + return nil, false, err } - return &record, nil + if record.Evicted { + return nil, true, nil + } + return &record.StmtRecord, false, nil } func (w *stmtParseWorker) needStop(record *StmtRecord) bool { diff --git a/pkg/util/stmtsummary/v2/reader_test.go b/pkg/util/stmtsummary/v2/reader_test.go index 906fc69d2d2a0..6f24b1c3422fe 100644 --- a/pkg/util/stmtsummary/v2/reader_test.go +++ b/pkg/util/stmtsummary/v2/reader_test.go @@ -266,6 +266,8 @@ func TestHistoryReader(t *testing.T) { require.NoError(t, err) _, err = file.WriteString("{\"begin\":1672129270,\"end\":1672129280,\"digest\":\"digest2\",\"exec_count\":20}\n") require.NoError(t, err) + _, err = file.WriteString("{\"begin\":1672129270,\"end\":1672129280,\"digest\":\"evicted_digest\",\"exec_count\":99,\"evicted\":true}\n") + require.NoError(t, err) require.NoError(t, file.Close()) file, err = os.Create(filename2) diff --git a/pkg/util/stmtsummary/v2/record_test.go b/pkg/util/stmtsummary/v2/record_test.go index fbb65feb5468c..4c24b5ece8e70 100644 --- a/pkg/util/stmtsummary/v2/record_test.go +++ b/pkg/util/stmtsummary/v2/record_test.go @@ -15,6 +15,7 @@ package stmtsummary import ( + "encoding/json" "testing" "github.com/stretchr/testify/require" @@ -83,4 +84,11 @@ func TestStmtRecord(t *testing.T) { require.Equal(t, info.TotalRUV2*2, record2.SumRUV2) require.Equal(t, info.CPUUsages.TidbCPUTime*2, record2.SumTidbCPU) require.Equal(t, info.CPUUsages.TikvCPUTime*2, record2.SumTikvCPU) + + b, err := marshalEvictedStmtRecord(record2) + require.NoError(t, err) + items := make(map[string]any) + require.NoError(t, json.Unmarshal(b, &items)) + require.Equal(t, true, items["evicted"]) + require.Equal(t, record2.Digest, items["digest"]) } diff --git a/pkg/util/stmtsummary/v2/stmtsummary.go b/pkg/util/stmtsummary/v2/stmtsummary.go index acefa0955e191..2eab73de1ca2f 100644 --- a/pkg/util/stmtsummary/v2/stmtsummary.go +++ b/pkg/util/stmtsummary/v2/stmtsummary.go @@ -41,6 +41,17 @@ const ( defaultMaxSQLLength = 32768 defaultRefreshInterval = 30 * 60 // 30 min defaultRotateCheckInterval = 1 // s + + // evictedLogChanCap bounds the buffer of per-record evicted entries waiting + // to be logged. When full, new evictions are dropped so Add() never blocks. + evictedLogChanCap = 1024 + + // evictedLogBatchSize and evictedLogFlushInterval bound the async logger's + // batching. They reduce write frequency under eviction bursts while keeping + // single-record latency low. + evictedLogBatchSize = 64 + evictedLogFlushInterval = 100 * time.Millisecond + evictedDropReportInterval = 30 * time.Second ) var ( @@ -86,12 +97,19 @@ type StmtSummary struct { optMaxStmtCount *atomic2.Uint32 optMaxSQLLength *atomic2.Uint32 optRefreshInterval *atomic2.Uint32 + optPersistEvicted *atomic2.Bool + optGroupByUser *atomic2.Bool window *stmtWindow windowLock sync.Mutex storage stmtStorage closeWg sync.WaitGroup closed atomic.Bool + + // evictedCh carries per-record evictions to the async logger. + // Eviction persistence is controlled by optPersistEvicted; sends are non-blocking. + evictedCh chan *StmtRecord + evictedDropped atomic.Uint64 } // NewStmtSummary creates a new StmtSummary from Config. @@ -113,7 +131,8 @@ func NewStmtSummary(cfg *Config) (*StmtSummary, error) { optMaxStmtCount: atomic2.NewUint32(defaultMaxStmtCount), optMaxSQLLength: atomic2.NewUint32(defaultMaxSQLLength), optRefreshInterval: atomic2.NewUint32(defaultRefreshInterval), - window: newStmtWindow(timeNow(), uint(defaultMaxStmtCount)), + optPersistEvicted: atomic2.NewBool(false), + optGroupByUser: atomic2.NewBool(false), storage: newStmtLogStorage(&log.Config{ File: log.FileLogConfig{ Filename: cfg.Filename, @@ -122,13 +141,20 @@ func NewStmtSummary(cfg *Config) (*StmtSummary, error) { MaxBackups: cfg.FileMaxBackups, }, }), + evictedCh: make(chan *StmtRecord, evictedLogChanCap), } + s.window = newStmtWindow(timeNow(), uint(defaultMaxStmtCount), s.onEvict) s.closeWg.Add(1) go func() { defer s.closeWg.Done() s.rotateLoop() }() + s.closeWg.Add(1) + go func() { + defer s.closeWg.Done() + s.evictedLogLoop() + }() return s, nil } @@ -146,9 +172,18 @@ func NewStmtSummary4Test(maxStmtCount uint) *StmtSummary { optMaxStmtCount: atomic2.NewUint32(defaultMaxStmtCount), optMaxSQLLength: atomic2.NewUint32(defaultMaxSQLLength), optRefreshInterval: atomic2.NewUint32(60 * 60 * 24 * 365), // 1 year - window: newStmtWindow(timeNow(), maxStmtCount), + optPersistEvicted: atomic2.NewBool(false), + optGroupByUser: atomic2.NewBool(false), storage: &mockStmtStorage{}, + evictedCh: make(chan *StmtRecord, evictedLogChanCap), } + ss.window = newStmtWindow(timeNow(), maxStmtCount, ss.onEvict) + + ss.closeWg.Add(1) + go func() { + defer ss.closeWg.Done() + ss.evictedLogLoop() + }() return ss } @@ -235,6 +270,40 @@ func (s *StmtSummary) SetRefreshInterval(v uint32) error { return nil } +// PersistEvicted reports whether per-record evictions are persisted. +func (s *StmtSummary) PersistEvicted() bool { + return s.optPersistEvicted.Load() +} + +// SetPersistEvicted enables or disables per-record eviction persistence. +func (s *StmtSummary) SetPersistEvicted(v bool) error { + s.optPersistEvicted.Store(v) + return nil +} + +// GroupByUser reports whether statement summaries are grouped by the +// executing user in addition to the usual digest/schema/plan tuple. +func (s *StmtSummary) GroupByUser() bool { + return s.optGroupByUser.Load() +} + +// SetGroupByUser toggles user-dimension grouping. Switching the flag clears +// the in-memory window because existing records were aggregated under a +// different grouping key; persisted records are unaffected. +func (s *StmtSummary) SetGroupByUser(v bool) error { + // Hold windowLock across the flag flip and clear so Add (which reads + // the flag under the same lock) cannot insert a record with the old + // grouping mode after the window is cleared. + s.windowLock.Lock() + defer s.windowLock.Unlock() + if s.optGroupByUser.Load() == v { + return nil + } + s.optGroupByUser.Store(v) + s.window.clear() + return nil +} + // Add adds a single stmtsummary.StmtExecInfo to the current statistics window // of StmtSummary. Before adding, it will check whether the current window has // expired, and if it has expired, the window will be persisted asynchronously @@ -245,11 +314,22 @@ func (s *StmtSummary) Add(info *stmtsummary.StmtExecInfo) { } k := stmtsummary.StmtDigestKeyPool.Get().(*stmtsummary.StmtDigestKey) - // Init hash value in advance, to reduce the time holding the lock. - k.Init(info.SchemaName, info.Digest, info.PrevSQLDigest, info.PlanDigest, info.ResourceGroupName) // Add info to the current statistics window. s.windowLock.Lock() + if s.closed.Load() { + s.windowLock.Unlock() + stmtsummary.StmtDigestKeyPool.Put(k) + return + } + // Decide userForKey under windowLock so SetGroupByUser's flag flip + clear + // is atomic w.r.t. Add; otherwise a post-clear insert could land under the + // wrong grouping mode. + userForKey := "" + if s.optGroupByUser.Load() { + userForKey = info.User + } + k.Init(info.SchemaName, info.Digest, info.PrevSQLDigest, info.PlanDigest, info.ResourceGroupName, userForKey) var record *lockedStmtRecord v, exist := s.window.lru.Get(k) if exist { @@ -306,11 +386,17 @@ func (s *StmtSummary) ClearInternal() { // Close closes the work of StmtSummary. func (s *StmtSummary) Close() { + s.windowLock.Lock() + if !s.closed.CompareAndSwap(false, true) { + s.windowLock.Unlock() + return + } + s.windowLock.Unlock() + if s.cancel != nil { s.cancel() s.closeWg.Wait() } - s.closed.Store(true) s.flush() } @@ -319,7 +405,7 @@ func (s *StmtSummary) flush() { s.windowLock.Lock() window := s.window - s.window = newStmtWindow(now, uint(s.MaxStmtCount())) + s.window = newStmtWindow(now, uint(s.MaxStmtCount()), s.onEvict) s.windowLock.Unlock() if window.lru.Size() > 0 { @@ -364,7 +450,7 @@ func (s *StmtSummary) updateMetrics() { func (s *StmtSummary) rotate(now time.Time) { w := s.window - s.window = newStmtWindow(now, uint(s.MaxStmtCount())) + s.window = newStmtWindow(now, uint(s.MaxStmtCount()), s.onEvict) size := w.lru.Size() if size > 0 { // Persist window asynchronously. @@ -376,6 +462,126 @@ func (s *StmtSummary) rotate(now time.Time) { } } +// onEvict is the LRU eviction hook installed on every stmtWindow. +// Called while the record's lock is held (see newStmtWindow). We copy the +// fields we need and hand the clone off to the async log goroutine. A +// non-blocking send is used so the hot Add() path never stalls on log I/O. +func (s *StmtSummary) onEvict(_ *stmtsummary.StmtDigestKey, r *StmtRecord, begin, end time.Time) bool { + if !s.optPersistEvicted.Load() { + return false + } + if s.evictedCh == nil { + return false + } + clone := cloneRecordForLog(r) + clone.Begin = begin.Unix() + clone.End = end.Unix() + select { + case s.evictedCh <- clone: + return true + default: + s.evictedDropped.Add(1) + metrics.StmtSummaryEvictedLogCounter.WithLabelValues( + metrics.StmtSummaryTypeV2, + metrics.StmtSummaryEvictedLogResultDropped, + ).Inc() + return false + } +} + +// evictedLogLoop drains evictedCh and writes each record to the stmt log. +// When group_by_user is also enabled, each logged record represents exactly +// one (digest, user) group that fell out of the LRU. +func (s *StmtSummary) evictedLogLoop() { + reportTicker := time.NewTicker(evictedDropReportInterval) + defer reportTicker.Stop() + + flushTimer := time.NewTimer(evictedLogFlushInterval) + if !flushTimer.Stop() { + <-flushTimer.C + } + defer flushTimer.Stop() + + var lastDropReport uint64 + report := func() { + cur := s.evictedDropped.Load() + if cur > lastDropReport { + logutil.BgLogger().Warn("stmt summary evicted log dropped records", + zap.Uint64("dropped_total", cur), + zap.Uint64("since_last_report", cur-lastDropReport), + ) + lastDropReport = cur + } + } + + stopFlushTimer := func() { + if !flushTimer.Stop() { + select { + case <-flushTimer.C: + default: + } + } + } + + batch := make([]*StmtRecord, 0, evictedLogBatchSize) + flush := func() { + if len(batch) == 0 { + return + } + s.storage.logEvicted(batch) + for i := range batch { + batch[i] = nil + } + batch = batch[:0] + stopFlushTimer() + } + appendRecord := func(r *StmtRecord) { + batch = append(batch, r) + if len(batch) == 1 { + flushTimer.Reset(evictedLogFlushInterval) + } + if len(batch) >= evictedLogBatchSize { + flush() + } + } + drainAvailable := func() { + for len(batch) > 0 && len(batch) < evictedLogBatchSize { + select { + case r := <-s.evictedCh: + appendRecord(r) + default: + return + } + } + } + + for { + select { + case <-s.ctx.Done(): + // Close sets closed while holding windowLock before canceling this + // context, and Add rechecks closed under the same lock. At this + // point no Add can enqueue more evicted records. + for { + select { + case r := <-s.evictedCh: + appendRecord(r) + default: + flush() + report() + return + } + } + case r := <-s.evictedCh: + appendRecord(r) + drainAvailable() + case <-flushTimer.C: + flush() + case <-reportTicker.C: + report() + } + } +} + // stmtWindow represents a single statistical window, which has a begin // time and an end time. Data within a single window is eliminated // according to the LRU strategy. All evicted data will be aggregated @@ -387,7 +593,14 @@ type stmtWindow struct { evictedCount atomic.Int64 // total number of LRU evictions in this window } -func newStmtWindow(begin time.Time, capacity uint) *stmtWindow { +// onEvictFn is invoked for every LRU eviction. The callback receives the +// locked record (caller holds r.Lock) so it can copy fields cheaply. It +// returns true when the record has been handed off for per-record persistence, +// in which case the caller can skip adding it to the persisted aggregate. +// Must not block. +type onEvictFn func(key *stmtsummary.StmtDigestKey, r *StmtRecord, begin, end time.Time) bool + +func newStmtWindow(begin time.Time, capacity uint, onEvict onEvictFn) *stmtWindow { w := &stmtWindow{ begin: begin, lru: kvcache.NewSimpleLRUCache(capacity, 0, 0), @@ -398,7 +611,12 @@ func newStmtWindow(begin time.Time, capacity uint) *stmtWindow { r := v.(*lockedStmtRecord) r.Lock() defer r.Unlock() - w.evicted.add(k.(*stmtsummary.StmtDigestKey), r.StmtRecord) + key := k.(*stmtsummary.StmtDigestKey) + queuedForEvictedLog := false + if onEvict != nil { + queuedForEvictedLog = onEvict(key, r.StmtRecord, w.begin, timeNow()) + } + w.evicted.add(key, r.StmtRecord, queuedForEvictedLog) }) return w } @@ -411,29 +629,32 @@ func (w *stmtWindow) clear() { type stmtStorage interface { persist(w *stmtWindow, end time.Time) + // logEvicted writes evicted records to durable storage. It may be + // called concurrently with persist; implementations must be safe to call + // from the evictedLogLoop goroutine. + logEvicted(records []*StmtRecord) sync() error } type stmtEvicted struct { sync.Mutex - keys map[string]struct{} + keys map[string]struct{} + // other contains all evicted records in the current window. other *StmtRecord + // otherForPersist contains records not covered by per-record evicted logs. + // When per-record evicted logging is disabled, it is equivalent to other. + otherForPersist *StmtRecord } func newStmtEvicted() *stmtEvicted { return &stmtEvicted{ - keys: make(map[string]struct{}), - other: &StmtRecord{ - AuthUsers: make(map[string]struct{}), - MinLatency: time.Duration(math.MaxInt64), - BackoffTypes: make(map[string]int), - FirstSeen: time.Now(), - LastSeen: time.Now(), - }, + keys: make(map[string]struct{}), + other: newEvictedAggregateRecord(), + otherForPersist: newEvictedAggregateRecord(), } } -func (e *stmtEvicted) add(key *stmtsummary.StmtDigestKey, record *StmtRecord) { +func (e *stmtEvicted) add(key *stmtsummary.StmtDigestKey, record *StmtRecord, queuedForEvictedLog bool) { if key == nil || record == nil { return } @@ -441,6 +662,9 @@ func (e *stmtEvicted) add(key *stmtsummary.StmtDigestKey, record *StmtRecord) { defer e.Unlock() e.keys[string(key.Hash())] = struct{}{} e.other.Merge(record) + if !queuedForEvictedLog { + e.otherForPersist.Merge(record) + } } func (e *stmtEvicted) count() int { @@ -449,6 +673,16 @@ func (e *stmtEvicted) count() int { return len(e.keys) } +func newEvictedAggregateRecord() *StmtRecord { + return &StmtRecord{ + AuthUsers: make(map[string]struct{}), + MinLatency: time.Duration(math.MaxInt64), + BackoffTypes: make(map[string]int), + FirstSeen: time.Now(), + LastSeen: time.Now(), + } +} + type lockedStmtRecord struct { sync.Mutex *StmtRecord @@ -457,6 +691,7 @@ type lockedStmtRecord struct { type mockStmtStorage struct { sync.Mutex windows []*stmtWindow + evicted []*StmtRecord } func (s *mockStmtStorage) persist(w *stmtWindow, _ time.Time) { @@ -465,10 +700,38 @@ func (s *mockStmtStorage) persist(w *stmtWindow, _ time.Time) { s.Unlock() } +func (s *mockStmtStorage) logEvicted(records []*StmtRecord) { + s.Lock() + s.evicted = append(s.evicted, records...) + s.Unlock() +} + func (*mockStmtStorage) sync() error { return nil } +// cloneRecordForLog returns a shallow copy of r with its two mutable maps +// (AuthUsers, BackoffTypes) cloned, so the async logger can marshal the +// snapshot without racing with further updates on the retained StmtRecord. +// Called with r's lock held (see onEvict). +func cloneRecordForLog(r *StmtRecord) *StmtRecord { + c := *r + if len(r.AuthUsers) > 0 { + c.AuthUsers = make(map[string]struct{}, len(r.AuthUsers)) + for u := range r.AuthUsers { + c.AuthUsers[u] = struct{}{} + } + } + if len(r.BackoffTypes) > 0 { + c.BackoffTypes = make(map[string]int, len(r.BackoffTypes)) + for k, v := range r.BackoffTypes { + c.BackoffTypes[k] = v + } + } + // IndexNames is a slice; shallow copy is fine because it is append-only. + return &c +} + /* Public proxy functions between v1 and v2 */ // Add wraps GlobalStmtSummary.Add and stmtsummary.StmtSummaryByDigestMap.AddStatement. @@ -544,3 +807,25 @@ func SetMaxSQLLength(v int) error { } return stmtsummary.StmtSummaryByDigestMap.SetMaxSQLLength(v) } + +// SetPersistEvicted toggles per-record eviction persistence. Only v2 +// (persistent) honors this flag; v1 has no log sink, so the call is a no-op +// for it. +func SetPersistEvicted(v bool) error { + if GlobalStmtSummary != nil { + return GlobalStmtSummary.SetPersistEvicted(v) + } + return nil +} + +// SetGroupByUser toggles the user dimension on both v1 and v2 so the sysvar +// setter can call one entry point regardless of which backend is active. +func SetGroupByUser(v bool) error { + if err := stmtsummary.StmtSummaryByDigestMap.SetGroupByUser(v); err != nil { + return err + } + if GlobalStmtSummary != nil { + return GlobalStmtSummary.SetGroupByUser(v) + } + return nil +} diff --git a/pkg/util/stmtsummary/v2/stmtsummary_test.go b/pkg/util/stmtsummary/v2/stmtsummary_test.go index 062a52e9c41b5..c395045e7b387 100644 --- a/pkg/util/stmtsummary/v2/stmtsummary_test.go +++ b/pkg/util/stmtsummary/v2/stmtsummary_test.go @@ -15,14 +15,20 @@ package stmtsummary import ( + "bytes" "encoding/json" "path/filepath" + "strings" "testing" + "time" "github.com/pingcap/tidb/pkg/metrics" + "github.com/pingcap/tidb/pkg/util/stmtsummary" "github.com/prometheus/client_golang/prometheus" dto "github.com/prometheus/client_model/go" "github.com/stretchr/testify/require" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" ) func readGaugeValue(t *testing.T, gauge prometheus.Gauge) float64 { @@ -32,6 +38,13 @@ func readGaugeValue(t *testing.T, gauge prometheus.Gauge) float64 { return m.GetGauge().GetValue() } +func readCounterValue(t *testing.T, counter prometheus.Counter) float64 { + t.Helper() + m := &dto.Metric{} + require.NoError(t, counter.Write(m)) + return m.GetCounter().GetValue() +} + func TestStmtWindow(t *testing.T) { ss := NewStmtSummary4Test(5) defer ss.Close() @@ -82,6 +95,157 @@ func TestStmtSummary(t *testing.T) { require.Equal(t, 0, w.lru.Size()) } +func TestStmtSummaryPersistEvicted(t *testing.T) { + begin := time.Date(2026, 5, 25, 10, 0, 0, 0, time.UTC) + evictAt := begin.Add(42 * time.Second) + now := begin + oldTimeNow := timeNow + timeNow = func() time.Time { + return now + } + t.Cleanup(func() { + timeNow = oldTimeNow + }) + + storage := &mockStmtStorage{} + ss := NewStmtSummary4Test(2) + ss.storage = storage + defer ss.Close() + require.NoError(t, ss.SetPersistEvicted(true)) + + // With capacity 2, the 3rd and later distinct digests evict older entries + // and should each land in storage.evicted. + ss.Add(GenerateStmtExecInfo4Test("digest1")) + ss.Add(GenerateStmtExecInfo4Test("digest2")) + now = evictAt + ss.Add(GenerateStmtExecInfo4Test("digest3")) // evicts digest1 + ss.Add(GenerateStmtExecInfo4Test("digest4")) // evicts digest2 + + // The log is async; wait briefly for drain. + require.Eventually(t, func() bool { + storage.Lock() + defer storage.Unlock() + return len(storage.evicted) == 2 + }, time.Second, 10*time.Millisecond, "expected 2 evicted records to be logged") + + storage.Lock() + digests := []string{storage.evicted[0].Digest, storage.evicted[1].Digest} + for _, record := range storage.evicted { + require.Equal(t, begin.Unix(), record.Begin) + require.Equal(t, evictAt.Unix(), record.End) + } + storage.Unlock() + require.ElementsMatch(t, []string{"digest1", "digest2"}, digests) + + // Disable and verify no further log writes. + require.NoError(t, ss.SetPersistEvicted(false)) + ss.Add(GenerateStmtExecInfo4Test("digest5")) // evicts digest3 + require.Never(t, func() bool { + storage.Lock() + defer storage.Unlock() + return len(storage.evicted) != 2 + }, 100*time.Millisecond, 10*time.Millisecond, "evicted count should remain 2 after disabling") +} + +func TestStmtSummaryPersistEvictedDoesNotPersistLoggedRecordsAsAggregate(t *testing.T) { + var logBuf bytes.Buffer + storage := &stmtLogStorage{ + logger: zap.New(zapcore.NewCore(&stmtLogEncoder{}, zapcore.AddSync(&logBuf), zapcore.InfoLevel)), + } + + ss := NewStmtSummary4Test(2) + ss.storage = storage + require.NoError(t, ss.SetPersistEvicted(true)) + + ss.Add(GenerateStmtExecInfo4Test("digest1")) + ss.Add(GenerateStmtExecInfo4Test("digest2")) + ss.Add(GenerateStmtExecInfo4Test("digest3")) // evicts digest1 + ss.Add(GenerateStmtExecInfo4Test("digest4")) // evicts digest2 + persistedBefore := readCounterValue(t, metrics.StmtSummaryEvictedLogCounter.WithLabelValues( + metrics.StmtSummaryTypeV2, + metrics.StmtSummaryEvictedLogResultPersisted, + )) + ss.Close() + persistedAfter := readCounterValue(t, metrics.StmtSummaryEvictedLogCounter.WithLabelValues( + metrics.StmtSummaryTypeV2, + metrics.StmtSummaryEvictedLogResultPersisted, + )) + require.Equal(t, 2.0, persistedAfter-persistedBefore) + + type loggedRecord struct { + Digest string `json:"digest"` + ExecCount int64 `json:"exec_count"` + Evicted bool `json:"evicted"` + } + + var totalExecCount int64 + evictedDigests := make([]string, 0, 2) + for _, line := range strings.Split(strings.TrimSpace(logBuf.String()), "\n") { + var record loggedRecord + require.NoError(t, json.Unmarshal([]byte(line), &record)) + totalExecCount += record.ExecCount + if record.Evicted { + evictedDigests = append(evictedDigests, record.Digest) + continue + } + require.NotEmpty(t, record.Digest, "logged evicted records should not also be persisted as the aggregate row") + } + + require.ElementsMatch(t, []string{"digest1", "digest2"}, evictedDigests) + require.Equal(t, int64(4), totalExecCount) +} + +func TestStmtSummaryGroupByUser(t *testing.T) { + ss := NewStmtSummary4Test(100) + defer ss.Close() + + // Two statements, same digest, different users: without the flag they + // should merge into one record. + ss.Add(stmtExecInfoWithUser("digest1", "alice")) + ss.Add(stmtExecInfoWithUser("digest1", "bob")) + require.Equal(t, 1, ss.window.lru.Size()) + + // Switching the flag on clears the window. Re-emitting produces two rows. + require.NoError(t, ss.SetGroupByUser(true)) + require.Equal(t, 0, ss.window.lru.Size()) + ss.Add(stmtExecInfoWithUser("digest1", "alice")) + ss.Add(stmtExecInfoWithUser("digest1", "bob")) + ss.Add(stmtExecInfoWithUser("digest1", "alice")) + require.Equal(t, 2, ss.window.lru.Size()) + + // When grouping by user, each record's AuthUsers must hold exactly one + // user — the one that groups it — so SAMPLE_USER naturally reflects the + // grouping dimension without a dedicated column. + users := map[string]int64{} + for _, v := range ss.window.lru.Values() { + r := v.(*lockedStmtRecord) + require.Len(t, r.AuthUsers, 1) + for u := range r.AuthUsers { + users[u] = r.ExecCount + } + } + require.Equal(t, int64(2), users["alice"]) + require.Equal(t, int64(1), users["bob"]) + + // Turning the flag off again clears and reverts to single-record merging. + require.NoError(t, ss.SetGroupByUser(false)) + ss.Add(stmtExecInfoWithUser("digest1", "alice")) + ss.Add(stmtExecInfoWithUser("digest1", "bob")) + require.Equal(t, 1, ss.window.lru.Size()) + for _, v := range ss.window.lru.Values() { + r := v.(*lockedStmtRecord) + require.Len(t, r.AuthUsers, 2) // both users merged when grouping is off + } +} + +// stmtExecInfoWithUser returns a StmtExecInfo whose digest and User fields are +// set; everything else is the generic test fixture. +func stmtExecInfoWithUser(digest, user string) *stmtsummary.StmtExecInfo { + info := GenerateStmtExecInfo4Test(digest) + info.User = user + return info +} + func TestWindowEvictedCountResetOnRotate(t *testing.T) { ss := NewStmtSummary4Test(2) defer ss.Close()