backupccl:SHOW BACKUP should read Files from the slim manifest and SST

Release note: None

pkg/ccl/backupccl/show.go

@@ -401,6 +401,8 @@ you must pass the 'encryption_info_dir' parameter that points to the directory o
 		)
 		info.collectionURI = dest[0]
 		info.subdir = computedSubdir
+		info.kmsEnv = &kmsEnv
+		info.enc = encryption
 
 		mkStore := p.ExecCfg().DistSQLSrv.ExternalStorageFromURI
 		incStores, cleanupFn, err := backupdest.MakeBackupDestinationStores(ctx, p.User(), mkStore,
@@ -459,9 +461,7 @@ you must pass the 'encryption_info_dir' parameter that points to the directory o
 			}
 		}
 		if _, ok := opts[backupOptCheckFiles]; ok {
-			fileSizes, err := checkBackupFiles(ctx, info,
-				p.ExecCfg().DistSQLSrv.ExternalStorageFromURI,
-				p.User())
+			fileSizes, err := checkBackupFiles(ctx, info, p.ExecCfg(), p.User(), encryption, &kmsEnv)
 			if err != nil {
 				return err
 			}
@@ -493,7 +493,7 @@ func getBackupInfoReader(p sql.PlanHookState, backup *tree.ShowBackup) backupInf
 		case tree.BackupRangeDetails:
 			shower = backupShowerRanges
 		case tree.BackupFileDetails:
-			shower = backupShowerFileSetup(backup.InCollection)
+			shower = backupShowerFileSetup(p, backup.InCollection)
 		case tree.BackupSchemaDetails:
 			shower = backupShowerDefault(p, true, backup.Options)
 		case tree.BackupValidateDetails:
@@ -511,8 +511,10 @@ func getBackupInfoReader(p sql.PlanHookState, backup *tree.ShowBackup) backupInf
 func checkBackupFiles(
 	ctx context.Context,
 	info backupInfo,
-	storeFactory cloud.ExternalStorageFromURIFactory,
+	execCfg *sql.ExecutorConfig,
 	user username.SQLUsername,
+	encryption *jobspb.BackupEncryptionOptions,
+	kmsEnv cloud.KMSEnv,
 ) ([][]int64, error) {
 	const maxMissingFiles = 10
 	missingFiles := make(map[string]struct{}, maxMissingFiles)
@@ -521,7 +523,7 @@ func checkBackupFiles(
 		// TODO (msbutler): Right now, checkLayer opens stores for each backup layer. In 22.2,
 		// once a backup chain cannot have mixed localities, only create stores for full backup
 		// and first incremental backup.
-		defaultStore, err := storeFactory(ctx, info.defaultURIs[layer], user)
+		defaultStore, err := execCfg.DistSQLSrv.ExternalStorageFromURI(ctx, info.defaultURIs[layer], user)
 		if err != nil {
 			return nil, err
 		}
@@ -562,16 +564,22 @@ func checkBackupFiles(
 		}
 
 		for locality, uri := range info.localityInfo[layer].URIsByOriginalLocalityKV {
-			store, err := storeFactory(ctx, uri, user)
+			store, err := execCfg.DistSQLSrv.ExternalStorageFromURI(ctx, uri, user)
 			if err != nil {
 				return nil, err
 			}
 			localityStores[locality] = store
 		}
 
 		// Check all backup SSTs.
-		fileSizes := make([]int64, len(info.manifests[layer].Files))
-		for i, f := range info.manifests[layer].Files {
+		fileSizes := make([]int64, 0)
+		it, err := makeBackupManifestFileIterator(ctx, execCfg.DistSQLSrv.ExternalStorage,
+			info.manifests[layer], encryption, kmsEnv)
+		if err != nil {
+			return nil, err
+		}
+		defer it.close()
+		for f, hasNext := it.next(); hasNext; f, hasNext = it.next() {
 			store := defaultStore
 			uri := info.defaultURIs[layer]
 			if _, ok := localityStores[f.LocalityKV]; ok {
@@ -590,7 +598,10 @@ func checkBackupFiles(
 				}
 				continue
 			}
-			fileSizes[i] = sz
+			fileSizes = append(fileSizes, sz)
+		}
+		if it.err() != nil {
+			return nil, it.err()
 		}
 
 		return fileSizes, nil
@@ -629,6 +640,7 @@ type backupInfo struct {
 	subdir        string
 	localityInfo  []jobspb.RestoreDetails_BackupLocalityInfo
 	enc           *jobspb.BackupEncryptionOptions
+	kmsEnv        cloud.KMSEnv
 	fileSizes     [][]int64
 }
 
@@ -694,7 +706,6 @@ func backupShowerDefault(p sql.PlanHookState, showSchemas bool, opts tree.KVOpti
 
 			var rows []tree.Datums
 			for layer, manifest := range info.manifests {
-				ctx, sp := tracing.ChildSpan(ctx, "backupccl.backupShowerDefault.fn.layer")
 				descriptors, err := backupinfo.BackupManifestDescriptors(&manifest)
 				if err != nil {
 					return nil, err
@@ -749,7 +760,8 @@ func backupShowerDefault(p sql.PlanHookState, showSchemas bool, opts tree.KVOpti
 				if len(info.fileSizes) > 0 {
 					fileSizes = info.fileSizes[layer]
 				}
-				tableSizes, err := getTableSizes(ctx, manifest.Files, fileSizes)
+
+				tableSizes, err := getTableSizes(ctx, p.ExecCfg().DistSQLSrv.ExternalStorage, info, manifest, fileSizes)
 				if err != nil {
 					return nil, err
 				}
@@ -917,7 +929,6 @@ func backupShowerDefault(p sql.PlanHookState, showSchemas bool, opts tree.KVOpti
 					}
 					rows = append(rows, row)
 				}
-				sp.Finish()
 			}
 			return rows, nil
 		},
@@ -932,16 +943,29 @@ type descriptorSize struct {
 // getLogicalSSTSize gets the total logical bytes stored in each SST. Note that a
 // BackupManifest_File identifies a span in an SST and there can be multiple
 // spans stored in an SST.
-func getLogicalSSTSize(ctx context.Context, files []backuppb.BackupManifest_File) map[string]int64 {
+func getLogicalSSTSize(
+	ctx context.Context,
+	storeFactory cloud.ExternalStorageFactory,
+	manifest backuppb.BackupManifest,
+	enc *jobspb.BackupEncryptionOptions,
+	kmsEnv cloud.KMSEnv,
+) (map[string]int64, error) {
 	ctx, span := tracing.ChildSpan(ctx, "backupccl.getLogicalSSTSize")
 	defer span.Finish()
-	_ = ctx
 
 	sstDataSize := make(map[string]int64)
-	for _, file := range files {
-		sstDataSize[file.Path] += file.EntryCounts.DataSize
+	it, err := makeBackupManifestFileIterator(ctx, storeFactory, manifest, enc, kmsEnv)
+	if err != nil {
+		return nil, err
+	}
+	defer it.close()
+	for f, hasNext := it.next(); hasNext; f, hasNext = it.next() {
+		sstDataSize[f.Path] += f.EntryCounts.DataSize
 	}
-	return sstDataSize
+	if it.err() != nil {
+		return nil, it.err()
+	}
+	return sstDataSize, nil
 }
 
 // approximateSpanPhysicalSize approximates the number of bytes written to disk for the span.
@@ -953,24 +977,39 @@ func approximateSpanPhysicalSize(
 
 // getTableSizes gathers row and size count for each table in the manifest
 func getTableSizes(
-	ctx context.Context, files []backuppb.BackupManifest_File, fileSizes []int64,
+	ctx context.Context,
+	storeFactory cloud.ExternalStorageFactory,
+	info backupInfo,
+	manifest backuppb.BackupManifest,
+	fileSizes []int64,
 ) (map[descpb.ID]descriptorSize, error) {
 	ctx, span := tracing.ChildSpan(ctx, "backupccl.getTableSizes")
 	defer span.Finish()
 
-	tableSizes := make(map[descpb.ID]descriptorSize)
-	if len(files) == 0 {
-		return tableSizes, nil
-	}
-	_, tenantID, err := keys.DecodeTenantPrefix(files[0].Span.Key)
+	logicalSSTSize, err := getLogicalSSTSize(ctx, storeFactory, manifest, info.enc, info.kmsEnv)
 	if err != nil {
 		return nil, err
 	}
-	showCodec := keys.MakeSQLCodec(tenantID)
 
-	logicalSSTSize := getLogicalSSTSize(ctx, files)
+	it, err := makeBackupManifestFileIterator(ctx, storeFactory, manifest, info.enc, info.kmsEnv)
+	if err != nil {
+		return nil, err
+	}
+	defer it.close()
+	tableSizes := make(map[descpb.ID]descriptorSize)
+	var tenantID roachpb.TenantID
+	var showCodec keys.SQLCodec
+	var idx int
+	for f, hasNext := it.next(); hasNext; f, hasNext = it.next() {
+		if !tenantID.IsSet() {
+			var err error
+			_, tenantID, err = keys.DecodeTenantPrefix(f.Span.Key)
+			if err != nil {
+				return nil, err
+			}
+			showCodec = keys.MakeSQLCodec(tenantID)
+		}
 
-	for i, file := range files {
 		// TODO(dan): This assumes each file in the backup only
 		// contains data from a single table, which is usually but
 		// not always correct. It does not account for a BACKUP that
@@ -980,18 +1019,23 @@ func getTableSizes(
 		// TODO(msbutler): after handling the todo above, understand whether
 		// we should return an error if a key does not have tableId. The lack
 		// of error handling let #77705 sneak by our unit tests.
-		_, tableID, err := showCodec.DecodeTablePrefix(file.Span.Key)
+		_, tableID, err := showCodec.DecodeTablePrefix(f.Span.Key)
 		if err != nil {
 			continue
 		}
 		s := tableSizes[descpb.ID(tableID)]
-		s.rowCount.Add(file.EntryCounts)
+		s.rowCount.Add(f.EntryCounts)
 		if len(fileSizes) > 0 {
-			s.fileSize += approximateSpanPhysicalSize(file.EntryCounts.DataSize, logicalSSTSize[file.Path],
-				fileSizes[i])
+			s.fileSize += approximateSpanPhysicalSize(f.EntryCounts.DataSize, logicalSSTSize[f.Path],
+				fileSizes[idx])
 		}
 		tableSizes[descpb.ID(tableID)] = s
+		idx++
 	}
+	if it.err() != nil {
+		return nil, it.err()
+	}
+
 	return tableSizes, nil
 }
 
@@ -1190,7 +1234,9 @@ var backupShowerDoctor = backupShower{
 	},
 }
 
-func backupShowerFileSetup(inCol tree.StringOrPlaceholderOptList) backupShower {
+func backupShowerFileSetup(
+	p sql.PlanHookState, inCol tree.StringOrPlaceholderOptList,
+) backupShower {
 	return backupShower{header: colinfo.ResultColumns{
 		{Name: "path", Typ: types.String},
 		{Name: "backup_type", Typ: types.String},
@@ -1224,8 +1270,20 @@ func backupShowerFileSetup(inCol tree.StringOrPlaceholderOptList) backupShower {
 					backupType = "incremental"
 				}
 
-				logicalSSTSize := getLogicalSSTSize(ctx, manifest.Files)
-				for j, file := range manifest.Files {
+				logicalSSTSize, err := getLogicalSSTSize(ctx, p.ExecCfg().DistSQLSrv.ExternalStorage, manifest,
+					info.enc, info.kmsEnv)
+				if err != nil {
+					return nil, err
+				}
+
+				it, err := makeBackupManifestFileIterator(ctx, p.ExecCfg().DistSQLSrv.ExternalStorage,
+					manifest, info.enc, info.kmsEnv)
+				if err != nil {
+					return nil, err
+				}
+				defer it.close()
+				var idx int
+				for file, hasNext := it.next(); hasNext; file, hasNext = it.next() {
 					filePath := file.Path
 					if inCol != nil {
 						filePath = path.Join(manifestDirs[i], filePath)
@@ -1240,7 +1298,7 @@ func backupShowerFileSetup(inCol tree.StringOrPlaceholderOptList) backupShower {
 					sz := int64(-1)
 					if len(info.fileSizes) > 0 {
 						sz = approximateSpanPhysicalSize(file.EntryCounts.DataSize,
-							logicalSSTSize[file.Path], info.fileSizes[i][j])
+							logicalSSTSize[file.Path], info.fileSizes[i][idx])
 					}
 					rows = append(rows, tree.Datums{
 						tree.NewDString(filePath),
@@ -1254,6 +1312,10 @@ func backupShowerFileSetup(inCol tree.StringOrPlaceholderOptList) backupShower {
 						tree.NewDString(locality),
 						tree.NewDInt(tree.DInt(sz)),
 					})
+					idx++
+				}
+				if it.err() != nil {
+					return nil, it.err()
 				}
 			}
 			return rows, nil