sql: implement persistedsqlstats flush logic

This commit implements the initial flush logic of the
persisted sql stats subsystem.

Release note: None

docs/generated/settings/settings-for-tenants.txt

@@ -129,6 +129,7 @@ sql.spatial.experimental_box2d_comparison_operators.enabled	boolean	false	enable
 sql.stats.automatic_collection.enabled	boolean	true	automatic statistics collection mode
 sql.stats.automatic_collection.fraction_stale_rows	float	0.2	target fraction of stale rows per table that will trigger a statistics refresh
 sql.stats.automatic_collection.min_stale_rows	integer	500	target minimum number of stale rows per table that will trigger a statistics refresh
+sql.stats.flush.interval	duration	1h0m0s	the interval at which SQL execution statistics are flushed to disk
 sql.stats.histogram_collection.enabled	boolean	true	histogram collection mode
 sql.stats.multi_column_collection.enabled	boolean	true	multi-column statistics collection mode
 sql.stats.post_events.enabled	boolean	false	if set, an event is logged for every CREATE STATISTICS job

docs/generated/settings/settings.html

@@ -133,6 +133,7 @@
 <tr><td><code>sql.stats.automatic_collection.enabled</code></td><td>boolean</td><td><code>true</code></td><td>automatic statistics collection mode</td></tr>
 <tr><td><code>sql.stats.automatic_collection.fraction_stale_rows</code></td><td>float</td><td><code>0.2</code></td><td>target fraction of stale rows per table that will trigger a statistics refresh</td></tr>
 <tr><td><code>sql.stats.automatic_collection.min_stale_rows</code></td><td>integer</td><td><code>500</code></td><td>target minimum number of stale rows per table that will trigger a statistics refresh</td></tr>
+<tr><td><code>sql.stats.flush.interval</code></td><td>duration</td><td><code>1h0m0s</code></td><td>the interval at which SQL execution statistics are flushed to disk</td></tr>
 <tr><td><code>sql.stats.histogram_collection.enabled</code></td><td>boolean</td><td><code>true</code></td><td>histogram collection mode</td></tr>
 <tr><td><code>sql.stats.multi_column_collection.enabled</code></td><td>boolean</td><td><code>true</code></td><td>multi-column statistics collection mode</td></tr>
 <tr><td><code>sql.stats.post_events.enabled</code></td><td>boolean</td><td><code>false</code></td><td>if set, an event is logged for every CREATE STATISTICS job</td></tr>

pkg/BUILD.bazel

@@ -295,6 +295,7 @@ ALL_TESTS = [
     "//pkg/sql/sqlliveness/slinstance:slinstance_test",
     "//pkg/sql/sqlliveness/slstorage:slstorage_test",
     "//pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil:sqlstatsutil_test",
+    "//pkg/sql/sqlstats/persistedsqlstats:persistedsqlstats_test",
     "//pkg/sql/stats:stats_test",
     "//pkg/sql/stmtdiagnostics:stmtdiagnostics_test",
     "//pkg/sql/tests:tests_test",

pkg/base/testing_knobs.go

@@ -39,4 +39,5 @@ type TestingKnobs struct {
 	BackupRestore           ModuleTestingKnobs
 	MigrationManager        ModuleTestingKnobs
 	IndexUsageStatsKnobs    ModuleTestingKnobs
+	SQLStatsKnobs           ModuleTestingKnobs
 }

pkg/server/BUILD.bazel

@@ -145,6 +145,7 @@ go_library(
         "//pkg/sql/sqlinstance/instanceprovider",
         "//pkg/sql/sqlliveness",
         "//pkg/sql/sqlliveness/slprovider",
+        "//pkg/sql/sqlstats/persistedsqlstats",
         "//pkg/sql/sqlutil",
         "//pkg/sql/stats",
         "//pkg/sql/stmtdiagnostics",

pkg/server/server_sql.go

@@ -77,6 +77,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlinstance/instanceprovider"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness/slprovider"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats/persistedsqlstats"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlutil"
 	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/sql/stmtdiagnostics"
@@ -548,6 +549,7 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*SQLServer, error) {
 	if cfg.TestingKnobs.JobsTestingKnobs != nil {
 		distSQLCfg.TestingKnobs.JobsTestingKnobs = cfg.TestingKnobs.JobsTestingKnobs
 	}
+
 	distSQLServer := distsql.NewServer(ctx, distSQLCfg, cfg.flowScheduler)
 	execinfrapb.RegisterDistSQLServer(cfg.grpcServer, distSQLServer)
 
@@ -708,6 +710,9 @@ func newSQLServer(ctx context.Context, cfg sqlServerArgs) (*SQLServer, error) {
 	if indexUsageStatsKnobs := cfg.TestingKnobs.IndexUsageStatsKnobs; indexUsageStatsKnobs != nil {
 		execCfg.IndexUsageStatsTestingKnobs = indexUsageStatsKnobs.(*idxusage.TestingKnobs)
 	}
+	if sqlStatsKnobs := cfg.TestingKnobs.SQLStatsKnobs; sqlStatsKnobs != nil {
+		execCfg.SQLStatsTestingKnobs = sqlStatsKnobs.(*persistedsqlstats.TestingKnobs)
+	}
 
 	statsRefresher := stats.MakeRefresher(
 		cfg.Settings,

pkg/sql/BUILD.bazel

@@ -354,6 +354,7 @@ go_library(
         "//pkg/sql/sqlfsm",
         "//pkg/sql/sqlliveness",
         "//pkg/sql/sqlstats",
+        "//pkg/sql/sqlstats/persistedsqlstats",
         "//pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil",
         "//pkg/sql/sqlstats/sslocal",
         "//pkg/sql/sqltelemetry",

pkg/sql/conn_executor.go

@@ -49,6 +49,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sessionphase"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlerrors"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats/persistedsqlstats"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats/sslocal"
 	"github.com/cockroachdb/cockroach/pkg/sql/stmtdiagnostics"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
@@ -321,7 +322,7 @@ func NewServer(cfg *ExecutorConfig, pool *mon.BytesMonitor) *Server {
 		nil, /* reportedProvider */
 	)
 	reportedSQLStatsController := reportedSQLStats.GetController(cfg.SQLStatusServer)
-	sqlStats := sslocal.New(
+	memSQLStats := sslocal.New(
 		cfg.Settings,
 		sqlstats.MaxMemSQLStatsStmtFingerprints,
 		sqlstats.MaxMemSQLStatsTxnFingerprints,
@@ -331,14 +332,11 @@ func NewServer(cfg *ExecutorConfig, pool *mon.BytesMonitor) *Server {
 		sqlstats.SQLStatReset,
 		reportedSQLStats,
 	)
-	sqlStatsController := sqlStats.GetController(cfg.SQLStatusServer)
 	s := &Server{
 		cfg:                     cfg,
 		Metrics:                 metrics,
 		InternalMetrics:         makeMetrics(cfg, true /* internal */),
 		pool:                    pool,
-		sqlStats:                sqlStats,
-		sqlStatsController:      sqlStatsController,
 		reportedStats:           reportedSQLStats,
 		reportedStatsController: reportedSQLStatsController,
 		reCache:                 tree.NewRegexpCache(512),
@@ -349,6 +347,20 @@ func NewServer(cfg *ExecutorConfig, pool *mon.BytesMonitor) *Server {
 		}),
 	}
 
+	sqlStatsInternalExecutor := MakeInternalExecutor(context.Background(), s, MemoryMetrics{}, cfg.Settings)
+	persistedSQLStats := persistedsqlstats.New(&persistedsqlstats.Config{
+		Settings:         s.cfg.Settings,
+		InternalExecutor: &sqlStatsInternalExecutor,
+		KvDB:             cfg.DB,
+		SQLIDContainer:   cfg.NodeID,
+		Knobs:            cfg.SQLStatsTestingKnobs,
+		FlushCounter:     metrics.StatsMetrics.SQLStatsFlushStarted,
+		FailureCounter:   metrics.StatsMetrics.SQLStatsFlushFailure,
+		FlushDuration:    metrics.StatsMetrics.SQLStatsFlushDuration,
+	}, memSQLStats)
+
+	s.sqlStats = persistedSQLStats
+	s.sqlStatsController = persistedSQLStats.GetController(cfg.SQLStatusServer)
 	return s
 }
 
@@ -395,7 +407,12 @@ func makeMetrics(cfg *ExecutorConfig, internal bool) Metrics {
 			),
 			ReportedSQLStatsMemoryCurBytesCount: metric.NewGauge(
 				getMetricMeta(MetaReportedSQLStatsMemCurBytes, internal)),
-			DiscardedStatsCount: metric.NewCounter(getMetricMeta(MetaDiscardedSQLStats, internal)),
+			DiscardedStatsCount:  metric.NewCounter(getMetricMeta(MetaDiscardedSQLStats, internal)),
+			SQLStatsFlushStarted: metric.NewCounter(getMetricMeta(MetaSQLStatsFlushStarted, internal)),
+			SQLStatsFlushFailure: metric.NewCounter(getMetricMeta(MetaSQLStatsFlushFailure, internal)),
+			SQLStatsFlushDuration: metric.NewLatency(
+				getMetricMeta(MetaSQLStatsFlushDuration, internal), 6*metricsSampleInterval,
+			),
 		},
 	}
 }
@@ -418,6 +435,11 @@ func (s *Server) GetSQLStatsController() *sslocal.Controller {
 	return s.sqlStatsController
 }
 
+// GetSQLStatsProvider returns the provider for the sqlstats subsystem.
+func (s *Server) GetSQLStatsProvider() sqlstats.Provider {
+	return s.sqlStats
+}
+
 // GetReportedSQLStatsController returns the sqlstats.Controller for the current
 // sql.Server's reported SQL Stats.
 func (s *Server) GetReportedSQLStatsController() *sslocal.Controller {
@@ -442,7 +464,7 @@ func (s *Server) GetUnscrubbedStmtStats(
 	ctx context.Context,
 ) ([]roachpb.CollectedStatementStatistics, error) {
 	var stmtStats []roachpb.CollectedStatementStatistics
-	stmtStatsVisitor := func(stat *roachpb.CollectedStatementStatistics) error {
+	stmtStatsVisitor := func(_ context.Context, stat *roachpb.CollectedStatementStatistics) error {
 		stmtStats = append(stmtStats, *stat)
 		return nil
 	}
@@ -462,7 +484,7 @@ func (s *Server) GetUnscrubbedTxnStats(
 	ctx context.Context,
 ) ([]roachpb.CollectedTransactionStatistics, error) {
 	var txnStats []roachpb.CollectedTransactionStatistics
-	txnStatsVisitor := func(_ roachpb.TransactionFingerprintID, stat *roachpb.CollectedTransactionStatistics) error {
+	txnStatsVisitor := func(_ context.Context, _ roachpb.TransactionFingerprintID, stat *roachpb.CollectedTransactionStatistics) error {
 		txnStats = append(txnStats, *stat)
 		return nil
 	}
@@ -490,7 +512,7 @@ func (s *Server) getScrubbedStmtStats(
 	salt := ClusterSecret.Get(&s.cfg.Settings.SV)
 
 	var scrubbedStats []roachpb.CollectedStatementStatistics
-	stmtStatsVisitor := func(stat *roachpb.CollectedStatementStatistics) error {
+	stmtStatsVisitor := func(_ context.Context, stat *roachpb.CollectedStatementStatistics) error {
 		// Scrub the statement itself.
 		scrubbedQueryStr, ok := scrubStmtStatKey(s.cfg.VirtualSchemas, stat.Key.Query)
 

pkg/sql/crdb_internal.go

@@ -916,7 +916,7 @@ CREATE TABLE crdb_internal.node_statement_statistics (
 
 		nodeID, _ := p.execCfg.NodeID.OptionalNodeID() // zero if not available
 
-		statementVisitor := func(stats *roachpb.CollectedStatementStatistics) error {
+		statementVisitor := func(_ context.Context, stats *roachpb.CollectedStatementStatistics) error {
 			anonymized := tree.DNull
 			anonStr, ok := scrubStmtStatKey(p.getVirtualTabler(), stats.Key.Query)
 			if ok {
@@ -1051,7 +1051,7 @@ CREATE TABLE crdb_internal.node_transaction_statistics (
 
 		nodeID, _ := p.execCfg.NodeID.OptionalNodeID() // zero if not available
 
-		transactionVisitor := func(txnFingerprintID roachpb.TransactionFingerprintID, stats *roachpb.CollectedTransactionStatistics) error {
+		transactionVisitor := func(_ context.Context, txnFingerprintID roachpb.TransactionFingerprintID, stats *roachpb.CollectedTransactionStatistics) error {
 			stmtFingerprintIDsDatum := tree.NewDArray(types.String)
 			for _, stmtFingerprintID := range stats.StatementFingerprintIDs {
 				if err := stmtFingerprintIDsDatum.Append(tree.NewDString(strconv.FormatUint(uint64(stmtFingerprintID), 10))); err != nil {

pkg/sql/exec_util.go

@@ -77,6 +77,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondatapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlliveness"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats/persistedsqlstats"
 	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/sql/stmtdiagnostics"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
@@ -855,6 +856,24 @@ var (
 		Measurement: "Discarded SQL Stats",
 		Unit:        metric.Unit_COUNT,
 	}
+	MetaSQLStatsFlushStarted = metric.Metadata{
+		Name:        "sql.stats.flush.count",
+		Help:        "Number of times SQL Stats are flushed to persistent storage",
+		Measurement: "SQL Stats Flush",
+		Unit:        metric.Unit_COUNT,
+	}
+	MetaSQLStatsFlushFailure = metric.Metadata{
+		Name:        "sql.stats.flush.error",
+		Help:        "Number of errors encountered when flushing SQL Stats",
+		Measurement: "SQL Stats Flush",
+		Unit:        metric.Unit_COUNT,
+	}
+	MetaSQLStatsFlushDuration = metric.Metadata{
+		Name:        "sql.stats.flush.duration",
+		Help:        "Time took to in nanoseconds to complete SQL Stats flush",
+		Measurement: "SQL Stats Flush",
+		Unit:        metric.Unit_NANOSECONDS,
+	}
 )
 
 func getMetricMeta(meta metric.Metadata, internal bool) metric.Metadata {
@@ -932,6 +951,7 @@ type ExecutorConfig struct {
 	TenantTestingKnobs            *TenantTestingKnobs
 	BackupRestoreTestingKnobs     *BackupRestoreTestingKnobs
 	IndexUsageStatsTestingKnobs   *idxusage.TestingKnobs
+	SQLStatsTestingKnobs          *persistedsqlstats.TestingKnobs
 	// HistogramWindowInterval is (server.Config).HistogramWindowInterval.
 	HistogramWindowInterval time.Duration
 

pkg/sql/executor_statement_metrics.go

@@ -65,6 +65,10 @@ type StatsMetrics struct {
 	ReportedSQLStatsMemoryCurBytesCount *metric.Gauge
 
 	DiscardedStatsCount *metric.Counter
+
+	SQLStatsFlushStarted  *metric.Counter
+	SQLStatsFlushFailure  *metric.Counter
+	SQLStatsFlushDuration *metric.Histogram
 }
 
 // StatsMetrics is part of the metric.Struct interface.

pkg/sql/logictest/testdata/logic_test/statement_statistics

@@ -1,11 +1,5 @@
 # LogicTest: local !3node-tenant(52763)
 
-# Disable SQL Stats flush to prevents stats from being cleared from the
-# in-memory store.
-
-statement ok
-SET CLUSTER SETTING sql.stats.flush.enabled = false;
-
 # Check that node_statement_statistics report per application
 
 statement ok

pkg/sql/sqlstats/persistedsqlstats/BUILD.bazel

@@ -0,0 +1,59 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "persistedsqlstats",
+    srcs = [
+        "cluster_settings.go",
+        "flush.go",
+        "provider.go",
+        "test_utils.go",
+    ],
+    importpath = "github.com/cockroachdb/cockroach/pkg/sql/sqlstats/persistedsqlstats",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/base",
+        "//pkg/kv",
+        "//pkg/roachpb:with-mocks",
+        "//pkg/security",
+        "//pkg/settings",
+        "//pkg/settings/cluster",
+        "//pkg/sql/sem/tree",
+        "//pkg/sql/sessiondata",
+        "//pkg/sql/sqlstats",
+        "//pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil",
+        "//pkg/sql/sqlstats/sslocal",
+        "//pkg/sql/sqlutil",
+        "//pkg/util/log",
+        "//pkg/util/metric",
+        "//pkg/util/stop",
+        "//pkg/util/timeutil",
+        "@com_github_cockroachdb_errors//:errors",
+    ],
+)
+
+go_test(
+    name = "persistedsqlstats_test",
+    srcs = [
+        "flush_test.go",
+        "main_test.go",
+    ],
+    deps = [
+        ":persistedsqlstats",
+        "//pkg/base",
+        "//pkg/roachpb:with-mocks",
+        "//pkg/security",
+        "//pkg/security/securitytest",
+        "//pkg/server",
+        "//pkg/sql",
+        "//pkg/sql/sqlstats",
+        "//pkg/testutils/serverutils",
+        "//pkg/testutils/sqlutils",
+        "//pkg/testutils/testcluster",
+        "//pkg/util/leaktest",
+        "//pkg/util/log",
+        "//pkg/util/stop",
+        "//pkg/util/syncutil",
+        "//pkg/util/timeutil",
+        "@com_github_stretchr_testify//require",
+    ],
+)

pkg/sql/sqlstats/persistedsqlstats/cluster_settings.go

@@ -0,0 +1,26 @@
+// Copyright 2021 The Cockroach Authors.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package persistedsqlstats
+
+import (
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/settings"
+)
+
+// SQLStatsFlushInterval is the cluster setting that controls how often the SQL
+// stats are flushed to system table.
+var SQLStatsFlushInterval = settings.RegisterDurationSetting(
+	"sql.stats.flush.interval",
+	"the interval at which SQL execution statistics are flushed to disk",
+	time.Hour,
+	settings.NonNegativeDurationWithMaximum(time.Hour*24),
+).WithPublic()