Merge pull request #154358 from spilchen/backport25.4-154136

release-25.4: sql/inspect: add progress tracking to INSPECT

Author: spilchen

pkg/jobs/jobspb/jobs.proto

@@ -1518,7 +1518,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
 }