feat: update FindPriorBackups to use backup index

This patch teaches FindPriorBackups (now more aptly renamed to
FindAllIncrementals) to use the backup index introduced in #150384.
Additional changes will still be required to fully fix the related
tickets, namely updating the logic to no longer need to load ALL
manifests of a chain into memory, only the manifests needed.

Epic: CRDB-47942

Informs: #150327, #150329, #150330, #150331

Release note: None

Author: kev-cao

pkg/backup/backup_test.go

@@ -9965,108 +9965,6 @@ func TestUserfileNormalizationIncrementalShowBackup(t *testing.T) {
 	sqlDB.Exec(t, query)
 }
 
-// TestBackupRestoreSeparateExplicitIsDefault tests that a backup/restore round
-// trip using the 'incremental_location' parameter restores the same db as a BR
-// round trip without the parameter, even when that location is in fact the default.
-func TestBackupRestoreSeparateExplicitIsDefault(t *testing.T) {
-	defer leaktest.AfterTest(t)()
-	defer log.Scope(t).Close(t)
-
-	skip.UnderRace(t, "multinode cluster setup times out under race, likely due to resource starvation.")
-
-	const numAccounts = 1
-	_, sqlDB, _, cleanupFn := backupRestoreTestSetup(t, multiNode, numAccounts, InitManualReplication)
-	defer cleanupFn()
-
-	const c1, c2, c3 = `nodelocal://1/full/`, `nodelocal://2/full/`, `nodelocal://3/full/`
-	const i1, i2, i3 = `nodelocal://1/full/incrementals`, `nodelocal://2/full/incrementals`, `nodelocal://3/full/incrementals`
-
-	collections := []string{
-		fmt.Sprintf("'%s?COCKROACH_LOCALITY=%s'", c1, url.QueryEscape("default")),
-		fmt.Sprintf("'%s?COCKROACH_LOCALITY=%s'", c2, url.QueryEscape("dc=dc1")),
-		fmt.Sprintf("'%s?COCKROACH_LOCALITY=%s'", c3, url.QueryEscape("dc=dc2")),
-	}
-
-	incrementals := []string{
-		fmt.Sprintf("'%s?COCKROACH_LOCALITY=%s'", i1, url.QueryEscape("default")),
-		fmt.Sprintf("'%s?COCKROACH_LOCALITY=%s'", i2, url.QueryEscape("dc=dc1")),
-		fmt.Sprintf("'%s?COCKROACH_LOCALITY=%s'", i3, url.QueryEscape("dc=dc2")),
-	}
-	tests := []struct {
-		dest []string
-		inc  []string
-	}{
-		{dest: []string{collections[0]}, inc: []string{incrementals[0]}},
-		{dest: collections, inc: incrementals},
-	}
-
-	for _, br := range tests {
-
-		dest := strings.Join(br.dest, ", ")
-		inc := strings.Join(br.inc, ", ")
-
-		if len(br.dest) > 1 {
-			dest = "(" + dest + ")"
-			inc = "(" + inc + ")"
-		}
-		// create db
-		sqlDB.Exec(t, `CREATE DATABASE fkdb`)
-		sqlDB.Exec(t, `CREATE TABLE fkdb.fk (ind INT)`)
-
-		for i := 0; i < 10; i++ {
-			sqlDB.Exec(t, `INSERT INTO fkdb.fk (ind) VALUES ($1)`, i)
-		}
-		fb := fmt.Sprintf("BACKUP DATABASE fkdb INTO %s", dest)
-		sqlDB.Exec(t, fb)
-
-		sqlDB.Exec(t, `INSERT INTO fkdb.fk (ind) VALUES ($1)`, 200)
-
-		sib := fmt.Sprintf("BACKUP DATABASE fkdb INTO LATEST IN %s WITH incremental_location=%s", dest, inc)
-		sqlDB.Exec(t, sib)
-		{
-			// Locality Aware Show Backup validation
-			// TODO (msbutler): move to data driven test after 22.1 backport
-
-			// Assert the localities field populates correctly (not null if backup is locality aware).
-			localities := sqlDB.QueryStr(t,
-				fmt.Sprintf("SELECT locality FROM [SHOW BACKUP FILES FROM LATEST IN %s]", dest))
-			expectedLocalities := map[string]bool{"default": true, "dc=dc1": true, "dc=dc2": true}
-			for _, locality := range localities {
-				if len(br.dest) > 1 {
-					_, ok := expectedLocalities[locality[0]]
-					require.Equal(t, true, ok)
-				} else {
-					require.Equal(t, "NULL", locality[0])
-				}
-			}
-			// Assert show backup still works.
-			sqlDB.Exec(t, fmt.Sprintf("SHOW BACKUPS IN %s", dest))
-			sqlDB.Exec(t, fmt.Sprintf("SHOW BACKUP FROM LATEST IN %s", dest))
-
-			// Locality aware show backups will eventually fail if not all localities are provided,
-			// but for now, they're ok.
-			if len(br.dest) > 1 {
-				sqlDB.Exec(t, fmt.Sprintf("SHOW BACKUP FROM LATEST IN %s", br.dest[1]))
-			}
-
-			sqlDB.Exec(t, fmt.Sprintf("SHOW BACKUP FROM LATEST IN %s WITH incremental_location= %s", dest, inc))
-		}
-		sir := fmt.Sprintf("RESTORE DATABASE fkdb FROM LATEST IN %s WITH new_db_name = 'inc_fkdb', incremental_location=%s", dest, inc)
-		sqlDB.Exec(t, sir)
-
-		ib := fmt.Sprintf("BACKUP DATABASE fkdb INTO LATEST IN %s", dest)
-		sqlDB.Exec(t, ib)
-		ir := fmt.Sprintf("RESTORE DATABASE fkdb FROM LATEST IN %s WITH new_db_name = 'trad_fkdb'", dest)
-		sqlDB.Exec(t, ir)
-
-		sqlDB.CheckQueryResults(t, `SELECT * FROM trad_fkdb.fk`, sqlDB.QueryStr(t, `SELECT * FROM inc_fkdb.fk`))
-
-		sqlDB.Exec(t, "DROP DATABASE fkdb")
-		sqlDB.Exec(t, "DROP DATABASE trad_fkdb;")
-		sqlDB.Exec(t, "DROP DATABASE inc_fkdb;")
-	}
-}
-
 // TestRestoreOnFailOrCancelAfterPause is a regression test that ensures that we
 // can cancel a paused restore job. Previously, this test would result in a nil
 // pointer exception when accessing the `r.execCfg` since the resumer was not

pkg/backup/backupdest/BUILD.bazel

@@ -26,10 +26,12 @@ go_library(
         "//pkg/sql",
         "//pkg/sql/pgwire/pgcode",
         "//pkg/sql/pgwire/pgerror",
+        "//pkg/util",
         "//pkg/util/ctxgroup",
         "//pkg/util/encoding",
         "//pkg/util/hlc",
         "//pkg/util/ioctx",
+        "//pkg/util/metamorphic",
         "//pkg/util/mon",
         "//pkg/util/protoutil",
         "//pkg/util/timeutil",

pkg/backup/backupdest/backup_destination.go

@@ -219,7 +219,15 @@ func ResolveDest(
 	}
 	defer incrementalStore.Close()
 
-	priors, err := FindPriorBackups(ctx, incrementalStore, OmitManifest)
+	rootStore, err := makeCloudStorage(ctx, collectionURI, user)
+	if err != nil {
+		return ResolvedDestination{}, err
+	}
+	defer rootStore.Close()
+	priors, err := FindAllIncrementalPaths(
+		ctx, execCfg, incrementalStore, rootStore,
+		chosenSuffix, OmitManifest, len(dest.IncrementalStorage) > 0,
+	)
 	if err != nil {
 		return ResolvedDestination{}, errors.Wrap(err, "adjusting backup destination to append new layer to existing backup")
 	}
@@ -537,10 +545,13 @@ func ListFullBackupsInCollection(
 // included in the result, otherwise they are filtered out.
 func ResolveBackupManifests(
 	ctx context.Context,
+	execCfg *sql.ExecutorConfig,
 	mem *mon.BoundAccount,
+	defaultCollectionURI string,
 	baseStores []cloud.ExternalStorage,
 	incStores []cloud.ExternalStorage,
 	mkStore cloud.ExternalStorageFromURIFactory,
+	resolvedSubdir string,
 	fullyResolvedBaseDirectory []string,
 	fullyResolvedIncrementalsDirectory []string,
 	endTime hlc.Timestamp,
@@ -575,7 +586,23 @@ func ResolveBackupManifests(
 
 	var incrementalBackups []string
 	if len(incStores) > 0 {
-		incrementalBackups, err = FindPriorBackups(ctx, incStores[0], includeManifest)
+		rootStore, err := mkStore(ctx, defaultCollectionURI, user)
+		if err != nil {
+			return nil, nil, nil, 0, err
+		}
+		defer rootStore.Close()
+		incrementalBackups, err = FindAllIncrementalPaths(
+			ctx,
+			execCfg,
+			incStores[0],
+			rootStore,
+			resolvedSubdir,
+			includeManifest,
+			// In the legacy path, there is no index to be used, so we may as well set
+			// customIncLocation to true rather than try to compute it from the
+			// parameters.
+			true,
+		)
 		if err != nil {
 			return nil, nil, nil, 0, err
 		}

pkg/backup/backupdest/backup_index.go

@@ -21,14 +21,25 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/metamorphic"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
 
+var (
+	ReadBackupIndexEnabled = settings.RegisterBoolSetting(
+		settings.ApplicationLevel,
+		"backup.index.read.enabled",
+		"if true, the backup index will be read when reading from a backup collection",
+		metamorphic.ConstantWithTestBool("backup.index.read.enabled", false),
+	)
+)
+
 // WriteBackupIndexMetadata writes an index file for the backup described by the
 // job details. The provided ExternalStorage needs to be rooted at the specific
 // directory that the index file should be written to.
@@ -68,7 +79,7 @@ func WriteBackupIndexMetadata(
 
 	path, err := backuputils.AbsoluteBackupPathInCollectionURI(details.CollectionURI, details.URI)
 	if err != nil {
-		return errors.Wrapf(err, "get relative backup path")
+		return err
 	}
 	metadata := &backuppb.BackupIndexMetadata{
 		StartTime: details.StartTime,
@@ -262,6 +273,28 @@ func GetBackupTreeIndexMetadata(
 	return indexes[:coveringIdx], nil
 }
 
+// ParseBackupFilePathFromIndexFileName parses the path to a backup given the
+// basename of its index file and its subdirectory. For full backups, the
+// returned path is relative to the root of the backup. For incremental
+// backups, the path is relative to the incremental storage location starting
+// from the subdir (e.g. 20250730/130000.00-20250730-120000.00). It expects
+// that subdir is resolved and not `LATEST`.
+//
+// Note: While the path is stored in the index file, we can take a shortcut here
+// and derive it from the filename solely because backup paths are
+// millisecond-precise and so are the timestamps encoded in the filename.
+func parseBackupFilePathFromIndexFileName(subdir, basename string) (string, error) {
+	start, end, err := parseIndexFilename(basename)
+	if err != nil {
+		return "", err
+	}
+	if start.IsZero() {
+		return subdir, nil
+	}
+
+	return ConstructDateBasedIncrementalFolderName(start, end), nil
+}
+
 // ParseIndexFilename parses the start and end timestamps from the index
 // filename.
 //

pkg/backup/backupdest/backup_index_test.go

@@ -12,7 +12,6 @@ import (
 	"fmt"
 	"io"
 	"maps"
-	"math/rand/v2"
 	"os"
 	"path"
 	"slices"
@@ -484,103 +483,6 @@ func TestIndexExists(t *testing.T) {
 	}
 }
 
-func TestListIndexes(t *testing.T) {
-	defer leaktest.AfterTest(t)()
-	defer log.Scope(t).Close(t)
-
-	ctx := context.Background()
-	dir, dirCleanup := testutils.TempDir(t)
-	defer dirCleanup()
-
-	st := cluster.MakeTestingClusterSettingsWithVersions(
-		clusterversion.Latest.Version(),
-		clusterversion.Latest.Version(),
-		true,
-	)
-	execCfg := &sql.ExecutorConfig{Settings: st}
-
-	const collectionURI = "nodelocal://1/test"
-	storage, err := cloud.ExternalStorageFromURI(
-		ctx,
-		collectionURI,
-		base.ExternalIODirConfig{},
-		st,
-		blobs.TestBlobServiceClient(dir),
-		username.RootUserName(),
-		nil, /* db */
-		nil, /* limiters */
-		cloud.NilMetrics,
-	)
-	require.NoError(t, err)
-	defer storage.Close()
-	storageFactory := func(
-		_ context.Context, _ string, _ username.SQLUsername, _ ...cloud.ExternalStorageOption,
-	) (cloud.ExternalStorage, error) {
-		return storage, nil
-	}
-
-	fullEnd := intToTime(2)
-	fullSubdir := fullEnd.GoTime().Format(backupbase.DateBasedIntoFolderName)
-	details := jobspb.BackupDetails{
-		Destination: jobspb.BackupDetails_Destination{
-			To:     []string{collectionURI},
-			Subdir: fullSubdir,
-		},
-		StartTime:     hlc.Timestamp{},
-		EndTime:       fullEnd,
-		CollectionURI: collectionURI,
-		// URI does not need to be set properly for this test since we are not
-		// reading the contents of the files.
-		URI: collectionURI + "/" + fullSubdir,
-	}
-	require.NoError(t, WriteBackupIndexMetadata(
-		ctx, execCfg, username.RootUserName(), storageFactory, details,
-	))
-
-	incsToWrite := [][2]int{
-		{2, 4},
-		{2, 6},
-		{4, 6},
-		{6, 8},
-	}
-	// We shuffle the order that we write backups in just to really stress the
-	// ordering logic in ListIndexes.
-	rand.Shuffle(len(incsToWrite), func(i, j int) {
-		incsToWrite[i], incsToWrite[j] = incsToWrite[j], incsToWrite[i]
-	})
-	for _, times := range incsToWrite {
-		start := intToTime(times[0])
-		end := intToTime(times[1])
-		details.StartTime = start
-		details.EndTime = end
-		require.NoError(
-			t, WriteBackupIndexMetadata(ctx, execCfg, username.RootUserName(), storageFactory, details),
-		)
-	}
-
-	indexes, err := ListIndexes(ctx, storage, fullSubdir)
-	require.NoError(t, err)
-
-	require.Len(t, indexes, len(incsToWrite)+1)
-	require.True(t, slices.IsSortedFunc(indexes, func(a, b string) int {
-		aStart, aEnd, err := parseIndexFilename(a)
-		require.NoError(t, err)
-		bStart, bEnd, err := parseIndexFilename(b)
-		require.NoError(t, err)
-		if aEnd.Before(bEnd) {
-			return -1
-		} else if aEnd.After(bEnd) {
-			return 1
-		}
-		// end times are equal, compare start times
-		if bStart.Before(aStart) {
-			return 1
-		} else {
-			return -1
-		}
-	}), "indexes are not sorted by end time and then start time")
-}
-
 func TestGetBackupTreeIndexMetadata(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)