sql: implement INSPECT command

This change implements the `INSPECT` command which runs inspect jobs to
perform consistency validation checks on tables and their indexes.

Informs: cockroachdb#148365, cockroachdb#155056
Epic: CRDB-30356

Release note (sql change): The `INSPECT` command runs consistency
validation check jobs against tables or databases and specified indexes.

Author: bghal

pkg/jobs/resultcols.go

@@ -38,6 +38,12 @@ var OnlineRestoreJobExecutionResultHeader = colinfo.ResultColumns{
 	{Name: "background_download_job_id", Typ: types.Int},
 }
 
+// InspectJobExecutionResultHeader is the header for INSPECT job results.
+var InspectJobExecutionResultHeader = colinfo.ResultColumns{
+	{Name: "job_id", Typ: types.Int},
+	{Name: "status", Typ: types.String},
+}
+
 // DetachedJobExecutionResultHeader is the header for various job commands when
 // job executes in detached mode (i.e. the caller doesn't wait for job completion).
 var DetachedJobExecutionResultHeader = colinfo.ResultColumns{

pkg/server/server.go

@@ -100,7 +100,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/flowinfra"
 	_ "github.com/cockroachdb/cockroach/pkg/sql/gcjob"    // register jobs declared outside of pkg/sql
 	_ "github.com/cockroachdb/cockroach/pkg/sql/importer" // register jobs/planHooks declared outside of pkg/sql
-	_ "github.com/cockroachdb/cockroach/pkg/sql/inspect"  // register jobs declared outside of pkg/sql
+	_ "github.com/cockroachdb/cockroach/pkg/sql/inspect"  // register job and planHook declared outside of pkg/sql
 	"github.com/cockroachdb/cockroach/pkg/sql/optionalnodeliveness"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire"
 	_ "github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scjob" // register jobs declared outside of pkg/sql

pkg/sql/BUILD.bazel

@@ -141,7 +141,7 @@ go_library(
         "information_schema.go",
         "insert.go",
         "insert_fast_path.go",
-        "inspect_node.go",
+        "inspect_job.go",
         "instrumentation.go",
         "internal.go",
         "internal_result_channel.go",

pkg/sql/alter_index.go

@@ -40,7 +40,7 @@ func (p *planner) AlterIndex(ctx context.Context, n *tree.AlterIndex) (planNode,
 		return nil, err
 	}
 
-	_, tableDesc, index, err := p.getTableAndIndex(ctx, &n.Index, privilege.CREATE, true /* skipCache */)
+	_, tableDesc, index, err := p.GetTableAndIndex(ctx, &n.Index, privilege.CREATE, true /* skipCache */)
 	if err != nil {
 		return nil, err
 	}

pkg/sql/catalog/descriptor.go

@@ -341,7 +341,7 @@ type DatabaseDescriptor interface {
 type TableDescriptor interface {
 	Descriptor
 
-	// TableDesc returns the backing protobuf for this database.
+	// TableDesc returns the backing protobuf for this table.
 	TableDesc() *descpb.TableDescriptor
 
 	// GetState returns the raw state of this descriptor. This information is

pkg/sql/comment_on_index.go

@@ -34,7 +34,7 @@ func (p *planner) CommentOnIndex(ctx context.Context, n *tree.CommentOnIndex) (p
 		return nil, err
 	}
 
-	_, tableDesc, index, err := p.getTableAndIndex(ctx, &n.Index, privilege.CREATE, true /* skipCache */)
+	_, tableDesc, index, err := p.GetTableAndIndex(ctx, &n.Index, privilege.CREATE, true /* skipCache */)
 	if err != nil {
 		return nil, err
 	}

pkg/sql/importer/import_job.go

@@ -328,24 +328,30 @@ func (r *importResumer) Resume(ctx context.Context, execCtx interface{}) error {
 		}
 		tblDesc := tabledesc.NewBuilder(table.Desc).BuildImmutableTable()
 		if len(tblDesc.PublicNonPrimaryIndexes()) > 0 {
-			jobID, err := sql.TriggerInspectJob(
+			checks, err := sql.InspectChecksForTable(ctx, nil /* p */, tblDesc)
+			if err != nil {
+				return err
+			}
+
+			job, err := sql.TriggerInspectJob(
 				ctx,
 				fmt.Sprintf("import-validation-%s", tblDesc.GetName()),
 				p.ExecCfg(),
-				tblDesc,
+				nil, /* txn */
+				checks,
 				setPublicTimestamp,
 			)
 			if err != nil {
 				return errors.Wrapf(err, "failed to trigger inspect for import validation for table %s", tblDesc.GetName())
 			}
-			log.Eventf(ctx, "triggered inspect job %d for import validation for table %s with AOST %s", jobID, tblDesc.GetName(), setPublicTimestamp)
+			log.Eventf(ctx, "triggered inspect job %d for import validation for table %s with AOST %s", job.ID(), tblDesc.GetName(), setPublicTimestamp)
 
 			// For sync mode, wait for the inspect job to complete.
 			if validationMode == ImportRowCountValidationSync {
-				if err := p.ExecCfg().JobRegistry.WaitForJobs(ctx, []jobspb.JobID{jobID}); err != nil {
-					return errors.Wrapf(err, "failed to wait for inspect job %d for table %s", jobID, tblDesc.GetName())
+				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())
 				}
-				log.Eventf(ctx, "inspect job %d completed for table %s", jobID, tblDesc.GetName())
+				log.Eventf(ctx, "inspect job %d completed for table %s", job.ID(), tblDesc.GetName())
 			}
 		}
 	}

pkg/sql/inspect/BUILD.bazel

@@ -7,6 +7,7 @@ go_library(
         "inspect_job.go",
         "inspect_logger.go",
         "inspect_metrics.go",
+        "inspect_plan.go",
         "inspect_processor.go",
         "issue.go",
         "log_sink.go",
@@ -18,6 +19,7 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/sql/inspect",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/clusterversion",
         "//pkg/jobs",
         "//pkg/jobs/jobfrontier",
         "//pkg/jobs/jobspb",
@@ -30,6 +32,7 @@ go_library(
         "//pkg/sql",
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/catenumpb",
+        "//pkg/sql/catalog/colinfo",
         "//pkg/sql/catalog/descpb",
         "//pkg/sql/catalog/descs",
         "//pkg/sql/execinfra",
@@ -39,6 +42,7 @@ go_library(
         "//pkg/sql/pgwire/pgcode",
         "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/physicalplan",
+        "//pkg/sql/privilege",
         "//pkg/sql/rowexec",
         "//pkg/sql/sem/idxtype",
         "//pkg/sql/sem/tree",

pkg/sql/inspect/index_consistency_check_test.go

@@ -438,8 +438,7 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 				FROM generate_series(1001, 2000) AS gs1;`,
 			)
 
-			// TODO(148365): Run INSPECT instead of SCRUB.
-			_, err = db.Exec(`SET enable_scrub_job=true`)
+			_, err = db.Exec(`SET enable_inspect_command=true`)
 			require.NoError(t, err)
 
 			// If not using timestamp before corruption, get current timestamp
@@ -451,8 +450,8 @@ func TestDetectIndexConsistencyErrors(t *testing.T) {
 
 			// 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)
+			inspectQuery := fmt.Sprintf(`INSPECT TABLE test.t AS OF SYSTEM TIME %s WITH OPTIONS INDEX ALL`, absoluteTimestamp)
+			_, err = db.Query(inspectQuery)
 			if tc.expectedErrRegex == "" {
 				require.NoError(t, err)
 				require.Equal(t, 0, issueLogger.numIssuesFound())
@@ -587,10 +586,9 @@ func TestIndexConsistencyWithReservedWordColumns(t *testing.T) {
 		`CREATE INDEX idx_having ON test.reserved_table ("having", "group")`,
 	)
 
-	// TODO(148365): Run INSPECT instead of SCRUB.
-	_, err := db.Exec(`SET enable_scrub_job=true`)
+	_, err := db.Exec(`SET enable_inspect_command=true`)
 	require.NoError(t, err)
-	_, err = db.Query(`EXPERIMENTAL SCRUB TABLE test.reserved_table WITH OPTIONS INDEX ALL`)
+	_, err = db.Query(`INSPECT 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_job.go

@@ -400,6 +400,19 @@ func (c *inspectResumer) maybeCleanupProtectedTimestamp(
 	}
 }
 
+// ReportResults implements the JobResultsReporter interface.
+func (r *inspectResumer) ReportResults(ctx context.Context, resultsCh chan<- tree.Datums) error {
+	select {
+	case resultsCh <- tree.Datums{
+		tree.NewDInt(tree.DInt(r.job.ID())),
+		tree.NewDString(string(r.job.State())),
+	}:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
 func init() {
 	createResumerFn := func(job *jobs.Job, settings *cluster.Settings) jobs.Resumer {
 		return &inspectResumer{job: job}