sql, sqlstats: Refactor Statement Recording

- Adds a new RecordedStatementStatsBuilder struct to make building
  RecordedStmtStats easier.

- Moves the recording, buffering, and synchronization of sql stats
  to the ingester and removes the logic from StatsCollecotr

Resolves: CRDB-57024
Epic: CRDB-55081
Release note: None

Author: kyle-a-wong

pkg/sql/conn_executor.go

@@ -1268,7 +1268,6 @@ func (s *Server) newConnExecutor(
 		s.sqlStatsIngester,
 		ex.phaseTimes,
 		s.localSqlStats.GetCounters(),
-		s.cfg.SQLStatsTestingKnobs,
 	)
 	ex.dataMutatorIterator.OnApplicationNameChange = func(newName string) {
 		ex.applicationName.Store(newName)

pkg/sql/executor_statement_metrics.go

@@ -13,7 +13,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/contentionpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/idxrecommendations"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
-	"github.com/cockroachdb/cockroach/pkg/sql/sessionphase"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/metric"
@@ -167,47 +166,16 @@ func (ex *connExecutor) recordStatementSummary(
 	stmtErr error,
 	stats topLevelQueryStats,
 ) appstatspb.StmtFingerprintID {
-	phaseTimes := ex.statsCollector.PhaseTimes()
-
-	// Collect the statistics.
-	idleLatRaw := phaseTimes.GetIdleLatency(ex.statsCollector.PreviousPhaseTimes())
-	idleLatSec := idleLatRaw.Seconds()
-	runLatRaw := phaseTimes.GetRunLatency()
-	runLatSec := runLatRaw.Seconds()
-	parseLatSec := phaseTimes.GetParsingLatency().Seconds()
-	planLatSec := phaseTimes.GetPlanningLatency().Seconds()
-	// We want to exclude any overhead to reduce possible confusion.
-	svcLatRaw := phaseTimes.GetServiceLatencyNoOverhead()
-	svcLatSec := svcLatRaw.Seconds()
-
-	// processing latency: contributing towards SQL results.
-	processingLatSec := parseLatSec + planLatSec + runLatSec
-
-	// overhead latency: txn/retry management, error checking, etc
-	execOverheadSec := svcLatSec - processingLatSec
 
 	stmt := &planner.stmt
 	flags := planner.curPlan.flags
 	ex.recordStatementLatencyMetrics(
-		stmt, flags, automaticRetryTxnCount+automaticRetryStmtCount, runLatRaw, svcLatRaw,
+		stmt, flags, automaticRetryTxnCount+automaticRetryStmtCount, ex.statsCollector.RunLatency(), ex.statsCollector.ServiceLatency(),
 	)
 
-	fullScan := flags.IsSet(planFlagContainsFullIndexScan) || flags.IsSet(planFlagContainsFullTableScan)
-
 	idxRecommendations := idxrecommendations.FormatIdxRecommendations(planner.instrumentation.indexRecs)
 	queryLevelStats, queryLevelStatsOk := planner.instrumentation.GetQueryLevelStats()
 
-	var sqlInstanceIDs []int64
-	var kvNodeIDs []int32
-	if queryLevelStatsOk {
-		sqlInstanceIDs = make([]int64, 0, len(queryLevelStats.SQLInstanceIDs))
-		for _, sqlInstanceID := range queryLevelStats.SQLInstanceIDs {
-			sqlInstanceIDs = append(sqlInstanceIDs, int64(sqlInstanceID))
-		}
-		kvNodeIDs = queryLevelStats.KVNodeIDs
-	}
-	startTime := phaseTimes.GetSessionPhaseTime(sessionphase.PlannerStartExecStmt).ToUTC()
-	implicitTxn := flags.IsSet(planFlagImplicitTxn)
 	stmtFingerprintID := planner.instrumentation.fingerprintId
 	autoRetryReason := ex.state.mu.autoRetryReason
 	if automaticRetryStmtCount > 0 {
@@ -221,51 +189,45 @@ func (ex *connExecutor) recordStatementSummary(
 	ex.metrics.EngineMetrics.StatementIndexBytesWritten.Inc(stats.indexBytesWritten)
 
 	if ex.statsCollector.EnabledForTransaction() {
-		recordedStmtStats := &sqlstats.RecordedStmtStats{
-			FingerprintID:        stmtFingerprintID,
-			QuerySummary:         stmt.StmtSummary,
-			Generic:              flags.IsSet(planFlagGeneric),
-			AppliedStmtHints:     len(stmt.Hints) > 0,
-			DistSQL:              flags.ShouldBeDistributed(),
-			Vec:                  flags.IsSet(planFlagVectorized),
-			ImplicitTxn:          implicitTxn,
-			PlanHash:             planner.instrumentation.planGist.Hash(),
-			SessionID:            ex.planner.extendedEvalCtx.SessionID,
-			StatementID:          stmt.QueryID,
-			AutoRetryCount:       automaticRetryTxnCount + automaticRetryStmtCount,
-			Failed:               stmtErr != nil,
-			AutoRetryReason:      autoRetryReason,
-			RowsAffected:         rowsAffected,
-			IdleLatencySec:       idleLatSec,
-			ParseLatencySec:      parseLatSec,
-			PlanLatencySec:       planLatSec,
-			RunLatencySec:        runLatSec,
-			ServiceLatencySec:    svcLatSec,
-			OverheadLatencySec:   execOverheadSec,
-			BytesRead:            stats.bytesRead,
-			RowsRead:             stats.rowsRead,
-			RowsWritten:          stats.rowsWritten,
-			Nodes:                sqlInstanceIDs,
-			KVNodeIDs:            kvNodeIDs,
-			StatementType:        stmt.AST.StatementType(),
-			PlanGist:             planner.instrumentation.planGist.String(),
-			StatementError:       stmtErr,
-			IndexRecommendations: idxRecommendations,
-			Query:                stmt.StmtNoConstants,
-			StartTime:            startTime,
-			EndTime:              startTime.Add(svcLatRaw),
-			FullScan:             fullScan,
-			ExecStats:            queryLevelStats,
+		b := sqlstats.NewRecordedStatementStatsBuilder(
+			stmtFingerprintID,
+			planner.SessionData().Database,
+			stmt.StmtNoConstants,
+			stmt.StmtSummary,
+			stmt.AST.StatementType(),
+			ex.statsCollector.CurrentApplicationName(),
+		).
+			QueryID(stmt.QueryID).
+			SessionID(ex.planner.extendedEvalCtx.SessionID).
+			PlanMetadata(
+				flags.IsSet(planFlagGeneric),
+				flags.ShouldBeDistributed(),
+				flags.IsSet(planFlagVectorized),
+				flags.IsSet(planFlagImplicitTxn),
+				flags.IsSet(planFlagContainsFullIndexScan) || flags.IsSet(planFlagContainsFullTableScan),
+			).
+			PlanGist(planner.instrumentation.planGist.String(), planner.instrumentation.planGist.Hash()).
+			LatencyRecorder(ex.statsCollector).
+			QueryLevelStats(stats.bytesRead, stats.rowsRead, stats.rowsWritten).
+			ExecStats(queryLevelStats).
 			// TODO(mgartner): Use a slice of struct{uint64, uint64} instead of
 			// converting to strings.
-			Indexes:       planner.instrumentation.indexesUsed.Strings(),
-			Database:      planner.SessionData().Database,
-			QueryTags:     stmt.QueryTags,
-			App:           ex.statsCollector.CurrentApplicationName(),
-			UnderOuterTxn: ex.extraTxnState.underOuterTxn,
+			Indexes(planner.instrumentation.indexesUsed.Strings()).
+			AutoRetry(automaticRetryTxnCount+automaticRetryStmtCount, autoRetryReason).
+			RowsAffected(rowsAffected).
+			IndexRecommendations(idxRecommendations).
+			QueryTags(stmt.QueryTags).
+			StatementError(stmtErr)
+
+		if ex.extraTxnState.underOuterTxn {
+			b.UnderOuterTxn()
+		}
+
+		if len(stmt.Hints) > 0 {
+			b.AppliedStatementHints()
 		}
 
-		ex.statsCollector.RecordStatement(ctx, recordedStmtStats)
+		ex.statsCollector.RecordStatement(ctx, b.Build())
 	}
 
 	// Record statement execution statistics if span is recorded and no error was
@@ -314,11 +276,16 @@ func (ex *connExecutor) recordStatementSummary(
 		ex.extraTxnState.transactionStatementsHash.Add(uint64(stmtFingerprintID))
 	}
 	ex.extraTxnState.numRows += rowsAffected
-	ex.extraTxnState.idleLatency += idleLatRaw
+	ex.extraTxnState.idleLatency += ex.statsCollector.IdleLatency()
 
 	if log.V(2) {
 		// ages since significant epochs
-		sessionAge := phaseTimes.GetSessionAge().Seconds()
+		sessionAge := ex.statsCollector.PhaseTimes().GetSessionAge().Seconds()
+		parseLatSec := ex.statsCollector.ParsingLatency().Seconds()
+		planLatSec := ex.statsCollector.PlanningLatency().Seconds()
+		runLatSec := ex.statsCollector.RunLatency().Seconds()
+		svcLatSec := ex.statsCollector.ServiceLatency().Seconds()
+		execOverheadSec := ex.statsCollector.ExecOverheadLatency().Seconds()
 
 		log.Dev.Infof(ctx,
 			"query stats: %d rows, %d retries, "+

pkg/sql/sqlstats/BUILD.bazel

@@ -22,6 +22,7 @@ go_library(
         "//pkg/sql/sem/tree",
         "//pkg/sql/sqlcommenter",
         "//pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil",
+        "//pkg/util/buildutil",
         "//pkg/util/log",
         "//pkg/util/metric",
         "//pkg/util/stop",

pkg/sql/sqlstats/sslocal/sql_stats_ingester_test.go

@@ -412,7 +412,6 @@ func TestStatsCollectorIngester(t *testing.T) {
 		ingester,
 		phaseTimes,
 		uniqueServerCounts,
-		nil, // knobs
 	)
 
 	sessionID := clusterunique.IDFromBytes([]byte("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"))

pkg/sql/sqlstats/sslocal/sql_stats_ingestor.go

@@ -265,6 +265,25 @@ func (i *SQLStatsIngester) ingest(ctx context.Context, events *eventBuffer) {
 	}
 }
 
+func (i *SQLStatsIngester) RecordStatement(statement *sqlstats.RecordedStmtStats) {
+	i.BufferStatement(statement)
+	if i.testingKnobs != nil && i.testingKnobs.SynchronousSQLStats {
+		// Flush buffer and wait for the stats ingester to finish writing.
+		i.guard.ForceSync()
+		<-i.syncStatsTestingCh
+	}
+}
+
+func (i *SQLStatsIngester) RecordTransaction(transaction *sqlstats.RecordedTxnStats) {
+	i.BufferTransaction(transaction)
+
+	if i.testingKnobs != nil && i.testingKnobs.SynchronousSQLStats {
+		// Flush buffer and wait for the stats ingester to finish writing.
+		i.guard.ForceSync()
+		<-i.syncStatsTestingCh
+	}
+}
+
 func (i *SQLStatsIngester) BufferStatement(statement *sqlstats.RecordedStmtStats) {
 	if i.testingKnobs != nil && i.testingKnobs.IngesterStmtInterceptor != nil {
 		i.testingKnobs.IngesterStmtInterceptor(statement.SessionID, statement)

pkg/sql/sqlstats/sslocal/sql_stats_test.go

@@ -471,7 +471,6 @@ func TestExplicitTxnFingerprintAccounting(t *testing.T) {
 		ingester,
 		sessionphase.NewTimes(),
 		sqlStats.GetCounters(),
-		nil, /* knobs */
 	)
 
 	recordStats := func(testCase *tc) {
@@ -595,7 +594,6 @@ func TestAssociatingStmtStatsWithTxnFingerprint(t *testing.T) {
 			ingester,
 			sessionphase.NewTimes(),
 			sqlStats.GetCounters(),
-			nil, /* knobs */
 		)
 
 		ingester.Start(ctx, stopper)

pkg/sql/sqlstats/sslocal/sslocal_stats_collector.go

@@ -7,9 +7,9 @@ package sslocal
 
 import (
 	"context"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
-	"github.com/cockroachdb/cockroach/pkg/sql/appstatspb"
 	"github.com/cockroachdb/cockroach/pkg/sql/clusterunique"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessionphase"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
@@ -35,12 +35,6 @@ import (
 //     persist statement and transaction insights to an in-memory cache.
 //     Events are sent to the insights subsystem for async processing.
 type StatsCollector struct {
-	// stmtFingerprintID is the fingerprint ID of the current statement we are
-	// recording. Note that we don't observe sql stats for all statements (e.g. COMMIT).
-	// If no stats have been attempted to be recorded yet for the current statement,
-	// this value will be 0.
-	stmtFingerprintID appstatspb.StmtFingerprintID
-
 	// phaseTimes tracks session-level phase times.
 	phaseTimes sessionphase.Times
 
@@ -67,8 +61,7 @@ type StatsCollector struct {
 
 	statsIngester *SQLStatsIngester
 
-	st    *cluster.Settings
-	knobs *sqlstats.TestingKnobs
+	st *cluster.Settings
 }
 
 // NewStatsCollector returns an instance of StatsCollector.
@@ -78,27 +71,20 @@ func NewStatsCollector(
 	ingester *SQLStatsIngester,
 	phaseTime *sessionphase.Times,
 	uniqueServerCounts *ssmemstorage.SQLStatsAtomicCounters,
-	knobs *sqlstats.TestingKnobs,
 ) *StatsCollector {
 	s := &StatsCollector{
 		flushTarget:        appStats,
 		phaseTimes:         *phaseTime,
 		uniqueServerCounts: uniqueServerCounts,
 		statsIngester:      ingester,
 		st:                 st,
-		knobs:              knobs,
 	}
 
 	s.sendStats = s.enabled()
 
 	return s
 }
 
-// StatementFingerprintID returns the fingerprint ID for the current statement.
-func (s *StatsCollector) StatementFingerprintID() appstatspb.StmtFingerprintID {
-	return s.stmtFingerprintID
-}
-
 // PhaseTimes returns the sessionphase.Times that this StatsCollector is
 // currently tracking.
 func (s *StatsCollector) PhaseTimes() *sessionphase.Times {
@@ -118,7 +104,6 @@ func (s *StatsCollector) PreviousPhaseTimes() *sessionphase.Times {
 // Found a bug again? Consider refactoring.
 func (s *StatsCollector) Reset(appStats *ssmemstorage.Container, phaseTime *sessionphase.Times) {
 	s.flushTarget = appStats
-	s.stmtFingerprintID = 0
 	s.previousPhaseTimes = s.phaseTimes
 	s.phaseTimes = *phaseTime
 }
@@ -166,14 +151,8 @@ func (s *StatsCollector) RecordStatement(_ctx context.Context, value *sqlstats.R
 	if !s.sendStats {
 		return
 	}
-	s.stmtFingerprintID = value.FingerprintID
-	s.statsIngester.BufferStatement(value)
 
-	if s.knobs != nil && s.knobs.SynchronousSQLStats {
-		// Flush buffer and wait for the stats ingester to finish writing.
-		s.statsIngester.guard.ForceSync()
-		<-s.statsIngester.syncStatsTestingCh
-	}
+	s.statsIngester.RecordStatement(value)
 }
 
 // RecordTransaction sends the transaction statistics to the stats ingester.
@@ -182,13 +161,7 @@ func (s *StatsCollector) RecordTransaction(_ctx context.Context, value *sqlstats
 		return
 	}
 
-	s.statsIngester.BufferTransaction(value)
-
-	if s.knobs != nil && s.knobs.SynchronousSQLStats {
-		// Flush buffer and wait for the stats ingester to finish writing.
-		s.statsIngester.guard.ForceSync()
-		<-s.statsIngester.syncStatsTestingCh
-	}
+	s.statsIngester.RecordTransaction(value)
 }
 
 func (s *StatsCollector) EnabledForTransaction() bool {
@@ -205,3 +178,41 @@ func (s *StatsCollector) EnabledForTransaction() bool {
 func (s *StatsCollector) CurrentApplicationName() string {
 	return s.flushTarget.ApplicationName()
 }
+
+func (s *StatsCollector) RunLatency() time.Duration {
+	return s.PhaseTimes().GetRunLatency()
+}
+
+func (s *StatsCollector) IdleLatency() time.Duration {
+	return s.PhaseTimes().GetIdleLatency(s.PreviousPhaseTimes())
+}
+
+func (s *StatsCollector) ServiceLatency() time.Duration {
+	return s.PhaseTimes().GetServiceLatencyNoOverhead()
+}
+
+func (s *StatsCollector) ParsingLatency() time.Duration {
+	return s.PhaseTimes().GetParsingLatency()
+}
+
+func (s *StatsCollector) PlanningLatency() time.Duration {
+	return s.PhaseTimes().GetPlanningLatency()
+}
+
+func (s *StatsCollector) ProcessingLatency() time.Duration {
+	return s.ParsingLatency() + s.PlanningLatency() + s.RunLatency()
+}
+
+func (s *StatsCollector) ExecOverheadLatency() time.Duration {
+	return s.ServiceLatency() - s.ProcessingLatency()
+}
+
+func (s *StatsCollector) StartTime() time.Time {
+	return s.PhaseTimes().GetSessionPhaseTime(sessionphase.PlannerStartExecStmt).ToUTC()
+}
+
+func (s *StatsCollector) EndTime() time.Time {
+	return s.StartTime().Add(s.ServiceLatency())
+}
+
+var _ sqlstats.StatementLatencyRecorder = &StatsCollector{}