feat(rest): implement create table

Author: HeartLinked

src/iceberg/catalog.h

@@ -27,6 +27,7 @@
 #include <vector>
 
 #include "iceberg/result.h"
+#include "iceberg/sort_order.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/type_fwd.h"
 
@@ -107,12 +108,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::unique_ptr<Table>> CreateTable(
       const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
-      const std::string& location,
+      const SortOrder& order, const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) = 0;
 
   /// \brief Update a table
@@ -131,13 +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 std::string& location,
+      const SortOrder& order, const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) = 0;
 
   /// \brief Check whether table exists

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -401,7 +401,7 @@ Result<std::vector<TableIdentifier>> InMemoryCatalog::ListTables(
 
 Result<std::unique_ptr<Table>> InMemoryCatalog::CreateTable(
     const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
-    const std::string& location,
+    const SortOrder& order, const std::string& location,
     const std::unordered_map<std::string, std::string>& properties) {
   std::unique_lock lock(mutex_);
   return NotImplemented("create table");
@@ -441,7 +441,7 @@ Result<std::unique_ptr<Table>> InMemoryCatalog::UpdateTable(
 
 Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
     const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
-    const std::string& location,
+    const SortOrder& order, const std::string& location,
     const std::unordered_map<std::string, std::string>& properties) {
   std::unique_lock lock(mutex_);
   return NotImplemented("stage create table");

src/iceberg/catalog/memory/in_memory_catalog.h

@@ -72,7 +72,7 @@ class ICEBERG_EXPORT InMemoryCatalog
 
   Result<std::unique_ptr<Table>> CreateTable(
       const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
-      const std::string& location,
+      const SortOrder& order, const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) override;
 
   Result<std::unique_ptr<Table>> UpdateTable(
@@ -82,7 +82,7 @@ class ICEBERG_EXPORT InMemoryCatalog
 
   Result<std::shared_ptr<Transaction>> StageCreateTable(
       const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
-      const std::string& location,
+      const SortOrder& order, const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) override;
 
   Result<bool> TableExists(const TableIdentifier& identifier) const 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 (auto spec_json_result = GetJsonValue<nlohmann::json>(json, kPartitionSpec);
+      spec_json_result.has_value()) {
+    ICEBERG_ASSIGN_OR_RAISE(
+        request.partition_spec,
+        PartitionSpecFromJson(request.schema, spec_json_result.value(),
+                              PartitionSpec::kInitialSpecId));
+  }
+
+  if (auto order_json_result = GetJsonValue<nlohmann::json>(json, kWriteOrder);
+      order_json_result.has_value()) {
+    ICEBERG_ASSIGN_OR_RAISE(request.write_order,
+                            SortOrderFromJson(order_json_result.value(), 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

@@ -77,8 +77,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,18 +103,21 @@ 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(paths), std::move(endpoints),
+                      std::move(file_io)));
 }
 
 RestCatalog::RestCatalog(std::unique_ptr<RestCatalogProperties> config,
                          std::unique_ptr<ResourcePaths> paths,
-                         std::unordered_set<Endpoint> endpoints)
+                         std::unordered_set<Endpoint> endpoints,
+                         std::shared_ptr<FileIO> file_io)
     : config_(std::move(config)),
       client_(std::make_unique<HttpClient>(config_->ExtractHeaders())),
       paths_(std::move(paths)),
       name_(config_->Get(RestCatalogProperties::kName)),
-      supported_endpoints_(std::move(endpoints)) {}
+      supported_endpoints_(std::move(endpoints)),
+      file_io_(std::move(file_io)) {}
 
 std::string_view RestCatalog::name() const { return name_; }
 
@@ -241,11 +244,50 @@ Result<std::vector<TableIdentifier>> RestCatalog::ListTables(
 }
 
 Result<std::unique_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 Schema& schema, const PartitionSpec& spec,
+    const 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));
+
+  // Schema, partition spec, and sort order are not copyable due to Lazy<> members,
+  // recreate them
+  auto new_schema = std::make_shared<Schema>(
+      std::vector<SchemaField>(schema.fields().begin(), schema.fields().end()),
+      schema.schema_id());
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      std::shared_ptr<PartitionSpec> partition_spec,
+      PartitionSpec::Make(
+          spec.spec_id(),
+          std::vector<PartitionField>(spec.fields().begin(), spec.fields().end()),
+          spec.last_assigned_field_id()));
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      std::shared_ptr<SortOrder> sort_order,
+      SortOrder::Make(order.order_id(), std::vector<SortField>(order.fields().begin(),
+                                                               order.fields().end())));
+
+  CreateTableRequest request{
+      .name = identifier.name,
+      .location = location,
+      .schema = new_schema,
+      .partition_spec = partition_spec,
+      .write_order = sort_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 std::make_unique<Table>(identifier, std::move(load_result.metadata),
+                                 std::move(load_result.metadata_location), file_io_,
+                                 std::static_pointer_cast<Catalog>(shared_from_this()));
 }
 
 Result<std::unique_ptr<Table>> RestCatalog::UpdateTable(
@@ -258,7 +300,7 @@ Result<std::unique_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::string& location,
+    [[maybe_unused]] const 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;
 
@@ -73,7 +76,7 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog {
 
   Result<std::unique_ptr<Table>> CreateTable(
       const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
-      const std::string& location,
+      const SortOrder& order, const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) override;
 
   Result<std::unique_ptr<Table>> UpdateTable(
@@ -83,7 +86,7 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog {
 
   Result<std::shared_ptr<Transaction>> StageCreateTable(
       const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
-      const std::string& location,
+      const SortOrder& order, const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) override;
 
   Result<bool> TableExists(const TableIdentifier& identifier) const override;
@@ -101,13 +104,14 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog {
  private:
   RestCatalog(std::unique_ptr<RestCatalogProperties> config,
               std::unique_ptr<ResourcePaths> paths,
-              std::unordered_set<Endpoint> endpoints);
+              std::unordered_set<Endpoint> endpoints, std::shared_ptr<FileIO> file_io);
 
   std::unique_ptr<RestCatalogProperties> config_;
   std::unique_ptr<HttpClient> client_;
   std::unique_ptr<ResourcePaths> paths_;
   std::string name_;
   std::unordered_set<Endpoint> supported_endpoints_;
+  std::shared_ptr<FileIO> file_io_;
 };
 
 }  // namespace iceberg::rest

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"
@@ -126,6 +127,28 @@ struct ICEBERG_REST_EXPORT RenameTableRequest {
   }
 };
 
+/// \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/table.cc

@@ -120,6 +120,8 @@ std::unique_ptr<Transaction> Table::NewTransaction() const {
 
 const std::shared_ptr<FileIO>& Table::io() const { return io_; }
 
+const std::shared_ptr<TableMetadata>& Table::metadata() const { return metadata_; }
+
 std::unique_ptr<TableScanBuilder> Table::NewScan() const {
   return std::make_unique<TableScanBuilder>(metadata_, io_);
 }