From 43fe422bd061e69901a856b491b888c1036c3e12 Mon Sep 17 00:00:00 2001 From: JmPotato Date: Wed, 1 Jul 2026 13:20:21 +0800 Subject: [PATCH 1/3] *: add tidb_paging_size_bytes and forward it to coprocessor RPC (#68091) close pingcap/tidb#68090 (cherry picked from commit 4ccb1aee018d416c212a9e2a40fb22870e317c9d) Signed-off-by: JmPotato --- DEPS.bzl | 4 +- go.mod | 2 +- go.sum | 8 +-- pkg/distsql/BUILD.bazel | 2 +- pkg/distsql/context/context.go | 1 + pkg/distsql/context/context_test.go | 1 + pkg/distsql/distsql.go | 2 +- pkg/distsql/request_builder.go | 1 + pkg/distsql/request_builder_test.go | 13 +++++ pkg/executor/distsql.go | 1 + pkg/kv/kv.go | 3 ++ pkg/session/session.go | 9 ++++ pkg/session/tidb_test.go | 32 ++++++++++++ pkg/sessionctx/vardef/tidb_vars.go | 5 ++ pkg/sessionctx/variable/session.go | 5 ++ pkg/sessionctx/variable/setvar_affect.go | 1 + pkg/sessionctx/variable/sysvar.go | 4 ++ pkg/store/copr/BUILD.bazel | 2 +- pkg/store/copr/copr_test/coprocessor_test.go | 40 +++++++++++++-- pkg/store/copr/coprocessor.go | 20 +++++--- pkg/store/copr/coprocessor_cache.go | 10 ++-- pkg/store/copr/coprocessor_cache_test.go | 43 ++++++++++++++++ pkg/store/copr/coprocessor_test.go | 51 +++++++++++++++++++ .../r/sessionctx/setvar.result | 22 ++++++++ .../integrationtest/t/sessionctx/setvar.test | 10 ++++ 25 files changed, 268 insertions(+), 24 deletions(-) diff --git a/DEPS.bzl b/DEPS.bzl index c2d9e319c7ce0..c042969418846 100644 --- a/DEPS.bzl +++ b/DEPS.bzl @@ -3519,8 +3519,8 @@ def go_deps(): name = "com_github_pingcap_kvproto", build_file_proto_mode = "disable_global", importpath = "github.com/pingcap/kvproto", - sum = "h1:fnagXtVQTLmi2sLH7QgvH1vHlFohDeQtRASxzXVk2ss=", - version = "v0.0.0-20260703015547-e800f9e5a427", + sum = "h1:PJIfTrhcVitRq5nJmzTDCV75YpaDJc+YKMtqgyBfH84=", + version = "v0.0.0-20260716111302-363178052f57", ) go_repository( name = "com_github_pingcap_log", diff --git a/go.mod b/go.mod index 74fd99e41872d..2a8d4bf5a7e9e 100644 --- a/go.mod +++ b/go.mod @@ -100,7 +100,7 @@ require ( github.com/pingcap/errors v0.11.5-0.20260508054701-306e305bcf41 github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 github.com/pingcap/fn v1.0.0 - github.com/pingcap/kvproto v0.0.0-20260703015547-e800f9e5a427 + github.com/pingcap/kvproto v0.0.0-20260716111302-363178052f57 github.com/pingcap/log v1.1.1-0.20250917021125-19901e015dc9 github.com/pingcap/metering_sdk v0.0.0-20260324055927-14fead745f1d github.com/pingcap/sysutil v1.0.1-0.20240311050922-ae81ee01f3a5 diff --git a/go.sum b/go.sum index e2cd71d5eeb3f..e2dc731b89029 100644 --- a/go.sum +++ b/go.sum @@ -726,10 +726,8 @@ github.com/pingcap/fn v1.0.0/go.mod h1:u9WZ1ZiOD1RpNhcI42RucFh/lBuzTu6rw88a+oF2Z github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989 h1:surzm05a8C9dN8dIUmo4Be2+pMRb6f55i+UIYrluu2E= github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989/go.mod h1:O17XtbryoCJhkKGbT62+L2OlrniwqiGLSqrmdHCMzZw= github.com/pingcap/kvproto v0.0.0-20191211054548-3c6b38ea5107/go.mod h1:WWLmULLO7l8IOcQG+t+ItJ3fEcrL5FxF0Wu+HrMy26w= -github.com/pingcap/kvproto v0.0.0-20260610100801-904083a187d5 h1:OgjuwE14rgqWIPSMyPvR4fMI9zA3TB7DZ535w/58yk4= -github.com/pingcap/kvproto v0.0.0-20260610100801-904083a187d5/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8= -github.com/pingcap/kvproto v0.0.0-20260703015547-e800f9e5a427 h1:fnagXtVQTLmi2sLH7QgvH1vHlFohDeQtRASxzXVk2ss= -github.com/pingcap/kvproto v0.0.0-20260703015547-e800f9e5a427/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8= +github.com/pingcap/kvproto v0.0.0-20260716111302-363178052f57 h1:PJIfTrhcVitRq5nJmzTDCV75YpaDJc+YKMtqgyBfH84= +github.com/pingcap/kvproto v0.0.0-20260716111302-363178052f57/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8= github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuRurdGxZXBz0At+9avep+ub7U1AGYLIMM= github.com/pingcap/log v1.1.0/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4= github.com/pingcap/log v1.1.1-0.20250917021125-19901e015dc9 h1:qG9BSvlWFEE5otQGamuWedx9LRm0nrHvsQRQiW8SxEs= @@ -897,8 +895,6 @@ 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.20260617064024-5526ae2a8546 h1:fjOCdCmNgUrRzwOlemKHnCItUorHKbNvYRxdXlr7KCc= github.com/tikv/client-go/v2 v2.0.8-0.20260617064024-5526ae2a8546/go.mod h1:WDAIZI2EmfkDX4BP947c5wXJiIM3GgTnaaCp3TqReBE= -github.com/tikv/pd/client v0.0.0-20260611063303-491a11c2f061 h1:G0s09nW/KYlNi17A5rp2EJ1lBTZsQXcH1qpptejC/5E= -github.com/tikv/pd/client v0.0.0-20260611063303-491a11c2f061/go.mod h1:/AgomU1MKQ1LV0kYcvfLrTX/Prn1jLekzvtRNCqNiO4= github.com/tikv/pd/client v0.0.0-20260714045825-19ee324ee9d8 h1:8oHjHFQea2yL+MUK6Z6d1p0dYI2rHkoCQVjw1Yo6Uss= github.com/tikv/pd/client v0.0.0-20260714045825-19ee324ee9d8/go.mod h1:UL+x0aP+9+BDig+LYZFIU1G5r/gH2fEsVKl5MTS1Ud8= github.com/timakin/bodyclose v0.0.0-20241222091800-1db5c5ca4d67 h1:9LPGD+jzxMlnk5r6+hJnar67cgpDIz/iyD+rfl5r2Vk= diff --git a/pkg/distsql/BUILD.bazel b/pkg/distsql/BUILD.bazel index 05416ed307778..5810c4ef8c6c4 100644 --- a/pkg/distsql/BUILD.bazel +++ b/pkg/distsql/BUILD.bazel @@ -68,7 +68,7 @@ go_test( embed = [":distsql"], flaky = True, race = "on", - shard_count = 33, + shard_count = 34, deps = [ "//pkg/distsql/context", "//pkg/errctx", diff --git a/pkg/distsql/context/context.go b/pkg/distsql/context/context.go index 65796efc7bff5..87b7ae7364b1c 100644 --- a/pkg/distsql/context/context.go +++ b/pkg/distsql/context/context.go @@ -74,6 +74,7 @@ type DistSQLContext struct { EnablePaging bool MinPagingSize int MaxPagingSize int + PagingSizeBytes int RequestSourceType string ExplicitRequestSourceType string StoreBatchSize int diff --git a/pkg/distsql/context/context_test.go b/pkg/distsql/context/context_test.go index 122eba4d8b624..bb6453b50359c 100644 --- a/pkg/distsql/context/context_test.go +++ b/pkg/distsql/context/context_test.go @@ -82,6 +82,7 @@ func TestContextDetach(t *testing.T) { EnablePaging: true, MinPagingSize: 1, MaxPagingSize: 1, + PagingSizeBytes: 1, RequestSourceType: "a", ExplicitRequestSourceType: "b", StoreBatchSize: 1, diff --git a/pkg/distsql/distsql.go b/pkg/distsql/distsql.go index 2f18414d56a80..0880e320bd2d3 100644 --- a/pkg/distsql/distsql.go +++ b/pkg/distsql/distsql.go @@ -128,7 +128,7 @@ func Select(ctx context.Context, dctx *distsqlctx.DistSQLContext, kvReq *kv.Requ sqlType: label, memTracker: kvReq.MemTracker, storeType: kvReq.StoreType, - paging: kvReq.Paging.Enable, + paging: kvReq.Paging.Enable || kvReq.Paging.PagingSizeBytes > 0, distSQLConcurrency: kvReq.Concurrency, }, nil } diff --git a/pkg/distsql/request_builder.go b/pkg/distsql/request_builder.go index c57bc7afed5f7..ad43c3d16433c 100644 --- a/pkg/distsql/request_builder.go +++ b/pkg/distsql/request_builder.go @@ -363,6 +363,7 @@ func (builder *RequestBuilder) SetFromSessionVars(dctx *distsqlctx.DistSQLContex builder.SetPaging(dctx.EnablePaging) builder.Request.Paging.MinPagingSize = uint64(dctx.MinPagingSize) builder.Request.Paging.MaxPagingSize = uint64(dctx.MaxPagingSize) + builder.Request.Paging.PagingSizeBytes = uint64(dctx.PagingSizeBytes) } builder.RequestSource.RequestSourceInternal = dctx.InRestrictedSQL builder.RequestSource.RequestSourceType = dctx.RequestSourceType diff --git a/pkg/distsql/request_builder_test.go b/pkg/distsql/request_builder_test.go index 8d5d507c3432b..d863d044e9757 100644 --- a/pkg/distsql/request_builder_test.go +++ b/pkg/distsql/request_builder_test.go @@ -651,6 +651,19 @@ func TestRequestBuilder8(t *testing.T) { require.Equal(t, expect, actual) } +func TestRequestBuilderKeepsPagingSizeBytesWhenPagingDisabled(t *testing.T) { + dctx := NewDistSQLContextForTest() + dctx.EnablePaging = false + dctx.PagingSizeBytes = 4 * 1024 * 1024 + + actual, err := (&RequestBuilder{}). + SetFromSessionVars(dctx). + Build() + require.NoError(t, err) + require.False(t, actual.Paging.Enable) + require.Equal(t, uint64(4*1024*1024), actual.Paging.PagingSizeBytes) +} + func TestRequestBuilderTiKVClientReadTimeout(t *testing.T) { dctx := NewDistSQLContextForTest() dctx.TiKVClientReadTimeout = 100 diff --git a/pkg/executor/distsql.go b/pkg/executor/distsql.go index 17d793a948288..4c04323332a83 100644 --- a/pkg/executor/distsql.go +++ b/pkg/executor/distsql.go @@ -1079,6 +1079,7 @@ func (e *IndexLookUpExecutor) buildIndexSelectResultForRange( if e.indexLookUpPushDown { // Paging and Cop-cache is not supported in index lookup push down. builder.Request.Paging.Enable = false + builder.Request.Paging.PagingSizeBytes = 0 builder.Request.Cacheable = false } diff --git a/pkg/kv/kv.go b/pkg/kv/kv.go index 0a3c7a86bed4f..adf5643fb87fa 100644 --- a/pkg/kv/kv.go +++ b/pkg/kv/kv.go @@ -637,6 +637,9 @@ type Request struct { // when enabled, this field is adjusted to be max(MaxPagingSize, paging.MinAllowedMaxPagingSize), // see paging.GrowPagingSize MaxPagingSize uint64 + // PagingSizeBytes is the byte budget per page. + // 0 means disabled (no byte-budget paging). + PagingSizeBytes uint64 } // RequestSource indicates whether the request is an internal request. RequestSource util.RequestSource diff --git a/pkg/session/session.go b/pkg/session/session.go index 983cc6bc2bdbe..661045eaa1c9e 100644 --- a/pkg/session/session.go +++ b/pkg/session/session.go @@ -3334,6 +3334,14 @@ func (s *session) GetDistSQLCtx() *distsqlctx.DistSQLContext { ruConsumptionReporter = rgCtl } } + pagingSizeBytes := vars.PagingSizeBytes + if pagingSizeBytes > 0 { + if !vardef.EnableResourceControl.Load() || dom == nil || sc.ResourceGroupName == "" { + pagingSizeBytes = 0 + } else if rg, ok := dom.InfoSchema().ResourceGroupByName(ast.NewCIStr(sc.ResourceGroupName)); !ok || rg.GetBurstLimitAdjusted() < 0 { + pagingSizeBytes = 0 + } + } return &distsqlctx.DistSQLContext{ WarnHandler: sc.WarnHandler, InRestrictedSQL: sc.InRestrictedSQL, @@ -3373,6 +3381,7 @@ func (s *session) GetDistSQLCtx() *distsqlctx.DistSQLContext { EnablePaging: vars.EnablePaging, MinPagingSize: vars.MinPagingSize, MaxPagingSize: vars.MaxPagingSize, + PagingSizeBytes: pagingSizeBytes, RequestSourceType: vars.RequestSourceType, ExplicitRequestSourceType: vars.ExplicitRequestSourceType, StoreBatchSize: vars.StoreBatchSize, diff --git a/pkg/session/tidb_test.go b/pkg/session/tidb_test.go index b72430bfed24d..ea3412448d9f5 100644 --- a/pkg/session/tidb_test.go +++ b/pkg/session/tidb_test.go @@ -164,6 +164,38 @@ func TestCrossKSSessionDistSQLCtxDoesNotExposeTypedNilRUReporter(t *testing.T) { require.True(t, dctx.RUConsumptionReporter == nil) } +func TestDistSQLCtxPagingSizeBytesRequiresHardCappedResourceGroup(t *testing.T) { + store, dom := CreateStoreAndBootstrap(t) + defer func() { require.NoError(t, store.Close()) }() + defer dom.Close() + + oldRCEnabled := vardef.EnableResourceControl.Load() + vardef.EnableResourceControl.Store(true) + defer vardef.EnableResourceControl.Store(oldRCEnabled) + + se, err := createSession(store) + require.NoError(t, err) + MustExec(t, se, "create resource group rg_paging_capped ru_per_sec=1000") + MustExec(t, se, "create resource group rg_paging_unlimited ru_per_sec=1000 burstable=unlimited") + + const pagingSizeBytes = 4 * 1024 * 1024 + se.sessionVars.PagingSizeBytes = pagingSizeBytes + + check := func(resourceGroupName string, rcEnabled bool, expected int) { + vardef.EnableResourceControl.Store(rcEnabled) + se.sessionVars.StmtCtx.ResetForRetry() + se.sessionVars.StmtCtx.ResourceGroupName = resourceGroupName + require.Equal(t, expected, se.GetDistSQLCtx().PagingSizeBytes) + } + + check("default", true, 0) + MustExec(t, se, "alter resource group `default` ru_per_sec=1000 burstable=off") + check("default", true, pagingSizeBytes) + check("rg_paging_capped", true, pagingSizeBytes) + check("rg_paging_unlimited", true, 0) + check("rg_paging_capped", false, 0) +} + func TestRUV2MetricsIsolatedPerStatementInExplicitTxn(t *testing.T) { store, dom := CreateStoreAndBootstrap(t) defer func() { require.NoError(t, store.Close()) }() diff --git a/pkg/sessionctx/vardef/tidb_vars.go b/pkg/sessionctx/vardef/tidb_vars.go index e6cdef4ce0d02..ddab918749818 100644 --- a/pkg/sessionctx/vardef/tidb_vars.go +++ b/pkg/sessionctx/vardef/tidb_vars.go @@ -538,6 +538,10 @@ const ( // TiDBMaxPagingSize is used to control the max paging size in the coprocessor paging protocol. TiDBMaxPagingSize = "tidb_max_paging_size" + // TiDBPagingSizeBytes is the byte budget per coprocessor page. + // 0 means disabled (no byte-budget paging). + TiDBPagingSizeBytes = "tidb_paging_size_bytes" + // TiDBEnableCascadesPlanner is used to control whether to enable the cascades planner. TiDBEnableCascadesPlanner = "tidb_enable_cascades_planner" @@ -1484,6 +1488,7 @@ const ( DefInitChunkSize = 32 DefMinPagingSize = int(paging.MinPagingSize) DefMaxPagingSize = int(paging.MinAllowedMaxPagingSize) + DefPagingSizeBytes = 0 DefMaxChunkSize = 1024 DefDMLBatchSize = 0 DefMaxPreparedStmtCount = -1 diff --git a/pkg/sessionctx/variable/session.go b/pkg/sessionctx/variable/session.go index 8065833c9f71d..8d8abe2174c05 100644 --- a/pkg/sessionctx/variable/session.go +++ b/pkg/sessionctx/variable/session.go @@ -1733,6 +1733,10 @@ type SessionVars struct { // NOTE: please don't change it directly. Use `SetResourceGroupName`, because it'll need to inc/dec the metrics ResourceGroupName string + // PagingSizeBytes is the byte budget per page. + // 0 means disabled. + PagingSizeBytes int + // PessimisticTransactionFairLocking controls whether fair locking for pessimistic transaction // is enabled. PessimisticTransactionFairLocking bool @@ -2446,6 +2450,7 @@ func NewSessionVars(hctx HookContext) *SessionVars { EnableLateMaterialization: vardef.DefTiDBOptEnableLateMaterialization, TiFlashComputeDispatchPolicy: tiflashcompute.DispatchPolicyConsistentHash, ResourceGroupName: resourcegroup.DefaultResourceGroupName, + PagingSizeBytes: vardef.DefPagingSizeBytes, DefaultCollationForUTF8MB4: mysql.DefaultCollationName, GroupConcatMaxLen: vardef.DefGroupConcatMaxLen, EnableRedactLog: vardef.DefTiDBRedactLog, diff --git a/pkg/sessionctx/variable/setvar_affect.go b/pkg/sessionctx/variable/setvar_affect.go index 258de00d355dd..774effc0299a6 100644 --- a/pkg/sessionctx/variable/setvar_affect.go +++ b/pkg/sessionctx/variable/setvar_affect.go @@ -92,6 +92,7 @@ var isHintUpdatableVerified = map[string]struct{}{ "tidb_max_chunk_size": {}, "tidb_min_paging_size": {}, "tidb_max_paging_size": {}, + "tidb_paging_size_bytes": {}, "tidb_enable_cascades_planner": {}, "tidb_merge_join_concurrency": {}, "tidb_index_merge_intersection_concurrency": {}, diff --git a/pkg/sessionctx/variable/sysvar.go b/pkg/sessionctx/variable/sysvar.go index c36a18b042d51..03cfdde8563a6 100644 --- a/pkg/sessionctx/variable/sysvar.go +++ b/pkg/sessionctx/variable/sysvar.go @@ -3014,6 +3014,10 @@ var defaultSysVars = []*SysVar{ s.MaxPagingSize = tidbOptPositiveInt32(val, vardef.DefMaxPagingSize) return nil }}, + {Scope: vardef.ScopeGlobal | vardef.ScopeSession, Name: vardef.TiDBPagingSizeBytes, Value: strconv.Itoa(vardef.DefPagingSizeBytes), Type: vardef.TypeUnsigned, MinValue: 0, MaxValue: math.MaxInt64, SetSession: func(s *SessionVars, val string) error { + s.PagingSizeBytes = int(TidbOptInt64(val, int64(vardef.DefPagingSizeBytes))) + return nil + }}, {Scope: vardef.ScopeSession, Name: vardef.TiDBMemoryDebugModeMinHeapInUse, Value: strconv.Itoa(0), Type: vardef.TypeInt, MinValue: math.MinInt64, MaxValue: math.MaxInt64, SetSession: func(s *SessionVars, val string) error { s.MemoryDebugModeMinHeapInUse = TidbOptInt64(val, 0) return nil diff --git a/pkg/store/copr/BUILD.bazel b/pkg/store/copr/BUILD.bazel index effaa8a063a0e..8ae06ab8309d1 100644 --- a/pkg/store/copr/BUILD.bazel +++ b/pkg/store/copr/BUILD.bazel @@ -94,7 +94,7 @@ go_test( embed = [":copr"], flaky = True, race = "on", - shard_count = 43, + shard_count = 44, deps = [ "//pkg/kv", "//pkg/store/driver/backoff", diff --git a/pkg/store/copr/copr_test/coprocessor_test.go b/pkg/store/copr/copr_test/coprocessor_test.go index dba6e71aeb397..7be57a51429f1 100644 --- a/pkg/store/copr/copr_test/coprocessor_test.go +++ b/pkg/store/copr/copr_test/coprocessor_test.go @@ -242,9 +242,10 @@ func TestBuildCopIteratorWithBatchStoreCopr(t *testing.T) { Concurrency: 15, StoreBatchSize: 3, Paging: struct { - Enable bool - MinPagingSize uint64 - MaxPagingSize uint64 + Enable bool + MinPagingSize uint64 + MaxPagingSize uint64 + PagingSizeBytes uint64 }{ Enable: true, MinPagingSize: 1, @@ -256,6 +257,39 @@ func TestBuildCopIteratorWithBatchStoreCopr(t *testing.T) { tasks = it.GetTasks() require.Equal(t, len(tasks), 4) + // byte-budget paging disables store batch without changing row-count paging. + ranges = copr.BuildKeyRanges("a", "c", "d", "e", "h", "x", "y", "z") + req = &kv.Request{ + Tp: kv.ReqTypeDAG, + StoreType: kv.TiKV, + KeyRanges: kv.NewNonParitionedKeyRangesWithHint(ranges, []int{1, 1, 3, 3}), + Concurrency: 15, + StoreBatchSize: 3, + } + req.Paging.PagingSizeBytes = uint64(4 * 1024 * 1024) + it, errRes = copClient.BuildCopIterator(ctx, req, vars, opt) + require.Nil(t, errRes) + tasks = it.GetTasks() + require.Equal(t, len(tasks), 4) + require.False(t, req.Paging.Enable) + require.Zero(t, req.Paging.MinPagingSize) + require.Zero(t, req.Paging.MaxPagingSize) + // The byte budget is kept for TiKV DAG requests. + require.Equal(t, uint64(4*1024*1024), req.Paging.PagingSizeBytes) + + // byte-budget paging only applies to TiKV DAG requests; a non-DAG request + // drops the budget so req.Paging.PagingSizeBytes is the single source of truth. + req = &kv.Request{ + Tp: kv.ReqTypeAnalyze, + StoreType: kv.TiKV, + KeyRanges: kv.NewNonParitionedKeyRangesWithHint(copr.BuildKeyRanges("a", "c"), nil), + Concurrency: 15, + } + req.Paging.PagingSizeBytes = uint64(4 * 1024 * 1024) + _, errRes = copClient.BuildCopIterator(ctx, req, vars, opt) + require.Nil(t, errRes) + require.Zero(t, req.Paging.PagingSizeBytes) + // only small tasks will be batched. ranges = copr.BuildKeyRanges("a", "b", "h", "i", "o", "p") req = &kv.Request{ diff --git a/pkg/store/copr/coprocessor.go b/pkg/store/copr/coprocessor.go index c94a3cf94a76a..89c8fd3f56bb6 100644 --- a/pkg/store/copr/coprocessor.go +++ b/pkg/store/copr/coprocessor.go @@ -123,6 +123,7 @@ func (c *CopClient) BuildCopIterator(ctx context.Context, req *kv.Request, vars eventCb := option.EventCb failpoint.Inject("DisablePaging", func(_ failpoint.Value) { req.Paging.Enable = false + req.Paging.PagingSizeBytes = 0 }) if req.StoreType == kv.TiDB { // coprocessor on TiDB doesn't support paging @@ -132,8 +133,14 @@ func (c *CopClient) BuildCopIterator(ctx context.Context, req *kv.Request, vars // coprocessor request but type is not DAG req.Paging.Enable = false } + // Byte-budget paging only applies to TiKV DAG requests. Zero it out for any + // other case so that req.Paging.PagingSizeBytes stays the single source of + // truth downstream, mirroring how req.Paging.Enable is handled above. + if req.StoreType != kv.TiKV || req.Tp != kv.ReqTypeDAG { + req.Paging.PagingSizeBytes = 0 + } failpoint.Inject("checkKeyRangeSortedForPaging", func(_ failpoint.Value) { - if req.Paging.Enable { + if req.Paging.Enable || req.Paging.PagingSizeBytes > 0 { if !req.KeyRanges.IsFullySorted() { logutil.BgLogger().Fatal("distsql request key range not sorted!") } @@ -637,7 +644,7 @@ func buildCopTasks(bo *Backoffer, ranges *KeyRanges, opt *buildCopTaskOpt) ([]*c chanSize := 2 // in paging request, a request will be returned in multi batches, // enlarge the channel size to avoid the request blocked by buffer full. - if req.Paging.Enable { + if req.Paging.Enable || req.Paging.PagingSizeBytes > 0 { chanSize = 18 } @@ -1661,7 +1668,7 @@ func (worker *copIteratorWorker) handleTaskOnce(bo *Backoffer, task *copTask) (* } }) - if task.paging { + if task.paging || worker.req.Paging.PagingSizeBytes > 0 { task.pagingTaskIdx = atomic.AddUint32(worker.pagingTaskIdx, 1) } @@ -1672,6 +1679,7 @@ func (worker *copIteratorWorker) handleTaskOnce(bo *Backoffer, task *copTask) (* Ranges: task.ranges.ToPBRanges(), SchemaVer: worker.req.SchemaVar, PagingSize: task.pagingSize, + PagingSizeBytes: worker.req.Paging.PagingSizeBytes, Tasks: task.ToPBBatchTasks(), ConnectionId: worker.req.ConnID, ConnectionAlias: worker.req.ConnAlias, @@ -2474,7 +2482,7 @@ func (worker *copIteratorWorker) handleCopCache(task *copTask, resp *copResponse // Cache hit and is valid: use cached data as response data and we don't update the cache. data := slices.Clone(cacheValue.Data) resp.pbResp.Data = data - if worker.req.Paging.Enable { + if worker.req.Paging.Enable || worker.req.Paging.PagingSizeBytes > 0 { var start, end []byte if cacheValue.PageStart != nil { start = slices.Clone(cacheValue.PageStart) @@ -2938,8 +2946,8 @@ func checkStoreBatchCopr(req *kv.Request) bool { // Disable batch copr for follower read return false } - // Disable batch copr when paging is enabled. - if req.Paging.Enable { + // Disable batch copr for paged requests. + if req.Paging.Enable || req.Paging.PagingSizeBytes > 0 { return false } // Disable it for internal requests to avoid regression. diff --git a/pkg/store/copr/coprocessor_cache.go b/pkg/store/copr/coprocessor_cache.go index e41d4bd60c19b..01da36c3c3df4 100644 --- a/pkg/store/copr/coprocessor_cache.go +++ b/pkg/store/copr/coprocessor_cache.go @@ -113,7 +113,7 @@ func coprCacheBuildKey(copReq *coprocessor.Request) ([]byte, error) { } totalLength += 2 + len(r.Start) + 2 + len(r.End) } - if copReq.PagingSize > 0 { + if copReq.PagingSize > 0 || copReq.PagingSizeBytes > 0 { totalLength++ } @@ -149,8 +149,12 @@ func coprCacheBuildKey(copReq *coprocessor.Request) ([]byte, error) { dest += len(r.End) } - // 1 byte when use paging protocol - if copReq.PagingSize > 0 { + // 1 byte marks a paging request (row-count or byte-budget). The exact + // PagingSize/PagingSizeBytes values are deliberately excluded from the key: a + // cached page is self-describing via its returned range, so requests that + // differ only in page granularity can safely share a cache entry, while + // non-paging requests stay in a separate key space. + if copReq.PagingSize > 0 || copReq.PagingSizeBytes > 0 { key[dest] = 1 } diff --git a/pkg/store/copr/coprocessor_cache_test.go b/pkg/store/copr/coprocessor_cache_test.go index 811960a3d56de..24929e8588305 100644 --- a/pkg/store/copr/coprocessor_cache_test.go +++ b/pkg/store/copr/coprocessor_cache_test.go @@ -25,6 +25,11 @@ import ( ) func TestBuildCacheKey(t *testing.T) { + const ( + bytePagingSize = 0x0102030405060708 + rowPagingSize = 0x1112131415161718 + ) + req := coprocessor.Request{ Tp: 0xAB, StartTs: 0xAABBCC, @@ -57,6 +62,27 @@ func TestBuildCacheKey(t *testing.T) { expectKey += "\x01\x01\x03" // EndKey require.EqualValues(t, []byte(expectKey), key) + // A paging request (row-count or byte-budget) only appends a single marker + // byte; the exact sizes are not part of the key, so every paging variant + // shares the same key space and they never collide with the non-paging key + // asserted above. + req.PagingSizeBytes = bytePagingSize + key, err = coprCacheBuildKey(&req) + require.NoError(t, err) + require.EqualValues(t, []byte(expectKey+"\x01"), key) + + req.PagingSize = rowPagingSize + req.PagingSizeBytes = 0 + key, err = coprCacheBuildKey(&req) + require.NoError(t, err) + require.EqualValues(t, []byte(expectKey+"\x01"), key) + + req.PagingSize = rowPagingSize + req.PagingSizeBytes = bytePagingSize + key, err = coprCacheBuildKey(&req) + require.NoError(t, err) + require.EqualValues(t, []byte(expectKey+"\x01"), key) + req = coprocessor.Request{ Tp: 0xABCC, // Tp too big StartTs: 0xAABBCC, @@ -208,6 +234,23 @@ func TestGetSet(t *testing.T) { v = cache.Get([]byte("foo")) require.NotNil(t, v) require.EqualValues(t, []byte("bar"), v.Data) + + t.Run("paging size bytes cache hit keeps range", func(t *testing.T) { + req := &kv.Request{} + req.Paging.PagingSizeBytes = 1024 + worker := &copIteratorWorker{req: req} + task := &copTask{} + resp := &copResponse{pbResp: &coprocessor.Response{IsCacheHit: true}} + cacheValue := &coprCacheValue{ + Data: []byte("cached"), + PageStart: []byte("m"), + PageEnd: []byte("z"), + } + require.NoError(t, worker.handleCopCache(task, resp, nil, cacheValue)) + require.EqualValues(t, []byte("cached"), resp.pbResp.Data) + require.Equal(t, []byte("m"), resp.pbResp.GetRange().GetStart()) + require.Equal(t, []byte("z"), resp.pbResp.GetRange().GetEnd()) + }) } func TestIssue24118(t *testing.T) { diff --git a/pkg/store/copr/coprocessor_test.go b/pkg/store/copr/coprocessor_test.go index cd00ba099a14d..805289f07cdd7 100644 --- a/pkg/store/copr/coprocessor_test.go +++ b/pkg/store/copr/coprocessor_test.go @@ -702,6 +702,57 @@ func TestBuildPagingTasksDisablePagingForSmallLimit(t *testing.T) { require.Equal(t, tasks[0].pagingSize, uint64(0)) } +func TestBuildCopTasksWithPagingSizeBytes(t *testing.T) { + mockClient, cluster, pdClient, err := testutils.NewMockTiKV("", nil) + require.NoError(t, err) + defer func() { + pdClient.Close() + err = mockClient.Close() + require.NoError(t, err) + }() + _, regionIDs, _ := testutils.BootstrapWithMultiRegions(cluster, []byte("g"), []byte("n"), []byte("t")) + + pdCli := tikv.NewCodecPDClient(tikv.ModeTxn, pdClient) + defer pdCli.Close() + cache := NewRegionCache(tikv.NewRegionCache(pdCli)) + defer cache.Close() + bo := backoff.NewBackofferWithVars(context.Background(), 3000, nil) + + // A byte budget lives on the request and is the single source of truth. + req := &kv.Request{KeepOrder: true} + req.Paging.PagingSizeBytes = uint64(4 * 1024 * 1024) + tasks, err := buildCopTasks(bo, buildCopRanges("a", "c"), &buildCopTaskOpt{ + req: req, + cache: cache, + respChan: true, + }) + require.NoError(t, err) + require.Len(t, tasks, 1) + taskEqual(t, tasks[0], regionIDs[0], 0, "a", "c") + // A byte budget alone must not turn on row-count paging at the task level, + // but it still enlarges the response channel like row-count paging does. + require.False(t, tasks[0].paging) + require.Equal(t, uint64(0), tasks[0].pagingSize) + require.Equal(t, 18, cap(tasks[0].respChan)) + + // Row-count paging with a tiny limit downgrades independently; the byte + // budget on the request is untouched by that downgrade. + req.Paging.Enable = true + req.Paging.MinPagingSize = paging.MinPagingSize + req.LimitSize = 1 + tasks, err = buildCopTasks(bo, buildCopRanges("a", "c"), &buildCopTaskOpt{ + req: req, + cache: cache, + respChan: true, + }) + require.NoError(t, err) + require.Len(t, tasks, 1) + require.False(t, tasks[0].paging) + require.Equal(t, uint64(0), tasks[0].pagingSize) + require.Equal(t, uint64(4*1024*1024), req.Paging.PagingSizeBytes) + require.Equal(t, 18, cap(tasks[0].respChan)) +} + func toCopRange(r kv.KeyRange) *coprocessor.KeyRange { coprRange := coprocessor.KeyRange{} coprRange.Start = r.StartKey diff --git a/tests/integrationtest/r/sessionctx/setvar.result b/tests/integrationtest/r/sessionctx/setvar.result index c3040508c636a..9955f315be966 100644 --- a/tests/integrationtest/r/sessionctx/setvar.result +++ b/tests/integrationtest/r/sessionctx/setvar.result @@ -1097,6 +1097,28 @@ set @@tidb_max_paging_size=default; select @@tidb_max_paging_size; @@tidb_max_paging_size 50000 +select /*+ set_var(tidb_paging_size_bytes=1048576) */ @@tidb_paging_size_bytes; +@@tidb_paging_size_bytes +1048576 +select @@tidb_paging_size_bytes; +@@tidb_paging_size_bytes +0 +set @@tidb_paging_size_bytes=default; +select @@tidb_paging_size_bytes; +@@tidb_paging_size_bytes +0 +set @@tidb_paging_size_bytes=0; +select @@tidb_paging_size_bytes; +@@tidb_paging_size_bytes +0 +set @@tidb_paging_size_bytes=4194304; +select @@tidb_paging_size_bytes; +@@tidb_paging_size_bytes +4194304 +set @@tidb_paging_size_bytes=default; +select @@tidb_paging_size_bytes; +@@tidb_paging_size_bytes +0 select /*+ set_var(tidb_enable_cascades_planner=0) */ @@tidb_enable_cascades_planner; @@tidb_enable_cascades_planner 0 diff --git a/tests/integrationtest/t/sessionctx/setvar.test b/tests/integrationtest/t/sessionctx/setvar.test index 392a1a25e84e0..b27581da2b028 100644 --- a/tests/integrationtest/t/sessionctx/setvar.test +++ b/tests/integrationtest/t/sessionctx/setvar.test @@ -452,6 +452,16 @@ select /*+ set_var(tidb_max_paging_size=20) */ @@tidb_max_paging_size; select @@tidb_max_paging_size; set @@tidb_max_paging_size=default; select @@tidb_max_paging_size; +select /*+ set_var(tidb_paging_size_bytes=1048576) */ @@tidb_paging_size_bytes; +select @@tidb_paging_size_bytes; +set @@tidb_paging_size_bytes=default; +select @@tidb_paging_size_bytes; +set @@tidb_paging_size_bytes=0; +select @@tidb_paging_size_bytes; +set @@tidb_paging_size_bytes=4194304; +select @@tidb_paging_size_bytes; +set @@tidb_paging_size_bytes=default; +select @@tidb_paging_size_bytes; select /*+ set_var(tidb_enable_cascades_planner=0) */ @@tidb_enable_cascades_planner; select @@tidb_enable_cascades_planner; set @@tidb_enable_cascades_planner=default; From a30c616413ab88d12e701467325e6172274ca743 Mon Sep 17 00:00:00 2001 From: Hitachi Date: Thu, 9 Jul 2026 19:15:13 +0800 Subject: [PATCH 2/3] store/copr: add RC paging pre-charge EMA and PredictedReadBytes hint (#67941) close pingcap/tidb#68085 (cherry picked from commit ab7d93b603ba83d398d1b9b0063c78eceaacfc20) Signed-off-by: JmPotato --- DEPS.bzl | 8 +- go.mod | 4 +- go.sum | 8 +- .../grafana/tidb_resource_control.json | 320 ++++++++++++++++++ .../grafana/tidb_resource_control.jsonnet | 107 +++++- ...b_resource_control_with_keyspace_name.json | 320 ++++++++++++++++++ ...esource_control_with_keyspace_name.jsonnet | 107 +++++- pkg/store/copr/BUILD.bazel | 5 +- pkg/store/copr/coprocessor.go | 39 +++ pkg/store/copr/coprocessor_test.go | 11 + pkg/store/copr/ema.go | 64 ++++ pkg/store/copr/ema_test.go | 183 ++++++++++ 12 files changed, 1163 insertions(+), 13 deletions(-) create mode 100644 pkg/store/copr/ema.go create mode 100644 pkg/store/copr/ema_test.go diff --git a/DEPS.bzl b/DEPS.bzl index c042969418846..172122768fe02 100644 --- a/DEPS.bzl +++ b/DEPS.bzl @@ -4206,16 +4206,16 @@ def go_deps(): build_tags = ["nextgen", "intest"], build_file_proto_mode = "disable_global", importpath = "github.com/tikv/client-go/v2", - sum = "h1:fjOCdCmNgUrRzwOlemKHnCItUorHKbNvYRxdXlr7KCc=", - version = "v2.0.8-0.20260617064024-5526ae2a8546", + sum = "h1:RSzuwb/YdgiMPQO7dRZ8ilE71YL4n03nGU9qrCXDZEg=", + version = "v2.0.8-0.20260716132210-520a5635e5e3", ) go_repository( name = "com_github_tikv_pd_client", build_tags = ["nextgen", "intest"], build_file_proto_mode = "disable_global", importpath = "github.com/tikv/pd/client", - sum = "h1:8oHjHFQea2yL+MUK6Z6d1p0dYI2rHkoCQVjw1Yo6Uss=", - version = "v0.0.0-20260714045825-19ee324ee9d8", + sum = "h1:Bk296Skcn1GfRARDmYAuGONfnybon7ZntmCTrlCnzpQ=", + version = "v0.0.0-20260716095117-1636e69a936b", ) go_repository( name = "com_github_timakin_bodyclose", diff --git a/go.mod b/go.mod index 2a8d4bf5a7e9e..751393da983ca 100644 --- a/go.mod +++ b/go.mod @@ -123,8 +123,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.20260617064024-5526ae2a8546 - github.com/tikv/pd/client v0.0.0-20260714045825-19ee324ee9d8 + github.com/tikv/client-go/v2 v2.0.8-0.20260716132210-520a5635e5e3 + github.com/tikv/pd/client v0.0.0-20260716095117-1636e69a936b 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 diff --git a/go.sum b/go.sum index e2dc731b89029..01af500254cdf 100644 --- a/go.sum +++ b/go.sum @@ -893,10 +893,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.20260617064024-5526ae2a8546 h1:fjOCdCmNgUrRzwOlemKHnCItUorHKbNvYRxdXlr7KCc= -github.com/tikv/client-go/v2 v2.0.8-0.20260617064024-5526ae2a8546/go.mod h1:WDAIZI2EmfkDX4BP947c5wXJiIM3GgTnaaCp3TqReBE= -github.com/tikv/pd/client v0.0.0-20260714045825-19ee324ee9d8 h1:8oHjHFQea2yL+MUK6Z6d1p0dYI2rHkoCQVjw1Yo6Uss= -github.com/tikv/pd/client v0.0.0-20260714045825-19ee324ee9d8/go.mod h1:UL+x0aP+9+BDig+LYZFIU1G5r/gH2fEsVKl5MTS1Ud8= +github.com/tikv/client-go/v2 v2.0.8-0.20260716132210-520a5635e5e3 h1:RSzuwb/YdgiMPQO7dRZ8ilE71YL4n03nGU9qrCXDZEg= +github.com/tikv/client-go/v2 v2.0.8-0.20260716132210-520a5635e5e3/go.mod h1:0IJooYYDE5i4jKOug+hnphJrmrFORf3VgwfpDWsQoTQ= +github.com/tikv/pd/client v0.0.0-20260716095117-1636e69a936b h1:Bk296Skcn1GfRARDmYAuGONfnybon7ZntmCTrlCnzpQ= +github.com/tikv/pd/client v0.0.0-20260716095117-1636e69a936b/go.mod h1:5xk417Ar8t3SNWBhEwAJ/uycGmTs8ZUQN3GgdUJRFhE= 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= diff --git a/pkg/metrics/grafana/tidb_resource_control.json b/pkg/metrics/grafana/tidb_resource_control.json index b175c12f92913..14db9cba12c5a 100644 --- a/pkg/metrics/grafana/tidb_resource_control.json +++ b/pkg/metrics/grafana/tidb_resource_control.json @@ -2459,6 +2459,326 @@ "titleSize": "h6", "type": "row" }, + { + "collapse": true, + "collapsed": true, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 0 + }, + "id": 61, + "panels": [ + { + "aliasColors": { }, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "description": "Rate of read coprocessor RPCs split by whether they carried a positive PredictedReadBytes hint and triggered request-side read-byte pre-charge.", + "fill": 1, + "fillGradient": 0, + "gridPos": { + "h": 7, + "w": 24, + "x": 0, + "y": 0 + }, + "id": 62, + "legend": { + "alignAsTable": true, + "avg": false, + "current": true, + "max": false, + "min": false, + "rightSide": true, + "show": true, + "sideWidth": null, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "links": [ ], + "nullPointMode": "null", + "percentage": false, + "pointradius": 5, + "points": false, + "renderer": "flot", + "repeat": null, + "seriesOverrides": [ ], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "expr": "sum(rate(resource_manager_client_request_cop_read_precharge_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group)", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-precharged", + "refId": "A" + }, + { + "expr": "sum(rate(resource_manager_client_request_cop_read_no_precharge_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group)", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-no-precharge", + "refId": "B" + } + ], + "thresholds": [ ], + "timeFrom": null, + "timeShift": null, + "title": "Cop Read Pre-charge Coverage", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [ ] + }, + "yaxes": [ + { + "format": "short", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": true + }, + { + "format": "short", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": true + } + ] + }, + { + "aliasColors": { }, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "description": "Bytes rate over pre-charged coprocessor reads: predicted pre-charge basis and actual bytes read.", + "fill": 1, + "fillGradient": 0, + "gridPos": { + "h": 7, + "w": 12, + "x": 0, + "y": 7 + }, + "id": 63, + "legend": { + "alignAsTable": true, + "avg": false, + "current": true, + "max": false, + "min": false, + "rightSide": true, + "show": true, + "sideWidth": null, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "links": [ ], + "nullPointMode": "null", + "percentage": false, + "pointradius": 5, + "points": false, + "renderer": "flot", + "repeat": null, + "seriesOverrides": [ ], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "expr": "sum(rate(resource_manager_client_request_paging_precharge_bytes_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group)", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-precharge-predicted", + "refId": "A" + }, + { + "expr": "sum(rate(resource_manager_client_request_paging_actual_bytes_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group)", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-precharge-actual", + "refId": "B" + } + ], + "thresholds": [ ], + "timeFrom": null, + "timeShift": null, + "title": "Paging Bytes", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [ ] + }, + "yaxes": [ + { + "format": "Bps", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": true + }, + { + "format": "Bps", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": true + } + ] + }, + { + "aliasColors": { }, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "description": "Quantiles of signed prediction residual (actual - predicted read bytes) for pre-charged coprocessor RPCs. Positive = under-prediction (extra debit), negative = over-prediction (refund).", + "fill": 1, + "fillGradient": 0, + "gridPos": { + "h": 7, + "w": 12, + "x": 12, + "y": 7 + }, + "id": 64, + "legend": { + "alignAsTable": true, + "avg": true, + "current": true, + "max": true, + "min": true, + "rightSide": true, + "show": true, + "sideWidth": null, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "links": [ ], + "nullPointMode": "null", + "percentage": false, + "pointradius": 5, + "points": false, + "renderer": "flot", + "repeat": null, + "seriesOverrides": [ ], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "expr": "histogram_quantile(0.99, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group, le))", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-99", + "refId": "A" + }, + { + "expr": "histogram_quantile(0.9, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group, le))", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-90", + "refId": "B" + }, + { + "expr": "histogram_quantile(0.5, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group, le))", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-50", + "refId": "C" + }, + { + "expr": "histogram_quantile(0.1, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group, le))", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-10", + "refId": "D" + }, + { + "expr": "histogram_quantile(0.01, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group, le))", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-01", + "refId": "E" + } + ], + "thresholds": [ ], + "timeFrom": null, + "timeShift": null, + "title": "Paging Prediction Residual Bytes", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [ ] + }, + "yaxes": [ + { + "format": "bytes", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": true + }, + { + "format": "bytes", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": true + } + ] + } + ], + "repeat": null, + "repeatIteration": null, + "repeatRowId": null, + "showTitle": true, + "title": "Paging Pre-charge", + "titleSize": "h6", + "type": "row" + }, { "collapse": true, "collapsed": true, diff --git a/pkg/metrics/grafana/tidb_resource_control.jsonnet b/pkg/metrics/grafana/tidb_resource_control.jsonnet index ec1c6e6a53349..cc16077092f21 100644 --- a/pkg/metrics/grafana/tidb_resource_control.jsonnet +++ b/pkg/metrics/grafana/tidb_resource_control.jsonnet @@ -827,6 +827,95 @@ local TokenRequestCountPanel = graphPanel.new( ) ); +//* ==============Panel (Paging Pre-charge)================== +//* Row Title: Paging Pre-charge +//* Description: The metrics about resource control paging pre-charge +//* Panels: 3 +//* ==============Panel (Paging Pre-charge)================== + +local pagingPrechargeRow = row.new(collapse=true, title="Paging Pre-charge"); + +local PagingRequestCountPanel = graphPanel.new( + title="Cop Read Pre-charge Coverage", + datasource=myDS, + legend_rightSide=true, + legend_current=true, + legend_alignAsTable=true, + legend_values=true, + format="short", + description="Rate of read coprocessor RPCs split by whether they carried a positive PredictedReadBytes hint and triggered request-side read-byte pre-charge.", +).addTarget( + prometheus.target( + 'sum(rate(resource_manager_client_request_cop_read_precharge_total{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group)', + legendFormat="{{instance}}-{{resource_group}}-precharged", + ) +).addTarget( + prometheus.target( + 'sum(rate(resource_manager_client_request_cop_read_no_precharge_total{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group)', + legendFormat="{{instance}}-{{resource_group}}-no-precharge", + ) +); + +local PagingBytesPanel = graphPanel.new( + title="Paging Bytes", + datasource=myDS, + legend_rightSide=true, + legend_current=true, + legend_alignAsTable=true, + legend_values=true, + format="Bps", + description="Bytes rate over pre-charged coprocessor reads: predicted pre-charge basis and actual bytes read.", +).addTarget( + prometheus.target( + 'sum(rate(resource_manager_client_request_paging_precharge_bytes_total{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group)', + legendFormat="{{instance}}-{{resource_group}}-precharge-predicted", + ) +).addTarget( + prometheus.target( + 'sum(rate(resource_manager_client_request_paging_actual_bytes_total{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group)', + legendFormat="{{instance}}-{{resource_group}}-precharge-actual", + ) +); + +local PagingPredictionResidualBytesPanel = graphPanel.new( + title="Paging Prediction Residual Bytes", + datasource=myDS, + legend_rightSide=true, + legend_min=true, + legend_max=true, + legend_avg=true, + legend_current=true, + legend_alignAsTable=true, + legend_values=true, + format="bytes", + description="Quantiles of signed prediction residual (actual - predicted read bytes) for pre-charged coprocessor RPCs. Positive = under-prediction (extra debit), negative = over-prediction (refund).", +).addTarget( + prometheus.target( + 'histogram_quantile(0.99, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group, le))', + legendFormat="{{instance}}-{{resource_group}}-99", + ) +).addTarget( + prometheus.target( + 'histogram_quantile(0.9, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group, le))', + legendFormat="{{instance}}-{{resource_group}}-90", + ) +).addTarget( + prometheus.target( + 'histogram_quantile(0.5, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group, le))', + legendFormat="{{instance}}-{{resource_group}}-50", + ) +).addTarget( + prometheus.target( + 'histogram_quantile(0.1, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group, le))', + legendFormat="{{instance}}-{{resource_group}}-10", + ) +).addTarget( + prometheus.target( + 'histogram_quantile(0.01, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group, le))', + legendFormat="{{instance}}-{{resource_group}}-01", + ) +); + //* ==============Panel (Runaway)================== //* Row Title: Runaway //* Description: The metrics about runaway resource control @@ -1500,6 +1589,12 @@ local rowPos = { x: 0, y: 0, w: rowW, h: rowH }; local leftPanelPos = { x: 0, y: 0, w: panelW, h: panelH }; local rightPanelPos = { x: panelW, y: 0, w: panelW, h: panelH }; local fullPanelPos = { x: 0, y: 0, w: rowW, h: panelH }; +local pagingLeftPanelPos = { x: 0, y: panelH, w: panelW, h: panelH }; +local pagingRightPanelPos = { x: panelW, y: panelH, w: panelW, h: panelH }; +local pagingPanelRow = pagingPrechargeRow/* Paging Pre-charge */ + .addPanel(PagingRequestCountPanel, gridPos=fullPanelPos) + .addPanel(PagingBytesPanel, gridPos=pagingLeftPanelPos) + .addPanel(PagingPredictionResidualBytesPanel, gridPos=pagingRightPanelPos); TiDBResourceControlDash .addPanel( @@ -1539,7 +1634,17 @@ TiDBResourceControlDash .addPanel(TokenRequestCountPanel, gridPos=rightPanelPos) , gridPos=rowPos -).addPanel( +){ + // Keep the paging row IDs at 61-64 so generating the dashboard does not + // renumber the existing Runaway and later panels. + panels+::: [ + pagingPanelRow { + gridPos: rowPos, + id: 61, + panels: std.mapWithIndex(function(i, panel) panel { id: 62 + i }, pagingPanelRow.panels), + }, + ], +}.addPanel( runawayRow/* Runaway */ .addPanel(QueryMaxDurationPanel, gridPos=leftPanelPos) .addPanel(RunawayEventPanel, gridPos=rightPanelPos) diff --git a/pkg/metrics/nextgengrafana/tidb_resource_control_with_keyspace_name.json b/pkg/metrics/nextgengrafana/tidb_resource_control_with_keyspace_name.json index cf0f9b977e30d..696f964b80307 100644 --- a/pkg/metrics/nextgengrafana/tidb_resource_control_with_keyspace_name.json +++ b/pkg/metrics/nextgengrafana/tidb_resource_control_with_keyspace_name.json @@ -2252,6 +2252,326 @@ "titleSize": "h6", "type": "row" }, + { + "collapse": true, + "collapsed": true, + "gridPos": { + "h": 1, + "w": 24, + "x": 0, + "y": 0 + }, + "id": 50, + "panels": [ + { + "aliasColors": { }, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "description": "Rate of read coprocessor RPCs split by whether they carried a positive PredictedReadBytes hint and triggered request-side read-byte pre-charge.", + "fill": 1, + "fillGradient": 0, + "gridPos": { + "h": 7, + "w": 24, + "x": 0, + "y": 0 + }, + "id": 51, + "legend": { + "alignAsTable": true, + "avg": false, + "current": true, + "max": false, + "min": false, + "rightSide": true, + "show": true, + "sideWidth": null, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "links": [ ], + "nullPointMode": "null", + "percentage": false, + "pointradius": 5, + "points": false, + "renderer": "flot", + "repeat": null, + "seriesOverrides": [ ], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "expr": "sum(rate(resource_manager_client_request_cop_read_precharge_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group)", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-precharged", + "refId": "A" + }, + { + "expr": "sum(rate(resource_manager_client_request_cop_read_no_precharge_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group)", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-no-precharge", + "refId": "B" + } + ], + "thresholds": [ ], + "timeFrom": null, + "timeShift": null, + "title": "Cop Read Pre-charge Coverage", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [ ] + }, + "yaxes": [ + { + "format": "short", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": true + }, + { + "format": "short", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": true + } + ] + }, + { + "aliasColors": { }, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "description": "Bytes rate over pre-charged coprocessor reads: predicted pre-charge basis and actual bytes read.", + "fill": 1, + "fillGradient": 0, + "gridPos": { + "h": 7, + "w": 12, + "x": 0, + "y": 7 + }, + "id": 52, + "legend": { + "alignAsTable": true, + "avg": false, + "current": true, + "max": false, + "min": false, + "rightSide": true, + "show": true, + "sideWidth": null, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "links": [ ], + "nullPointMode": "null", + "percentage": false, + "pointradius": 5, + "points": false, + "renderer": "flot", + "repeat": null, + "seriesOverrides": [ ], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "expr": "sum(rate(resource_manager_client_request_paging_precharge_bytes_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group)", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-precharge-predicted", + "refId": "A" + }, + { + "expr": "sum(rate(resource_manager_client_request_paging_actual_bytes_total{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group)", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-precharge-actual", + "refId": "B" + } + ], + "thresholds": [ ], + "timeFrom": null, + "timeShift": null, + "title": "Paging Bytes", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [ ] + }, + "yaxes": [ + { + "format": "Bps", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": true + }, + { + "format": "Bps", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": true + } + ] + }, + { + "aliasColors": { }, + "bars": false, + "dashLength": 10, + "dashes": false, + "datasource": "${DS_TEST-CLUSTER}", + "description": "Quantiles of signed prediction residual (actual - predicted read bytes) for pre-charged coprocessor RPCs. Positive = under-prediction (extra debit), negative = over-prediction (refund).", + "fill": 1, + "fillGradient": 0, + "gridPos": { + "h": 7, + "w": 12, + "x": 12, + "y": 7 + }, + "id": 53, + "legend": { + "alignAsTable": true, + "avg": true, + "current": true, + "max": true, + "min": true, + "rightSide": true, + "show": true, + "sideWidth": null, + "total": false, + "values": true + }, + "lines": true, + "linewidth": 1, + "links": [ ], + "nullPointMode": "null", + "percentage": false, + "pointradius": 5, + "points": false, + "renderer": "flot", + "repeat": null, + "seriesOverrides": [ ], + "spaceLength": 10, + "stack": false, + "steppedLine": false, + "targets": [ + { + "expr": "histogram_quantile(0.99, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group, le))", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-99", + "refId": "A" + }, + { + "expr": "histogram_quantile(0.9, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group, le))", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-90", + "refId": "B" + }, + { + "expr": "histogram_quantile(0.5, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group, le))", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-50", + "refId": "C" + }, + { + "expr": "histogram_quantile(0.1, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group, le))", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-10", + "refId": "D" + }, + { + "expr": "histogram_quantile(0.01, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$tidb_instance\", resource_group=~\"$resource_group\"}[1m])) by (instance, resource_group, le))", + "format": "time_series", + "intervalFactor": 2, + "legendFormat": "{{instance}}-{{resource_group}}-01", + "refId": "E" + } + ], + "thresholds": [ ], + "timeFrom": null, + "timeShift": null, + "title": "Paging Prediction Residual Bytes", + "tooltip": { + "shared": true, + "sort": 0, + "value_type": "individual" + }, + "type": "graph", + "xaxis": { + "buckets": null, + "mode": "time", + "name": null, + "show": true, + "values": [ ] + }, + "yaxes": [ + { + "format": "bytes", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": true + }, + { + "format": "bytes", + "label": null, + "logBase": 1, + "max": null, + "min": null, + "show": true + } + ] + } + ], + "repeat": null, + "repeatIteration": null, + "repeatRowId": null, + "showTitle": true, + "title": "Paging Pre-charge", + "titleSize": "h6", + "type": "row" + }, { "collapse": true, "collapsed": true, diff --git a/pkg/metrics/nextgengrafana/tidb_resource_control_with_keyspace_name.jsonnet b/pkg/metrics/nextgengrafana/tidb_resource_control_with_keyspace_name.jsonnet index cae6c82d80feb..9896091882acd 100644 --- a/pkg/metrics/nextgengrafana/tidb_resource_control_with_keyspace_name.jsonnet +++ b/pkg/metrics/nextgengrafana/tidb_resource_control_with_keyspace_name.jsonnet @@ -766,6 +766,95 @@ local TokenRequestCountPanel = graphPanel.new( ) ); +//* ==============Panel (Paging Pre-charge)================== +//* Row Title: Paging Pre-charge +//* Description: The metrics about resource control paging pre-charge +//* Panels: 3 +//* ==============Panel (Paging Pre-charge)================== + +local pagingPrechargeRow = row.new(collapse=true, title="Paging Pre-charge"); + +local PagingRequestCountPanel = graphPanel.new( + title="Cop Read Pre-charge Coverage", + datasource=myDS, + legend_rightSide=true, + legend_current=true, + legend_alignAsTable=true, + legend_values=true, + format="short", + description="Rate of read coprocessor RPCs split by whether they carried a positive PredictedReadBytes hint and triggered request-side read-byte pre-charge.", +).addTarget( + prometheus.target( + 'sum(rate(resource_manager_client_request_cop_read_precharge_total{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group)', + legendFormat="{{instance}}-{{resource_group}}-precharged", + ) +).addTarget( + prometheus.target( + 'sum(rate(resource_manager_client_request_cop_read_no_precharge_total{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group)', + legendFormat="{{instance}}-{{resource_group}}-no-precharge", + ) +); + +local PagingBytesPanel = graphPanel.new( + title="Paging Bytes", + datasource=myDS, + legend_rightSide=true, + legend_current=true, + legend_alignAsTable=true, + legend_values=true, + format="Bps", + description="Bytes rate over pre-charged coprocessor reads: predicted pre-charge basis and actual bytes read.", +).addTarget( + prometheus.target( + 'sum(rate(resource_manager_client_request_paging_precharge_bytes_total{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group)', + legendFormat="{{instance}}-{{resource_group}}-precharge-predicted", + ) +).addTarget( + prometheus.target( + 'sum(rate(resource_manager_client_request_paging_actual_bytes_total{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group)', + legendFormat="{{instance}}-{{resource_group}}-precharge-actual", + ) +); + +local PagingPredictionResidualBytesPanel = graphPanel.new( + title="Paging Prediction Residual Bytes", + datasource=myDS, + legend_rightSide=true, + legend_min=true, + legend_max=true, + legend_avg=true, + legend_current=true, + legend_alignAsTable=true, + legend_values=true, + format="bytes", + description="Quantiles of signed prediction residual (actual - predicted read bytes) for pre-charged coprocessor RPCs. Positive = under-prediction (extra debit), negative = over-prediction (refund).", +).addTarget( + prometheus.target( + 'histogram_quantile(0.99, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group, le))', + legendFormat="{{instance}}-{{resource_group}}-99", + ) +).addTarget( + prometheus.target( + 'histogram_quantile(0.9, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group, le))', + legendFormat="{{instance}}-{{resource_group}}-90", + ) +).addTarget( + prometheus.target( + 'histogram_quantile(0.5, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group, le))', + legendFormat="{{instance}}-{{resource_group}}-50", + ) +).addTarget( + prometheus.target( + 'histogram_quantile(0.1, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group, le))', + legendFormat="{{instance}}-{{resource_group}}-10", + ) +).addTarget( + prometheus.target( + 'histogram_quantile(0.01, sum(rate(resource_manager_client_request_paging_prediction_residual_bytes_bucket{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$tidb_instance", resource_group=~"$resource_group"}[1m])) by (instance, resource_group, le))', + legendFormat="{{instance}}-{{resource_group}}-01", + ) +); + //* ==============Panel (Runaway)================== //* Row Title: Runaway //* Description: The metrics about runaway resource control @@ -1210,6 +1299,12 @@ local rowPos = { x: 0, y: 0, w: rowW, h: rowH }; local leftPanelPos = { x: 0, y: 0, w: panelW, h: panelH }; local rightPanelPos = { x: panelW, y: 0, w: panelW, h: panelH }; local fullPanelPos = { x: 0, y: 0, w: rowW, h: panelH }; +local pagingLeftPanelPos = { x: 0, y: panelH, w: panelW, h: panelH }; +local pagingRightPanelPos = { x: panelW, y: panelH, w: panelW, h: panelH }; +local pagingPanelRow = pagingPrechargeRow/* Paging Pre-charge */ + .addPanel(PagingRequestCountPanel, gridPos=fullPanelPos) + .addPanel(PagingBytesPanel, gridPos=pagingLeftPanelPos) + .addPanel(PagingPredictionResidualBytesPanel, gridPos=pagingRightPanelPos); TiDBResourceControlDash .addPanel( @@ -1247,7 +1342,17 @@ TiDBResourceControlDash .addPanel(TokenRequestCountPanel, gridPos=rightPanelPos) , gridPos=rowPos -).addPanel( +){ + // Keep the paging row IDs at 50-53 so generating the dashboard does not + // renumber the existing Runaway and later panels. + panels+::: [ + pagingPanelRow { + gridPos: rowPos, + id: 50, + panels: std.mapWithIndex(function(i, panel) panel { id: 51 + i }, pagingPanelRow.panels), + }, + ], +}.addPanel( runawayRow/* Runaway */ .addPanel(QueryMaxDurationPanel, gridPos=leftPanelPos) .addPanel(RunawayEventPanel, gridPos=rightPanelPos) diff --git a/pkg/store/copr/BUILD.bazel b/pkg/store/copr/BUILD.bazel index 8ae06ab8309d1..27ff496c3ca3a 100644 --- a/pkg/store/copr/BUILD.bazel +++ b/pkg/store/copr/BUILD.bazel @@ -7,6 +7,7 @@ go_library( "batch_request_sender.go", "coprocessor.go", "coprocessor_cache.go", + "ema.go", "key_ranges.go", "mpp.go", "mpp_probe.go", @@ -86,6 +87,7 @@ go_test( "batch_coprocessor_test.go", "coprocessor_cache_test.go", "coprocessor_test.go", + "ema_test.go", "key_ranges_test.go", "main_test.go", "mpp_probe_test.go", @@ -94,7 +96,7 @@ go_test( embed = [":copr"], flaky = True, race = "on", - shard_count = 44, + shard_count = 50, deps = [ "//pkg/kv", "//pkg/store/driver/backoff", @@ -107,6 +109,7 @@ go_test( "@com_github_pingcap_failpoint//:failpoint", "@com_github_pingcap_kvproto//pkg/coprocessor", "@com_github_pingcap_kvproto//pkg/errorpb", + "@com_github_pingcap_kvproto//pkg/kvrpcpb", "@com_github_pingcap_kvproto//pkg/metapb", "@com_github_pingcap_kvproto//pkg/mpp", "@com_github_stathat_consistent//:consistent", diff --git a/pkg/store/copr/coprocessor.go b/pkg/store/copr/coprocessor.go index 89c8fd3f56bb6..421a6dd616b3c 100644 --- a/pkg/store/copr/coprocessor.go +++ b/pkg/store/copr/coprocessor.go @@ -60,6 +60,7 @@ import ( "github.com/pingcap/tidb/pkg/util/tracing" "github.com/pingcap/tidb/pkg/util/trxevents" "github.com/pingcap/tipb/go-tipb" + clientgoconfig "github.com/tikv/client-go/v2/config" "github.com/tikv/client-go/v2/metrics" "github.com/tikv/client-go/v2/tikv" "github.com/tikv/client-go/v2/tikvrpc" @@ -217,6 +218,7 @@ func (c *CopClient) BuildCopIterator(ctx context.Context, req *kv.Request, vars rpcCancel: tikv.NewRPCanceller(), buildTaskElapsed: *buildOpt.elapsed, runawayChecker: req.RunawayChecker, + ema: newRUEMA(req.Paging.PagingSizeBytes), } // Pipelined-dml can flush locks when it is still reading. // The coprocessor of the txn should not be blocked by itself. @@ -1008,6 +1010,9 @@ type copIterator struct { runawayChecker resourcegroup.RunawayChecker stats *copIteratorRuntimeStats + + // One EMA per copIterator (logical scan), shared across workers. + ema *ruEMA } type liteCopIteratorWorker struct { @@ -1041,6 +1046,8 @@ type copIteratorWorker struct { storeBatchedNum *atomic.Uint64 storeBatchedFallbackNum *atomic.Uint64 stats *copIteratorRuntimeStats + + ema *ruEMA } // copIteratorTaskSender sends tasks to taskCh then wait for the workers to exit. @@ -1236,6 +1243,7 @@ func newCopIteratorWorker(it *copIterator, taskCh <-chan *copTask) *copIteratorW storeBatchedNum: &it.storeBatchedNum, storeBatchedFallbackNum: &it.storeBatchedFallbackNum, stats: it.stats, + ema: it.ema, } } @@ -1659,6 +1667,15 @@ func (worker *copIteratorWorker) handleTask(ctx context.Context, task *copTask, } } +func (worker *copIteratorWorker) predictedReadBytesForTask(task *copTask) uint64 { + // Byte-budget paging is independent from row-count paging, so a request-level + // byte budget must still carry the pre-charge hint when task.paging is false. + if !task.paging && worker.req.Paging.PagingSizeBytes == 0 { + return 0 + } + return worker.ema.Predict() +} + // handleTaskOnce handles single copTask, successful results are send to channel. // If error happened, returns error. If region split or meet lock, returns the remain tasks. func (worker *copIteratorWorker) handleTaskOnce(bo *Backoffer, task *copTask) (*copTaskResult, error) { @@ -1709,6 +1726,7 @@ func (worker *copIteratorWorker) handleTaskOnce(bo *Backoffer, task *copTask) (* BucketsVersion: task.bucketsVer, }) req.InputRequestSource = task.requestSource.GetRequestSource() + req.PredictedReadBytes = worker.predictedReadBytesForTask(task) if task.firstReadType != "" { req.ReadType = task.firstReadType req.IsRetryRequest = true @@ -1875,6 +1893,22 @@ func appendScanDetail(logStr string, columnFamily string, scanInfo *kvrpcpb.Scan return logStr } +// pagingResponseReadBytes returns the MVCC bytes basis the EMA observes. +// Mirrors client-go's resourcecontrol.MakeResponseInfo so the EMA learns +// the same quantity PD bills against. +func pagingResponseReadBytes(pbResp *coprocessor.Response) uint64 { + if pbResp == nil { + return 0 + } + if scanDetail := pbResp.GetExecDetailsV2().GetScanDetailV2(); scanDetail != nil { + if clientgoconfig.NextGen { + return max(scanDetail.GetTotalVersionsSize(), scanDetail.GetProcessedVersionsSize()) + } + return scanDetail.GetProcessedVersionsSize() + } + return 0 +} + func (worker *copIteratorWorker) handleCopPagingResult(bo *Backoffer, rpcCtx *tikv.RPCContext, resp *copResponse, cacheKey []byte, cacheValue *coprCacheValue, task *copTask, costTime time.Duration) (*copTaskResult, error) { result, err := worker.handleCopResponse(bo, rpcCtx, resp, cacheKey, cacheValue, task, costTime) if err != nil { @@ -1890,11 +1924,16 @@ func (worker *copIteratorWorker) handleCopPagingResult(bo *Backoffer, rpcCtx *ti pagingRange := resp.pbResp.Range // only paging requests need to calculate the next ranges if pagingRange == nil { + // The final page has no remaining range; skipping it keeps the EMA slightly conservative. // If the storage engine doesn't support paging protocol, it should have return all the region data. // So we finish here. return result, nil } + if readBytes := pagingResponseReadBytes(resp.pbResp); readBytes > 0 { + worker.ema.Observe(readBytes, time.Now()) + } + // calculate next ranges and grow the paging size task.ranges = worker.calculateRemain(task.ranges, pagingRange, worker.req.Desc) if task.ranges.Len() == 0 { diff --git a/pkg/store/copr/coprocessor_test.go b/pkg/store/copr/coprocessor_test.go index 805289f07cdd7..52dae0474e31c 100644 --- a/pkg/store/copr/coprocessor_test.go +++ b/pkg/store/copr/coprocessor_test.go @@ -700,6 +700,11 @@ func TestBuildPagingTasksDisablePagingForSmallLimit(t *testing.T) { taskEqual(t, tasks[0], regionIDs[0], 0, "a", "c") require.False(t, tasks[0].paging) require.Equal(t, tasks[0].pagingSize, uint64(0)) + + ema := newRUEMA(0) + ema.Observe(1_048_576, time.Now()) + worker := &copIteratorWorker{req: req, ema: ema} + require.Zero(t, worker.predictedReadBytesForTask(tasks[0])) } func TestBuildCopTasksWithPagingSizeBytes(t *testing.T) { @@ -735,6 +740,10 @@ func TestBuildCopTasksWithPagingSizeBytes(t *testing.T) { require.Equal(t, uint64(0), tasks[0].pagingSize) require.Equal(t, 18, cap(tasks[0].respChan)) + ema := newRUEMA(req.Paging.PagingSizeBytes) + worker := &copIteratorWorker{req: req, ema: ema} + require.Equal(t, uint64(4*1024*1024), worker.predictedReadBytesForTask(tasks[0])) + // Row-count paging with a tiny limit downgrades independently; the byte // budget on the request is untouched by that downgrade. req.Paging.Enable = true @@ -751,6 +760,8 @@ func TestBuildCopTasksWithPagingSizeBytes(t *testing.T) { require.Equal(t, uint64(0), tasks[0].pagingSize) require.Equal(t, uint64(4*1024*1024), req.Paging.PagingSizeBytes) require.Equal(t, 18, cap(tasks[0].respChan)) + worker = &copIteratorWorker{req: req, ema: newRUEMA(req.Paging.PagingSizeBytes)} + require.Equal(t, uint64(4*1024*1024), worker.predictedReadBytesForTask(tasks[0])) } func toCopRange(r kv.KeyRange) *coprocessor.KeyRange { diff --git a/pkg/store/copr/ema.go b/pkg/store/copr/ema.go new file mode 100644 index 0000000000000..dbeaf3ea16d9e --- /dev/null +++ b/pkg/store/copr/ema.go @@ -0,0 +1,64 @@ +// 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 copr + +import ( + "math" + "sync" + "time" +) + +// defaultRUEMATau is the decay time constant τ: sample weight drops to +// 1/e after τ. +const defaultRUEMATau = time.Second + +// ruEMA is a time-aware exponential moving average (EMA) weighting older +// samples by exp(-Δt/τ) so long gaps decay stale samples. Safe for concurrent +// Observe/Predict. +type ruEMA struct { + mu sync.Mutex + tau time.Duration + value float64 + lastObsAt time.Time +} + +func newRUEMA(seedReadBytes uint64) *ruEMA { + // lastObsAt is intentionally left as the zero time. If there is no seed, + // the first Observe gets alpha ~= 1 and seeds value from the first sample; + // if there is a byte-budget seed, the first real sample replaces it. + return &ruEMA{tau: defaultRUEMATau, value: float64(seedReadBytes)} +} + +func (e *ruEMA) Observe(bytes uint64, now time.Time) { + e.mu.Lock() + defer e.mu.Unlock() + dt := now.Sub(e.lastObsAt) + if dt < 0 { + dt = 0 + } + alpha := 1 - math.Exp(-float64(dt)/float64(e.tau)) + e.value += alpha * (float64(bytes) - e.value) + // Don't rewind on out-of-order Observes. + if now.After(e.lastObsAt) { + e.lastObsAt = now + } +} + +// Predict returns the current estimate. +func (e *ruEMA) Predict() uint64 { + e.mu.Lock() + defer e.mu.Unlock() + return uint64(e.value) +} diff --git a/pkg/store/copr/ema_test.go b/pkg/store/copr/ema_test.go new file mode 100644 index 0000000000000..f4619594355ec --- /dev/null +++ b/pkg/store/copr/ema_test.go @@ -0,0 +1,183 @@ +// 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 copr + +import ( + "sync" + "testing" + "time" + + "github.com/pingcap/kvproto/pkg/coprocessor" + "github.com/pingcap/kvproto/pkg/kvrpcpb" + "github.com/stretchr/testify/require" + clientgoconfig "github.com/tikv/client-go/v2/config" +) + +func TestRUEMASeedAndConverge(t *testing.T) { + t.Run("seeded prediction", func(t *testing.T) { + const pageSizeBytes uint64 = 4 * 1024 * 1024 + e := newRUEMA(pageSizeBytes) + require.Equal(t, pageSizeBytes, e.Predict(), "fresh seeded EMA predicts the requested page size") + + now := time.Now() + e.Observe(1_000_000, now) + require.Equal(t, uint64(1_000_000), e.Predict(), + "first real sample replaces the seed via a ~infinite dt (alpha≈1)") + }) + + t.Run("unseeded first observation", func(t *testing.T) { + e := newRUEMA(0) + require.Zero(t, e.Predict(), "fresh unseeded EMA: no prediction") + + now := time.Now() + e.Observe(1_000_000, now) + require.Equal(t, uint64(1_000_000), e.Predict(), + "first sample seeds the EMA via a ~infinite dt (alpha≈1)") + + e.Observe(1_000_000, now.Add(100*time.Millisecond)) + require.InDelta(t, float64(1_000_000), float64(e.Predict()), 1, + "steady input: prediction stays at the input") + }) +} + +func TestRUEMATracksShift(t *testing.T) { + e := newRUEMA(0) + now := time.Now() + for i := 0; i < 5; i++ { + e.Observe(100_000, now.Add(time.Duration(i)*100*time.Millisecond)) + } + require.InDelta(t, float64(100_000), float64(e.Predict()), 1, + "EMA should have converged to 100K after 5 identical samples") + + // Shift workload to 500K per RPC. After enough new samples with tau=1s + // and 100 ms gaps, the EMA should follow the new regime. + for i := 5; i < 20; i++ { + e.Observe(500_000, now.Add(time.Duration(i)*100*time.Millisecond)) + } + require.Greater(t, e.Predict(), uint64(400_000), + "EMA should have shifted well above the old steady value") + require.LessOrEqual(t, e.Predict(), uint64(500_000), + "EMA should not overshoot the new steady value") +} + +func TestRUEMALargeGapCollapsesWeight(t *testing.T) { + e := newRUEMA(0) + now := time.Now() + e.Observe(100_000, now) + // A gap much larger than tau (default 1s) means alpha ≈ 1, so the new + // sample should dominate the EMA almost entirely. + e.Observe(1_000_000, now.Add(10*time.Second)) + require.InDelta(t, float64(1_000_000), float64(e.Predict()), 1_000, + "a gap >> tau should collapse the older sample's weight") +} + +func TestPagingResponseReadBytes(t *testing.T) { + // A nil response or one without ScanDetailV2 should produce 0 so the + // caller can skip the observation cleanly. + require.Zero(t, pagingResponseReadBytes(nil)) + require.Zero(t, pagingResponseReadBytes(&coprocessor.Response{})) + + tests := []struct { + name string + scanDetail *kvrpcpb.ScanDetailV2 + classicWant uint64 + nextGenWant uint64 + }{ + { + name: "classic uses processed versions size", + scanDetail: &kvrpcpb.ScanDetailV2{ + ProcessedVersionsSize: 1_048_576, + TotalVersionsSize: 2_097_152, + }, + classicWant: 1_048_576, + nextGenWant: 2_097_152, + }, + { + name: "nextgen keeps compatibility when total is smaller", + scanDetail: &kvrpcpb.ScanDetailV2{ + ProcessedVersionsSize: 1_048_576, + TotalVersionsSize: 512 * 1024, + }, + classicWant: 1_048_576, + nextGenWant: 1_048_576, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + resp := &coprocessor.Response{ + ExecDetailsV2: &kvrpcpb.ExecDetailsV2{ + ScanDetailV2: tt.scanDetail, + }, + } + want := tt.classicWant + if clientgoconfig.NextGen { + want = tt.nextGenWant + } + require.Equal(t, want, pagingResponseReadBytes(resp)) + }) + } +} + +func TestRUEMAConcurrentObserveAndPredict(t *testing.T) { + // One EMA per copIterator is shared by multiple workers, so Observe and + // Predict must be race-free. Run with `go test -race` to exercise the + // mutex; this test guarantees no panic/deadlock and that readiness is + // eventually observed from a reader goroutine. + e := newRUEMA(0) + const writers = 8 + const iters = 200 + done := make(chan struct{}) + go func() { + // Reader: hammer Predict while writers fan in samples. + for { + select { + case <-done: + return + default: + _ = e.Predict() + } + } + }() + base := time.Now() + var wg sync.WaitGroup + for w := 0; w < writers; w++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + for i := 0; i < iters; i++ { + e.Observe(uint64(100_000+id*1_000+i), + base.Add(time.Duration(i)*time.Millisecond)) + } + }(w) + } + wg.Wait() + close(done) + require.Greater(t, e.Predict(), uint64(0), "prediction must be positive after observations") +} + +func TestRUEMANonMonotonicTime(t *testing.T) { + // If for any reason a later observation carries an earlier timestamp + // (clock skew, test fixture), Observe must not blow up with a negative + // Δt. The behavior we want: treat the gap as zero and use the new + // sample only minimally. + e := newRUEMA(0) + now := time.Now() + e.Observe(100_000, now) + e.Observe(500_000, now.Add(-1*time.Second)) + // With Δt clamped to 0, alpha = 0, so the EMA should stay at the first + // sample's value. + require.InDelta(t, float64(100_000), float64(e.Predict()), 1, + "negative Δt should be clamped so the new sample has ~zero weight") +} From 0d734469b54d735006728f71d85365cd84f734ac Mon Sep 17 00:00:00 2001 From: JmPotato Date: Wed, 15 Jul 2026 14:03:22 +0800 Subject: [PATCH 3/3] session: use resource group runtime state for paging (#69633) ref pingcap/tidb#68085 (cherry picked from commit 10292a4f8697f6fe7ae3965f3ebd6656c1ab37ed) Signed-off-by: JmPotato --- pkg/session/session.go | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/pkg/session/session.go b/pkg/session/session.go index 661045eaa1c9e..d15ccaa925beb 100644 --- a/pkg/session/session.go +++ b/pkg/session/session.go @@ -3335,12 +3335,8 @@ func (s *session) GetDistSQLCtx() *distsqlctx.DistSQLContext { } } pagingSizeBytes := vars.PagingSizeBytes - if pagingSizeBytes > 0 { - if !vardef.EnableResourceControl.Load() || dom == nil || sc.ResourceGroupName == "" { - pagingSizeBytes = 0 - } else if rg, ok := dom.InfoSchema().ResourceGroupByName(ast.NewCIStr(sc.ResourceGroupName)); !ok || rg.GetBurstLimitAdjusted() < 0 { - pagingSizeBytes = 0 - } + if pagingSizeBytes > 0 && (!vardef.EnableResourceControl.Load() || !resourceGroupAllowsPagingSizeBytes(dom, sc.ResourceGroupName)) { + pagingSizeBytes = 0 } return &distsqlctx.DistSQLContext{ WarnHandler: sc.WarnHandler, @@ -3414,6 +3410,19 @@ func (s *session) GetDistSQLCtx() *distsqlctx.DistSQLContext { return dctx } +func resourceGroupAllowsPagingSizeBytes(dom *domain.Domain, resourceGroupName string) bool { + if dom == nil || resourceGroupName == "" { + return false + } + if rgCtl := dom.ResourceGroupsController(); rgCtl != nil { + if state, ok := rgCtl.GetResourceGroupRuntimeState(resourceGroupName); ok { + return state.HasLimitedBurst + } + } + rg, ok := dom.InfoSchema().ResourceGroupByName(ast.NewCIStr(resourceGroupName)) + return ok && rg.GetBurstLimitAdjusted() >= 0 +} + // GetRangerCtx returns the context used in `ranger` related functions func (s *session) GetRangerCtx() *rangerctx.RangerContext { vars := s.GetSessionVars()