sql: add support for meta-checks and post-checks

Previously the checks were independent and could only process individual
spans.
The row count check will be dependent on other checks that perform the
scan of each span and will need to aggregate the row counts after all
the spans have been processed.
This change sets up interfaces for meta-checks (dependent on other
checks running on the spans) and post-checks (run after all spans have
been processed.

Part of: #155472
Epic: CRDB-55075

Release note: None

Author: bghal

pkg/jobs/jobspb/jobs.go

@@ -172,3 +172,9 @@ var _ redact.SafeFormatter = (*RowLevelTTLProcessorProgress)(nil)
 func (r *RowLevelTTLProcessorProgress) SafeFormat(p redact.SafePrinter, _ rune) {
 	p.SafeString(redact.SafeString(r.String()))
 }
+
+// UpdateWithSpanProgress updates the InspectProgress with data from the
+// check-specific fields in the span progress update.
+func (p *InspectProgress) UpdateWithSpanProgress(prog *InspectProcessorProgress) {
+	return
+}

pkg/sql/inspect/index_consistency_check.go

@@ -57,21 +57,6 @@ func (c *indexConsistencyCheckApplicability) AppliesTo(
 	return spanContainsTable(c.tableID, codec, span)
 }
 
-// checkState represents the state of an index consistency check.
-type checkState int
-
-const (
-	// checkNotStarted indicates Start() has not been called yet.
-	checkNotStarted checkState = iota
-	// checkHashMatched indicates the hash precheck passed - no corruption detected,
-	// so the full check can be skipped.
-	checkHashMatched
-	// checkRunning indicates the full check is actively running and may produce more results.
-	checkRunning
-	// checkDone indicates the check has finished (iterator exhausted or error occurred).
-	checkDone
-)
-
 // indexConsistencyCheck verifies consistency between a table's primary index
 // and a specified secondary index by streaming rows from both sides of a
 // query. It reports an issue if a key exists in the primary but not the

pkg/sql/inspect/inspect_processor.go

@@ -100,6 +100,12 @@ type inspectProcessor struct {
 		// worker goroutines.
 		syncutil.Mutex
 	}
+
+	// test, if set, supports test-only behavior.
+	test *struct {
+		// progress is the job progress used to persist state after a span is processed.
+		progress *jobspb.InspectProgress
+	}
 }
 
 var _ execinfra.Processor = (*inspectProcessor)(nil)
@@ -202,6 +208,11 @@ func (p *inspectProcessor) runInspect(ctx context.Context, output execinfra.RowR
 		return err
 	}
 
+	// Run any post-checks after all spans have been processed.
+	if err := p.processPostChecks(ctx); 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
@@ -268,7 +279,7 @@ func (p *inspectProcessor) processSpan(
 	asOfToUse := p.getTimestampForSpan()
 
 	// Only create checks that apply to this span.
-	var checks []inspectCheck
+	var checks inspectChecks
 	for _, factory := range p.checkFactories {
 		check := factory(asOfToUse)
 		applies, err := check.AppliesTo(p.cfg.Codec, span)
@@ -279,6 +290,16 @@ func (p *inspectProcessor) processSpan(
 			checks = append(checks, check)
 		}
 	}
+	checks.sortStable()
+
+	// Provide the meta-checks with a reference to all the other checks.
+	for _, check := range checks {
+		if metaCheck, ok := check.(inspectMetaCheck); ok {
+			if err := metaCheck.RegisterChecks(checks); err != nil {
+				return err
+			}
+		}
+	}
 
 	runner := inspectRunner{
 		checks: checks,
@@ -316,14 +337,74 @@ func (p *inspectProcessor) processSpan(
 		}
 	}
 
+	progressMsg := &jobspb.InspectProcessorProgress{
+		ChecksCompleted: 0, // No additional checks completed, just marking span done
+		Finished:        false,
+	}
+
+	for _, check := range checks {
+		if check, ok := check.(inspectMetaCheck); ok {
+			err := check.MetaCheck(ctx, progressMsg, p.loggerBundle)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
 	// Report span completion for checkpointing.
-	if err := p.sendSpanCompletionProgress(ctx, output, span, false /* finished */); err != nil {
+	if err := p.sendSpanCompletionProgress(ctx, output, span, false /* finished */, progressMsg); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// processPostChecks runs any post-checks using the accumulated job progress.
+func (p *inspectProcessor) processPostChecks(ctx context.Context) (err error) {
+	progress, err := p.loadProgress(ctx, p.spec.JobID)
+	if err != nil {
 		return err
 	}
 
+	for _, factory := range p.checkFactories {
+		check := factory(p.getTimestampForSpan())
+		if postCheck, ok := check.(inspectPostCheck); ok {
+			issues, err := postCheck.Issues(ctx, progress)
+			if err != nil {
+				return err
+			}
+			for _, issue := range issues {
+				err = p.loggerBundle.logIssue(ctx, issue)
+				if err != nil {
+					return errors.Wrapf(err, "error logging inspect issue")
+				}
+			}
+		}
+	}
+
 	return nil
 }
 
+func (p *inspectProcessor) loadProgress(
+	ctx context.Context, jobID jobspb.JobID,
+) (*jobspb.InspectProgress, error) {
+	if p.test != nil {
+		return p.test.progress, nil
+	}
+
+	job, err := p.cfg.JobRegistry.LoadJob(ctx, p.spec.JobID)
+	if err != nil {
+		return nil, err
+	}
+
+	progress, ok := job.Progress().Details.(*jobspb.Progress_Inspect)
+	if !ok {
+		return nil, errors.New("job is not an inspect job")
+	}
+
+	return progress.Inspect, nil
+}
+
 // sendInspectProgress marshals and sends inspect processor progress via BulkProcessorProgress.
 func (p *inspectProcessor) sendInspectProgress(
 	ctx context.Context, output execinfra.RowReceiver, checksCompleted int64, finished bool,
@@ -365,15 +446,18 @@ func (p *inspectProcessor) getTimestampForSpan() hlc.Timestamp {
 // 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,
+	ctx context.Context,
+	output execinfra.RowReceiver,
+	completedSpan roachpb.Span,
+	finished bool,
+	progressMsg *jobspb.InspectProcessorProgress,
 ) error {
 	if p.spansProcessedCtr != nil {
 		p.spansProcessedCtr.Inc(1)
 	}
-	progressMsg := &jobspb.InspectProcessorProgress{
-		ChecksCompleted: 0, // No additional checks completed, just marking span done
-		Finished:        finished,
-	}
+
+	progressMsg.ChecksCompleted = 0 // No additional checks completed, just marking span done
+	progressMsg.Finished = finished
 
 	progressAny, err := pbtypes.MarshalAny(progressMsg)
 	if err != nil {
@@ -466,7 +550,6 @@ func buildInspectCheckFactories(
 					asOf:         asOf,
 				}
 			})
-
 		default:
 			return nil, errors.AssertionFailedf("unsupported inspect check type: %v", specCheck.Type)
 		}