Merge #155284 #155439

155284: backfill: add a separate control for number of vectors merged per batch r=mw5h a=mw5h

Add the 'bulkio.index_backfill.vector_merge_batch_size', which controls the number of rows to merge into vector indexes per transaction while the index is being created. This is analogous to the 'bulkio.index_backfill.merge_batch_size' setting, but it only applies when the target index is a vector index. By default, 3 vectors will be merged per transaction to reduce contention with fixup tasks.

Fixes: #155283
Release note (sql change): Added the
bulkio.index_backfill.vector_merge_batch_size cluster setting to control how many vectors to merge into a vector index per transaction during create operations. By default, this defaults to 3.

155439: roachtest/inspect: use INSPECT instead of SCRUB in inspect test r=spilchen a=spilchen

Now that the INSPECT command has been implemented, we can use INSPECT instead of SCRUB to start inspect in the inspect throughput test.

Informs #154457

Epic: CRDB-30356

Release note: none

Co-authored-by: Matt White <matt.white@cockroachlabs.com>
Co-authored-by: Matt Spilchen <matt.spilchen@cockroachlabs.com>

Author: 3 people

pkg/cmd/roachtest/tests/inspect_throughput.go

@@ -33,9 +33,7 @@ func registerInspectThoughput(r registry.Registry) {
 	r.Add(makeInspectThroughputTest(r, 12, 8, 500_000_000, 3*time.Hour, 1))
 
 	// Long run: 12 nodes × 8 CPUs, 1B rows, 2 index checks (runs INSPECT twice: 1 index, then 2 indexes), ~5 hours (in v25.4)
-	// TODO(148365): Until we have INSPECT syntax, we cannot execute the INSPECT
-	// job on a subset of indexes. Set this to 2 when INSPECT SQL is available.
-	const indexesForLongRun = 1
+	const indexesForLongRun = 2
 	r.Add(makeInspectThroughputTest(r, 12, 8, 1_000_000_000, 8*time.Hour, indexesForLongRun))
 }
 
@@ -212,8 +210,8 @@ func makeInspectThroughputTest(
 				t.Fatal(err)
 			}
 
-			// Enable scrub jobs for EXPERIMENTAL SCRUB to use job system.
-			if _, err := db.Exec("SET enable_scrub_job = on"); err != nil {
+			// Enable INSPECT feature flag.
+			if _, err := db.Exec("USE bulkingest; SET enable_inspect_command = true"); err != nil {
 				t.Fatal(err)
 			}
 
@@ -226,9 +224,8 @@ func makeInspectThroughputTest(
 				tickHistogram(cfg.metricName)
 				before := timeutil.Now()
 
-				// TODO(148365): Update to use INSPECT syntax when SQL is ready.
-				scrubSQL := fmt.Sprintf("EXPERIMENTAL SCRUB TABLE bulkingest.bulkingest WITH OPTIONS INDEX (%s)", cfg.indexListSQL)
-				if _, err := db.Exec(scrubSQL); err != nil {
+				inspectSQL := fmt.Sprintf("INSPECT TABLE bulkingest.bulkingest WITH OPTIONS INDEX (%s)", cfg.indexListSQL)
+				if _, err := db.Exec(inspectSQL); err != nil {
 					t.Fatal(err)
 				}
 

pkg/sql/backfill/mvcc_index_merger.go

@@ -47,6 +47,17 @@ var indexBackfillMergeBatchSize = settings.RegisterIntSetting(
 	settings.NonNegativeInt, /* validateFn */
 )
 
+// indexBackfillVectorMergeBatchSize is the maximum number of vectors we attempt to merge
+// in a single transaction during the merging process. This is a much smaller number because
+// fixups create a lot of conflicts with big batches.
+var indexBackfillVectorMergeBatchSize = settings.RegisterIntSetting(
+	settings.ApplicationLevel,
+	"bulkio.index_backfill.vector_merge_batch_size",
+	"the number of vectors we merge between temporary and adding indexes in a single batch",
+	3,
+	settings.NonNegativeInt, /* validateFn */
+)
+
 // indexBackfillMergeBatchBytes is the maximum number of bytes we attempt to
 // merge from the temporary index in a single transaction during the merging
 // process.
@@ -315,7 +326,12 @@ func (ibm *IndexBackfillMerger) scan(
 			}
 		}
 	}
-	chunkSize := indexBackfillMergeBatchSize.Get(&ibm.flowCtx.Cfg.Settings.SV)
+	var chunkSize int64
+	if _, ok := ibm.VectorIndexes[ibm.spec.AddedIndexes[spanIdx]]; ok {
+		chunkSize = indexBackfillVectorMergeBatchSize.Get(&ibm.flowCtx.Cfg.Settings.SV)
+	} else {
+		chunkSize = indexBackfillMergeBatchSize.Get(&ibm.flowCtx.Cfg.Settings.SV)
+	}
 	chunkBytes := indexBackfillMergeBatchBytes.Get(&ibm.flowCtx.Cfg.Settings.SV)
 
 	var br *kvpb.BatchResponse

pkg/sql/inspect/inspect_job_test.go

@@ -31,9 +31,6 @@ import (
 // triggered by statements that run in implicit transactions. It verifies that the job
 // starts correctly, that errors or timeouts propagate to the user, and that
 // client-visible semantics (like statement timeout or job failure) behave as expected.
-//
-// Note: This test currently uses SCRUB to trigger a job, but is not testing SCRUB
-// itself. The goal is to verify general execution semantics for async job statements.
 func TestInspectJobImplicitTxnSemantics(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -71,7 +68,7 @@ func TestInspectJobImplicitTxnSemantics(t *testing.T) {
 
 	runner.Exec(t, `
 		CREATE DATABASE db;
-		SET enable_scrub_job = true;
+		SET enable_inspect_command = true;
 		CREATE TABLE db.t (
 			id INT PRIMARY KEY,
 			val INT
@@ -115,7 +112,7 @@ func TestInspectJobImplicitTxnSemantics(t *testing.T) {
 					onInspectErrorToReturn.Store(&tc.onStartError)
 					defer func() { onInspectErrorToReturn.Store(nil) }()
 				}
-				_, err := db.Exec("EXPERIMENTAL SCRUB TABLE db.t AS OF SYSTEM TIME '-1us'")
+				_, err := db.Exec("INSPECT TABLE db.t AS OF SYSTEM TIME '-1us'")
 				pauseJobStart.Store(false)
 				if tc.expectedErrRegex != "" {
 					require.Error(t, err)
@@ -202,7 +199,7 @@ func TestInspectJobProtectedTimestamp(t *testing.T) {
 			runner := sqlutils.MakeSQLRunner(db)
 			runner.Exec(t, `
 				CREATE DATABASE db;
-				SET enable_scrub_job = true;
+				SET enable_inspect_command = true;
 				CREATE TABLE db.t (
 					id INT PRIMARY KEY,
 					val INT
@@ -216,7 +213,7 @@ func TestInspectJobProtectedTimestamp(t *testing.T) {
 			// Start INSPECT job with AS OF timestamp in a goroutine
 			errCh := make(chan error, 1)
 			go func() {
-				_, err := db.Exec("EXPERIMENTAL SCRUB TABLE db.t AS OF SYSTEM TIME '-1us'")
+				_, err := db.Exec("INSPECT TABLE db.t AS OF SYSTEM TIME '-1us'")
 				errCh <- err
 			}()
 
@@ -368,12 +365,11 @@ func TestInspectProgressWithMultiRangeTable(t *testing.T) {
 	})
 
 	// Start the INSPECT job.
-	// TODO(148365): Run INSPECT instead of SCRUB.
 	t.Logf("Starting INSPECT job on table with %d ranges distributed across %d nodes", rangeCount, nodeCount)
 	_, err := db.Exec(`
-		SET enable_scrub_job = true;
+		SET enable_inspect_command = true;
 		COMMIT;
-		EXPERIMENTAL SCRUB TABLE multi_range_table`)
+		INSPECT TABLE multi_range_table`)
 	require.NoError(t, err)
 
 	var jobID int64

pkg/sql/inspect/inspect_metrics_test.go

@@ -35,7 +35,7 @@ func TestInspectMetrics(t *testing.T) {
 	runner := sqlutils.MakeSQLRunner(db)
 	runner.Exec(t, `
 		CREATE DATABASE db;
-		SET enable_scrub_job = true;
+		SET enable_inspect_command = true;
 		CREATE TABLE db.t (
 			id INT PRIMARY KEY,
 			val INT
@@ -52,7 +52,7 @@ func TestInspectMetrics(t *testing.T) {
 	initialIssuesFound := metrics.IssuesFound.Count()
 
 	// First run: no corruption, should succeed without issues
-	runner.Exec(t, "EXPERIMENTAL SCRUB TABLE db.t")
+	runner.Exec(t, "INSPECT TABLE db.t")
 	require.Equal(t, initialRuns+1, metrics.Runs.Count(), "Runs counter should increment")
 	require.Equal(t, initialRunsWithIssues, metrics.RunsWithIssues.Count(), "RunsWithIssues should not increment")
 	require.Equal(t, initialIssuesFound, metrics.IssuesFound.Count(), "IssuesFound should not increment")
@@ -72,7 +72,7 @@ func TestInspectMetrics(t *testing.T) {
 	require.NoError(t, err)
 
 	// Second run: with corruption, should detect the missing index entry.
-	_, err = db.Exec("EXPERIMENTAL SCRUB TABLE db.t")
+	_, err = db.Exec("INSPECT TABLE db.t")
 	require.Error(t, err, "INSPECT should fail when corruption is detected")
 	require.Contains(t, err.Error(), "INSPECT found inconsistencies")
 	require.Equal(t, initialRuns+2, metrics.Runs.Count(),
@@ -91,7 +91,7 @@ func TestInspectMetrics(t *testing.T) {
 		CREATE INDEX i2 on db.t2 (val);
 		INSERT INTO db.t2 VALUES (1, 2), (2, 3);
 	`)
-	runner.Exec(t, "EXPERIMENTAL SCRUB TABLE db.t2")
+	runner.Exec(t, "INSPECT TABLE db.t2")
 	require.Equal(t, initialRuns+3, metrics.Runs.Count(),
 		"Runs counter should increment for third job execution")
 	require.Equal(t, initialRunsWithIssues+1, metrics.RunsWithIssues.Count(),