stmtsummary: add request-units info in statements_summary

pkg/executor/adapter.go

@@ -1894,6 +1894,10 @@ func (a *ExecStmt) SummaryStmt(succ bool) {
 	if tikvExecDetailRaw != nil {
 		tikvExecDetail = *(tikvExecDetailRaw.(*util.ExecDetails))
 	}
+	var ruDetail *util.RUDetails
+	if ruDetailRaw := a.GoCtx.Value(util.RUDetailsCtxKey); ruDetailRaw != nil {
+		ruDetail = ruDetailRaw.(*util.RUDetails)
+	}
 
 	if stmtCtx.WaitLockLeaseTime > 0 {
 		if execDetail.BackoffSleep == nil {
@@ -1949,6 +1953,8 @@ func (a *ExecStmt) SummaryStmt(succ bool) {
 		Prepared:            a.isPreparedStmt,
 		KeyspaceName:        keyspaceName,
 		KeyspaceID:          keyspaceID,
+		RUDetail:            ruDetail,
+		ResourceGroupName:   sessVars.StmtCtx.ResourceGroupName,
 	}
 	if a.retryCount > 0 {
 		stmtExecInfo.ExecRetryTime = costTime - sessVars.DurationParse - sessVars.DurationCompile - time.Since(a.retryStartTime)

pkg/infoschema/tables.go

@@ -1368,6 +1368,13 @@ var tableStatementsSummaryCols = []columnInfo{
 	{name: stmtsummary.Charset, tp: mysql.TypeVarchar, size: 64, comment: "Sampled charset"},
 	{name: stmtsummary.Collation, tp: mysql.TypeVarchar, size: 64, comment: "Sampled collation"},
 	{name: stmtsummary.PlanHint, tp: mysql.TypeVarchar, size: 64, comment: "Sampled plan hint"},
+	{name: stmtsummary.MaxRequestUnitReadStr, tp: mysql.TypeDouble, flag: mysql.NotNullFlag | mysql.UnsignedFlag, size: 22, comment: "Max read request-unit cost of these statements"},
+	{name: stmtsummary.AvgRequestUnitReadStr, tp: mysql.TypeDouble, flag: mysql.NotNullFlag | mysql.UnsignedFlag, size: 22, comment: "Average read request-unit cost of these statements"},
+	{name: stmtsummary.MaxRequestUnitWriteStr, tp: mysql.TypeDouble, flag: mysql.NotNullFlag | mysql.UnsignedFlag, size: 22, comment: "Max write request-unit cost of these statements"},
+	{name: stmtsummary.AvgRequestUnitWriteStr, tp: mysql.TypeDouble, flag: mysql.NotNullFlag | mysql.UnsignedFlag, size: 22, comment: "Average write request-unit cost of these statements"},
+	{name: stmtsummary.MaxQueuedRcTimeStr, tp: mysql.TypeLonglong, size: 22, flag: mysql.NotNullFlag | mysql.UnsignedFlag, comment: "Max time of waiting for available request-units"},
+	{name: stmtsummary.AvgQueuedRcTimeStr, tp: mysql.TypeLonglong, size: 22, flag: mysql.NotNullFlag | mysql.UnsignedFlag, comment: "Max time of waiting for available request-units"},
+	{name: stmtsummary.ResourceGroupName, tp: mysql.TypeVarchar, size: 64, comment: "Bind resource group name"},
 }
 
 var tableStorageStatsCols = []columnInfo{

pkg/planner/optimize.go

@@ -195,7 +195,7 @@ func Optimize(ctx context.Context, sctx sessionctx.Context, node ast.Node, is in
 				// if we are in a txn, should update the txn resource name to let the txn
 				// commit with the hint resource group.
 				if txn, err := sctx.Txn(false); err == nil && txn != nil && txn.Valid() {
-					kv.SetTxnResourceGroup(txn, sessVars.ResourceGroupName)
+					kv.SetTxnResourceGroup(txn, sessVars.StmtCtx.ResourceGroupName)
 				}
 			} else {
 				err := infoschema.ErrResourceGroupSupportDisabled

pkg/util/stmtsummary/evicted.go

@@ -381,4 +381,8 @@ func addInfo(addTo *stmtSummaryByDigestElement, addWith *stmtSummaryByDigestElem
 	addTo.sumWriteSQLRespTotal += addWith.sumWriteSQLRespTotal
 
 	addTo.sumErrors += addWith.sumErrors
+
+	addTo.StmtRUSummary.Merge(&addWith.StmtRUSummary)
+	// resourceGroupName might not be inited because when it is a evicted item.
+	addTo.resourceGroupName = addWith.resourceGroupName
 }

pkg/util/stmtsummary/reader.go

@@ -313,6 +313,13 @@ const (
 	Charset                           = "CHARSET"
 	Collation                         = "COLLATION"
 	PlanHint                          = "PLAN_HINT"
+	AvgRequestUnitReadStr             = "AVG_REQUEST_UNIT_READ"
+	MaxRequestUnitReadStr             = "MAX_REQUEST_UNIT_READ"
+	AvgRequestUnitWriteStr            = "AVG_REQUEST_UNIT_WRITE"
+	MaxRequestUnitWriteStr            = "MAX_REQUEST_UNIT_WRITE"
+	AvgQueuedRcTimeStr                = "AVG_QUEUED_RC_TIME"
+	MaxQueuedRcTimeStr                = "MAX_QUEUED_RC_TIME"
+	ResourceGroupName                 = "RESOURCE_GROUP"
 )
 
 type columnValueFactory func(reader *stmtSummaryReader, ssElement *stmtSummaryByDigestElement, ssbd *stmtSummaryByDigest) interface{}
@@ -632,4 +639,25 @@ var columnValueFactoryMap = map[string]columnValueFactory{
 	PlanHint: func(_ *stmtSummaryReader, ssElement *stmtSummaryByDigestElement, _ *stmtSummaryByDigest) interface{} {
 		return ssElement.planHint
 	},
+	AvgRequestUnitReadStr: func(_ *stmtSummaryReader, ssElement *stmtSummaryByDigestElement, _ *stmtSummaryByDigest) interface{} {
+		return avgSumFloat(ssElement.SumRRU, ssElement.execCount)
+	},
+	MaxRequestUnitReadStr: func(_ *stmtSummaryReader, ssElement *stmtSummaryByDigestElement, _ *stmtSummaryByDigest) interface{} {
+		return ssElement.MaxRRU
+	},
+	AvgRequestUnitWriteStr: func(_ *stmtSummaryReader, ssElement *stmtSummaryByDigestElement, _ *stmtSummaryByDigest) interface{} {
+		return avgSumFloat(ssElement.SumWRU, ssElement.execCount)
+	},
+	MaxRequestUnitWriteStr: func(_ *stmtSummaryReader, ssElement *stmtSummaryByDigestElement, _ *stmtSummaryByDigest) interface{} {
+		return ssElement.MaxWRU
+	},
+	AvgQueuedRcTimeStr: func(_ *stmtSummaryReader, ssElement *stmtSummaryByDigestElement, _ *stmtSummaryByDigest) interface{} {
+		return avgInt(int64(ssElement.SumRUWaitDuration), ssElement.execCount)
+	},
+	MaxQueuedRcTimeStr: func(_ *stmtSummaryReader, ssElement *stmtSummaryByDigestElement, _ *stmtSummaryByDigest) interface{} {
+		return int64(ssElement.MaxRUWaitDuration)
+	},
+	ResourceGroupName: func(_ *stmtSummaryReader, ssElement *stmtSummaryByDigestElement, _ *stmtSummaryByDigest) interface{} {
+		return ssElement.resourceGroupName
+	},
 }

pkg/util/stmtsummary/statement_summary.go

@@ -50,6 +50,8 @@ type stmtSummaryByDigestKey struct {
 	planDigest string
 	// `hash` is the hash value of this object.
 	hash []byte
+	// `resourceGroupName` is the resource group's name of this statement is bind to.
+	resourceGroupName string
 }
 
 // Hash implements SimpleLRUCache.Key.
@@ -62,6 +64,7 @@ func (key *stmtSummaryByDigestKey) Hash() []byte {
 		key.hash = append(key.hash, hack.Slice(key.schemaName)...)
 		key.hash = append(key.hash, hack.Slice(key.prevDigest)...)
 		key.hash = append(key.hash, hack.Slice(key.planDigest)...)
+		key.hash = append(key.hash, hack.Slice(key.resourceGroupName)...)
 	}
 	return key.hash
 }
@@ -211,6 +214,9 @@ type stmtSummaryByDigestElement struct {
 	// pessimistic execution retry information.
 	execRetryCount uint
 	execRetryTime  time.Duration
+	// request-units
+	resourceGroupName string
+	StmtRUSummary
 }
 
 // StmtExecInfo records execution information of each statement.
@@ -244,11 +250,13 @@ type StmtExecInfo struct {
 	ExecRetryCount      uint
 	ExecRetryTime       time.Duration
 	execdetails.StmtExecDetails
-	ResultRows      int64
-	TiKVExecDetails util.ExecDetails
-	Prepared        bool
-	KeyspaceName    string
-	KeyspaceID      uint32
+	ResultRows        int64
+	TiKVExecDetails   util.ExecDetails
+	Prepared          bool
+	KeyspaceName      string
+	KeyspaceID        uint32
+	ResourceGroupName string
+	RUDetail          *util.RUDetails
 }
 
 // newStmtSummaryByDigestMap creates an empty stmtSummaryByDigestMap.
@@ -299,10 +307,11 @@ func (ssMap *stmtSummaryByDigestMap) AddStatement(sei *StmtExecInfo) {
 	historySize := ssMap.historySize()
 
 	key := &stmtSummaryByDigestKey{
-		schemaName: sei.SchemaName,
-		digest:     sei.Digest,
-		prevDigest: sei.PrevSQLDigest,
-		planDigest: sei.PlanDigest,
+		schemaName:        sei.SchemaName,
+		digest:            sei.Digest,
+		prevDigest:        sei.PrevSQLDigest,
+		planDigest:        sei.PlanDigest,
+		resourceGroupName: sei.ResourceGroupName,
 	}
 	// Calculate hash value in advance, to reduce the time holding the lock.
 	key.Hash()
@@ -651,20 +660,21 @@ func newStmtSummaryByDigestElement(sei *StmtExecInfo, beginTime int64, intervalS
 		// PrevSQL is already truncated to cfg.Log.QueryLogMaxLen.
 		prevSQL: sei.PrevSQL,
 		// samplePlan needs to be decoded so it can't be truncated.
-		samplePlan:       samplePlan,
-		sampleBinaryPlan: binPlan,
-		planHint:         planHint,
-		indexNames:       sei.StmtCtx.IndexNames,
-		minLatency:       sei.TotalLatency,
-		firstSeen:        sei.StartTime,
-		lastSeen:         sei.StartTime,
-		backoffTypes:     make(map[string]int),
-		authUsers:        make(map[string]struct{}),
-		planInCache:      false,
-		planCacheHits:    0,
-		planInBinding:    false,
-		prepared:         sei.Prepared,
-		minResultRows:    math.MaxInt64,
+		samplePlan:        samplePlan,
+		sampleBinaryPlan:  binPlan,
+		planHint:          planHint,
+		indexNames:        sei.StmtCtx.IndexNames,
+		minLatency:        sei.TotalLatency,
+		firstSeen:         sei.StartTime,
+		lastSeen:          sei.StartTime,
+		backoffTypes:      make(map[string]int),
+		authUsers:         make(map[string]struct{}),
+		planInCache:       false,
+		planCacheHits:     0,
+		planInBinding:     false,
+		prepared:          sei.Prepared,
+		minResultRows:     math.MaxInt64,
+		resourceGroupName: sei.ResourceGroupName,
 	}
 	ssElement.add(sei, intervalSeconds)
 	return ssElement
@@ -888,6 +898,9 @@ func (ssElement *stmtSummaryByDigestElement) add(sei *StmtExecInfo, intervalSeco
 	ssElement.sumPDTotal += time.Duration(atomic.LoadInt64(&sei.TiKVExecDetails.WaitPDRespDuration))
 	ssElement.sumBackoffTotal += time.Duration(atomic.LoadInt64(&sei.TiKVExecDetails.BackoffDuration))
 	ssElement.sumWriteSQLRespTotal += sei.StmtExecDetails.WriteSQLRespDuration
+
+	// request-units
+	ssElement.StmtRUSummary.Add(sei.RUDetail)
 }
 
 // Truncate SQL to maxSQLLength.
@@ -949,9 +962,63 @@ func avgFloat(sum int64, count int64) float64 {
 	return 0
 }
 
+func avgSumFloat(sum float64, count int64) float64 {
+	if count > 0 {
+		return sum / float64(count)
+	}
+	return 0
+}
+
 func convertEmptyToNil(str string) interface{} {
 	if str == "" {
 		return nil
 	}
 	return str
 }
+
+// StmtRUSummary is the request-units summary for each type of statements.
+type StmtRUSummary struct {
+	SumRRU            float64       `json:"sum_rru"`
+	SumWRU            float64       `json:"sum_wru"`
+	SumRUWaitDuration time.Duration `json:"sum_ru_wait_duration"`
+	MaxRRU            float64       `json:"max_rru"`
+	MaxWRU            float64       `json:"max_wru"`
+	MaxRUWaitDuration time.Duration `json:"max_ru_wait_duration"`
+}
+
+// Add add a new sample value to the ru summary record.
+func (s *StmtRUSummary) Add(info *util.RUDetails) {
+	if info != nil {
+		rru := info.RRU()
+		s.SumRRU += rru
+		if s.MaxRRU < rru {
+			s.MaxRRU = rru
+		}
+		wru := info.WRU()
+		s.SumWRU += wru
+		if s.MaxWRU < wru {
+			s.MaxWRU = wru
+		}
+		ruWaitDur := info.RUWaitDuration()
+		s.SumRUWaitDuration += ruWaitDur
+		if s.MaxRUWaitDuration < ruWaitDur {
+			s.MaxRUWaitDuration = ruWaitDur
+		}
+	}
+}
+
+// Merge merges the value of 2 ru summary records.
+func (s *StmtRUSummary) Merge(other *StmtRUSummary) {
+	s.SumRRU += other.SumRRU
+	s.SumWRU += other.SumWRU
+	s.SumRUWaitDuration += other.SumRUWaitDuration
+	if s.MaxRRU < other.MaxRRU {
+		s.MaxRRU = other.MaxRRU
+	}
+	if s.MaxWRU < other.MaxWRU {
+		s.MaxWRU = other.MaxWRU
+	}
+	if s.MaxRUWaitDuration < other.MaxRUWaitDuration {
+		s.MaxRUWaitDuration = other.MaxRUWaitDuration
+	}
+}