sql/inspect: skip INSPECT checks for non-applicable spans earlier

INSPECT jobs generate a set of checks along with a separate list of spans,
which only apply to some of the checks. Previously, filtering to ensure a span
applied to a check happened late in the process.

This commit moves that filtering earlier, so checks immediately skip spans
that are not relevant. This improves efficiency and prepares the system for
upcoming progress tracking logic, which will be added in a separate PR.

Informs: #148297
Epic: CRDB-30356

Release note: none

Author: spilchen

pkg/sql/inspect/BUILD.bazel

@@ -18,6 +18,7 @@ go_library(
     deps = [
         "//pkg/jobs",
         "//pkg/jobs/jobspb",
+        "//pkg/keys",
         "//pkg/kv/kvserver/protectedts/ptpb",
         "//pkg/roachpb",
         "//pkg/security/username",

pkg/sql/inspect/index_consistency_check.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
@@ -30,13 +31,28 @@ import (
 	"github.com/cockroachdb/redact"
 )
 
-// indexConsistencyCheck verifies consistency between a table’s primary index
+// indexConsistencyCheckApplicability is a lightweight version that only implements applicability logic.
+type indexConsistencyCheckApplicability struct {
+	tableID descpb.ID
+}
+
+var _ inspectCheckApplicability = (*indexConsistencyCheckApplicability)(nil)
+
+// AppliesTo implements the inspectCheckApplicability interface.
+func (c *indexConsistencyCheckApplicability) AppliesTo(
+	codec keys.SQLCodec, span roachpb.Span,
+) (bool, error) {
+	return spanContainsTable(c.tableID, codec, span)
+}
+
+// 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
 // secondary, or vice versa.
 type indexConsistencyCheck struct {
+	indexConsistencyCheckApplicability
+
 	flowCtx *execinfra.FlowCtx
-	tableID descpb.ID
 	indexID descpb.IndexID
 	asOf    hlc.Timestamp
 
@@ -57,6 +73,7 @@ type indexConsistencyCheck struct {
 }
 
 var _ inspectCheck = (*indexConsistencyCheck)(nil)
+var _ inspectCheckApplicability = (*indexConsistencyCheck)(nil)
 
 // Started implements the inspectCheck interface.
 func (c *indexConsistencyCheck) Started() bool {
@@ -67,14 +84,9 @@ func (c *indexConsistencyCheck) Started() bool {
 func (c *indexConsistencyCheck) Start(
 	ctx context.Context, cfg *execinfra.ServerConfig, span roachpb.Span, workerIndex int,
 ) error {
-	// Early out for spans that don't apply to the table we are running the check on.
-	_, tableID, err := cfg.Codec.DecodeTablePrefix(span.Key)
-	if err != nil {
+	if err := assertCheckApplies(c, cfg.Codec, span); err != nil {
 		return err
 	}
-	if descpb.ID(tableID) != c.tableID {
-		return nil
-	}
 
 	// Load up the index and table descriptors.
 	if err := c.loadCatalogInfo(ctx); err != nil {

pkg/sql/inspect/inspect_processor.go

@@ -190,10 +190,24 @@ func getInspectLogger(flowCtx *execinfra.FlowCtx, jobID jobspb.JobID) inspectLog
 func (p *inspectProcessor) processSpan(
 	ctx context.Context, span roachpb.Span, workerIndex int,
 ) (err error) {
-	checks := make([]inspectCheck, len(p.checkFactories))
-	for i, factory := range p.checkFactories {
-		checks[i] = factory()
+	// Only create checks that apply to this span
+	var checks []inspectCheck
+	for _, factory := range p.checkFactories {
+		check := factory()
+		applies, err := check.AppliesTo(p.cfg.Codec, span)
+		if err != nil {
+			return err
+		}
+		if applies {
+			checks = append(checks, check)
+		}
 	}
+
+	// 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,
@@ -263,8 +277,10 @@ func buildInspectCheckFactories(
 		case jobspb.InspectCheckIndexConsistency:
 			checkFactories = append(checkFactories, func() inspectCheck {
 				return &indexConsistencyCheck{
+					indexConsistencyCheckApplicability: indexConsistencyCheckApplicability{
+						tableID: tableID,
+					},
 					flowCtx: flowCtx,
-					tableID: tableID,
 					indexID: indexID,
 					asOf:    spec.InspectDetails.AsOf,
 				}

pkg/sql/inspect/inspect_processor_test.go

@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
@@ -112,6 +113,13 @@ type testingInspectCheck struct {
 }
 
 var _ inspectCheck = (*testingInspectCheck)(nil)
+var _ inspectCheckApplicability = (*testingInspectCheck)(nil)
+
+// AppliesTo implements the inspectCheckApplicability interface.
+func (t *testingInspectCheck) AppliesTo(codec keys.SQLCodec, span roachpb.Span) (bool, error) {
+	// For testing, assume all checks apply to all spans
+	return true, nil
+}
 
 // Started implements the inspectCheck interface.
 func (t *testingInspectCheck) Started() bool {

pkg/sql/inspect/runner.go

@@ -8,11 +8,35 @@ package inspect
 import (
 	"context"
 
+	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/errors"
 )
 
+// inspectCheckApplicability defines the interface for determining if a check
+// applies to a span.
+type inspectCheckApplicability interface {
+	// AppliesTo reports whether this check applies to the given span.
+	AppliesTo(codec keys.SQLCodec, span roachpb.Span) (bool, error)
+}
+
+// assertCheckApplies is a helper that calls AppliesTo and asserts the check applies.
+func assertCheckApplies(
+	check inspectCheckApplicability, codec keys.SQLCodec, span roachpb.Span,
+) error {
+	applies, err := check.AppliesTo(codec, span)
+	if err != nil {
+		return err
+	}
+	if !applies {
+		return errors.AssertionFailedf(
+			"check does not apply to this span: span=%s", span.String())
+	}
+	return nil
+}
+
 // inspectCheck defines a single validation operation used by the INSPECT system.
 // Each check represents a specific type of data validation, such as index consistency.
 //
@@ -25,6 +49,8 @@ import (
 // under the hood to detect inconsistencies. All results are surfaced through the
 // inspectIssue type.
 type inspectCheck interface {
+	inspectCheckApplicability
+
 	// Started reports whether the check has been initialized.
 	Started() bool
 
@@ -117,3 +143,12 @@ func (c *inspectRunner) Close(ctx context.Context) error {
 	}
 	return retErr
 }
+
+// spanContainsTable checks if the given span contains data for the specified table.
+func spanContainsTable(tableID descpb.ID, codec keys.SQLCodec, span roachpb.Span) (bool, error) {
+	_, spanTableID, err := codec.DecodeTablePrefix(span.Key)
+	if err != nil {
+		return false, err
+	}
+	return descpb.ID(spanTableID) == tableID, nil
+}

pkg/sql/inspect/runner_test.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"testing"
 
+	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
@@ -28,6 +29,13 @@ type mockInspectCheck struct {
 }
 
 var _ inspectCheck = &mockInspectCheck{}
+var _ inspectCheckApplicability = &mockInspectCheck{}
+
+// AppliesTo implements the inspectCheckApplicability interface.
+func (m *mockInspectCheck) AppliesTo(codec keys.SQLCodec, span roachpb.Span) (bool, error) {
+	// For testing, assume all checks apply to all spans
+	return true, nil
+}
 
 func (m *mockInspectCheck) Started() bool {
 	return m.started