sqlstats: generalize insight.ConcurrentBufferIngester

This commit copies the ConcurrentBufferIngester from insights into the sqlstats
pkg in preparation to have it consume the sql execution event and handle the
writing to both sqlstats and insights systems, instead of just for insights. At
this stage we are preserving how this ingester consumes events. The ingester is
still only used for insights writing.

Previously, the insights registry had the responsibility of buffering statements
by session id and processing them once the entire transaction completes (via
receicing a transaction event). The responsibility of buffering statements by
session has now moved to the ingester. As a result we remove the
`ObserveStatement` function from the insights registry which was only used to
insert into the statement buffer.

Part of: cockroachdb#141024

Epic: none
Release note: None

Author: xinhaoz

Diff for: pkg/sql/conn_executor.go

@@ -343,6 +343,9 @@ type Server struct {
 	// sqlStatsController is the control-plane interface for sqlStats.
 	sqlStatsController *persistedsqlstats.Controller
 
+	// sqlStatsIngester provides the interface to consume stats about a sql execution.
+	sqlStatsIngester *sslocal.SQLConcurrentBufferIngester
+
 	// schemaTelemetryController is the control-plane interface for schema
 	// telemetry.
 	schemaTelemetryController *schematelemetrycontroller.Controller
@@ -439,12 +442,6 @@ func NewServer(cfg *ExecutorConfig, pool *mon.BytesMonitor) *Server {
 	metrics := makeMetrics(false /* internal */, &cfg.Settings.SV)
 	serverMetrics := makeServerMetrics(cfg)
 	insightsProvider := insights.New(cfg.Settings, serverMetrics.InsightsMetrics, cfg.InsightsTestingKnobs)
-	// TODO(117690): Unify StmtStatsEnable and TxnStatsEnable into a single cluster setting.
-	sqlstats.TxnStatsEnable.SetOnChange(&cfg.Settings.SV, func(_ context.Context) {
-		if !sqlstats.TxnStatsEnable.Get(&cfg.Settings.SV) {
-			insightsProvider.Writer().Clear()
-		}
-	})
 	reportedSQLStats := sslocal.New(
 		cfg.Settings,
 		sqlstats.MaxMemReportedSQLStatsStmtFingerprints,
@@ -466,6 +463,13 @@ func NewServer(cfg *ExecutorConfig, pool *mon.BytesMonitor) *Server {
 		reportedSQLStats,
 		cfg.SQLStatsTestingKnobs,
 	)
+	sqlStatsIngester := sslocal.NewSQLConcurrentBufferIngester(insightsProvider)
+	// TODO(117690): Unify StmtStatsEnable and TxnStatsEnable into a single cluster setting.
+	sqlstats.TxnStatsEnable.SetOnChange(&cfg.Settings.SV, func(_ context.Context) {
+		if !sqlstats.TxnStatsEnable.Get(&cfg.Settings.SV) {
+			sqlStatsIngester.Clear()
+		}
+	})
 	s := &Server{
 		cfg:                     cfg,
 		Metrics:                 metrics,
@@ -474,6 +478,7 @@ func NewServer(cfg *ExecutorConfig, pool *mon.BytesMonitor) *Server {
 		pool:                    pool,
 		localSqlStats:           memSQLStats,
 		reportedStats:           reportedSQLStats,
+		sqlStatsIngester:        sqlStatsIngester,
 		reportedStatsController: reportedSQLStatsController,
 		insights:                insightsProvider,
 		reCache:                 tree.NewRegexpCache(512),
@@ -657,7 +662,7 @@ func (s *Server) Start(ctx context.Context, stopper *stop.Stopper) {
 	// should be accounted for in their costs.
 	ctx = multitenant.WithTenantCostControlExemption(ctx)
 
-	s.insights.Start(ctx, stopper)
+	s.sqlStatsIngester.Start(ctx, stopper)
 	s.sqlStats.Start(ctx, stopper)
 
 	s.schemaTelemetryController.Start(ctx, stopper)
@@ -1211,14 +1216,10 @@ func (s *Server) newConnExecutor(
 	ex.applicationStats = applicationStats
 	// We ignore statements and transactions run by the internal executor by
 	// passing a nil writer.
-	var writer *insights.ConcurrentBufferIngester
-	if !ex.sessionData().Internal {
-		writer = ex.server.insights.Writer()
-	}
 	ex.statsCollector = sslocal.NewStatsCollector(
 		s.cfg.Settings,
 		applicationStats,
-		writer,
+		s.sqlStatsIngester,
 		ex.phaseTimes,
 		s.localSqlStats.GetCounters(),
 		underOuterTxn,

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

@@ -7,7 +7,6 @@ go_library(
     srcs = [
         "causes.go",
         "detector.go",
-        "ingester.go",
         "insights.go",
         "pool.go",
         "provider.go",
@@ -24,14 +23,12 @@ go_library(
         "//pkg/settings/cluster",
         "//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/util/quantile",
-        "//pkg/util/stop",
         "//pkg/util/syncutil",
         "//pkg/util/uuid",
         "@com_github_cockroachdb_redact//:redact",
@@ -44,7 +41,6 @@ go_test(
     srcs = [
         "causes_test.go",
         "detector_test.go",
-        "ingester_test.go",
         "insights_test.go",
         "registry_test.go",
         "store_test.go",
@@ -59,8 +55,6 @@ go_test(
         "//pkg/sql/pgwire/pgcode",
         "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/sqlstats",
-        "//pkg/util/leaktest",
-        "//pkg/util/log",
         "//pkg/util/stop",
         "//pkg/util/uint128",
         "//pkg/util/uuid",