sql/inspect: add checkpointing to INSPECT jobs

Previously, INSPECT jobs did not support checkpointing, meaning job restarts
would reprocess all data from the beginning. This change adds span based
checkpointing to allow jobs to resume from their last completed spans.

Progress is tracked using jobfrontier to store completed spans. On job
restart, the resumer loads completed spans and filters them from the work to
be done.

Closes #148297

Release note: None

Epic: CRDB-30356

Author: spilchen

pkg/sql/inspect/BUILD.bazel

@@ -18,6 +18,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/jobs",
+        "//pkg/jobs/jobfrontier",
         "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/kv/kvserver/protectedts/ptpb",
@@ -47,7 +48,8 @@ go_library(
         "//pkg/util/ctxgroup",
         "//pkg/util/hlc",
         "//pkg/util/log",
-        "//pkg/util/stop",
+        "//pkg/util/protoutil",
+        "//pkg/util/span",
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",
         "//pkg/util/tracing",
@@ -77,6 +79,7 @@ go_test(
     deps = [
         "//pkg/base",
         "//pkg/jobs",
+        "//pkg/jobs/jobfrontier",
         "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/kv",

pkg/sql/inspect/inspect_job.go

@@ -60,17 +60,22 @@ func (c *inspectResumer) Resume(ctx context.Context, execCtx interface{}) error
 		return err
 	}
 
-	// TODO(149460): add a goroutine that will replan the job on topology changes
-	plan, planCtx, err := c.planInspectProcessors(ctx, jobExecCtx, pkSpans)
+	progressTracker, cleanupProgress, remainingSpans, err := c.setupProgressTrackingAndFilter(ctx, execCfg, pkSpans)
 	if err != nil {
 		return err
 	}
+	defer cleanupProgress()
+
+	// If all spans are completed, job is done.
+	if len(remainingSpans) == 0 {
+		log.Dev.Infof(ctx, "all spans already completed, INSPECT job finished")
+		return nil
+	}
 
-	progressTracker, cleanupProgress, err := c.setupProgressTracking(ctx, execCfg, pkSpans)
+	plan, planCtx, err := c.planInspectProcessors(ctx, jobExecCtx, remainingSpans)
 	if err != nil {
 		return err
 	}
-	defer cleanupProgress()
 
 	if err := c.runInspectPlan(ctx, jobExecCtx, planCtx, plan, progressTracker); err != nil {
 		return err
@@ -202,7 +207,7 @@ func (c *inspectResumer) runInspectPlan(
 	metadataCallbackWriter := sql.NewMetadataOnlyMetadataCallbackWriter(
 		func(ctx context.Context, meta *execinfrapb.ProducerMetadata) error {
 			if meta.BulkProcessorProgress != nil {
-				return progressTracker.handleProcessorProgress(ctx, meta)
+				return progressTracker.handleProgressUpdate(ctx, meta)
 			}
 			return nil
 		})
@@ -228,43 +233,71 @@ func (c *inspectResumer) runInspectPlan(
 	return metadataCallbackWriter.Err()
 }
 
-// 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(
+// 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(), error) {
-	fractionInterval := fractionUpdateInterval.Get(&execCfg.Settings.SV)
-	details := c.job.Details().(jobspb.InspectDetails)
+) (*inspectProgressTracker, func(), []roachpb.Span, 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(
+		c.job,
+		&execCfg.Settings.SV,
+		execCfg.InternalDB,
+	)
+	completedSpans, err := progressTracker.initTracker(ctx)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	remainingSpans := c.filterCompletedSpans(pkSpans, completedSpans)
 
-	// Build applicability checkers for progress tracking
-	applicabilityCheckers, err := buildApplicabilityCheckers(details)
+	applicabilityCheckers, err := buildApplicabilityCheckers(c.job.Details().(jobspb.InspectDetails))
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
 	}
 
-	// Calculate total applicable checks: only count checks that will actually run
+	// 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)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
+	}
+	completedCheckCount, err := countApplicableChecks(completedSpans, applicabilityCheckers, execCfg.Codec)
+	if err != nil {
+		return nil, 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
+	if err := progressTracker.initJobProgress(ctx, totalCheckCount, completedCheckCount); err != nil {
+		return nil, nil, nil, err
+	}
+	cleanup := func() {
+		progressTracker.terminateTracker()
 	}
 
-	// Start background progress updates
-	progressTracker.startBackgroundUpdates(ctx)
+	return progressTracker, cleanup, remainingSpans, nil
+}
 
-	// Return cleanup function
-	cleanup := func() {
-		progressTracker.terminateTracker(ctx)
+// filterCompletedSpans removes spans that are already completed from the list to process.
+func (c *inspectResumer) filterCompletedSpans(
+	allSpans []roachpb.Span, completedSpans []roachpb.Span,
+) []roachpb.Span {
+	if len(completedSpans) == 0 {
+		return allSpans
+	}
+
+	completedGroup := roachpb.SpanGroup{}
+	completedGroup.Add(completedSpans...)
+
+	var remainingSpans []roachpb.Span
+	for _, span := range allSpans {
+		// Check if this span is fully contained in completed spans.
+		// We need to check if the entire span is covered by completed spans.
+		if !completedGroup.Encloses(span) {
+			remainingSpans = append(remainingSpans, span)
+		}
 	}
 
-	return progressTracker, cleanup, nil
+	return remainingSpans
 }
 
 // buildApplicabilityCheckers creates lightweight applicability checkers from InspectDetails.

pkg/sql/inspect/inspect_processor.go

@@ -45,6 +45,14 @@ var (
 		10*time.Second,
 		settings.DurationWithMinimum(1*time.Millisecond),
 	)
+
+	checkpointInterval = settings.RegisterDurationSetting(
+		settings.ApplicationLevel,
+		"sql.inspect.checkpoint_interval",
+		"the amount of time between INSPECT job checkpoint updates",
+		30*time.Second,
+		settings.DurationWithMinimum(1*time.Millisecond),
+	)
 )
 
 type inspectCheckFactory func(asOf hlc.Timestamp) inspectCheck
@@ -211,7 +219,7 @@ func (p *inspectProcessor) processSpan(
 ) (err error) {
 	asOfToUse := p.getTimestampForSpan()
 
-	// Only create checks that apply to this span
+	// Only create checks that apply to this span.
 	var checks []inspectCheck
 	for _, factory := range p.checkFactories {
 		check := factory(asOfToUse)
@@ -224,11 +232,6 @@ func (p *inspectProcessor) processSpan(
 		}
 	}
 
-	// If no checks apply to this span, there's nothing to do
-	if len(checks) == 0 {
-		return nil
-	}
-
 	runner := inspectRunner{
 		checks: checks,
 		logger: p.logger,
@@ -264,6 +267,11 @@ func (p *inspectProcessor) processSpan(
 		}
 	}
 
+	// Report span completion for checkpointing.
+	if err := p.sendSpanCompletionProgress(ctx, output, span, false /* finished */); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -305,6 +313,36 @@ func (p *inspectProcessor) getTimestampForSpan() hlc.Timestamp {
 	return p.spec.InspectDetails.AsOf
 }
 
+// sendSpanCompletionProgress sends progress indicating a span has been completed.
+// This is used for checkpointing to track which spans are done.
+func (p *inspectProcessor) sendSpanCompletionProgress(
+	ctx context.Context, output execinfra.RowReceiver, completedSpan roachpb.Span, finished bool,
+) error {
+	progressMsg := &jobspb.InspectProcessorProgress{
+		ChecksCompleted: 0, // No additional checks completed, just marking span done
+		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{
+			CompletedSpans:  []roachpb.Span{completedSpan}, // Mark this span as completed
+			ProgressDetails: *progressAny,
+			NodeID:          p.flowCtx.NodeID.SQLInstanceID(),
+			FlowID:          p.flowCtx.ID,
+			ProcessorID:     p.processorID,
+			Drained:         finished,
+		},
+	}
+
+	output.Push(nil, meta)
+	return nil
+}
+
 // newInspectProcessor constructs a new inspectProcessor from the given InspectSpec.
 // It parses the spec to generate a set of inspectCheck factories, sets up the span source,
 // and wires in logging and concurrency controls.