sql/inspect: make AS OF SYSTEM TIME optional in INSPECT jobs

Previously, INSPECT jobs required specifying an `AS OF SYSTEM TIME`
timestamp. This commit removes that requirement.

If the timestamp is left empty, the INSPECT processor will now default
to using the current time when querying for index inconsistencies. A
fresh timestamp will be acquired for each span. No protected timestamp
record will be used in this mode.

Fixes #148675

Epic: CRDB-30356
Release note: none

Author: spilchen

pkg/jobs/jobspb/jobs.proto

@@ -1483,6 +1483,7 @@ message InspectDetails {
   repeated Check checks = 1;
 
   // AsOf specifies the timestamp at which the inspect checks should be performed.
+  // If empty, the current timestamp will be used for each span processed.
   util.hlc.Timestamp as_of = 2 [(gogoproto.nullable) = false];
 
   // ProtectedTimestampRecord is the id of the protected timestamp record that

pkg/sql/inspect/BUILD.bazel

@@ -95,6 +95,7 @@ go_test(
         "//pkg/testutils/skip",
         "//pkg/testutils/sqlutils",
         "//pkg/testutils/testcluster",
+        "//pkg/util/hlc",
         "//pkg/util/leaktest",
         "//pkg/util/log",
         "//pkg/util/syncutil",

pkg/sql/inspect/index_consistency_check_test.go

@@ -571,7 +571,7 @@ func TestIndexConsistencyWithReservedWordColumns(t *testing.T) {
 	// TODO(148365): Run INSPECT instead of SCRUB.
 	_, err := db.Exec(`SET enable_scrub_job=true`)
 	require.NoError(t, err)
-	_, err = db.Query(`EXPERIMENTAL SCRUB TABLE test.reserved_table AS OF SYSTEM TIME '-1us' WITH OPTIONS INDEX ALL`)
+	_, err = db.Query(`EXPERIMENTAL SCRUB TABLE test.reserved_table WITH OPTIONS INDEX ALL`)
 	require.NoError(t, err, "should succeed on table with reserved word column names")
 	require.Equal(t, 0, issueLogger.numIssuesFound(), "No issues should be found in happy path test")
 

pkg/sql/inspect/inspect_processor.go

@@ -20,6 +20,7 @@ import (
 	"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/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
@@ -36,7 +37,7 @@ var (
 	)
 )
 
-type inspectCheckFactory func() inspectCheck
+type inspectCheckFactory func(asOf hlc.Timestamp) inspectCheck
 
 type inspectProcessor struct {
 	processorID    int32
@@ -47,6 +48,7 @@ type inspectProcessor struct {
 	spanSrc        spanSource
 	logger         inspectLogger
 	concurrency    int
+	clock          *hlc.Clock
 }
 
 var _ execinfra.Processor = (*inspectProcessor)(nil)
@@ -190,10 +192,12 @@ func getInspectLogger(flowCtx *execinfra.FlowCtx, jobID jobspb.JobID) inspectLog
 func (p *inspectProcessor) processSpan(
 	ctx context.Context, span roachpb.Span, workerIndex int,
 ) (err error) {
+	asOfToUse := p.getTimestampForSpan()
+
 	// Only create checks that apply to this span
 	var checks []inspectCheck
 	for _, factory := range p.checkFactories {
-		check := factory()
+		check := factory(asOfToUse)
 		applies, err := check.AppliesTo(p.cfg.Codec, span)
 		if err != nil {
 			return err
@@ -231,6 +235,16 @@ func (p *inspectProcessor) processSpan(
 	return nil
 }
 
+// getTimestampForSpan returns the timestamp to use for processing a span.
+// If AsOf is empty, it returns the current timestamp from the processor's clock.
+// Otherwise, it returns the specified AsOf timestamp.
+func (p *inspectProcessor) getTimestampForSpan() hlc.Timestamp {
+	if p.spec.InspectDetails.AsOf.IsEmpty() {
+		return p.clock.Now()
+	}
+	return p.spec.InspectDetails.AsOf
+}
+
 // 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.
@@ -240,9 +254,6 @@ func (p *inspectProcessor) processSpan(
 func newInspectProcessor(
 	ctx context.Context, flowCtx *execinfra.FlowCtx, processorID int32, spec execinfrapb.InspectSpec,
 ) (execinfra.Processor, error) {
-	if spec.InspectDetails.AsOf.IsEmpty() {
-		return nil, errors.AssertionFailedf("ASOF time must be set for INSPECT")
-	}
 	checkFactories, err := buildInspectCheckFactories(flowCtx, spec)
 	if err != nil {
 		return nil, err
@@ -257,6 +268,7 @@ func newInspectProcessor(
 		spanSrc:        newSliceSpanSource(spec.Spans),
 		logger:         getInspectLogger(flowCtx, spec.JobID),
 		concurrency:    getProcessorConcurrency(flowCtx),
+		clock:          flowCtx.Cfg.DB.KV().Clock(),
 	}, nil
 }
 
@@ -275,14 +287,14 @@ func buildInspectCheckFactories(
 		indexID := specCheck.IndexID
 		switch specCheck.Type {
 		case jobspb.InspectCheckIndexConsistency:
-			checkFactories = append(checkFactories, func() inspectCheck {
+			checkFactories = append(checkFactories, func(asOf hlc.Timestamp) inspectCheck {
 				return &indexConsistencyCheck{
 					indexConsistencyCheckApplicability: indexConsistencyCheckApplicability{
 						tableID: tableID,
 					},
 					flowCtx: flowCtx,
 					indexID: indexID,
-					asOf:    spec.InspectDetails.AsOf,
+					asOf:    asOf,
 				}
 			})
 

pkg/sql/inspect/inspect_processor_test.go

@@ -12,11 +12,13 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/stretchr/testify/require"
@@ -228,19 +230,29 @@ func runProcessorAndWait(t *testing.T, proc *inspectProcessor, expectErr bool) {
 
 // makeProcessor will create an inspect processor for test.
 func makeProcessor(
-	t *testing.T, checkFactory inspectCheckFactory, src spanSource, concurrency int,
+	t *testing.T,
+	checkFactory inspectCheckFactory,
+	src spanSource,
+	concurrency int,
+	asOf hlc.Timestamp,
 ) (*inspectProcessor, *testIssueCollector) {
 	t.Helper()
+	clock := hlc.NewClockForTesting(nil)
 	logger := &testIssueCollector{}
 	proc := &inspectProcessor{
-		spec:           execinfrapb.InspectSpec{},
+		spec: execinfrapb.InspectSpec{
+			InspectDetails: jobspb.InspectDetails{
+				AsOf: asOf,
+			},
+		},
 		checkFactories: []inspectCheckFactory{checkFactory},
 		cfg: &execinfra.ServerConfig{
 			Settings: cluster.MakeTestingClusterSettings(),
 		},
 		spanSrc:     src,
 		logger:      logger,
 		concurrency: concurrency,
+		clock:       clock,
 	}
 	return proc, logger
 }
@@ -318,12 +330,12 @@ func TestInspectProcessor_ControlFlow(t *testing.T) {
 
 	for _, tc := range tests {
 		t.Run(tc.desc, func(t *testing.T) {
-			factory := func() inspectCheck {
+			factory := func(asOf hlc.Timestamp) inspectCheck {
 				return &testingInspectCheck{
 					configs: tc.configs,
 				}
 			}
-			proc, _ := makeProcessor(t, factory, tc.spanSrc, len(tc.configs))
+			proc, _ := makeProcessor(t, factory, tc.spanSrc, len(tc.configs), hlc.Timestamp{})
 			runProcessorAndWait(t, proc, tc.expectErr)
 		})
 	}
@@ -337,7 +349,7 @@ func TestInspectProcessor_EmitIssues(t *testing.T) {
 		mode:     spanModeNormal,
 		maxSpans: 1,
 	}
-	factory := func() inspectCheck {
+	factory := func(asOf hlc.Timestamp) inspectCheck {
 		return &testingInspectCheck{
 			configs: []testingCheckConfig{
 				{
@@ -350,9 +362,78 @@ func TestInspectProcessor_EmitIssues(t *testing.T) {
 			},
 		}
 	}
-	proc, logger := makeProcessor(t, factory, spanSrc, 1)
+	proc, logger := makeProcessor(t, factory, spanSrc, 1, hlc.Timestamp{})
 
 	runProcessorAndWait(t, proc, true /* expectErr */)
 
 	require.Equal(t, 2, logger.numIssuesFound())
 }
+
+func TestInspectProcessor_AsOfTime(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	tests := []struct {
+		name            string
+		asOf            hlc.Timestamp
+		verifyTimestamp func(t *testing.T, actualTime time.Time, capturedTimestamp hlc.Timestamp, testStartTime time.Time)
+	}{
+		{
+			name: "empty timestamp uses clock time",
+			asOf: hlc.Timestamp{}, // Empty timestamp
+			verifyTimestamp: func(t *testing.T, actualTime time.Time, capturedTimestamp hlc.Timestamp, testStartTime time.Time) {
+				// Verify that the AOST time in the issue is >= the test start time
+				require.True(t, actualTime.After(testStartTime) || actualTime.Equal(testStartTime),
+					"AOST time (%v) should be >= test start time (%v)", actualTime, testStartTime)
+				// Also verify the timestamp was not empty
+				require.False(t, capturedTimestamp.IsEmpty(),
+					"Captured timestamp should not be empty when AsOf is not specified")
+			},
+		},
+		{
+			name: "specific timestamp is preserved",
+			asOf: hlc.Timestamp{WallTime: 12345},
+			verifyTimestamp: func(t *testing.T, actualTime time.Time, capturedTimestamp hlc.Timestamp, testStartTime time.Time) {
+				// Verify that the exact timestamp is preserved
+				require.Equal(t, hlc.Timestamp{WallTime: 12345}.GoTime(), actualTime)
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			spanSrc := &testingSpanSource{
+				mode:     spanModeNormal,
+				maxSpans: 1,
+			}
+
+			// Record start time before creating the processor
+			testStartTime := time.Now()
+
+			var capturedTimestamp hlc.Timestamp
+			factory := func(asOf hlc.Timestamp) inspectCheck {
+				capturedTimestamp = asOf
+				return &testingInspectCheck{
+					configs: []testingCheckConfig{
+						{
+							mode: checkModeNone,
+							issues: []*inspectIssue{
+								{ErrorType: "test_error", PrimaryKey: "pk1", AOST: asOf.GoTime()},
+							},
+						},
+					},
+				}
+			}
+
+			proc, logger := makeProcessor(t, factory, spanSrc, 1, tc.asOf)
+
+			runProcessorAndWait(t, proc, true /* expectErr */)
+
+			require.Equal(t, 1, logger.numIssuesFound())
+
+			// Run the test-specific timestamp verification
+			actualTime := logger.issue(0).AOST
+			tc.verifyTimestamp(t, actualTime, capturedTimestamp, testStartTime)
+		})
+	}
+}

pkg/sql/scrub.go

@@ -227,11 +227,14 @@ func (n *scrubNode) startScrubTable(
 			return pgerror.Newf(pgcode.InvalidTransactionState,
 				"cannot run within a multi-statement transaction")
 		}
-		if !hasTS {
-			return pgerror.Newf(pgcode.Syntax,
-				"SCRUB with inspect jobs requires AS OF SYSTEM TIME")
+
+		// If AS OF SYSTEM TIME is not provided, we pass in an empty timestamp to
+		// force the job to use the current time.
+		var asOfForJob hlc.Timestamp
+		if hasTS {
+			asOfForJob = ts
 		}
-		return n.runScrubTableJob(ctx, p, tableDesc, ts)
+		return n.runScrubTableJob(ctx, p, tableDesc, asOfForJob)
 	}
 	// Process SCRUB options. These are only present during a SCRUB TABLE
 	// statement.