sql/inspect: use ASOF time for INSPECT

Previously, inspect issues used the current time for the AOST field
instead of the timestamp at which the data was actually scanned. This
change updates index consistency checks to use time thats flowed into
the job. When calling INSPECT from SCRUB, we use the time specified in
the SCRUB command.

Informs #148573

Release note: none
Epic: CRDB-30356

Author: spilchen

pkg/jobs/jobspb/jobs.proto

@@ -1446,6 +1446,9 @@ message InspectDetails {
 
   // Checks is the list of individual checks this job will perform.
   repeated Check checks = 1;
+
+  // AsOf specifies the timestamp at which the inspect checks should be performed.
+  util.hlc.Timestamp as_of = 2 [(gogoproto.nullable) = false];
 }
 
 message UpdateTableMetadataCacheDetails {}

pkg/sql/inspect/BUILD.bazel

@@ -40,6 +40,7 @@ go_library(
         "//pkg/sql/spanutils",
         "//pkg/sql/types",
         "//pkg/util/ctxgroup",
+        "//pkg/util/hlc",
         "//pkg/util/log",
         "//pkg/util/timeutil",
         "//pkg/util/tracing",

pkg/sql/inspect/index_consistency_check.go

@@ -25,7 +25,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondatapb"
 	"github.com/cockroachdb/cockroach/pkg/sql/spanutils"
-	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
 )
@@ -38,6 +38,7 @@ type indexConsistencyCheck struct {
 	flowCtx *execinfra.FlowCtx
 	tableID descpb.ID
 	indexID descpb.IndexID
+	asOf    hlc.Timestamp
 
 	tableDesc     catalog.TableDescriptor
 	secIndex      catalog.Index
@@ -194,16 +195,21 @@ func (c *indexConsistencyCheck) Start(
 		colNames(pkColumns), colNames(otherColumns), c.tableDesc.GetID(), c.secIndex, c.priIndex.GetID(), predicate,
 	)
 
+	// Wrap the query with AS OF SYSTEM TIME to ensure it uses the specified timestamp
+	// TODO(148573): use a protected timestamp record for this timestamp.
+	queryWithAsOf := fmt.Sprintf("SELECT * FROM (%s) AS OF SYSTEM TIME %s", checkQuery, c.asOf.AsOfSystemTime())
+
 	// Store the query for error reporting
-	c.lastQuery = checkQuery
+	c.lastQuery = queryWithAsOf
 
+	// Execute the query with AS OF SYSTEM TIME embedded in the SQL
 	it, err := c.flowCtx.Cfg.DB.Executor().QueryIteratorEx(
 		ctx, "inspect-index-consistency-check", nil, /* txn */
 		sessiondata.InternalExecutorOverride{
 			User:             username.NodeUserName(),
 			QualityOfService: &sessiondatapb.BulkLowQoS,
 		},
-		checkQuery,
+		queryWithAsOf,
 		queryArgs...,
 	)
 	if err != nil {
@@ -243,9 +249,8 @@ func (c *indexConsistencyCheck) Next(
 		details["query"] = c.lastQuery // Store the query that caused the error
 
 		return &inspectIssue{
-			ErrorType: InternalError,
-			// TODO(148573): Use the timestamp that we create a protected timestamp for.
-			AOST:       timeutil.Now(),
+			ErrorType:  InternalError,
+			AOST:       c.asOf.GoTime(),
 			DatabaseID: c.tableDesc.GetParentID(),
 			SchemaID:   c.tableDesc.GetParentSchemaID(),
 			ObjectID:   c.tableDesc.GetID(),
@@ -308,9 +313,8 @@ func (c *indexConsistencyCheck) Next(
 	details["index_name"] = c.secIndex.GetName()
 
 	return &inspectIssue{
-		ErrorType: errorType,
-		// TODO(148573): Use the timestamp that we create a protected timestamp for.
-		AOST:       timeutil.Now(),
+		ErrorType:  errorType,
+		AOST:       c.asOf.GoTime(),
 		DatabaseID: c.tableDesc.GetParentID(),
 		SchemaID:   c.tableDesc.GetParentSchemaID(),
 		ObjectID:   c.tableDesc.GetID(),

pkg/sql/inspect/index_consistency_check_test.go

@@ -149,6 +149,8 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 		// Each element corresponds to the issue at the same index in expectedIssues.
 		// For non-internal-error issues, the corresponding element should be nil.
 		expectedInternalErrorPatterns []map[string]string
+		// useTimestampBeforeCorruption uses a timestamp from before corruption is introduced
+		useTimestampBeforeCorruption bool
 	}{
 		{
 			desc: "happy path sanity",
@@ -236,6 +238,15 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 			},
 			postIndexSQL: "DELETE FROM test.t", /* delete all rows to test hasRows=false code path */
 		},
+		{
+			desc:          "timestamp before corruption - no issues found",
+			splitRangeDDL: "ALTER TABLE test.t SPLIT AT VALUES (500)",
+			indexDDL: []string{
+				"CREATE INDEX idx_t_a ON test.t (a) STORING (c)",
+			},
+			danglingIndexEntryInsertQuery: "SELECT 15, 30, 300, 'corrupt', 'e_3', 300.5", // Add dangling entry after timestamp
+			useTimestampBeforeCorruption:  true,                                          // Use timestamp from before corruption
+		},
 	} {
 		t.Run(tc.desc, func(t *testing.T) {
 			issueLogger.reset()
@@ -288,6 +299,19 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 				r.Exec(t, tc.postIndexSQL)
 			}
 
+			// Get timestamp before corruption if needed
+			var expectedASOFTime time.Time
+
+			if tc.useTimestampBeforeCorruption {
+				// Get timestamp before corruption
+				r.QueryRow(t, "SELECT now()::timestamp").Scan(&expectedASOFTime)
+				expectedASOFTime = expectedASOFTime.UTC()
+
+				// Sleep for 1 millisecond to ensure corruption happens after timestamp
+				// This should be long enough to guarantee a different timestamp
+				time.Sleep(1 * time.Millisecond)
+			}
+
 			tableDesc := desctestutils.TestingGetPublicTableDescriptor(kvDB, codec, "test", "t")
 			secIndex := tableDesc.PublicNonPrimaryIndexes()[0]
 
@@ -341,7 +365,18 @@ func TestDetectIndexConsistencyErrors(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.t WITH OPTIONS INDEX ALL`)
+
+			// If not using timestamp before corruption, get current timestamp
+			if !tc.useTimestampBeforeCorruption {
+				// Convert relative timestamp to absolute timestamp using CRDB
+				r.QueryRow(t, "SELECT (now() + '-1us')::timestamp").Scan(&expectedASOFTime)
+				expectedASOFTime = expectedASOFTime.UTC()
+			}
+
+			// Use the absolute timestamp in nanoseconds for inspect command
+			absoluteTimestamp := fmt.Sprintf("'%d'", expectedASOFTime.UnixNano())
+			scrubQuery := fmt.Sprintf(`EXPERIMENTAL SCRUB TABLE test.t AS OF SYSTEM TIME %s WITH OPTIONS INDEX ALL`, absoluteTimestamp)
+			_, err = db.Query(scrubQuery)
 			if tc.expectedErrRegex == "" {
 				require.NoError(t, err)
 				require.Equal(t, 0, issueLogger.numIssuesFound())
@@ -380,7 +415,7 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 				require.NotEqual(t, 0, foundIssue.DatabaseID, "expected issue to have a database ID: %s", expectedIssue)
 				require.NotEqual(t, 0, foundIssue.SchemaID, "expected issue to have a schema ID: %s", expectedIssue)
 				require.NotEqual(t, 0, foundIssue.ObjectID, "expected issue to have an object ID: %s", expectedIssue)
-				require.NotEqual(t, time.Time{}, foundIssue.AOST, "expected issue to have an AOST time: %s", expectedIssue)
+				require.Equal(t, expectedASOFTime, foundIssue.AOST.UTC())
 
 				// Additional validation for internal errors
 				if foundIssue.ErrorType == "internal_error" {
@@ -460,7 +495,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 WITH OPTIONS INDEX ALL`)
+	_, err = db.Query(`EXPERIMENTAL SCRUB TABLE test.reserved_table AS OF SYSTEM TIME '-1us' 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_job_test.go

@@ -107,7 +107,7 @@ func TestInspectJobImplicitTxnSemantics(t *testing.T) {
 					onInspectErrorToReturn.Store(&tc.onStartError)
 					defer func() { onInspectErrorToReturn.Store(nil) }()
 				}
-				_, err := db.Exec("EXPERIMENTAL SCRUB TABLE db.t")
+				_, err := db.Exec("EXPERIMENTAL SCRUB TABLE db.t AS OF SYSTEM TIME '-1us'")
 				pauseJobStart.Store(false)
 				if tc.expectedErrRegex != "" {
 					require.Error(t, err)

pkg/sql/inspect/inspect_processor.go

@@ -218,6 +218,9 @@ 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
@@ -252,6 +255,7 @@ func buildInspectCheckFactories(
 					flowCtx: flowCtx,
 					tableID: specCheck.TableID,
 					indexID: specCheck.IndexID,
+					asOf:    spec.InspectDetails.AsOf,
 				}
 			})
 

pkg/sql/logictest/testdata/logic_test/distsql_inspect

@@ -47,12 +47,12 @@ subtest end
 subtest scrub_job_implicit_txn
 
 statement ok
-EXPERIMENTAL SCRUB TABLE data;
+EXPERIMENTAL SCRUB TABLE data AS OF SYSTEM TIME '-1us';
 
 query TTB
 SELECT description, status, finished IS NOT NULL AS finished FROM [SHOW JOBS] WHERE job_type = 'INSPECT' ORDER BY created DESC LIMIT 1
 ----
-EXPERIMENTAL SCRUB TABLE data  succeeded  true
+EXPERIMENTAL SCRUB TABLE data AS OF SYSTEM TIME '-1us'  succeeded  true
 
 subtest end
 

pkg/sql/logictest/testdata/logic_test/inspect

@@ -20,12 +20,12 @@ subtest end
 subtest scrub_job_implicit_txn
 
 statement ok
-EXPERIMENTAL SCRUB TABLE t1;
+EXPERIMENTAL SCRUB TABLE t1 AS OF SYSTEM TIME '-1us';
 
 query TTB
 SELECT description, status, finished IS NOT NULL AS finished FROM [SHOW JOBS] WHERE job_type = 'INSPECT' ORDER BY created DESC LIMIT 1
 ----
-EXPERIMENTAL SCRUB TABLE t1  succeeded  true
+EXPERIMENTAL SCRUB TABLE t1 AS OF SYSTEM TIME '-1us'  succeeded  true
 
 subtest end
 

pkg/sql/scrub.go

@@ -217,14 +217,22 @@ func (n *scrubNode) startScrubDatabase(ctx context.Context, p *planner, name *tr
 func (n *scrubNode) startScrubTable(
 	ctx context.Context, p *planner, tableDesc catalog.TableDescriptor, tableName *tree.TableName,
 ) error {
-	if p.extendedEvalCtx.SessionData().EnableScrubJob {
-		return n.runScrubTableJob(ctx, p, tableDesc)
-	}
-
 	ts, hasTS, err := p.getTimestamp(ctx, n.n.AsOf)
 	if err != nil {
 		return err
 	}
+
+	if p.extendedEvalCtx.SessionData().EnableScrubJob {
+		if !p.extendedEvalCtx.TxnIsSingleStmt {
+			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")
+		}
+		return n.runScrubTableJob(ctx, p, tableDesc, ts)
+	}
 	// Process SCRUB options. These are only present during a SCRUB TABLE
 	// statement.
 	var indexesSet bool
@@ -466,7 +474,7 @@ func createConstraintCheckOperations(
 }
 
 func (n *scrubNode) runScrubTableJob(
-	ctx context.Context, p *planner, tableDesc catalog.TableDescriptor,
+	ctx context.Context, p *planner, tableDesc catalog.TableDescriptor, asOf hlc.Timestamp,
 ) error {
 	// Consistency check is done async via a job.
 	jobID := p.ExecCfg().JobRegistry.MakeJobID()
@@ -490,18 +498,14 @@ func (n *scrubNode) runScrubTableJob(
 					IndexID: secIndexes[0].GetID(),
 				},
 			},
+			AsOf: asOf,
 		},
 		Progress:      jobspb.InspectProgress{},
 		CreatedBy:     nil,
 		Username:      username.NodeUserName(),
 		DescriptorIDs: descpb.IDs{tableDesc.GetID()},
 	}
 
-	if !p.extendedEvalCtx.TxnIsSingleStmt {
-		return pgerror.Newf(pgcode.InvalidTransactionState,
-			"cannot run within a multi-statement transaction")
-	}
-
 	var sj *jobs.StartableJob
 	if err := p.ExecCfg().InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) (err error) {
 		return p.ExecCfg().JobRegistry.CreateStartableJobWithTxn(ctx, &sj, jobID, txn, jr)