executor: add explicit request source for point get request (#45526)

ref #44769

ddl/tests/resourcegroup/resource_group_test.go

@@ -63,7 +63,9 @@ func TestResourceGroupBasic(t *testing.T) {
 
 	// test default resource group.
 	tk.MustQuery("select * from information_schema.resource_groups where name = 'default'").Check(testkit.Rows("default UNLIMITED MEDIUM YES <nil> <nil>"))
-	tk.MustExec("alter resource group `default` RU_PER_SEC=1000 PRIORITY=LOW")
+	tk.MustExec("alter resource group `default` PRIORITY=LOW")
+	tk.MustQuery("select * from information_schema.resource_groups where name = 'default'").Check(testkit.Rows("default UNLIMITED LOW YES <nil> <nil>"))
+	tk.MustExec("alter resource group `default` ru_per_sec=1000")
 	tk.MustQuery("select * from information_schema.resource_groups where name = 'default'").Check(testkit.Rows("default 1000 LOW YES <nil> <nil>"))
 	tk.MustContainErrMsg("drop resource group `default`", "can't drop reserved resource group")
 

executor/builder.go

@@ -1919,6 +1919,8 @@ func (b *executorBuilder) getSnapshot() (kv.Snapshot, error) {
 	replicaReadType := sessVars.GetReplicaRead()
 	snapshot.SetOption(kv.ReadReplicaScope, b.readReplicaScope)
 	snapshot.SetOption(kv.TaskID, sessVars.StmtCtx.TaskID)
+	snapshot.SetOption(kv.ResourceGroupName, sessVars.ResourceGroupName)
+	snapshot.SetOption(kv.ExplicitRequestSourceType, sessVars.ExplicitRequestSourceType)
 
 	if replicaReadType.IsClosestRead() && b.readReplicaScope != kv.GlobalTxnScope {
 		snapshot.SetOption(kv.MatchStoreLabels, []*metapb.StoreLabel{
@@ -5210,7 +5212,6 @@ func (b *executorBuilder) buildBatchPointGet(plan *plannercore.BatchPointGetPlan
 	if e.Ctx().GetSessionVars().IsReplicaReadClosestAdaptive() {
 		e.snapshot.SetOption(kv.ReplicaReadAdjuster, newReplicaReadAdjuster(e.Ctx(), plan.GetAvgRowSize()))
 	}
-	e.snapshot.SetOption(kv.ResourceGroupName, b.ctx.GetSessionVars().ResourceGroupName)
 	if e.RuntimeStats() != nil {
 		snapshotStats := &txnsnapshot.SnapshotRuntimeStats{}
 		e.stats = &runtimeStatsWithSnapshot{

executor/point_get.go

@@ -75,7 +75,6 @@ func (b *executorBuilder) buildPointGet(p *plannercore.PointGetPlan) exec.Execut
 	if b.ctx.GetSessionVars().IsReplicaReadClosestAdaptive() {
 		e.snapshot.SetOption(kv.ReplicaReadAdjuster, newReplicaReadAdjuster(e.Ctx(), p.GetAvgRowSize()))
 	}
-	e.snapshot.SetOption(kv.ResourceGroupName, b.ctx.GetSessionVars().ResourceGroupName)
 	if e.RuntimeStats() != nil {
 		snapshotStats := &txnsnapshot.SnapshotRuntimeStats{}
 		e.stats = &runtimeStatsWithSnapshot{

meta/meta.go

@@ -19,6 +19,7 @@ import (
 	"encoding/binary"
 	"encoding/json"
 	"fmt"
+	"math"
 	"strconv"
 	"strings"
 	"sync"
@@ -84,7 +85,7 @@ var (
 	// the default meta of the `default` group
 	defaultRGroupMeta = &model.ResourceGroupInfo{
 		ResourceGroupSettings: &model.ResourceGroupSettings{
-			RURate:     1000000,
+			RURate:     math.MaxInt32,
 			BurstLimit: -1,
 			Priority:   model.MediumPriorityValue,
 		},

metrics/grafana/tidb.json

@@ -19827,7 +19827,7 @@
           "dashLength": 10,
           "dashes": false,
           "datasource": "${DS_TEST-CLUSTER}",
-          "description": "The hit/miss count of kv request that read from closest tikv. null means the request don't not support follower read",
+          "description": "The hit/miss count of kv requests that read from closest tikv. null means the request don't not support follower read",
           "fieldConfig": {
             "defaults": {},
             "overrides": []
@@ -19982,7 +19982,7 @@
           "timeFrom": null,
           "timeRegions": [],
           "timeShift": null,
-          "title": "Coprocssor Response Size Per Instance and TiKV",
+          "title": "Coprocessor Response Size Per Instance and TiKV",
           "tooltip": {
             "shared": true,
             "sort": 0,

session/sessiontest/session_test.go

@@ -2453,6 +2453,10 @@ func TestRequestSource(t *testing.T) {
 					requestSource = r.GetContext().GetRequestSource()
 				case *coprocessor.Request:
 					requestSource = r.GetContext().GetRequestSource()
+				case *kvrpcpb.GetRequest:
+					requestSource = r.GetContext().GetRequestSource()
+				case *kvrpcpb.BatchGetRequest:
+					requestSource = r.GetContext().GetRequestSource()
 				}
 				require.Equal(t, source, requestSource)
 				return next(target, req)
@@ -2468,4 +2472,6 @@ func TestRequestSource(t *testing.T) {
 	tk.MustExecWithContext(insertCtx, "insert into t values(1, 1)")
 	selectCtx := interceptor.WithRPCInterceptor(context.Background(), withCheckInterceptor("external_Select_lightning"))
 	tk.MustExecWithContext(selectCtx, "select count(*) from t;")
+	tk.MustQueryWithContext(selectCtx, "select b from t where a = 1;")
+	tk.MustQueryWithContext(selectCtx, "select b from t where a in (1, 2, 3);")
 }

sessiontxn/internal/txn.go

@@ -72,6 +72,9 @@ func GetSnapshotWithTS(s sessionctx.Context, ts uint64, interceptor kv.SnapshotI
 	if tp := s.GetSessionVars().RequestSourceType; tp != "" {
 		snap.SetOption(kv.RequestSourceType, tp)
 	}
+	if tp := s.GetSessionVars().ExplicitRequestSourceType; tp != "" {
+		snap.SetOption(kv.ExplicitRequestSourceType, tp)
+	}
 	if s.GetSessionVars().LoadBasedReplicaReadThreshold > 0 {
 		snap.SetOption(kv.LoadBasedReplicaReadThreshold, s.GetSessionVars().LoadBasedReplicaReadThreshold)
 	}

store/driver/txn/snapshot.go

@@ -129,6 +129,8 @@ func (s *tikvSnapshot) SetOption(opt int, val interface{}) {
 		s.KVSnapshot.SetRequestSourceInternal(val.(bool))
 	case kv.RequestSourceType:
 		s.KVSnapshot.SetRequestSourceType(val.(string))
+	case kv.ExplicitRequestSourceType:
+		s.KVSnapshot.SetExplicitRequestSourceType(val.(string))
 	case kv.ReplicaReadAdjuster:
 		s.KVSnapshot.SetReplicaReadAdjuster(val.(txnkv.ReplicaReadAdjuster))
 	case kv.ScanBatchSize: