Merge #144063 #144111 #145274

144063: roachtest: add compaction to backup fixtures r=msbutler a=kev-cao

This commit enables compaction in the generation of backup fixtures.

Epic: None

Release note: None

144111: sql: move BackupRestoreTestingKnobs to exec_util_backup.go r=mgartner a=mgartner

`BackupRestoreTestingKnobs` has been moved to a new file,
`exec_util_backup.go` and `CODEOWNERS` has been updated.

Epic: None
Release note: None


145274: vecindex: fix bugs found during testing of vector indexes r=drewkimball a=andy-kimball

#### cspann: do not overwrite root level when searching targets

In the case where the root partition is in a non-Ready state, we need
to search its target partitions. Previously, we were overwriting the
root partition's level, which should be the same as the target partition's
level. However, if there's a bug, it can be different, and overriding
the root level causes cascading problems downstream, including a panic
that can take down the server. This commit adds code to check the level
and assert that it's equal to the root partition level rather than
silently overwriting it.

#### vecindex: clone treeKey when creating fixups

Previously, the C-SPANN index assumed that it owned the memory for
the vector index prefix value (called the "tree key" in C-SPANN).
However, the backfiller assumed the opposite, and reuses the memory
between SearchForInsert calls. This commit resolves the conflict by
always cloning the tree key when it's enqueued as part of a fixup,
so that callers can reuse the memory.

Fixes: #145261

#### cspann: pass tree key to delete vector fixup

Previously, the delete vector fixup did not take a tree key parameter,
which meant it always searched the default tree for the vector to
delete. This commit fixes that by passing the tree key to the fixup.

Release justification: Vector indexing is a business priority. These changes are all in the vecindex package, for a feature that is protected by a default-off feature flag.

Co-authored-by: Kevin Cao <[email protected]>
Co-authored-by: Marcus Gartner <[email protected]>
Co-authored-by: Andrew Kimball <[email protected]>

Author: 4 people

.github/CODEOWNERS

@@ -73,6 +73,7 @@
 /pkg/sql/control_job*        @cockroachdb/sql-queries-prs @cockroachdb/jobs-prs
 /pkg/sql/job_exec_context*   @cockroachdb/sql-queries-prs @cockroachdb/jobs-prs
 /pkg/sql/delegate/*job*.go   @cockroachdb/jobs-prs @cockroachdb/disaster-recovery
+#!/pkg/sql/BUILD.bazel       @cockroachdb/sql-queries-noreview
 
 /pkg/sql/importer/           @cockroachdb/sql-queries-prs
 /pkg/sql/importer/export*    @cockroachdb/cdc-prs
@@ -258,8 +259,9 @@
 /pkg/ccl/changefeedccl/      @cockroachdb/cdc-prs
 
 /pkg/crosscluster/       @cockroachdb/disaster-recovery
-/pkg/backup/          @cockroachdb/disaster-recovery
-/pkg//backup/*_job.go  @cockroachdb/disaster-recovery @cockroachdb/jobs-prs
+/pkg/backup/                 @cockroachdb/disaster-recovery
+/pkg//backup/*_job.go        @cockroachdb/disaster-recovery @cockroachdb/jobs-prs
+/pkg/sql/exec_util_backup.go @cockroachdb/disaster-recovery
 /pkg/revert/          @cockroachdb/disaster-recovery
 /pkg/ccl/storageccl/         @cockroachdb/disaster-recovery
 /pkg/ccl/cloudccl/           @cockroachdb/disaster-recovery

pkg/cmd/roachtest/tests/backup_fixtures.go

@@ -20,6 +20,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/roachtestutil/clusterupgrade"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/spec"
 	"github.com/cockroachdb/cockroach/pkg/cmd/roachtest/test"
+	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachprod"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/blobfixture"
 	"github.com/cockroachdb/cockroach/pkg/roachprod/install"
@@ -35,6 +37,8 @@ type TpccFixture struct {
 	WorkloadWarehouses     int
 	MinutesPerIncremental  int
 	IncrementalChainLength int
+	CompactionThreshold    int
+	CompactionWindow       int
 	RestoredSizeEstimate   string
 }
 
@@ -45,6 +49,8 @@ var TinyFixture = TpccFixture{
 	ImportWarehouses:       10,
 	WorkloadWarehouses:     10,
 	IncrementalChainLength: 4,
+	CompactionThreshold:    3,
+	CompactionWindow:       2,
 	RestoredSizeEstimate:   "700MiB",
 }
 
@@ -55,6 +61,8 @@ var SmallFixture = TpccFixture{
 	ImportWarehouses:       5000,
 	WorkloadWarehouses:     1000,
 	IncrementalChainLength: 48,
+	CompactionThreshold:    3,
+	CompactionWindow:       2,
 	RestoredSizeEstimate:   "350GiB",
 }
 
@@ -65,6 +73,8 @@ var MediumFixture = TpccFixture{
 	ImportWarehouses:       30000,
 	WorkloadWarehouses:     5000,
 	IncrementalChainLength: 400,
+	CompactionThreshold:    3,
+	CompactionWindow:       2,
 	RestoredSizeEstimate:   "2TiB",
 }
 
@@ -77,6 +87,8 @@ var LargeFixture = TpccFixture{
 	ImportWarehouses:       300000,
 	WorkloadWarehouses:     7500,
 	IncrementalChainLength: 400,
+	CompactionThreshold:    3,
+	CompactionWindow:       2,
 	RestoredSizeEstimate:   "20TiB",
 }
 
@@ -205,37 +217,136 @@ func (bd *backupDriver) runWorkload(ctx context.Context) (func(), error) {
 // scheduleBackups begins the backup schedule.
 func (bd *backupDriver) scheduleBackups(ctx context.Context) {
 	bd.t.L().Printf("creating backup schedule", bd.sp.fixture.WorkloadWarehouses)
-
-	createScheduleStatement := CreateScheduleStatement(bd.registry.URI(bd.fixture.DataPath))
 	conn := bd.c.Conn(ctx, bd.t.L(), 1)
+	defer conn.Close()
+	if bd.sp.fixture.CompactionThreshold > 0 {
+		bd.t.L().Printf(
+			"enabling compaction with threshold %d and window size %d",
+			bd.sp.fixture.CompactionThreshold, bd.sp.fixture.CompactionWindow,
+		)
+		_, err := conn.Exec(fmt.Sprintf(
+			"SET CLUSTER SETTING backup.compaction.threshold = %d", bd.sp.fixture.CompactionThreshold,
+		))
+		require.NoError(bd.t, err)
+		_, err = conn.Exec(fmt.Sprintf(
+			"SET CLUSTER SETTING backup.compaction.window_size = %d", bd.sp.fixture.CompactionWindow,
+		))
+		require.NoError(bd.t, err)
+	}
+	createScheduleStatement := CreateScheduleStatement(bd.registry.URI(bd.fixture.DataPath))
 	_, err := conn.Exec(createScheduleStatement)
 	require.NoError(bd.t, err)
 }
 
 // monitorBackups pauses the schedule once the target number of backups in the
 // chain have been taken.
-func (bd *backupDriver) monitorBackups(ctx context.Context) {
-	sql := sqlutils.MakeSQLRunner(bd.c.Conn(ctx, bd.t.L(), 1))
+func (bd *backupDriver) monitorBackups(ctx context.Context) error {
+	conn := bd.c.Conn(ctx, bd.t.L(), 1)
+	defer conn.Close()
+	sql := sqlutils.MakeSQLRunner(conn)
 	fixtureURI := bd.registry.URI(bd.fixture.DataPath)
-	for {
+	const (
+		WaitingFirstFull = iota
+		RunningIncrementals
+		WaitingCompactions
+		Done
+	)
+	state := WaitingFirstFull
+	for state != Done {
 		time.Sleep(1 * time.Minute)
-		var activeScheduleCount int
-		scheduleCountQuery := fmt.Sprintf(`SELECT count(*) FROM [SHOW SCHEDULES] WHERE label='%s' AND schedule_status='ACTIVE'`, scheduleLabel)
-		sql.QueryRow(bd.t, scheduleCountQuery).Scan(&activeScheduleCount)
-		if activeScheduleCount < 2 {
-			bd.t.L().Printf(`First full backup still running`)
-			continue
+		compSuccess, compRunning, compFailed, err := bd.compactionJobStates(sql)
+		if err != nil {
+			return err
+		}
+		switch state {
+		case WaitingFirstFull:
+			var activeScheduleCount int
+			scheduleCountQuery := fmt.Sprintf(
+				`SELECT count(*) FROM [SHOW SCHEDULES] WHERE label='%s' AND schedule_status='ACTIVE'`, scheduleLabel,
+			)
+			sql.QueryRow(bd.t, scheduleCountQuery).Scan(&activeScheduleCount)
+			if activeScheduleCount < 2 {
+				bd.t.L().Printf(`First full backup still running`)
+			} else {
+				state = RunningIncrementals
+			}
+		case RunningIncrementals:
+			var backupCount int
+			backupCountQuery := fmt.Sprintf(
+				`SELECT count(DISTINCT end_time) FROM [SHOW BACKUP FROM LATEST IN '%s']`, fixtureURI.String(),
+			)
+			sql.QueryRow(bd.t, backupCountQuery).Scan(&backupCount)
+			bd.t.L().Printf(`%d scheduled backups taken`, backupCount)
+
+			if bd.sp.fixture.CompactionThreshold > 0 {
+				bd.t.L().Printf("%d compaction jobs succeeded, %d running", len(compSuccess), len(compRunning))
+				if len(compFailed) > 0 {
+					return errors.Newf("compaction jobs with ids %v failed", compFailed)
+				}
+			}
+
+			if backupCount >= bd.sp.fixture.IncrementalChainLength {
+				pauseSchedulesQuery := fmt.Sprintf(
+					`PAUSE SCHEDULES WITH x AS (SHOW SCHEDULES) SELECT id FROM x WHERE label = '%s'`, scheduleLabel,
+				)
+				sql.Exec(bd.t, pauseSchedulesQuery)
+				if len(compRunning) > 0 {
+					state = WaitingCompactions
+				} else {
+					state = Done
+				}
+			}
+		case WaitingCompactions:
+			if len(compFailed) > 0 {
+				return errors.Newf("compaction jobs with ids %v failed", compFailed)
+			} else if len(compRunning) > 0 {
+				bd.t.L().Printf("waiting for %d compaction jobs to finish", len(compRunning))
+			} else {
+				state = Done
+			}
+		}
+	}
+	return nil
+}
+
+// compactionJobStates returns the state of the compaction jobs, returning
+// the IDs of the jobs that succeeded, are running, and failed.
+func (bd *backupDriver) compactionJobStates(
+	sql *sqlutils.SQLRunner,
+) ([]jobspb.JobID, []jobspb.JobID, []jobspb.JobID, error) {
+	if bd.sp.fixture.CompactionThreshold == 0 {
+		return nil, nil, nil, nil
+	}
+	type Job struct {
+		jobID  jobspb.JobID
+		status jobs.State
+	}
+	compactionQuery := `SELECT job_id, status FROM [SHOW JOBS] WHERE job_type = 'BACKUP' AND
+	description ILIKE 'COMPACT BACKUPS%'`
+	rows := sql.Query(bd.t, compactionQuery)
+	defer rows.Close()
+	var compJobs []Job
+	for rows.Next() {
+		var job Job
+		if err := rows.Scan(&job.jobID, &job.status); err != nil {
+			return nil, nil, nil, errors.Wrapf(err, "error scanning compaction job")
 		}
-		var backupCount int
-		backupCountQuery := fmt.Sprintf(`SELECT count(DISTINCT end_time) FROM [SHOW BACKUP FROM LATEST IN '%s']`, fixtureURI.String())
-		sql.QueryRow(bd.t, backupCountQuery).Scan(&backupCount)
-		bd.t.L().Printf(`%d scheduled backups taken`, backupCount)
-		if backupCount >= bd.sp.fixture.IncrementalChainLength {
-			pauseSchedulesQuery := fmt.Sprintf(`PAUSE SCHEDULES WITH x AS (SHOW SCHEDULES) SELECT id FROM x WHERE label = '%s'`, scheduleLabel)
-			sql.Exec(bd.t, pauseSchedulesQuery)
-			break
+		compJobs = append(compJobs, job)
+	}
+	var successes, running, failures []jobspb.JobID
+	for _, job := range compJobs {
+		switch job.status {
+		case jobs.StateSucceeded:
+			successes = append(successes, job.jobID)
+		case jobs.StateRunning:
+			running = append(running, job.jobID)
+		case jobs.StateFailed:
+			failures = append(failures, job.jobID)
+		default:
+			bd.t.L().Printf(`unexpected compaction job %d in state %s`, job.jobID, job.status)
 		}
 	}
+	return successes, running, failures, nil
 }
 
 func fixtureDirectory() string {
@@ -357,7 +468,7 @@ func registerBackupFixtures(r registry.Registry) {
 				require.NoError(t, err)
 
 				bd.scheduleBackups(ctx)
-				bd.monitorBackups(ctx)
+				require.NoError(t, bd.monitorBackups(ctx))
 
 				stopWorkload()
 

pkg/sql/BUILD.bazel

@@ -114,6 +114,7 @@ go_library(
         "exec_factory_util.go",
         "exec_log.go",
         "exec_util.go",
+        "exec_util_backup.go",
         "execute.go",
         "executor_statement_metrics.go",
         "explain_bundle.go",

pkg/sql/backfill/backfill_test.go

@@ -181,3 +181,63 @@ func TestConcurrentOperationsDuringVectorIndexCreation(t *testing.T) {
 	sqlDB.QueryRow(t, `SELECT id FROM vectors@vec_idx ORDER BY vec <-> '[1, 2, 3]' LIMIT 1`).Scan(&id)
 	require.Equal(t, 1, id)
 }
+
+// Regression for issue #145261: vector index backfill with a prefix column
+// crashes the node.
+func TestVectorIndexWithPrefixBackfill(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	srv, db, _ := serverutils.StartServer(t, base.TestServerArgs{})
+	defer srv.Stopper().Stop(ctx)
+	sqlDB := sqlutils.MakeSQLRunner(db)
+
+	// Enable vector indexes.
+	sqlDB.Exec(t, `SET CLUSTER SETTING feature.vector_index.enabled = true`)
+
+	// Create a table with a vector column + a prefix column.
+	sqlDB.Exec(t, `
+		CREATE TABLE items (
+			id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+			customer_id INT NOT NULL,
+			name TEXT,
+			embedding VECTOR(3)
+		);
+	`)
+
+	// Generate 10 customer id's, each with 100 vectors.
+	sqlDB.Exec(t, `
+		INSERT INTO items (customer_id, name, embedding)
+		SELECT
+			(i % 10) + 1 AS customer_id,
+			'Item ' || i,
+			ARRAY[random(), random(), random()]::vector
+		FROM generate_series(1, 1000) AS s(i);
+	`)
+
+	// Create the vector index with a small partition size so that the trees
+	// have more levels and splits to get there.
+	sqlDB.Exec(t, `
+		CREATE VECTOR INDEX ON items (customer_id, embedding)
+		WITH (min_partition_size=2, max_partition_size=8, build_beam_size=2);
+	`)
+
+	// Ensure that each customer has 100 vectors.
+	for i := range 10 {
+		func() {
+			rows := sqlDB.Query(t, `
+				SELECT id FROM items
+				WHERE customer_id = $1
+				ORDER BY embedding <-> $2
+				LIMIT 200`, i+1, "[0, 0, 0]")
+			defer rows.Close()
+
+			count := 0
+			for rows.Next() {
+				count++
+			}
+			require.Equal(t, 100, count)
+		}()
+	}
+}

pkg/sql/exec_util.go

@@ -21,7 +21,6 @@ import (
 	"time"
 
 	apd "github.com/cockroachdb/apd/v3"
-	"github.com/cockroachdb/cockroach/pkg/backup/backuppb"
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/cloud/externalconn"
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
@@ -41,7 +40,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/rangecache"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/rangefeed"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/rangefeed/rangefeedcache"
-	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvserverbase"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/protectedts"
 	"github.com/cockroachdb/cockroach/pkg/multitenant"
@@ -1901,67 +1899,6 @@ type SchemaTelemetryTestingKnobs struct {
 // ModuleTestingKnobs implements the base.ModuleTestingKnobs interface.
 func (*SchemaTelemetryTestingKnobs) ModuleTestingKnobs() {}
 
-// BackupRestoreTestingKnobs contains knobs for backup and restore behavior.
-//
-// TODO (msbutler): move these to backup
-type BackupRestoreTestingKnobs struct {
-	// AfterBackupChunk is called after each chunk of a backup is completed.
-	AfterBackupChunk func()
-
-	// AfterBackupCheckpoint if set will be called after a BACKUP-CHECKPOINT
-	// is written.
-	AfterBackupCheckpoint func()
-
-	// AfterLoadingCompactionManifestOnResume is run once the backup manifest has been
-	// loaded/created on the resumption of a compaction job.
-	AfterLoadingCompactionManifestOnResume func(manifest *backuppb.BackupManifest)
-
-	// CaptureResolvedTableDescSpans allows for intercepting the spans which are
-	// resolved during backup planning, and will eventually be backed up during
-	// execution.
-	CaptureResolvedTableDescSpans func([]roachpb.Span)
-
-	// RunAfterSplitAndScatteringEntry allows blocking the RESTORE job after a
-	// single RestoreSpanEntry has been split and scattered.
-	RunAfterSplitAndScatteringEntry func(ctx context.Context)
-
-	// RunAfterProcessingRestoreSpanEntry allows blocking the RESTORE job after a
-	// single RestoreSpanEntry has been processed and added to the SSTBatcher.
-	RunAfterProcessingRestoreSpanEntry func(ctx context.Context, entry *execinfrapb.RestoreSpanEntry) error
-
-	// RunAfterExportingSpanEntry allows blocking the BACKUP job after a single
-	// span has been exported.
-	RunAfterExportingSpanEntry func(ctx context.Context, response *kvpb.ExportResponse)
-
-	// BackupMonitor is used to overwrite the monitor used by backup during
-	// testing. This is typically the bulk mem monitor if not
-	// specified here.
-	BackupMemMonitor *mon.BytesMonitor
-
-	RestoreDistSQLRetryPolicy *retry.Options
-
-	RunBeforeRestoreFlow func() error
-
-	RunAfterRestoreFlow func() error
-
-	BackupDistSQLRetryPolicy *retry.Options
-
-	RunBeforeBackupFlow func() error
-
-	RunAfterBackupFlow func() error
-
-	RunAfterRetryIteration func(err error) error
-
-	RunAfterRestoreProcDrains func()
-
-	RunBeforeResolvingCompactionDest func() error
-}
-
-var _ base.ModuleTestingKnobs = &BackupRestoreTestingKnobs{}
-
-// ModuleTestingKnobs implements the base.ModuleTestingKnobs interface.
-func (*BackupRestoreTestingKnobs) ModuleTestingKnobs() {}
-
 // StreamingTestingKnobs contains knobs for streaming behavior.
 type StreamingTestingKnobs struct {
 	// RunAfterReceivingEvent allows blocking the stream ingestion processor after