diff --git a/pkg/executor/select.go b/pkg/executor/select.go
index 3ba9027f569bd..8ec4866e59368 100644
--- a/pkg/executor/select.go
+++ b/pkg/executor/select.go
@@ -1213,7 +1213,9 @@ func ResetContextOfStmt(ctx sessionctx.Context, s ast.StmtNode) (err error) {
 		sc.PrevLastInsertID = vars.StmtCtx.PrevLastInsertID
 	}
 	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/BUILD.bazel b/pkg/metrics/BUILD.bazel
index bf61ff52c766e..0cbdb20518474 100644
--- a/pkg/metrics/BUILD.bazel
+++ b/pkg/metrics/BUILD.bazel
@@ -23,6 +23,7 @@ go_library(
         "server.go",
         "session.go",
         "sli.go",
+        "stmtsummary.go",
         "stats.go",
         "telemetry.go",
         "topsql.go",
@@ -62,6 +63,8 @@ go_test(
         "//pkg/statistics/handle/cache",
         "//pkg/testkit/testsetup",
         "@com_github_pingcap_errors//:errors",
+        "@com_github_prometheus_client_golang//prometheus",
+        "@com_github_prometheus_client_model//go",
         "@com_github_stretchr_testify//require",
         "@org_uber_go_goleak//:goleak",
     ],
diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go
index f1f6264bc9b61..1cd444f4d5f91 100644
--- a/pkg/metrics/metrics.go
+++ b/pkg/metrics/metrics.go
@@ -97,6 +97,7 @@ func InitMetrics() {
 	InitResourceGroupMetrics()
 	InitGlobalSortMetrics()
 	InitInfoSchemaV2Metrics()
+	InitStmtSummaryMetrics()
 	timermetrics.InitTimerMetrics()
 
 	// For now, those metrics are initialized but not registered.
@@ -319,6 +320,11 @@ func RegisterMetrics() {
 	prometheus.MustRegister(IndexLookRowsCounter)
 	prometheus.MustRegister(IndexLookUpExecutorRowNumber)
 	prometheus.MustRegister(IndexLookUpCopTaskCount)
+
+	// StmtSummary
+	prometheus.MustRegister(StmtSummaryWindowRecordCount)
+	prometheus.MustRegister(StmtSummaryWindowEvictedCount)
+	prometheus.MustRegister(StmtSummaryEvictedLogCounter)
 }
 
 var mode struct {
diff --git a/pkg/metrics/metrics_internal_test.go b/pkg/metrics/metrics_internal_test.go
index 21318c0d4f78f..13f8ba0f2cf9c 100644
--- a/pkg/metrics/metrics_internal_test.go
+++ b/pkg/metrics/metrics_internal_test.go
@@ -18,6 +18,8 @@ import (
 	"testing"
 
 	"github.com/pingcap/errors"
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/require"
 )
 
@@ -25,3 +27,54 @@ func TestRetLabel(t *testing.T) {
 	require.Equal(t, opSucc, RetLabel(nil))
 	require.Equal(t, opFailed, RetLabel(errors.New("test error")))
 }
+
+func readGaugeValue(t *testing.T, gauge prometheus.Gauge) float64 {
+	t.Helper()
+	m := &dto.Metric{}
+	require.NoError(t, gauge.Write(m))
+	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)
+	close(ch)
+
+	count := 0
+	for range ch {
+		count++
+	}
+	return count
+}
+
+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))
+	require.Equal(t, 1, countCollectedMetrics(StmtSummaryWindowEvictedCount))
+	require.Equal(t, 3.0, readGaugeValue(t, StmtSummaryWindowRecordCount.WithLabelValues(StmtSummaryTypeV1)))
+	require.Equal(t, 1.0, readGaugeValue(t, StmtSummaryWindowEvictedCount.WithLabelValues(StmtSummaryTypeV1)))
+
+	SetStmtSummaryWindowMetrics(StmtSummaryTypeV2, 5, 2)
+	require.Equal(t, 2, countCollectedMetrics(StmtSummaryWindowRecordCount))
+	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
new file mode 100644
index 0000000000000..6a93ae55dd2b7
--- /dev/null
+++ b/pkg/metrics/stmtsummary.go
@@ -0,0 +1,128 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metrics
+
+import (
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// Statement summary metrics.
+const (
+	// StmtSummaryTypeV1 marks metrics reported by the legacy statement summary implementation.
+	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 (
+	// StmtSummaryWindowRecordCount is a gauge that tracks the number of statement
+	// summary records in the current statement summary window.
+	StmtSummaryWindowRecordCount *prometheus.GaugeVec
+
+	// StmtSummaryWindowEvictedCount is a gauge that tracks the number of LRU
+	// evictions that have occurred in the current statement summary window.
+	// 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
+	stmtSummaryWindowEvictedCountV2 prometheus.Gauge
+
+	stmtSummaryWindowMetricsMu sync.Mutex
+)
+
+// InitStmtSummaryMetrics initializes statement summary metrics.
+func InitStmtSummaryMetrics() {
+	StmtSummaryWindowRecordCount = NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "tidb",
+			Subsystem: "stmt_summary",
+			Name:      "window_record_count",
+			Help:      "The number of statement summary records currently tracked by statement summary.",
+		}, []string{LblType})
+
+	StmtSummaryWindowEvictedCount = NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "tidb",
+			Subsystem: "stmt_summary",
+			Name:      "window_evicted_count",
+			Help:      "The number of LRU evictions in the current statement summary window.",
+		}, []string{LblType})
+
+	StmtSummaryEvictedLogCounter = 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
+	stmtSummaryWindowEvictedCountV1 = nil
+	stmtSummaryWindowEvictedCountV2 = nil
+	stmtSummaryWindowMetricsMu.Unlock()
+}
+
+// SetStmtSummaryWindowMetrics reports statement summary window metrics for a given implementation type.
+func SetStmtSummaryWindowMetrics(typ string, recordCount, evictedCount float64) {
+	switch typ {
+	case StmtSummaryTypeV1:
+		recordGauge, evictedGauge := getStmtSummaryWindowMetricsLocked(typ)
+		recordGauge.Set(recordCount)
+		evictedGauge.Set(evictedCount)
+	case StmtSummaryTypeV2:
+		recordGauge, evictedGauge := getStmtSummaryWindowMetricsLocked(typ)
+		recordGauge.Set(recordCount)
+		evictedGauge.Set(evictedCount)
+	default:
+		StmtSummaryWindowRecordCount.WithLabelValues(typ).Set(recordCount)
+		StmtSummaryWindowEvictedCount.WithLabelValues(typ).Set(evictedCount)
+	}
+}
+
+func getStmtSummaryWindowMetricsLocked(typ string) (prometheus.Gauge, prometheus.Gauge) {
+	stmtSummaryWindowMetricsMu.Lock()
+	defer stmtSummaryWindowMetricsMu.Unlock()
+
+	switch typ {
+	case StmtSummaryTypeV1:
+		if stmtSummaryWindowRecordCountV1 == nil {
+			stmtSummaryWindowRecordCountV1 = StmtSummaryWindowRecordCount.WithLabelValues(typ)
+			stmtSummaryWindowEvictedCountV1 = StmtSummaryWindowEvictedCount.WithLabelValues(typ)
+		}
+		return stmtSummaryWindowRecordCountV1, stmtSummaryWindowEvictedCountV1
+	case StmtSummaryTypeV2:
+		if stmtSummaryWindowRecordCountV2 == nil {
+			stmtSummaryWindowRecordCountV2 = StmtSummaryWindowRecordCount.WithLabelValues(typ)
+			stmtSummaryWindowEvictedCountV2 = StmtSummaryWindowEvictedCount.WithLabelValues(typ)
+		}
+		return stmtSummaryWindowRecordCountV2, stmtSummaryWindowEvictedCountV2
+	default:
+		return nil, nil
+	}
+}
diff --git a/pkg/sessionctx/sessionstates/session_states_test.go b/pkg/sessionctx/sessionstates/session_states_test.go
index 2caea78e5575b..5b23b0e8ce091 100644
--- a/pkg/sessionctx/sessionstates/session_states_test.go
+++ b/pkg/sessionctx/sessionstates/session_states_test.go
@@ -684,6 +684,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/variable/session.go b/pkg/sessionctx/variable/session.go
index 340ab1355452a..1c8b051e5d4fd 100644
--- a/pkg/sessionctx/variable/session.go
+++ b/pkg/sessionctx/variable/session.go
@@ -3062,7 +3062,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 92b73c4b6ed2f..671215f328337 100644
--- a/pkg/sessionctx/variable/sysvar.go
+++ b/pkg/sessionctx/variable/sysvar.go
@@ -880,6 +880,14 @@ var defaultSysVars = []*SysVar{
 		SetGlobal: func(_ context.Context, s *SessionVars, val string) error {
 			return stmtsummaryv2.SetMaxSQLLength(TidbOptInt(val, DefTiDBStmtSummaryMaxSQLLength))
 		}},
+	{Scope: ScopeGlobal, Name: TiDBStmtSummaryPersistEvicted, Value: BoolToOnOff(DefTiDBStmtSummaryPersistEvicted), Type: TypeBool, AllowEmpty: true,
+		SetGlobal: func(_ context.Context, s *SessionVars, val string) error {
+			return stmtsummaryv2.SetPersistEvicted(TiDBOptOn(val))
+		}},
+	{Scope: ScopeGlobal, Name: TiDBStmtSummaryGroupByUser, Value: BoolToOnOff(DefTiDBStmtSummaryGroupByUser), Type: TypeBool, AllowEmpty: true,
+		SetGlobal: func(_ context.Context, s *SessionVars, val string) error {
+			return stmtsummaryv2.SetGroupByUser(TiDBOptOn(val))
+		}},
 	{Scope: ScopeGlobal, Name: TiDBCapturePlanBaseline, Value: DefTiDBCapturePlanBaseline, Type: TypeBool, AllowEmptyAll: true},
 	{Scope: ScopeGlobal, Name: TiDBEvolvePlanTaskMaxTime, Value: strconv.Itoa(DefTiDBEvolvePlanTaskMaxTime), Type: TypeInt, MinValue: -1, MaxValue: math.MaxInt64},
 	{Scope: ScopeGlobal, Name: TiDBEvolvePlanTaskStartTime, Value: DefTiDBEvolvePlanTaskStartTime, Type: TypeTime},
diff --git a/pkg/sessionctx/variable/tidb_vars.go b/pkg/sessionctx/variable/tidb_vars.go
index f730d61d4fb67..1f856c3d61899 100644
--- a/pkg/sessionctx/variable/tidb_vars.go
+++ b/pkg/sessionctx/variable/tidb_vars.go
@@ -667,6 +667,12 @@ 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 are persisted to the statement summary log.
+	TiDBStmtSummaryPersistEvicted = "tidb_stmt_summary_persist_evicted"
+
+	// TiDBStmtSummaryGroupByUser indicates whether statement summaries are grouped by executing user.
+	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"
 
@@ -1510,6 +1516,8 @@ const (
 	DefTiDBStmtSummaryHistorySize                     = 24
 	DefTiDBStmtSummaryMaxStmtCount                    = 3000
 	DefTiDBStmtSummaryMaxSQLLength                    = 4096
+	DefTiDBStmtSummaryPersistEvicted                  = false
+	DefTiDBStmtSummaryGroupByUser                     = false
 	DefTiDBCapturePlanBaseline                        = Off
 	DefTiDBIgnoreInlistPlanDigest                     = true
 	DefTiDBEnableIndexMerge                           = true
diff --git a/pkg/util/stmtsummary/BUILD.bazel b/pkg/util/stmtsummary/BUILD.bazel
index 675df9fd991e5..c2be72944348a 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 = 24,
+    shard_count = 26,
     deps = [
         "//pkg/meta/model",
         "//pkg/parser/auth",
diff --git a/pkg/util/stmtsummary/evicted_test.go b/pkg/util/stmtsummary/evicted_test.go
index ed80c6af12b7d..ae022858eee3f 100644
--- a/pkg/util/stmtsummary/evicted_test.go
+++ b/pkg/util/stmtsummary/evicted_test.go
@@ -50,7 +50,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
 }
@@ -189,7 +189,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 2ac2a007577b2..9410b3910f196 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"
 	"maps"
 	"math"
@@ -54,8 +55,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 {
@@ -66,6 +75,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
@@ -301,6 +317,7 @@ func newStmtSummaryByDigestMap() *stmtSummaryByDigestMap {
 		optRefreshInterval:     atomic2.NewInt64(1800),
 		optHistorySize:         atomic2.NewInt32(24),
 		optMaxSQLLength:        atomic2.NewInt32(4096),
+		optGroupByUser:         atomic2.NewBool(false),
 		other:                  ssbde,
 	}
 	newSsMap.summaryMap.SetOnEvict(func(k kvcache.Key, v kvcache.Value) {
@@ -330,8 +347,6 @@ func (ssMap *stmtSummaryByDigestMap) AddStatement(sei *StmtExecInfo) {
 	historySize := ssMap.historySize()
 
 	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
 	// Enclose the block in a function to ensure the lock will always be released.
@@ -347,6 +362,15 @@ func (ssMap *stmtSummaryByDigestMap) AddStatement(sei *StmtExecInfo) {
 			return nil, 0
 		}
 
+		// 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)
+
 		if ssMap.beginTimeForCurInterval+intervalSeconds <= now {
 			// `beginTimeForCurInterval` is a multiple of intervalSeconds, so that when the interval is a multiple
 			// of 60 (or 600, 1800, 3600, etc), begin time shows 'XX:XX:00', not 'XX:XX:01'~'XX:XX:59'.
@@ -381,6 +405,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
@@ -507,6 +536,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 64bb4e0b78abf..07369c671a2b3 100644
--- a/pkg/util/stmtsummary/statement_summary_test.go
+++ b/pkg/util/stmtsummary/statement_summary_test.go
@@ -78,7 +78,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{
@@ -443,7 +443,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)
@@ -453,7 +453,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)
@@ -463,7 +463,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)
@@ -484,7 +484,7 @@ func TestAddStatement(t *testing.T) {
 		planDigest:  "",
 	}
 	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)
@@ -1029,7 +1029,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)
@@ -1073,7 +1073,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)
 
@@ -1277,7 +1277,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)
@@ -1324,7 +1324,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)
@@ -1393,7 +1393,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)
@@ -1408,7 +1408,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)
 }
@@ -1421,7 +1421,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)
@@ -1467,7 +1467,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)
@@ -1518,3 +1518,81 @@ func TestAccessPrivilege(t *testing.T) {
 	datums = reader.GetStmtSummaryHistoryRows()
 	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 3a2d48bae7f0d..4ee01e1d8159b 100644
--- a/pkg/util/stmtsummary/v2/BUILD.bazel
+++ b/pkg/util/stmtsummary/v2/BUILD.bazel
@@ -14,6 +14,7 @@ go_library(
     deps = [
         "//pkg/config",
         "//pkg/meta/model",
+        "//pkg/metrics",
         "//pkg/parser/auth",
         "//pkg/parser/mysql",
         "//pkg/sessionctx/stmtctx",
@@ -47,16 +48,22 @@ go_test(
     ],
     embed = [":stmtsummary"],
     flaky = True,
-    shard_count = 13,
+    shard_count = 18,
     deps = [
         "//pkg/meta/model",
+        "//pkg/metrics",
         "//pkg/parser/auth",
         "//pkg/parser/model",
         "//pkg/testkit/testsetup",
         "//pkg/types",
         "//pkg/util",
         "//pkg/util/set",
+        "//pkg/util/stmtsummary",
+        "@com_github_prometheus_client_golang//prometheus",
+        "@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..8ab8a7abd59c6 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,8 +72,44 @@ 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)
+	b, err := marshalStmtRecord(r)
 	if err != nil {
 		logutil.BgLogger().Warn("failed to marshal statement summary", zap.Error(err))
 		return
@@ -79,6 +117,14 @@ func (s *stmtLogStorage) log(r *StmtRecord) {
 	s.logger.Info(string(b))
 }
 
+func marshalStmtRecord(r *StmtRecord) ([]byte, error) {
+	return json.Marshal(r)
+}
+
+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 cf5dded90c7ad..7e3b1cb314fc7 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 fb4d9cb712fdd..c4c78a140c3b0 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"
@@ -77,4 +78,17 @@ func TestStmtRecord(t *testing.T) {
 	require.Equal(t, info.RUDetail.RUWaitDuration()*2, record2.SumRUWaitDuration)
 	require.Equal(t, info.CPUUsages.TidbCPUTime*2, record2.SumTidbCPU)
 	require.Equal(t, info.CPUUsages.TikvCPUTime*2, record2.SumTikvCPU)
+
+	b, err := marshalStmtRecord(record2)
+	require.NoError(t, err)
+	items := make(map[string]any)
+	require.NoError(t, json.Unmarshal(b, &items))
+	require.Equal(t, record2.Digest, items["digest"])
+
+	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 5ce7c7da8fb59..751c4a96beca7 100644
--- a/pkg/util/stmtsummary/v2/stmtsummary.go
+++ b/pkg/util/stmtsummary/v2/stmtsummary.go
@@ -25,6 +25,7 @@ import (
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/pkg/config"
+	"github.com/pingcap/tidb/pkg/metrics"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util/kvcache"
@@ -41,6 +42,17 @@ const (
 	defaultMaxSQLLength        = 4096
 	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 +98,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 +132,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 +142,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 +173,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 +271,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 +315,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 +387,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 +406,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 {
@@ -390,14 +477,25 @@ func (s *StmtSummary) rotateLoop() {
 			if now.After(s.window.begin.Add(time.Duration(s.RefreshInterval()) * time.Second)) {
 				s.rotate(now)
 			}
+			s.updateMetrics()
 			s.windowLock.Unlock()
 		}
 	}
 }
 
+// updateMetrics reports the current window's record count and eviction count
+// to Prometheus gauges. Must be called with windowLock held.
+func (s *StmtSummary) updateMetrics() {
+	metrics.SetStmtSummaryWindowMetrics(
+		metrics.StmtSummaryTypeV2,
+		float64(s.window.lru.Size()),
+		float64(s.window.evictedCount.Load()),
+	)
+}
+
 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.
@@ -409,27 +507,161 @@ 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
 // into stmtEvicted.
 type stmtWindow struct {
-	begin   time.Time
-	lru     *kvcache.SimpleLRUCache // *StmtDigestKey => *lockedStmtRecord
-	evicted *stmtEvicted
+	begin        time.Time
+	lru          *kvcache.SimpleLRUCache // *StmtDigestKey => *lockedStmtRecord
+	evicted      *stmtEvicted
+	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),
 		evicted: newStmtEvicted(),
 	}
 	w.lru.SetOnEvict(func(k kvcache.Key, v kvcache.Value) {
+		w.evictedCount.Add(1)
 		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
 }
@@ -437,32 +669,37 @@ func newStmtWindow(begin time.Time, capacity uint) *stmtWindow {
 func (w *stmtWindow) clear() {
 	w.lru.DeleteAll()
 	w.evicted = newStmtEvicted()
+	w.evictedCount.Store(0)
 }
 
 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.Unix(math.MaxInt64, 0),
-		},
+		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
 	}
@@ -470,6 +707,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 {
@@ -478,6 +718,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
@@ -486,6 +736,7 @@ type lockedStmtRecord struct {
 type mockStmtStorage struct {
 	sync.Mutex
 	windows []*stmtWindow
+	evicted []*StmtRecord
 }
 
 func (s *mockStmtStorage) persist(w *stmtWindow, _ time.Time) {
@@ -494,10 +745,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.
@@ -574,6 +853,28 @@ 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
+}
+
 // GetMoreThanCntBindableStmt wraps GlobalStmtSummary.GetMoreThanCntBindableStmt and
 // stmtsummary.StmtSummaryByDigestMap.GetMoreThanCntBindableStmt.
 func GetMoreThanCntBindableStmt(frequency int64) []*stmtsummary.BindableStmt {
diff --git a/pkg/util/stmtsummary/v2/stmtsummary_test.go b/pkg/util/stmtsummary/v2/stmtsummary_test.go
index cc489195425df..39cc8c53d5647 100644
--- a/pkg/util/stmtsummary/v2/stmtsummary_test.go
+++ b/pkg/util/stmtsummary/v2/stmtsummary_test.go
@@ -15,11 +15,35 @@
 package stmtsummary
 
 import (
+	"bytes"
+	"encoding/json"
+	"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 {
+	t.Helper()
+	m := &dto.Metric{}
+	require.NoError(t, gauge.Write(m))
+	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()
@@ -66,6 +90,200 @@ 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()
+	require.NoError(t, ss.SetMaxStmtCount(2))
+	metrics.SetStmtSummaryWindowMetrics(metrics.StmtSummaryTypeV2, 0, 0)
+	t.Cleanup(func() {
+		metrics.SetStmtSummaryWindowMetrics(metrics.StmtSummaryTypeV2, 0, 0)
+	})
+
+	// Fill the LRU cache and trigger evictions.
+	ss.Add(GenerateStmtExecInfo4Test("digest1"))
+	ss.Add(GenerateStmtExecInfo4Test("digest2"))
+	ss.Add(GenerateStmtExecInfo4Test("digest3")) // evicts digest1
+	ss.Add(GenerateStmtExecInfo4Test("digest4")) // evicts digest2
+	require.Equal(t, 2, ss.window.lru.Size())
+	require.Equal(t, int64(2), ss.window.evictedCount.Load())
+	ss.windowLock.Lock()
+	ss.updateMetrics()
+	ss.windowLock.Unlock()
+	require.Equal(t, 2.0, readGaugeValue(t, metrics.StmtSummaryWindowRecordCount.WithLabelValues(metrics.StmtSummaryTypeV2)))
+	require.Equal(t, 2.0, readGaugeValue(t, metrics.StmtSummaryWindowEvictedCount.WithLabelValues(metrics.StmtSummaryTypeV2)))
+
+	// Rotate creates a new window with a fresh counter.
+	ss.rotate(timeNow())
+	require.Equal(t, int64(0), ss.window.evictedCount.Load())
+	ss.windowLock.Lock()
+	ss.updateMetrics()
+	ss.windowLock.Unlock()
+	require.Equal(t, 0.0, readGaugeValue(t, metrics.StmtSummaryWindowEvictedCount.WithLabelValues(metrics.StmtSummaryTypeV2)))
+
+	// Add more records in the new window.
+	ss.Add(GenerateStmtExecInfo4Test("digest5"))
+	ss.Add(GenerateStmtExecInfo4Test("digest6"))
+	ss.Add(GenerateStmtExecInfo4Test("digest7")) // evicts digest5
+	require.Equal(t, int64(1), ss.window.evictedCount.Load())
+	require.Equal(t, 2, ss.window.lru.Size())
+	ss.windowLock.Lock()
+	ss.updateMetrics()
+	ss.windowLock.Unlock()
+	require.Equal(t, 2.0, readGaugeValue(t, metrics.StmtSummaryWindowRecordCount.WithLabelValues(metrics.StmtSummaryTypeV2)))
+	require.Equal(t, 1.0, readGaugeValue(t, metrics.StmtSummaryWindowEvictedCount.WithLabelValues(metrics.StmtSummaryTypeV2)))
+}
+
 func TestStmtSummaryFlush(t *testing.T) {
 	storage := &mockStmtStorage{}
 	ss := NewStmtSummary4Test(1000)