sql/inspect: introduce interface for INSPECT check infrastructure

Add a new interface to support INSPECT operations that perform checks emitting
rows for each issue found. This interface follows a row iterator pattern,
enabling future implementation of specific checks.

Informs: #146863
Epic: CRDB-30356
Release note: none

Author: spilchen

pkg/sql/inspect/BUILD.bazel

@@ -5,6 +5,9 @@ go_library(
     srcs = [
         "inspect_job.go",
         "inspect_processor.go",
+        "issue.go",
+        "log_sink.go",
+        "runner.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/sql/inspect",
     visibility = ["//visibility:public"],
@@ -14,6 +17,7 @@ go_library(
         "//pkg/roachpb",
         "//pkg/settings/cluster",
         "//pkg/sql",
+        "//pkg/sql/catalog/descpb",
         "//pkg/sql/catalog/descs",
         "//pkg/sql/execinfra",
         "//pkg/sql/execinfrapb",
@@ -25,28 +29,35 @@ go_library(
         "//pkg/util/log",
         "//pkg/util/tracing",
         "@com_github_cockroachdb_errors//:errors",
+        "@com_github_cockroachdb_redact//:redact",
     ],
 )
 
 go_test(
     name = "inspect_test",
     srcs = [
         "inspect_job_test.go",
+        "issue_test.go",
         "main_test.go",
+        "runner_test.go",
     ],
+    embed = [":inspect"],
     deps = [
         "//pkg/base",
+        "//pkg/roachpb",
         "//pkg/security/securityassets",
         "//pkg/security/securitytest",
         "//pkg/server",
         "//pkg/sql",
+        "//pkg/sql/execinfra",
         "//pkg/testutils",
         "//pkg/testutils/serverutils",
         "//pkg/testutils/sqlutils",
         "//pkg/testutils/testcluster",
         "//pkg/util/leaktest",
         "//pkg/util/log",
         "@com_github_cockroachdb_errors//:errors",
+        "@com_github_cockroachdb_redact//:redact",
         "@com_github_stretchr_testify//require",
     ],
 )

pkg/sql/inspect/issue.go

@@ -0,0 +1,67 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package inspect
+
+import (
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
+	"github.com/cockroachdb/redact"
+)
+
+// inspectIssue represents a single validation failure detected by an inspectCheck.
+// These issues correspond to rows written into the system.inspect_errors table.
+//
+// Each issue identifies the object where the inconsistency was found, the specific
+// problem type, and a JSON-serializable details map with check-specific context.
+type inspectIssue struct {
+	// ErrorType is a machine-readable string describing the type of validation failure
+	ErrorType redact.RedactableString
+
+	// DatabaseID is the descriptor ID of the database containing the object in error.
+	// May be 0 if not applicable.
+	DatabaseID descpb.ID
+
+	// SchemaID is the descriptor ID of the schema containing the object in error.
+	// May be 0 if not applicable.
+	SchemaID descpb.ID
+
+	// ObjectID is the descriptor ID of the thing where the error occurred.
+	// Usually this is the ID of the table.
+	ObjectID descpb.ID
+
+	// PrimaryKey is the primary key of the row involved in the issue, rendered as a string.
+	// If the error does not relate to a specific row, this may be empty.
+	PrimaryKey string
+
+	// Details contains additional structured metadata describing the issue.
+	// The contents vary by check type.
+	Details map[redact.RedactableString]interface{}
+}
+
+var _ redact.SafeFormatter = (*inspectIssue)(nil)
+
+// SafeFormat implements the redact.SafeFormatter interface.
+func (i inspectIssue) SafeFormat(w redact.SafePrinter, _ rune) {
+	var buf redact.StringBuilder
+	buf.Printf("{type=")
+	buf.Print(i.ErrorType)
+	if i.DatabaseID != 0 {
+		buf.Printf(" db=%d", i.DatabaseID)
+	}
+	if i.SchemaID != 0 {
+		buf.Printf(" schema=%d", i.SchemaID)
+	}
+	if i.ObjectID != 0 {
+		buf.Printf(" obj=%d", i.ObjectID)
+	}
+	buf.Printf(" pk=%q", i.PrimaryKey)
+	if i.Details != nil {
+		buf.Printf(" details=%+v", i.Details)
+	}
+	buf.Printf("}")
+	w.Print(buf)
+}
+
+func (i inspectIssue) String() string { return redact.StringWithoutMarkers(i) }

pkg/sql/inspect/issue_test.go

@@ -0,0 +1,70 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package inspect
+
+import (
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/redact"
+	"github.com/stretchr/testify/require"
+)
+
+func TestIssueRedaction(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	testCases := []struct {
+		desc       string
+		issue      *inspectIssue
+		redactable string
+		redacted   string
+	}{
+		{
+			desc: "no details",
+			issue: &inspectIssue{
+				ErrorType:  "testing_error",
+				DatabaseID: 10,
+				ObjectID:   101,
+				PrimaryKey: "/Table/101/c=10",
+			},
+			redactable: "{type=testing_error db=10 obj=101 pk=‹\"/Table/101/c=10\"›}",
+			redacted:   "{type=testing_error db=10 obj=101 pk=‹×›}",
+		},
+		{
+			desc: "with details",
+			issue: &inspectIssue{
+				ErrorType:  "testing_system_error",
+				SchemaID:   100,
+				PrimaryKey: "/System/100/",
+				Details: map[redact.RedactableString]interface{}{
+					"field1": 10,
+					"field2": "strval",
+				},
+			},
+			redactable: "{type=testing_system_error schema=100 pk=‹\"/System/100/\"› details=map[field1:10 field2:‹strval›]}",
+			redacted:   "{type=testing_system_error schema=100 pk=‹×› details=map[field1:10 field2:‹×›]}",
+		},
+		{
+			desc: "details that has a redactable string value",
+			issue: &inspectIssue{
+				ErrorType: "testing_system_error",
+				Details: map[redact.RedactableString]interface{}{
+					"field1": redact.RedactableString("redactable string"),
+				},
+			},
+			redactable: "{type=testing_system_error pk=‹\"\"› details=map[field1:redactable string]}",
+			redacted:   "{type=testing_system_error pk=‹×› details=map[field1:redactable string]}",
+		},
+	}
+	for _, c := range testCases {
+		t.Run(c.desc, func(t *testing.T) {
+			require.EqualValues(t, c.redactable, redact.Sprint(c.issue))
+			require.EqualValues(t, c.redacted, redact.Sprint(c.issue).Redact())
+		})
+	}
+}

pkg/sql/inspect/log_sink.go

@@ -0,0 +1,23 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package inspect
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+)
+
+// logSink will report any inspect errors directly to cockroach.log.
+type logSink struct{}
+
+var _ inspectLogger = &logSink{}
+
+// logIssue implements the inspectLogger interface.
+func (c *logSink) logIssue(ctx context.Context, issue *inspectIssue) error {
+	log.Errorf(ctx, "inspect issue: %+v", issue)
+	return nil
+}

pkg/sql/inspect/runner.go

@@ -0,0 +1,108 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package inspect
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
+	"github.com/cockroachdb/errors"
+)
+
+// inspectCheck defines a single validation operation used by the INSPECT system.
+// Each check represents a specific type of data validation, such as index consistency.
+//
+// A check is stateful. It must be initialized with Start(), which prepares it to
+// produce results. Once started, repeated calls to Next() yield zero or more
+// inspectIssue results until Done() returns true. After completion, Close() releases
+// any associated resources.
+//
+// Checks are expected to run on a single node and may execute SQL queries or scans
+// under the hood to detect inconsistencies. All results are surfaced through the
+// inspectIssue type.
+type inspectCheck interface {
+	// Started reports whether the check has been initialized.
+	Started() bool
+
+	// Start prepares the check to begin returning results.
+	Start(ctx context.Context, cfg *execinfra.ServerConfig, span roachpb.Span, workerIndex int) error
+
+	// Next returns the next inspect error, if any.
+	// Returns (nil, nil) when there are no errors for the current row.
+	Next(ctx context.Context, cfg *execinfra.ServerConfig) (*inspectIssue, error)
+
+	// Done reports whether the check has produced all results.
+	Done(ctx context.Context) bool
+
+	// Close cleans up resources for the check.
+	Close(ctx context.Context) error
+}
+
+// inspectLogger records issues found by inspect checks. Implementations of this
+// interface define how inspectIssue results are handled.
+type inspectLogger interface {
+	logIssue(ctx context.Context, issue *inspectIssue) error
+}
+
+// inspectRunner coordinates the execution of a set of inspectChecks.
+//
+// It manages the lifecycle of each check, including initialization,
+// iteration, and cleanup. Each call to Step processes one unit of
+// work: either advancing a check by one result or moving on to the next
+// check if the current one is finished.
+//
+// When a validation issue is found, the runner calls the provided
+// inspectLogger to record it.
+type inspectRunner struct {
+	// checks holds the list of checks to run. Each check is run to completion
+	// before moving on to the next.
+	checks []inspectCheck
+
+	// logger records issues reported by the checks.
+	logger inspectLogger
+}
+
+// Step advances execution by processing one result from the current inspectCheck.
+//
+// If the current check is not yet started, it is initialized. If it has more results,
+// Step retrieves the next result and logs it if an issue is found. If the check is done,
+// it is closed and removed from the queue.
+//
+// Returns true if a check was advanced or an issue was found. Returns false when all
+// checks are complete. If an error occurs at any stage, it is returned immediately.
+func (c *inspectRunner) Step(
+	ctx context.Context, cfg *execinfra.ServerConfig, span roachpb.Span, workerIndex int,
+) (bool, error) {
+	for len(c.checks) > 0 {
+		check := c.checks[0]
+		if !check.Started() {
+			if err := check.Start(ctx, cfg, span, workerIndex); err != nil {
+				return false, err
+			}
+		}
+
+		if !check.Done(ctx) {
+			issue, err := check.Next(ctx, cfg)
+			if err != nil {
+				return false, err
+			}
+			if issue != nil {
+				err = c.logger.logIssue(ctx, issue)
+				if err != nil {
+					return false, errors.Wrapf(err, "error logging inspect issue")
+				}
+			}
+			return true, nil
+		}
+
+		if err := check.Close(ctx); err != nil {
+			return false, err
+		}
+		c.checks = c.checks[1:]
+	}
+	return false, nil
+}