sql/inspect: record table version in INSPECT job to detect schema drift

Previously, the INSPECT job only recorded the table IDs involved in the
job. However, there is a time gap between when the job record is created
and when the job is executed, during which schema changes amy occur.
These changes can invalidate assumptions made at job creation (e.g.
index ID existence).

This change introduces table version tracking into the job record.
Before executing any checks, the job will verify that the current table
versions match those stored at creation. If a mismatch is detected, the
job will fail early.

This mechanism mirrors similar protections implemented in other jobs,
such as TTL jobs.

Informs #158197
Release note: none

Author: spilchen

pkg/jobs/jobspb/jobs.proto

@@ -1474,6 +1474,14 @@ message InspectDetails {
       (gogoproto.customname) = "IndexID",
       (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb.IndexID"
     ];
+
+    // TableVersion records the descriptor version observed when the check was
+    // planned. It is used at execution time to detect concurrent schema changes
+    // and abort the job before running inconsistent checks.
+    uint64 table_version = 4 [
+      (gogoproto.customname) = "TableVersion",
+      (gogoproto.casttype) = "github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb.DescriptorVersion"
+    ];
   }
 
   // Checks is the list of individual checks this job will perform.

pkg/sql/importer/import_job.go

@@ -327,32 +327,45 @@ func (r *importResumer) Resume(ctx context.Context, execCtx interface{}) error {
 		if err != nil {
 			return err
 		}
-		tblDesc := tabledesc.NewBuilder(table.Desc).BuildImmutableTable()
-		if len(tblDesc.PublicNonPrimaryIndexes()) > 0 {
-			checks, err := inspect.ChecksForTable(ctx, nil /* p */, tblDesc)
+
+		var checks []*jobspb.InspectDetails_Check
+		var tableName string
+		if err := p.ExecCfg().InternalDB.DescsTxn(ctx, func(
+			ctx context.Context, txn descs.Txn,
+		) error {
+			// INSPECT requires the latest descriptor. The one cached in the job is
+			// out of date as it has an old table version.
+			tblDesc, err := txn.Descriptors().ByIDWithoutLeased(txn.KV()).WithoutNonPublic().Get().Table(ctx, table.Desc.ID)
 			if err != nil {
 				return err
 			}
+			tableName = tblDesc.GetName()
+			checks, err = inspect.ChecksForTable(ctx, nil /* p */, tblDesc)
+			return err
+		}); err != nil {
+			return err
+		}
 
+		if len(checks) > 0 {
 			job, err := inspect.TriggerJob(
 				ctx,
-				fmt.Sprintf("import-validation-%s", tblDesc.GetName()),
+				fmt.Sprintf("import-validation-%s", tableName),
 				p.ExecCfg(),
 				nil, /* txn */
 				checks,
 				setPublicTimestamp,
 			)
 			if err != nil {
-				return errors.Wrapf(err, "failed to trigger inspect for import validation for table %s", tblDesc.GetName())
+				return errors.Wrapf(err, "failed to trigger inspect for import validation for table %s", tableName)
 			}
-			log.Eventf(ctx, "triggered inspect job %d for import validation for table %s with AOST %s", job.ID(), tblDesc.GetName(), setPublicTimestamp)
+			log.Eventf(ctx, "triggered inspect job %d for import validation for table %s with AOST %s", job.ID(), tableName, setPublicTimestamp)
 
 			// For sync mode, wait for the inspect job to complete.
 			if validationMode == ImportRowCountValidationSync {
 				if err := p.ExecCfg().JobRegistry.WaitForJobs(ctx, []jobspb.JobID{job.ID()}); err != nil {
-					return errors.Wrapf(err, "failed to wait for inspect job %d for table %s", job.ID(), tblDesc.GetName())
+					return errors.Wrapf(err, "failed to wait for inspect job %d for table %s", job.ID(), tableName)
 				}
-				log.Eventf(ctx, "inspect job %d completed for table %s", job.ID(), tblDesc.GetName())
+				log.Eventf(ctx, "inspect job %d completed for table %s", job.ID(), tableName)
 			}
 		}
 	}

pkg/sql/inspect/index_consistency_check.go

@@ -10,6 +10,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
@@ -80,7 +81,10 @@ type indexConsistencyCheck struct {
 
 	flowCtx *execinfra.FlowCtx
 	indexID descpb.IndexID
-	asOf    hlc.Timestamp
+	// tableVersion is the descriptor version recorded when the check was planned.
+	// It is used to detect concurrent schema changes for non-AS OF inspections.
+	tableVersion descpb.DescriptorVersion
+	asOf         hlc.Timestamp
 
 	tableDesc catalog.TableDescriptor
 	secIndex  catalog.Index
@@ -448,6 +452,17 @@ func (c *indexConsistencyCheck) loadCatalogInfo(ctx context.Context) error {
 		if err != nil {
 			return err
 		}
+		if c.tableVersion != 0 && c.tableDesc.GetVersion() != c.tableVersion {
+			return errors.WithHintf(
+				errors.Newf(
+					"table %s [%d] has had a schema change since the job has started at %s",
+					c.tableDesc.GetName(),
+					c.tableDesc.GetID(),
+					c.tableDesc.GetModificationTime().GoTime().Format(time.RFC3339),
+				),
+				"use AS OF SYSTEM TIME to avoid schema changes during inspection",
+			)
+		}
 
 		c.priIndex = c.tableDesc.GetPrimaryIndex()
 

pkg/sql/inspect/index_consistency_check_test.go

@@ -7,6 +7,7 @@ package inspect
 
 import (
 	"context"
+	gosql "database/sql"
 	"fmt"
 	"testing"
 	"time"
@@ -766,3 +767,74 @@ func TestMissingIndexEntryWithHistoricalQuery(t *testing.T) {
 	require.Contains(t, err.Error(), "INSPECT found inconsistencies",
 		"INSPECT should detect the missing index entry that existed at the historical timestamp")
 }
+
+func TestInspectWithoutASOFSchemaChange(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	unblockCh := make(chan struct{})
+	s, db, _ := serverutils.StartServer(t, base.TestServerArgs{
+		Knobs: base.TestingKnobs{
+			Inspect: &sql.InspectTestingKnobs{
+				OnInspectJobStart: func() error {
+					// Block the INSPECT job from starting until the DROP INDEX completes.
+					const maxWait = 30 * time.Second
+					select {
+					case <-unblockCh:
+						return nil
+					case <-time.After(maxWait):
+						return errors.New("timed out waiting for DROP INDEX to complete")
+					}
+				},
+			},
+		},
+	})
+	defer s.Stopper().Stop(ctx)
+	r := sqlutils.MakeSQLRunner(db)
+
+	r.ExecMultiple(t,
+		`SET CLUSTER SETTING jobs.registry.interval.adopt = '500ms'`,
+		`SET enable_inspect_command = true`,
+		`CREATE DATABASE t`,
+		`CREATE TABLE t.pk_swap (
+                       old_pk INT PRIMARY KEY,
+                       new_pk INT UNIQUE NOT NULL,
+                       payload INT NOT NULL,
+                       INDEX payload_idx (payload)
+              )`,
+		`INSERT INTO t.pk_swap SELECT i, i+100, i+200 FROM generate_series(1, 5) AS g(i)`,
+	)
+
+	r.Exec(t, `INSPECT TABLE t.pk_swap WITH OPTIONS INDEX ALL, DETACHED`)
+	r.Exec(t, `DROP INDEX t.pk_swap@payload_idx`)
+	// Unblock the INSPECT job now that the DROP INDEX has completed.
+	close(unblockCh)
+
+	var jobID int64
+	require.Eventually(t, func() bool {
+		row := db.QueryRow(`SELECT job_id FROM [SHOW JOBS] WHERE job_type = 'INSPECT' ORDER BY created DESC LIMIT 1`)
+		return row.Scan(&jobID) == nil
+	}, 10*time.Second, 100*time.Millisecond)
+
+	var (
+		status string
+		jobErr gosql.NullString
+	)
+	require.Eventually(t, func() bool {
+		row := db.QueryRow(`SELECT status, error FROM [SHOW JOBS] WHERE job_id = $1`, jobID)
+		if err := row.Scan(&status, &jobErr); err != nil {
+			t.Logf("error polling job status: %v", err)
+			return false
+		}
+		return status != "running" && status != "pending"
+	}, 60*time.Second, time.Second)
+
+	require.Equal(t, "failed", status, "detached job should fail after schema change")
+	require.Regexp(t, `table pk_swap \[\d+\] has had a schema change since the job has started at .+\nHINT: use AS OF SYSTEM TIME to avoid schema changes during inspection`, jobErr.String)
+
+	rows, err := db.Query(`SHOW INSPECT ERRORS FOR TABLE t.pk_swap WITH DETAILS`)
+	require.NoError(t, err)
+	defer rows.Close()
+	require.False(t, rows.Next(), "no inspect errors are recorded when the job aborts before running checks")
+}

pkg/sql/inspect/inspect_job_entry.go

@@ -148,7 +148,12 @@ func ChecksForTable(
 			}
 			continue
 		}
-		check := jobspb.InspectDetails_Check{Type: jobspb.InspectCheckIndexConsistency, TableID: table.GetID(), IndexID: index.GetID()}
+		check := jobspb.InspectDetails_Check{
+			Type:         jobspb.InspectCheckIndexConsistency,
+			TableID:      table.GetID(),
+			IndexID:      index.GetID(),
+			TableVersion: table.GetVersion(),
+		}
 		checks = append(checks, &check)
 	}
 
@@ -184,7 +189,12 @@ func checksByIndexNames(
 			return nil, pgerror.Newf(pgcode.InvalidName, "index %q on table %q is not supported for index consistency checking", index.GetName(), table.GetName())
 		}
 
-		checks = append(checks, &jobspb.InspectDetails_Check{Type: jobspb.InspectCheckIndexConsistency, TableID: table.GetID(), IndexID: index.GetID()})
+		checks = append(checks, &jobspb.InspectDetails_Check{
+			Type:         jobspb.InspectCheckIndexConsistency,
+			TableID:      table.GetID(),
+			IndexID:      index.GetID(),
+			TableVersion: table.GetVersion(),
+		})
 	}
 
 	return checks, nil

pkg/sql/inspect/inspect_processor.go

@@ -452,16 +452,18 @@ func buildInspectCheckFactories(
 	for _, specCheck := range spec.InspectDetails.Checks {
 		tableID := specCheck.TableID
 		indexID := specCheck.IndexID
+		tableVersion := specCheck.TableVersion
 		switch specCheck.Type {
 		case jobspb.InspectCheckIndexConsistency:
 			checkFactories = append(checkFactories, func(asOf hlc.Timestamp) inspectCheck {
 				return &indexConsistencyCheck{
 					indexConsistencyCheckApplicability: indexConsistencyCheckApplicability{
 						tableID: tableID,
 					},
-					flowCtx: flowCtx,
-					indexID: indexID,
-					asOf:    asOf,
+					flowCtx:      flowCtx,
+					indexID:      indexID,
+					tableVersion: tableVersion,
+					asOf:         asOf,
 				}
 			})