Merge #152304

152304: sql: report inspect errors r=bghal a=bghal

sql: report INSPECT errors to table

The inspect job logs errors. This change persists the errors to the
`system.inspect_errors` table where they will be viewed with the `SHOW
INSPECT ERRORS` command.

Fixes: #148301
Epic: CRDB-30356

Release note: None


Co-authored-by: Brendan Gerrity <[email protected]>

Author: craig[bot]

pkg/sql/inspect/BUILD.bazel

@@ -5,11 +5,13 @@ go_library(
     srcs = [
         "index_consistency_check.go",
         "inspect_job.go",
+        "inspect_logger.go",
         "inspect_processor.go",
         "issue.go",
         "log_sink.go",
         "runner.go",
         "span_source.go",
+        "table_sink.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/sql/inspect",
     visibility = ["//visibility:public"],
@@ -58,6 +60,7 @@ go_test(
         "issue_test.go",
         "main_test.go",
         "runner_test.go",
+        "table_sink_test.go",
     ],
     embed = [":inspect"],
     exec_properties = select({
@@ -66,6 +69,7 @@ go_test(
     }),
     deps = [
         "//pkg/base",
+        "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/kv",
         "//pkg/roachpb",
@@ -75,6 +79,7 @@ go_test(
         "//pkg/settings/cluster",
         "//pkg/sql",
         "//pkg/sql/catalog",
+        "//pkg/sql/catalog/descs",
         "//pkg/sql/catalog/desctestutils",
         "//pkg/sql/execinfra",
         "//pkg/sql/execinfrapb",

pkg/sql/inspect/inspect_logger.go

@@ -0,0 +1,47 @@
+// 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/errors"
+)
+
+// inspectLogger records issues found by inspect checks. Implementations of this
+// interface define how inspectIssue results are handled.
+type inspectLogger interface {
+	// logIssue records an inspectIssue found by a check.
+	logIssue(ctx context.Context, issue *inspectIssue) error
+
+	// hasIssues returns true if any issues have been logged.
+	hasIssues() bool
+}
+
+// inspectLoggers manages a collection of inspectLogger instances.
+type inspectLoggers []inspectLogger
+
+var _ inspectLogger = inspectLoggers{}
+
+func (l inspectLoggers) logIssue(ctx context.Context, issue *inspectIssue) error {
+	var retErr error
+
+	for _, logger := range l {
+		if err := logger.logIssue(ctx, issue); err != nil {
+			retErr = errors.CombineErrors(retErr, err)
+		}
+	}
+	return retErr
+}
+
+func (l inspectLoggers) hasIssues() bool {
+	for _, logger := range l {
+		if logger.hasIssues() {
+			return true
+		}
+	}
+	return false
+}

pkg/sql/inspect/inspect_processor.go

@@ -167,13 +167,21 @@ func getProcessorConcurrency(flowCtx *execinfra.FlowCtx) int {
 }
 
 // getInspectLogger returns a logger for the inspect processor.
-func getInspectLogger(flowCtx *execinfra.FlowCtx) inspectLogger {
+func getInspectLogger(flowCtx *execinfra.FlowCtx, jobID jobspb.JobID) inspectLogger {
+	loggers := inspectLoggers{
+		&logSink{},
+		&tableSink{
+			db:    flowCtx.Cfg.DB,
+			jobID: jobID,
+		},
+	}
+
 	knobs := flowCtx.Cfg.ExecutorConfig.(*sql.ExecutorConfig).InspectTestingKnobs
-	if knobs == nil || knobs.InspectIssueLogger == nil {
-		// TODO(148301): Implement a proper logger that writes to system.inspect_errors.
-		return &logSink{}
+	if knobs != nil && knobs.InspectIssueLogger != nil {
+		loggers = append(loggers, knobs.InspectIssueLogger.(inspectLogger))
 	}
-	return knobs.InspectIssueLogger.(inspectLogger)
+
+	return loggers
 }
 
 // processSpan executes all configured inspect checks against a single span.
@@ -225,14 +233,15 @@ func newInspectProcessor(
 	if err != nil {
 		return nil, err
 	}
+
 	return &inspectProcessor{
 		spec:           spec,
 		processorID:    processorID,
 		flowCtx:        flowCtx,
 		checkFactories: checkFactories,
 		cfg:            flowCtx.Cfg,
 		spanSrc:        newSliceSpanSource(spec.Spans),
-		logger:         getInspectLogger(flowCtx),
+		logger:         getInspectLogger(flowCtx, spec.JobID),
 		concurrency:    getProcessorConcurrency(flowCtx),
 	}, nil
 }

pkg/sql/inspect/log_sink.go

@@ -7,25 +7,26 @@ package inspect
 
 import (
 	"context"
+	"sync/atomic"
 
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 )
 
 // logSink will report any inspect errors directly to cockroach.log.
 type logSink struct {
-	foundIssue bool
+	foundIssue atomic.Bool
 }
 
 var _ inspectLogger = &logSink{}
 
 // logIssue implements the inspectLogger interface.
 func (c *logSink) logIssue(ctx context.Context, issue *inspectIssue) error {
-	c.foundIssue = true
+	c.foundIssue.Store(true)
 	log.Dev.Errorf(ctx, "inspect issue: %+v", issue)
 	return nil
 }
 
 // hasIssues implements the inspectLogger interface.
 func (c *logSink) hasIssues() bool {
-	return c.foundIssue
+	return c.foundIssue.Load()
 }

pkg/sql/inspect/runner.go

@@ -42,16 +42,6 @@ type inspectCheck interface {
 	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 records an inspectIssue found by a check.
-	logIssue(ctx context.Context, issue *inspectIssue) error
-
-	// hasIssues returns true if any issues have been logged.
-	hasIssues() bool
-}
-
 // inspectRunner coordinates the execution of a set of inspectChecks.
 //
 // It manages the lifecycle of each check, including initialization,

pkg/sql/inspect/table_sink.go

@@ -0,0 +1,69 @@
+// 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"
+	gojson "encoding/json"
+	"sync/atomic"
+
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/security/username"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
+	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
+	"github.com/cockroachdb/cockroach/pkg/sql/sessiondatapb"
+)
+
+// tableSink will report any inspect errors directly to system.inspect_errors.
+type tableSink struct {
+	foundIssue atomic.Bool
+	db         descs.DB
+	jobID      jobspb.JobID
+}
+
+var _ inspectLogger = &tableSink{}
+
+// logIssue implements the inspectLogger interface.
+func (c *tableSink) logIssue(ctx context.Context, issue *inspectIssue) error {
+	c.foundIssue.Store(true)
+
+	detailsBytes, err := gojson.Marshal(issue.Details)
+	if err != nil {
+		return err
+	}
+
+	executor := c.db.Executor()
+
+	if _, err = executor.ExecEx(
+		ctx,
+		"insert-inspect-error",
+		nil, /* txn */
+		sessiondata.InternalExecutorOverride{
+			User:             username.NodeUserName(),
+			QualityOfService: &sessiondatapb.BulkLowQoS,
+		},
+		`INSERT INTO system.inspect_errors
+			(job_id, error_type, aost, database_id, schema_id, id, primary_key, details)
+			 VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
+		c.jobID,
+		string(issue.ErrorType),
+		issue.AOST,
+		issue.DatabaseID,
+		issue.SchemaID,
+		issue.ObjectID,
+		issue.PrimaryKey,
+		string(detailsBytes),
+	); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// hasIssues implements the inspectLogger interface.
+func (c *tableSink) hasIssues() bool {
+	return c.foundIssue.Load()
+}

pkg/sql/inspect/table_sink_test.go

@@ -0,0 +1,87 @@
+// 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"
+	gojson "encoding/json"
+	"testing"
+	"time"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/redact"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTableSink(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	s, db, _ := serverutils.StartServer(t, base.TestServerArgs{})
+	defer s.Stopper().Stop(context.Background())
+	runner := sqlutils.MakeSQLRunner(db)
+
+	const jobID jobspb.JobID = 50
+
+	var tableLogger inspectLogger = &tableSink{
+		db:    s.InternalDB().(descs.DB),
+		jobID: jobID,
+	}
+
+	issue := inspectIssue{
+		ErrorType:  MissingSecondaryIndexEntry,
+		DatabaseID: 1,
+		SchemaID:   2,
+		ObjectID:   3,
+		PrimaryKey: "key",
+		Details: map[redact.RedactableString]interface{}{
+			"foo":  `"bar"`,
+			"biz":  "\u2603baz",
+			"titi": "toto\n",
+		},
+	}
+
+	require.NoError(t, tableLogger.logIssue(context.Background(), &issue))
+
+	// Query the system.inspect_errors table and expect one entry
+	var count int
+	runner.QueryRow(t, `SELECT count(*) FROM system.inspect_errors WHERE job_id = $1`, jobID).Scan(&count)
+	require.Equal(t, 1, count, "Expected exactly one entry in system.inspect_errors")
+
+	// Compare the entry against the test instance
+	var actualJobID int64
+	var actualErrorType, actualPrimaryKey string
+	var actualAOST time.Time
+	var actualDatabaseID, actualSchemaID, actualObjectID int64
+	var actualDetailsBytes []byte
+
+	runner.QueryRow(t, `SELECT job_id, error_type, aost, database_id, schema_id, id, primary_key, details
+		FROM system.inspect_errors WHERE job_id = $1`, jobID).Scan(
+		&actualJobID, &actualErrorType, &actualAOST, &actualDatabaseID, &actualSchemaID,
+		&actualObjectID, &actualPrimaryKey, &actualDetailsBytes)
+
+	require.Equal(t, int64(jobID), actualJobID, "job_id should match")
+	require.Equal(t, string(issue.ErrorType), actualErrorType, "error_type should match")
+	require.Equal(t, issue.AOST, actualAOST, "aost should match")
+	require.Equal(t, int64(issue.DatabaseID), actualDatabaseID, "database_id should match")
+	require.Equal(t, int64(issue.SchemaID), actualSchemaID, "schema_id should match")
+	require.Equal(t, int64(issue.ObjectID), actualObjectID, "id should match")
+	require.Equal(t, issue.PrimaryKey, actualPrimaryKey, "primary_key should match")
+
+	var detailsMap map[string]interface{}
+	err := gojson.Unmarshal(actualDetailsBytes, &detailsMap)
+	require.NoError(t, err)
+	require.Len(t, detailsMap, 3)
+	require.Equal(t, `"bar"`, detailsMap["foo"], "details[foo] should match")
+	require.Equal(t, "\u2603baz", detailsMap["biz"], "details[biz] should match")
+	require.Equal(t, "toto\n", detailsMap["titi"], "details[titi] should match")
+}