Merge #149894 #150427

149894: sql/inspect: implement INSPECT processor stub r=spilchen a=spilchen

This commit completes the implementation of the INSPECT processor, which executes the per-node portion of an INSPECT job. The processor follows a producer-worker model: the producer enqueues spans, which are consumed by multiple concurrent workers. By default, the number of workers matches the output of runtime.GOMAXPROCS(0), but it can be overridden via the sql.inspect.processor_concurrency setting.

The processor uses inspectRunner to perform various checks. This commit introduces a stub for an index consistency check and a testing-only check to validate processor logic.

Informs: #148299

Epic: CRDB-30356

Release note: none

150427: roachtest: unblock rust-postgres test case fixed by #22463 r=spilchen a=spilchen

A specific test case within the rust-postgres was blocklisted due to its reliance on CITEXT, which was recently implemented in #22463.  This commit removes the blocklist entry to reflect the success.

Closes #149973

Epic: none
Release note: none

Co-authored-by: Matt Spilchen <matt.spilchen@cockroachlabs.com>

Author: craig[bot]

pkg/cmd/roachtest/tests/rust_postgres_blocklist.go

@@ -61,7 +61,6 @@ var rustPostgresBlocklist = blocklist{
 	"types.ltxtquery":                       "unsupported datatype - https://github.com/cockroachdb/cockroach/issues/44657",
 	"types.ltxtquery_any":                   "unsupported datatype - https://github.com/cockroachdb/cockroach/issues/44657",
 	"types.test_array_vec_params":           "default int size (int4 vs int8) mismatch",
-	"types.test_citext_params":              "unsupported citext type alias - https://github.com/cockroachdb/cockroach/issues/22463",
 	"types.test_hstore_params":              "unsupported datatype - https://github.com/cockroachdb/cockroach/issues/41284",
 	"types.test_i16_params":                 "default int size (int4 vs int8) mismatch",
 	"types.test_i32_params":                 "default int size (int4 vs int8) mismatch",

pkg/sql/inspect/BUILD.bazel

@@ -8,13 +8,15 @@ go_library(
         "issue.go",
         "log_sink.go",
         "runner.go",
+        "span_source.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/sql/inspect",
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/roachpb",
+        "//pkg/settings",
         "//pkg/settings/cluster",
         "//pkg/sql",
         "//pkg/sql/catalog/descpb",
@@ -26,6 +28,7 @@ go_library(
         "//pkg/sql/rowexec",
         "//pkg/sql/sem/tree",
         "//pkg/sql/types",
+        "//pkg/util/ctxgroup",
         "//pkg/util/log",
         "//pkg/util/tracing",
         "@com_github_cockroachdb_errors//:errors",
@@ -37,6 +40,7 @@ go_test(
     name = "inspect_test",
     srcs = [
         "inspect_job_test.go",
+        "inspect_processor_test.go",
         "issue_test.go",
         "main_test.go",
         "runner_test.go",
@@ -48,14 +52,17 @@ go_test(
         "//pkg/security/securityassets",
         "//pkg/security/securitytest",
         "//pkg/server",
+        "//pkg/settings/cluster",
         "//pkg/sql",
         "//pkg/sql/execinfra",
+        "//pkg/sql/execinfrapb",
         "//pkg/testutils",
         "//pkg/testutils/serverutils",
         "//pkg/testutils/sqlutils",
         "//pkg/testutils/testcluster",
         "//pkg/util/leaktest",
         "//pkg/util/log",
+        "//pkg/util/syncutil",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_redact//:redact",
         "@com_github_stretchr_testify//require",

pkg/sql/inspect/inspect_processor.go

@@ -7,19 +7,43 @@ package inspect
 
 import (
 	"context"
+	"runtime"
 
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/rowexec"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
+	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
+	"github.com/cockroachdb/errors"
 )
 
+var (
+	processorConcurrencyOverride = settings.RegisterIntSetting(
+		settings.ApplicationLevel,
+		"sql.inspect.processor_concurrency",
+		"sets the number of concurrent processors for INSPECT jobs. "+
+			"0 uses the default based on GOMAXPROCS. Values above GOMAXPROCS are capped.",
+		0,
+		settings.NonNegativeInt,
+	)
+)
+
+type inspectCheckFactory func() inspectCheck
+
 type inspectProcessor struct {
-	processorID int32
-	flowCtx     *execinfra.FlowCtx
-	spec        execinfrapb.InspectSpec
+	processorID    int32
+	flowCtx        *execinfra.FlowCtx
+	spec           execinfrapb.InspectSpec
+	cfg            *execinfra.ServerConfig
+	checkFactories []inspectCheckFactory
+	spanSrc        spanSource
+	logger         inspectLogger
+	concurrency    int
 }
 
 var _ execinfra.Processor = (*inspectProcessor)(nil)
@@ -59,21 +83,153 @@ func (p *inspectProcessor) Run(ctx context.Context, output execinfra.RowReceiver
 	output.ProducerDone()
 }
 
+// runInspect starts a set of worker goroutines to process spans concurrently.
+// Each span is read from a buffered channel and passed to processSpan.
+// The function blocks until all spans are processed or an error occurs.
 func (p *inspectProcessor) runInspect(ctx context.Context, output execinfra.RowReceiver) error {
-	log.Infof(ctx, "INSPECT processor started processorID=%d", p.processorID)
+	log.Infof(ctx, "INSPECT processor started processorID=%d concurrency=%d", p.processorID, p.concurrency)
+
+	group := ctxgroup.WithContext(ctx)
+
+	if p.concurrency == 0 {
+		return errors.AssertionFailedf("must have at least one worker")
+	}
+	spanCh := make(chan roachpb.Span, p.concurrency)
+
+	// Launch worker goroutines. Each worker reads spans from spanCh and processes
+	// them.
+	for i := 0; i < p.concurrency; i++ {
+		workerIndex := i
+		group.GoCtx(func(ctx context.Context) error {
+			for {
+				select {
+				case <-ctx.Done():
+					// If the context is canceled (e.g., due to an error in another worker), exit.
+					return ctx.Err()
+				case span, ok := <-spanCh:
+					if !ok {
+						// Channel is closed, no more spans to process.
+						return nil
+					}
+					if err := p.processSpan(ctx, span, workerIndex); err != nil {
+						// On error, return it. ctxgroup will cancel all other goroutines.
+						return err
+					}
+				}
+			}
+		})
+	}
+
+	// Producer goroutine: feeds all spans into the channel for workers to consume.
+	group.GoCtx(func(ctx context.Context) error {
+		defer close(spanCh)
+		for {
+			span, ok, err := p.spanSrc.NextSpan(ctx)
+			if err != nil {
+				return err
+			}
+			if !ok {
+				return nil // done
+			}
+			select {
+			case <-ctx.Done():
+				// Exit early if context is canceled.
+				return ctx.Err()
+			case spanCh <- span:
+				// Send span to the workers.
+			}
+		}
+	})
+
+	// Wait for all goroutines to finish.
+	return group.Wait()
+}
+
+// getProcessorConcurrency returns the number of concurrent workers to use for
+// INSPECT processing. If the cluster setting is non-zero, it uses the minimum
+// of the override and GOMAXPROCS. Otherwise, it defaults to GOMAXPROCS.
+func getProcessorConcurrency(flowCtx *execinfra.FlowCtx) int {
+	override := int(processorConcurrencyOverride.Get(&flowCtx.Cfg.Settings.SV))
+	if override > 0 {
+		return min(runtime.GOMAXPROCS(0), override)
+	}
+	return runtime.GOMAXPROCS(0)
+}
+
+// processSpan executes all configured inspect checks against a single span.
+// It instantiates a fresh set of checks from the configured factories and uses
+// an inspectRunner to drive their execution.
+func (p *inspectProcessor) processSpan(
+	ctx context.Context, span roachpb.Span, workerIndex int,
+) error {
+	checks := make([]inspectCheck, len(p.checkFactories))
+	for i, factory := range p.checkFactories {
+		checks[i] = factory()
+	}
+	runner := inspectRunner{
+		checks: checks,
+		logger: p.logger,
+	}
+
+	// Process all checks on the given span.
+	for {
+		ok, err := runner.Step(ctx, p.cfg, span, workerIndex)
+		if err != nil {
+			return err
+		}
+		if !ok {
+			break
+		}
+	}
 	return nil
 }
 
+// newInspectProcessor constructs a new inspectProcessor from the given InspectSpec.
+// It parses the spec to generate a set of inspectCheck factories, sets up the span source,
+// and wires in logging and concurrency controls.
+//
+// The returned processor is ready for integration into a distributed flow, but will not
+// begin execution until Run is called.
 func newInspectProcessor(
 	ctx context.Context, flowCtx *execinfra.FlowCtx, processorID int32, spec execinfrapb.InspectSpec,
 ) (execinfra.Processor, error) {
+	checkFactories, err := buildInspectCheckFactories(spec)
+	if err != nil {
+		return nil, err
+	}
 	return &inspectProcessor{
-		spec:        spec,
-		processorID: processorID,
-		flowCtx:     flowCtx,
+		spec:           spec,
+		processorID:    processorID,
+		flowCtx:        flowCtx,
+		checkFactories: checkFactories,
+		cfg:            flowCtx.Cfg,
+		spanSrc:        newSliceSpanSource(spec.Spans),
+		// TODO(148301): log to cockroach.log for now, but later log to system.inspect_errors
+		logger:      &logSink{},
+		concurrency: getProcessorConcurrency(flowCtx),
 	}, nil
 }
 
+// buildInspectCheckFactories returns a slice of factory functions that create
+// inspectCheck instances at runtime. Each factory corresponds to one check entry
+// defined in the job's InspectSpec.
+//
+// This indirection ensures that each check instance is freshly created per span,
+// avoiding shared state across concurrent workers.
+func buildInspectCheckFactories(spec execinfrapb.InspectSpec) ([]inspectCheckFactory, error) {
+	checkFactories := make([]inspectCheckFactory, 0, len(spec.InspectDetails.Checks))
+	for _, specCheck := range spec.InspectDetails.Checks {
+		switch specCheck.Type {
+		case jobspb.InspectCheckIndexConsistency:
+			// TODO(148863): implement the index consistency checker. No-op for now.
+
+		default:
+			return nil, errors.AssertionFailedf("unsupported inspect check type: %v", specCheck.Type)
+		}
+	}
+	return checkFactories, nil
+}
+
 func init() {
 	rowexec.NewInspectProcessor = newInspectProcessor
 }