Merge #154136 #154255

154136: sql/inspect: add progress tracking to INSPECT r=spilchen a=spilchen

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 periodically (every 10 seconds) update the job data with the progress update.

Informs #148297

Release note: none
Epic: CRDB-30356

154255: sql/import: pass deep copy of BulkOpSummary to progress r=ZhouXing19 a=ZhouXing19

Fixes: #153480

Prior to this change, we passed the `BulkOpSummary` map directly from `accumulatedBulkSummary` to `prog.Summary` in #152745. This caused a `concurrent map iteration and map write` panic because the map was being updated via `metaFn` while simultaneously being marshaled into a protobuf during `jobs.Update` calls from `FractionProgressed`.

This change creates a deep copy of the map when assigning from `accumulatedBulkSummary` to `prog.Summary`, eliminating the concurrency issue by ensuring each goroutine operates on separate map instances.

Release Notes: None

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

Author: 3 people

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/kv/kvpb/api.go

@@ -2133,6 +2133,21 @@ func (b *BulkOpSummary) Add(other BulkOpSummary) {
 	}
 }
 
+// DeepCopy returns a deep copy of the original BulkOpSummary.
+func (b *BulkOpSummary) DeepCopy() BulkOpSummary {
+	cpy := BulkOpSummary{
+		DataSize:    b.DataSize,
+		SSTDataSize: b.SSTDataSize,
+	}
+	if b.EntryCounts != nil {
+		cpy.EntryCounts = make(map[uint64]int64, len(b.EntryCounts))
+		for k, v := range b.EntryCounts {
+			cpy.EntryCounts[k] = v
+		}
+	}
+	return cpy
+}
+
 // MustSetValue is like SetValue, except it resets the enum and panics if the
 // provided value is not a valid variant type.
 func (e *RangeFeedEvent) MustSetValue(value interface{}) {

pkg/kv/kvpb/api.proto

@@ -1935,6 +1935,7 @@ message BulkOpSummary {
   // generation logic is also available in the BulkOpSummaryID helper. It does
   // not take MVCC range tombstones into account.
   map<uint64, int64> entry_counts = 5;
+  // Please update BulkOpSummary.DeepCopy() if new fields are added.
 }
 
 // ExportResponse is the response to an Export() operation.

pkg/sql/importer/import_processor_planning.go

@@ -176,7 +176,7 @@ func distImport(
 			}
 
 			accumulatedBulkSummary.Lock()
-			prog.Summary = accumulatedBulkSummary.BulkOpSummary
+			prog.Summary = accumulatedBulkSummary.BulkOpSummary.DeepCopy()
 			accumulatedBulkSummary.Unlock()
 			return overall / float32(len(from))
 		},

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
 }