backupccl: add SHOW BACKUPS IN Y and SHOW BACKUP x IN Y

docs/generated/sql/bnf/show_backup.bnf

@@ -1,3 +1,4 @@
 show_backup_stmt ::=
 	'SHOW' 'BACKUP' location opt_with_options
+	| 'SHOW' 'BACKUP' location 'IN' location opt_with_options
 	| 'SHOW' 'BACKUP' 'SCHEMAS' location opt_with_options

docs/generated/sql/bnf/stmt_block.bnf

@@ -530,7 +530,9 @@ use_stmt ::=
 	'USE' var_value
 
 show_backup_stmt ::=
-	'SHOW' 'BACKUP' string_or_placeholder opt_with_options
+	'SHOW' 'BACKUPS' 'IN' string_or_placeholder
+	| 'SHOW' 'BACKUP' string_or_placeholder opt_with_options
+	| 'SHOW' 'BACKUP' string_or_placeholder 'IN' string_or_placeholder opt_with_options
 	| 'SHOW' 'BACKUP' 'SCHEMAS' string_or_placeholder opt_with_options
 
 show_columns_stmt ::=
@@ -705,6 +707,7 @@ unreserved_keyword ::=
 	| 'AUTOMATIC'
 	| 'AUTHORIZATION'
 	| 'BACKUP'
+	| 'BACKUPS'
 	| 'BEFORE'
 	| 'BEGIN'
 	| 'BINARY'

pkg/ccl/backupccl/show.go

@@ -10,6 +10,8 @@ package backupccl
 
 import (
 	"context"
+	"net/url"
+	"path"
 	"strings"
 	"time"
 
@@ -50,11 +52,23 @@ func showBackupPlanHook(
 		return nil, nil, nil, false, err
 	}
 
+	if backup.Path == nil && backup.InCollection != nil {
+		return showBackupsInCollectionPlanHook(ctx, backup, p)
+	}
+
 	toFn, err := p.TypeAsString(ctx, backup.Path, "SHOW BACKUP")
 	if err != nil {
 		return nil, nil, nil, false, err
 	}
 
+	var inColFn func() (string, error)
+	if backup.InCollection != nil {
+		inColFn, err = p.TypeAsString(ctx, backup.InCollection, "SHOW BACKUP")
+		if err != nil {
+			return nil, nil, nil, false, err
+		}
+	}
+
 	expected := map[string]sql.KVStringOptValidate{
 		backupOptEncPassphrase:  sql.KVStringOptRequireValue,
 		backupOptEncKMS:         sql.KVStringOptRequireValue,
@@ -89,6 +103,19 @@ func showBackupPlanHook(
 			return err
 		}
 
+		if inColFn != nil {
+			collection, err := inColFn()
+			if err != nil {
+				return err
+			}
+			parsed, err := url.Parse(collection)
+			if err != nil {
+				return err
+			}
+			parsed.Path = path.Join(parsed.Path, str)
+			str = parsed.String()
+		}
+
 		store, err := p.ExecCfg().DistSQLSrv.ExternalStorageFromURI(ctx, str, p.User())
 		if err != nil {
 			return errors.Wrapf(err, "make storage")
@@ -380,6 +407,42 @@ var backupShowerFiles = backupShower{
 	},
 }
 
+// showBackupPlanHook implements PlanHookFn.
+func showBackupsInCollectionPlanHook(
+	ctx context.Context, backup *tree.ShowBackup, p sql.PlanHookState,
+) (sql.PlanHookRowFn, sqlbase.ResultColumns, []sql.PlanNode, bool, error) {
+
+	collectionFn, err := p.TypeAsString(ctx, backup.InCollection, "SHOW BACKUPS")
+	if err != nil {
+		return nil, nil, nil, false, err
+	}
+
+	fn := func(ctx context.Context, _ []sql.PlanNode, resultsCh chan<- tree.Datums) error {
+		ctx, span := tracing.ChildSpan(ctx, backup.StatementTag())
+		defer tracing.FinishSpan(span)
+
+		collection, err := collectionFn()
+		if err != nil {
+			return err
+		}
+
+		store, err := p.ExecCfg().DistSQLSrv.ExternalStorageFromURI(ctx, collection, p.User())
+		if err != nil {
+			return errors.Wrapf(err, "connect to external storage")
+		}
+		defer store.Close()
+		res, err := store.ListFiles(ctx, "/*/*/BACKUP")
+		if err != nil {
+			return err
+		}
+		for _, i := range res {
+			resultsCh <- tree.Datums{tree.NewDString(strings.TrimSuffix(i, "/BACKUP"))}
+		}
+		return nil
+	}
+	return fn, sqlbase.ResultColumns{{Name: "path", Typ: types.String}}, nil, false, nil
+}
+
 func init() {
 	sql.AddPlanHook(showBackupPlanHook)
 }

pkg/ccl/backupccl/show_test.go

@@ -284,3 +284,41 @@ func TestShowBackup(t *testing.T) {
 func eqWhitespace(a, b string) bool {
 	return strings.Replace(a, "\t", "", -1) == strings.Replace(b, "\t", "", -1)
 }
+
+func TestShowBackups(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	const numAccounts = 11
+	_, _, sqlDB, tempDir, cleanupFn := BackupRestoreTestSetup(t, singleNode, numAccounts, InitNone)
+	_, _, sqlDBRestore, cleanupEmptyCluster := backupRestoreTestSetupEmpty(t, singleNode, tempDir, InitNone)
+	defer cleanupFn()
+	defer cleanupEmptyCluster()
+
+	const full = LocalFoo + "/full"
+
+	// Make an initial backup.
+	sqlDB.Exec(t, `BACKUP data.bank INTO $1`, full)
+	// Add Incremental changes to it 3 times.
+	sqlDB.Exec(t, `BACKUP data.bank INTO LATEST IN $1`, full)
+	sqlDB.Exec(t, `BACKUP data.bank INTO LATEST IN $1`, full)
+	sqlDB.Exec(t, `BACKUP data.bank INTO LATEST IN $1`, full)
+	// Make a second full backup, add changes to it twice.
+	sqlDB.Exec(t, `BACKUP data.bank INTO $1`, full)
+	sqlDB.Exec(t, `BACKUP data.bank INTO LATEST IN $1`, full)
+	sqlDB.Exec(t, `BACKUP data.bank INTO LATEST IN $1`, full)
+	// Make a third full backup, add changes to it.
+	sqlDB.Exec(t, `BACKUP data.bank INTO $1`, full)
+	sqlDB.Exec(t, `BACKUP data.bank INTO LATEST IN $1`, full)
+
+	rows := sqlDBRestore.QueryStr(t, `SHOW BACKUPS IN $1`, full)
+
+	// assert that we see the three, and only the three, full backups.
+	require.Equal(t, 3, len(rows))
+
+	// check that we can show the inc layers in the individual full backups.
+	b1 := sqlDBRestore.QueryStr(t, `SHOW BACKUP $1 IN $2`, rows[0][0], full)
+	require.Equal(t, 4, len(b1))
+	b2 := sqlDBRestore.QueryStr(t, `SHOW BACKUP $1 IN $2`, rows[1][0], full)
+	require.Equal(t, 3, len(b2))
+}

pkg/sql/parser/parse_test.go

@@ -1512,6 +1512,11 @@ func TestParse(t *testing.T) {
 		{`SHOW BACKUP FILES 'bar'`},
 		{`SHOW BACKUP FILES 'bar' WITH foo = 'bar'`},
 
+		{`SHOW BACKUPS IN 'bar'`},
+		{`SHOW BACKUPS IN $1`},
+		{`SHOW BACKUP 'foo' IN 'bar'`},
+		{`SHOW BACKUP $1 IN $2 WITH foo = 'bar'`},
+
 		{`BACKUP TABLE foo TO 'bar' AS OF SYSTEM TIME '1' INCREMENTAL FROM 'baz'`},
 		{`BACKUP TABLE foo TO $1 INCREMENTAL FROM 'bar', $2, 'baz'`},
 
@@ -2249,10 +2254,10 @@ $function$`,
 		{`BACKUP foo TO 'bar' WITH OPTIONS (detached, KMS = ('foo', 'bar'), revision_history)`,
 			`BACKUP TABLE foo TO 'bar' WITH revision_history, detached, kms=('foo', 'bar')`},
 
-		{`RESTORE foo FROM 'bar' WITH OPTIONS (encryption_passphrase='secret', into_db='baz', 
+		{`RESTORE foo FROM 'bar' WITH OPTIONS (encryption_passphrase='secret', into_db='baz',
 skip_missing_foreign_keys, skip_missing_sequences, skip_missing_sequence_owners, skip_missing_views, detached)`,
 			`RESTORE TABLE foo FROM 'bar' WITH encryption_passphrase='secret', into_db='baz', skip_missing_foreign_keys, skip_missing_sequence_owners, skip_missing_sequences, skip_missing_views, detached`},
-		{`RESTORE foo FROM 'bar' WITH ENCRYPTION_PASSPHRASE = 'secret', INTO_DB=baz, 
+		{`RESTORE foo FROM 'bar' WITH ENCRYPTION_PASSPHRASE = 'secret', INTO_DB=baz,
 SKIP_MISSING_FOREIGN_KEYS, SKIP_MISSING_SEQUENCES, SKIP_MISSING_SEQUENCE_OWNERS, SKIP_MISSING_VIEWS`,
 			`RESTORE TABLE foo FROM 'bar' WITH encryption_passphrase='secret', into_db='baz', skip_missing_foreign_keys, skip_missing_sequence_owners, skip_missing_sequences, skip_missing_views`},
 

pkg/sql/parser/sql.y

@@ -568,7 +568,7 @@ func (u *sqlSymUnion) executorType() tree.ScheduledJobExecutorType {
 %token <str> ALL ALTER ALWAYS ANALYSE ANALYZE AND AND_AND ANY ANNOTATE_TYPE ARRAY AS ASC
 %token <str> ASYMMETRIC AT ATTRIBUTE AUTHORIZATION AUTOMATIC
 
-%token <str> BACKUP BEFORE BEGIN BETWEEN BIGINT BIGSERIAL BINARY BIT
+%token <str> BACKUP BACKUPS BEFORE BEGIN BETWEEN BIGINT BIGSERIAL BINARY BIT
 %token <str> BUCKET_COUNT
 %token <str> BOOLEAN BOTH BOX2D BUNDLE BY
 
@@ -4204,14 +4204,29 @@ show_histogram_stmt:
 // %Text: SHOW BACKUP [SCHEMAS|FILES|RANGES] <location>
 // %SeeAlso: WEBDOCS/show-backup.html
 show_backup_stmt:
-  SHOW BACKUP string_or_placeholder opt_with_options
+  SHOW BACKUPS IN string_or_placeholder
+ {
+    $$.val = &tree.ShowBackup{
+      InCollection:    $4.expr(),
+    }
+  }
+| SHOW BACKUP string_or_placeholder opt_with_options
   {
     $$.val = &tree.ShowBackup{
       Details: tree.BackupDefaultDetails,
       Path:    $3.expr(),
       Options: $4.kvOptions(),
     }
   }
+| SHOW BACKUP string_or_placeholder IN string_or_placeholder opt_with_options
+  {
+    $$.val = &tree.ShowBackup{
+      Details: tree.BackupDefaultDetails,
+      Path:    $3.expr(),
+      InCollection: $5.expr(),
+      Options: $6.kvOptions(),
+    }
+  }
 | SHOW BACKUP SCHEMAS string_or_placeholder opt_with_options
   {
     $$.val = &tree.ShowBackup{
@@ -11026,6 +11041,7 @@ unreserved_keyword:
 | AUTOMATIC
 | AUTHORIZATION
 | BACKUP
+| BACKUPS
 | BEFORE
 | BEGIN
 | BINARY

pkg/sql/sem/tree/show.go

@@ -84,13 +84,19 @@ const (
 // ShowBackup represents a SHOW BACKUP statement.
 type ShowBackup struct {
 	Path                 Expr
+	InCollection         Expr
 	Details              BackupDetails
 	ShouldIncludeSchemas bool
 	Options              KVOptions
 }
 
 // Format implements the NodeFormatter interface.
 func (node *ShowBackup) Format(ctx *FmtCtx) {
+	if node.InCollection != nil && node.Path == nil {
+		ctx.WriteString("SHOW BACKUPS IN ")
+		ctx.FormatNode(node.InCollection)
+		return
+	}
 	ctx.WriteString("SHOW BACKUP ")
 	if node.Details == BackupRangeDetails {
 		ctx.WriteString("RANGES ")
@@ -101,6 +107,10 @@ func (node *ShowBackup) Format(ctx *FmtCtx) {
 		ctx.WriteString("SCHEMAS ")
 	}
 	ctx.FormatNode(node.Path)
+	if node.InCollection != nil {
+		ctx.WriteString(" IN ")
+		ctx.FormatNode(node.InCollection)
+	}
 	if len(node.Options) > 0 {
 		ctx.WriteString(" WITH ")
 		ctx.FormatNode(&node.Options)