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/schema.cc

@@ -33,8 +33,11 @@
 
 namespace iceberg {
 
-Schema::Schema(std::vector<SchemaField> fields, std::optional<int32_t> schema_id)
-    : StructType(std::move(fields)), schema_id_(schema_id) {}
+Schema::Schema(std::vector<SchemaField> fields, std::optional<int32_t> schema_id,
+               std::vector<int32_t> identifier_field_ids)
+    : StructType(std::move(fields)),
+      schema_id_(schema_id),
+      identifier_field_ids_(std::move(identifier_field_ids)) {}
 
 std::optional<int32_t> Schema::schema_id() const { return schema_id_; }
 
@@ -179,4 +182,20 @@ Result<std::unique_ptr<Schema>> Schema::Project(
                         std::nullopt);
 }
 
+std::vector<int32_t> Schema::IdentifierFieldIds() const { return identifier_field_ids_; }
+
+Result<std::vector<std::string>> Schema::IdentifierFieldNames() const {
+  using std::ranges::to;
+  using std::views::transform;
+  std::vector<std::string> names;
+  for (auto id : identifier_field_ids_) {
+    ICEBERG_ASSIGN_OR_RAISE(auto field, FindFieldById(id));
+    if (!field.has_value()) {
+      return InvalidSchema("Can not find field to the specified id: {}", id);
+    }
+    names.emplace_back(field.value().get().name());
+  }
+  return names;
+}
+
 }  // namespace iceberg

src/iceberg/schema.h

@@ -49,7 +49,8 @@ class ICEBERG_EXPORT Schema : public StructType {
   static constexpr int32_t kInvalidColumnId = -1;
 
   explicit Schema(std::vector<SchemaField> fields,
-                  std::optional<int32_t> schema_id = std::nullopt);
+                  std::optional<int32_t> schema_id = std::nullopt,
+                  std::vector<int32_t> identifier_field_ids = {});
 
   /// \brief Get the schema ID.
   ///
@@ -100,6 +101,9 @@ class ICEBERG_EXPORT Schema : public StructType {
   Result<std::unique_ptr<Schema>> Project(
       const std::unordered_set<int32_t>& field_ids) const;
 
+  std::vector<int32_t> IdentifierFieldIds() const;
+  Result<std::vector<std::string>> IdentifierFieldNames() const;
+
   friend bool operator==(const Schema& lhs, const Schema& rhs) { return lhs.Equals(rhs); }
 
  private:
@@ -116,6 +120,8 @@ class ICEBERG_EXPORT Schema : public StructType {
       const Schema&);
 
   const std::optional<int32_t> schema_id_;
+  std::vector<int32_t> identifier_field_ids_;
+
   /// Mapping from field id to field.
   Lazy<InitIdToFieldMap> id_to_field_;
   /// Mapping from field name to field id.

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