session,store: remove RUv2 rpc interceptor (#67508)

DEPS.bzl

@@ -7805,13 +7805,13 @@ def go_deps():
         build_tags = ["nextgen", "intest"],
         build_file_proto_mode = "disable_global",
         importpath = "github.com/tikv/client-go/v2",
-        sha256 = "46ee8c64e0f95ad1514e824506e49ea1d9f75329ee6419aae4b0817336550557",
-        strip_prefix = "github.com/tikv/client-go/v2@v2.0.8-0.20260401083018-b7f9a9e9d2ab",
+        sha256 = "a5e70899561f437abf34016a1fd907ab6394ab4771f75ba509b61dd5bb806351",
+        strip_prefix = "github.com/tikv/client-go/v2@v2.0.8-0.20260414033830-1adc54c38a51",
         urls = [
-            "http://bazel-cache.pingcap.net:8080/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20260401083018-b7f9a9e9d2ab.zip",
-            "http://ats.apps.svc/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20260401083018-b7f9a9e9d2ab.zip",
-            "https://cache.hawkingrei.com/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20260401083018-b7f9a9e9d2ab.zip",
-            "https://storage.googleapis.com/pingcapmirror/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20260401083018-b7f9a9e9d2ab.zip",
+            "http://bazel-cache.pingcap.net:8080/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20260414033830-1adc54c38a51.zip",
+            "http://ats.apps.svc/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20260414033830-1adc54c38a51.zip",
+            "https://cache.hawkingrei.com/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20260414033830-1adc54c38a51.zip",
+            "https://storage.googleapis.com/pingcapmirror/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20260414033830-1adc54c38a51.zip",
         ],
     )
     go_repository(

go.mod

@@ -124,8 +124,8 @@ require (
 	github.com/stathat/consistent v1.0.0
 	github.com/stretchr/testify v1.11.1
 	github.com/tiancaiamao/appdash v0.0.0-20181126055449-889f96f722a2
-	github.com/tikv/client-go/v2 v2.0.8-0.20260401083018-b7f9a9e9d2ab
-	github.com/tikv/pd/client v0.0.0-20260401072359-048f0d8f6f71
+	github.com/tikv/client-go/v2 v2.0.8-0.20260414033830-1adc54c38a51
+	github.com/tikv/pd/client v0.0.0-20260404141330-8a6813497b52
 	github.com/timakin/bodyclose v0.0.0-20241222091800-1db5c5ca4d67
 	github.com/twmb/murmur3 v1.1.6
 	github.com/uber/jaeger-client-go v2.22.1+incompatible

go.sum

@@ -895,10 +895,10 @@ github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4=
 github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 github.com/tidwall/sjson v1.2.5 h1:kLy8mja+1c9jlljvWTlSazM7cKDRfJuR/bOJhcY5NcY=
 github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6jwd28=
-github.com/tikv/client-go/v2 v2.0.8-0.20260401083018-b7f9a9e9d2ab h1:t9Kh7tVsSSUMuSPpHChPF6W3VtN4Kq7Gi8EgWPbYRyY=
-github.com/tikv/client-go/v2 v2.0.8-0.20260401083018-b7f9a9e9d2ab/go.mod h1:lfRxHwyBp1rjTmNC04SUZ+dqk7i1R1AeJ2zraMQaNvY=
-github.com/tikv/pd/client v0.0.0-20260401072359-048f0d8f6f71 h1:5hCQ6J2fwUpYqIgQGR625bW98wvYS9FUpTiVszIbVSg=
-github.com/tikv/pd/client v0.0.0-20260401072359-048f0d8f6f71/go.mod h1:4kxXuAQAREpH+lVbydVwGNNDmcwdj0RG4Ofwky08W/k=
+github.com/tikv/client-go/v2 v2.0.8-0.20260414033830-1adc54c38a51 h1:HiKahyqWSI5aijRsSk1JciFg61D5a0x3TFyleYERBl4=
+github.com/tikv/client-go/v2 v2.0.8-0.20260414033830-1adc54c38a51/go.mod h1:PILS4Yr8mWPD7J6W0+hVq4Z+lwhTIYxPYUA/OTxPSvg=
+github.com/tikv/pd/client v0.0.0-20260404141330-8a6813497b52 h1:fXIMowblD3qdfHXJYGJpe7SbBlTO4S9GPVZZvL3CPG8=
+github.com/tikv/pd/client v0.0.0-20260404141330-8a6813497b52/go.mod h1:I2yRx/Yf8Y8kgM5f3VNp4a8fWpnjPC4TxWk554AY8bM=
 github.com/timakin/bodyclose v0.0.0-20241222091800-1db5c5ca4d67 h1:9LPGD+jzxMlnk5r6+hJnar67cgpDIz/iyD+rfl5r2Vk=
 github.com/timakin/bodyclose v0.0.0-20241222091800-1db5c5ca4d67/go.mod h1:mkjARE7Yr8qU23YcGMSALbIxTQ9r9QBVahQOBRfU460=
 github.com/tjfoc/gmsm v1.3.2/go.mod h1:HaUcFuY0auTiaHB9MHFGCPx5IaLhTUd2atbCFBQXn9w=

pkg/ddl/backfilling_operators.go

@@ -51,6 +51,7 @@ import (
 	"github.com/pingcap/tidb/pkg/util/chunk"
 	contextutil "github.com/pingcap/tidb/pkg/util/context"
 	"github.com/pingcap/tidb/pkg/util/dbterror"
+	"github.com/pingcap/tidb/pkg/util/execdetails"
 	"github.com/pingcap/tidb/pkg/util/intest"
 	"github.com/pingcap/tidb/pkg/util/logutil"
 	"github.com/pingcap/tidb/pkg/util/size"
@@ -585,7 +586,7 @@ func (w *tableScanWorker) scanRecords(task TableScanTask, sender func(IndexRecor
 				terror.Call(rs.Close)
 				return err
 			}
-			w.collector.Accepted(execDetails.UnpackedBytesReceivedKVTotal)
+			w.collector.Accepted(execdetails.LoadTiKVExecDetails(&execDetails).UnpackedBytesReceivedKVTotal)
 			execDetails = kvutil.ExecDetails{}
 
 			_, tableScanRowCount := distsqlCtx.RuntimeStatsColl.GetCopCountAndRows(tableScanCopID)

pkg/distsql/context/BUILD.bazel

@@ -18,7 +18,6 @@ go_library(
         "//pkg/util/tiflash",
         "//pkg/util/topsql/stmtstats",
         "@com_github_tikv_client_go_v2//kv",
-        "@com_github_tikv_client_go_v2//tikvrpc/interceptor",
         "@org_uber_go_atomic//:atomic",
     ],
 )
@@ -43,7 +42,6 @@ go_test(
         "//pkg/util/tiflash",
         "//pkg/util/topsql/stmtstats",
         "@com_github_tikv_client_go_v2//kv",
-        "@com_github_tikv_client_go_v2//tikvrpc/interceptor",
         "@org_uber_go_atomic//:atomic",
     ],
 )

pkg/distsql/context/context.go

@@ -29,7 +29,6 @@ import (
 	"github.com/pingcap/tidb/pkg/util/tiflash"
 	"github.com/pingcap/tidb/pkg/util/topsql/stmtstats"
 	tikvstore "github.com/tikv/client-go/v2/kv"
-	"github.com/tikv/client-go/v2/tikvrpc/interceptor"
 	"go.uber.org/atomic"
 )
 
@@ -46,7 +45,6 @@ type DistSQLContext struct {
 	KVVars                 *tikvstore.Variables
 	KvExecCounter          *stmtstats.KvExecCounter
 	RUV2Metrics            *execdetails.RUV2Metrics
-	RUV2RPCInterceptor     interceptor.RPCInterceptor
 	SessionMemTracker      *memory.Tracker
 
 	Location         *time.Location

pkg/distsql/context/context_test.go

@@ -32,7 +32,6 @@ import (
 	"github.com/pingcap/tidb/pkg/util/tiflash"
 	"github.com/pingcap/tidb/pkg/util/topsql/stmtstats"
 	tikvstore "github.com/tikv/client-go/v2/kv"
-	"github.com/tikv/client-go/v2/tikvrpc/interceptor"
 	"go.uber.org/atomic"
 )
 
@@ -55,10 +54,7 @@ func TestContextDetach(t *testing.T) {
 		KVVars:                 kvVars,
 		KvExecCounter:          &stmtstats.KvExecCounter{},
 		RUV2Metrics:            execdetails.NewRUV2Metrics(),
-		RUV2RPCInterceptor: interceptor.NewRPCInterceptor("test", func(next interceptor.RPCInterceptorFunc) interceptor.RPCInterceptorFunc {
-			return next
-		}),
-		SessionMemTracker: &memory.Tracker{},
+		SessionMemTracker:      &memory.Tracker{},
 
 		Location:         time.Local,
 		RuntimeStatsColl: &execdetails.RuntimeStatsColl{},
@@ -113,7 +109,6 @@ func TestContextDetach(t *testing.T) {
 			"$.RunawayChecker",
 			"$.RUConsumptionReporter",
 			"$.ExecDetails",
-			"$.RUV2RPCInterceptor",
 		}))
 
 	staticObj := obj.Detach()
@@ -130,7 +125,6 @@ func TestContextDetach(t *testing.T) {
 			"$.RUConsumptionReporter",
 			"$.ExecDetails",
 			"$.RUV2Metrics",
-			"$.RUV2RPCInterceptor",
 			"$.KVVars.Killed",
 			"$.KvExecCounter",
 			"$.SessionMemTracker",

pkg/distsql/distsql.go

@@ -78,7 +78,6 @@ func Select(ctx context.Context, dctx *distsqlctx.DistSQLContext, kvReq *kv.Requ
 	}
 
 	ctx = WithSQLKvExecCounterInterceptor(ctx, dctx.KvExecCounter)
-	ctx = WithRUV2MetricsInterceptor(ctx, dctx.RUV2RPCInterceptor)
 	option := &kv.ClientSendOption{
 		SessionMemTracker:          dctx.SessionMemTracker,
 		EnabledRateLimitAction:     enabledRateLimitAction,
@@ -178,7 +177,6 @@ func SelectWithRuntimeStats(ctx context.Context, dctx *distsqlctx.DistSQLContext
 func Analyze(ctx context.Context, client kv.Client, kvReq *kv.Request, vars any,
 	isRestrict bool, dctx *distsqlctx.DistSQLContext) (SelectResult, error) {
 	ctx = WithSQLKvExecCounterInterceptor(ctx, dctx.KvExecCounter)
-	ctx = WithRUV2MetricsInterceptor(ctx, dctx.RUV2RPCInterceptor)
 	failpoint.Inject("mockAnalyzeRequestWaitForCancel", func(val failpoint.Value) {
 		if val.(bool) {
 			<-ctx.Done()
@@ -290,12 +288,3 @@ func WithSQLKvExecCounterInterceptor(ctx context.Context, counter *stmtstats.KvE
 	}
 	return ctx
 }
-
-// WithRUV2MetricsInterceptor binds an interceptor for client-go to collect
-// statement-level RUv2 request counters and TiKV ExecDetailsV2-based metrics.
-func WithRUV2MetricsInterceptor(ctx context.Context, ruv2Interceptor interceptor.RPCInterceptor) context.Context {
-	if ruv2Interceptor != nil {
-		return interceptor.WithRPCInterceptor(ctx, ruv2Interceptor)
-	}
-	return ctx
-}

pkg/executor/BUILD.bazel

@@ -442,6 +442,7 @@ go_test(
         "//pkg/executor/internal/testutil",
         "//pkg/executor/join",
         "//pkg/executor/sortexec",
+        "//pkg/executor/staticrecordset",
         "//pkg/executor/unionexec",
         "//pkg/expression",
         "//pkg/expression/aggregation",

pkg/executor/adapter.go

@@ -185,8 +185,10 @@ func (a *recordSet) Next(ctx context.Context, req *chunk.Chunk) (err error) {
 	if len(a.traceID) > 0 {
 		ctx = tikvtrace.ContextWithTraceID(ctx, a.traceID)
 	}
+	ctx = inheritStmtRUV2Context(ctx, a.stmt)
 
 	err = a.stmt.next(ctx, a.executor, req)
+	syncRUV2MetricsAfterExec(a.stmt)
 	if err != nil {
 		a.lastErrs = append(a.lastErrs, err)
 		return err
@@ -204,6 +206,25 @@ func (a *recordSet) Next(ctx context.Context, req *chunk.Chunk) (err error) {
 	return nil
 }
 
+func inheritStmtRUV2Context(ctx context.Context, stmt *ExecStmt) context.Context {
+	if stmt == nil || stmt.GoCtx == nil {
+		return ctx
+	}
+	return execdetails.ContextWithInheritedRUV2Details(ctx, stmt.GoCtx)
+}
+
+func syncRUV2MetricsAfterExec(stmt *ExecStmt) {
+	if stmt == nil || stmt.GoCtx == nil {
+		return
+	}
+	sessVars := stmt.Ctx.GetSessionVars()
+	if sessVars.RUV2Metrics == nil {
+		return
+	}
+	ruDetail, _ := stmt.GoCtx.Value(util.RUDetailsCtxKey).(*util.RUDetails)
+	execdetails.SyncRUV2MetricsFromRUDetails(sessVars.RUV2Metrics, ruDetail)
+}
+
 // NewChunk create a chunk base on top-level executor's exec.NewFirstChunk().
 func (a *recordSet) NewChunk(alloc chunk.Allocator) *chunk.Chunk {
 	if alloc == nil {
@@ -259,7 +280,7 @@ func (a *recordSet) TryDetach() (sqlexec.RecordSet, bool, error) {
 	if !ok {
 		return nil, false, nil
 	}
-	return staticrecordset.New(a.Fields(), e, a.stmt.GetTextToLog(false)), true, nil
+	return staticrecordset.New(a.Fields(), e, a.stmt.GetTextToLog(false), a.stmt.GoCtx), true, nil
 }
 
 // GetExecutor4Test exports the internal executor for test purpose.
@@ -1110,6 +1131,7 @@ func (a *ExecStmt) handleNoDelayExecutor(ctx context.Context, e exec.Executor) (
 	}
 
 	err = a.next(ctx, e, exec.TryNewCacheChunk(e))
+	syncRUV2MetricsAfterExec(a)
 	if err != nil {
 		return nil, err
 	}
@@ -1717,6 +1739,7 @@ func (a *ExecStmt) finalizeStatementRUV2Metrics() {
 	if ruDetail == nil {
 		return
 	}
+	execdetails.SyncRUV2MetricsFromRUDetails(sessVars.RUV2Metrics, ruDetail)
 
 	weights := sessVars.RUV2Weights()
 	tidbRU := sessVars.RUV2Metrics.CalculateRUValues(weights)
@@ -1951,7 +1974,7 @@ func (a *ExecStmt) updateNetworkTrafficStatsAndMetrics() {
 	hasMPPTraffic := a.updateMPPNetworkTraffic()
 	tikvExecDetailRaw := a.GoCtx.Value(util.ExecDetailsKey)
 	if tikvExecDetailRaw != nil {
-		tikvExecDetail := tikvExecDetailRaw.(*util.ExecDetails)
+		tikvExecDetail := execdetails.LoadTiKVExecDetails(tikvExecDetailRaw.(*util.ExecDetails))
 		executor_metrics.ExecutorNetworkTransmissionSentTiKVTotal.Add(float64(tikvExecDetail.UnpackedBytesSentKVTotal))
 		executor_metrics.ExecutorNetworkTransmissionSentTiKVCrossZone.Add(float64(tikvExecDetail.UnpackedBytesSentKVCrossZone))
 		executor_metrics.ExecutorNetworkTransmissionReceivedTiKVTotal.Add(float64(tikvExecDetail.UnpackedBytesReceivedKVTotal))

pkg/executor/adapter_test.go

@@ -20,10 +20,12 @@ import (
 	"slices"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/pingcap/failpoint"
+	"github.com/pingcap/kvproto/pkg/kvrpcpb"
 	rmpb "github.com/pingcap/kvproto/pkg/resource_manager"
 	"github.com/pingcap/tidb/pkg/config"
 	distsqlctx "github.com/pingcap/tidb/pkg/distsql/context"
@@ -453,7 +455,7 @@ func TestWriteSlowLog(t *testing.T) {
 	checkWriteSlowLog(true)
 }
 
-func TestFinishExecuteStmtReportsTiDBRUV2WithoutSyncingRUDetails(t *testing.T) {
+func TestFinishExecuteStmtSyncsTiDBRUV2FromRUDetails(t *testing.T) {
 	original := config.GetGlobalConfig()
 	originalGenerateBinaryPlan := variable.GenerateBinaryPlan.Load()
 	t.Cleanup(func() {
@@ -488,11 +490,19 @@ func TestFinishExecuteStmtReportsTiDBRUV2WithoutSyncingRUDetails(t *testing.T) {
 	sessVars.RUV2Metrics.AddResultChunkCells(100)
 	sessVars.RUV2Metrics.AddPlanCnt(2)
 	sessVars.RUV2Metrics.AddSessionParserTotal(3)
-
-	expected := sessVars.RUV2Metrics.CalculateRUValues(sessVars.RUV2Weights())
 	ruDetails := goCtx.Value(util.RUDetailsCtxKey).(*util.RUDetails)
 	ruDetails.AddTiKVRUV2(23456)
+	rawRUV2 := &kvrpcpb.RUV2{
+		ReadRpcCount:                 5,
+		WriteRpcCount:                7,
+		StorageProcessedKeysBatchGet: 11,
+	}
+	ruDetails.AddRUV2(rawRUV2)
 	ruDetails.UpdateTiFlash(&rmpb.Consumption{RRU: 345, WRU: 67})
+	// Build expected metrics by cloning the current state and manually adding
+	// the RUV2 counters (without draining ruDetails, since FinishExecuteStmt will drain).
+	expected := sessVars.RUV2Metrics.Clone()
+	execdetails.UpdateRUV2MetricsFromRUV2(expected, rawRUV2)
 
 	execStmt := &executor.ExecStmt{
 		Ctx:      ctx,
@@ -502,9 +512,12 @@ func TestFinishExecuteStmtReportsTiDBRUV2WithoutSyncingRUDetails(t *testing.T) {
 	execStmt.FinishExecuteStmt(0, nil, false)
 
 	require.Equal(t, float64(23456), ruDetails.TiKVRUV2())
+	require.Equal(t, int64(5), sessVars.RUV2Metrics.ResourceManagerReadCnt())
+	require.Equal(t, int64(7), sessVars.RUV2Metrics.ResourceManagerWriteCnt())
+	require.Equal(t, int64(11), sessVars.RUV2Metrics.TiKVStorageProcessedKeysBatchGet())
 	require.Equal(t, "rg1", reporter.group)
 	require.Equal(t, float64(23456), reporter.tikvRUV2)
-	require.Equal(t, expected, reporter.tidbRUV2)
+	require.Equal(t, expected.CalculateRUValues(sessVars.RUV2Weights()), reporter.tidbRUV2)
 	require.Equal(t, float64(412), reporter.tiflashRU)
 
 	t.Run("stmt summary ignores optimistic autocommit retry count", func(t *testing.T) {
@@ -569,6 +582,60 @@ func TestFinishExecuteStmtReportsTiDBRUV2WithoutSyncingRUDetails(t *testing.T) {
 		require.Zero(t, reporter.tidbRUV2)
 		require.Zero(t, reporter.tiflashRU)
 	})
+
+	t.Run("network traffic stats are read atomically", func(t *testing.T) {
+		reporter := &mockRUV2ConsumptionReporter{}
+		ctx := &mockRUV2ReportingContext{
+			Context:  mock.NewContext(),
+			reporter: reporter,
+		}
+		sessVars := ctx.GetSessionVars()
+		sessVars.StartTime = time.Now()
+		sessVars.StmtCtx.StmtType = "Select"
+		sessVars.StmtCtx.OriginalSQL = "select 1"
+		sessVars.StmtCtx.ResetSQLDigest(sessVars.StmtCtx.OriginalSQL)
+		sessVars.RUV2Metrics = execdetails.NewRUV2Metrics()
+
+		goCtx := execdetails.ContextWithInitializedExecDetails(context.Background())
+		tikvExecDetail := goCtx.Value(util.ExecDetailsKey).(*util.ExecDetails)
+		execStmt := &executor.ExecStmt{
+			Ctx:      ctx,
+			GoCtx:    goCtx,
+			StmtNode: &ast.SelectStmt{},
+		}
+
+		done := make(chan struct{})
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case <-done:
+					return
+				default:
+					atomic.AddInt64(&tikvExecDetail.WaitKVRespDuration, int64(time.Millisecond))
+					atomic.AddInt64(&tikvExecDetail.WaitPDRespDuration, int64(time.Millisecond))
+					atomic.AddInt64(&tikvExecDetail.BackoffDuration, int64(time.Millisecond))
+					atomic.AddInt64(&tikvExecDetail.UnpackedBytesSentKVTotal, 1)
+					atomic.AddInt64(&tikvExecDetail.UnpackedBytesReceivedKVTotal, 1)
+					atomic.AddInt64(&tikvExecDetail.UnpackedBytesSentKVCrossZone, 1)
+					atomic.AddInt64(&tikvExecDetail.UnpackedBytesReceivedKVCrossZone, 1)
+					atomic.AddInt64(&tikvExecDetail.UnpackedBytesSentMPPTotal, 1)
+					atomic.AddInt64(&tikvExecDetail.UnpackedBytesReceivedMPPTotal, 1)
+					atomic.AddInt64(&tikvExecDetail.UnpackedBytesSentMPPCrossZone, 1)
+					atomic.AddInt64(&tikvExecDetail.UnpackedBytesReceivedMPPCrossZone, 1)
+				}
+			}
+		}()
+
+		for range 64 {
+			execStmt.FinishExecuteStmt(0, nil, false)
+		}
+
+		close(done)
+		wg.Wait()
+	})
 }
 
 func TestSlowLogMaxPerSec(t *testing.T) {

pkg/executor/explain.go

@@ -143,7 +143,7 @@ func (e *ExplainExec) executeAnalyzeExec(ctx context.Context) (err error) {
 			if ruDetailsRaw != nil {
 				ruDetails = ruDetailsRaw.(*clientutil.RUDetails).Clone()
 			}
-			ruv2Metrics := execdetails.RUV2MetricsFromContext(ctx)
+			ruv2Metrics := execdetails.SyncRUV2MetricsFromContext(ctx)
 			if ruDetails != nil || ruv2Metrics != nil {
 				ruVersion := rmclient.DefaultRUVersion
 				if do := domain.GetDomain(e.Ctx()); do != nil {