Merge pull request #154701 from spilchen/backport25.4-154276

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

Author: rafiss

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.