feat(rest): parse storage-credentials from LoadTableResponse

plusplusjiajia · plusplusjiajia · commit c1216fa220aa · 2026-06-14T16:54:21.000+08:00
diff --git a/src/iceberg/catalog/rest/json_serde.cc b/src/iceberg/catalog/rest/json_serde.cc
@@ -71,6 +71,8 @@ constexpr std::string_view kSource = "source";
 constexpr std::string_view kDestination = "destination";
 constexpr std::string_view kMetadata = "metadata";
 constexpr std::string_view kConfig = "config";
+constexpr std::string_view kStorageCredentials = "storage-credentials";
+constexpr std::string_view kPrefix = "prefix";
 constexpr std::string_view kIdentifiers = "identifiers";
 constexpr std::string_view kOverrides = "overrides";
 constexpr std::string_view kDefaults = "defaults";
@@ -695,6 +697,16 @@ nlohmann::json ToJson(const LoadTableResult& result) {
   SetOptionalStringField(json, kMetadataLocation, result.metadata_location);
   json[kMetadata] = ToJson(*result.metadata);
   SetContainerField(json, kConfig, result.config);
+  if (!result.storage_credentials.empty()) {
+    nlohmann::json creds = nlohmann::json::array();
+    for (const auto& cred : result.storage_credentials) {
+      nlohmann::json entry;
+      entry[kPrefix] = cred.prefix;
+      SetContainerField(entry, kConfig, cred.config);
+      creds.push_back(std::move(entry));
+    }
+    json[kStorageCredentials] = std::move(creds);
+  }
   return json;
 }
 
@@ -707,6 +719,27 @@ Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
   ICEBERG_ASSIGN_OR_RAISE(result.config,
                           GetJsonValueOrDefault<decltype(result.config)>(json, kConfig));
+  if (auto it = json.find(kStorageCredentials); it != json.end() && !it->is_null()) {
+    if (!it->is_array()) {
+      return JsonParseError("Cannot parse storage credentials from non-array: {}",
+                            SafeDumpJson(*it));
+    }
+    for (const auto& entry : *it) {
+      StorageCredential cred;
+      ICEBERG_ASSIGN_OR_RAISE(cred.prefix, GetJsonValue<std::string>(entry, kPrefix));
+      ICEBERG_ASSIGN_OR_RAISE(cred.config,
+                              GetJsonValue<decltype(cred.config)>(entry, kConfig));
+      // prefix and config are required by the REST spec; non-empty matches the
+      // Java reference implementation (Credential.validate()).
+      if (cred.prefix.empty()) {
+        return JsonParseError("Invalid storage credential: prefix must be non-empty");
+      }
+      if (cred.config.empty()) {
+        return JsonParseError("Invalid storage credential: config must be non-empty");
+      }
+      result.storage_credentials.push_back(std::move(cred));
+    }
+  }
   ICEBERG_RETURN_UNEXPECTED(result.Validate());
   return result;
 }
diff --git a/src/iceberg/catalog/rest/rest_catalog.cc b/src/iceberg/catalog/rest/rest_catalog.cc
@@ -479,10 +479,16 @@ Result<std::string> RestCatalog::LoadTableInternal(
   return response.body();
 }
 
-Result<std::shared_ptr<Table>> RestCatalog::LoadTable(const TableIdentifier& identifier) {
+Result<LoadTableResult> RestCatalog::LoadTableResultFor(
+    const TableIdentifier& identifier) {
   ICEBERG_ASSIGN_OR_RAISE(const auto body, LoadTableInternal(identifier));
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(body));
-  ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
+  return LoadTableResultFromJson(json);
+}
+
+Result<std::shared_ptr<Table>> RestCatalog::LoadTable(const TableIdentifier& identifier) {
+  ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFor(identifier));
+  // Binds the catalog's default FileIO; use LoadTableResultFor() for vended creds.
   // TODO: Support table-specific auth config and per-table FileIO from the REST
   // load response when table-scoped REST operations are introduced.
   return Table::Make(identifier, std::move(load_result.metadata),
diff --git a/src/iceberg/catalog/rest/rest_catalog.h b/src/iceberg/catalog/rest/rest_catalog.h
@@ -95,6 +95,10 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
 
   Result<std::shared_ptr<Table>> LoadTable(const TableIdentifier& identifier) override;
 
+  /// \brief Like LoadTable() but returns the raw LoadTableResult with any vended
+  /// `storage-credentials`, without binding a FileIO.
+  Result<LoadTableResult> LoadTableResultFor(const TableIdentifier& identifier);
+
   Result<std::shared_ptr<Table>> RegisterTable(
       const TableIdentifier& identifier,
       const std::string& metadata_file_location) override;
diff --git a/src/iceberg/catalog/rest/types.cc b/src/iceberg/catalog/rest/types.cc
@@ -86,7 +86,8 @@ bool CreateTableRequest::operator==(const CreateTableRequest& other) const {
 }
 
 bool LoadTableResult::operator==(const LoadTableResult& other) const {
-  if (metadata_location != other.metadata_location || config != other.config) {
+  if (metadata_location != other.metadata_location || config != other.config ||
+      storage_credentials != other.storage_credentials) {
     return false;
   }
 
diff --git a/src/iceberg/catalog/rest/types.h b/src/iceberg/catalog/rest/types.h
@@ -180,12 +180,23 @@ struct ICEBERG_REST_EXPORT CreateTableRequest {
 /// \brief An opaque token that allows clients to make use of pagination for list APIs.
 using PageToken = std::string;
 
+/// \brief A short-lived credential vended by a REST catalog for a storage
+/// location ``prefix`` (clients pick the longest matching prefix); ``config``
+/// holds backend properties such as ``"s3.access-key-id"`` (Iceberg REST spec).
+struct ICEBERG_REST_EXPORT StorageCredential {
+  std::string prefix;
+  std::unordered_map<std::string, std::string> config;
+
+  bool operator==(const StorageCredential& other) const = default;
+};
+
 /// \brief Result body for table create/load/register APIs.
 struct ICEBERG_REST_EXPORT LoadTableResult {
   std::string metadata_location;
   std::shared_ptr<TableMetadata> metadata;  // required
   std::unordered_map<std::string, std::string> config;
-  // TODO(Li Feiyang): Add std::shared_ptr<StorageCredential> storage_credential;
+  /// \brief Vended storage credentials, one per URI prefix; empty if none.
+  std::vector<StorageCredential> storage_credentials;
 
   /// \brief Validates the LoadTableResult.
   Status Validate() const {
diff --git a/src/iceberg/test/rest_json_serde_test.cc b/src/iceberg/test/rest_json_serde_test.cc
@@ -82,6 +82,11 @@ static std::shared_ptr<TableMetadata> MakeSimpleTableMetadata() {
   });
 }
 
+std::string LoadTableJsonWithCredentials(std::string_view storage_credentials) {
+  return std::string(R"({"storage-credentials":)") + std::string(storage_credentials) +
+         R"(,"metadata":{"format-version":2,"table-uuid":"test","location":"s3://test","last-sequence-number":0,"last-column-id":1,"last-updated-ms":0,"schemas":[{"type":"struct","schema-id":1,"fields":[{"id":1,"name":"id","type":"int","required":true}]}],"current-schema-id":1,"partition-specs":[{"spec-id":0,"fields":[]}],"default-spec-id":0,"last-partition-id":0,"sort-orders":[{"order-id":0,"fields":[]}],"default-sort-order-id":0}})";
+}
+
 // Test parameter structure for roundtrip tests
 template <typename Model>
 struct JsonRoundTripParam {
@@ -1116,7 +1121,17 @@ INSTANTIATE_TEST_SUITE_P(
             .model = {.metadata_location = "s3://bucket/metadata/v1.json",
                       .metadata = MakeSimpleTableMetadata(),
                       .config = {{"warehouse", "s3://bucket/warehouse"},
-                                 {"foo", "bar"}}}}),
+                                 {"foo", "bar"}}}},
+        LoadTableResultParam{
+            .test_name = "WithStorageCredentials",
+            .expected_json_str =
+                R"({"metadata":{"current-schema-id":1,"current-snapshot-id":null,"default-sort-order-id":0,"default-spec-id":0,"format-version":2,"last-column-id":1,"last-partition-id":0,"last-sequence-number":0,"last-updated-ms":0,"location":"s3://bucket/test","metadata-log":[],"partition-specs":[{"fields":[],"spec-id":0}],"partition-statistics":[],"properties":{},"refs":{},"schemas":[{"fields":[{"id":1,"name":"id","required":true,"type":"int"}],"schema-id":1,"type":"struct"}],"snapshot-log":[],"snapshots":[],"sort-orders":[{"fields":[],"order-id":0}],"statistics":[],"table-uuid":"test-uuid-1234"},"storage-credentials":[{"config":{"s3.access-key-id":"AKIAtest","s3.region":"us-east-1","s3.secret-access-key":"secret"},"prefix":"s3"}]})",
+            .model =
+                {.metadata = MakeSimpleTableMetadata(),
+                 .storage_credentials = {{.prefix = "s3",
+                                          .config = {{"s3.access-key-id", "AKIAtest"},
+                                                     {"s3.secret-access-key", "secret"},
+                                                     {"s3.region", "us-east-1"}}}}}}),
     [](const ::testing::TestParamInfo<LoadTableResultParam>& info) {
       return info.param.test_name;
     });
@@ -1145,7 +1160,18 @@ INSTANTIATE_TEST_SUITE_P(
             .json_str =
                 R"({"metadata":{"format-version":2,"table-uuid":"test-uuid-1234","location":"s3://bucket/test","last-sequence-number":0,"last-updated-ms":0,"last-column-id":1,"schemas":[{"type":"struct","schema-id":1,"fields":[{"id":1,"name":"id","type":"int","required":true}]}],"current-schema-id":1,"partition-specs":[{"spec-id":0,"fields":[]}],"default-spec-id":0,"last-partition-id":0,"sort-orders":[{"order-id":0,"fields":[]}],"default-sort-order-id":0,"properties":{}},"config":{"warehouse":"s3://bucket/warehouse"}})",
             .expected_model = {.metadata = MakeSimpleTableMetadata(),
-                               .config = {{"warehouse", "s3://bucket/warehouse"}}}}),
+                               .config = {{"warehouse", "s3://bucket/warehouse"}}}},
+        LoadTableResultDeserializeParam{
+            .test_name = "WithStorageCredentials",
+            .json_str =
+                R"({"metadata":{"format-version":2,"table-uuid":"test-uuid-1234","location":"s3://bucket/test","last-sequence-number":0,"last-updated-ms":0,"last-column-id":1,"schemas":[{"type":"struct","schema-id":1,"fields":[{"id":1,"name":"id","type":"int","required":true}]}],"current-schema-id":1,"partition-specs":[{"spec-id":0,"fields":[]}],"default-spec-id":0,"last-partition-id":0,"sort-orders":[{"order-id":0,"fields":[]}],"default-sort-order-id":0,"properties":{}},"storage-credentials":[{"prefix":"s3","config":{"s3.access-key-id":"AKIAtest","s3.secret-access-key":"secret","s3.session-token":"token","s3.region":"us-east-1"}}]})",
+            .expected_model =
+                {.metadata = MakeSimpleTableMetadata(),
+                 .storage_credentials = {{.prefix = "s3",
+                                          .config = {{"s3.access-key-id", "AKIAtest"},
+                                                     {"s3.secret-access-key", "secret"},
+                                                     {"s3.session-token", "token"},
+                                                     {"s3.region", "us-east-1"}}}}}}),
     [](const ::testing::TestParamInfo<LoadTableResultDeserializeParam>& info) {
       return info.param.test_name;
     });
@@ -1184,7 +1210,28 @@ INSTANTIATE_TEST_SUITE_P(
         LoadTableResultInvalidParam{
             .test_name = "InvalidMetadataContent",
             .invalid_json_str = R"({"metadata":{"format-version":"invalid"}})",
-            .expected_error_message = "type must be number, but is string"}),
+            .expected_error_message = "type must be number, but is string"},
+        LoadTableResultInvalidParam{
+            .test_name = "StorageCredentialsNotArray",
+            .invalid_json_str = LoadTableJsonWithCredentials(R"("oops")"),
+            .expected_error_message = "Cannot parse storage credentials from non-array"},
+        LoadTableResultInvalidParam{
+            .test_name = "StorageCredentialMissingPrefix",
+            .invalid_json_str = LoadTableJsonWithCredentials(R"([{"config":{"k":"v"}}])"),
+            .expected_error_message = "Missing 'prefix'"},
+        LoadTableResultInvalidParam{
+            .test_name = "StorageCredentialMissingConfig",
+            .invalid_json_str = LoadTableJsonWithCredentials(R"([{"prefix":"s3"}])"),
+            .expected_error_message = "Missing 'config'"},
+        LoadTableResultInvalidParam{.test_name = "StorageCredentialEmptyPrefix",
+                                    .invalid_json_str = LoadTableJsonWithCredentials(
+                                        R"([{"prefix":"","config":{"k":"v"}}])"),
+                                    .expected_error_message = "prefix must be non-empty"},
+        LoadTableResultInvalidParam{
+            .test_name = "StorageCredentialEmptyConfig",
+            .invalid_json_str =
+                LoadTableJsonWithCredentials(R"([{"prefix":"s3","config":{}}])"),
+            .expected_error_message = "config must be non-empty"}),
     [](const ::testing::TestParamInfo<LoadTableResultInvalidParam>& info) {
       return info.param.test_name;
     });

Original file line number	Diff line number	Diff line change
`@@ -86,7 +86,8 @@ bool CreateTableRequest::operator==(const CreateTableRequest& other) const {`
`86`	`86`	`}`
`87`	`87`
`88`	`88`	`bool LoadTableResult::operator==(const LoadTableResult& other) const {`
`89`		`- if (metadata_location != other.metadata_location \|\| config != other.config) {`
	`89`	`+ if (metadata_location != other.metadata_location \|\| config != other.config \|\|`
	`90`	`+ storage_credentials != other.storage_credentials) {`
`90`	`91`	`return false;`
`91`	`92`	`}`
`92`	`93`