feat: Impl InMemoryCatalog's UpdateTable

Author: WZhuo

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -21,10 +21,13 @@
 
 #include <algorithm>
 #include <iterator>  // IWYU pragma: keep
+#include <memory>
 
-#include "iceberg/exception.h"
 #include "iceberg/table.h"
+#include "iceberg/table_identifier.h"
 #include "iceberg/table_metadata.h"
+#include "iceberg/table_requirement.h"
+#include "iceberg/table_update.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg {
@@ -121,6 +124,13 @@ class ICEBERG_EXPORT InMemoryNamespace {
   /// \return The metadata location if the table exists; error otherwise.
   Result<std::string> GetTableMetadataLocation(const TableIdentifier& table_ident) const;
 
+  /// \brief Updates the metadata location for the specified table.
+  ///
+  /// \param table_ident The identifier of the table.
+  /// \param metadata_location The new metadata location.
+  Status UpdateTableMetadataLocation(const TableIdentifier& table_ident,
+                                     const std::string& metadata_location);
+
   /// \brief Internal utility for retrieving a namespace node pointer from the tree.
   ///
   /// \tparam NamespacePtr The type of the namespace node pointer.
@@ -279,7 +289,7 @@ Result<std::vector<std::string>> InMemoryNamespace::ListTables(
   return table_names;
 }
 
-Status InMemoryNamespace::RegisterTable(TableIdentifier const& table_ident,
+Status InMemoryNamespace::RegisterTable(const TableIdentifier& table_ident,
                                         const std::string& metadata_location) {
   const auto ns = GetNamespace(this, table_ident.ns);
   ICEBERG_RETURN_UNEXPECTED(ns);
@@ -290,21 +300,21 @@ Status InMemoryNamespace::RegisterTable(TableIdentifier const& table_ident,
   return {};
 }
 
-Status InMemoryNamespace::UnregisterTable(TableIdentifier const& table_ident) {
+Status InMemoryNamespace::UnregisterTable(const TableIdentifier& table_ident) {
   const auto ns = GetNamespace(this, table_ident.ns);
   ICEBERG_RETURN_UNEXPECTED(ns);
   ns.value()->table_metadata_locations_.erase(table_ident.name);
   return {};
 }
 
-Result<bool> InMemoryNamespace::TableExists(TableIdentifier const& table_ident) const {
+Result<bool> InMemoryNamespace::TableExists(const TableIdentifier& table_ident) const {
   const auto ns = GetNamespace(this, table_ident.ns);
   ICEBERG_RETURN_UNEXPECTED(ns);
   return ns.value()->table_metadata_locations_.contains(table_ident.name);
 }
 
 Result<std::string> InMemoryNamespace::GetTableMetadataLocation(
-    TableIdentifier const& table_ident) const {
+    const TableIdentifier& table_ident) const {
   const auto ns = GetNamespace(this, table_ident.ns);
   ICEBERG_RETURN_UNEXPECTED(ns);
   const auto it = ns.value()->table_metadata_locations_.find(table_ident.name);
@@ -314,17 +324,24 @@ Result<std::string> InMemoryNamespace::GetTableMetadataLocation(
   return it->second;
 }
 
+Status InMemoryNamespace::UpdateTableMetadataLocation(
+    const TableIdentifier& table_ident, const std::string& metadata_location) {
+  ICEBERG_ASSIGN_OR_RAISE(auto ns, GetNamespace(this, table_ident.ns));
+  ns->table_metadata_locations_[table_ident.name] = metadata_location;
+  return {};
+}
+
 std::shared_ptr<InMemoryCatalog> InMemoryCatalog::Make(
-    std::string const& name, std::shared_ptr<FileIO> const& file_io,
-    std::string const& warehouse_location,
-    std::unordered_map<std::string, std::string> const& properties) {
+    const std::string& name, const std::shared_ptr<FileIO>& file_io,
+    const std::string& warehouse_location,
+    const std::unordered_map<std::string, std::string>& properties) {
   return std::make_shared<InMemoryCatalog>(name, file_io, warehouse_location, properties);
 }
 
 InMemoryCatalog::InMemoryCatalog(
-    std::string const& name, std::shared_ptr<FileIO> const& file_io,
-    std::string const& warehouse_location,
-    std::unordered_map<std::string, std::string> const& properties)
+    const std::string& name, const std::shared_ptr<FileIO>& file_io,
+    const std::string& warehouse_location,
+    const std::unordered_map<std::string, std::string>& properties)
     : catalog_name_(std::move(name)),
       properties_(std::move(properties)),
       file_io_(std::move(file_io)),
@@ -394,7 +411,32 @@ Result<std::unique_ptr<Table>> InMemoryCatalog::UpdateTable(
     const TableIdentifier& identifier,
     const std::vector<std::unique_ptr<TableRequirement>>& requirements,
     const std::vector<std::unique_ptr<TableUpdate>>& updates) {
-  return NotImplemented("update table");
+  std::unique_lock lock(mutex_);
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata_location,
+                          root_namespace_->GetTableMetadataLocation(identifier));
+
+  ICEBERG_ASSIGN_OR_RAISE(auto base,
+                          TableMetadataUtil::Read(*file_io_, metadata_location));
+  base->metadata_file_location = std::move(metadata_location);
+
+  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_RETURN_UNEXPECTED(
+      TableMetadataUtil::Write(*file_io_, base.get(), updated.get()));
+  ICEBERG_RETURN_UNEXPECTED(root_namespace_->UpdateTableMetadataLocation(
+      identifier, updated->metadata_file_location));
+  TableMetadataUtil::DeleteRemovedMetadataFiles(*file_io_, base.get(), updated.get());
+
+  return std::make_unique<Table>(identifier, std::move(updated), file_io_,
+                                 std::static_pointer_cast<Catalog>(shared_from_this()));
 }
 
 Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
@@ -435,9 +477,9 @@ Result<std::unique_ptr<Table>> InMemoryCatalog::LoadTable(
 
   ICEBERG_ASSIGN_OR_RAISE(auto metadata,
                           TableMetadataUtil::Read(*file_io_, metadata_location.value()));
+  metadata->metadata_file_location = metadata_location.value();
 
-  return std::make_unique<Table>(identifier, std::move(metadata),
-                                 metadata_location.value(), file_io_,
+  return std::make_unique<Table>(identifier, std::move(metadata), file_io_,
                                  std::static_pointer_cast<Catalog>(shared_from_this()));
 }
 

src/iceberg/table.cc

@@ -28,17 +28,16 @@
 #include "iceberg/table_scan.h"
 #include "iceberg/update/update_properties.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/timepoint.h"
 
 namespace iceberg {
 
 Table::~Table() = default;
 
 Table::Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
-             std::string metadata_location, std::shared_ptr<FileIO> io,
-             std::shared_ptr<Catalog> catalog)
+             std::shared_ptr<FileIO> io, std::shared_ptr<Catalog> catalog)
     : identifier_(std::move(identifier)),
       metadata_(std::move(metadata)),
-      metadata_location_(std::move(metadata_location)),
       io_(std::move(io)),
       catalog_(std::move(catalog)),
       properties_(TableProperties::FromMap(metadata_->properties)),
@@ -52,9 +51,9 @@ Status Table::Refresh() {
   }
 
   ICEBERG_ASSIGN_OR_RAISE(auto refreshed_table, catalog_->LoadTable(identifier_));
-  if (metadata_location_ != refreshed_table->metadata_location_) {
+  if (metadata_->metadata_file_location !=
+      refreshed_table->metadata_->metadata_file_location) {
     metadata_ = std::move(refreshed_table->metadata_);
-    metadata_location_ = std::move(refreshed_table->metadata_location_);
     io_ = std::move(refreshed_table->io_);
     properties_ = std::move(refreshed_table->properties_);
     metadata_cache_ = std::make_unique<TableMetadataCache>(metadata_.get());
@@ -91,8 +90,14 @@ Table::sort_orders() const {
 
 const TableProperties& Table::properties() const { return *properties_; }
 
+const std::string& Table::metadata_file_location() const {
+  return metadata_->metadata_file_location;
+}
+
 const std::string& Table::location() const { return metadata_->location; }
 
+const TimePointMs& Table::last_updated_ms() const { return metadata_->last_updated_ms; }
+
 Result<std::shared_ptr<Snapshot>> Table::current_snapshot() const {
   return metadata_->Snapshot();
 }

src/iceberg/table.h

@@ -29,6 +29,7 @@
 #include "iceberg/snapshot.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/type_fwd.h"
+#include "iceberg/util/timepoint.h"
 
 namespace iceberg {
 
@@ -45,8 +46,7 @@ class ICEBERG_EXPORT Table {
   /// \param[in] catalog The catalog that this table belongs to. If null, the table will
   /// be read-only.
   Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
-        std::string metadata_location, std::shared_ptr<FileIO> io,
-        std::shared_ptr<Catalog> catalog);
+        std::shared_ptr<FileIO> io, std::shared_ptr<Catalog> catalog);
 
   /// \brief Return the identifier of this table
   const TableIdentifier& name() const { return identifier_; }
@@ -84,9 +84,17 @@ class ICEBERG_EXPORT Table {
   /// \brief Return a map of string properties for this table
   const TableProperties& properties() const;
 
+  /// \brief Return the table's metadata file location
+  const std::string& metadata_file_location() const;
+
   /// \brief Return the table's base location
   const std::string& location() const;
 
+  /// \brief Get the time when this table was last updated
+  ///
+  /// \return the time when this table was last updated
+  const TimePointMs& last_updated_ms() const;
+
   /// \brief Return the table's current snapshot, return NotFoundError if not found
   Result<std::shared_ptr<Snapshot>> current_snapshot() const;
 
@@ -127,7 +135,6 @@ class ICEBERG_EXPORT Table {
  private:
   const TableIdentifier identifier_;
   std::shared_ptr<TableMetadata> metadata_;
-  std::string metadata_location_;
   std::shared_ptr<FileIO> io_;
   std::shared_ptr<Catalog> catalog_;
   std::unique_ptr<TableProperties> properties_;