Merge #154635 #154678 #154873

154635: sql: skip flaky ngpsql roachtests r=rafiss a=bghal

The tests are timing out.

```
  Failed Import_string_array [30 s]
  Stack Trace:
```

```
  Failed Prepended_messages [30 s]
  Stack Trace:
```

Fixes: #154004 
Fixes: #153691

Release note: None


154678: sql: give cluster time to upgrade under test r=rafiss a=bghal

In local 3-node testing the upgrade takes ~80s to get to the cluster
version bump so relaxing the retries helps the tests pass.

Epic: none
Fixes: #154550
Fixes: #154077
Fixes: #155046

Release note: None


154873: sql: implement `INSPECT` command r=bghal a=bghal

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

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

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


Co-authored-by: Brendan Gerrity <[email protected]>

Author: craig[bot]

pkg/cmd/roachtest/tests/npgsql_blocklist.go

@@ -662,6 +662,8 @@ var npgsqlBlocklist = blocklist{
 
 var npgsqlIgnoreList = blocklist{
 	`Npgsql.Tests.ConnectionTests(Multiplexing).Fail_connect_then_succeed(True)`:                     "flaky",
+	`Npgsql.Tests.CopyTests(Multiplexing).Import_string_array`:                                       "flaky",
+	`Npgsql.Tests.CopyTests(Multiplexing).Prepended_messages`:                                        "flaky",
 	`Npgsql.Tests.TransactionTests(NonMultiplexing).Failed_transaction_on_close_with_custom_timeout`: "flaky",
 	`Npgsql.Tests.TransactionTests(Multiplexing).Failed_transaction_on_close_with_custom_timeout`:    "flaky",
 }

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")