feat: first version of rest catalog

Author: HeartLinked

src/iceberg/catalog/rest/CMakeLists.txt

@@ -15,7 +15,12 @@
 # specific language governing permissions and limitations
 # under the License.
 
-set(ICEBERG_REST_SOURCES rest_catalog.cc json_internal.cc)
+set(ICEBERG_REST_SOURCES
+    catalog.cc
+    json_internal.cc
+    config.cc
+    http_client_internal.cc
+    resource_paths.cc)
 
 set(ICEBERG_REST_STATIC_BUILD_INTERFACE_LIBS)
 set(ICEBERG_REST_SHARED_BUILD_INTERFACE_LIBS)

src/iceberg/catalog/rest/catalog.cc

@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/catalog.h"
+
+#include <memory>
+#include <utility>
+
+#include <cpr/cpr.h>
+
+#include "iceberg/catalog/rest/config.h"
+#include "iceberg/catalog/rest/constant.h"
+#include "iceberg/catalog/rest/endpoint_util.h"
+#include "iceberg/catalog/rest/http_client_interal.h"
+#include "iceberg/catalog/rest/json_internal.h"
+#include "iceberg/catalog/rest/types.h"
+#include "iceberg/json_internal.h"
+#include "iceberg/result.h"
+#include "iceberg/table.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest {
+
+Result<std::unique_ptr<RestCatalog>> RestCatalog::Make(const RestCatalogConfig& config) {
+  // Create ResourcePaths and validate URI
+  ICEBERG_ASSIGN_OR_RAISE(auto paths, ResourcePaths::Make(config));
+
+  ICEBERG_ASSIGN_OR_RAISE(auto tmp_client, HttpClient::Make(config));
+
+  const std::string endpoint = paths->V1Config();
+  cpr::Parameters params;
+  ICEBERG_ASSIGN_OR_RAISE(const auto& response, tmp_client->Get(endpoint, params));
+  switch (response.status_code) {
+    case cpr::status::HTTP_OK: {
+      ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.text));
+      ICEBERG_ASSIGN_OR_RAISE(auto server_config, CatalogConfigFromJson(json));
+      // Merge server config into client config, server config overrides > client config
+      // properties > server config defaults
+      auto final_props = std::move(server_config.defaults);
+      for (const auto& kv : config.configs()) {
+        final_props.insert_or_assign(kv.first, kv.second);
+      }
+
+      for (const auto& kv : server_config.overrides) {
+        final_props.insert_or_assign(kv.first, kv.second);
+      }
+      auto final_config = RestCatalogConfig::FromMap(final_props);
+      ICEBERG_ASSIGN_OR_RAISE(auto client, HttpClient::Make(*final_config));
+      ICEBERG_ASSIGN_OR_RAISE(auto final_paths, ResourcePaths::Make(*final_config));
+      return std::unique_ptr<RestCatalog>(new RestCatalog(
+          std::move(final_config), std::move(client), std::move(*final_paths)));
+    };
+    default: {
+      ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.text));
+      ICEBERG_ASSIGN_OR_RAISE(auto list_response, ErrorResponseFromJson(json));
+      return UnknownError("Error listing namespaces: {}", list_response.error.message);
+    }
+  }
+}
+
+RestCatalog::RestCatalog(std::unique_ptr<RestCatalogConfig> config,
+                         std::unique_ptr<HttpClient> client, ResourcePaths paths)
+    : config_(std::move(config)), client_(std::move(client)), paths_(std::move(paths)) {}
+
+std::string_view RestCatalog::name() const {
+  auto it = config_->configs().find(std::string(RestCatalogConfig::kName));
+  if (it == config_->configs().end() || it->second.empty()) {
+    return {""};
+  }
+  return std::string_view(it->second);
+}
+
+Result<std::vector<Namespace>> RestCatalog::ListNamespaces(const Namespace& ns) const {
+  const std::string endpoint = paths_.V1Namespaces();
+  std::vector<Namespace> result;
+  std::string next_token;
+  while (true) {
+    cpr::Parameters params;
+    if (!ns.levels.empty()) {
+      params.Add({std::string(kQueryParamParent), EncodeNamespaceForUrl(ns)});
+    }
+    if (!next_token.empty()) {
+      params.Add({std::string(kQueryParamPageToken), next_token});
+    }
+    ICEBERG_ASSIGN_OR_RAISE(const auto& response, client_->Get(endpoint, params));
+    switch (response.status_code) {
+      case cpr::status::HTTP_OK: {
+        ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.text));
+        ICEBERG_ASSIGN_OR_RAISE(auto list_response, ListNamespacesResponseFromJson(json));
+        result.insert(result.end(), list_response.namespaces.begin(),
+                      list_response.namespaces.end());
+        if (list_response.next_page_token.empty()) {
+          return result;
+        }
+        next_token = list_response.next_page_token;
+        continue;
+      }
+      case cpr::status::HTTP_NOT_FOUND: {
+        return NoSuchNamespace("Namespace not found");
+      }
+      default:
+        ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.text));
+        ICEBERG_ASSIGN_OR_RAISE(auto list_response, ErrorResponseFromJson(json));
+        return UnknownError("Error listing namespaces: {}", list_response.error.message);
+    }
+  }
+}
+
+Status RestCatalog::CreateNamespace(
+    const Namespace& ns, const std::unordered_map<std::string, std::string>& properties) {
+  return NotImplemented("Not implemented");
+}
+
+Result<std::unordered_map<std::string, std::string>> RestCatalog::GetNamespaceProperties(
+    const Namespace& ns) const {
+  return NotImplemented("Not implemented");
+}
+
+Status RestCatalog::DropNamespace(const Namespace& ns) {
+  return NotImplemented("Not implemented");
+}
+
+Result<bool> RestCatalog::NamespaceExists(const Namespace& ns) const {
+  return NotImplemented("Not implemented");
+}
+
+Status RestCatalog::UpdateNamespaceProperties(
+    const Namespace& ns, const std::unordered_map<std::string, std::string>& updates,
+    const std::unordered_set<std::string>& removals) {
+  return NotImplemented("Not implemented");
+}
+
+Result<std::vector<TableIdentifier>> RestCatalog::ListTables(const Namespace& ns) const {
+  return NotImplemented("Not implemented");
+}
+
+Result<std::unique_ptr<Table>> RestCatalog::CreateTable(
+    const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
+    const std::string& location,
+    const std::unordered_map<std::string, std::string>& properties) {
+  return NotImplemented("Not implemented");
+}
+
+Result<std::unique_ptr<Table>> RestCatalog::UpdateTable(
+    const TableIdentifier& identifier,
+    const std::vector<std::unique_ptr<TableRequirement>>& requirements,
+    const std::vector<std::unique_ptr<TableUpdate>>& updates) {
+  return NotImplemented("Not implemented");
+}
+
+Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
+    const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
+    const std::string& location,
+    const std::unordered_map<std::string, std::string>& properties) {
+  return NotImplemented("Not implemented");
+}
+
+Status RestCatalog::DropTable(const TableIdentifier& identifier, bool purge) {
+  return NotImplemented("Not implemented");
+}
+
+Result<bool> RestCatalog::TableExists(const TableIdentifier& identifier) const {
+  return NotImplemented("Not implemented");
+}
+
+Status RestCatalog::RenameTable(const TableIdentifier& from, const TableIdentifier& to) {
+  return NotImplemented("Not implemented");
+}
+
+Result<std::unique_ptr<Table>> RestCatalog::LoadTable(const TableIdentifier& identifier) {
+  return NotImplemented("Not implemented");
+}
+
+Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(
+    const TableIdentifier& identifier, const std::string& metadata_file_location) {
+  return NotImplemented("Not implemented");
+}
+
+std::unique_ptr<RestCatalog::TableBuilder> RestCatalog::BuildTable(
+    const TableIdentifier& identifier, const Schema& schema) const {
+  return nullptr;
+}
+
+}  // namespace iceberg::rest

src/iceberg/catalog/rest/catalog.h

@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <memory>
+#include <string>
+
+#include "iceberg/catalog.h"
+#include "iceberg/catalog/rest/config.h"
+#include "iceberg/catalog/rest/http_client_interal.h"
+#include "iceberg/catalog/rest/iceberg_rest_export.h"
+#include "iceberg/catalog/rest/resource_paths.h"
+#include "iceberg/result.h"
+
+/// \file iceberg/catalog/rest/catalog.h
+/// RestCatalog implementation for Iceberg REST API.
+
+namespace iceberg::rest {
+
+class ICEBERG_REST_EXPORT RestCatalog : public Catalog {
+ public:
+  RestCatalog(const RestCatalog&) = delete;
+  RestCatalog& operator=(const RestCatalog&) = delete;
+  RestCatalog(RestCatalog&&) = delete;
+  RestCatalog& operator=(RestCatalog&&) = delete;
+
+  /// \brief Create a RestCatalog instance
+  ///
+  /// \param config the configuration for the RestCatalog
+  /// \return a unique_ptr to RestCatalog instance
+  static Result<std::unique_ptr<RestCatalog>> Make(const RestCatalogConfig& config);
+
+  std::string_view name() const override;
+
+  Result<std::vector<Namespace>> ListNamespaces(const Namespace& ns) const override;
+
+  Status CreateNamespace(
+      const Namespace& ns,
+      const std::unordered_map<std::string, std::string>& properties) override;
+
+  Result<std::unordered_map<std::string, std::string>> GetNamespaceProperties(
+      const Namespace& ns) const override;
+
+  Status DropNamespace(const Namespace& ns) override;
+
+  Result<bool> NamespaceExists(const Namespace& ns) const override;
+
+  Status UpdateNamespaceProperties(
+      const Namespace& ns, const std::unordered_map<std::string, std::string>& updates,
+      const std::unordered_set<std::string>& removals) override;
+
+  Result<std::vector<TableIdentifier>> ListTables(const Namespace& ns) const override;
+
+  Result<std::unique_ptr<Table>> CreateTable(
+      const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
+      const std::string& location,
+      const std::unordered_map<std::string, std::string>& properties) override;
+
+  Result<std::unique_ptr<Table>> UpdateTable(
+      const TableIdentifier& identifier,
+      const std::vector<std::unique_ptr<TableRequirement>>& requirements,
+      const std::vector<std::unique_ptr<TableUpdate>>& updates) override;
+
+  Result<std::shared_ptr<Transaction>> StageCreateTable(
+      const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
+      const std::string& location,
+      const std::unordered_map<std::string, std::string>& properties) override;
+
+  Result<bool> TableExists(const TableIdentifier& identifier) const override;
+
+  Status RenameTable(const TableIdentifier& from, const TableIdentifier& to) override;
+
+  Status DropTable(const TableIdentifier& identifier, bool purge) override;
+
+  Result<std::unique_ptr<Table>> LoadTable(const TableIdentifier& identifier) override;
+
+  Result<std::shared_ptr<Table>> RegisterTable(
+      const TableIdentifier& identifier,
+      const std::string& metadata_file_location) override;
+
+  std::unique_ptr<RestCatalog::TableBuilder> BuildTable(
+      const TableIdentifier& identifier, const Schema& schema) const override;
+
+ private:
+  RestCatalog(std::unique_ptr<RestCatalogConfig> config,
+              std::unique_ptr<HttpClient> client, ResourcePaths paths);
+
+  std::unique_ptr<RestCatalogConfig> config_;
+  std::unique_ptr<HttpClient> client_;
+  ResourcePaths paths_;
+};
+
+}  // namespace iceberg::rest

src/iceberg/catalog/rest/config.cc

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/config.h"
+
+#include "iceberg/catalog/rest/constant.h"
+
+namespace iceberg::rest {
+
+std::unique_ptr<RestCatalogConfig> RestCatalogConfig::default_properties() {
+  return std::make_unique<RestCatalogConfig>();
+}
+
+std::unique_ptr<RestCatalogConfig> RestCatalogConfig::FromMap(
+    const std::unordered_map<std::string, std::string>& properties) {
+  auto rest_catalog_config = std::make_unique<RestCatalogConfig>();
+  rest_catalog_config->configs_ = properties;
+  return rest_catalog_config;
+}
+
+Result<cpr::Header> RestCatalogConfig::GetExtraHeaders() const {
+  cpr::Header headers;
+
+  headers[std::string(kHeaderContentType)] = std::string(kMimeTypeApplicationJson);
+  headers[std::string(kHeaderUserAgent)] = std::string(kUserAgent);
+  headers[std::string(kHeaderXClientVersion)] = std::string(kClientVersion);
+
+  constexpr std::string_view prefix = "header.";
+  for (const auto& [key, value] : configs_) {
+    if (key.starts_with(prefix)) {
+      std::string_view header_name = std::string_view(key).substr(prefix.length());
+
+      if (header_name.empty()) {
+        return InvalidArgument("Header name cannot be empty after '{}' prefix", prefix);
+      }
+
+      if (value.empty()) {
+        return InvalidArgument("Header value for '{}' cannot be empty", header_name);
+      }
+      headers[std::string(header_name)] = value;
+    }
+  }
+  return headers;
+}
+
+}  // namespace iceberg::rest