feat: implement create (stage) table for InMemoryCatalog (#416)

Author: WZhuo

src/iceberg/CMakeLists.txt

@@ -66,6 +66,7 @@ set(ICEBERG_SOURCES
     sort_order.cc
     statistics_file.cc
     table.cc
+    table_identifier.cc
     table_metadata.cc
     table_properties.cc
     table_requirement.cc
@@ -86,6 +87,7 @@ set(ICEBERG_SOURCES
     util/decimal.cc
     util/gzip_internal.cc
     util/murmurhash3_internal.cc
+    util/property_util.cc
     util/snapshot_util.cc
     util/temporal_util.cc
     util/timepoint.cc

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -26,7 +26,10 @@
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_metadata.h"
 #include "iceberg/table_requirement.h"
+#include "iceberg/table_requirements.h"
 #include "iceberg/table_update.h"
+#include "iceberg/transaction.h"
+#include "iceberg/util/checked_cast.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg {
@@ -318,7 +321,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;
 }
@@ -405,32 +408,68 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::CreateTable(
     const std::string& location,
     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 table_metadata, TableMetadata::Make(*schema, *spec, *order,
+                                                                   location, properties));
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto metadata_file_location,
+      TableMetadataUtil::Write(*file_io_, nullptr, "", *table_metadata));
+  ICEBERG_RETURN_UNEXPECTED(
+      root_namespace_->UpdateTableMetadataLocation(identifier, metadata_file_location));
+  return Table::Make(identifier, std::move(table_metadata),
+                     std::move(metadata_file_location), file_io_, shared_from_this());
 }
 
 Result<std::shared_ptr<Table>> InMemoryCatalog::UpdateTable(
     const TableIdentifier& identifier,
     const std::vector<std::unique_ptr<TableRequirement>>& requirements,
     const std::vector<std::unique_ptr<TableUpdate>>& updates) {
   std::unique_lock lock(mutex_);
-  ICEBERG_ASSIGN_OR_RAISE(auto base_metadata_location,
-                          root_namespace_->GetTableMetadataLocation(identifier));
-
-  ICEBERG_ASSIGN_OR_RAISE(auto base,
-                          TableMetadataUtil::Read(*file_io_, base_metadata_location));
+  auto base_metadata_location = root_namespace_->GetTableMetadataLocation(identifier);
+  std::unique_ptr<TableMetadata> base;
+  std::unique_ptr<TableMetadataBuilder> builder;
+  ICEBERG_ASSIGN_OR_RAISE(auto is_create, TableRequirements::IsCreate(requirements));
+  if (is_create) {
+    if (base_metadata_location.has_value()) {
+      return AlreadyExists("Table already exists: {}", identifier);
+    }
+    int8_t format_version = TableMetadata::kDefaultTableFormatVersion;
+    for (const auto& update : updates) {
+      if (update->kind() == TableUpdate::Kind::kUpgradeFormatVersion) {
+        format_version =
+            internal::checked_cast<const table::UpgradeFormatVersion&>(*update)
+                .format_version();
+      }
+    }
+    builder = TableMetadataBuilder::BuildFromEmpty(format_version);
+  } else {
+    ICEBERG_RETURN_UNEXPECTED(base_metadata_location);
+    ICEBERG_ASSIGN_OR_RAISE(
+        base, TableMetadataUtil::Read(*file_io_, base_metadata_location.value()));
+    builder = TableMetadataBuilder::BuildFrom(base.get());
+  }
 
   for (const auto& requirement : requirements) {
     ICEBERG_RETURN_UNEXPECTED(requirement->Validate(base.get()));
   }
 
-  auto builder = TableMetadataBuilder::BuildFrom(base.get());
   for (const auto& update : updates) {
     update->ApplyTo(*builder);
   }
   ICEBERG_ASSIGN_OR_RAISE(auto updated, builder->Build());
   ICEBERG_ASSIGN_OR_RAISE(
       auto new_metadata_location,
-      TableMetadataUtil::Write(*file_io_, base.get(), base_metadata_location, *updated));
+      TableMetadataUtil::Write(
+          *file_io_, base.get(),
+          base_metadata_location.has_value() ? base_metadata_location.value() : "",
+          *updated));
   ICEBERG_RETURN_UNEXPECTED(
       root_namespace_->UpdateTableMetadataLocation(identifier, new_metadata_location));
   TableMetadataUtil::DeleteRemovedMetadataFiles(*file_io_, base.get(), *updated);
@@ -445,7 +484,21 @@ Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
     const std::string& location,
     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 table_metadata,
+      TableMetadata::Make(*schema, *spec, *order, base_location, properties));
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto table, StagedTable::Make(identifier, std::move(table_metadata), "", file_io_,
+                                    shared_from_this()));
+  return Transaction::Make(std::move(table), Transaction::Kind::kCreate,
+                           /* auto_commit */ false);
 }
 
 Result<bool> InMemoryCatalog::TableExists(const TableIdentifier& identifier) const {
@@ -495,7 +548,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/meson.build

@@ -88,6 +88,7 @@ iceberg_sources = files(
     'sort_order.cc',
     'statistics_file.cc',
     'table.cc',
+    'table_identifier.cc',
     'table_metadata.cc',
     'table_properties.cc',
     'table_requirement.cc',
@@ -108,6 +109,7 @@ iceberg_sources = files(
     'util/decimal.cc',
     'util/gzip_internal.cc',
     'util/murmurhash3_internal.cc',
+    'util/property_util.cc',
     'util/snapshot_util.cc',
     'util/temporal_util.cc',
     'util/timepoint.cc',

src/iceberg/schema.cc

@@ -235,7 +235,7 @@ Result<std::vector<std::string>> Schema::IdentifierFieldNames() const {
   for (auto id : identifier_field_ids_) {
     ICEBERG_ASSIGN_OR_RAISE(auto name, FindColumnNameById(id));
     if (!name.has_value()) {
-      return InvalidSchema("Cannot find the field of the specified field id: {}", id);
+      return InvalidSchema("Cannot find identifier field id: {}", id);
     }
     names.emplace_back(name.value());
   }

src/iceberg/table_identifier.cc

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/table_identifier.h"
+
+#include "iceberg/util/formatter_internal.h"
+
+namespace iceberg {
+
+std::string Namespace::ToString() const { return FormatRange(levels, ".", "", ""); }
+
+std::string TableIdentifier::ToString() const {
+  if (!ns.levels.empty()) {
+    return std::format("{}.{}", ns.ToString(), name);
+  } else {
+    return name;
+  }
+}
+
+}  // namespace iceberg

src/iceberg/table_identifier.h

@@ -27,6 +27,7 @@
 
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
+#include "iceberg/util/formatter.h"  // IWYU pragma: keep
 
 namespace iceberg {
 
@@ -35,8 +36,12 @@ struct ICEBERG_EXPORT Namespace {
   std::vector<std::string> levels;
 
   bool operator==(const Namespace& other) const { return levels == other.levels; }
+
+  std::string ToString() const;
 };
 
+ICEBERG_EXPORT inline std::string ToString(const Namespace& ns) { return ns.ToString(); }
+
 /// \brief Identifies a table in iceberg catalog.
 struct ICEBERG_EXPORT TableIdentifier {
   Namespace ns;
@@ -53,6 +58,12 @@ struct ICEBERG_EXPORT TableIdentifier {
     }
     return {};
   }
+
+  std::string ToString() const;
 };
 
+ICEBERG_EXPORT inline std::string ToString(const TableIdentifier& ident) {
+  return ident.ToString();
+}
+
 }  // namespace iceberg