backup: write index full subdirs in reverse chronological order

pkg/backup/backupbase/constants.go

@@ -63,11 +63,7 @@ const (
 
 	// BackupIndexDirectoryName is the path from the root of the backup collection
 	// 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"
+	BackupIndexDirectoryPath = backupMetadataDirectory + "/index/"
 
 	// BackupIndexFilenameTimestampFormat is the format used for the human
 	// readable start and end times in the index file names.

pkg/backup/backupinfo/BUILD.bazel

@@ -72,6 +72,7 @@ go_test(
         "//pkg/backup/backupbase",
         "//pkg/backup/backuppb",
         "//pkg/backup/backuptestutils",
+        "//pkg/backup/backuputils",
         "//pkg/base",
         "//pkg/blobs",
         "//pkg/ccl",

pkg/backup/backupinfo/backup_index.go

@@ -332,7 +332,7 @@ func ListSubdirsFromIndex(ctx context.Context, store cloud.ExternalStorage) ([]s
 		"/",
 		func(indexSubdir string) error {
 			indexSubdir = strings.TrimSuffix(indexSubdir, "/")
-			subdir, err := unflattenIndexSubdir(indexSubdir)
+			subdir, err := convertIndexSubdirToSubdir(indexSubdir)
 			if err != nil {
 				return err
 			}
@@ -342,6 +342,10 @@ func ListSubdirsFromIndex(ctx context.Context, store cloud.ExternalStorage) ([]s
 	); err != nil {
 		return nil, errors.Wrapf(err, "listing index subdirs")
 	}
+	// Index subdirs are in descending order and we want chronological order.
+	// TODO (kev-cao): In the new faster `SHOW BACKUP` effort, we will want
+	// the subdirs in descending order.
+	slices.Reverse(subdirs)
 	return subdirs, nil
 }
 
@@ -422,49 +426,71 @@ func getBackupIndexFileName(startTime, endTime hlc.Timestamp) string {
 // 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, error) {
-	flattened, err := flattenSubdirForIndex(subdir)
+	flattened, err := convertSubdirToIndexSubdir(subdir)
 	if err != nil {
 		return "", err
 	}
 	return path.Join(backupbase.BackupIndexDirectoryPath, flattened), nil
 }
 
-// flattenSubdirForIndex flattens a full backup subdirectory to be used in the
-// index. Note that this path does not contain a trailing or leading slash.
+// convertSubdirToIndexSubdir flattens a full backup subdirectory to be used in
+// the index. Note that this path does not contain a trailing or leading slash.
 // It assumes subdir is not `LATEST` and has been resolved.
 // We flatten the subdir so that when listing from the index, we can list with
-// the `index/` prefix and delimit on `/`. e.g.:
+// the index prefix and delimit on `/`. e.g.:
 //
-// index/
+// metadata/index/
 //
-//	|_ 2025-08-13-120000.00/
+//	|_ <desc_end_time>_20250813-120000.00/
 //	|  |_ <index_meta>.pb
-//	|_ 2025-08-14-120000.00/
+//	|_ <desc_end_time>_20250814-120000.00/
 //	|  |_ <index_meta>.pb
-//	|_ 2025-08-14-120000.00/
+//	|_ <desc_end_time>_20250814-120000.00/
 //		 |_ <index_meta>.pb
 //
 // Listing on `index/` and delimiting on `/` will return the subdirectories
 // without listing the files in them.
-func flattenSubdirForIndex(subdir string) (string, error) {
+func convertSubdirToIndexSubdir(subdir string) (string, error) {
 	subdirTime, err := time.Parse(backupbase.DateBasedIntoFolderName, subdir)
 	if err != nil {
 		return "", errors.Wrapf(
-			err, "subdir does not match format '%s'", backupbase.DateBasedIntoFolderName,
+			err, "invalid subdir format: %s", subdir,
 		)
 	}
-	return subdirTime.Format(backupbase.BackupIndexFlattenedSubdir), nil
+	return fmt.Sprintf(
+		"%s_%s",
+		backuputils.EncodeDescendingTS(subdirTime),
+		subdirTime.Format(backupbase.BackupIndexFilenameTimestampFormat),
+	), 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)
+// convertIndexSubdirToSubdir converts an index subdir back to the
+// original full backup subdir.
+func convertIndexSubdirToSubdir(flattened string) (string, error) {
+	parts := strings.Split(flattened, "_")
+	if len(parts) != 2 {
+		return "", errors.Newf(
+			"invalid index subdir format: %s", flattened,
+		)
+	}
+	descSubdirTime, err := backuputils.DecodeDescendingTS(parts[0])
+	if err != nil {
+		return "", errors.Wrapf(
+			err, "index subdir %s could not be decoded", flattened,
+		)
+	}
+	// Validate that the two parts of the index subdir correspond to the same time.
+	subdirTime, err := time.Parse(backupbase.BackupIndexFilenameTimestampFormat, parts[1])
 	if err != nil {
 		return "", errors.Wrapf(
-			err, "index subdir does not match format %s", backupbase.BackupIndexFlattenedSubdir,
+			err, "index subdir %s could not be decoded", flattened,
+		)
+	}
+	if !descSubdirTime.Equal(subdirTime) {
+		return "", errors.Newf(
+			"index subdir %s has mismatched timestamps", flattened,
 		)
 	}
-	unflattened := subdirTime.Format(backupbase.DateBasedIntoFolderName)
+	unflattened := descSubdirTime.Format(backupbase.DateBasedIntoFolderName)
 	return unflattened, nil
 }

pkg/backup/backupinfo/backup_index_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/backup/backupbase"
 	"github.com/cockroachdb/cockroach/pkg/backup/backuppb"
 	"github.com/cockroachdb/cockroach/pkg/backup/backuptestutils"
+	"github.com/cockroachdb/cockroach/pkg/backup/backuputils"
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/blobs"
 	"github.com/cockroachdb/cockroach/pkg/cloud"
@@ -95,29 +96,6 @@ func TestGetBackupIndexFileName(t *testing.T) {
 	)
 }
 
-func TestGetBackupIndexFilePath(t *testing.T) {
-	defer leaktest.AfterTest(t)()
-	defer log.Scope(t).Close(t)
-
-	start, end := hlc.Timestamp{WallTime: 10}, hlc.Timestamp{WallTime: 20}
-	t.Run("fails if subdir is 'LATEST' and unresolved", func(t *testing.T) {
-		_, err := getBackupIndexFilePath("LATEST", start, end)
-		require.Error(t, err)
-	})
-	t.Run("returns correct path for resolved subdir", func(t *testing.T) {
-		subdir := "/2025/07/17-152115.00"
-		flattenedSubdir := "2025-07-17-152115.00"
-		indexPath, err := getBackupIndexFilePath(subdir, start, end)
-		require.NoError(t, err)
-		require.True(
-			t, strings.HasPrefix(
-				indexPath,
-				path.Join(backupbase.BackupIndexDirectoryPath, flattenedSubdir),
-			),
-		)
-	})
-}
-
 func TestWriteBackupIndexMetadata(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -224,7 +202,7 @@ func TestWriteBackupIndexMetadataWithLocalityAwareBackups(t *testing.T) {
 	))
 }
 
-func TestWriteBackupindexMetadataWithSpecifiedIncrementalLocation(t *testing.T) {
+func TestWriteBackupIndexMetadataWithSpecifiedIncrementalLocation(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
@@ -648,6 +626,66 @@ func TestGetBackupTreeIndexMetadata(t *testing.T) {
 	}
 }
 
+func TestConvertIndexSubdirToSubdir(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	endTime := time.Date(2025, 12, 5, 0, 0, 0, 0, time.UTC)
+	endTimeAsSubdir := endTime.Format(backupbase.DateBasedIntoFolderName)
+	endTimeDescEnc := backuputils.EncodeDescendingTS(endTime)
+	endTimeIndexSuffix := endTime.Format(backupbase.BackupIndexFilenameTimestampFormat)
+
+	testcases := []struct {
+		name           string
+		indexSubdir    string
+		expectedSubdir string
+		error          string
+	}{
+		{
+			name:           "valid index subdir",
+			indexSubdir:    endTimeDescEnc + "_" + endTimeIndexSuffix,
+			expectedSubdir: endTimeAsSubdir,
+		},
+		{
+			name:        "index subdir missing two parts",
+			indexSubdir: endTimeDescEnc,
+			error:       "invalid index subdir format",
+		},
+		{
+			name:        "index subdir with extra parts",
+			indexSubdir: endTimeDescEnc + "_" + endTimeIndexSuffix + "_extra",
+			error:       "invalid index subdir format",
+		},
+		{
+			name:        "index subdir with invalid descending timestamp",
+			indexSubdir: "invalid" + "_" + endTimeIndexSuffix,
+			error:       "could not be decoded",
+		},
+		{
+			name:        "index subdir with invalid timestamp suffix",
+			indexSubdir: endTimeDescEnc + "_invalid",
+			error:       "could not be decoded",
+		},
+		{
+			name:        "index subdir with mismatched timestamps",
+			indexSubdir: endTimeDescEnc + "_" + endTime.Add(time.Second).Format(backupbase.BackupIndexFilenameTimestampFormat),
+			error:       "mismatched timestamps",
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			subdir, err := convertIndexSubdirToSubdir(tc.indexSubdir)
+			if tc.error != "" {
+				require.ErrorContains(t, err, tc.error)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, tc.expectedSubdir, subdir)
+		})
+	}
+}
+
 type fakeExternalStorage struct {
 	cloud.ExternalStorage
 	files map[string]*closableBytesWriter

pkg/backup/backuputils/utils.go

@@ -111,6 +111,19 @@ func EncodeDescendingTS(ts time.Time) string {
 	return hex.EncodeToString(buffer)
 }
 
+// DecodeDescendingTS decodes a time.Time encoded with EncodeDescendingTS.
+func DecodeDescendingTS(encoded string) (time.Time, error) {
+	buffer, err := hex.DecodeString(encoded)
+	if err != nil {
+		return time.Time{}, err
+	}
+	_, tsMillis, err := encoding.DecodeUvarintDescending(buffer)
+	if err != nil {
+		return time.Time{}, err
+	}
+	return time.UnixMilli(int64(tsMillis)), nil
+}
+
 // AbsoluteBackupPathInCollectionURI returns the absolute path of a backup
 // assuming the root is the collection URI. Backup URI represents the URI that
 // points to the directory containing the backup manifest of the backup. Since