rest: rebuild resource paths from final config

Author: plusplusjiajia

src/iceberg/catalog/rest/rest_catalog.cc

@@ -117,7 +117,7 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
   ICEBERG_ASSIGN_OR_RAISE(auto server_config, FetchServerConfig(*paths, config));
 
   std::unique_ptr<RestCatalogProperties> final_config = RestCatalogProperties::FromMap(
-      MergeConfigs(server_config.overrides, config.configs(), server_config.defaults));
+      MergeConfigs(server_config.defaults, config.configs(), server_config.overrides));
 
   std::unordered_set<Endpoint> endpoints;
   if (!server_config.endpoints.empty()) {
@@ -132,7 +132,11 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
 
   // Update resource paths based on the final config
   ICEBERG_ASSIGN_OR_RAISE(auto final_uri, final_config->Uri());
-  ICEBERG_RETURN_UNEXPECTED(paths->SetBaseUri(std::string(TrimTrailingSlash(final_uri))));
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto final_paths,
+      ResourcePaths::Make(std::string(TrimTrailingSlash(final_uri)),
+                          final_config->Get(RestCatalogProperties::kPrefix)));
+  paths = std::move(final_paths);
 
   return std::shared_ptr<RestCatalog>(
       new RestCatalog(std::move(final_config), std::move(file_io), std::move(paths),

src/iceberg/test/rest_util_test.cc

@@ -19,14 +19,134 @@
 
 #include "iceberg/catalog/rest/rest_util.h"
 
+#include <arpa/inet.h>
 #include <gtest/gtest.h>
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <unistd.h>
 
+#include <string>
+#include <thread>
+
+#include "iceberg/catalog/rest/catalog_properties.h"
 #include "iceberg/catalog/rest/endpoint.h"
+#include "iceberg/catalog/rest/rest_catalog.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/test/matchers.h"
+#include "iceberg/test/std_io.h"
 
 namespace iceberg::rest {
 
+namespace {
+
+class StubHttpServer {
+ public:
+  explicit StubHttpServer(std::string config_json) : config_json_(std::move(config_json)) {}
+
+  ~StubHttpServer() { Join(); }
+
+  void Start() {
+    listen_fd_ = socket(AF_INET, SOCK_STREAM, 0);
+    ASSERT_GE(listen_fd_, 0);
+
+    int opt = 1;
+    setsockopt(listen_fd_, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
+
+    sockaddr_in addr{};
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    addr.sin_port = htons(0);  // ephemeral port
+    ASSERT_EQ(bind(listen_fd_, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)), 0);
+    ASSERT_EQ(listen(listen_fd_, 16), 0);
+
+    socklen_t len = sizeof(addr);
+    ASSERT_EQ(getsockname(listen_fd_, reinterpret_cast<sockaddr*>(&addr), &len), 0);
+    port_ = ntohs(addr.sin_port);
+
+    server_thread_ = std::thread([this]() { ServeTwoRequests(); });
+  }
+
+  void Join() {
+    if (server_thread_.joinable()) {
+      server_thread_.join();
+    }
+    if (listen_fd_ >= 0) {
+      shutdown(listen_fd_, SHUT_RDWR);
+      close(listen_fd_);
+      listen_fd_ = -1;
+    }
+  }
+
+  uint16_t port() const { return port_; }
+
+  const std::string& first_path() const { return first_path_; }
+  const std::string& second_path() const { return second_path_; }
+
+ private:
+  static std::string buildJsonResponse(const std::string& body) {
+    std::string resp = "HTTP/1.1 200 OK\r\n";
+    resp += "Content-Type: application/json\r\n";
+    resp += "Connection: close\r\n";
+    resp += "Content-Length: " + std::to_string(body.size()) + "\r\n";
+    resp += "\r\n";
+    resp += body;
+    return resp;
+  }
+
+  static std::string extractPathFromRequest(const std::string& req) {
+    auto line_end = req.find("\r\n");
+    std::string first_line = (line_end == std::string::npos) ? req : req.substr(0, line_end);
+    auto sp1 = first_line.find(' ');
+    auto sp2 = (sp1 == std::string::npos) ? std::string::npos : first_line.find(' ', sp1 + 1);
+    if (sp1 != std::string::npos && sp2 != std::string::npos && sp2 > sp1 + 1) {
+      return first_line.substr(sp1 + 1, sp2 - (sp1 + 1));
+    }
+    return {};
+  }
+
+  void ServeOneRequest(std::string& out_path) {
+    sockaddr_in client_addr{};
+    socklen_t client_len = sizeof(client_addr);
+    int fd = accept(listen_fd_, reinterpret_cast<sockaddr*>(&client_addr), &client_len);
+    ASSERT_GE(fd, 0);
+
+    std::string req;
+    char buf[4096];
+    ssize_t n = read(fd, buf, sizeof(buf));
+    if (n > 0) {
+      req.assign(buf, buf + n);
+    }
+
+    out_path = extractPathFromRequest(req);
+
+    std::string resp;
+    if (out_path == "/v1/config") {
+      resp = buildJsonResponse(config_json_);
+    } else {
+      resp = buildJsonResponse(R"({"namespaces":[]})");
+    }
+
+    (void)write(fd, resp.data(), resp.size());
+    shutdown(fd, SHUT_RDWR);
+    close(fd);
+  }
+
+  void ServeTwoRequests() {
+    ServeOneRequest(first_path_);
+    ServeOneRequest(second_path_);
+  }
+
+  std::string config_json_;
+  int listen_fd_ = -1;
+  uint16_t port_ = 0;
+  std::thread server_thread_;
+
+  std::string first_path_;
+  std::string second_path_;
+};
+
+}  // namespace
+
 TEST(RestUtilTest, TrimTrailingSlash) {
   EXPECT_EQ(TrimTrailingSlash("https://foo"), "https://foo");
   EXPECT_EQ(TrimTrailingSlash("https://foo/"), "https://foo");
@@ -154,4 +274,28 @@ TEST(RestUtilTest, MergeConfigs) {
   EXPECT_EQ(merged_empty["key"], "value");
 }
 
+TEST(RestUtilTest, PrefixIsTakenFromFinalConfig) {
+  StubHttpServer server(R"({"defaults":{},"overrides":{"prefix":"serverp"}})");
+  server.Start();
+
+  auto config = RestCatalogProperties::default_properties();
+  config->Set(RestCatalogProperties::kUri,
+              std::string("http://127.0.0.1:") + std::to_string(server.port()))
+      .Set(RestCatalogProperties::kName, std::string("test_catalog"))
+      .Set(RestCatalogProperties::kWarehouse, std::string("wh"))
+      .Set(RestCatalogProperties::kPrefix, std::string("clientp"));
+
+  auto catalog_result = RestCatalog::Make(*config, std::make_shared<test::StdFileIO>());
+  ASSERT_THAT(catalog_result, IsOk());
+  auto& catalog = catalog_result.value();
+
+  Namespace root{.levels = {}};
+  auto list_result = catalog->ListNamespaces(root);
+  ASSERT_THAT(list_result, IsOk());
+
+  server.Join();
+  EXPECT_EQ(server.first_path(), "/v1/config");
+  EXPECT_EQ(server.second_path(), "/v1/serverp/namespaces");
+}
+
 }  // namespace iceberg::rest