sql/inspect: add progress tracking to INSPECT

Previously, when you ran an INSPECT job we didn't maintain any kind of
progress tracking. This commit adds it. The granularity that we use for
progress tracking is with checks x spans. Note: each check won't always
use all spans, so we only count spans that apply to the check.

The processors flow back the completed checks (per span) they have done.
A goroutine runs at the coordinator that will peridocially (every 10
seconds) update the job data with the progress update.

Informs #148297

Release note: none
Epic: CRDB-30356

Author: spilchen

pkg/jobs/jobspb/jobs.proto

@@ -1538,7 +1538,13 @@ message HotRangesLoggerProgress {
 }
 
 message InspectProgress {
+  int64 job_total_check_count = 1;
+  int64 job_completed_check_count = 2;
+}
 
+message InspectProcessorProgress {
+  int64 checks_completed = 1;  // Number of checks completed since last update.
+  bool finished = 2;           // Processor finished all checks.
 }
 
 message ImportRollbackProgress {}

pkg/sql/inspect/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "inspect_processor.go",
         "issue.go",
         "log_sink.go",
+        "progress.go",
         "runner.go",
         "span_source.go",
         "table_sink.go",
@@ -46,10 +47,13 @@ go_library(
         "//pkg/util/ctxgroup",
         "//pkg/util/hlc",
         "//pkg/util/log",
+        "//pkg/util/stop",
+        "//pkg/util/syncutil",
         "//pkg/util/timeutil",
         "//pkg/util/tracing",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",
+        "@com_github_gogo_protobuf//types",
     ],
 )
 
@@ -61,6 +65,7 @@ go_test(
         "inspect_processor_test.go",
         "issue_test.go",
         "main_test.go",
+        "progress_test.go",
         "runner_test.go",
         "table_sink_test.go",
     ],
@@ -71,13 +76,15 @@ go_test(
     }),
     deps = [
         "//pkg/base",
+        "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/kv",
         "//pkg/kv/kvserver/protectedts",
         "//pkg/roachpb",
         "//pkg/security/securityassets",
         "//pkg/security/securitytest",
+        "//pkg/security/username",
         "//pkg/server",
         "//pkg/settings/cluster",
         "//pkg/sql",
@@ -99,8 +106,10 @@ go_test(
         "//pkg/util/leaktest",
         "//pkg/util/log",
         "//pkg/util/syncutil",
+        "//pkg/util/uuid",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",
+        "@com_github_gogo_protobuf//types",
         "@com_github_stretchr_testify//require",
     ],
 )

pkg/sql/inspect/index_consistency_check_test.go

@@ -518,10 +518,12 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 
 			// Validate job status matches expected outcome
 			var jobStatus string
-			r.QueryRow(t, `SELECT status FROM [SHOW JOBS] WHERE job_type = 'INSPECT' ORDER BY job_id DESC LIMIT 1`).Scan(&jobStatus)
+			var fractionCompleted float64
+			r.QueryRow(t, `SELECT status, fraction_completed FROM [SHOW JOBS] WHERE job_type = 'INSPECT' ORDER BY job_id DESC LIMIT 1`).Scan(&jobStatus, &fractionCompleted)
 
 			if tc.expectedErrRegex == "" {
 				require.Equal(t, "succeeded", jobStatus, "expected job to succeed when no issues found")
+				require.InEpsilon(t, 1.0, fractionCompleted, 0.01, "progress should reach 100%% on successful completion")
 			} else {
 				require.Equal(t, "failed", jobStatus, "expected job to fail when inconsistencies found")
 			}
@@ -575,8 +577,10 @@ func TestIndexConsistencyWithReservedWordColumns(t *testing.T) {
 	require.NoError(t, err, "should succeed on table with reserved word column names")
 	require.Equal(t, 0, issueLogger.numIssuesFound(), "No issues should be found in happy path test")
 
-	// Verify job succeeded
+	// Verify job succeeded and progress reached 100%
 	var jobStatus string
-	r.QueryRow(t, `SELECT status FROM [SHOW JOBS] WHERE job_type = 'INSPECT' ORDER BY job_id DESC LIMIT 1`).Scan(&jobStatus)
+	var fractionCompleted float64
+	r.QueryRow(t, `SELECT status, fraction_completed FROM [SHOW JOBS] WHERE job_type = 'INSPECT' ORDER BY job_id DESC LIMIT 1`).Scan(&jobStatus, &fractionCompleted)
 	require.Equal(t, "succeeded", jobStatus, "INSPECT job should succeed")
+	require.InEpsilon(t, 1.0, fractionCompleted, 0.01, "progress should reach 100%% on successful completion")
 }

pkg/sql/inspect/inspect_job.go

@@ -18,7 +18,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
-	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/sql/physicalplan"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
@@ -67,13 +66,19 @@ func (c *inspectResumer) Resume(ctx context.Context, execCtx interface{}) error
 		return err
 	}
 
-	if err := c.runInspectPlan(ctx, jobExecCtx, planCtx, plan); err != nil {
+	progressTracker, cleanupProgress, err := c.setupProgressTracking(ctx, execCfg, pkSpans)
+	if err != nil {
+		return err
+	}
+	defer cleanupProgress()
+
+	if err := c.runInspectPlan(ctx, jobExecCtx, planCtx, plan, progressTracker); err != nil {
 		return err
 	}
 
 	c.maybeCleanupProtectedTimestamp(ctx, execCfg)
 
-	return c.markJobComplete(ctx)
+	return nil
 }
 
 // OnFailOrCancel implements the Resumer interface
@@ -181,19 +186,26 @@ func (c *inspectResumer) planInspectProcessors(
 }
 
 // runInspectPlan executes the distributed physical plan for the INSPECT job.
-// It sets up a metadata-only DistSQL receiver to collect any execution errors,
-// then runs the plan using the provided planning context and evaluation context.
-// This function returns any error surfaced via metadata from the processors.
+// It sets up a metadata-only DistSQL receiver to collect any execution errors
+// and progress updates, then runs the plan using the provided planning context
+// and evaluation context. This function returns any error surfaced via metadata
+// from the processors.
 func (c *inspectResumer) runInspectPlan(
 	ctx context.Context,
 	jobExecCtx sql.JobExecContext,
 	planCtx *sql.PlanningCtx,
 	plan *sql.PhysicalPlan,
+	progressTracker *inspectProgressTracker,
 ) error {
 	execCfg := jobExecCtx.ExecCfg()
 
 	metadataCallbackWriter := sql.NewMetadataOnlyMetadataCallbackWriter(
-		func(context.Context, *execinfrapb.ProducerMetadata) error { return nil })
+		func(ctx context.Context, meta *execinfrapb.ProducerMetadata) error {
+			if meta.BulkProcessorProgress != nil {
+				return progressTracker.handleProcessorProgress(ctx, meta)
+			}
+			return nil
+		})
 
 	distSQLReceiver := sql.MakeDistSQLReceiver(
 		ctx,
@@ -216,18 +228,62 @@ func (c *inspectResumer) runInspectPlan(
 	return metadataCallbackWriter.Err()
 }
 
-func (c *inspectResumer) markJobComplete(ctx context.Context) error {
-	// TODO(148297): add fine-grained progress reporting
-	return c.job.NoTxn().Update(ctx,
-		func(_ isql.Txn, md jobs.JobMetadata, ju *jobs.JobUpdater) error {
-			progress := md.Progress
-			progress.Progress = &jobspb.Progress_FractionCompleted{
-				FractionCompleted: 1,
-			}
-			ju.UpdateProgress(progress)
-			return nil
-		},
-	)
+// setupProgressTracking initializes and starts progress tracking for the INSPECT job.
+// It calculates the total number of applicable checks, initializes progress to 0%, and starts
+// the background update goroutine. Returns the progress tracker and cleanup function.
+func (c *inspectResumer) setupProgressTracking(
+	ctx context.Context, execCfg *sql.ExecutorConfig, pkSpans []roachpb.Span,
+) (*inspectProgressTracker, func(), error) {
+	fractionInterval := fractionUpdateInterval.Get(&execCfg.Settings.SV)
+	details := c.job.Details().(jobspb.InspectDetails)
+
+	// Build applicability checkers for progress tracking
+	applicabilityCheckers, err := buildApplicabilityCheckers(details)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Calculate total applicable checks: only count checks that will actually run
+	totalCheckCount, err := countApplicableChecks(pkSpans, applicabilityCheckers, execCfg.Codec)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Create and initialize progress tracker
+	progressTracker := newInspectProgressTracker(c.job, fractionInterval)
+	if err := progressTracker.initJobProgress(ctx, totalCheckCount); err != nil {
+		progressTracker.terminateTracker(ctx)
+		return nil, nil, err
+	}
+
+	// Start background progress updates
+	progressTracker.startBackgroundUpdates(ctx)
+
+	// Return cleanup function
+	cleanup := func() {
+		progressTracker.terminateTracker(ctx)
+	}
+
+	return progressTracker, cleanup, nil
+}
+
+// buildApplicabilityCheckers creates lightweight applicability checkers from InspectDetails.
+// These are used only for progress calculation and don't require the full check machinery.
+func buildApplicabilityCheckers(
+	details jobspb.InspectDetails,
+) ([]inspectCheckApplicability, error) {
+	checkers := make([]inspectCheckApplicability, 0, len(details.Checks))
+	for _, specCheck := range details.Checks {
+		switch specCheck.Type {
+		case jobspb.InspectCheckIndexConsistency:
+			checkers = append(checkers, &indexConsistencyCheckApplicability{
+				tableID: specCheck.TableID,
+			})
+		default:
+			return nil, errors.AssertionFailedf("unsupported inspect check type: %v", specCheck.Type)
+		}
+	}
+	return checkers, nil
 }
 
 // maybeProtectTimestamp creates a protected timestamp record for the AsOf

pkg/sql/inspect/inspect_processor.go

@@ -8,6 +8,7 @@ package inspect
 import (
 	"context"
 	"runtime"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -24,6 +25,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
+	pbtypes "github.com/gogo/protobuf/types"
 )
 
 var (
@@ -35,6 +37,14 @@ var (
 		0,
 		settings.NonNegativeInt,
 	)
+
+	fractionUpdateInterval = settings.RegisterDurationSetting(
+		settings.ApplicationLevel,
+		"sql.inspect.fraction_update_interval",
+		"the amount of time between INSPECT job percent complete progress updates",
+		10*time.Second,
+		settings.DurationWithMinimum(1*time.Millisecond),
+	)
 )
 
 type inspectCheckFactory func(asOf hlc.Timestamp) inspectCheck
@@ -116,7 +126,7 @@ func (p *inspectProcessor) runInspect(ctx context.Context, output execinfra.RowR
 						// Channel is closed, no more spans to process.
 						return nil
 					}
-					if err := p.processSpan(ctx, span, workerIndex); err != nil {
+					if err := p.processSpan(ctx, span, workerIndex, output); err != nil {
 						// On error, return it. ctxgroup will cancel all other goroutines.
 						return err
 					}
@@ -150,6 +160,12 @@ func (p *inspectProcessor) runInspect(ctx context.Context, output execinfra.RowR
 	if err := group.Wait(); err != nil {
 		return err
 	}
+
+	// Send final completion message to indicate this processor is finished
+	if err := p.sendInspectProgress(ctx, output, 0, true /* finished */); err != nil {
+		return err
+	}
+
 	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")
@@ -188,9 +204,10 @@ func getInspectLogger(flowCtx *execinfra.FlowCtx, jobID jobspb.JobID) inspectLog
 
 // processSpan executes all configured inspect checks against a single span.
 // It instantiates a fresh set of checks from the configured factories and uses
-// an inspectRunner to drive their execution.
+// an inspectRunner to drive their execution. After processing the span, it
+// sends progress metadata to the coordinator.
 func (p *inspectProcessor) processSpan(
-	ctx context.Context, span roachpb.Span, workerIndex int,
+	ctx context.Context, span roachpb.Span, workerIndex int, output execinfra.RowReceiver,
 ) (err error) {
 	asOfToUse := p.getTimestampForSpan()
 
@@ -222,6 +239,10 @@ func (p *inspectProcessor) processSpan(
 		}
 	}()
 
+	// Keep track of completed checks for progress updates.
+	initialCheckCount := len(checks)
+	lastSeenCheckCount := initialCheckCount
+
 	// Process all checks on the given span.
 	for {
 		ok, stepErr := runner.Step(ctx, p.cfg, span, workerIndex)
@@ -231,7 +252,46 @@ func (p *inspectProcessor) processSpan(
 		if !ok {
 			break
 		}
+
+		// Check if any inspections have completed (when the count decreases).
+		currentCheckCount := runner.CheckCount()
+		if currentCheckCount < lastSeenCheckCount {
+			checksCompleted := lastSeenCheckCount - currentCheckCount
+			lastSeenCheckCount = currentCheckCount
+			if err := p.sendInspectProgress(ctx, output, int64(checksCompleted), false /* finished */); err != nil {
+				return err
+			}
+		}
 	}
+
+	return nil
+}
+
+// sendInspectProgress marshals and sends inspect processor progress via BulkProcessorProgress.
+func (p *inspectProcessor) sendInspectProgress(
+	ctx context.Context, output execinfra.RowReceiver, checksCompleted int64, finished bool,
+) error {
+	progressMsg := &jobspb.InspectProcessorProgress{
+		ChecksCompleted: checksCompleted,
+		Finished:        finished,
+	}
+
+	progressAny, err := pbtypes.MarshalAny(progressMsg)
+	if err != nil {
+		return errors.Wrapf(err, "unable to marshal inspect processor progress")
+	}
+
+	meta := &execinfrapb.ProducerMetadata{
+		BulkProcessorProgress: &execinfrapb.RemoteProducerMetadata_BulkProcessorProgress{
+			ProgressDetails: *progressAny,
+			NodeID:          p.flowCtx.NodeID.SQLInstanceID(),
+			FlowID:          p.flowCtx.ID,
+			ProcessorID:     p.processorID,
+			Drained:         finished,
+		},
+	}
+
+	output.Push(nil, meta)
 	return nil
 }