sqlstats: remove move persistedsqlstats controller

Let's remove this controller struct and just use the PersistedSQLStats provider
directly for clarity.

Epic: none
Release note: None

Author: xinhaoz

pkg/ccl/serverccl/diagnosticsccl/reporter_test.go

@@ -68,7 +68,7 @@ func TestTenantReport(t *testing.T) {
 	setupCluster(t, tenantDB)
 
 	// Clear the SQL stat pool before getting diagnostics.
-	rt.server.SQLServer().(*sql.Server).GetSQLStatsController().ResetLocalSQLStats(ctx)
+	require.NoError(t, rt.server.SQLServer().(*sql.Server).GetLocalSQLStatsProvider().Reset(ctx))
 	reporter.ReportDiagnostics(ctx)
 
 	require.Equal(t, 1, rt.diagServer.NumRequests())
@@ -171,7 +171,7 @@ func TestServerReport(t *testing.T) {
 
 		node := rt.server.MetricsRecorder().GenerateNodeStatus(ctx)
 		// Clear the SQL stat pool before getting diagnostics.
-		rt.server.SQLServer().(*sql.Server).GetSQLStatsController().ResetLocalSQLStats(ctx)
+		require.NoError(t, rt.server.SQLServer().(*sql.Server).GetLocalSQLStatsProvider().Reset(ctx))
 		rt.server.DiagnosticsReporter().(*diagnostics.Reporter).ReportDiagnostics(ctx)
 
 		keyCounts := make(map[roachpb.StoreID]int64)
@@ -495,7 +495,7 @@ func TestUsageQuantization(t *testing.T) {
 	ts := s.ApplicationLayer()
 
 	// Flush the SQL stat pool.
-	ts.SQLServer().(*sql.Server).GetSQLStatsController().ResetLocalSQLStats(ctx)
+	require.NoError(t, ts.SQLServer().(*sql.Server).GetLocalSQLStatsProvider().Reset(ctx))
 
 	// Collect a round of statistics.
 	ts.DiagnosticsReporter().(*diagnostics.Reporter).ReportDiagnostics(ctx)

pkg/server/application_api/stats_test.go

@@ -63,7 +63,7 @@ CREATE TABLE t.test (x INT PRIMARY KEY);
 	sqlServer := s.SQLServer().(*sql.Server)
 
 	// Flush stats at the beginning of the test.
-	sqlServer.GetSQLStatsController().ResetLocalSQLStats(ctx)
+	require.NoError(t, sqlServer.GetLocalSQLStatsProvider().Reset(ctx))
 	sqlServer.GetReportedSQLStatsController().ResetLocalSQLStats(ctx)
 
 	// Run some queries mixed with diagnostics, and ensure that the statistics
@@ -213,7 +213,7 @@ func TestSQLStatCollection(t *testing.T) {
 
 	// Reset the SQL statistics, which will dump stats into the
 	// reported statistics pool.
-	sqlServer.GetSQLStatsController().ResetLocalSQLStats(ctx)
+	require.NoError(t, sqlServer.GetLocalSQLStatsProvider().Reset(ctx))
 
 	// Query the reported statistics.
 	stats, err = sqlServer.GetScrubbedReportingStats(ctx, 1000, true)
@@ -271,7 +271,7 @@ func TestSQLStatCollection(t *testing.T) {
 	}
 
 	// Flush the SQL stats again.
-	sqlServer.GetSQLStatsController().ResetLocalSQLStats(ctx)
+	require.NoError(t, sqlServer.GetLocalSQLStatsProvider().Reset(ctx))
 
 	// Find our statement stat from the reported stats pool.
 	stats, err = sqlServer.GetScrubbedReportingStats(ctx, 1000, true)
@@ -396,7 +396,7 @@ func TestScrubbedReportingStatsLimit(t *testing.T) {
 	sqlRunner := sqlutils.MakeSQLRunner(sqlDB)
 	sqlServer := srv.ApplicationLayer().SQLServer().(*sql.Server)
 	// Flush stats at the beginning of the test.
-	sqlServer.GetSQLStatsController().ResetLocalSQLStats(ctx)
+	require.NoError(t, sqlServer.GetLocalSQLStatsProvider().Reset(ctx))
 	sqlServer.GetReportedSQLStatsController().ResetLocalSQLStats(ctx)
 
 	hashedAppName := "hashed app name"
@@ -409,13 +409,13 @@ func TestScrubbedReportingStatsLimit(t *testing.T) {
 	sqlRunner.Exec(t, `DELETE FROM t.test WHERE x=5`)
 
 	// verify that with low limit, number of stats is within that limit
-	sqlServer.GetSQLStatsController().ResetLocalSQLStats(ctx)
+	require.NoError(t, sqlServer.GetLocalSQLStatsProvider().Reset(ctx))
 	stats, err := sqlServer.GetScrubbedReportingStats(ctx, 5, true)
 	require.NoError(t, err)
 	require.LessOrEqual(t, len(stats), 5)
 
 	// verify that with high limit, the number of	queries is as much as the above
-	sqlServer.GetSQLStatsController().ResetLocalSQLStats(ctx)
+	require.NoError(t, sqlServer.GetLocalSQLStatsProvider().Reset(ctx))
 	stats, err = sqlServer.GetScrubbedReportingStats(ctx, 1000, true)
 	require.NoError(t, err)
 	require.GreaterOrEqual(t, len(stats), 7)

pkg/server/server_sql.go

@@ -1146,7 +1146,7 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*SQLServer, error) {
 		execCfg,
 	)
 
-	distSQLServer.ServerConfig.SQLStatsController = pgServer.SQLServer.GetSQLStatsController()
+	distSQLServer.ServerConfig.SQLStatsController = pgServer.SQLServer.GetSQLStatsProvider()
 	distSQLServer.ServerConfig.SchemaTelemetryController = pgServer.SQLServer.GetSchemaTelemetryController()
 	distSQLServer.ServerConfig.IndexUsageStatsController = pgServer.SQLServer.GetIndexUsageStatsController()
 

pkg/server/sql_stats.go

@@ -89,12 +89,13 @@ func (s *statusServer) ResetSQLStats(
 	}
 
 	response := &serverpb.ResetSQLStatsResponse{}
-	controller := s.sqlServer.pgServer.SQLServer.GetSQLStatsController()
+	localSQLStats := s.sqlServer.pgServer.SQLServer.GetLocalSQLStatsProvider()
+	persistedSQLStats := s.sqlServer.pgServer.SQLServer.GetSQLStatsProvider()
 
-	// If we need to reset persisted stats, we delegate to SQLStatsController,
+	// If we need to reset persisted stats, we delegate to persisted sql stats,
 	// which will trigger a system table truncation and RPC fanout under the hood.
 	if req.ResetPersistedStats {
-		if err := controller.ResetClusterSQLStats(ctx); err != nil {
+		if err := persistedSQLStats.ResetClusterSQLStats(ctx); err != nil {
 			return nil, err
 		}
 
@@ -113,8 +114,8 @@ func (s *statusServer) ResetSQLStats(
 			return nil, status.Error(codes.InvalidArgument, err.Error())
 		}
 		if local {
-			controller.ResetLocalSQLStats(ctx)
-			return response, nil
+			err := localSQLStats.Reset(ctx)
+			return response, err
 		}
 		status, err := s.dialNode(ctx, requestedNodeID)
 		if err != nil {

pkg/sql/conn_executor.go

@@ -340,9 +340,6 @@ type Server struct {
 	// node. Newly collected statistics flow into localSqlStats.
 	localSqlStats *sslocal.SQLStats
 
-	// sqlStatsController is the control-plane interface for sqlStats.
-	sqlStatsController *persistedsqlstats.Controller
-
 	// sqlStatsIngester provides the interface to consume stats about a sql execution.
 	sqlStatsIngester *sslocal.SQLStatsIngester
 
@@ -507,6 +504,7 @@ func NewServer(cfg *ExecutorConfig, pool *mon.BytesMonitor) *Server {
 		ClusterID:               s.cfg.NodeInfo.LogicalClusterID,
 		SQLIDContainer:          cfg.NodeInfo.NodeID,
 		JobRegistry:             s.cfg.JobRegistry,
+		FanoutServer:            cfg.SQLStatusServer,
 		Knobs:                   cfg.SQLStatsTestingKnobs,
 		FlushesSuccessful:       serverMetrics.StatsMetrics.SQLStatsFlushesSuccessful,
 		FlushDoneSignalsIgnored: serverMetrics.StatsMetrics.SQLStatsFlushDoneSignalsIgnored,
@@ -516,7 +514,6 @@ func NewServer(cfg *ExecutorConfig, pool *mon.BytesMonitor) *Server {
 	}, memSQLStats)
 
 	s.sqlStats = persistedSQLStats
-	s.sqlStatsController = persistedSQLStats.GetController(cfg.SQLStatusServer)
 	schemaTelemetryIEMonitor := MakeInternalExecutorMemMonitor(MemoryMetrics{}, s.GetExecutorConfig().Settings)
 	schemaTelemetryIEMonitor.StartNoReserved(context.Background(), s.GetBytesMonitor())
 	s.schemaTelemetryController = schematelemetrycontroller.NewController(
@@ -700,12 +697,6 @@ func (s *Server) Start(ctx context.Context, stopper *stop.Stopper) {
 	s.txnIDCache.Start(ctx, stopper)
 }
 
-// GetSQLStatsController returns the persistedsqlstats.Controller for current
-// sql.Server's SQL Stats.
-func (s *Server) GetSQLStatsController() *persistedsqlstats.Controller {
-	return s.sqlStatsController
-}
-
 // GetSchemaTelemetryController returns the schematelemetryschedule.Controller
 // for current sql.Server's schema telemetry.
 func (s *Server) GetSchemaTelemetryController() *schematelemetrycontroller.Controller {
@@ -3840,7 +3831,7 @@ func (ex *connExecutor) initEvalCtx(ctx context.Context, evalCtx *extendedEvalCo
 			SessionDataStack:               ex.sessionDataStack,
 			ReCache:                        ex.server.reCache,
 			ToCharFormatCache:              ex.server.toCharFormatCache,
-			SQLStatsController:             ex.server.sqlStatsController,
+			SQLStatsController:             ex.server.sqlStats,
 			SchemaTelemetryController:      ex.server.schemaTelemetryController,
 			IndexUsageStatsController:      ex.server.indexUsageStatsController,
 			ConsistencyChecker:             p.execCfg.ConsistencyChecker,

pkg/sql/planner.go

@@ -497,17 +497,14 @@ func internalExtendedEvalCtx(
 	evalContextTestingKnobs := execCfg.EvalContextTestingKnobs
 
 	var indexUsageStats *idxusage.LocalIndexUsageStats
-	var sqlStatsController eval.SQLStatsController
 	var schemaTelemetryController eval.SchemaTelemetryController
 	var indexUsageStatsController eval.IndexUsageStatsController
 	var sqlStatsProvider *persistedsqlstats.PersistedSQLStats
 	var localSqlStatsProvider *sslocal.SQLStats
 	if ief := execCfg.InternalDB; ief != nil {
 		if ief.server != nil {
 			indexUsageStats = ief.server.indexUsageStats
-			sqlStatsController = ief.server.sqlStatsController
 			schemaTelemetryController = ief.server.schemaTelemetryController
-			indexUsageStatsController = ief.server.indexUsageStatsController
 			sqlStatsProvider = ief.server.sqlStats
 			localSqlStatsProvider = ief.server.localSqlStats
 		} else {
@@ -517,7 +514,6 @@ func internalExtendedEvalCtx(
 			indexUsageStats = idxusage.NewLocalIndexUsageStats(&idxusage.Config{
 				Setting: execCfg.Settings,
 			})
-			sqlStatsController = &persistedsqlstats.Controller{}
 			schemaTelemetryController = &schematelemetrycontroller.Controller{}
 			indexUsageStatsController = &idxusage.Controller{}
 			sqlStatsProvider = &persistedsqlstats.PersistedSQLStats{}
@@ -534,7 +530,7 @@ func internalExtendedEvalCtx(
 			TestingKnobs:                   evalContextTestingKnobs,
 			StmtTimestamp:                  stmtTimestamp,
 			TxnTimestamp:                   txnTimestamp,
-			SQLStatsController:             sqlStatsController,
+			SQLStatsController:             sqlStatsProvider,
 			SchemaTelemetryController:      schemaTelemetryController,
 			IndexUsageStatsController:      indexUsageStatsController,
 			ConsistencyChecker:             execCfg.ConsistencyChecker,

pkg/sql/sqlstats/persistedsqlstats/BUILD.bazel

@@ -8,7 +8,6 @@ go_library(
         "cluster_settings.go",
         "compaction_exec.go",
         "compaction_scheduling.go",
-        "controller.go",
         "flush.go",
         "provider.go",
         "scheduled_job_monitor.go",

pkg/sql/sqlstats/persistedsqlstats/bench_test.go

@@ -385,7 +385,6 @@ func BenchmarkSqlStatsMaxFlushTime(b *testing.B) {
 
 	sqlStats := s.SQLServer().(*sql.Server).GetLocalSQLStatsProvider()
 	pss := s.SQLServer().(*sql.Server).GetSQLStatsProvider()
-	controller := s.SQLServer().(*sql.Server).GetSQLStatsController()
 	stmtFingerprintLimit := sqlstats.MaxMemSQLStatsStmtFingerprints.Get(&s.ClusterSettings().SV)
 	txnFingerprintLimit := sqlstats.MaxMemSQLStatsTxnFingerprints.Get(&s.ClusterSettings().SV)
 
@@ -435,7 +434,7 @@ func BenchmarkSqlStatsMaxFlushTime(b *testing.B) {
 	b.Run(fmt.Sprintf("single-application/writes=insert/%d-fingerprints", totalFingerprints), func(b *testing.B) {
 		b.StopTimer()
 		for i := 0; i < b.N; i++ {
-			require.NoError(b, controller.ResetClusterSQLStats(ctx))
+			require.NoError(b, pss.ResetClusterSQLStats(ctx))
 			fillBenchAppMemStats()
 			b.StartTimer()
 			require.True(b, pss.MaybeFlush(ctx, s.AppStopper()))

pkg/sql/sqlstats/persistedsqlstats/controller.go

@@ -87,8 +87,7 @@ func TestPersistedSQLStatsReset(t *testing.T) {
 	checkInsertedTxnStats(t, appName, observer, expectedStmtFingerprintToFingerprintID)
 
 	// Resets cluster wide SQL stats.
-	sqlStatsController := server.SQLServer().(*sql.Server).GetSQLStatsController()
-	require.NoError(t, sqlStatsController.ResetClusterSQLStats(ctx))
+	require.NoError(t, sqlStats.ResetClusterSQLStats(ctx))
 
 	var count int
 	observer.QueryRow(t,