sql: add support for span checks and cluster 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 span-level (dependent on other
checks running on the spans) and cluster-level checks (run after all spans have
been processed).

Part of: #155472
Epic: CRDB-55075

Release note: None

Author: bghal

pkg/sql/inspect/index_consistency_check.go

@@ -58,20 +58,9 @@ 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
-)
+func (c *indexConsistencyCheckApplicability) IsSpanLevel() bool {
+	return false
+}
 
 // indexConsistencyCheck verifies consistency between a table's primary index
 // and a specified secondary index by streaming rows from both sides of a

pkg/sql/inspect/inspect_processor.go

@@ -274,7 +274,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)
@@ -322,8 +322,20 @@ func (p *inspectProcessor) processSpan(
 		}
 	}
 
+	// Process the span-level checks.
+	progressMsg := &jobspb.InspectProcessorProgress{}
+	for _, check := range checks {
+		if check, ok := check.(inspectSpanCheck); ok {
+			progressMsg.ChecksCompleted++
+			err := check.CheckSpan(ctx, checks, p.loggerBundle, progressMsg)
+			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, progressMsg); err != nil {
 		return err
 	}
 
@@ -371,15 +383,14 @@ 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,
+	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,
-	}
 
 	progressAny, err := pbtypes.MarshalAny(progressMsg)
 	if err != nil {
@@ -393,7 +404,7 @@ func (p *inspectProcessor) sendSpanCompletionProgress(
 			NodeID:          p.flowCtx.NodeID.SQLInstanceID(),
 			FlowID:          p.flowCtx.ID,
 			ProcessorID:     p.processorID,
-			Drained:         finished,
+			Drained:         progressMsg.Finished,
 		},
 	}
 
@@ -423,7 +434,7 @@ func newInspectProcessor(
 ) (execinfra.Processor, error) {
 	execCfg := flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig)
 
-	checkFactories, err := buildInspectCheckFactories(execCfg, spec)
+	checkFactories, err := buildInspectCheckFactories(execCfg, spec.InspectDetails)
 	if err != nil {
 		return nil, err
 	}
@@ -438,7 +449,7 @@ func newInspectProcessor(
 		checkFactories:      checkFactories,
 		cfg:                 flowCtx.Cfg,
 		spanSrc:             newSliceSpanSource(spec.Spans),
-		loggerBundle:        getInspectLogger(execCfg, spec.JobID),
+		loggerBundle:        getInspectLogger(flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig), spec.JobID),
 		concurrency:         getProcessorConcurrency(flowCtx),
 		clock:               flowCtx.Cfg.DB.KV().Clock(),
 		spansProcessedCtr:   spansProcessedCtr,
@@ -453,10 +464,10 @@ func newInspectProcessor(
 // This indirection ensures that each check instance is freshly created per span,
 // avoiding shared state across concurrent workers.
 func buildInspectCheckFactories(
-	execCfg *sql.ExecutorConfig, spec execinfrapb.InspectSpec,
+	execCfg *sql.ExecutorConfig, details jobspb.InspectDetails,
 ) ([]inspectCheckFactory, error) {
-	checkFactories := make([]inspectCheckFactory, 0, len(spec.InspectDetails.Checks))
-	for _, specCheck := range spec.InspectDetails.Checks {
+	checkFactories := make([]inspectCheckFactory, 0, len(details.Checks))
+	for _, specCheck := range details.Checks {
 		tableID := specCheck.TableID
 		indexID := specCheck.IndexID
 		tableVersion := specCheck.TableVersion
@@ -473,7 +484,6 @@ func buildInspectCheckFactories(
 					asOf:         asOf,
 				}
 			})
-
 		default:
 			return nil, errors.AssertionFailedf("unsupported inspect check type: %v", specCheck.Type)
 		}

pkg/sql/inspect/inspect_processor_test.go

@@ -7,7 +7,6 @@ package inspect
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -24,6 +23,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
+	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 )
 
@@ -126,6 +126,10 @@ func (t *testingInspectCheck) AppliesTo(codec keys.SQLCodec, span roachpb.Span)
 	return true, nil
 }
 
+func (c *testingInspectCheck) IsSpanLevel() bool {
+	return false
+}
+
 // Started implements the inspectCheck interface.
 func (t *testingInspectCheck) Started() bool {
 	return t.started
@@ -205,6 +209,73 @@ func (t *testingInspectCheck) Close(context.Context) error {
 	return nil
 }
 
+// testingInspectCheckSpan is a test implementation of inspectCheckSpan.
+type testingInspectCheckSpan struct {
+	started bool
+
+	spanIssues int64
+}
+
+var _ inspectCheckApplicability = (*testingInspectCheckSpan)(nil)
+
+var _ inspectSpanCheck = (*testingInspectCheckSpan)(nil)
+
+func (t *testingInspectCheckSpan) AppliesTo(codec keys.SQLCodec, span roachpb.Span) (bool, error) {
+	// For testing, assume all checks apply to all spans
+	return true, nil
+}
+
+func (c *testingInspectCheckSpan) IsSpanLevel() bool {
+	return true
+}
+
+func (t *testingInspectCheckSpan) AppliesToCluster() (bool, error) {
+	// For testing it runs.
+	return true, nil
+}
+
+func (t *testingInspectCheckSpan) Started() bool {
+	return t.started
+}
+
+func (t *testingInspectCheckSpan) Start(
+	ctx context.Context, cfg *execinfra.ServerConfig, span roachpb.Span, workerIndex int,
+) error {
+	t.started = true
+	return nil
+}
+
+func (*testingInspectCheckSpan) Next(
+	ctx context.Context, cfg *execinfra.ServerConfig,
+) (*inspectIssue, error) {
+	return nil, nil
+}
+
+func (*testingInspectCheckSpan) Done(ctx context.Context) bool {
+	return true
+}
+
+func (*testingInspectCheckSpan) Close(ctx context.Context) error {
+	return nil
+}
+
+func (t *testingInspectCheckSpan) RegisterChecksForSpan(checks inspectChecks) error {
+	return nil
+}
+
+func (t *testingInspectCheckSpan) CheckSpan(
+	ctx context.Context, logger *inspectLoggerBundle, msg *jobspb.InspectProcessorProgress,
+) error {
+	for i := t.spanIssues; i > 0; i-- {
+		err := logger.logIssue(ctx, &inspectIssue{})
+		if err != nil {
+			return errors.Wrapf(err, "error logging inspect issue")
+		}
+	}
+
+	return nil
+}
+
 // discardRowReceiver is a minimal RowReceiver that discards all data.
 type discardRowReceiver struct{}
 
@@ -249,7 +320,7 @@ func runProcessorAndWait(t *testing.T, proc *inspectProcessor, expectErr bool) {
 // makeProcessor will create an inspect processor for test.
 func makeProcessor(
 	t *testing.T,
-	checkFactory inspectCheckFactory,
+	checkFactories []inspectCheckFactory,
 	src spanSource,
 	concurrency int,
 	asOf hlc.Timestamp,
@@ -274,7 +345,7 @@ func makeProcessor(
 				AsOf: asOf,
 			},
 		},
-		checkFactories: []inspectCheckFactory{checkFactory},
+		checkFactories: checkFactories,
 		cfg:            flowCtx.Cfg,
 		flowCtx:        flowCtx,
 		spanSrc:        src,
@@ -358,12 +429,14 @@ func TestInspectProcessor_ControlFlow(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.desc, func(t *testing.T) {
-			factory := func(asOf hlc.Timestamp) inspectCheck {
-				return &testingInspectCheck{
-					configs: tc.configs,
-				}
+			factories := []inspectCheckFactory{
+				func(asOf hlc.Timestamp) inspectCheck {
+					return &testingInspectCheck{
+						configs: tc.configs,
+					}
+				},
 			}
-			proc, _ := makeProcessor(t, factory, tc.spanSrc, len(tc.configs), hlc.Timestamp{})
+			proc, _ := makeProcessor(t, factories, tc.spanSrc, len(tc.configs), hlc.Timestamp{})
 			runProcessorAndWait(t, proc, tc.expectErr)
 		})
 	}
@@ -377,20 +450,22 @@ func TestInspectProcessor_EmitIssues(t *testing.T) {
 		mode:     spanModeNormal,
 		maxSpans: 1,
 	}
-	factory := func(asOf hlc.Timestamp) inspectCheck {
-		return &testingInspectCheck{
-			configs: []testingCheckConfig{
-				{
-					mode: checkModeNone,
-					issues: []*inspectIssue{
-						{ErrorType: "test_error", PrimaryKey: "pk1"},
-						{ErrorType: "test_error", PrimaryKey: "pk2"},
+	factories := []inspectCheckFactory{
+		func(asOf hlc.Timestamp) inspectCheck {
+			return &testingInspectCheck{
+				configs: []testingCheckConfig{
+					{
+						mode: checkModeNone,
+						issues: []*inspectIssue{
+							{ErrorType: "test_error", PrimaryKey: "pk1"},
+							{ErrorType: "test_error", PrimaryKey: "pk2"},
+						},
 					},
 				},
-			},
-		}
+			}
+		},
 	}
-	proc, logger := makeProcessor(t, factory, spanSrc, 1, hlc.Timestamp{})
+	proc, logger := makeProcessor(t, factories, spanSrc, 1, hlc.Timestamp{})
 
 	runProcessorAndWait(t, proc, true /* expectErr */)
 
@@ -439,21 +514,23 @@ func TestInspectProcessor_AsOfTime(t *testing.T) {
 			testStartTime := time.Now()
 
 			var capturedTimestamp hlc.Timestamp
-			factory := func(asOf hlc.Timestamp) inspectCheck {
-				capturedTimestamp = asOf
-				return &testingInspectCheck{
-					configs: []testingCheckConfig{
-						{
-							mode: checkModeNone,
-							issues: []*inspectIssue{
-								{ErrorType: "test_error", PrimaryKey: "pk1", AOST: asOf.GoTime()},
+			factories := []inspectCheckFactory{
+				func(asOf hlc.Timestamp) inspectCheck {
+					capturedTimestamp = asOf
+					return &testingInspectCheck{
+						configs: []testingCheckConfig{
+							{
+								mode: checkModeNone,
+								issues: []*inspectIssue{
+									{ErrorType: "test_error", PrimaryKey: "pk1", AOST: asOf.GoTime()},
+								},
 							},
 						},
-					},
-				}
+					}
+				},
 			}
 
-			proc, logger := makeProcessor(t, factory, spanSrc, 1, tc.asOf)
+			proc, logger := makeProcessor(t, factories, spanSrc, 1, tc.asOf)
 
 			runProcessorAndWait(t, proc, true /* expectErr */)
 
@@ -465,3 +542,25 @@ func TestInspectProcessor_AsOfTime(t *testing.T) {
 		})
 	}
 }
+
+func TestInspectProcessor_SpanCheck(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	spanSrc := &testingSpanSource{
+		mode:     spanModeNormal,
+		maxSpans: 1,
+	}
+	factories := []inspectCheckFactory{
+		func(asOf hlc.Timestamp) inspectCheck {
+			return &testingInspectCheckSpan{
+				spanIssues: 2,
+			}
+		},
+	}
+	proc, logger := makeProcessor(t, factories, spanSrc, 1, hlc.Timestamp{})
+
+	runProcessorAndWait(t, proc, true /* expectErr */)
+
+	require.Equal(t, 2, logger.numIssuesFound())
+}