feat: implement transaction api (#418)

- Introduce `Transaction` class to manage multi-operation table updates.
- Add `Table::NewTransaction()` and `Table::NewUpdateProperties()` to
initiate updates.
- Move `PendingUpdate` and `UpdateProperties` to `src/iceberg/update/`
and refactor them to use the transaction mechanism.
- Add `StagedTable` to represent tables with uncommitted changes.
- Update `InMemoryCatalog` and `RestCatalog` to align with the new
update flow.
- Refactor `Table` to support metadata refresh and location management
within transactions.
- Add comprehensive tests for transactions and property updates.

Author: wgtmac

example/demo_example.cc

@@ -58,7 +58,13 @@ int main(int argc, char** argv) {
   }
 
   auto table = std::move(load_result.value());
-  auto scan_result = table->NewScan()->Build();
+  auto scan_builder = table->NewScan();
+  if (!scan_builder.has_value()) {
+    std::cerr << "Failed to create scan builder: " << scan_builder.error().message
+              << std::endl;
+    return 1;
+  }
+  auto scan_result = scan_builder.value()->Build();
   if (!scan_result.has_value()) {
     std::cerr << "Failed to build scan: " << scan_result.error().message << std::endl;
     return 1;

src/iceberg/CMakeLists.txt

@@ -72,9 +72,11 @@ set(ICEBERG_SOURCES
     table_requirements.cc
     table_scan.cc
     table_update.cc
+    transaction.cc
     transform.cc
     transform_function.cc
     type.cc
+    update/pending_update.cc
     update/update_properties.cc
     util/bucket_util.cc
     util/conversions.cc

src/iceberg/catalog.h

@@ -110,7 +110,7 @@ class ICEBERG_EXPORT Catalog {
   /// \param location a location for the table; leave empty if unspecified
   /// \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(
+  virtual Result<std::shared_ptr<Table>> CreateTable(
       const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
       const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) = 0;
@@ -121,7 +121,7 @@ class ICEBERG_EXPORT Catalog {
   /// \param requirements a list of table requirements
   /// \param updates a list of table updates
   /// \return a Table instance or ErrorKind::kAlreadyExists if the table already exists
-  virtual Result<std::unique_ptr<Table>> UpdateTable(
+  virtual Result<std::shared_ptr<Table>> UpdateTable(
       const TableIdentifier& identifier,
       const std::vector<std::unique_ptr<TableRequirement>>& requirements,
       const std::vector<std::unique_ptr<TableUpdate>>& updates) = 0;
@@ -175,7 +175,7 @@ class ICEBERG_EXPORT Catalog {
   /// \param identifier a table identifier
   /// \return instance of Table implementation referred to by identifier or
   /// ErrorKind::kNoSuchTable if the table does not exist
-  virtual Result<std::unique_ptr<Table>> LoadTable(const TableIdentifier& identifier) = 0;
+  virtual Result<std::shared_ptr<Table>> LoadTable(const TableIdentifier& identifier) = 0;
 
   /// \brief Register a table with the catalog if it does not exist
   ///

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -399,15 +399,15 @@ Result<std::vector<TableIdentifier>> InMemoryCatalog::ListTables(
   return table_idents;
 }
 
-Result<std::unique_ptr<Table>> InMemoryCatalog::CreateTable(
+Result<std::shared_ptr<Table>> InMemoryCatalog::CreateTable(
     const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
     const std::string& location,
     const std::unordered_map<std::string, std::string>& properties) {
   std::unique_lock lock(mutex_);
   return NotImplemented("create table");
 }
 
-Result<std::unique_ptr<Table>> InMemoryCatalog::UpdateTable(
+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) {
@@ -434,9 +434,8 @@ Result<std::unique_ptr<Table>> InMemoryCatalog::UpdateTable(
       root_namespace_->UpdateTableMetadataLocation(identifier, new_metadata_location));
   TableMetadataUtil::DeleteRemovedMetadataFiles(*file_io_, base.get(), *updated);
 
-  return std::make_unique<Table>(identifier, std::move(updated),
-                                 std::move(new_metadata_location), file_io_,
-                                 std::static_pointer_cast<Catalog>(shared_from_this()));
+  return Table::Make(identifier, std::move(updated), std::move(new_metadata_location),
+                     file_io_, shared_from_this());
 }
 
 Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
@@ -464,7 +463,7 @@ Status InMemoryCatalog::RenameTable(const TableIdentifier& from,
   return NotImplemented("rename table");
 }
 
-Result<std::unique_ptr<Table>> InMemoryCatalog::LoadTable(
+Result<std::shared_ptr<Table>> InMemoryCatalog::LoadTable(
     const TableIdentifier& identifier) {
   if (!file_io_) [[unlikely]] {
     return InvalidArgument("file_io is not set for catalog {}", catalog_name_);
@@ -479,9 +478,8 @@ Result<std::unique_ptr<Table>> InMemoryCatalog::LoadTable(
 
   ICEBERG_ASSIGN_OR_RAISE(auto metadata,
                           TableMetadataUtil::Read(*file_io_, metadata_location));
-  return std::make_unique<Table>(identifier, std::move(metadata),
-                                 std::move(metadata_location), file_io_,
-                                 std::static_pointer_cast<Catalog>(shared_from_this()));
+  return Table::Make(identifier, std::move(metadata), std::move(metadata_location),
+                     file_io_, shared_from_this());
 }
 
 Result<std::shared_ptr<Table>> InMemoryCatalog::RegisterTable(
@@ -500,9 +498,8 @@ Result<std::shared_ptr<Table>> InMemoryCatalog::RegisterTable(
   if (!root_namespace_->RegisterTable(identifier, metadata_file_location)) {
     return UnknownError("The registry failed.");
   }
-  return std::make_unique<Table>(identifier, std::move(metadata), metadata_file_location,
-                                 file_io_,
-                                 std::static_pointer_cast<Catalog>(shared_from_this()));
+  return Table::Make(identifier, std::move(metadata), metadata_file_location, file_io_,
+                     shared_from_this());
 }
 
 }  // namespace iceberg

src/iceberg/catalog/memory/in_memory_catalog.h

@@ -70,12 +70,12 @@ class ICEBERG_EXPORT InMemoryCatalog
 
   Result<std::vector<TableIdentifier>> ListTables(const Namespace& ns) const override;
 
-  Result<std::unique_ptr<Table>> CreateTable(
+  Result<std::shared_ptr<Table>> CreateTable(
       const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
       const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) override;
 
-  Result<std::unique_ptr<Table>> UpdateTable(
+  Result<std::shared_ptr<Table>> UpdateTable(
       const TableIdentifier& identifier,
       const std::vector<std::unique_ptr<TableRequirement>>& requirements,
       const std::vector<std::unique_ptr<TableUpdate>>& updates) override;
@@ -91,7 +91,7 @@ class ICEBERG_EXPORT InMemoryCatalog
 
   Status RenameTable(const TableIdentifier& from, const TableIdentifier& to) override;
 
-  Result<std::unique_ptr<Table>> LoadTable(const TableIdentifier& identifier) override;
+  Result<std::shared_ptr<Table>> LoadTable(const TableIdentifier& identifier) override;
 
   Result<std::shared_ptr<Table>> RegisterTable(
       const TableIdentifier& identifier,

src/iceberg/catalog/rest/rest_catalog.cc

@@ -240,15 +240,15 @@ Result<std::vector<TableIdentifier>> RestCatalog::ListTables(
   return NotImplemented("Not implemented");
 }
 
-Result<std::unique_ptr<Table>> RestCatalog::CreateTable(
+Result<std::shared_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::unordered_map<std::string, std::string>& properties) {
   return NotImplemented("Not implemented");
 }
 
-Result<std::unique_ptr<Table>> RestCatalog::UpdateTable(
+Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
     [[maybe_unused]] const TableIdentifier& identifier,
     [[maybe_unused]] const std::vector<std::unique_ptr<TableRequirement>>& requirements,
     [[maybe_unused]] const std::vector<std::unique_ptr<TableUpdate>>& updates) {
@@ -278,7 +278,7 @@ Status RestCatalog::RenameTable([[maybe_unused]] const TableIdentifier& from,
   return NotImplemented("Not implemented");
 }
 
-Result<std::unique_ptr<Table>> RestCatalog::LoadTable(
+Result<std::shared_ptr<Table>> RestCatalog::LoadTable(
     [[maybe_unused]] const TableIdentifier& identifier) {
   return NotImplemented("Not implemented");
 }

src/iceberg/catalog/rest/rest_catalog.h

@@ -71,12 +71,12 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog {
 
   Result<std::vector<TableIdentifier>> ListTables(const Namespace& ns) const override;
 
-  Result<std::unique_ptr<Table>> CreateTable(
+  Result<std::shared_ptr<Table>> CreateTable(
       const TableIdentifier& identifier, const Schema& schema, const PartitionSpec& spec,
       const std::string& location,
       const std::unordered_map<std::string, std::string>& properties) override;
 
-  Result<std::unique_ptr<Table>> UpdateTable(
+  Result<std::shared_ptr<Table>> UpdateTable(
       const TableIdentifier& identifier,
       const std::vector<std::unique_ptr<TableRequirement>>& requirements,
       const std::vector<std::unique_ptr<TableUpdate>>& updates) override;
@@ -92,7 +92,7 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog {
 
   Status DropTable(const TableIdentifier& identifier, bool purge) override;
 
-  Result<std::unique_ptr<Table>> LoadTable(const TableIdentifier& identifier) override;
+  Result<std::shared_ptr<Table>> LoadTable(const TableIdentifier& identifier) override;
 
   Result<std::shared_ptr<Table>> RegisterTable(
       const TableIdentifier& identifier,

src/iceberg/meson.build

@@ -94,9 +94,11 @@ iceberg_sources = files(
     'table_requirements.cc',
     'table_scan.cc',
     'table_update.cc',
+    'transaction.cc',
     'transform.cc',
     'transform_function.cc',
     'type.cc',
+    'update/pending_update.cc',
     'update/update_properties.cc',
     'util/bucket_util.cc',
     'util/conversions.cc',
@@ -175,7 +177,6 @@ install_headers(
         'name_mapping.h',
         'partition_field.h',
         'partition_spec.h',
-        'pending_update.h',
         'result.h',
         'schema_field.h',
         'schema.h',
@@ -196,7 +197,6 @@ install_headers(
         'transform.h',
         'type_fwd.h',
         'type.h',
-        'update/update_properties.h',
     ],
     subdir: 'iceberg',
 )
@@ -205,6 +205,7 @@ subdir('catalog')
 subdir('expression')
 subdir('manifest')
 subdir('row')
+subdir('update')
 subdir('util')
 
 if get_option('tests').enabled()