Merge #154010 #154071

154010: sql/inspect: make AS OF SYSTEM TIME optional in INSPECT jobs r=spilchen a=spilchen

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

154071: startup: unskip TestStartupFailureRandomRange; disable UA improve logs r=rickystewart a=nicktrav

TestStartupFailureRandomRange is marked as a nightly-only test. However, given the intricacies of how various environment variables are passed into the test infrastructure, this test is effectively never run.

Update the skip logic to build on top of #153135, allowing the test to be run in nightly-only stress runs.

Update the test to disable multi-tenant mode. This currently causes the test to fail.

Fixes #123908.

Release note: None.

Epic: None.

Co-authored-by: Matt Spilchen <[email protected]>
Co-authored-by: Nick Travers <[email protected]>

Author: 3 people

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.

pkg/util/startup/startup_test.go

@@ -85,18 +85,13 @@ func TestStartupFailure(t *testing.T) {
 func TestStartupFailureRandomRange(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
-	// This test takes 30s and so we don't want it to run in the "blocking path"
-	// of CI at all, and we also don't want to stress it in nightlies as part of
-	// a big package (where it will take a lot of time that could be spent running
-	// "faster" tests). In this package, it is the only test and so it's fine to
-	// run it under nightly (skipping race builds because with many nodes they are
-	// very resource intensive and tend to collapse).
+	// This test takes 30s+, so we don't want it to run in the "blocking path" of
+	// CI. We also skip race builds as the test uses multiple nodes, which can
+	// cause the test to grind to a halt and flake out.
 	skip.UnderRace(t, "6 nodes with replication is too slow for race")
-	skip.WithIssue(t, 9999999999, "nicktrav will have a fix shortly")
-	// TODO(nicktrav): re-enable only under nightlies once the fix is out.
-	//if !skip.NightlyStress() {
-	//	skip.IgnoreLint(t, "test takes 30s to run due to circuit breakers and timeouts")
-	//}
+	if !skip.Stress() {
+		skip.IgnoreLint(t, "test takes 30s to run due to circuit breakers and timeouts")
+	}
 
 	rng, seed := randutil.NewTestRand()
 	t.Log("TestStartupFailureRandomRange using seed", seed)
@@ -148,6 +143,11 @@ func runCircuitBreakerTestForKey(
 	args := base.TestClusterArgs{
 		ServerArgsPerNode:   make(map[int]base.TestServerArgs),
 		ReusableListenerReg: lReg,
+		// TODO(travers): This test is has a lingering issue when run in UA mode
+		// that needs to be addressed before the following can be removed.
+		ServerArgs: base.TestServerArgs{
+			DefaultTestTenant: base.TestIsSpecificToStorageLayerAndNeedsASystemTenant,
+		},
 	}
 	var enableFaults atomic.Bool
 	for i := 0; i < nodes; i++ {
@@ -229,6 +229,7 @@ func runCircuitBreakerTestForKey(
 		return d.StartKey
 	}
 
+	t.Log("segmenting ranges")
 	var rangeSpans []roachpb.Span
 	r, err := c.QueryContext(ctx, "select range_id, start_key, end_key from crdb_internal.ranges_no_leases order by start_key")
 	require.NoError(t, err, "failed to query ranges")
@@ -243,9 +244,11 @@ func runCircuitBreakerTestForKey(
 		})
 	}
 	good, bad := faultyRangeSelector(rangeSpans)
+	t.Logf("prepping %d good ranges", len(good))
 	for _, span := range good {
 		prepRange(span.Key, false)
 	}
+	t.Logf("prepping %d faulty ranges", len(good))
 	var ranges []string
 	for _, span := range bad {
 		prepRange(span.Key, true)
@@ -254,27 +257,33 @@ func runCircuitBreakerTestForKey(
 	rangesList := fmt.Sprintf("[%s]", strings.Join(ranges, ", "))
 
 	// Remove nodes permanently to only leave quorum on planned ranges.
+	t.Log("stopping n3 and n4")
 	tc.StopServer(3)
 	tc.StopServer(4)
 
 	// Stop node with replicas that would leave ranges without quorum.
+	t.Log("stopping n5")
 	tc.StopServer(5)
 
 	// Probe compromised ranges to trigger circuit breakers on them. If we don't
 	// do this, then restart queries will wait for quorum to be reestablished with
 	// restarting node without failing.
+	t.Logf("waiting for %d compromised ranges to trigger CBs", len(bad))
 	var wg sync.WaitGroup
 	wg.Add(len(bad))
 	for _, span := range bad {
 		go func(key roachpb.Key) {
 			defer wg.Done()
+			t.Logf("waiting for compromised range: %s", key)
 			_ = db.Put(context.Background(), keys.RangeProbeKey(roachpb.RKey(key)), "")
+			t.Logf("done waiting for compromised range: %s", key)
 		}(span.Key)
 	}
 	wg.Wait()
 
 	// Restart node and check that it succeeds in reestablishing range quorum
 	// necessary for startup actions.
+	t.Log("starting n5")
 	require.NoError(t, lReg.MustGet(t, 5).Reopen())
 	err = tc.RestartServer(5)
 	require.NoError(t, err, "restarting server with range(s) %s tripping circuit breaker", rangesList)