sql/inspect: fix progress tracking bugs causing over 100% completion

Fix two bugs in INSPECT job progress tracking that caused inaccurate completion
percentages, sometimes exceeding 100%.

The first issue was in inspect_processor.go, where processSpan() checked the
loop termination condition (!ok) before accounting for the final completed
check. When the last check finished we would exit the loop before reporting
the final check as complete.

The second issue was in inspect_job.go, where initProgressFromPlan() calculated
total checks using PK spans rather than the partitioned spans actually used by
processors. The number of total checks that we think need to be complete were
much lower than the partitioned spans. This caused us to report completion
rates in excess of 100%.

Epic: CRDB-30356

Informs: #154457

Release note: none

Author: spilchen

pkg/sql/inspect/index_consistency_check_test.go

@@ -31,6 +31,21 @@ import (
 
 const expectedInspectFoundInconsistencies = "INSPECT found inconsistencies"
 
+// requireCheckCountsMatch verifies that the job's total check count equals its completed check count.
+// This is used to verify that progress tracking correctly counted all checks.
+func requireCheckCountsMatch(t *testing.T, r *sqlutils.SQLRunner, jobID int64) {
+	t.Helper()
+	var totalChecks, completedChecks int64
+	r.QueryRow(t, `
+		SELECT
+			(crdb_internal.pb_to_json('cockroach.sql.jobs.jobspb.Progress', value)->'inspect'->>'jobTotalCheckCount')::INT,
+			(crdb_internal.pb_to_json('cockroach.sql.jobs.jobspb.Progress', value)->'inspect'->>'jobCompletedCheckCount')::INT
+		FROM system.job_info
+		WHERE job_id = $1 AND info_key = 'legacy_progress'
+	`, jobID).Scan(&totalChecks, &completedChecks)
+	require.Equal(t, totalChecks, completedChecks, "total checks should equal completed checks when job succeeds")
+}
+
 // encodeSecondaryIndexEntry encodes row data into a secondary index entry.
 // The datums must be ordered according to the table's public columns.
 // Returns the encoded index entry, expecting exactly one entry to be produced.
@@ -517,13 +532,15 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 			}
 
 			// Validate job status matches expected outcome
+			var jobID int64
 			var jobStatus string
 			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)
+			r.QueryRow(t, `SELECT job_id, status, fraction_completed FROM [SHOW JOBS] WHERE job_type = 'INSPECT' ORDER BY job_id DESC LIMIT 1`).Scan(&jobID, &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")
+				requireCheckCountsMatch(t, r, jobID)
 			} else {
 				require.Equal(t, "failed", jobStatus, "expected job to fail when inconsistencies found")
 			}
@@ -578,9 +595,11 @@ func TestIndexConsistencyWithReservedWordColumns(t *testing.T) {
 	require.Equal(t, 0, issueLogger.numIssuesFound(), "No issues should be found in happy path test")
 
 	// Verify job succeeded and progress reached 100%
+	var jobID int64
 	var jobStatus string
 	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)
+	r.QueryRow(t, `SELECT job_id, status, fraction_completed FROM [SHOW JOBS] WHERE job_type = 'INSPECT' ORDER BY job_id DESC LIMIT 1`).Scan(&jobID, &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")
+	requireCheckCountsMatch(t, r, jobID)
 }

pkg/sql/inspect/inspect_job.go

@@ -62,12 +62,14 @@ func (c *inspectResumer) Resume(ctx context.Context, execCtx interface{}) error
 		return err
 	}
 
-	progressTracker, cleanupProgress, remainingSpans, err := c.setupProgressTrackingAndFilter(ctx, execCfg, pkSpans)
+	progressTracker, completedSpans, cleanupProgress, err := c.setupProgressTracking(ctx, execCfg)
 	if err != nil {
 		return err
 	}
 	defer cleanupProgress()
 
+	remainingSpans := c.filterCompletedSpans(pkSpans, completedSpans)
+
 	// If all spans are completed, job is done.
 	if len(remainingSpans) == 0 {
 		log.Dev.Infof(ctx, "all spans already completed, INSPECT job finished")
@@ -79,6 +81,13 @@ func (c *inspectResumer) Resume(ctx context.Context, execCtx interface{}) error
 		return err
 	}
 
+	// After planning, we have the finalized set of spans to process (adjacent
+	// spans on the same node are merged). Compute the checks to run and initialize
+	// progress tracking from the plan.
+	if err := c.initProgressFromPlan(ctx, execCfg, progressTracker, plan, completedSpans); err != nil {
+		return err
+	}
+
 	if err := c.runInspectPlan(ctx, jobExecCtx, planCtx, plan, progressTracker); err != nil {
 		return err
 	}
@@ -240,11 +249,11 @@ func (c *inspectResumer) runInspectPlan(
 	return metadataCallbackWriter.Err()
 }
 
-// setupProgressTrackingAndFilter initializes progress tracking and returns
-// it, along with a cleanup function and the remaining spans to process.
-func (c *inspectResumer) setupProgressTrackingAndFilter(
-	ctx context.Context, execCfg *sql.ExecutorConfig, pkSpans []roachpb.Span,
-) (*inspectProgressTracker, func(), []roachpb.Span, error) {
+// setupProgressTracking initializes progress tracking and returns
+// it, along with a cleanup function.
+func (c *inspectResumer) setupProgressTracking(
+	ctx context.Context, execCfg *sql.ExecutorConfig,
+) (*inspectProgressTracker, []roachpb.Span, func(), error) {
 	// Create and initialize the tracker. We use the completed spans from the job
 	// (if any) to filter out the spans we need to process in this run of the job.
 	progressTracker := newInspectProgressTracker(
@@ -256,32 +265,53 @@ func (c *inspectResumer) setupProgressTrackingAndFilter(
 	if err != nil {
 		return nil, nil, nil, err
 	}
-	remainingSpans := c.filterCompletedSpans(pkSpans, completedSpans)
+
+	cleanup := func() {
+		progressTracker.terminateTracker()
+	}
+
+	return progressTracker, completedSpans, cleanup, nil
+}
+
+// initProgressFromPlan initializes job progress based on the actual partitioned spans
+// that will be processed.
+func (c *inspectResumer) initProgressFromPlan(
+	ctx context.Context,
+	execCfg *sql.ExecutorConfig,
+	progressTracker *inspectProgressTracker,
+	plan *sql.PhysicalPlan,
+	completedPartitionedSpans []roachpb.Span,
+) error {
+	// Extract all spans from the plan processors.
+	var remainingPartitionedSpans []roachpb.Span
+	for i := range plan.Processors {
+		if plan.Processors[i].Spec.Core.Inspect != nil {
+			remainingPartitionedSpans = append(remainingPartitionedSpans, plan.Processors[i].Spec.Core.Inspect.Spans...)
+		}
+	}
 
 	applicabilityCheckers, err := buildApplicabilityCheckers(c.job.Details().(jobspb.InspectDetails))
 	if err != nil {
-		return nil, nil, nil, err
+		return err
 	}
 
 	// Calculate total applicable checks on ALL spans (not just remaining ones)
 	// This ensures consistent progress calculation across job restarts.
-	totalCheckCount, err := countApplicableChecks(pkSpans, applicabilityCheckers, execCfg.Codec)
+	completedCheckCount, err := countApplicableChecks(completedPartitionedSpans, applicabilityCheckers, execCfg.Codec)
 	if err != nil {
-		return nil, nil, nil, err
+		return err
 	}
-	completedCheckCount, err := countApplicableChecks(completedSpans, applicabilityCheckers, execCfg.Codec)
+	remainingCheckCount, err := countApplicableChecks(remainingPartitionedSpans, applicabilityCheckers, execCfg.Codec)
 	if err != nil {
-		return nil, nil, nil, err
+		return err
 	}
 
-	if err := progressTracker.initJobProgress(ctx, totalCheckCount, completedCheckCount); err != nil {
-		return nil, nil, nil, err
-	}
-	cleanup := func() {
-		progressTracker.terminateTracker()
-	}
+	totalCheckCount := completedCheckCount + remainingCheckCount
+
+	log.Dev.Infof(ctx, "INSPECT progress init: %d partitioned spans, %d total checks (%d remaining + %d completed)",
+		len(remainingPartitionedSpans), totalCheckCount, remainingCheckCount, completedCheckCount)
 
-	return progressTracker, cleanup, remainingSpans, nil
+	return progressTracker.initJobProgress(ctx, totalCheckCount, completedCheckCount)
 }
 
 // filterCompletedSpans removes spans that are already completed from the list to process.

pkg/sql/inspect/inspect_job_test.go

@@ -129,11 +129,12 @@ func TestInspectJobImplicitTxnSemantics(t *testing.T) {
 			}()
 
 			// Wait for the job to finish.
+			var jobID int64
 			var status string
 			var fractionCompleted float64
 			testutils.SucceedsSoon(t, func() error {
-				row := db.QueryRow(`SELECT status, fraction_completed FROM [SHOW JOBS] WHERE job_type = 'INSPECT' ORDER BY job_id DESC LIMIT 1`)
-				if err := row.Scan(&status, &fractionCompleted); err != nil {
+				row := db.QueryRow(`SELECT job_id, status, fraction_completed FROM [SHOW JOBS] WHERE job_type = 'INSPECT' ORDER BY job_id DESC LIMIT 1`)
+				if err := row.Scan(&jobID, &status, &fractionCompleted); err != nil {
 					return err
 				}
 				if status == "succeeded" || status == "failed" {
@@ -304,3 +305,90 @@ func TestInspectJobProtectedTimestamp(t *testing.T) {
 		})
 	}
 }
+
+// TestInspectProgressWithMultiRangeTable is a regression test for the bug where
+// INSPECT progress could exceed 100% when a table had many ranges. This test
+// creates a multi-node cluster with a multi-range table and verifies that:
+// 1. The progress never exceeds 100%
+// 2. The final progress is exactly 100% when the job completes
+// 3. The total check count is based on partitioned spans, not PK spans
+func TestInspectProgressWithMultiRangeTable(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	skip.UnderShort(t)
+
+	ctx := context.Background()
+	const numNodes = 3
+	tc := serverutils.StartCluster(t, numNodes, base.TestClusterArgs{})
+	defer tc.Stopper().Stop(ctx)
+
+	db := tc.ServerConn(0)
+	runner := sqlutils.MakeSQLRunner(db)
+
+	// Create a table and split it into multiple ranges.
+	runner.Exec(t, `
+		CREATE DATABASE testdb;
+		USE testdb;
+		CREATE TABLE multi_range_table (
+			id INT PRIMARY KEY,
+			val1 INT,
+			val2 STRING,
+			INDEX idx_val1 (val1),
+			INDEX idx_val2 (val2)
+		);
+	`)
+
+	// Insert data to create multiple ranges. We'll insert enough data and then
+	// manually split to ensure multiple ranges.
+	runner.Exec(t, `INSERT INTO multi_range_table SELECT i, i*2, md5(i::STRING) FROM generate_series(1, 1000) AS g(i)`)
+	for i := 100; i <= 900; i += 100 {
+		runner.Exec(t, `ALTER TABLE multi_range_table SPLIT AT VALUES ($1)`, i)
+	}
+	runner.Exec(t, `ALTER TABLE multi_range_table SCATTER`)
+
+	// Wait for scatter to distribute ranges across nodes.
+	var rangeCount, nodeCount int
+	testutils.SucceedsSoon(t, func() error {
+		// Count total ranges and verify distribution.
+		runner.QueryRow(t, `
+			WITH r AS (SHOW RANGES FROM TABLE multi_range_table WITH DETAILS)
+			SELECT count(DISTINCT range_id), count(DISTINCT lease_holder)
+			FROM r
+		`).Scan(&rangeCount, &nodeCount)
+
+		if rangeCount <= 5 {
+			return errors.Newf("waiting for splits: only %d ranges", rangeCount)
+		}
+
+		if nodeCount < 2 {
+			return errors.Newf("waiting for scatter to multiple nodes: ranges only on %d node(s)", nodeCount)
+		}
+
+		return nil
+	})
+
+	// Start the INSPECT job.
+	// TODO(148365): Run INSPECT instead of SCRUB.
+	t.Logf("Starting INSPECT job on table with %d ranges distributed across %d nodes", rangeCount, nodeCount)
+	_, err := db.Exec(`
+		SET enable_scrub_job = true;
+		COMMIT;
+		EXPERIMENTAL SCRUB TABLE multi_range_table`)
+	require.NoError(t, err)
+
+	var jobID int64
+	var status string
+	var fractionCompleted float64
+	runner.QueryRow(t, `
+		SELECT job_id, status, fraction_completed
+		FROM [SHOW JOBS]
+		WHERE job_type = 'INSPECT'
+		ORDER BY created DESC
+		LIMIT 1
+	`).Scan(&jobID, &status, &fractionCompleted)
+	t.Logf("Job %d: status=%s, fraction_completed=%.4f", jobID, status, fractionCompleted)
+
+	require.Equal(t, "succeeded", status, "INSPECT job should succeed")
+	require.InEpsilon(t, 1.0, fractionCompleted, 0.01,
+		"progress should be ~100%% at completion, got %.2f%%", fractionCompleted*100)
+}

pkg/sql/inspect/inspect_processor.go

@@ -114,7 +114,7 @@ func (p *inspectProcessor) Run(ctx context.Context, output execinfra.RowReceiver
 // Each span is read from a buffered channel and passed to processSpan.
 // The function blocks until all spans are processed or an error occurs.
 func (p *inspectProcessor) runInspect(ctx context.Context, output execinfra.RowReceiver) error {
-	log.Dev.Infof(ctx, "INSPECT processor started processorID=%d concurrency=%d", p.processorID, p.concurrency)
+	log.Dev.Infof(ctx, "INSPECT processor started processorID=%d concurrency=%d spans=%d", p.processorID, p.concurrency, len(p.spec.Spans))
 
 	group := ctxgroup.WithContext(ctx)
 
@@ -262,9 +262,6 @@ func (p *inspectProcessor) processSpan(
 		if stepErr != nil {
 			return stepErr
 		}
-		if !ok {
-			break
-		}
 
 		// Check if any inspections have completed (when the count decreases).
 		currentCheckCount := runner.CheckCount()
@@ -275,6 +272,10 @@ func (p *inspectProcessor) processSpan(
 				return err
 			}
 		}
+
+		if !ok {
+			break
+		}
 	}
 
 	// Report span completion for checkpointing.