polish impls

Author: wgtmac

src/iceberg/catalog/rest/catalog_properties.cc

@@ -37,18 +37,7 @@ std::unique_ptr<RestCatalogProperties> RestCatalogProperties::FromMap(
 
 std::unordered_map<std::string, std::string> RestCatalogProperties::ExtractHeaders()
     const {
-  std::unordered_map<std::string, std::string> headers;
-  constexpr std::string_view prefix = "header.";
-  for (const auto& [key, value] : configs_) {
-    if (key.starts_with(prefix)) {
-      std::string header_name = key.substr(prefix.length());
-      if (header_name.empty() || value.empty()) {
-        continue;
-      }
-      headers[header_name] = value;
-    }
-  }
-  return headers;
+  return Extract(kHeaderPrefix);
 }
 
 Result<std::string_view> RestCatalogProperties::Uri() const {

src/iceberg/catalog/rest/catalog_properties.h

@@ -41,15 +41,14 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
 
   /// \brief The URI of the REST catalog server.
   inline static Entry<std::string> kUri{"uri", ""};
-
   /// \brief The name of the catalog.
   inline static Entry<std::string> kName{"name", ""};
-
   /// \brief The warehouse path.
   inline static Entry<std::string> kWarehouse{"warehouse", ""};
-
   /// \brief The optional prefix for REST API paths.
   inline static Entry<std::string> kPrefix{"prefix", ""};
+  /// \brief The prefix for HTTP headers.
+  inline static constexpr std::string_view kHeaderPrefix = "header.";
 
   /// \brief Create a default RestCatalogProperties instance.
   static std::unique_ptr<RestCatalogProperties> default_properties();
@@ -59,9 +58,6 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
       const std::unordered_map<std::string, std::string>& properties);
 
   /// \brief Returns HTTP headers to be added to every request.
-  ///
-  /// This includes any key prefixed with "header." in the properties.
-  /// \return A map of headers with the prefix removed from the keys.
   std::unordered_map<std::string, std::string> ExtractHeaders() const;
 
   /// \brief Get the URI of the REST catalog server.

src/iceberg/catalog/rest/error_handlers.cc

@@ -21,6 +21,8 @@
 
 #include <string_view>
 
+#include "iceberg/catalog/rest/types.h"
+
 namespace iceberg::rest {
 
 namespace {
@@ -58,7 +60,7 @@ Status DefaultErrorHandler::Accept(const ErrorModel& error) const {
       return ServiceUnavailable("Service unavailable: {}", error.message);
   }
 
-  return RestError("Unable to process: {}", error.message);
+  return RestError("Code: {}, message: {}", error.code, error.message);
 }
 
 const std::shared_ptr<NamespaceErrorHandler>& NamespaceErrorHandler::Instance() {

src/iceberg/catalog/rest/error_handlers.h

@@ -22,7 +22,7 @@
 #include <memory>
 
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
-#include "iceberg/catalog/rest/types.h"
+#include "iceberg/catalog/rest/type_fwd.h"
 #include "iceberg/result.h"
 
 /// \file iceberg/catalog/rest/error_handlers.h

src/iceberg/catalog/rest/http_client.cc

@@ -23,8 +23,10 @@
 #include <nlohmann/json.hpp>
 
 #include "iceberg/catalog/rest/constant.h"
+#include "iceberg/catalog/rest/error_handlers.h"
 #include "iceberg/catalog/rest/json_internal.h"
 #include "iceberg/json_internal.h"
+#include "iceberg/result.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg::rest {
@@ -47,11 +49,8 @@ class HttpResponse::Impl {
 };
 
 HttpResponse::HttpResponse(HttpResponse&&) noexcept = default;
-
 HttpResponse& HttpResponse::operator=(HttpResponse&&) noexcept = default;
-
 HttpResponse::HttpResponse() = default;
-
 HttpResponse::~HttpResponse() = default;
 
 int32_t HttpResponse::status_code() const { return impl_->status_code(); }
@@ -101,6 +100,7 @@ bool IsSuccessful(int32_t status_code) {
 Status HandleFailureResponse(const cpr::Response& response,
                              const ErrorHandler& error_handler) {
   if (!IsSuccessful(response.status_code)) {
+    // TODO(gangwu): response status code is lost, wrap it with RestError.
     ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.text));
     ICEBERG_ASSIGN_OR_RAISE(auto error_response, ErrorResponseFromJson(json));
     return error_handler.Accept(error_response.error);
@@ -137,30 +137,34 @@ Result<HttpResponse> HttpClient::Get(
     const std::string& path, const std::unordered_map<std::string, std::string>& params,
     const std::unordered_map<std::string, std::string>& headers,
     const ErrorHandler& error_handler) {
-  std::scoped_lock<std::mutex> lock(session_mutex_);
+  cpr::Response response;
+  {
+    std::scoped_lock<std::mutex> lock(session_mutex_);
+    PrepareSession(path, headers, params);
+    response = session_->Get();
+  }
 
-  PrepareSession(path, headers, params);
-  cpr::Response response = session_->Get();
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
-  auto impl = std::make_unique<HttpResponse::Impl>(std::move(response));
   HttpResponse http_response;
-  http_response.impl_ = std::move(impl);
+  http_response.impl_ = std::make_unique<HttpResponse::Impl>(std::move(response));
   return http_response;
 }
 
 Result<HttpResponse> HttpClient::Post(
     const std::string& path, const std::string& body,
     const std::unordered_map<std::string, std::string>& headers,
     const ErrorHandler& error_handler) {
-  std::scoped_lock<std::mutex> lock(session_mutex_);
+  cpr::Response response;
+  {
+    std::scoped_lock<std::mutex> lock(session_mutex_);
+    PrepareSession(path, headers);
+    session_->SetBody(cpr::Body{body});
+    response = session_->Post();
+  }
 
-  PrepareSession(path, headers);
-  session_->SetBody(cpr::Body{body});
-  cpr::Response response = session_->Post();
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
-  auto impl = std::make_unique<HttpResponse::Impl>(std::move(response));
   HttpResponse http_response;
-  http_response.impl_ = std::move(impl);
+  http_response.impl_ = std::make_unique<HttpResponse::Impl>(std::move(response));
   return http_response;
 }
 
@@ -169,52 +173,61 @@ Result<HttpResponse> HttpClient::PostForm(
     const std::unordered_map<std::string, std::string>& form_data,
     const std::unordered_map<std::string, std::string>& headers,
     const ErrorHandler& error_handler) {
-  std::scoped_lock<std::mutex> lock(session_mutex_);
+  cpr::Response response;
 
-  // Override default Content-Type (application/json) with form-urlencoded
-  auto form_headers = headers;
-  form_headers[kHeaderContentType] = kMimeTypeFormUrlEncoded;
+  {
+    std::scoped_lock<std::mutex> lock(session_mutex_);
 
-  PrepareSession(path, form_headers);
-  std::vector<cpr::Pair> pair_list;
-  pair_list.reserve(form_data.size());
-  for (const auto& [key, val] : form_data) {
-    pair_list.emplace_back(key, val);
+    // Override default Content-Type (application/json) with form-urlencoded
+    auto form_headers = headers;
+    form_headers[kHeaderContentType] = kMimeTypeFormUrlEncoded;
+
+    PrepareSession(path, form_headers);
+    std::vector<cpr::Pair> pair_list;
+    pair_list.reserve(form_data.size());
+    for (const auto& [key, val] : form_data) {
+      pair_list.emplace_back(key, val);
+    }
+    session_->SetPayload(cpr::Payload(pair_list.begin(), pair_list.end()));
+
+    response = session_->Post();
   }
-  session_->SetPayload(cpr::Payload(pair_list.begin(), pair_list.end()));
-  cpr::Response response = session_->Post();
+
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
-  auto impl = std::make_unique<HttpResponse::Impl>(std::move(response));
   HttpResponse http_response;
-  http_response.impl_ = std::move(impl);
+  http_response.impl_ = std::make_unique<HttpResponse::Impl>(std::move(response));
   return http_response;
 }
 
 Result<HttpResponse> HttpClient::Head(
     const std::string& path, const std::unordered_map<std::string, std::string>& headers,
     const ErrorHandler& error_handler) {
-  std::scoped_lock<std::mutex> lock(session_mutex_);
+  cpr::Response response;
+  {
+    std::scoped_lock<std::mutex> lock(session_mutex_);
+    PrepareSession(path, headers);
+    response = session_->Head();
+  }
 
-  PrepareSession(path, headers);
-  cpr::Response response = session_->Head();
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
-  auto impl = std::make_unique<HttpResponse::Impl>(std::move(response));
   HttpResponse http_response;
-  http_response.impl_ = std::move(impl);
+  http_response.impl_ = std::make_unique<HttpResponse::Impl>(std::move(response));
   return http_response;
 }
 
 Result<HttpResponse> HttpClient::Delete(
     const std::string& path, const std::unordered_map<std::string, std::string>& headers,
     const ErrorHandler& error_handler) {
-  std::scoped_lock<std::mutex> lock(session_mutex_);
+  cpr::Response response;
+  {
+    std::scoped_lock<std::mutex> lock(session_mutex_);
+    PrepareSession(path, headers);
+    response = session_->Delete();
+  }
 
-  PrepareSession(path, headers);
-  cpr::Response response = session_->Delete();
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
-  auto impl = std::make_unique<HttpResponse::Impl>(std::move(response));
   HttpResponse http_response;
-  http_response.impl_ = std::move(impl);
+  http_response.impl_ = std::make_unique<HttpResponse::Impl>(std::move(response));
   return http_response;
 }
 

src/iceberg/catalog/rest/http_client.h

@@ -25,8 +25,8 @@
 #include <string>
 #include <unordered_map>
 
-#include "iceberg/catalog/rest/error_handlers.h"
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
+#include "iceberg/catalog/rest/type_fwd.h"
 #include "iceberg/result.h"
 
 /// \file iceberg/catalog/rest/http_client.h
@@ -114,12 +114,10 @@ class ICEBERG_REST_EXPORT HttpClient {
 
   std::unordered_map<std::string, std::string> default_headers_;
 
-  // Mutex to protect the non-thread-safe cpr::Session.
-  mutable std::mutex session_mutex_;
-
   // TODO(Li Feiyang): use connection pool to support external multi-threaded concurrent
   // calls
   std::unique_ptr<cpr::Session> session_;
+  mutable std::mutex session_mutex_;
 };
 
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/meson.build

@@ -65,6 +65,7 @@ install_headers(
         'resource_paths.h',
         'rest_catalog.h',
         'rest_util.h',
+        'type_fwd.h',
         'types.h',
     ],
     subdir: 'iceberg/catalog/rest',