sql/contention: add blocking statement to contention event tables

pkg/kv/kvpb/api.proto

@@ -3900,6 +3900,12 @@ message ContentionEvent {
   // IsLatch, if true, indicates that the contention event was due to waiting to
   // acquire a latch.
   bool is_latch = 4;
+
+  // BlockingStmtFingerprintID is the fingerprint ID of the statement that held
+  // the contended lock.
+  uint64 blocking_stmt_fingerprint_id = 5 [
+    (gogoproto.customname) = "BlockingStmtFingerprintID"
+  ];
 }
 
 // QuorumReplicationFlowAdmissionEvent is a message that is recorded on traces by a

pkg/sql/crdb_internal.go

@@ -7912,6 +7912,7 @@ CREATE TABLE crdb_internal.transaction_contention_events (
 
     blocking_txn_id              UUID NOT NULL,
     blocking_txn_fingerprint_id  BYTES NOT NULL,
+    blocking_stmt_fingerprint_id BYTES,
 
     waiting_txn_id               UUID NOT NULL,
     waiting_txn_fingerprint_id   BYTES NOT NULL,
@@ -7957,17 +7958,24 @@ CREATE TABLE crdb_internal.transaction_contention_events (
 			return nil, nil, err
 		}
 
-		const numDatums = 15
+		const numDatums = 16
 		row := make(tree.Datums, numDatums)
 		worker := func(ctx context.Context, pusher rowPusher) error {
 			for i := range resp.Events {
 				collectionTs, err := tree.MakeDTimestampTZ(resp.Events[i].CollectionTs, time.Microsecond)
 				if err != nil {
 					return err
 				}
-				blockingFingerprintID := tree.NewDBytes(
+				blockingTxnFingerprintID := tree.NewDBytes(
 					tree.DBytes(sqlstatsutil.EncodeUint64ToBytes(uint64(resp.Events[i].BlockingTxnFingerprintID))))
 
+				blockingStmtFingerprintID := tree.DNull
+				if resp.Events[i].BlockingEvent.BlockingStmtFingerprintID != 0 {
+					blockingStmtFingerprintID = tree.NewDBytes(
+						tree.DBytes(sqlstatsutil.EncodeUint64ToBytes(
+							resp.Events[i].BlockingEvent.BlockingStmtFingerprintID)))
+				}
+
 				waitingFingerprintID := tree.NewDBytes(
 					tree.DBytes(sqlstatsutil.EncodeUint64ToBytes(uint64(resp.Events[i].WaitingTxnFingerprintID))))
 
@@ -7998,18 +8006,19 @@ CREATE TABLE crdb_internal.transaction_contention_events (
 				row = append(row[:0],
 					collectionTs, // collection_ts
 					tree.NewDUuid(tree.DUuid{UUID: resp.Events[i].BlockingEvent.TxnMeta.ID}), // blocking_txn_id
-					blockingFingerprintID, // blocking_fingerprint_id
+					blockingTxnFingerprintID,                                     // blocking_txn_fingerprint_id
+					blockingStmtFingerprintID,                                    // blocking_stmt_fingerprint_id
 					tree.NewDUuid(tree.DUuid{UUID: resp.Events[i].WaitingTxnID}), // waiting_txn_id
-					waitingFingerprintID,        // waiting_fingerprint_id
-					contentionDuration,          // contention_duration
-					contendingKey,               // contending_key,
-					contendingPrettyKey,         // contending_pretty_key
-					waitingStmtId,               // waiting_stmt_id
-					waitingStmtFingerprintID,    // waiting_stmt_fingerprint_id
-					tree.NewDString(dbName),     // database_name
-					tree.NewDString(schemaName), // schema_name
-					tree.NewDString(tableName),  // table_name
-					tree.NewDString(indexName),  // index_name
+					waitingFingerprintID,                                         // waiting_fingerprint_id
+					contentionDuration,                                           // contention_duration
+					contendingKey,                                                // contending_key,
+					contendingPrettyKey,                                          // contending_pretty_key
+					waitingStmtId,                                                // waiting_stmt_id
+					waitingStmtFingerprintID,                                     // waiting_stmt_fingerprint_id
+					tree.NewDString(dbName),                                      // database_name
+					tree.NewDString(schemaName),                                  // schema_name
+					tree.NewDString(tableName),                                   // table_name
+					tree.NewDString(indexName),                                   // index_name
 					tree.NewDString(resp.Events[i].ContentionType.String()),
 				)
 				if buildutil.CrdbTestBuild {

pkg/sql/crdb_internal_test.go

@@ -1925,3 +1925,81 @@ func TestSupportedCRDBInternalTablesNotChanged(t *testing.T) {
 		}
 	}
 }
+
+func TestBlockingStmtFingerprintIDInContentionEventsTable(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	srv, sqlDB, _ := serverutils.StartServer(t, base.TestServerArgs{
+		DefaultTestTenant: base.TestIsSpecificToStorageLayerAndNeedsASystemTenant,
+	})
+	defer srv.Stopper().Stop(ctx)
+	s := srv.ApplicationLayer()
+
+	_, err := sqlDB.Exec("SET CLUSTER SETTING sql.contention.event_store.resolution_interval = '10ms'")
+	require.NoError(t, err)
+
+	testCases := []struct {
+		name                      string
+		key                       string
+		blockingStmtFingerprintID uint64
+		expected                  gosql.NullString
+	}{
+		{
+			name:                      "non-null",
+			key:                       "/Animals/Cats",
+			blockingStmtFingerprintID: 9005,
+			expected:                  gosql.NullString{String: fmt.Sprintf("%016x", uint64(9005)), Valid: true},
+		},
+		{
+			name: "null",
+			key:  "/Animals/Dogs",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			keyEscaped := fmt.Sprintf("\"%s\"", tc.key)
+			s.ExecutorConfig().(sql.ExecutorConfig).ContentionRegistry.AddContentionEvent(contentionpb.ExtendedContentionEvent{
+				BlockingEvent: kvpb.ContentionEvent{
+					Key: roachpb.Key(tc.key),
+					TxnMeta: enginepb.TxnMeta{
+						Key: roachpb.Key(tc.key),
+						ID:  uuid.MakeV4(),
+					},
+					Duration:                  1 * time.Minute,
+					BlockingStmtFingerprintID: tc.blockingStmtFingerprintID,
+				},
+				BlockingTxnFingerprintID: 9001,
+				WaitingTxnID:             uuid.MakeV4(),
+				WaitingTxnFingerprintID:  9002,
+				WaitingStmtID:            clusterunique.ID{Uint128: uint128.Uint128{Lo: 9003, Hi: 1004}},
+				WaitingStmtFingerprintID: 9004,
+				ContentionType:           contentionpb.ContentionType_LOCK_WAIT,
+			})
+
+			// Contention flush can take some time to flush the events.
+			testutils.SucceedsSoon(t, func() error {
+				row := sqlDB.QueryRow(`SELECT
+    encode(blocking_stmt_fingerprint_id, 'hex')
+		FROM crdb_internal.transaction_contention_events
+		WHERE contending_pretty_key = $1`, keyEscaped)
+
+				var actual gosql.NullString
+				err = row.Scan(&actual)
+				if err != nil {
+					return err
+				}
+				if actual != tc.expected {
+					return errors.Newf(
+						"unexpected blocking stmt fingerprint id: got %+v, expected %+v",
+						actual,
+						tc.expected,
+					)
+				}
+				return nil
+			})
+		})
+	}
+}