feat(rest): implement stage-create table (#485)

Author: HeartLinked

src/iceberg/catalog/rest/rest_catalog.cc

@@ -44,6 +44,7 @@
 #include "iceberg/table.h"
 #include "iceberg/table_requirement.h"
 #include "iceberg/table_update.h"
+#include "iceberg/transaction.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg::rest {
@@ -274,11 +275,11 @@ Result<std::vector<TableIdentifier>> RestCatalog::ListTables(const Namespace& ns
   return result;
 }
 
-Result<std::shared_ptr<Table>> RestCatalog::CreateTable(
+Result<LoadTableResult> RestCatalog::CreateTableInternal(
     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) {
+    const std::unordered_map<std::string, std::string>& properties, bool stage_create) {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::CreateTable());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Tables(identifier.ns));
 
@@ -288,7 +289,7 @@ Result<std::shared_ptr<Table>> RestCatalog::CreateTable(
       .schema = schema,
       .partition_spec = spec,
       .write_order = order,
-      .stage_create = false,
+      .stage_create = stage_create,
       .properties = properties,
   };
 
@@ -298,10 +299,19 @@ Result<std::shared_ptr<Table>> RestCatalog::CreateTable(
       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());
+  return LoadTableResultFromJson(json);
+}
+
+Result<std::shared_ptr<Table>> RestCatalog::CreateTable(
+    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_ASSIGN_OR_RAISE(auto result,
+                          CreateTableInternal(identifier, schema, spec, order, location,
+                                              properties, /*stage_create=*/false));
+  return Table::Make(identifier, std::move(result.metadata),
+                     std::move(result.metadata_location), file_io_, shared_from_this());
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
@@ -335,13 +345,19 @@ Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
 }
 
 Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
-    [[maybe_unused]] const TableIdentifier& identifier,
-    [[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");
+    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_ASSIGN_OR_RAISE(auto result,
+                          CreateTableInternal(identifier, schema, spec, order, location,
+                                              properties, /*stage_create=*/true));
+  ICEBERG_ASSIGN_OR_RAISE(auto staged_table,
+                          StagedTable::Make(identifier, std::move(result.metadata),
+                                            std::move(result.metadata_location), file_io_,
+                                            shared_from_this()));
+  return Transaction::Make(std::move(staged_table), Transaction::Kind::kCreate,
+                           /*auto_commit=*/false);
 }
 
 Status RestCatalog::DropTable(const TableIdentifier& identifier, bool purge) {
@@ -393,9 +409,6 @@ Result<std::string> RestCatalog::LoadTableInternal(
 }
 
 Result<std::shared_ptr<Table>> RestCatalog::LoadTable(const TableIdentifier& identifier) {
-  ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::LoadTable());
-  ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Table(identifier));
-
   ICEBERG_ASSIGN_OR_RAISE(const auto body, LoadTableInternal(identifier));
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(body));
   ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));

src/iceberg/catalog/rest/rest_catalog.h

@@ -110,6 +110,12 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
 
   Result<std::string> LoadTableInternal(const TableIdentifier& identifier) const;
 
+  Result<LoadTableResult> CreateTableInternal(
+      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, bool stage_create);
+
   std::unique_ptr<RestCatalogProperties> config_;
   std::shared_ptr<FileIO> file_io_;
   std::unique_ptr<HttpClient> client_;

src/iceberg/catalog/rest/type_fwd.h

@@ -25,6 +25,7 @@
 namespace iceberg::rest {
 
 struct ErrorResponse;
+struct LoadTableResult;
 
 class Endpoint;
 class ErrorHandler;

src/iceberg/test/rest_catalog_test.cc

@@ -52,6 +52,7 @@
 #include "iceberg/test/std_io.h"
 #include "iceberg/test/test_resource.h"
 #include "iceberg/test/util/docker_compose_util.h"
+#include "iceberg/transaction.h"
 
 namespace iceberg::rest {
 
@@ -639,4 +640,51 @@ TEST_F(RestCatalogIntegrationTest, RegisterTable) {
   EXPECT_NE(table->name(), registered_table->name());
 }
 
+TEST_F(RestCatalogIntegrationTest, StageCreateTable) {
+  auto catalog_result = CreateCatalog();
+  ASSERT_THAT(catalog_result, IsOk());
+  auto& catalog = catalog_result.value();
+
+  // Create namespace
+  Namespace ns{.levels = {"test_stage_create"}};
+  auto status = catalog->CreateNamespace(ns, {});
+  EXPECT_THAT(status, IsOk());
+
+  // Stage create table
+  auto schema = CreateDefaultSchema();
+  auto partition_spec = PartitionSpec::Unpartitioned();
+  auto sort_order = SortOrder::Unsorted();
+
+  TableIdentifier table_id{.ns = ns, .name = "staged_table"};
+  std::unordered_map<std::string, std::string> table_properties{{"key1", "value1"}};
+  auto txn_result = catalog->StageCreateTable(table_id, schema, partition_spec,
+                                              sort_order, "", table_properties);
+  ASSERT_THAT(txn_result, IsOk());
+  auto& txn = txn_result.value();
+
+  // Verify the staged table in transaction
+  EXPECT_NE(txn->table(), nullptr);
+  EXPECT_EQ(txn->table()->name(), table_id);
+
+  // Table should NOT exist in catalog yet (staged but not committed)
+  auto exists_result = catalog->TableExists(table_id);
+  ASSERT_THAT(exists_result, IsOk());
+  EXPECT_FALSE(exists_result.value());
+
+  // Commit the transaction
+  auto commit_result = txn->Commit();
+  ASSERT_THAT(commit_result, IsOk());
+  auto& committed_table = commit_result.value();
+
+  // Verify table now exists
+  exists_result = catalog->TableExists(table_id);
+  ASSERT_THAT(exists_result, IsOk());
+  EXPECT_TRUE(exists_result.value());
+
+  // Verify table properties
+  EXPECT_EQ(committed_table->name(), table_id);
+  auto& props = committed_table->metadata()->properties.configs();
+  EXPECT_EQ(props.at("key1"), "value1");
+}
+
 }  // namespace iceberg::rest