backup: teach SHOW BACKUPS to use backup index

`SHOW BACKUPS` previously would find all full backups by delimiting on
the `data/` prefix and listing all objects from the root store. This was
not very performant as while the data SSTs in all the backups would not
be listed, all of the other files would be. This commit teaches `SHOW
BACKUPS` to use the new backup index introduced in #150384 when listing
backups. This method relies on listing the top-level contents in the
`index/` directory and delimiting on `/`, which means only the required
objects are listed.

This behavior is gated behind the `WITH INDEX` option that has been
added to `SHOW BACKUPS`. This is because in 25.4, it is still possible
to restore from backups made before the backup index was added. Since
the new behavior relies entirely on the `index/`, we would miss any
backups that do not have a corresponding index item created. To capture
all backups in a collection, we'd need to list both from the `index/`
and via the legacy path, which would be counterintuitive. Instead, we
will instead rely on `WITH INDEX` until v26.1, at which point all
restorable backups should have an index created for them. At that time,
we will make `WITH INDEX` the default and use `WITH LEGACY LISTING` to
list older backups.

Epic: CRDB-47942

Fixes: #150328

Release note (general change): Added `WITH INDEX` option to `SHOW
BACKUPS` for faster listing of 25.4+ backups.

Author: kev-cao

docs/generated/sql/bnf/show_backup.bnf

@@ -1,5 +1,5 @@
 show_backup_stmt ::=
-	'SHOW' 'BACKUPS' 'IN' collectionURI
+	'SHOW' 'BACKUPS' 'IN' collectionURI opt_with_show_backups_options
 	| 'SHOW' 'BACKUP' 'SCHEMAS' 'FROM' subdirectory 'IN' collectionURI 'WITH' show_backup_options ( ( ',' show_backup_options ) )*
 	| 'SHOW' 'BACKUP' 'SCHEMAS' 'FROM' subdirectory 'IN' collectionURI 'WITH' 'OPTIONS' '(' show_backup_options ( ( ',' show_backup_options ) )* ')'
 	| 'SHOW' 'BACKUP' 'SCHEMAS' 'FROM' subdirectory 'IN' collectionURI 

docs/generated/sql/bnf/stmt_block.bnf

@@ -840,7 +840,7 @@ use_stmt ::=
 	'USE' var_value
 
 show_backup_stmt ::=
-	'SHOW' 'BACKUPS' 'IN' string_or_placeholder_opt_list
+	'SHOW' 'BACKUPS' 'IN' string_or_placeholder_opt_list opt_with_show_backups_options
 	| 'SHOW' 'BACKUP' show_backup_details 'FROM' string_or_placeholder 'IN' string_or_placeholder_opt_list opt_with_show_backup_options
 	| 'SHOW' 'BACKUP' string_or_placeholder 'IN' string_or_placeholder_opt_list opt_with_show_backup_options
 
@@ -2114,6 +2114,11 @@ var_value ::=
 	a_expr
 	| extra_var_value
 
+opt_with_show_backups_options ::=
+	'WITH' show_backups_options_list
+	| 'WITH' 'OPTIONS' '(' show_backups_options_list ')'
+	| 
+
 show_backup_details ::=
 	'SCHEMAS'
 
@@ -3003,6 +3008,9 @@ extra_var_value ::=
 	| 'NONE'
 	| cockroachdb_extra_reserved_keyword
 
+show_backups_options_list ::=
+	( show_backups_options ) ( ( ',' show_backups_options ) )*
+
 show_backup_options_list ::=
 	( show_backup_options ) ( ( ',' show_backup_options ) )*
 
@@ -3633,6 +3641,9 @@ for_locking_item ::=
 var_list ::=
 	( var_value ) ( ( ',' var_value ) )*
 
+show_backups_options ::=
+	'INDEX'
+
 show_backup_options ::=
 	'AS_JSON'
 	| 'CHECK_FILES'

pkg/backup/backup_test.go

@@ -1759,30 +1759,34 @@ func TestBackupRestoreResume(t *testing.T) {
 				if err != nil {
 					t.Fatal(err)
 				}
-				backupDir := dir + "/backup" + "-" + item.testName
+				collectionURI := fmt.Sprintf("nodelocal://1/%s", item.testName)
+				subdir := time.Now().Format(backupbase.DateBasedIntoFolderName)
 
-				if err := os.MkdirAll(backupDir+item.checkpointDirectory, 0755); err != nil {
+				checkpointDir := path.Join(dir, item.testName, subdir, item.checkpointDirectory)
+				if err := os.MkdirAll(checkpointDir, 0755); err != nil {
 					t.Fatal(err)
 				}
-				checkpointFile := backupDir + item.checkpointDirectory + "/" + backupinfo.BackupManifestCheckpointName
+				checkpointFile := path.Join(checkpointDir, backupinfo.BackupManifestCheckpointName)
 				if err := os.WriteFile(checkpointFile, mockManifest, 0644); err != nil {
 					t.Fatal(err)
 				}
-				uri := "nodelocal://1/backup" + "-" + item.testName
 				createAndWaitForJob(
 					t, sqlDB, []descpb.ID{backupTableDesc.GetID()},
 					jobspb.BackupDetails{
+						Destination: jobspb.BackupDetails_Destination{
+							Subdir: subdir,
+						},
 						EndTime:       srv.Clock().Now(),
-						CollectionURI: uri,
-						URI:           uri,
+						CollectionURI: collectionURI,
+						URI:           fmt.Sprintf("%s%s", collectionURI, subdir),
 					},
 					jobspb.BackupProgress{},
 					roachpb.Version{},
 				)
 
 				// If the backup properly took the (incorrect) checkpoint into account, it
 				// won't have tried to re-export any keys within backupCompletedSpan.
-				backupManifestFile := backupDir + "/" + backupbase.BackupManifestName
+				backupManifestFile := path.Join(dir, item.testName, subdir, backupbase.BackupManifestName)
 				backupManifestBytes, err := os.ReadFile(backupManifestFile)
 				if err != nil {
 					t.Fatal(err)

pkg/backup/backupbase/constants.go

@@ -68,6 +68,10 @@ const (
 	// to the directory containing the index files for the backup collection.
 	BackupIndexDirectoryPath = "index/"
 
+	// BackupIndexFlattenedSubdir is the format used for the top-level
+	// subdirectories within the index directory.
+	BackupIndexFlattenedSubdir = "2006-01-02-150405.00"
+
 	// BackupIndexFilenameTimestampFormat is the format used for the human
 	// readable start and end times in the index file names.
 	BackupIndexFilenameTimestampFormat = "20060102-150405.00"

pkg/backup/backupdest/backup_destination.go

@@ -507,8 +507,12 @@ func GetURIsByLocalityKV(
 // ListFullBackupsInCollection lists full backup paths in the collection
 // of an export store
 func ListFullBackupsInCollection(
-	ctx context.Context, store cloud.ExternalStorage,
+	ctx context.Context, store cloud.ExternalStorage, useIndex bool,
 ) ([]string, error) {
+	if useIndex {
+		return ListSubdirsFromIndex(ctx, store)
+	}
+
 	var backupPaths []string
 	if err := store.List(ctx, "", backupbase.ListingDelimDataSlash, func(f string) error {
 		if backupPathRE.MatchString(f) {

pkg/backup/backupdest/backup_index.go

@@ -109,9 +109,13 @@ func WriteBackupIndexMetadata(
 // words, we can remove these checks in v26.2+.
 func IndexExists(ctx context.Context, store cloud.ExternalStorage, subdir string) (bool, error) {
 	var indexExists bool
+	indexDir, err := indexSubdir(subdir)
+	if err != nil {
+		return false, err
+	}
 	if err := store.List(
 		ctx,
-		indexSubdir(subdir),
+		indexDir,
 		"/",
 		func(file string) error {
 			indexExists = true
@@ -139,9 +143,13 @@ func ListIndexes(
 	ctx context.Context, store cloud.ExternalStorage, subdir string,
 ) ([]string, error) {
 	var indexBasenames []string
+	indexDir, err := indexSubdir(subdir)
+	if err != nil {
+		return nil, err
+	}
 	if err := store.List(
 		ctx,
-		indexSubdir(subdir)+"/",
+		indexDir+"/",
 		"",
 		func(file string) error {
 			indexBasenames = append(indexBasenames, path.Base(file))
@@ -215,8 +223,12 @@ func GetBackupTreeIndexMetadata(
 	g := ctxgroup.WithContext(ctx)
 	for i, basename := range indexBasenames {
 		g.GoCtx(func(ctx context.Context) error {
+			indexDir, err := indexSubdir(subdir)
+			if err != nil {
+				return err
+			}
 			reader, size, err := store.ReadFile(
-				ctx, path.Join(indexSubdir(subdir), basename), cloud.ReadOptions{},
+				ctx, path.Join(indexDir, basename), cloud.ReadOptions{},
 			)
 			if err != nil {
 				return errors.Wrapf(err, "reading index file %s", basename)
@@ -292,6 +304,30 @@ func parseIndexFilename(basename string) (start time.Time, end time.Time, err er
 	return start, end, nil
 }
 
+// ListSubdirsFromIndex lists the paths of all full backup subdirectories that
+// have an entry in the index. The store should be rooted at the default
+// collection URI. The subdirs are returned in chronological order.
+func ListSubdirsFromIndex(ctx context.Context, store cloud.ExternalStorage) ([]string, error) {
+	var subdirs []string
+	if err := store.List(
+		ctx,
+		backupbase.BackupIndexDirectoryPath,
+		"/",
+		func(indexSubdir string) error {
+			indexSubdir = strings.TrimSuffix(indexSubdir, "/")
+			subdir, err := unflattenIndexSubdir(indexSubdir)
+			if err != nil {
+				return err
+			}
+			subdirs = append(subdirs, subdir)
+			return nil
+		},
+	); err != nil {
+		return nil, errors.Wrapf(err, "listing index subdirs")
+	}
+	return subdirs, nil
+}
+
 // shouldWriteIndex determines if a backup index file should be written for a
 // given backup. The rule is:
 //  1. An index should only be written on a v25.4+ cluster.
@@ -338,8 +374,12 @@ func getBackupIndexFilePath(subdir string, startTime, endTime hlc.Timestamp) (st
 	if strings.EqualFold(subdir, backupbase.LatestFileName) {
 		return "", errors.AssertionFailedf("expected subdir to be resolved and not be 'LATEST'")
 	}
+	indexDir, err := indexSubdir(subdir)
+	if err != nil {
+		return "", err
+	}
 	return backuputils.JoinURLPath(
-		indexSubdir(subdir),
+		indexDir,
 		getBackupIndexFileName(startTime, endTime),
 	), nil
 }
@@ -364,8 +404,12 @@ func getBackupIndexFileName(startTime, endTime hlc.Timestamp) string {
 // path for a given full backup subdir. The path is relative to the root of the
 // collection URI and does not contain a trailing slash. It assumes that subdir
 // has been resolved and is not `LATEST`.
-func indexSubdir(subdir string) string {
-	return path.Join(backupbase.BackupIndexDirectoryPath, flattenSubdirForIndex(subdir))
+func indexSubdir(subdir string) (string, error) {
+	flattened, err := flattenSubdirForIndex(subdir)
+	if err != nil {
+		return "", err
+	}
+	return path.Join(backupbase.BackupIndexDirectoryPath, flattened), nil
 }
 
 // flattenSubdirForIndex flattens a full backup subdirectory to be used in the
@@ -385,11 +429,21 @@ func indexSubdir(subdir string) string {
 //
 // Listing on `index/` and delimiting on `/` will return the subdirectories
 // without listing the files in them.
-func flattenSubdirForIndex(subdir string) string {
-	return strings.ReplaceAll(
-		// Trimming any trailing and leading slashes guarantees a specific format when
-		// returning the flattened subdir, so callers can expect a consistent result.
-		strings.TrimSuffix(strings.TrimPrefix(subdir, "/"), "/"),
-		"/", "-",
-	)
+func flattenSubdirForIndex(subdir string) (string, error) {
+	subdirTime, err := time.Parse(backupbase.DateBasedIntoFolderName, subdir)
+	if err != nil {
+		return "", errors.Wrapf(err, "parsing subdir %q for flattening", subdir)
+	}
+	return subdirTime.Format(backupbase.BackupIndexFlattenedSubdir), nil
+}
+
+// unflattenIndexSubdir is the inverse of flattenSubdirForIndex. It converts a
+// flattened index subdir back to the original full backup subdir.
+func unflattenIndexSubdir(flattened string) (string, error) {
+	subdirTime, err := time.Parse(backupbase.BackupIndexFlattenedSubdir, flattened)
+	if err != nil {
+		return "", errors.Wrapf(err, "parsing flattened index subdir %q for unflattening", flattened)
+	}
+	unflattened := subdirTime.Format(backupbase.DateBasedIntoFolderName)
+	return unflattened, nil
 }

pkg/backup/backupdest/backup_index_test.go

@@ -262,7 +262,7 @@ func TestDontWriteBackupIndexMetadata(t *testing.T) {
 		return externalStorage, nil
 	}
 
-	subdir := "2025/07/18-143826.00"
+	subdir := "/2025/07/18-143826.00"
 	details := jobspb.BackupDetails{
 		Destination: jobspb.BackupDetails_Destination{
 			To:     []string{"nodelocal://1/backup"},
@@ -345,8 +345,8 @@ func TestIndexExists(t *testing.T) {
 	ctx := context.Background()
 
 	const collectionURI = "nodelocal://1/backup"
-	const subdir1 = "/2025/07/18-222500.00/"
-	const subdir2 = "/2025/07/19-123456.00/"
+	const subdir1 = "/2025/07/18-222500.00"
+	const subdir2 = "/2025/07/19-123456.00"
 	st := cluster.MakeTestingClusterSettingsWithVersions(
 		clusterversion.Latest.Version(),
 		clusterversion.Latest.Version(),

pkg/backup/show.go

@@ -1392,7 +1392,7 @@ func showBackupsInCollectionPlanHook(
 			return errors.Wrapf(err, "connect to external storage")
 		}
 		defer store.Close()
-		res, err := backupdest.ListFullBackupsInCollection(ctx, store)
+		res, err := backupdest.ListFullBackupsInCollection(ctx, store, showStmt.Options.Index)
 		if err != nil {
 			return err
 		}

pkg/backup/show_test.go

@@ -455,9 +455,11 @@ func TestShowBackups(t *testing.T) {
 	sqlDB.Exec(t, `BACKUP data.bank INTO LATEST IN $1 WITH incremental_location = $2`, full, remoteInc)
 
 	rows := sqlDBRestore.QueryStr(t, `SHOW BACKUPS IN $1`, full)
+	rowsUsingIndex := sqlDBRestore.QueryStr(t, `SHOW BACKUPS IN $1 WITH INDEX`, full)
 
 	// assert that we see the three, and only the three, full backups.
 	require.Equal(t, 3, len(rows))
+	require.Equal(t, rows, rowsUsingIndex)
 
 	// check that we can show the inc layers in the individual full backups.
 	b1 := sqlDBRestore.QueryStr(t, `SELECT * FROM [SHOW BACKUP FROM $1 IN $2] WHERE object_type='table'`, rows[0][0], full)

pkg/sql/parser/sql.y

@@ -1444,7 +1444,7 @@ func (u *sqlSymUnion) doBlockOption() tree.DoBlockOption {
 %type <*tree.TenantReplicationOptions> opt_with_replication_options replication_options replication_options_list source_replication_options source_replication_options_list
 %type <tree.ShowBackupDetails> show_backup_details
 %type <*tree.ShowJobOptions> show_job_options show_job_options_list
-%type <*tree.ShowBackupOptions> opt_with_show_backup_options show_backup_options show_backup_options_list
+%type <*tree.ShowBackupOptions> opt_with_show_backup_options show_backup_options show_backup_options_list opt_with_show_backups_options show_backups_options show_backups_options_list
 %type <*tree.CopyOptions> opt_with_copy_options copy_options copy_options_list copy_generic_options copy_generic_options_list
 %type <str> import_format
 %type <str> storage_parameter_key
@@ -3839,7 +3839,7 @@ alter_external_connection_stmt:
 				 ConnectionLabelSpec: *($4.labelSpec()),
 		     As: $6.expr(),
 		}
-	} 
+	}
 | ALTER EXTERNAL CONNECTION IF EXISTS /*$6=*/label_spec AS /*$8=*/string_or_placeholder
 	{
 		   $$.val = &tree.AlterExternalConnection{
@@ -8771,10 +8771,11 @@ show_histogram_stmt:
 // %Text: SHOW BACKUP [SCHEMAS|FILES|RANGES] <location>
 // %SeeAlso: WEBDOCS/show-backup.html
 show_backup_stmt:
-  SHOW BACKUPS IN string_or_placeholder_opt_list
+  SHOW BACKUPS IN string_or_placeholder_opt_list opt_with_show_backups_options
  {
     $$.val = &tree.ShowBackup{
       InCollection:    $4.stringOrPlaceholderOptList(),
+      Options: *$5.showBackupOptions(),
     }
   }
 | SHOW BACKUP show_backup_details FROM string_or_placeholder IN string_or_placeholder_opt_list opt_with_show_backup_options
@@ -8857,6 +8858,38 @@ show_backup_details:
 	$$.val = tree.BackupValidateDetails
 	}
 
+opt_with_show_backups_options:
+  WITH show_backups_options_list
+  {
+    $$.val = $2.showBackupOptions()
+  }
+| WITH OPTIONS '(' show_backups_options_list ')'
+  {
+    $$.val = $4.showBackupOptions()
+  }
+| /* EMPTY */
+  {
+    $$.val = &tree.ShowBackupOptions{}
+  }
+
+show_backups_options_list:
+  show_backups_options
+  {
+    $$.val = $1.showBackupOptions()
+  }
+| show_backups_options_list ',' show_backups_options
+  {
+    if err := $1.showBackupOptions().CombineWith($3.showBackupOptions()); err != nil {
+      return setErr(sqllex, err)
+    }
+  }
+
+show_backups_options:
+ INDEX
+ {
+    $$.val = &tree.ShowBackupOptions{Index: true}
+ }
+
 opt_with_show_backup_options:
   WITH show_backup_options_list
   {