sql: add support for span checks and cluster checks

Previously the checks were independent and could only process individual
spans.
The row count check will be dependent on other checks that perform the
scan of each span and will need to aggregate the row counts after all
the spans have been processed.
This change sets up interfaces for span-level (dependent on other
checks running on the spans) and cluster-level checks (run after all spans have
been processed).

Part of: #155472
Epic: CRDB-55075

Release note: None

Author: bghal

pkg/sql/inspect/BUILD.bazel

@@ -3,6 +3,7 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "inspect",
     srcs = [
+        "cluster_runner.go",
         "index_consistency_check.go",
         "inspect_job_entry.go",
         "inspect_logger.go",
@@ -72,6 +73,7 @@ go_test(
     name = "inspect_test",
     size = "large",
     srcs = [
+        "cluster_runner_test.go",
         "index_consistency_check_test.go",
         "inspect_metrics_test.go",
         "inspect_processor_test.go",

pkg/sql/inspect/cluster_runner.go

@@ -0,0 +1,72 @@
+// Copyright 2026 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"
+)
+
+type clusterRunner struct {
+	// checks holds the list of checks to run. Each check is run to completion
+	// before moving on to the next.
+	checks []inspectClusterCheck
+
+	// logger records issues reported by the checks.
+	logger *inspectLoggerBundle
+
+	progressTracker *inspectProgressTracker
+}
+
+// processClusterChecks runs any cluster-level checks using the accumulated job
+// progress.
+func (c *clusterRunner) Step(ctx context.Context) (bool, error) {
+	for len(c.checks) > 0 {
+		check := c.checks[0]
+
+		if !check.Started() {
+			if err := check.StartCluster(ctx, c.progressTracker.mu.cachedProgress.GetInspect()); err != nil {
+				return false, errors.Wrapf(err, "error starting cluster inspect check")
+			}
+		}
+
+		for !check.DoneCluster(ctx) {
+			issue, err := check.NextCluster(ctx)
+			if err != nil {
+				return false, errors.Wrapf(err, "error running cluster inspect check")
+			}
+			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.CloseCluster(ctx); err != nil {
+			return false, errors.Wrapf(err, "error closing cluster inspect check")
+		}
+
+		c.checks = c.checks[1:]
+	}
+
+	return false, nil
+}
+
+// Close cleans up all checks in the runner. It will attempt to close each check,
+// even if errors occur during closing. If multiple checks fail to close, then
+// a combined error is returned.
+func (c *clusterRunner) Close(ctx context.Context) error {
+	var retErr error
+	for _, check := range c.checks {
+		if err := check.CloseCluster(ctx); err != nil {
+			retErr = errors.CombineErrors(retErr, err)
+		}
+	}
+	return retErr
+}