insights: insights and sqlstats subsystem should consume the same types

Previously, insights consumed its own data type containing per-execution
statisitcs about statements and transactions for outlier analysis. These
should only be created if we determine that the execution should be cached
in our insights registry.

SQL stats and insights will now consume the same data types to process and
perform their respective recordings: `sqlstats.RecordedStmtStats` and
`sqlstats.RecordedTxnStats`, which contain per execution statistics. This
is supporting work in moving sql stats recording to to occur in the same
async routine as insights.

This commit mostly deals with changing logic in the insights package and related
testing to perform analysis on `sqlstats.Recorded{Stmt,Txn}Stats` instead of
`insights.{Statement,Transaction}` types. We also change the function signatures
for the sqlstats containers to accept pointers to `sqlstats.Recorded*` objects
instead of by value.

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

Author: xinhaoz

Diff for: pkg/sql/conn_executor_exec.go

@@ -4329,7 +4329,7 @@ func (ex *connExecutor) recordTransactionFinish(
 	txnRetryLat := ex.phaseTimes.GetTransactionRetryLatency()
 	commitLat := ex.phaseTimes.GetCommitLatency()
 
-	recordedTxnStats := sqlstats.RecordedTxnStats{
+	recordedTxnStats := &sqlstats.RecordedTxnStats{
 		FingerprintID:           transactionFingerprintID,
 		SessionID:               ex.planner.extendedEvalCtx.SessionID,
 		TransactionID:           ev.txnID,
@@ -4373,7 +4373,7 @@ func (ex *connExecutor) recordTransactionFinish(
 		ex.planner.logTransaction(ctx,
 			int(ex.extraTxnState.txnCounter.Load()),
 			transactionFingerprintID,
-			&recordedTxnStats,
+			recordedTxnStats,
 			ex.extraTxnState.telemetrySkippedTxns,
 		)
 	}

Diff for: pkg/sql/exec_util.go

@@ -1745,7 +1745,7 @@ type ExecutorTestingKnobs struct {
 
 	// OnRecordTxnFinish, if set, will be called as we record a transaction
 	// finishing.
-	OnRecordTxnFinish func(isInternal bool, phaseTimes *sessionphase.Times, stmt string, txnStats sqlstats.RecordedTxnStats)
+	OnRecordTxnFinish func(isInternal bool, phaseTimes *sessionphase.Times, stmt string, txnStats *sqlstats.RecordedTxnStats)
 
 	// UseTransactionDescIDGenerator is used to force descriptor ID generation
 	// to use a transaction, and, in doing so, more deterministically allocate

Diff for: pkg/sql/executor_statement_metrics.go

@@ -167,7 +167,7 @@ func (ex *connExecutor) recordStatementSummary(
 	implicitTxn := flags.IsSet(planFlagImplicitTxn)
 	stmtFingerprintID := appstatspb.ConstructStatementFingerprintID(
 		stmt.StmtNoConstants, implicitTxn, planner.SessionData().Database)
-	recordedStmtStats := sqlstats.RecordedStmtStats{
+	recordedStmtStats := &sqlstats.RecordedStmtStats{
 		FingerprintID:        stmtFingerprintID,
 		QuerySummary:         stmt.StmtSummary,
 		DistSQL:              flags.ShouldBeDistributed(),
@@ -207,6 +207,7 @@ func (ex *connExecutor) recordStatementSummary(
 	}
 
 	err := ex.statsCollector.RecordStatement(ctx, recordedStmtStats)
+
 	if err != nil {
 		if log.V(1) {
 			log.Warningf(ctx, "failed to record statement: %s", err)

Diff for: pkg/sql/instrumentation_test.go

@@ -207,12 +207,12 @@ func TestSampledStatsCollectionOnNewFingerprint(t *testing.T) {
 
 	testApp := `sampling-test`
 	ctx := context.Background()
-	var collectedTxnStats []sqlstats.RecordedTxnStats
+	var collectedTxnStats []*sqlstats.RecordedTxnStats
 	s := serverutils.StartServerOnly(t, base.TestServerArgs{
 		Knobs: base.TestingKnobs{
 			SQLExecutor: &ExecutorTestingKnobs{
 				DisableProbabilisticSampling: true,
-				OnRecordTxnFinish: func(isInternal bool, _ *sessionphase.Times, stmt string, txnStats sqlstats.RecordedTxnStats) {
+				OnRecordTxnFinish: func(isInternal bool, _ *sessionphase.Times, stmt string, txnStats *sqlstats.RecordedTxnStats) {
 					// We won't run into a race here because we'll only observe
 					// txns from a single connection.
 					if txnStats.Application == testApp {

Diff for: pkg/sql/sqlstats/insights/BUILD.bazel

@@ -54,6 +54,10 @@ go_test(
         "//pkg/settings/cluster",
         "//pkg/sql/appstatspb",
         "//pkg/sql/clusterunique",
+        "//pkg/sql/execstats",
+        "//pkg/sql/pgwire/pgcode",
+        "//pkg/sql/pgwire/pgerror",
+        "//pkg/sql/sqlstats",
         "//pkg/util/leaktest",
         "//pkg/util/log",
         "//pkg/util/stop",

Diff for: pkg/sql/sqlstats/insights/detector.go

@@ -11,13 +11,14 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/appstatspb"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
 	"github.com/cockroachdb/cockroach/pkg/util/quantile"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 )
 
 type detector interface {
 	enabled() bool
-	isSlow(*Statement) bool
+	isSlow(stats *sqlstats.RecordedStmtStats) bool
 }
 
 var _ detector = &compositeDetector{}
@@ -37,7 +38,7 @@ func (d *compositeDetector) enabled() bool {
 	return false
 }
 
-func (d *compositeDetector) isSlow(statement *Statement) bool {
+func (d *compositeDetector) isSlow(statement *sqlstats.RecordedStmtStats) bool {
 	// Because some detectors may need to observe all statements to build up
 	// their baseline sense of what "normal" is, we avoid short-circuiting.
 	result := false
@@ -69,18 +70,18 @@ func (d *AnomalyDetector) enabled() bool {
 	return AnomalyDetectionEnabled.Get(&d.settings.SV)
 }
 
-func (d *AnomalyDetector) isSlow(stmt *Statement) (decision bool) {
+func (d *AnomalyDetector) isSlow(stmt *sqlstats.RecordedStmtStats) (decision bool) {
 	if !d.enabled() {
 		return
 	}
 
-	d.withFingerprintLatencySummary(stmt, func(latencySummary *quantile.Stream) {
-		latencySummary.Insert(stmt.LatencyInSeconds)
+	d.withFingerprintLatencySummary(stmt.FingerprintID, stmt.ServiceLatencySec, func(latencySummary *quantile.Stream) {
+		latencySummary.Insert(stmt.ServiceLatencySec)
 		p50 := latencySummary.Query(0.5, true)
 		p99 := latencySummary.Query(0.99, true)
-		decision = stmt.LatencyInSeconds >= p99 &&
-			stmt.LatencyInSeconds >= 2*p50 &&
-			stmt.LatencyInSeconds >= AnomalyDetectionLatencyThreshold.Get(&d.settings.SV).Seconds()
+		decision = stmt.ServiceLatencySec >= p99 &&
+			stmt.ServiceLatencySec >= 2*p50 &&
+			stmt.ServiceLatencySec >= AnomalyDetectionLatencyThreshold.Get(&d.settings.SV).Seconds()
 	})
 
 	return
@@ -104,21 +105,23 @@ func (d *AnomalyDetector) GetPercentileValues(id appstatspb.StmtFingerprintID) P
 }
 
 func (d *AnomalyDetector) withFingerprintLatencySummary(
-	stmt *Statement, consumer func(latencySummary *quantile.Stream),
+	stmtFingerprintID appstatspb.StmtFingerprintID,
+	latencySecs float64,
+	consumer func(latencySummary *quantile.Stream),
 ) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	var latencySummary *quantile.Stream
 
-	if element, ok := d.mu.index[stmt.FingerprintID]; ok {
+	if element, ok := d.mu.index[stmtFingerprintID]; ok {
 		// We are already tracking latencies for this fingerprint.
 		latencySummary = element.Value.(latencySummaryEntry).value
 		d.store.MoveToFront(element) // Mark this latency summary as recently used.
-	} else if stmt.LatencyInSeconds >= AnomalyDetectionLatencyThreshold.Get(&d.settings.SV).Seconds() {
+	} else if latencySecs >= AnomalyDetectionLatencyThreshold.Get(&d.settings.SV).Seconds() {
 		// We want to start tracking latencies for this fingerprint.
 		latencySummary = quantile.NewTargeted(desiredQuantiles)
-		entry := latencySummaryEntry{key: stmt.FingerprintID, value: latencySummary}
-		d.mu.index[stmt.FingerprintID] = d.store.PushFront(entry)
+		entry := latencySummaryEntry{key: stmtFingerprintID, value: latencySummary}
+		d.mu.index[stmtFingerprintID] = d.store.PushFront(entry)
 		d.metrics.Fingerprints.Inc(1)
 		d.metrics.Memory.Inc(latencySummary.ByteSize())
 	} else {
@@ -160,18 +163,14 @@ func (d *latencyThresholdDetector) enabled() bool {
 	return LatencyThreshold.Get(&d.st.SV) > 0
 }
 
-func (d *latencyThresholdDetector) isSlow(s *Statement) bool {
-	return d.enabled() && s.LatencyInSeconds >= LatencyThreshold.Get(&d.st.SV).Seconds()
-}
-
-func isFailed(s *Statement) bool {
-	return s.Status == Statement_Failed
+func (d *latencyThresholdDetector) isSlow(s *sqlstats.RecordedStmtStats) bool {
+	return d.enabled() && s.ServiceLatencySec >= LatencyThreshold.Get(&d.st.SV).Seconds()
 }
 
 var prefixesToIgnore = []string{"SET ", "EXPLAIN "}
 
 // shouldIgnoreStatement returns true if we don't want to analyze the statement.
-func shouldIgnoreStatement(s *Statement) bool {
+func shouldIgnoreStatement(s *sqlstats.RecordedStmtStats) bool {
 	for _, start := range prefixesToIgnore {
 		if strings.HasPrefix(s.Query, start) {
 			return true

Diff for: pkg/sql/sqlstats/insights/detector_test.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/appstatspb"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
 	"github.com/stretchr/testify/require"
 )
 
@@ -34,17 +35,17 @@ func TestAnyDetector(t *testing.T) {
 
 	t.Run("isSlow is false without any detectors", func(t *testing.T) {
 		detector := &compositeDetector{}
-		require.False(t, detector.isSlow(&Statement{}))
+		require.False(t, detector.isSlow(&sqlstats.RecordedStmtStats{}))
 	})
 
 	t.Run("isSlow is false without any concerned detectors", func(t *testing.T) {
 		detector := &compositeDetector{[]detector{&fakeDetector{}, &fakeDetector{}}}
-		require.False(t, detector.isSlow(&Statement{}))
+		require.False(t, detector.isSlow(&sqlstats.RecordedStmtStats{}))
 	})
 
 	t.Run("isSlow is true with at least one concerned detector", func(t *testing.T) {
 		detector := &compositeDetector{[]detector{&fakeDetector{stubIsSlow: true}, &fakeDetector{}}}
-		require.True(t, detector.isSlow(&Statement{}))
+		require.True(t, detector.isSlow(&sqlstats.RecordedStmtStats{}))
 	})
 
 	t.Run("isSlow consults all detectors without short-circuiting", func(t *testing.T) {
@@ -55,7 +56,7 @@ func TestAnyDetector(t *testing.T) {
 		d2 := &fakeDetector{stubIsSlow: true}
 
 		detector := &compositeDetector{[]detector{d1, d2}}
-		detector.isSlow(&Statement{})
+		detector.isSlow(&sqlstats.RecordedStmtStats{})
 		require.True(t, d1.isSlowCalled, "the first detector should be consulted")
 		require.True(t, d2.isSlowCalled, "the second detector should be consulted")
 	})
@@ -106,16 +107,18 @@ func TestLatencyQuantileDetector(t *testing.T) {
 		for _, test := range tests {
 			t.Run(test.name, func(t *testing.T) {
 				d := newAnomalyDetector(st, NewMetrics())
+				stmtWithSeedLatency := &sqlstats.RecordedStmtStats{ServiceLatencySec: test.seedLatency.Seconds()}
 				for i := 0; i < 1000; i++ {
-					d.isSlow(&Statement{LatencyInSeconds: test.seedLatency.Seconds()})
+					d.isSlow(stmtWithSeedLatency)
 				}
-				require.Equal(t, test.isSlow, d.isSlow(&Statement{LatencyInSeconds: test.candidateLatency.Seconds()}))
+				// Now determine if the candidate latency is slow.
+				require.Equal(t, test.isSlow, d.isSlow(&sqlstats.RecordedStmtStats{ServiceLatencySec: test.candidateLatency.Seconds()}))
 			})
 		}
 	})
 
 	// Testing the slow and failure detectors at the same time.
-	t.Run("isSlow and isFailed", func(t *testing.T) {
+	t.Run("isSlow with slow and failed statement", func(t *testing.T) {
 		ctx := context.Background()
 		st := cluster.MakeTestingClusterSettings()
 		AnomalyDetectionEnabled.Override(ctx, &st.SV, true)
@@ -127,46 +130,41 @@ func TestLatencyQuantileDetector(t *testing.T) {
 			candidateLatency time.Duration
 			status           Statement_Status
 			isSlow           bool
-			isFailed         bool
 		}{{
 			name:             "slow and failed statement",
 			seedLatency:      100 * time.Millisecond,
 			candidateLatency: 200 * time.Millisecond,
 			status:           Statement_Failed,
 			isSlow:           true,
-			isFailed:         true,
 		}, {
 			name:             "slow and non-failed statement",
 			seedLatency:      100 * time.Millisecond,
 			candidateLatency: 200 * time.Millisecond,
 			status:           Statement_Completed,
 			isSlow:           true,
-			isFailed:         false,
 		}, {
 			name:             "fast and non-failed statement",
 			seedLatency:      100 * time.Millisecond,
 			candidateLatency: 50 * time.Millisecond,
 			status:           Statement_Completed,
 			isSlow:           false,
-			isFailed:         false,
 		}, {
 			name:             "fast and failed statement",
 			seedLatency:      100 * time.Millisecond,
 			candidateLatency: 50 * time.Millisecond,
 			status:           Statement_Failed,
 			isSlow:           false,
-			isFailed:         true,
 		}}
 
 		for _, test := range tests {
 			t.Run(test.name, func(t *testing.T) {
 				d := newAnomalyDetector(st, NewMetrics())
+				stmtWithSeedLatency := &sqlstats.RecordedStmtStats{ServiceLatencySec: test.seedLatency.Seconds()}
 				for i := 0; i < 1000; i++ {
-					d.isSlow(&Statement{LatencyInSeconds: test.seedLatency.Seconds()})
+					d.isSlow(stmtWithSeedLatency)
 				}
-				stmt := &Statement{LatencyInSeconds: test.candidateLatency.Seconds(), Status: test.status}
-				require.Equal(t, test.isSlow, d.isSlow(stmt))
-				require.Equal(t, test.isFailed, isFailed(stmt))
+				// Now determine if the candidate latency is slow.
+				require.Equal(t, test.isSlow, d.isSlow(&sqlstats.RecordedStmtStats{ServiceLatencySec: test.candidateLatency.Seconds()}))
 			})
 		}
 	})
@@ -223,10 +221,11 @@ func TestLatencyQuantileDetector(t *testing.T) {
 				d := newAnomalyDetector(st, metrics)
 				// Show the detector `test.fingerprints` distinct fingerprints.
 				for i := 0; i < test.fingerprints; i++ {
-					d.isSlow(&Statement{
-						LatencyInSeconds: AnomalyDetectionLatencyThreshold.Get(&st.SV).Seconds(),
-						FingerprintID:    appstatspb.StmtFingerprintID(i),
-					})
+					stmt := &sqlstats.RecordedStmtStats{
+						ServiceLatencySec: AnomalyDetectionLatencyThreshold.Get(&st.SV).Seconds(),
+						FingerprintID:     appstatspb.StmtFingerprintID(i),
+					}
+					d.isSlow(stmt)
 				}
 				test.assertion(t, metrics)
 			})
@@ -241,10 +240,11 @@ func BenchmarkLatencyQuantileDetector(b *testing.B) {
 	settings := cluster.MakeTestingClusterSettings()
 	AnomalyDetectionEnabled.Override(context.Background(), &settings.SV, true)
 	d := newAnomalyDetector(settings, NewMetrics())
+	stmt := &sqlstats.RecordedStmtStats{
+		ServiceLatencySec: random.Float64(),
+	}
 	for i := 0; i < b.N; i++ {
-		d.isSlow(&Statement{
-			LatencyInSeconds: random.Float64(),
-		})
+		d.isSlow(stmt)
 	}
 }
 
@@ -267,21 +267,21 @@ func TestLatencyThresholdDetector(t *testing.T) {
 		st := cluster.MakeTestingClusterSettings()
 		LatencyThreshold.Override(context.Background(), &st.SV, 0)
 		detector := latencyThresholdDetector{st: st}
-		require.False(t, detector.isSlow(&Statement{LatencyInSeconds: 1}))
+		require.False(t, detector.isSlow(&sqlstats.RecordedStmtStats{ServiceLatencySec: 1}))
 	})
 
 	t.Run("isSlow false when fast enough", func(t *testing.T) {
 		st := cluster.MakeTestingClusterSettings()
 		LatencyThreshold.Override(context.Background(), &st.SV, 1*time.Second)
 		detector := latencyThresholdDetector{st: st}
-		require.False(t, detector.isSlow(&Statement{LatencyInSeconds: 0.5}))
+		require.False(t, detector.isSlow(&sqlstats.RecordedStmtStats{ServiceLatencySec: 0.5}))
 	})
 
 	t.Run("isSlow true beyond threshold", func(t *testing.T) {
 		st := cluster.MakeTestingClusterSettings()
 		LatencyThreshold.Override(context.Background(), &st.SV, 1*time.Second)
 		detector := latencyThresholdDetector{st: st}
-		require.True(t, detector.isSlow(&Statement{LatencyInSeconds: 1}))
+		require.True(t, detector.isSlow(&sqlstats.RecordedStmtStats{ServiceLatencySec: 1}))
 	})
 }
 
@@ -295,7 +295,7 @@ func (f *fakeDetector) enabled() bool {
 	return f.stubEnabled
 }
 
-func (f *fakeDetector) isSlow(*Statement) bool {
+func (f *fakeDetector) isSlow(stats *sqlstats.RecordedStmtStats) bool {
 	f.isSlowCalled = true
 	return f.stubIsSlow
 }

Diff for: pkg/sql/sqlstats/insights/ingester.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/sql/clusterunique"
 	"github.com/cockroachdb/cockroach/pkg/sql/contention/contentionutils"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 )
 
@@ -69,8 +70,8 @@ var eventBufferPool = sync.Pool{
 
 type event struct {
 	sessionID   clusterunique.ID
-	transaction *Transaction
-	statement   *Statement
+	transaction *sqlstats.RecordedTxnStats
+	statement   *sqlstats.RecordedStmtStats
 }
 
 type BufferOpt func(i *ConcurrentBufferIngester)
@@ -167,7 +168,7 @@ func (i *ConcurrentBufferIngester) ingest(events *eventBuffer) {
 }
 
 func (i *ConcurrentBufferIngester) ObserveStatement(
-	sessionID clusterunique.ID, statement *Statement,
+	sessionID clusterunique.ID, statement *sqlstats.RecordedStmtStats,
 ) {
 	if !i.registry.enabled() {
 		return
@@ -187,7 +188,7 @@ func (i *ConcurrentBufferIngester) ObserveStatement(
 }
 
 func (i *ConcurrentBufferIngester) ObserveTransaction(
-	sessionID clusterunique.ID, transaction *Transaction,
+	sessionID clusterunique.ID, transaction *sqlstats.RecordedTxnStats,
 ) {
 	if !i.registry.enabled() {
 		return