feat(rest): implement create table

Author: HeartLinked

src/iceberg/catalog.h

@@ -107,11 +107,13 @@ class ICEBERG_EXPORT Catalog {
   /// \param identifier a table identifier
   /// \param schema a schema
   /// \param spec a partition spec
+  /// \param order a sort order
   /// \param location a location for the table; leave empty if unspecified
   /// \param properties a string map of table properties
   /// \return a Table instance or ErrorKind::kAlreadyExists if the table already exists
   virtual Result<std::shared_ptr<Table>> CreateTable(
-      const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
+      const TableIdentifier& identifier, const std::shared_ptr<Schema>& schema,
+      const std::shared_ptr<PartitionSpec>& spec, const std::shared_ptr<SortOrder>& order,
       const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) = 0;
 
@@ -131,12 +133,14 @@ class ICEBERG_EXPORT Catalog {
   /// \param identifier a table identifier
   /// \param schema a schema
   /// \param spec a partition spec
+  /// \param order a sort order
   /// \param location a location for the table; leave empty if unspecified
   /// \param properties a string map of table properties
   /// \return a Transaction to create the table or ErrorKind::kAlreadyExists if the
   /// table already exists
   virtual Result<std::shared_ptr<Transaction>> StageCreateTable(
-      const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
+      const TableIdentifier& identifier, const std::shared_ptr<Schema>& schema,
+      const std::shared_ptr<PartitionSpec>& spec, const std::shared_ptr<SortOrder>& order,
       const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) = 0;
 

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -400,7 +400,8 @@ Result<std::vector<TableIdentifier>> InMemoryCatalog::ListTables(
 }
 
 Result<std::shared_ptr<Table>> InMemoryCatalog::CreateTable(
-    const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
+    const TableIdentifier& identifier, const std::shared_ptr<Schema>& schema,
+    const std::shared_ptr<PartitionSpec>& spec, const std::shared_ptr<SortOrder>& order,
     const std::string& location,
     const std::unordered_map<std::string, std::string>& properties) {
   std::unique_lock lock(mutex_);
@@ -439,7 +440,8 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::UpdateTable(
 }
 
 Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
-    const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
+    const TableIdentifier& identifier, const std::shared_ptr<Schema>& schema,
+    const std::shared_ptr<PartitionSpec>& spec, const std::shared_ptr<SortOrder>& order,
     const std::string& location,
     const std::unordered_map<std::string, std::string>& properties) {
   std::unique_lock lock(mutex_);

src/iceberg/catalog/memory/in_memory_catalog.h

@@ -71,7 +71,8 @@ class ICEBERG_EXPORT InMemoryCatalog
   Result<std::vector<TableIdentifier>> ListTables(const Namespace& ns) const override;
 
   Result<std::shared_ptr<Table>> CreateTable(
-      const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
+      const TableIdentifier& identifier, const std::shared_ptr<Schema>& schema,
+      const std::shared_ptr<PartitionSpec>& spec, const std::shared_ptr<SortOrder>& order,
       const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) override;
 
@@ -81,7 +82,8 @@ class ICEBERG_EXPORT InMemoryCatalog
       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 TableIdentifier& identifier, const std::shared_ptr<Schema>& schema,
+      const std::shared_ptr<PartitionSpec>& spec, const std::shared_ptr<SortOrder>& order,
       const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) override;
 

src/iceberg/catalog/rest/json_internal.cc

@@ -19,6 +19,7 @@
 
 #include "iceberg/catalog/rest/json_internal.h"
 
+#include <memory>
 #include <string>
 #include <utility>
 #include <vector>
@@ -27,6 +28,8 @@
 
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/json_internal.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/sort_order.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/util/json_util_internal.h"
 #include "iceberg/util/macros.h"
@@ -336,6 +339,57 @@ Result<ListTablesResponse> ListTablesResponseFromJson(const nlohmann::json& json
   return response;
 }
 
+nlohmann::json ToJson(const CreateTableRequest& request) {
+  nlohmann::json json;
+  json[kName] = request.name;
+  SetOptionalStringField(json, kLocation, request.location);
+  if (request.schema) {
+    json[kSchema] = ToJson(*request.schema);
+  }
+  if (request.partition_spec) {
+    json[kPartitionSpec] = ToJson(*request.partition_spec);
+  }
+  if (request.write_order) {
+    json[kWriteOrder] = ToJson(*request.write_order);
+  }
+  if (request.stage_create) {
+    json[kStageCreate] = request.stage_create;
+  }
+  SetContainerField(json, kProperties, request.properties);
+  return json;
+}
+
+Result<CreateTableRequest> CreateTableRequestFromJson(const nlohmann::json& json) {
+  CreateTableRequest request;
+  ICEBERG_ASSIGN_OR_RAISE(request.name, GetJsonValue<std::string>(json, kName));
+  ICEBERG_ASSIGN_OR_RAISE(request.location,
+                          GetJsonValueOrDefault<std::string>(json, kLocation));
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_json, GetJsonValue<nlohmann::json>(json, kSchema));
+  ICEBERG_ASSIGN_OR_RAISE(request.schema, SchemaFromJson(schema_json));
+
+  if (json.contains(kPartitionSpec)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_spec,
+                            GetJsonValue<nlohmann::json>(json, kPartitionSpec));
+    ICEBERG_ASSIGN_OR_RAISE(request.partition_spec,
+                            PartitionSpecFromJson(request.schema, partition_spec,
+                                                  PartitionSpec::kInitialSpecId));
+  }
+  if (json.contains(kWriteOrder)) {
+    ICEBERG_ASSIGN_OR_RAISE(auto sort_order_json,
+                            GetJsonValue<nlohmann::json>(json, kWriteOrder));
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order,
+                            SortOrderFromJson(sort_order_json, request.schema));
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(request.stage_create,
+                          GetJsonValueOrDefault<bool>(json, kStageCreate, false));
+  ICEBERG_ASSIGN_OR_RAISE(
+      request.properties,
+      GetJsonValueOrDefault<decltype(request.properties)>(json, kProperties));
+  ICEBERG_RETURN_UNEXPECTED(request.Validate());
+  return request;
+}
+
 #define ICEBERG_DEFINE_FROM_JSON(Model)                       \
   template <>                                                 \
   Result<Model> FromJson<Model>(const nlohmann::json& json) { \
@@ -354,5 +408,6 @@ ICEBERG_DEFINE_FROM_JSON(ListTablesResponse)
 ICEBERG_DEFINE_FROM_JSON(LoadTableResult)
 ICEBERG_DEFINE_FROM_JSON(RegisterTableRequest)
 ICEBERG_DEFINE_FROM_JSON(RenameTableRequest)
+ICEBERG_DEFINE_FROM_JSON(CreateTableRequest)
 
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/json_internal.h

@@ -55,6 +55,7 @@ ICEBERG_DECLARE_JSON_SERDE(ListTablesResponse)
 ICEBERG_DECLARE_JSON_SERDE(LoadTableResult)
 ICEBERG_DECLARE_JSON_SERDE(RegisterTableRequest)
 ICEBERG_DECLARE_JSON_SERDE(RenameTableRequest)
+ICEBERG_DECLARE_JSON_SERDE(CreateTableRequest)
 
 #undef ICEBERG_DECLARE_JSON_SERDE
 

src/iceberg/catalog/rest/rest_catalog.cc

@@ -40,6 +40,7 @@
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
+#include "iceberg/sort_order.h"
 #include "iceberg/table.h"
 #include "iceberg/util/macros.h"
 
@@ -77,8 +78,8 @@ Result<CatalogConfig> FetchServerConfig(const ResourcePaths& paths,
 
 RestCatalog::~RestCatalog() = default;
 
-Result<std::unique_ptr<RestCatalog>> RestCatalog::Make(
-    const RestCatalogProperties& config) {
+Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
+    const RestCatalogProperties& config, std::shared_ptr<FileIO> file_io) {
   ICEBERG_ASSIGN_OR_RAISE(auto uri, config.Uri());
   ICEBERG_ASSIGN_OR_RAISE(
       auto paths, ResourcePaths::Make(std::string(TrimTrailingSlash(uri)),
@@ -103,14 +104,17 @@ Result<std::unique_ptr<RestCatalog>> RestCatalog::Make(
   ICEBERG_ASSIGN_OR_RAISE(auto final_uri, final_config->Uri());
   ICEBERG_RETURN_UNEXPECTED(paths->SetBaseUri(std::string(TrimTrailingSlash(final_uri))));
 
-  return std::unique_ptr<RestCatalog>(
-      new RestCatalog(std::move(final_config), std::move(paths), std::move(endpoints)));
+  return std::shared_ptr<RestCatalog>(
+      new RestCatalog(std::move(final_config), std::move(file_io), std::move(paths),
+                      std::move(endpoints)));
 }
 
 RestCatalog::RestCatalog(std::unique_ptr<RestCatalogProperties> config,
+                         std::shared_ptr<FileIO> file_io,
                          std::unique_ptr<ResourcePaths> paths,
                          std::unordered_set<Endpoint> endpoints)
     : config_(std::move(config)),
+      file_io_(std::move(file_io)),
       client_(std::make_unique<HttpClient>(config_->ExtractHeaders())),
       paths_(std::move(paths)),
       name_(config_->Get(RestCatalogProperties::kName)),
@@ -241,11 +245,33 @@ Result<std::vector<TableIdentifier>> RestCatalog::ListTables(
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::CreateTable(
-    [[maybe_unused]] const TableIdentifier& identifier,
-    [[maybe_unused]] const Schema& schema, [[maybe_unused]] const PartitionSpec& spec,
-    [[maybe_unused]] const std::string& location,
-    [[maybe_unused]] const std::unordered_map<std::string, std::string>& properties) {
-  return NotImplemented("Not implemented");
+    const TableIdentifier& identifier, const std::shared_ptr<Schema>& schema,
+    const std::shared_ptr<PartitionSpec>& spec, const std::shared_ptr<SortOrder>& order,
+    const std::string& location,
+    const std::unordered_map<std::string, std::string>& properties) {
+  ICEBERG_RETURN_UNEXPECTED(CheckEndpoint(supported_endpoints_, Endpoint::CreateTable()));
+  ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Tables(identifier.ns));
+
+  CreateTableRequest request{
+      .name = identifier.name,
+      .location = location,
+      .schema = schema,
+      .partition_spec = spec,
+      .write_order = order,
+      .stage_create = false,
+      .properties = properties,
+  };
+
+  ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
+  ICEBERG_ASSIGN_OR_RAISE(
+      const auto response,
+      client_->Post(path, json_request, /*headers=*/{}, *TableErrorHandler::Instance()));
+
+  ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
+  ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
+  return Table::Make(identifier, load_result.metadata,
+                     std::move(load_result.metadata_location), file_io_,
+                     shared_from_this());
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
@@ -257,7 +283,9 @@ Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
 
 Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
     [[maybe_unused]] const TableIdentifier& identifier,
-    [[maybe_unused]] const Schema& schema, [[maybe_unused]] const PartitionSpec& spec,
+    [[maybe_unused]] const std::shared_ptr<Schema>& schema,
+    [[maybe_unused]] const std::shared_ptr<PartitionSpec>& spec,
+    [[maybe_unused]] const std::shared_ptr<SortOrder>& order,
     [[maybe_unused]] const std::string& location,
     [[maybe_unused]] const std::unordered_map<std::string, std::string>& properties) {
   return NotImplemented("Not implemented");

src/iceberg/catalog/rest/rest_catalog.h

@@ -20,8 +20,8 @@
 #pragma once
 
 #include <memory>
-#include <set>
 #include <string>
+#include <unordered_set>
 
 #include "iceberg/catalog.h"
 #include "iceberg/catalog/rest/endpoint.h"
@@ -35,7 +35,8 @@
 namespace iceberg::rest {
 
 /// \brief Rest catalog implementation.
-class ICEBERG_REST_EXPORT RestCatalog : public Catalog {
+class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
+                                        public std::enable_shared_from_this<RestCatalog> {
  public:
   ~RestCatalog() override;
 
@@ -47,8 +48,10 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog {
   /// \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 RestCatalogProperties& config);
+  /// \param file_io the FileIO instance to use for table operations
+  /// \return a shared_ptr to RestCatalog instance
+  static Result<std::shared_ptr<RestCatalog>> Make(const RestCatalogProperties& config,
+                                                   std::shared_ptr<FileIO> file_io);
 
   std::string_view name() const override;
 
@@ -72,7 +75,8 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog {
   Result<std::vector<TableIdentifier>> ListTables(const Namespace& ns) const override;
 
   Result<std::shared_ptr<Table>> CreateTable(
-      const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
+      const TableIdentifier& identifier, const std::shared_ptr<Schema>& schema,
+      const std::shared_ptr<PartitionSpec>& spec, const std::shared_ptr<SortOrder>& order,
       const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) override;
 
@@ -82,7 +86,8 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog {
       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 TableIdentifier& identifier, const std::shared_ptr<Schema>& schema,
+      const std::shared_ptr<PartitionSpec>& spec, const std::shared_ptr<SortOrder>& order,
       const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) override;
 
@@ -100,10 +105,11 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog {
 
  private:
   RestCatalog(std::unique_ptr<RestCatalogProperties> config,
-              std::unique_ptr<ResourcePaths> paths,
+              std::shared_ptr<FileIO> file_io, std::unique_ptr<ResourcePaths> paths,
               std::unordered_set<Endpoint> endpoints);
 
   std::unique_ptr<RestCatalogProperties> config_;
+  std::shared_ptr<FileIO> file_io_;
   std::unique_ptr<HttpClient> client_;
   std::unique_ptr<ResourcePaths> paths_;
   std::string name_;

src/iceberg/catalog/rest/types.h

@@ -27,6 +27,7 @@
 #include "iceberg/catalog/rest/endpoint.h"
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
 #include "iceberg/result.h"
+#include "iceberg/schema.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/type_fwd.h"
 #include "iceberg/util/macros.h"
@@ -138,6 +139,28 @@ struct ICEBERG_REST_EXPORT RenameTableRequest {
   bool operator==(const RenameTableRequest&) const = default;
 };
 
+/// \brief Request to create a table.
+struct ICEBERG_REST_EXPORT CreateTableRequest {
+  std::string name;  // required
+  std::string location;
+  std::shared_ptr<Schema> schema;  // required
+  std::shared_ptr<PartitionSpec> partition_spec;
+  std::shared_ptr<SortOrder> write_order;
+  bool stage_create = false;
+  std::unordered_map<std::string, std::string> properties;
+
+  /// \brief Validates the CreateTableRequest.
+  Status Validate() const {
+    if (name.empty()) {
+      return Invalid("Missing table name");
+    }
+    if (!schema) {
+      return Invalid("Missing schema");
+    }
+    return {};
+  }
+};
+
 /// \brief An opaque token that allows clients to make use of pagination for list APIs.
 using PageToken = std::string;
 

src/iceberg/json_internal.cc

@@ -220,18 +220,6 @@ Result<std::unique_ptr<SortOrder>> SortOrderFromJson(
   return SortOrder::Make(*current_schema, order_id, std::move(sort_fields));
 }
 
-Result<std::unique_ptr<SortOrder>> SortOrderFromJson(const nlohmann::json& json) {
-  ICEBERG_ASSIGN_OR_RAISE(auto order_id, GetJsonValue<int32_t>(json, kOrderId));
-  ICEBERG_ASSIGN_OR_RAISE(auto fields, GetJsonValue<nlohmann::json>(json, kFields));
-
-  std::vector<SortField> sort_fields;
-  for (const auto& field_json : fields) {
-    ICEBERG_ASSIGN_OR_RAISE(auto sort_field, SortFieldFromJson(field_json));
-    sort_fields.push_back(std::move(*sort_field));
-  }
-  return SortOrder::Make(order_id, std::move(sort_fields));
-}
-
 nlohmann::json ToJson(const SchemaField& field) {
   nlohmann::json json;
   json[kId] = field.field_id();

src/iceberg/test/mock_catalog.h

@@ -56,7 +56,8 @@ class MockCatalog : public Catalog {
               (const, override));
 
   MOCK_METHOD((Result<std::shared_ptr<Table>>), CreateTable,
-              (const TableIdentifier&, const Schema&, const PartitionSpec&,
+              (const TableIdentifier&, const std::shared_ptr<Schema>&,
+               const std::shared_ptr<PartitionSpec>&, const std::shared_ptr<SortOrder>&,
                const std::string&, (const std::unordered_map<std::string, std::string>&)),
               (override));
 
@@ -67,7 +68,8 @@ class MockCatalog : public Catalog {
               (override));
 
   MOCK_METHOD((Result<std::shared_ptr<Transaction>>), StageCreateTable,
-              (const TableIdentifier&, const Schema&, const PartitionSpec&,
+              (const TableIdentifier&, const std::shared_ptr<Schema>&,
+               const std::shared_ptr<PartitionSpec>&, const std::shared_ptr<SortOrder>&,
                const std::string&, (const std::unordered_map<std::string, std::string>&)),
               (override));