Merge #59299

59299: execstats: add cumulative contention time to the TraceAnalyzer r=RaduBerinde a=asubiotto

Depends on #59132 
Closes #56797 

This commit adds the summing of each processor's contention time to the
TraceAnalyzer. This stat is returned as a node and query level stat.

Additionally, this commit adds a ContentionTime field to the
StatementStatistics protobuf used in the DB Console.

Release note: None

Co-authored-by: Alfonso Subiotto Marques <alfonso@cockroachlabs.com>

pkg/roachpb/app_stats.go

@@ -131,14 +131,15 @@ func (s *StatementStatistics) Add(other *StatementStatistics) {
 	s.RowsRead.Add(other.RowsRead, s.Count, other.Count)
 
 	// Execution stats collected using a sampling approach.
-	statCollectionCount := s.ExecStatCollectionCount
-	if statCollectionCount == 0 && other.ExecStatCollectionCount == 0 {
+	execStatCollectionCount := s.ExecStatCollectionCount
+	if execStatCollectionCount == 0 && other.ExecStatCollectionCount == 0 {
 		// If both are zero, artificially set the receiver's count to one to avoid
 		// division by zero in Add.
-		statCollectionCount = 1
+		execStatCollectionCount = 1
 	}
-	s.BytesSentOverNetwork.Add(other.BytesSentOverNetwork, s.ExecStatCollectionCount, statCollectionCount)
-	s.MaxMemUsage.Add(other.MaxMemUsage, s.ExecStatCollectionCount, statCollectionCount)
+	s.BytesSentOverNetwork.Add(other.BytesSentOverNetwork, s.ExecStatCollectionCount, execStatCollectionCount)
+	s.MaxMemUsage.Add(other.MaxMemUsage, s.ExecStatCollectionCount, execStatCollectionCount)
+	s.ContentionTime.Add(other.ContentionTime, s.ExecStatCollectionCount, execStatCollectionCount)
 
 	if other.SensitiveInfo.LastErr != "" {
 		s.SensitiveInfo.LastErr = other.SensitiveInfo.LastErr
@@ -167,5 +168,6 @@ func (s *StatementStatistics) AlmostEqual(other *StatementStatistics, eps float6
 		s.BytesRead.AlmostEqual(other.BytesRead, eps) &&
 		s.RowsRead.AlmostEqual(other.RowsRead, eps) &&
 		s.BytesSentOverNetwork.AlmostEqual(other.BytesSentOverNetwork, eps) &&
-		s.MaxMemUsage.AlmostEqual(other.MaxMemUsage, eps)
+		s.MaxMemUsage.AlmostEqual(other.MaxMemUsage, eps) &&
+		s.ContentionTime.AlmostEqual(other.ContentionTime, eps)
 }

pkg/roachpb/app_stats.proto

@@ -101,8 +101,12 @@ message StatementStatistics {
   // the mean of the following NumericStat values:
   //  bytes_sent_over_network
   //  max_mem_usage
+  //  contention_time
   optional int64 exec_stat_collection_count = 19 [(gogoproto.nullable) = false];
 
+  // ContentionTime collects the time this statement spent contending.
+  optional NumericStat contention_time = 20 [(gogoproto.nullable) = false];
+
   // Note: be sure to update `sql/app_stats.go` and the comment above
   // exec_stat_collection_count when adding/removing fields here!
 }

pkg/sql/execstats/traceanalyzer.go

@@ -82,24 +82,30 @@ func NewFlowMetadata(flows map[roachpb.NodeID]*execinfrapb.FlowSpec) *FlowMetada
 	return a
 }
 
-// NodeLevelStats returns all the flow level stats that correspond to the given traces and flow metadata.
+// NodeLevelStats returns all the flow level stats that correspond to the given
+// traces and flow metadata.
+// TODO(asubiotto): Flatten this struct, we're currently allocating a map per
+//  stat.
 type NodeLevelStats struct {
 	NetworkBytesSentGroupedByNode map[roachpb.NodeID]int64
 	MaxMemoryUsageGroupedByNode   map[roachpb.NodeID]int64
 	KVBytesReadGroupedByNode      map[roachpb.NodeID]int64
 	KVRowsReadGroupedByNode       map[roachpb.NodeID]int64
 	KVTimeGroupedByNode           map[roachpb.NodeID]time.Duration
 	NetworkMessagesGroupedByNode  map[roachpb.NodeID]int64
+	ContentionTimeGroupedByNode   map[roachpb.NodeID]time.Duration
 }
 
-// QueryLevelStats returns all the query level stats that correspond to the given traces and flow metadata.
+// QueryLevelStats returns all the query level stats that correspond to the
+// given traces and flow metadata.
 type QueryLevelStats struct {
 	NetworkBytesSent int64
 	MaxMemUsage      int64
 	KVBytesRead      int64
 	KVRowsRead       int64
 	KVTime           time.Duration
 	NetworkMessages  int64
+	ContentionTime   time.Duration
 }
 
 // TraceAnalyzer is a struct that helps calculate top-level statistics from a
@@ -177,6 +183,7 @@ func (a *TraceAnalyzer) ProcessStats() error {
 		KVRowsReadGroupedByNode:       make(map[roachpb.NodeID]int64),
 		KVTimeGroupedByNode:           make(map[roachpb.NodeID]time.Duration),
 		NetworkMessagesGroupedByNode:  make(map[roachpb.NodeID]int64),
+		ContentionTimeGroupedByNode:   make(map[roachpb.NodeID]time.Duration),
 	}
 	var errs error
 
@@ -188,6 +195,7 @@ func (a *TraceAnalyzer) ProcessStats() error {
 		a.nodeLevelStats.KVBytesReadGroupedByNode[stats.nodeID] += int64(stats.stats.KV.BytesRead.Value())
 		a.nodeLevelStats.KVRowsReadGroupedByNode[stats.nodeID] += int64(stats.stats.KV.TuplesRead.Value())
 		a.nodeLevelStats.KVTimeGroupedByNode[stats.nodeID] += stats.stats.KV.KVTime.Value()
+		a.nodeLevelStats.ContentionTimeGroupedByNode[stats.nodeID] += stats.stats.KV.ContentionTime.Value()
 	}
 
 	// Process streamStats.
@@ -278,6 +286,10 @@ func (a *TraceAnalyzer) ProcessStats() error {
 	for _, networkMessages := range a.nodeLevelStats.NetworkMessagesGroupedByNode {
 		a.queryLevelStats.NetworkMessages += networkMessages
 	}
+
+	for _, contentionTime := range a.nodeLevelStats.KVTimeGroupedByNode {
+		a.queryLevelStats.ContentionTime += contentionTime
+	}
 	return errs
 }
 

pkg/sql/execstats/traceanalyzer_test.go

@@ -185,6 +185,11 @@ func TestTraceAnalyzer(t *testing.T) {
 }
 
 func TestTraceAnalyzerProcessStats(t *testing.T) {
+	const (
+		node1Time      = 3 * time.Second
+		node2Time      = 5 * time.Second
+		cumulativeTime = node1Time + node2Time
+	)
 	a := &execstats.TraceAnalyzer{FlowMetadata: &execstats.FlowMetadata{}}
 	a.AddComponentStats(
 		1, /* nodeID */
@@ -194,7 +199,8 @@ func TestTraceAnalyzerProcessStats(t *testing.T) {
 				1, /* processorID */
 			),
 			KV: execinfrapb.KVStats{
-				KVTime: optional.MakeTimeValue(3 * time.Second),
+				KVTime:         optional.MakeTimeValue(node1Time),
+				ContentionTime: optional.MakeTimeValue(node1Time),
 			},
 		},
 	)
@@ -207,12 +213,16 @@ func TestTraceAnalyzerProcessStats(t *testing.T) {
 				2, /* processorID */
 			),
 			KV: execinfrapb.KVStats{
-				KVTime: optional.MakeTimeValue(5 * time.Second),
+				KVTime:         optional.MakeTimeValue(node2Time),
+				ContentionTime: optional.MakeTimeValue(node2Time),
 			},
 		},
 	)
 
-	expected := execstats.QueryLevelStats{KVTime: 8 * time.Second}
+	expected := execstats.QueryLevelStats{
+		KVTime:         cumulativeTime,
+		ContentionTime: cumulativeTime,
+	}
 
 	assert.NoError(t, a.ProcessStats())
 	if got := a.GetQueryLevelStats(); !reflect.DeepEqual(got, expected) {

pkg/sql/instrumentation.go

@@ -244,6 +244,9 @@ func (ih *instrumentationHelper) Finish(
 			stmtStats.mu.data.MaxMemUsage.Record(
 				stmtStats.mu.data.ExecStatCollectionCount, float64(queryLevelStats.MaxMemUsage),
 			)
+			stmtStats.mu.data.ContentionTime.Record(
+				stmtStats.mu.data.ExecStatCollectionCount, queryLevelStats.ContentionTime.Seconds(),
+			)
 			stmtStats.mu.Unlock()
 		}
 	}

pkg/ui/src/util/appStats.ts

@@ -111,6 +111,15 @@ export function addStatementStats(
     exec_stat_collection_count: a.exec_stat_collection_count.add(
       b.exec_stat_collection_count,
     ),
+    contention_time:
+      a.contention_time && b.contention_time
+        ? addNumericStats(
+            a.contention_time,
+            b.contention_time,
+            execStatCountA,
+            execStatCountB,
+          )
+        : null,
   };
 }
 

pkg/ui/src/views/statements/statements.spec.tsx

@@ -493,6 +493,7 @@ function makeStats() {
     bytes_sent_over_network: makeStat(),
     max_mem_usage: makeStat(),
     exec_stat_collection_count: Long.fromNumber(10),
+    contention_time: makeStat(),
   };
 }