cliccl: add load show backups to display backup collection

Previously, users can list backups created by `BACKUP INTO`
with `SHOW BACKUP IN`in a sql session. But this listing task
can be also done offline without a running cluster.

This PR updates `load show` with `backups` subcommand, which allows
users to list backups in a backup collection created by
`BACKUP INTO`. With the same purpose as other `load
show` subcommands, this update allows users to list backups without
 running `SHOW BACKUP IN` in a sql session.

see cockroachdb#61131 cockroachdb#61829 to checkout other `load show` subcommand.

Release note (cli change): Add `load show backups` to
display backup collection. Previously, users can list backups
created by `BACKUP INTO` via `SHOW BACKUP IN`in a sql
session. But this listing task can be also done
offline without a running cluster. Now, users are able to list
backups in a collection with
`cockroach load show backups <collection_url>.

pkg/ccl/backupccl/backup_destination.go

@@ -272,7 +272,7 @@ func resolveBackupCollection(
 		chosenSuffix = strings.TrimPrefix(subdir, "/")
 		chosenSuffix = "/" + chosenSuffix
 	} else {
-		chosenSuffix = endTime.GoTime().Format(dateBasedIntoFolderName)
+		chosenSuffix = endTime.GoTime().Format(DateBasedIntoFolderName)
 	}
 	return collectionURI, chosenSuffix, nil
 }

pkg/ccl/backupccl/manifest_handling.go

@@ -71,11 +71,15 @@ const (
 const (
 	// BackupFormatDescriptorTrackingVersion added tracking of complete DBs.
 	BackupFormatDescriptorTrackingVersion uint32 = 1
+
 	// DateBasedIncFolderName is the date format used when creating sub-directories
 	// storing incremental backups for auto-appendable backups.
 	// It is exported for testing backup inspection tooling.
-	DateBasedIncFolderName  = "/20060102/150405.00"
-	dateBasedIntoFolderName = "/2006/01/02-150405.00"
+	DateBasedIncFolderName = "/20060102/150405.00"
+	// DateBasedIntoFolderName is the date format used when creating sub-directories
+	// for storing backups in a collection.
+	// Also exported for testing backup inspection tooling.
+	DateBasedIntoFolderName = "/2006/01/02-150405.00"
 	latestFileName          = "LATEST"
 )
 
@@ -1001,3 +1005,18 @@ func checkForPreviousBackup(
 func tempCheckpointFileNameForJob(jobID jobspb.JobID) string {
 	return fmt.Sprintf("%s-%d", backupManifestCheckpointName, jobID)
 }
+
+// ListFullBackupsInCollection lists full backup paths in the collection
+// of an export store
+func ListFullBackupsInCollection(
+	ctx context.Context, store cloud.ExternalStorage,
+) ([]string, error) {
+	backupPaths, err := store.ListFiles(ctx, "/*/*/*/"+backupManifestName)
+	if err != nil {
+		return nil, err
+	}
+	for i, backupPath := range backupPaths {
+		backupPaths[i] = strings.TrimSuffix(backupPath, "/"+backupManifestName)
+	}
+	return backupPaths, nil
+}

pkg/ccl/backupccl/show.go

@@ -532,12 +532,12 @@ func showBackupsInCollectionPlanHook(
 			return errors.Wrapf(err, "connect to external storage")
 		}
 		defer store.Close()
-		res, err := store.ListFiles(ctx, "/*/*/*/"+backupManifestName)
+		res, err := ListFullBackupsInCollection(ctx, store)
 		if err != nil {
 			return err
 		}
 		for _, i := range res {
-			resultsCh <- tree.Datums{tree.NewDString(strings.TrimSuffix(i, "/"+backupManifestName))}
+			resultsCh <- tree.Datums{tree.NewDString(i)}
 		}
 		return nil
 	}

pkg/ccl/cliccl/load.go

@@ -51,6 +51,14 @@ func init() {
 		RunE:  cli.MaybeDecorateGRPCError(runLoadShowSummary),
 	}
 
+	loadShowBackupsCmd := &cobra.Command{
+		Use:   "backups <backup_path>",
+		Short: "show backups in collections",
+		Long:  "Shows full backups in a backup collections.",
+		Args:  cobra.ExactArgs(1),
+		RunE:  cli.MaybeDecorateGRPCError(runLoadShowBackups),
+	}
+
 	loadShowIncrementalCmd := &cobra.Command{
 		Use:   "incremental <backup_path>",
 		Short: "show incremental backups",
@@ -88,9 +96,13 @@ func init() {
 
 	cli.AddCmd(loadCmds)
 	loadCmds.AddCommand(loadShowCmds)
-	loadShowCmds.AddCommand(loadShowSummaryCmd)
-	loadShowCmds.AddCommand(loadShowIncrementalCmd)
+	loadShowCmds.AddCommand([]*cobra.Command{
+		loadShowSummaryCmd,
+		loadShowBackupsCmd,
+		loadShowIncrementalCmd,
+	}...)
 	loadShowSummaryCmd.Flags().AddFlagSet(loadFlags)
+	loadShowBackupsCmd.Flags().AddFlagSet(loadFlags)
 	loadShowIncrementalCmd.Flags().AddFlagSet(loadFlags)
 }
 
@@ -134,6 +146,36 @@ func runLoadShowSummary(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
+func runLoadShowBackups(cmd *cobra.Command, args []string) error {
+
+	path := args[0]
+	if !strings.Contains(path, "://") {
+		path = cloudimpl.MakeLocalStorageURI(path)
+	}
+	ctx := context.Background()
+	store, err := cloudimpl.ExternalStorageFromURI(ctx, path, base.ExternalIODirConfig{},
+		cluster.NoSettings, newBlobFactory, security.RootUserName(), nil /*Internal Executor*/, nil /*kvDB*/)
+	if err != nil {
+		return errors.Wrapf(err, "connect to external storage")
+	}
+	defer store.Close()
+
+	backupPaths, err := backupccl.ListFullBackupsInCollection(ctx, store)
+	if err != nil {
+		return errors.Wrapf(err, "list full backups in collection")
+	}
+
+	if len(backupPaths) == 0 {
+		fmt.Println("no backups found.")
+	}
+
+	for _, backupPath := range backupPaths {
+		fmt.Println("./" + backupPath)
+	}
+
+	return nil
+}
+
 func runLoadShowIncremental(cmd *cobra.Command, args []string) error {
 
 	path := args[0]

pkg/ccl/cliccl/load_test.go

@@ -110,6 +110,53 @@ Tables:
 	})
 }
 
+func TestLoadShowBackups(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	c := cli.NewCLITest(cli.TestCLIParams{T: t, NoServer: true})
+	defer c.Cleanup()
+
+	ctx := context.Background()
+	dir, cleanFn := testutils.TempDir(t)
+	defer cleanFn()
+	srv, db, _ := serverutils.StartServer(t, base.TestServerArgs{ExternalIODir: dir, Insecure: true})
+	defer srv.Stopper().Stop(ctx)
+
+	sqlDB := sqlutils.MakeSQLRunner(db)
+	sqlDB.Exec(t, `CREATE DATABASE testDB`)
+	sqlDB.Exec(t, `USE testDB`)
+	const backupPath = "nodelocal://0/fooFolder"
+
+	t.Run("show-backups-without-backups-in-collection", func(t *testing.T) {
+		out, err := c.RunWithCapture(fmt.Sprintf("load show backups %s --external-io-dir=%s", backupPath, dir))
+		require.NoError(t, err)
+		expectedOutput := "no backups found.\n"
+		checkExpectedOutput(t, expectedOutput, out)
+	})
+
+	intoTS := hlc.Timestamp{WallTime: timeutil.Now().UnixNano()}
+	initAOST := intoTS.AsOfSystemTime()
+	sqlDB.Exec(t, fmt.Sprintf(`BACKUP DATABASE testDB INTO $1 AS OF SYSTEM TIME '%s'`, initAOST), backupPath)
+	intoTS1 := hlc.Timestamp{WallTime: timeutil.Now().UnixNano()}
+	intoAOST1 := intoTS1.AsOfSystemTime()
+	sqlDB.Exec(t, fmt.Sprintf(`BACKUP DATABASE testDB INTO $1 AS OF SYSTEM TIME '%s'`, intoAOST1), backupPath)
+	intoTS2 := hlc.Timestamp{WallTime: timeutil.Now().UnixNano()}
+	intoAOST2 := intoTS2.AsOfSystemTime()
+	sqlDB.Exec(t, fmt.Sprintf(`BACKUP DATABASE testDB INTO $1 AS OF SYSTEM TIME '%s'`, intoAOST2), backupPath)
+
+	t.Run("show-backups-with-backups-in-collection", func(t *testing.T) {
+		out, err := c.RunWithCapture(fmt.Sprintf("load show backups %s --external-io-dir=%s", backupPath, dir))
+		require.NoError(t, err)
+
+		expectedOutput := fmt.Sprintf(".%s\n.%s\n.%s\n",
+			intoTS.GoTime().Format(backupccl.DateBasedIntoFolderName),
+			intoTS1.GoTime().Format(backupccl.DateBasedIntoFolderName),
+			intoTS2.GoTime().Format(backupccl.DateBasedIntoFolderName))
+		checkExpectedOutput(t, expectedOutput, out)
+	})
+}
+
 func TestLoadShowIncremental(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)