sql/inspect: convert internal errors to inspect issues

Previously, internal errors during index consistency checks would fail the
entire job. Now these errors are converted to structured inspect issues with
detailed context.

Closes #148299

Release note: none

Epic: None

Author: spilchen

pkg/sql/inspect/index_consistency_check.go

@@ -49,6 +49,10 @@ type indexConsistencyCheck struct {
 	// queries join. The actual resulting rows from the RowContainer is
 	// twice this plus the error_type column.
 	columns []catalog.Column
+
+	// lastQuery stores the SQL query executed for this check to help
+	// debug internal errors by providing context about the span bounds.
+	lastQuery string
 }
 
 var _ inspectCheck = (*indexConsistencyCheck)(nil)
@@ -190,6 +194,9 @@ func (c *indexConsistencyCheck) Start(
 		colNames(pkColumns), colNames(otherColumns), c.tableDesc.GetID(), c.secIndex, c.priIndex.GetID(), predicate,
 	)
 
+	// Store the query for error reporting
+	c.lastQuery = checkQuery
+
 	it, err := c.flowCtx.Cfg.DB.Executor().QueryIteratorEx(
 		ctx, "inspect-index-consistency-check", nil, /* txn */
 		sessiondata.InternalExecutorOverride{
@@ -221,7 +228,29 @@ func (c *indexConsistencyCheck) Next(
 
 	ok, err := c.rowIter.Next(ctx)
 	if err != nil {
-		return nil, errors.Wrap(err, "fetching next row in index consistency check")
+		// Close the iterator to prevent further usage. The close may emit the
+		// internal error too, but we only need to capture it once.
+		_ = c.Close(ctx)
+		c.exhaustedIter = true
+
+		// Convert internal errors to inspect issues rather than failing the entire job.
+		// This allows us to capture and log data corruption or encoding errors as
+		// structured issues for investigation.
+		details := make(map[redact.RedactableString]interface{})
+		details["error_message"] = err.Error()
+		details["error_type"] = "internal_query_error"
+		details["index_name"] = c.secIndex.GetName()
+		details["query"] = c.lastQuery // Store the query that caused the error
+
+		return &inspectIssue{
+			ErrorType: InternalError,
+			// TODO(148573): Use the timestamp that we create a protected timestamp for.
+			AOST:       timeutil.Now(),
+			DatabaseID: c.tableDesc.GetParentID(),
+			SchemaID:   c.tableDesc.GetParentSchemaID(),
+			ObjectID:   c.tableDesc.GetID(),
+			Details:    details,
+		}, nil
 	}
 	if !ok {
 		c.exhaustedIter = true
@@ -303,10 +332,12 @@ func (c *indexConsistencyCheck) Done(context.Context) bool {
 // Close implements the inspectCheck interface.
 func (c *indexConsistencyCheck) Close(context.Context) error {
 	if c.rowIter != nil {
-		if err := c.rowIter.Close(); err != nil {
+		// Clear the iter ahead of close to ensure we only attempt the close once.
+		it := c.rowIter
+		c.rowIter = nil
+		if err := it.Close(); err != nil {
 			return errors.Wrap(err, "closing index consistency check iterator")
 		}
-		c.rowIter = nil
 	}
 	return nil
 }

pkg/sql/inspect/index_consistency_check_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/redact"
 	"github.com/stretchr/testify/require"
 )
 
@@ -144,6 +145,10 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 		// expectedErrRegex is the regex pattern that the error message should match.
 		// If empty then no error is expected.
 		expectedErrRegex string
+		// expectedInternalErrorPatterns contains patterns for validating internal error details.
+		// Each element corresponds to the issue at the same index in expectedIssues.
+		// For non-internal-error issues, the corresponding element should be nil.
+		expectedInternalErrorPatterns []map[string]string
 	}{
 		{
 			desc: "happy path sanity",
@@ -192,16 +197,24 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 			expectedErrRegex: expectedInspectFoundInconsistencies,
 		},
 		{
-			desc:          "2 ranges, secondary index on 'a', 1 dangling entry",
+			desc:          "2 ranges, secondary index on 'a', 1 dangling entry with internal error",
 			splitRangeDDL: "ALTER TABLE test.t SPLIT AT VALUES (500)",
 			indexDDL: []string{
 				"CREATE INDEX idx_t_a ON test.t (a) STORING (f)",
 			},
 			danglingIndexEntryInsertQuery: "SELECT 3, 30, 300, 'd_3', 'e_3', -56.712",
-			// TODO(148299): We eventually want to detect internal errors. Currently
-			// this test passes by expecting the internal error pattern rather than
-			// detecting the dangling entry.
-			expectedErrRegex: "error decoding 6 bytes: float64 value should be exactly 8 bytes: 5",
+			expectedIssues: []inspectIssue{
+				{ErrorType: "internal_error"},
+			},
+			expectedErrRegex: expectedInspectFoundInconsistencies,
+			expectedInternalErrorPatterns: []map[string]string{
+				{
+					"error_message": "error decoding.*float64",
+					"error_type":    "internal_query_error",
+					"index_name":    "idx_t_a",
+					"query":         "FROM.*table_",
+				},
+			},
 		},
 		{
 			desc:          "2 ranges, secondary index on 'b' storing 'f', 1 dangling entry",
@@ -368,6 +381,28 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 				require.NotEqual(t, 0, foundIssue.SchemaID, "expected issue to have a schema ID: %s", expectedIssue)
 				require.NotEqual(t, 0, foundIssue.ObjectID, "expected issue to have an object ID: %s", expectedIssue)
 				require.NotEqual(t, time.Time{}, foundIssue.AOST, "expected issue to have an AOST time: %s", expectedIssue)
+
+				// Additional validation for internal errors
+				if foundIssue.ErrorType == "internal_error" {
+					require.NotNil(t, foundIssue.Details, "internal error should have details")
+
+					// Validate patterns if provided for this specific issue
+					if tc.expectedInternalErrorPatterns != nil && i < len(tc.expectedInternalErrorPatterns) &&
+						tc.expectedInternalErrorPatterns[i] != nil {
+						expectedPatterns := tc.expectedInternalErrorPatterns[i]
+
+						// Validate each expected pattern
+						for detailKey, expectedPattern := range expectedPatterns {
+							redactableKey := redact.RedactableString(detailKey)
+							require.Contains(t, foundIssue.Details, redactableKey, "internal error should contain detail key: %s", detailKey)
+
+							detailValue, ok := foundIssue.Details[redactableKey].(string)
+							require.True(t, ok, "detail value for key %s should be a string", detailKey)
+							require.Regexp(t, expectedPattern, detailValue,
+								"detail %s should match pattern %s, got: %s", detailKey, expectedPattern, detailValue)
+						}
+					}
+				}
 			}
 
 			// Validate job status matches expected outcome

pkg/sql/inspect/issue.go

@@ -97,4 +97,9 @@ const (
 	// DanglingSecondaryIndexEntry occurs when a secondary index entry exists
 	// but the corresponding row in the primary index is missing.
 	DanglingSecondaryIndexEntry inspectErrorType = "dangling_secondary_index_entry"
+
+	// InternalError occurs when an internal error (e.g., data corruption, encoding
+	// issues) prevents the check from completing normally. These errors indicate
+	// potential data corruption or other serious issues that require investigation.
+	InternalError inspectErrorType = "internal_error"
 )