sql/inspect: support checking multiple indexes in one INSPECT job

Previously, there was an assert that made sure when building the INSPECT
job that we only had one check. This removes that assert so that we can
have multiple checks. When you run INSPECT against many indexes on a
table or against a database, we will have multiple checks, so it was
necessary for that case.

Closes #148300

Release note: none
Epic: none

Author: spilchen

pkg/sql/inspect/index_consistency_check_test.go

@@ -128,6 +128,9 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 		splitRangeDDL string
 		// indexDDL is the DDL to create the indexes on the table.
 		indexDDL []string
+		// corruptionTargetIndex specifies which secondary index to corrupt (0-based position).
+		// If not specified, defaults to 0 (first index).
+		corruptionTargetIndex int
 		// missingIndexEntrySelector defines a SQL predicate that selects rows
 		// whose secondary index entries will be manually deleted to simulate
 		// missing index entries (i.e., present in the primary index but not in the
@@ -247,6 +250,57 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 			danglingIndexEntryInsertQuery: "SELECT 15, 30, 300, 'corrupt', 'e_3', 300.5", // Add dangling entry after timestamp
 			useTimestampBeforeCorruption:  true,                                          // Use timestamp from before corruption
 		},
+		{
+			desc:          "2 indexes, corrupt second index, missing entry",
+			splitRangeDDL: "ALTER TABLE test.t SPLIT AT VALUES (500)",
+			indexDDL: []string{
+				"CREATE INDEX idx_t_a ON test.t (a)",
+				"CREATE INDEX idx_t_b ON test.t (b) STORING (e)",
+			},
+			corruptionTargetIndex:     1, // Target second index (idx_t_b)
+			missingIndexEntrySelector: "a = 7",
+			expectedIssues: []inspectIssue{
+				{
+					ErrorType:  "missing_secondary_index_entry",
+					PrimaryKey: "e'(7, \\'d_7\\')'",
+					Details: map[redact.RedactableString]interface{}{
+						"index_name": "idx_t_b",
+					},
+				},
+			},
+			expectedErrRegex: expectedInspectFoundInconsistencies,
+		},
+		{
+			desc:          "3 indexes, corrupt middle index, dangling entry",
+			splitRangeDDL: "ALTER TABLE test.t SPLIT AT VALUES (333),(666)",
+			indexDDL: []string{
+				"CREATE INDEX idx_t_a ON test.t (a)",
+				"CREATE INDEX idx_t_b ON test.t (b) STORING (c)",
+				"CREATE INDEX idx_t_c ON test.t (c) STORING (f)",
+			},
+			corruptionTargetIndex:         1, // Target second index (middle one)
+			danglingIndexEntryInsertQuery: "SELECT 25, 50, 500, 'corrupt_middle', 'e_25', 125.5",
+			expectedIssues: []inspectIssue{
+				{
+					ErrorType:  "dangling_secondary_index_entry",
+					PrimaryKey: "e'(25, \\'corrupt_middle\\')'",
+					Details: map[redact.RedactableString]interface{}{
+						"index_name": "idx_t_b",
+					},
+				},
+			},
+			expectedErrRegex: expectedInspectFoundInconsistencies,
+		},
+		{
+			desc: "multiple indexes, no corruption - all should be checked",
+			indexDDL: []string{
+				"CREATE INDEX idx_t_a ON test.t (a)",
+				"CREATE INDEX idx_t_b ON test.t (b)",
+				"CREATE INDEX idx_t_c ON test.t (c)",
+			},
+			// No corruptionTargetIndex specified, no corruption
+			missingIndexEntrySelector: "", // No corruption
+		},
 	} {
 		t.Run(tc.desc, func(t *testing.T) {
 			issueLogger.reset()
@@ -313,7 +367,14 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 			}
 
 			tableDesc := desctestutils.TestingGetPublicTableDescriptor(kvDB, codec, "test", "t")
-			secIndex := tableDesc.PublicNonPrimaryIndexes()[0]
+
+			// Select target index based on corruptionTargetIndex with bounds checking
+			indexes := tableDesc.PublicNonPrimaryIndexes()
+			targetIndexPos := tc.corruptionTargetIndex
+			if targetIndexPos < 0 || targetIndexPos >= len(indexes) {
+				targetIndexPos = 0 // Default to first index for safety/backward compatibility
+			}
+			secIndex := indexes[targetIndexPos]
 
 			// Apply test-specific corruption based on configured selectors:
 			// - If missingIndexEntrySelector is set, we delete the secondary index entries
@@ -438,6 +499,21 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 						}
 					}
 				}
+
+				// Validate Details if provided in expected issue
+				if expectedIssue.Details != nil {
+					require.NotNil(t, foundIssue.Details, "issue should have details when expected")
+
+					// Check that all expected detail keys and values match
+					for expectedKey, expectedValue := range expectedIssue.Details {
+						require.Contains(t, foundIssue.Details, expectedKey,
+							"issue should contain detail key: %s", expectedKey)
+
+						actualValue := foundIssue.Details[expectedKey]
+						require.Equal(t, expectedValue, actualValue,
+							"detail %s should be %v, got %v", expectedKey, expectedValue, actualValue)
+					}
+				}
 			}
 
 			// Validate job status matches expected outcome

pkg/sql/inspect/inspect_job.go

@@ -13,14 +13,14 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/sql/physicalplan"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
-	"github.com/cockroachdb/errors"
 )
 
 type inspectResumer struct {
@@ -84,10 +84,17 @@ func (c *inspectResumer) getPrimaryIndexSpans(
 ) ([]roachpb.Span, error) {
 	details := c.job.Details().(jobspb.InspectDetails)
 
-	spans := make([]roachpb.Span, 0, len(details.Checks))
+	// Deduplicate by table ID to avoid processing the same span multiple times
+	// when there are multiple checks on the same table.
+	uniqueTableIDs := make(map[descpb.ID]struct{})
+	for i := range details.Checks {
+		uniqueTableIDs[details.Checks[i].TableID] = struct{}{}
+	}
+
+	spans := make([]roachpb.Span, 0, len(uniqueTableIDs))
 	err := execCfg.InternalDB.DescsTxn(ctx, func(ctx context.Context, txn descs.Txn) error {
-		for i := range details.Checks {
-			desc, err := txn.Descriptors().ByIDWithLeased(txn.KV()).WithoutNonPublic().Get().Table(ctx, details.Checks[i].TableID)
+		for tableID := range uniqueTableIDs {
+			desc, err := txn.Descriptors().ByIDWithLeased(txn.KV()).WithoutNonPublic().Get().Table(ctx, tableID)
 			if err != nil {
 				return err
 			}
@@ -104,9 +111,6 @@ func (c *inspectResumer) getPrimaryIndexSpans(
 func (c *inspectResumer) planInspectProcessors(
 	ctx context.Context, jobExecCtx sql.JobExecContext, entirePKSpans []roachpb.Span,
 ) (*sql.PhysicalPlan, *sql.PlanningCtx, error) {
-	if len(entirePKSpans) > 1 {
-		return nil, nil, errors.AssertionFailedf("we only support one check: %d", len(entirePKSpans))
-	}
 	distSQLPlanner := jobExecCtx.DistSQLPlanner()
 	planCtx, _, err := distSQLPlanner.SetupAllNodesPlanning(ctx, jobExecCtx.ExtendedEvalContext(), jobExecCtx.ExecCfg())
 	if err != nil {

pkg/sql/inspect/inspect_processor.go

@@ -257,13 +257,15 @@ func buildInspectCheckFactories(
 ) ([]inspectCheckFactory, error) {
 	checkFactories := make([]inspectCheckFactory, 0, len(spec.InspectDetails.Checks))
 	for _, specCheck := range spec.InspectDetails.Checks {
+		tableID := specCheck.TableID
+		indexID := specCheck.IndexID
 		switch specCheck.Type {
 		case jobspb.InspectCheckIndexConsistency:
 			checkFactories = append(checkFactories, func() inspectCheck {
 				return &indexConsistencyCheck{
 					flowCtx: flowCtx,
-					tableID: specCheck.TableID,
-					indexID: specCheck.IndexID,
+					tableID: tableID,
+					indexID: indexID,
 					asOf:    spec.InspectDetails.AsOf,
 				}
 			})

pkg/sql/scrub.go

@@ -497,27 +497,29 @@ func TriggerInspectJob(
 	// Consistency check is done async via a job.
 	jobID := execCfg.JobRegistry.MakeJobID()
 
-	// TODO(148300): just grab the first secondary index and use that for the
-	// consistency check.
-	// TODO(148365): When INSPECT is added, we want to skip unsupported indexes
-	// and return a NOTICE.
 	secIndexes := tableDesc.PublicNonPrimaryIndexes()
 	if len(secIndexes) == 0 {
 		return jobID, errors.AssertionFailedf("must have at least one secondary index")
 	}
 
+	// Create checks for all secondary indexes
+	// TODO(148365): When INSPECT is added, we want to skip unsupported indexes
+	// and return a NOTICE.
+	checks := make([]*jobspb.InspectDetails_Check, 0, len(secIndexes))
+	for _, index := range secIndexes {
+		checks = append(checks, &jobspb.InspectDetails_Check{
+			Type:    jobspb.InspectCheckIndexConsistency,
+			TableID: tableDesc.GetID(),
+			IndexID: index.GetID(),
+		})
+	}
+
 	// TODO(sql-queries): add row count check when that is implemented.
 	jr := jobs.Record{
 		Description: jobRecordDescription,
 		Details: jobspb.InspectDetails{
-			Checks: []*jobspb.InspectDetails_Check{
-				{
-					Type:    jobspb.InspectCheckIndexConsistency,
-					TableID: tableDesc.GetID(),
-					IndexID: secIndexes[0].GetID(),
-				},
-			},
-			AsOf: asOf,
+			Checks: checks,
+			AsOf:   asOf,
 		},
 		Progress:      jobspb.InspectProgress{},
 		CreatedBy:     nil,