feat: transactional UpdateProperties method support

Author: shuxu.li

src/iceberg/base_transaction.cc

@@ -19,8 +19,6 @@
 
 #include "iceberg/base_transaction.h"
 
-#include <utility>
-
 #include "iceberg/catalog.h"
 #include "iceberg/pending_update.h"
 #include "iceberg/table.h"
@@ -30,14 +28,14 @@
 
 namespace iceberg {
 
-BaseTransaction::BaseTransaction(std::shared_ptr<Table> table,
+BaseTransaction::BaseTransaction(std::shared_ptr<const Table> table,
                                  std::shared_ptr<Catalog> catalog)
     : table_(std::move(table)), catalog_(std::move(catalog)) {
   ICEBERG_DCHECK(table_ != nullptr, "table must not be null");
   ICEBERG_DCHECK(catalog_ != nullptr, "catalog must not be null");
 }
 
-const std::shared_ptr<Table>& BaseTransaction::table() const { return table_; }
+const std::shared_ptr<const Table>& BaseTransaction::table() const { return table_; }
 
 std::shared_ptr<PropertiesUpdate> BaseTransaction::UpdateProperties() {
   return RegisterUpdate<PropertiesUpdate>();

src/iceberg/base_transaction.h

@@ -29,10 +29,10 @@ namespace iceberg {
 /// \brief Base class for transaction implementations
 class BaseTransaction : public Transaction {
  public:
-  BaseTransaction(std::shared_ptr<Table> table, std::shared_ptr<Catalog> catalog);
+  BaseTransaction(std::shared_ptr<const Table> table, std::shared_ptr<Catalog> catalog);
   ~BaseTransaction() override = default;
 
-  const std::shared_ptr<Table>& table() const override;
+  const std::shared_ptr<const Table>& table() const override;
 
   std::shared_ptr<PropertiesUpdate> UpdateProperties() override;
 
@@ -48,7 +48,7 @@ class BaseTransaction : public Transaction {
     return update;
   }
 
-  std::shared_ptr<Table> table_;
+  std::shared_ptr<const Table> table_;
   std::shared_ptr<Catalog> catalog_;
   std::vector<std::shared_ptr<PendingUpdate>> pending_updates_;
 };

src/iceberg/pending_update.cc

@@ -27,10 +27,6 @@
 
 namespace iceberg {
 
-// ============================================================================
-// UpdateProperties implementation
-// ============================================================================
-
 PropertiesUpdate& PropertiesUpdate::Set(std::string const& key,
                                         std::string const& value) {
   updates_[key] = value;

src/iceberg/table.cc

@@ -19,6 +19,9 @@
 
 #include "iceberg/table.h"
 
+#include <algorithm>
+
+#include "iceberg/base_transaction.h"
 #include "iceberg/catalog.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/schema.h"
@@ -114,7 +117,7 @@ std::unique_ptr<UpdateProperties> Table::UpdateProperties() const {
 }
 
 std::unique_ptr<Transaction> Table::NewTransaction() const {
-  throw NotImplemented("Table::NewTransaction is not implemented");
+  return std::make_unique<BaseTransaction>(shared_from_this(), catalog_);
 }
 
 const std::shared_ptr<FileIO>& Table::io() const { return io_; }

src/iceberg/table.h

@@ -33,7 +33,7 @@
 namespace iceberg {
 
 /// \brief Represents an Iceberg table
-class ICEBERG_EXPORT Table {
+class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
  public:
   ~Table();
 

src/iceberg/test/transaction_pending_update_test.cc

@@ -17,34 +17,20 @@
  * under the License.
  */
 
-#include <chrono>
-#include <memory>
-#include <string>
-#include <utility>
-#include <vector>
-
-#include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
 #include "iceberg/base_transaction.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/pending_update.h"
-#include "iceberg/snapshot.h"
 #include "iceberg/sort_order.h"
 #include "iceberg/table.h"
-#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/test/matchers.h"
 #include "iceberg/test/mock_catalog.h"
-#include "iceberg/transaction.h"
-#include "iceberg/util/macros.h"
 
 namespace iceberg {
 namespace {
-
 using ::testing::ElementsAre;
 using ::testing::NiceMock;
 
@@ -80,17 +66,17 @@ std::shared_ptr<Table> CreateTestTable(const TableIdentifier& identifier,
   return std::make_shared<Table>(identifier, metadata, "s3://bucket/table/metadata.json",
                                  nullptr, catalog);
 }
+}  // namespace
 
 TEST(TransactionPendingUpdateTest, CommitSetPropertiesUsesCatalog) {
   auto metadata = CreateBaseMetadata();
   const auto identifier = MakeIdentifier();
   auto catalog = std::make_shared<NiceMock<MockCatalog>>();
   auto table =
       CreateTestTable(identifier, std::make_shared<TableMetadata>(*metadata), catalog);
-  BaseTransaction transaction(table, catalog);
-
-  auto update_properties = transaction.UpdateProperties();
-  update_properties->Set("write.metadata.delete-after-commit.enabled", "true");
+  auto transaction = table->NewTransaction();
+  auto update_properties = transaction->UpdateProperties();
+  update_properties->Set("new-key", "new-value");
 
   EXPECT_CALL(*catalog,
               UpdateTable(::testing::Eq(identifier), ::testing::_, ::testing::_))
@@ -104,13 +90,13 @@ TEST(TransactionPendingUpdateTest, CommitSetPropertiesUsesCatalog) {
             dynamic_cast<const table::SetProperties*>(updates.front().get());
         EXPECT_NE(set_update, nullptr);
         const auto& updated = set_update->updated();
-        auto it = updated.find("write.metadata.delete-after-commit.enabled");
+        auto it = updated.find("new-key");
         EXPECT_NE(it, updated.end());
-        EXPECT_EQ("true", it->second);
+        EXPECT_EQ("new-value", it->second);
         return Result<std::unique_ptr<Table>>(std::unique_ptr<Table>());
       });
 
-  EXPECT_THAT(transaction.CommitTransaction(), IsOk());
+  EXPECT_THAT(transaction->CommitTransaction(), IsOk());
 }
 
 TEST(TransactionPendingUpdateTest, RemovePropertiesSkipsMissingKeys) {
@@ -119,9 +105,9 @@ TEST(TransactionPendingUpdateTest, RemovePropertiesSkipsMissingKeys) {
   auto catalog = std::make_shared<NiceMock<MockCatalog>>();
   auto table =
       CreateTestTable(identifier, std::make_shared<TableMetadata>(*metadata), catalog);
-  BaseTransaction transaction(table, catalog);
+  auto transaction = table->NewTransaction();
 
-  auto update_properties = transaction.UpdateProperties();
+  auto update_properties = transaction->UpdateProperties();
   update_properties->Remove("missing").Remove("existing");
 
   EXPECT_CALL(*catalog,
@@ -138,7 +124,7 @@ TEST(TransactionPendingUpdateTest, RemovePropertiesSkipsMissingKeys) {
         return Result<std::unique_ptr<Table>>(std::unique_ptr<Table>());
       });
 
-  EXPECT_THAT(transaction.CommitTransaction(), IsOk());
+  EXPECT_THAT(transaction->CommitTransaction(), IsOk());
 }
 
 TEST(TransactionPendingUpdateTest, AggregatesMultiplePendingUpdates) {
@@ -147,9 +133,9 @@ TEST(TransactionPendingUpdateTest, AggregatesMultiplePendingUpdates) {
   auto catalog = std::make_shared<NiceMock<MockCatalog>>();
   auto table =
       CreateTestTable(identifier, std::make_shared<TableMetadata>(*metadata), catalog);
-  BaseTransaction transaction(table, catalog);
+  auto transaction = table->NewTransaction();
 
-  auto update_properties = transaction.UpdateProperties();
+  auto update_properties = transaction->UpdateProperties();
   update_properties->Set("new-key", "new-value").Remove("existing");
 
   EXPECT_CALL(*catalog,
@@ -176,8 +162,7 @@ TEST(TransactionPendingUpdateTest, AggregatesMultiplePendingUpdates) {
         return Result<std::unique_ptr<Table>>(std::unique_ptr<Table>());
       });
 
-  EXPECT_THAT(transaction.CommitTransaction(), IsOk());
+  EXPECT_THAT(transaction->CommitTransaction(), IsOk());
 }
 
-}  // namespace
 }  // namespace iceberg

src/iceberg/transaction.h

@@ -36,7 +36,7 @@ class ICEBERG_EXPORT Transaction {
   /// \brief Return the Table that this transaction will update
   ///
   /// \return this transaction's table
-  virtual const std::shared_ptr<Table>& table() const = 0;
+  virtual const std::shared_ptr<const Table>& table() const = 0;
 
   /// \brief Create a new update properties operation
   ///