feat: add table create/replace/update interface to catalog

Author: shuxu.li

src/iceberg/catalog.h

@@ -140,6 +140,21 @@ class ICEBERG_EXPORT Catalog {
       const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) = 0;
 
+  /// \brief Start a transaction to replace a table
+  ///
+  /// \param identifier a table identifier
+  /// \param schema a schema
+  /// \param spec a partition spec
+  /// \param location a location for the table; leave empty if unspecified
+  /// \param properties a string map of table properties
+  /// \param orCreate whether to create the table if not exists
+  /// \return a Transaction to replace the table or ErrorKind::kNotFound if the table
+  /// doesn't exist and orCreate is false
+  virtual Result<std::shared_ptr<Transaction>> StageReplaceTable(
+      const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
+      const std::string& location,
+      const std::unordered_map<std::string, std::string>& properties, bool orCreate) = 0;
+
   /// \brief Check whether table exists
   ///
   /// \param identifier a table identifier
@@ -161,6 +176,15 @@ class ICEBERG_EXPORT Catalog {
   ///         - On failure, contains error information.
   virtual Status DropTable(const TableIdentifier& identifier, bool purge) = 0;
 
+  /// \brief Rename a table
+  ///
+  /// \param from the current table identifier
+  /// \param to the new table identifier
+  /// \return Status indicating the outcome of the operation.
+  ///         - On success, the table was renamed.
+  ///         - On failure, contains error information.
+  virtual Status RenameTable(const TableIdentifier& from, const TableIdentifier& to) = 0;
+
   /// \brief Load a table
   ///
   /// \param identifier a table identifier
@@ -223,6 +247,16 @@ class ICEBERG_EXPORT Catalog {
     ///
     /// \return the Transaction to create the table
     virtual std::unique_ptr<Transaction> StageCreate() = 0;
+
+    /// \brief Starts a transaction to replace the table
+    ///
+    /// \return the Transaction to replace the table
+    virtual std::unique_ptr<Transaction> StageReplace() = 0;
+
+    /// \brief Starts a transaction to create or replace the table
+    ///
+    /// \breturn the Transaction to create or replace the table
+    virtual std::unique_ptr<Transaction> StageCreateOrReplace() = 0;
   };
 
   /// \brief Instantiate a builder to either create a table or start a create/replace

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -404,6 +404,13 @@ Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
   return NotImplemented("stage create table");
 }
 
+Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageReplaceTable(
+    const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
+    const std::string& location,
+    const std::unordered_map<std::string, std::string>& properties, bool orCreate) {
+  return NotImplemented("stage replace table");
+}
+
 Result<bool> InMemoryCatalog::TableExists(const TableIdentifier& identifier) const {
   std::unique_lock lock(mutex_);
   return root_namespace_->TableExists(identifier);
@@ -415,6 +422,12 @@ Status InMemoryCatalog::DropTable(const TableIdentifier& identifier, bool purge)
   return root_namespace_->UnregisterTable(identifier);
 }
 
+Status InMemoryCatalog::RenameTable(const TableIdentifier& from,
+                                    const TableIdentifier& to) {
+  std::unique_lock lock(mutex_);
+  return NotImplemented("rename table");
+}
+
 Result<std::unique_ptr<Table>> InMemoryCatalog::LoadTable(
     const TableIdentifier& identifier) {
   if (!file_io_) [[unlikely]] {

src/iceberg/catalog/memory/in_memory_catalog.h

@@ -85,10 +85,18 @@ class ICEBERG_EXPORT InMemoryCatalog
       const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) override;
 
+  Result<std::shared_ptr<Transaction>> StageReplaceTable(
+      const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
+      const std::string& location,
+      const std::unordered_map<std::string, std::string>& properties,
+      bool orCreate) override;
+
   Result<bool> TableExists(const TableIdentifier& identifier) const override;
 
   Status DropTable(const TableIdentifier& identifier, bool purge) override;
 
+  Status RenameTable(const TableIdentifier& from, const TableIdentifier& to) override;
+
   Result<std::unique_ptr<Table>> LoadTable(const TableIdentifier& identifier) override;
 
   Result<std::shared_ptr<Table>> RegisterTable(

src/iceberg/table.cc

@@ -133,6 +133,10 @@ const std::vector<SnapshotLogEntry>& Table::history() const {
   return metadata_->snapshot_log;
 }
 
+std::shared_ptr<Transaction> Table::NewTransaction() {
+  throw NotImplemented("Table::NewTransaction is not implemented");
+}
+
 const std::shared_ptr<FileIO>& Table::io() const { return io_; }
 
 std::unique_ptr<TableScanBuilder> Table::NewScan() const {

src/iceberg/table.h

@@ -109,6 +109,11 @@ class ICEBERG_EXPORT Table {
   /// filter data.
   virtual std::unique_ptr<TableScanBuilder> NewScan() const;
 
+  /// \brief Create a new transaction for this table
+  ///
+  /// \return a shared pointer to the new Transaction
+  virtual std::shared_ptr<Transaction> NewTransaction();
+
   /// \brief Returns a FileIO to read and write table data and metadata files
   const std::shared_ptr<FileIO>& io() const;
 

src/iceberg/test/mock_catalog.h

@@ -71,10 +71,19 @@ class MockCatalog : public Catalog {
                const std::string&, (const std::unordered_map<std::string, std::string>&)),
               (override));
 
+  MOCK_METHOD((Result<std::shared_ptr<Transaction>>), StageReplaceTable,
+              (const TableIdentifier&, const Schema&, const PartitionSpec&,
+               const std::string&, (const std::unordered_map<std::string, std::string>&),
+               bool),
+              (override));
+
   MOCK_METHOD(Result<bool>, TableExists, (const TableIdentifier&), (const, override));
 
   MOCK_METHOD(Status, DropTable, (const TableIdentifier&, bool), (override));
 
+  MOCK_METHOD(Status, RenameTable, (const TableIdentifier&, const TableIdentifier&),
+              (override));
+
   MOCK_METHOD((Result<std::unique_ptr<Table>>), LoadTable, (const TableIdentifier&),
               (override));
 

src/iceberg/transaction.h

@@ -21,6 +21,7 @@
 #pragma once
 
 #include <memory>
+#include <vector>
 
 #include "iceberg/iceberg_export.h"
 #include "iceberg/type_fwd.h"
@@ -37,17 +38,28 @@ class ICEBERG_EXPORT Transaction {
   /// \return this transaction's table
   virtual const std::shared_ptr<Table>& table() const = 0;
 
-  /// \brief Create a new append API to add files to this table
+  /// \brief Create a new append API to add data files to this table
   ///
-  /// \return a new AppendFiles
+  /// \return a new AppendFiles instance
   virtual std::shared_ptr<AppendFiles> NewAppend() = 0;
 
+  /// \brief Apply multiple metadata updates to this transaction
+  ///
+  /// \param requirements the table requirements to validate
+  /// \param updates the table updates to apply
+  /// \return Status::OK if all updates were queued successfully
+  virtual Status UpdateTable(
+      const std::vector<std::unique_ptr<TableRequirement>>& requirements,
+      std::vector<std::unique_ptr<TableUpdate>> updates) = 0;
+
   /// \brief Apply the pending changes from all actions and commit
   ///
-  /// May throw ValidationException if any update cannot be applied to the current table
-  /// metadata. May throw CommitFailedException if the updates cannot be committed due to
-  /// conflicts.
-  virtual void CommitTransaction() = 0;
+  /// This method applies all pending data operations and metadata updates in the
+  /// transaction and commits them to the table in a single atomic operation.
+  ///
+  /// \return Status::OK if the transaction was committed successfully, or an error
+  ///         status if validation failed or the commit encountered conflicts
+  virtual Status CommitTransaction() = 0;
 };
 
 }  // namespace iceberg