feat(rest): support storage credentials in scan planning responses

Parse storage-credentials from PlanTableScanResponse,
FetchPlanningResultResponse, and FetchScanTasksResponse. When credentials
are present, build a scan-scoped FileIO via MakeTableFileIO and expose it
through RestTableScan::effective_io() for callers to use when reading
the returned scan tasks.

Author: Sandeep Gottimukkala

src/iceberg/catalog/rest/json_serde.cc

@@ -531,6 +531,15 @@ Result<nlohmann::json> ScanTaskFieldsToJson(
     json[kFileScanTasks] = std::move(tasks_json);
   }
 
+  if (!response.storage_credentials.empty()) {
+    nlohmann::json creds_json = nlohmann::json::array();
+    for (const auto& cred : response.storage_credentials) {
+      ICEBERG_ASSIGN_OR_RAISE(auto entry, StorageCredentialToJson(cred));
+      creds_json.push_back(std::move(entry));
+    }
+    json[kStorageCredentials] = std::move(creds_json);
+  }
+
   return json;
 }
 
@@ -571,6 +580,21 @@ Status ScanTaskFieldsFromJson(
         FileScanTasksFromJson(file_scan_tasks_json, response.delete_files,
                               partition_specs_by_id, schema));
   }
+
+  // 4. storage_credentials
+  if (json.contains(kStorageCredentials)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto creds_json,
+                            GetJsonValue<nlohmann::json>(json, kStorageCredentials));
+    if (!creds_json.is_array()) {
+      return JsonParseError("Cannot parse storage credentials from non-array: {}",
+                            SafeDumpJson(creds_json));
+    }
+    for (const auto& entry : creds_json) {
+      ICEBERG_ASSIGN_OR_RAISE(auto cred, StorageCredentialFromJson(entry));
+      response.storage_credentials.push_back(std::move(cred));
+    }
+  }
+
   return {};
 }
 

src/iceberg/catalog/rest/rest_catalog.cc

@@ -43,8 +43,8 @@
 #include "iceberg/catalog/rest/rest_util.h"
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/json_serde_internal.h"
-#include "iceberg/logging/logger.h"
 #include "iceberg/logging/log_level.h"
+#include "iceberg/logging/logger.h"
 #include "iceberg/metrics/metrics_reporters.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
@@ -934,6 +934,8 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
         .session = table_session,
         .supported_endpoints = supported_endpoints_,
         .identifier = identifier,
+        .catalog_config = config_.configs(),
+        .table_config = table_config,
     };
     return RestTable::Make(identifier, std::move(result.metadata),
                            std::move(result.metadata_location), std::move(table_io),

src/iceberg/catalog/rest/rest_table_scan.cc

@@ -29,6 +29,7 @@
 #include "iceberg/catalog/rest/http_client.h"
 #include "iceberg/catalog/rest/json_serde_internal.h"
 #include "iceberg/catalog/rest/resource_paths.h"
+#include "iceberg/catalog/rest/rest_file_io.h"
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/json_serde_internal.h"
 #include "iceberg/partition_spec.h"
@@ -115,8 +116,7 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanTableScan(
     }
   }
 
-  ICEBERG_ASSIGN_OR_RAISE(auto path,
-                          rest_context_.paths->Plan(rest_context_.identifier));
+  ICEBERG_ASSIGN_OR_RAISE(auto path, rest_context_.paths->Plan(rest_context_.identifier));
   ICEBERG_ASSIGN_OR_RAISE(auto request_json, ToJson(request));
   ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(request_json));
   ICEBERG_ASSIGN_OR_RAISE(
@@ -132,6 +132,7 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanTableScan(
 
   switch (result.plan_status) {
     case PlanStatus::kCompleted: {
+      ICEBERG_RETURN_UNEXPECTED(ApplyStorageCredentials(result.storage_credentials));
       auto tasks = ResolveScanTasks(result.plan_tasks, result.file_scan_tasks, specs);
       if (!tasks.has_value()) CancelPlanning(plan_id);
       return tasks;
@@ -140,7 +141,7 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::PlanTableScan(
       return FetchPlanningResult(plan_id, specs);
     case PlanStatus::kFailed:
       return IOError("Scan planning failed: {}",
-                      result.error ? result.error->message : "unknown error");
+                     result.error ? result.error->message : "unknown error");
     case PlanStatus::kCancelled:
       return IOError("Scan planning was cancelled for plan_id={}", plan_id);
   }
@@ -171,6 +172,7 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::FetchPlanningR
 
     switch (result.plan_status) {
       case PlanStatus::kCompleted: {
+        ICEBERG_RETURN_UNEXPECTED(ApplyStorageCredentials(result.storage_credentials));
         auto tasks = ResolveScanTasks(result.plan_tasks, result.file_scan_tasks, specs);
         if (!tasks.has_value()) CancelPlanning(plan_id);
         return tasks;
@@ -181,9 +183,8 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::FetchPlanningR
                               .count();
         if (elapsed_ms >= kMaxWaitTimeMs) {
           CancelPlanning(plan_id);
-          return IOError(
-              "Scan planning timed out after {}ms waiting for plan_id={}", elapsed_ms,
-              plan_id);
+          return IOError("Scan planning timed out after {}ms waiting for plan_id={}",
+                         elapsed_ms, plan_id);
         }
         std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
         delay_ms = std::min(delay_ms * 2, kMaxSleepMs);
@@ -192,15 +193,15 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::FetchPlanningR
       case PlanStatus::kFailed:
         CancelPlanning(plan_id);
         return IOError("Scan planning failed: {}",
-                        result.error ? result.error->message : "unknown error");
+                       result.error ? result.error->message : "unknown error");
       case PlanStatus::kCancelled:
         return IOError("Scan planning was cancelled for plan_id={}", plan_id);
     }
   }
 
   CancelPlanning(plan_id);
-  return IOError("Scan planning exceeded max retries ({}) for plan_id={}",
-                          kMaxRetries, plan_id);
+  return IOError("Scan planning exceeded max retries ({}) for plan_id={}", kMaxRetries,
+                 plan_id);
 }
 
 Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::FetchScanTasks(
@@ -212,15 +213,15 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> RestTableScan::FetchScanTasks
                           rest_context_.paths->FetchScanTasks(rest_context_.identifier));
   FetchScanTasksRequest request{.planTask = plan_task};
   ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
-  ICEBERG_ASSIGN_OR_RAISE(
-      const auto response,
-      rest_context_.client->Post(path, json_request, /*headers=*/{},
-                                 *PlanTaskErrorHandler::Instance(),
-                                 *rest_context_.session));
+  ICEBERG_ASSIGN_OR_RAISE(const auto response,
+                          rest_context_.client->Post(path, json_request, /*headers=*/{},
+                                                     *PlanTaskErrorHandler::Instance(),
+                                                     *rest_context_.session));
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
   ICEBERG_ASSIGN_OR_RAISE(auto result,
                           FetchScanTasksResponseFromJson(json, specs, *schema_));
   ICEBERG_RETURN_UNEXPECTED(result.Validate());
+  ICEBERG_RETURN_UNEXPECTED(ApplyStorageCredentials(result.storage_credentials));
 
   return ResolveScanTasks(result.plan_tasks, result.file_scan_tasks, specs);
 }
@@ -253,18 +254,31 @@ void RestTableScan::CancelPlanning(const std::string& plan_id) const {
   if (!path.has_value()) return;
 
   // Best-effort: ignore errors.
-  std::ignore = rest_context_.client->Delete(*path, /*params=*/{}, /*headers=*/{},
-                                             *PlanErrorHandler::Instance(),
-                                             *rest_context_.session);
+  std::ignore =
+      rest_context_.client->Delete(*path, /*params=*/{}, /*headers=*/{},
+                                   *PlanErrorHandler::Instance(), *rest_context_.session);
+}
+
+const std::shared_ptr<FileIO>& RestTableScan::effective_io() const {
+  return scan_io_ ? scan_io_ : io_;
+}
+
+Status RestTableScan::ApplyStorageCredentials(
+    const std::vector<StorageCredential>& credentials) const {
+  if (credentials.empty()) return {};
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto io, MakeTableFileIO(rest_context_.catalog_config, rest_context_.table_config,
+                               credentials));
+  scan_io_ = std::move(io);
+  return {};
 }
 
 // RestTableScanBuilder
 
-RestTableScanBuilder::RestTableScanBuilder(std::shared_ptr<TableMetadata> metadata,
-                                           std::shared_ptr<FileIO> io,
-                                           std::string table_name,
-                                           std::shared_ptr<MetricsReporter> metrics_reporter,
-                                           RestScanContext rest_context)
+RestTableScanBuilder::RestTableScanBuilder(
+    std::shared_ptr<TableMetadata> metadata, std::shared_ptr<FileIO> io,
+    std::string table_name, std::shared_ptr<MetricsReporter> metrics_reporter,
+    RestScanContext rest_context)
     : DataTableScanBuilder(std::move(metadata), std::move(io), std::move(table_name),
                            std::move(metrics_reporter)),
       rest_context_(std::move(rest_context)) {}

src/iceberg/catalog/rest/rest_table_scan.h

@@ -29,6 +29,7 @@
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
 #include "iceberg/metrics/metrics_reporter.h"
 #include "iceberg/result.h"
+#include "iceberg/storage_credential.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_scan.h"
 #include "iceberg/type_fwd.h"
@@ -52,6 +53,11 @@ struct ICEBERG_REST_EXPORT RestScanContext {
   std::shared_ptr<auth::AuthSession> session;
   std::unordered_set<Endpoint> supported_endpoints;
   TableIdentifier identifier;
+  /// Catalog-level config, used with table_config to build a scan-scoped FileIO
+  /// when the server vends storage credentials in a planning response.
+  std::unordered_map<std::string, std::string> catalog_config;
+  /// Table-level config merged with catalog_config for scan-scoped FileIO creation.
+  std::unordered_map<std::string, std::string> table_config;
 };
 
 /// \brief A DataTableScan that delegates PlanFiles() to the REST catalog server
@@ -69,6 +75,13 @@ class ICEBERG_REST_EXPORT RestTableScan : public DataTableScan {
   /// \brief Plans files via the REST scan planning endpoints.
   Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles() const override;
 
+  /// \brief Returns the FileIO to use when reading scan results.
+  ///
+  /// If the server vended storage credentials during planning, returns a FileIO
+  /// initialised with those credentials; otherwise returns the table's FileIO.
+  /// Must be called after PlanFiles().
+  const std::shared_ptr<FileIO>& effective_io() const;
+
  private:
   RestTableScan(std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
                 std::shared_ptr<FileIO> io, internal::TableScanContext context,
@@ -98,14 +111,19 @@ class ICEBERG_REST_EXPORT RestTableScan : public DataTableScan {
   /// DELETE /plan/{plan_id}; best-effort, errors are silently ignored.
   void CancelPlanning(const std::string& plan_id) const;
 
+  /// Builds a scan-scoped FileIO from vended credentials and caches it in scan_io_.
+  /// No-op if credentials is empty.
+  Status ApplyStorageCredentials(const std::vector<StorageCredential>& credentials) const;
+
   RestScanContext rest_context_;
+  mutable std::shared_ptr<FileIO> scan_io_;
 };
 
 /// \brief Builder that produces a RestTableScan with the REST HTTP context injected.
 class ICEBERG_REST_EXPORT RestTableScanBuilder : public DataTableScanBuilder {
  public:
-  RestTableScanBuilder(std::shared_ptr<TableMetadata> metadata, std::shared_ptr<FileIO> io,
-                       std::string table_name,
+  RestTableScanBuilder(std::shared_ptr<TableMetadata> metadata,
+                       std::shared_ptr<FileIO> io, std::string table_name,
                        std::shared_ptr<MetricsReporter> metrics_reporter,
                        RestScanContext rest_context);
 

src/iceberg/catalog/rest/types.cc

@@ -210,6 +210,7 @@ bool OptionalSharedPtrVectorEqual(
 template <typename Response>
 bool ScanTaskFieldsEqual(const Response& lhs, const Response& rhs) {
   return lhs.plan_tasks == rhs.plan_tasks &&
+         lhs.storage_credentials == rhs.storage_credentials &&
          SharedPtrVectorEqual(lhs.delete_files, rhs.delete_files) &&
          OptionalSharedPtrVectorEqual(lhs.file_scan_tasks, rhs.file_scan_tasks,
                                       FileScanTaskEqual);

src/iceberg/catalog/rest/types.h

@@ -337,7 +337,7 @@ struct ICEBERG_REST_EXPORT PlanTableScanResponse {
   PlanStatus plan_status = PlanStatus::kCompleted;
   std::string plan_id;
   std::optional<ErrorResponse> error;
-  // TODO(sandeepg): Add storage credentials and bind scan FileIO to them.
+  std::vector<StorageCredential> storage_credentials;
 
   Status Validate() const;
 
@@ -352,7 +352,7 @@ struct ICEBERG_REST_EXPORT FetchPlanningResultResponse {
   std::vector<std::shared_ptr<DataFile>> delete_files;
   PlanStatus plan_status = PlanStatus::kCompleted;
   std::optional<ErrorResponse> error;
-  // TODO(sandeepg): Add storage credentials and bind scan FileIO to them.
+  std::vector<StorageCredential> storage_credentials;
 
   Status Validate() const;
 
@@ -373,6 +373,7 @@ struct ICEBERG_REST_EXPORT FetchScanTasksResponse {
   std::optional<std::vector<std::string>> plan_tasks;
   std::optional<std::vector<std::shared_ptr<FileScanTask>>> file_scan_tasks;
   std::vector<std::shared_ptr<DataFile>> delete_files;
+  std::vector<StorageCredential> storage_credentials;
 
   Status Validate() const;