feat: InMemoryCatalog create table

Author: WZhuo

src/iceberg/catalog.h

@@ -105,14 +105,15 @@ class ICEBERG_EXPORT Catalog {
   /// \brief Create a table
   ///
   /// \param identifier a table identifier
+  /// \param location a location for the table; leave empty if unspecified
   /// \param schema a schema
   /// \param spec a partition spec
-  /// \param location a location for the table; leave empty if unspecified
+  /// \param sort_order a sort order
   /// \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 TableIdentifier& identifier, const std::string& location,
+      const Schema& schema, const PartitionSpec& spec, const SortOrder& sort_order,
       const std::unordered_map<std::string, std::string>& properties) = 0;
 
   /// \brief Update a table
@@ -129,15 +130,16 @@ class ICEBERG_EXPORT Catalog {
   /// \brief Start a transaction to create a table
   ///
   /// \param identifier a table identifier
+  /// \param location a location for the table; leave empty if unspecified
   /// \param schema a schema
   /// \param spec a partition spec
-  /// \param location a location for the table; leave empty if unspecified
+  /// \param sort_order a sort order
   /// \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 TableIdentifier& identifier, const std::string& location,
+      const Schema& schema, const PartitionSpec& spec, const SortOrder& sort_order,
       const std::unordered_map<std::string, std::string>& properties) = 0;
 
   /// \brief Check whether table exists

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -22,6 +22,7 @@
 #include <algorithm>
 #include <iterator>
 
+#include "iceberg/sort_order.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_metadata.h"
@@ -318,7 +319,7 @@ Result<std::string> InMemoryNamespace::GetTableMetadataLocation(
   ICEBERG_RETURN_UNEXPECTED(ns);
   const auto it = ns.value()->table_metadata_locations_.find(table_ident.name);
   if (it == ns.value()->table_metadata_locations_.end()) {
-    return NotFound("{} does not exist", table_ident.name);
+    return NotFound("Table does not exist: {}", table_ident);
   }
   return it->second;
 }
@@ -400,11 +401,25 @@ 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 TableIdentifier& identifier, const std::string& location, const Schema& schema,
+    const PartitionSpec& spec, const SortOrder& sort_order,
     const std::unordered_map<std::string, std::string>& properties) {
   std::unique_lock lock(mutex_);
-  return NotImplemented("create table");
+  if (root_namespace_->TableExists(identifier).value_or(false)) {
+    return AlreadyExists("Table already exists: {}", identifier);
+  }
+
+  std::string base_location =
+      location.empty() ? warehouse_location_ + "/" + identifier.ToString() : location;
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata, TableMetadata::Make(base_location, schema, spec,
+                                                             sort_order, properties));
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata_file_location,
+                          TableMetadataUtil::Write(*file_io_, nullptr, "", *metadata));
+  ICEBERG_RETURN_UNEXPECTED(
+      root_namespace_->UpdateTableMetadataLocation(identifier, metadata_file_location));
+  return std::make_unique<Table>(identifier, std::move(metadata),
+                                 std::move(metadata_file_location), file_io_,
+                                 std::static_pointer_cast<Catalog>(shared_from_this()));
 }
 
 Result<std::unique_ptr<Table>> InMemoryCatalog::UpdateTable(
@@ -440,11 +455,20 @@ 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 TableIdentifier& identifier, const std::string& location, const Schema& schema,
+    const PartitionSpec& spec, const SortOrder& sort_order,
     const std::unordered_map<std::string, std::string>& properties) {
   std::unique_lock lock(mutex_);
-  return NotImplemented("stage create table");
+  if (root_namespace_->TableExists(identifier).value_or(false)) {
+    return AlreadyExists("Table already exists: {}", identifier);
+  }
+
+  std::string base_location =
+      location.empty() ? warehouse_location_ + "/" + identifier.ToString() : location;
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata, TableMetadata::Make(base_location, schema, spec,
+                                                             sort_order, properties));
+  // TODO(zhuo.wang) Init transaction with metadata
+  return nullptr;
 }
 
 Result<bool> InMemoryCatalog::TableExists(const TableIdentifier& identifier) const {
@@ -495,7 +519,7 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::RegisterTable(
 
   std::unique_lock lock(mutex_);
   if (!root_namespace_->NamespaceExists(identifier.ns)) {
-    return NoSuchNamespace("table namespace does not exist.");
+    return NoSuchNamespace("Table namespace does not exist: {}", identifier.ns);
   }
   if (!root_namespace_->RegisterTable(identifier, metadata_file_location)) {
     return UnknownError("The registry failed.");

src/iceberg/catalog/memory/in_memory_catalog.h

@@ -71,8 +71,8 @@ class ICEBERG_EXPORT InMemoryCatalog
   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 TableIdentifier& identifier, const std::string& location,
+      const Schema& schema, const PartitionSpec& spec, const SortOrder& sort_order,
       const std::unordered_map<std::string, std::string>& properties) override;
 
   Result<std::unique_ptr<Table>> UpdateTable(
@@ -81,8 +81,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 std::string& location,
+      const TableIdentifier& identifier, const std::string& location,
+      const Schema& schema, const PartitionSpec& spec, const SortOrder& sort_order,
       const std::unordered_map<std::string, std::string>& properties) override;
 
   Result<bool> TableExists(const TableIdentifier& identifier) const override;

src/iceberg/catalog/rest/rest_catalog.cc

@@ -242,8 +242,9 @@ 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::string& location, [[maybe_unused]] const Schema& schema,
+    [[maybe_unused]] const PartitionSpec& spec,
+    [[maybe_unused]] const SortOrder& sort_order,
     [[maybe_unused]] const std::unordered_map<std::string, std::string>& properties) {
   return NotImplemented("Not implemented");
 }
@@ -257,8 +258,9 @@ 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 std::string& location, [[maybe_unused]] const Schema& schema,
+    [[maybe_unused]] const PartitionSpec& spec,
+    [[maybe_unused]] const SortOrder& sort_order,
     [[maybe_unused]] const std::unordered_map<std::string, std::string>& properties) {
   return NotImplemented("Not implemented");
 }

src/iceberg/catalog/rest/rest_catalog.h

@@ -20,7 +20,6 @@
 #pragma once
 
 #include <memory>
-#include <set>
 #include <string>
 
 #include "iceberg/catalog.h"
@@ -72,8 +71,8 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog {
   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 TableIdentifier& identifier, const std::string& location,
+      const Schema& schema, const PartitionSpec& spec, const SortOrder& sort_order,
       const std::unordered_map<std::string, std::string>& properties) override;
 
   Result<std::unique_ptr<Table>> UpdateTable(
@@ -82,8 +81,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 std::string& location,
+      const TableIdentifier& identifier, const std::string& location,
+      const Schema& schema, const PartitionSpec& spec, const SortOrder& sort_order,
       const std::unordered_map<std::string, std::string>& properties) override;
 
   Result<bool> TableExists(const TableIdentifier& identifier) const override;

src/iceberg/table_identifier.h

@@ -22,6 +22,8 @@
 /// \file iceberg/table_identifier.h
 /// A TableIdentifier is a unique identifier for a table
 
+#include <format>
+#include <sstream>
 #include <string>
 #include <vector>
 
@@ -35,6 +37,15 @@ struct ICEBERG_EXPORT Namespace {
   std::vector<std::string> levels;
 
   bool operator==(const Namespace& other) const { return levels == other.levels; }
+
+  std::string ToString() const {
+    std::ostringstream oss;
+    for (size_t i = 0; i < levels.size(); ++i) {
+      if (i) oss << '.';
+      oss << levels[i];
+    }
+    return oss.str();
+  }
 };
 
 /// \brief Identifies a table in iceberg catalog.
@@ -53,6 +64,27 @@ struct ICEBERG_EXPORT TableIdentifier {
     }
     return {};
   }
+
+  std::string ToString() const { return ns.ToString() + '.' + name; }
 };
 
 }  // namespace iceberg
+
+namespace std {
+
+template <>
+struct formatter<iceberg::Namespace> : std::formatter<std::string> {
+  constexpr auto parse(format_parse_context& ctx) { return ctx.begin(); }
+  auto format(const iceberg::Namespace& ns, format_context& ctx) const {
+    return std::formatter<std::string>::format(ns.ToString(), ctx);
+  }
+};
+
+template <>
+struct formatter<iceberg::TableIdentifier> : std::formatter<std::string> {
+  constexpr auto parse(format_parse_context& ctx) { return ctx.begin(); }
+  auto format(const iceberg::TableIdentifier& id, format_context& ctx) const {
+    return std::formatter<std::string>::format(id.ToString(), ctx);
+  }
+};
+}  // namespace std

src/iceberg/table_metadata.cc

@@ -36,16 +36,19 @@
 #include "iceberg/exception.h"
 #include "iceberg/file_io.h"
 #include "iceberg/json_internal.h"
+#include "iceberg/metrics_config.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
 #include "iceberg/snapshot.h"
 #include "iceberg/sort_order.h"
 #include "iceberg/table_properties.h"
 #include "iceberg/table_update.h"
+#include "iceberg/util/error_collector.h"
 #include "iceberg/util/gzip_internal.h"
 #include "iceberg/util/location_util.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/type_util.h"
 #include "iceberg/util/uuid.h"
 namespace iceberg {
 namespace {
@@ -64,6 +67,47 @@ std::string ToString(const MetadataLogEntry& entry) {
                      entry.metadata_file);
 }
 
+Result<std::unique_ptr<TableMetadata>> TableMetadata::Make(
+    const std::string& location, const iceberg::Schema& schema,
+    const iceberg::PartitionSpec& spec, const iceberg::SortOrder& sort_order,
+    const std::unordered_map<std::string, std::string>& properties, int format_version) {
+  for (const auto& [key, _] : properties) {
+    if (TableProperties::reserved_properties().contains(key)) {
+      return InvalidArgument(
+          "Table properties should not contain reserved properties, but got {}", key);
+    }
+  }
+
+  int last_column_id = 0;
+  auto next_id = [&last_column_id]() -> int32_t { return ++last_column_id; };
+
+  auto fresh_schema =
+      IdAssigner::AssignFreshIds(Schema::kInitialSchemaId, schema, next_id);
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto fresh_spec,
+      PartitionSpec::Make(
+          *fresh_schema, PartitionSpec::kInitialSpecId,
+          std::vector<PartitionField>(spec.fields().begin(), spec.fields().end()),
+          false));
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto fresh_order,
+      SortOrder::Make(sort_order.is_unsorted() ? sort_order.order_id()
+                                               : SortOrder::kInitialSortOrderId,
+                      std::vector<SortField>(sort_order.fields().begin(),
+                                             sort_order.fields().end())));
+  ICEBERG_RETURN_UNEXPECTED(
+      MetricsConfig::VerifyReferencedColumns(properties, *fresh_schema));
+
+  return TableMetadataBuilder::BuildFromEmpty(format_version)
+      ->SetLocation(location)
+      .SetCurrentSchema(fresh_schema, next_id())
+      .AddPartitionSpec(std::move(fresh_spec))
+      .AddSortOrder(std::move(fresh_order))
+      .SetProperties(properties)
+      .Build();
+}
+
 Result<std::shared_ptr<Schema>> TableMetadata::Schema() const {
   return SchemaById(current_schema_id);
 }
@@ -417,6 +461,33 @@ struct TableMetadataBuilder::Impl {
 
     metadata.last_updated_ms = kInvalidLastUpdatedMs;
   }
+
+  // TODO(zhuo.wang) Do validation
+  Status AddSchema(const std::shared_ptr<Schema>& schema, int new_last_column_id) {
+    if (new_last_column_id < metadata.last_column_id) {
+      return InvalidArgument("Invalid last column ID: {} < {} (previous last column ID)",
+                             new_last_column_id, metadata.last_column_id);
+    }
+    metadata.schemas.push_back(schema);
+    schemas_by_id.emplace(schema->schema_id().value(), schema);
+    metadata.current_schema_id = schema->schema_id().value();
+
+    return {};
+  }
+
+  // TODO(zhuo.wang) Do validation
+  Status AddPartitionSpec(const std::shared_ptr<PartitionSpec>& spec) {
+    metadata.partition_specs.push_back(spec);
+    specs_by_id.emplace(spec->spec_id(), spec);
+    return {};
+  }
+
+  // TODO(zhuo.wang) Do validation
+  Status AddSortOrder(const std::shared_ptr<SortOrder>& order) {
+    metadata.sort_orders.push_back(order);
+    sort_orders_by_id.emplace(order->order_id(), order);
+    return {};
+  }
 };
 
 TableMetadataBuilder::TableMetadataBuilder(int8_t format_version)
@@ -529,7 +600,7 @@ TableMetadataBuilder& TableMetadataBuilder::UpgradeFormatVersion(
 
 TableMetadataBuilder& TableMetadataBuilder::SetCurrentSchema(
     std::shared_ptr<Schema> schema, int32_t new_last_column_id) {
-  throw IcebergError(std::format("{} not implemented", __FUNCTION__));
+  BUILDER_RETURN(impl_->AddSchema(schema, new_last_column_id));
 }
 
 TableMetadataBuilder& TableMetadataBuilder::SetCurrentSchema(int32_t schema_id) {
@@ -551,7 +622,7 @@ TableMetadataBuilder& TableMetadataBuilder::SetDefaultPartitionSpec(int32_t spec
 
 TableMetadataBuilder& TableMetadataBuilder::AddPartitionSpec(
     std::shared_ptr<PartitionSpec> spec) {
-  throw IcebergError(std::format("{} not implemented", __FUNCTION__));
+  BUILDER_RETURN(impl_->AddPartitionSpec(spec));
 }
 
 TableMetadataBuilder& TableMetadataBuilder::RemovePartitionSpecs(
@@ -746,7 +817,8 @@ TableMetadataBuilder& TableMetadataBuilder::RemoveProperties(
 }
 
 TableMetadataBuilder& TableMetadataBuilder::SetLocation(std::string_view location) {
-  throw IcebergError(std::format("{} not implemented", __FUNCTION__));
+  impl_->metadata.location = location;
+  return *this;
 }
 
 TableMetadataBuilder& TableMetadataBuilder::AddEncryptionKey(

src/iceberg/table_metadata.h

@@ -124,6 +124,12 @@ struct ICEBERG_EXPORT TableMetadata {
   /// A `long` higher than all assigned row IDs
   int64_t next_row_id;
 
+  static Result<std::unique_ptr<TableMetadata>> Make(
+      const std::string& location, const iceberg::Schema& schema,
+      const iceberg::PartitionSpec& spec, const iceberg::SortOrder& sort_order,
+      const std::unordered_map<std::string, std::string>& properties,
+      int format_version = kDefaultTableFormatVersion);
+
   /// \brief Get the current schema, return NotFoundError if not found
   Result<std::shared_ptr<iceberg::Schema>> Schema() const;
   /// \brief Get the current schema by ID, return NotFoundError if not found