Merge #143016

143016: sslocal: stats collector cleanup r=xinhaoz a=xinhaoz

Commit 1
----------
### sqlstats: reduce stats collector calls from conn exec
Previously, separate functions in the stats collector were required to
send execution statistics for statements and transactions to sqlstats
and insights respectively.

This commit consolidates the sending of execution events to sqlstats
and insights into one function on the stats collector per event type
(one for statements and one for transactions).

Epic: none
Part of: #141024
Release note: None


Commit 2
----------
### sqlstats: add FingerprintID to sqlstats.RecordedTxnStats

This commit adds the `FingerprintID` field to sqlstats.RecordedTxnStats.
Epic: none
Part of: #141024

Release note: None

Commit 3
----------
### sqlstats,insights: add struct conversion functions
This commit adds functions converting sqlstats.Recorded{Stmt,Txn}Stats
to `insight.{Statement,Transaction}`.
Next: insights will only create insight objects if we are putting them
into the insights cache, after analyzing `sqlstats.Recorded*` objects.

Epic: none

Release note: None

Co-authored-by: Xin Hao Zhang <[email protected]>

Author: craig[bot]

pkg/sql/conn_executor_exec.go

@@ -4330,6 +4330,7 @@ func (ex *connExecutor) recordTransactionFinish(
 	commitLat := ex.phaseTimes.GetCommitLatency()
 
 	recordedTxnStats := sqlstats.RecordedTxnStats{
+		FingerprintID:           transactionFingerprintID,
 		SessionID:               ex.planner.extendedEvalCtx.SessionID,
 		TransactionID:           ev.txnID,
 		TransactionTimeSec:      elapsedTime.Seconds(),
@@ -4376,13 +4377,7 @@ func (ex *connExecutor) recordTransactionFinish(
 		)
 	}
 
-	ex.statsCollector.ObserveTransaction(ctx, transactionFingerprintID, recordedTxnStats)
-
-	return ex.statsCollector.RecordTransaction(
-		ctx,
-		transactionFingerprintID,
-		recordedTxnStats,
-	)
+	return ex.statsCollector.RecordTransaction(ctx, recordedTxnStats)
 }
 
 // Records a SERIALIZATION_CONFLICT contention event to the contention registry event

pkg/sql/executor_statement_metrics.go

@@ -235,8 +235,6 @@ func (ex *connExecutor) recordStatementSummary(
 		}
 	}
 
-	ex.statsCollector.ObserveStatement(stmtFingerprintID, recordedStmtStats)
-
 	// Do some transaction level accounting for the transaction this statement is
 	// a part of.
 

pkg/sql/sqlstats/insights/BUILD.bazel

@@ -14,6 +14,7 @@ go_library(
         "registry.go",
         "store.go",
         "test_utils.go",
+        "util.go",
     ],
     embed = [":insights_go_proto"],
     importpath = "github.com/cockroachdb/cockroach/pkg/sql/sqlstats/insights",
@@ -24,6 +25,8 @@ go_library(
         "//pkg/sql/appstatspb",
         "//pkg/sql/clusterunique",
         "//pkg/sql/contention/contentionutils",
+        "//pkg/sql/pgwire/pgerror",
+        "//pkg/sql/sqlstats",
         "//pkg/util/cache",
         "//pkg/util/intsets",
         "//pkg/util/metric",

pkg/sql/sqlstats/insights/ingester.go

@@ -51,8 +51,8 @@ type eventBufChPayload struct {
 	events        *eventBuffer
 }
 
-// ConcurrentBufferIngester buffers the "events" it sees (via ObserveStatement
-// and ObserveTransaction) and passes them along to the underlying registry
+// ConcurrentBufferIngester buffers the "events" it sees (via observeStatement
+// and observeTransaction) and passes them along to the underlying registry
 // once its buffer is full. (Or once a timeout has passed, for low-traffic
 // clusters and tests.)
 //

pkg/sql/sqlstats/insights/util.go

@@ -0,0 +1,121 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package insights
+
+import (
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
+	"github.com/cockroachdb/redact"
+)
+
+func MakeTxnInsight(value sqlstats.RecordedTxnStats) *Transaction {
+	var retryReason string
+	if value.AutoRetryReason != nil {
+		retryReason = value.AutoRetryReason.Error()
+	}
+
+	var cpuSQLNanos int64
+	if value.ExecStats.CPUTime.Nanoseconds() >= 0 {
+		cpuSQLNanos = value.ExecStats.CPUTime.Nanoseconds()
+	}
+
+	var errorCode string
+	var errorMsg redact.RedactableString
+	if value.TxnErr != nil {
+		errorCode = pgerror.GetPGCode(value.TxnErr).String()
+		errorMsg = redact.Sprint(value.TxnErr)
+	}
+
+	status := Transaction_Failed
+	if value.Committed {
+		status = Transaction_Completed
+	}
+
+	var user, appName string
+	if value.SessionData != nil {
+		user = value.SessionData.User().Normalized()
+		appName = value.SessionData.ApplicationName
+	}
+
+	insight := &Transaction{
+		ID:              value.TransactionID,
+		FingerprintID:   value.FingerprintID,
+		UserPriority:    value.Priority.String(),
+		ImplicitTxn:     value.ImplicitTxn,
+		Contention:      &value.ExecStats.ContentionTime,
+		StartTime:       value.StartTime,
+		EndTime:         value.EndTime,
+		User:            user,
+		ApplicationName: appName,
+		RowsRead:        value.RowsRead,
+		RowsWritten:     value.RowsWritten,
+		RetryCount:      value.RetryCount,
+		AutoRetryReason: retryReason,
+		CPUSQLNanos:     cpuSQLNanos,
+		LastErrorCode:   errorCode,
+		LastErrorMsg:    errorMsg,
+		Status:          status,
+	}
+
+	return insight
+}
+
+func MakeStmtInsight(value sqlstats.RecordedStmtStats) *Statement {
+	var autoRetryReason string
+	if value.AutoRetryReason != nil {
+		autoRetryReason = value.AutoRetryReason.Error()
+	}
+
+	var contention *time.Duration
+	var cpuSQLNanos int64
+	if value.ExecStats != nil {
+		contention = &value.ExecStats.ContentionTime
+		cpuSQLNanos = value.ExecStats.CPUTime.Nanoseconds()
+	}
+
+	var errorCode string
+	var errorMsg redact.RedactableString
+	if value.StatementError != nil {
+		errorCode = pgerror.GetPGCode(value.StatementError).String()
+		errorMsg = redact.Sprint(value.StatementError)
+	}
+
+	insight := &Statement{
+		ID:                   value.StatementID,
+		FingerprintID:        value.FingerprintID,
+		LatencyInSeconds:     value.ServiceLatencySec,
+		Query:                value.Query,
+		Status:               getInsightStatus(value.StatementError),
+		StartTime:            value.StartTime,
+		EndTime:              value.EndTime,
+		FullScan:             value.FullScan,
+		PlanGist:             value.PlanGist,
+		Retries:              int64(value.AutoRetryCount),
+		AutoRetryReason:      autoRetryReason,
+		RowsRead:             value.RowsRead,
+		RowsWritten:          value.RowsWritten,
+		Nodes:                value.Nodes,
+		KVNodeIDs:            value.KVNodeIDs,
+		Contention:           contention,
+		IndexRecommendations: value.IndexRecommendations,
+		Database:             value.Database,
+		CPUSQLNanos:          cpuSQLNanos,
+		ErrorCode:            errorCode,
+		ErrorMsg:             errorMsg,
+	}
+
+	return insight
+}
+
+func getInsightStatus(statementError error) Statement_Status {
+	if statementError == nil {
+		return Statement_Completed
+	}
+
+	return Statement_Failed
+}

pkg/sql/sqlstats/persistedsqlstats/bench_test.go

@@ -412,8 +412,10 @@ func BenchmarkSqlStatsMaxFlushTime(b *testing.B) {
 		}
 
 		for i := int64(1); i <= txnFingerprintLimit; i++ {
-			mockTxnValue := sqlstats.RecordedTxnStats{}
-			err := appContainer.RecordTransaction(ctx, appstatspb.TransactionFingerprintID(i), mockTxnValue)
+			mockTxnValue := sqlstats.RecordedTxnStats{
+				FingerprintID: appstatspb.TransactionFingerprintID(i),
+			}
+			err := appContainer.RecordTransaction(ctx, mockTxnValue)
 			if errors.Is(err, ssmemstorage.ErrFingerprintLimitReached) {
 				break
 			}

pkg/sql/sqlstats/sslocal/BUILD.bazel

@@ -20,7 +20,6 @@ go_library(
         "//pkg/settings/cluster",
         "//pkg/sql/appstatspb",
         "//pkg/sql/clusterunique",
-        "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/sessionphase",
         "//pkg/sql/sqlstats",
         "//pkg/sql/sqlstats/insights",
@@ -32,7 +31,6 @@ go_library(
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",
         "@com_github_cockroachdb_errors//:errors",
-        "@com_github_cockroachdb_redact//:redact",
     ],
 )
 

pkg/sql/sqlstats/sslocal/sql_stats_test.go

@@ -475,7 +475,8 @@ func TestExplicitTxnFingerprintAccounting(t *testing.T) {
 			statsCollector.EndTransaction(ctx, txnFingerprintID)
 			require.NoError(t,
 				statsCollector.
-					RecordTransaction(ctx, txnFingerprintID, sqlstats.RecordedTxnStats{
+					RecordTransaction(ctx, sqlstats.RecordedTxnStats{
+						FingerprintID: txnFingerprintID,
 						SessionData: &sessiondata.SessionData{
 							SessionData: sessiondatapb.SessionData{
 								UserProto:       username.RootUserName().EncodeProto(),
@@ -607,7 +608,8 @@ func TestAssociatingStmtStatsWithTxnFingerprint(t *testing.T) {
 
 			transactionFingerprintID := appstatspb.TransactionFingerprintID(txnFingerprintIDHash.Sum())
 			statsCollector.EndTransaction(ctx, transactionFingerprintID)
-			err := statsCollector.RecordTransaction(ctx, transactionFingerprintID, sqlstats.RecordedTxnStats{
+			err := statsCollector.RecordTransaction(ctx, sqlstats.RecordedTxnStats{
+				FingerprintID: transactionFingerprintID,
 				SessionData: &sessiondata.SessionData{
 					SessionData: sessiondatapb.SessionData{
 						UserProto:       username.RootUserName().EncodeProto(),