Merge #152112 #154525

152112: sql/schemachanger: improve performance of TestBackupSuccess r=fqazi a=fqazi

Previously, the successful backup tests would for each stage of a schema
change would run the full schema change, taking a backup at the target
stage. This patch modifies the TestBackupSuccess variants of the schema
changer test to start the server once, take all the backups and then
restore the stages we should be testing.

Testing show around a ~30% improvement with this change in execution
time.

Fixes: #152076

Release note: None


154525: sql/inspect: add job metrics for INSPECT jobs r=spilchen a=spilchen

Implements three metrics for INSPECT jobs:
- jobs.inspect.runs: Total number of INSPECT job executions
- jobs.inspect.runs_with_issues: Count of jobs that found inconsistencies
- jobs.inspect.issues_found: Aggregate count of issues found across all processors

Closes #153884

Epic: CRDB-30356
Release note: none

Co-authored-by: Faizan Qazi <[email protected]>
Co-authored-by: Matt Spilchen <[email protected]>

Author: 3 people

docs/generated/metrics/metrics.yaml

@@ -5290,6 +5290,14 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.issues_found
+      exported_name: jobs_inspect_issues_found
+      description: Total count of issues found by INSPECT jobs
+      y_axis_label: Issues
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: jobs.inspect.protected_age_sec
       exported_name: jobs_inspect_protected_age_sec
       labeled_name: 'jobs.protected_age_sec{type: inspect}'
@@ -5335,6 +5343,22 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.runs
+      exported_name: jobs_inspect_runs
+      description: Number of INSPECT jobs executed
+      y_axis_label: Jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: jobs.inspect.runs_with_issues
+      exported_name: jobs_inspect_runs_with_issues
+      description: Number of INSPECT jobs that found at least one issue
+      y_axis_label: Jobs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: jobs.key_visualizer.currently_idle
       exported_name: jobs_key_visualizer_currently_idle
       labeled_name: 'jobs{type: key_visualizer, status: currently_idle}'

pkg/ccl/schemachangerccl/multiregion_testcluster_factory.go

@@ -56,10 +56,8 @@ func (f MultiRegionTestClusterFactory) WithSchemaLockDisabled() sctest.TestServe
 	return f
 }
 
-// Run implements the sctest.TestServerFactory interface.
-func (f MultiRegionTestClusterFactory) Run(
-	ctx context.Context, t *testing.T, fn func(_ serverutils.TestServerInterface, _ *gosql.DB),
-) {
+// Start implements the sctest.TestServerFactory interface.
+func (f MultiRegionTestClusterFactory) Start(ctx context.Context, t *testing.T) sctest.TestServer {
 	const numServers = 3
 	knobs := base.TestingKnobs{
 		SQLEvalContext: &eval.TestingKnobs{
@@ -83,6 +81,19 @@ func (f MultiRegionTestClusterFactory) Run(
 	}
 	sql.CreateTableWithSchemaLocked.Override(ctx, &st.SV, !f.schemaLockedDisabled)
 	c, db, _ := multiregionccltestutils.TestingCreateMultiRegionCluster(t, numServers, knobs, multiregionccltestutils.WithSettings(st))
-	defer c.Stopper().Stop(ctx)
-	fn(c.Server(0), db)
+	return sctest.TestServer{
+		Server: c.Server(0),
+		DB:     db,
+		Stopper: func(t *testing.T) {
+			c.Stopper().Stop(ctx)
+		},
+	}
+}
+
+func (f MultiRegionTestClusterFactory) Run(
+	ctx context.Context, t *testing.T, fn func(s serverutils.TestServerInterface, tdb *gosql.DB),
+) {
+	s := f.Start(ctx, t)
+	defer s.Stopper(t)
+	fn(s.Server, s.DB)
 }

pkg/jobs/metrics.go

@@ -42,6 +42,7 @@ type Metrics struct {
 	Changefeed   metric.Struct
 	StreamIngest metric.Struct
 	Backup       metric.Struct
+	Inspect      metric.Struct
 
 	// AdoptIterations counts the number of adopt loops executed by Registry.
 	AdoptIterations *metric.Counter
@@ -402,6 +403,9 @@ func (m *Metrics) init(histogramWindowInterval time.Duration, lookup *cidr.Looku
 	if MakeBackupMetricsHook != nil {
 		m.Backup = MakeBackupMetricsHook(histogramWindowInterval)
 	}
+	if MakeInspectMetricsHook != nil {
+		m.Inspect = MakeInspectMetricsHook(histogramWindowInterval)
+	}
 
 	m.AdoptIterations = metric.NewCounter(metaAdoptIterations)
 	m.ClaimedJobs = metric.NewCounter(metaClaimedJobs)
@@ -482,6 +486,9 @@ var MakeRowLevelTTLMetricsHook func(time.Duration) metric.Struct
 // MakeBackupMetricsHook allows for registration of backup metrics.
 var MakeBackupMetricsHook func(time.Duration) metric.Struct
 
+// MakeInspectMetricsHook allows for registration of inspect metrics.
+var MakeInspectMetricsHook func(time.Duration) metric.Struct
+
 // JobTelemetryMetrics is a telemetry metrics for individual job types.
 type JobTelemetryMetrics struct {
 	Successful telemetry.Counter

pkg/sql/inspect/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "index_consistency_check.go",
         "inspect_job.go",
         "inspect_logger.go",
+        "inspect_metrics.go",
         "inspect_processor.go",
         "issue.go",
         "log_sink.go",
@@ -48,6 +49,7 @@ go_library(
         "//pkg/util/ctxgroup",
         "//pkg/util/hlc",
         "//pkg/util/log",
+        "//pkg/util/metric",
         "//pkg/util/protoutil",
         "//pkg/util/span",
         "//pkg/util/syncutil",
@@ -64,6 +66,7 @@ go_test(
     srcs = [
         "index_consistency_check_test.go",
         "inspect_job_test.go",
+        "inspect_metrics_test.go",
         "inspect_processor_test.go",
         "issue_test.go",
         "main_test.go",

pkg/sql/inspect/inspect_job.go

@@ -38,6 +38,8 @@ func (c *inspectResumer) Resume(ctx context.Context, execCtx interface{}) error
 	jobExecCtx := execCtx.(sql.JobExecContext)
 	execCfg := jobExecCtx.ExecCfg()
 
+	execCfg.JobRegistry.MetricsStruct().Inspect.(*InspectMetrics).Runs.Inc(1)
+
 	if err := c.maybeRunOnJobStartHook(execCfg); err != nil {
 		return err
 	}
@@ -93,6 +95,11 @@ func (c *inspectResumer) OnFailOrCancel(
 	jobExecCtx := execCtx.(sql.JobExecContext)
 	execCfg := jobExecCtx.ExecCfg()
 	c.maybeCleanupProtectedTimestamp(ctx, execCfg)
+
+	// Record RunsWithIssues metric if the job failed due to finding inconsistencies.
+	if jobErr != nil && errors.Is(jobErr, errInspectFoundInconsistencies) {
+		execCfg.JobRegistry.MetricsStruct().Inspect.(*InspectMetrics).RunsWithIssues.Inc(1)
+	}
 	return nil
 }
 

pkg/sql/inspect/inspect_logger.go

@@ -7,7 +7,9 @@ package inspect
 
 import (
 	"context"
+	"sync/atomic"
 
+	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/cockroachdb/errors"
 )
 
@@ -45,3 +47,23 @@ func (l inspectLoggers) hasIssues() bool {
 	}
 	return false
 }
+
+// metricsLogger increments metrics when issues are logged.
+type metricsLogger struct {
+	foundIssue     atomic.Bool
+	issuesFoundCtr *metric.Counter
+}
+
+var _ inspectLogger = &metricsLogger{}
+
+// logIssue implements the inspectLogger interface.
+func (m *metricsLogger) logIssue(ctx context.Context, issue *inspectIssue) error {
+	m.foundIssue.Store(true)
+	m.issuesFoundCtr.Inc(1)
+	return nil
+}
+
+// hasIssues implements the inspectLogger interface.
+func (m *metricsLogger) hasIssues() bool {
+	return m.foundIssue.Load()
+}

pkg/sql/inspect/inspect_metrics.go

@@ -0,0 +1,59 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package inspect
+
+import (
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/util/metric"
+)
+
+// InspectMetrics holds metrics for INSPECT job operations.
+type InspectMetrics struct {
+	Runs           *metric.Counter
+	RunsWithIssues *metric.Counter
+	IssuesFound    *metric.Counter
+}
+
+var _ metric.Struct = (*InspectMetrics)(nil)
+
+// MetricStruct implements the metric.Struct interface.
+func (InspectMetrics) MetricStruct() {}
+
+var (
+	metaInspectRuns = metric.Metadata{
+		Name:        "jobs.inspect.runs",
+		Help:        "Number of INSPECT jobs executed",
+		Measurement: "Jobs",
+		Unit:        metric.Unit_COUNT,
+	}
+	metaInspectRunsWithIssues = metric.Metadata{
+		Name:        "jobs.inspect.runs_with_issues",
+		Help:        "Number of INSPECT jobs that found at least one issue",
+		Measurement: "Jobs",
+		Unit:        metric.Unit_COUNT,
+	}
+	metaInspectIssuesFound = metric.Metadata{
+		Name:        "jobs.inspect.issues_found",
+		Help:        "Total count of issues found by INSPECT jobs",
+		Measurement: "Issues",
+		Unit:        metric.Unit_COUNT,
+	}
+)
+
+// MakeInspectMetrics instantiates the metrics for INSPECT jobs.
+func MakeInspectMetrics(histogramWindow time.Duration) metric.Struct {
+	return &InspectMetrics{
+		Runs:           metric.NewCounter(metaInspectRuns),
+		RunsWithIssues: metric.NewCounter(metaInspectRunsWithIssues),
+		IssuesFound:    metric.NewCounter(metaInspectIssuesFound),
+	}
+}
+
+func init() {
+	jobs.MakeInspectMetricsHook = MakeInspectMetrics
+}

pkg/sql/inspect/inspect_metrics_test.go

@@ -0,0 +1,101 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package inspect
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/desctestutils"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/stretchr/testify/require"
+)
+
+// TestInspectMetrics verifies that INSPECT job metrics are correctly updated
+// when jobs run.
+func TestInspectMetrics(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	s, db, _ := serverutils.StartServer(t, base.TestServerArgs{
+		DefaultTestTenant: base.TestIsSpecificToStorageLayerAndNeedsASystemTenant,
+	})
+	defer s.Stopper().Stop(ctx)
+
+	runner := sqlutils.MakeSQLRunner(db)
+	runner.Exec(t, `
+		CREATE DATABASE db;
+		SET enable_scrub_job = true;
+		CREATE TABLE db.t (
+			id INT PRIMARY KEY,
+			val INT
+		);
+		CREATE INDEX i1 on db.t (val);
+		INSERT INTO db.t VALUES (1, 2), (2, 3);
+	`)
+
+	execCfg := s.ApplicationLayer().ExecutorConfig().(sql.ExecutorConfig)
+	metrics := execCfg.JobRegistry.MetricsStruct().Inspect.(*InspectMetrics)
+
+	initialRuns := metrics.Runs.Count()
+	initialRunsWithIssues := metrics.RunsWithIssues.Count()
+	initialIssuesFound := metrics.IssuesFound.Count()
+
+	// First run: no corruption, should succeed without issues
+	runner.Exec(t, "EXPERIMENTAL SCRUB TABLE db.t")
+	require.Equal(t, initialRuns+1, metrics.Runs.Count(), "Runs counter should increment")
+	require.Equal(t, initialRunsWithIssues, metrics.RunsWithIssues.Count(), "RunsWithIssues should not increment")
+	require.Equal(t, initialIssuesFound, metrics.IssuesFound.Count(), "IssuesFound should not increment")
+
+	// Create corruption: delete a secondary index entry for row (1, 2)
+	// This creates a "missing_secondary_index_entry" issue - the primary key exists
+	// but the corresponding secondary index entry is missing.
+	kvDB := s.DB()
+	codec := s.Codec()
+	tableDesc := desctestutils.TestingGetPublicTableDescriptor(kvDB, codec, "db", "t")
+	secIndex := tableDesc.PublicNonPrimaryIndexes()[0] // i1 index on (val)
+	row := []tree.Datum{
+		tree.NewDInt(1), // id
+		tree.NewDInt(2), // val
+	}
+	err := deleteSecondaryIndexEntry(ctx, row, kvDB, tableDesc, secIndex)
+	require.NoError(t, err)
+
+	// Second run: with corruption, should detect the missing index entry.
+	_, err = db.Exec("EXPERIMENTAL SCRUB TABLE db.t")
+	require.Error(t, err, "INSPECT should fail when corruption is detected")
+	require.Contains(t, err.Error(), "INSPECT found inconsistencies")
+	require.Equal(t, initialRuns+2, metrics.Runs.Count(),
+		"Runs counter should increment for each job execution")
+	require.Equal(t, initialRunsWithIssues+1, metrics.RunsWithIssues.Count(),
+		"RunsWithIssues should increment when issues are found")
+	require.Equal(t, initialIssuesFound+1, metrics.IssuesFound.Count(),
+		"IssuesFound should increment when issues are detected")
+
+	// Third run: on a different clean table to verify RunsWithIssues doesn't increment again.
+	runner.Exec(t, `
+		CREATE TABLE db.t2 (
+			id INT PRIMARY KEY,
+			val INT
+		);
+		CREATE INDEX i2 on db.t2 (val);
+		INSERT INTO db.t2 VALUES (1, 2), (2, 3);
+	`)
+	runner.Exec(t, "EXPERIMENTAL SCRUB TABLE db.t2")
+	require.Equal(t, initialRuns+3, metrics.Runs.Count(),
+		"Runs counter should increment for third job execution")
+	require.Equal(t, initialRunsWithIssues+1, metrics.RunsWithIssues.Count(),
+		"RunsWithIssues should NOT increment for successful job")
+	require.Equal(t, initialIssuesFound+1, metrics.IssuesFound.Count(),
+		"IssuesFound should NOT increment for successful job")
+}

pkg/sql/inspect/inspect_processor.go

@@ -29,6 +29,10 @@ import (
 )
 
 var (
+	// errInspectFoundInconsistencies is a sentinel error used to mark errors
+	// returned when INSPECT jobs find data inconsistencies.
+	errInspectFoundInconsistencies = pgerror.New(pgcode.DataException, "INSPECT found inconsistencies")
+
 	processorConcurrencyOverride = settings.RegisterIntSetting(
 		settings.ApplicationLevel,
 		"sql.inspect.processor_concurrency",
@@ -176,7 +180,7 @@ func (p *inspectProcessor) runInspect(ctx context.Context, output execinfra.RowR
 
 	log.Dev.Infof(ctx, "INSPECT processor completed processorID=%d issuesFound=%t", p.processorID, p.logger.hasIssues())
 	if p.logger.hasIssues() {
-		return pgerror.Newf(pgcode.DataException, "INSPECT found inconsistencies")
+		return errInspectFoundInconsistencies
 	}
 	return nil
 }
@@ -194,15 +198,21 @@ func getProcessorConcurrency(flowCtx *execinfra.FlowCtx) int {
 
 // getInspectLogger returns a logger for the inspect processor.
 func getInspectLogger(flowCtx *execinfra.FlowCtx, jobID jobspb.JobID) inspectLogger {
+	execCfg := flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig)
+	metrics := execCfg.JobRegistry.MetricsStruct().Inspect.(*InspectMetrics)
+
 	loggers := inspectLoggers{
 		&logSink{},
 		&tableSink{
 			db:    flowCtx.Cfg.DB,
 			jobID: jobID,
 		},
+		&metricsLogger{
+			issuesFoundCtr: metrics.IssuesFound,
+		},
 	}
 
-	knobs := flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig).InspectTestingKnobs
+	knobs := execCfg.InspectTestingKnobs
 	if knobs != nil && knobs.InspectIssueLogger != nil {
 		loggers = append(loggers, knobs.InspectIssueLogger.(inspectLogger))
 	}