feat: transactional UpdateProperties method support

Author: shuxu.li

src/iceberg/CMakeLists.txt

@@ -19,7 +19,6 @@ set(ICEBERG_INCLUDES "$<BUILD_INTERFACE:${PROJECT_BINARY_DIR}/src>"
                      "$<BUILD_INTERFACE:${PROJECT_SOURCE_DIR}/src>")
 set(ICEBERG_SOURCES
     arrow_c_data_guard_internal.cc
-    base_transaction.cc
     catalog/memory/in_memory_catalog.cc
     expression/aggregate.cc
     expression/binder.cc
@@ -73,6 +72,7 @@ set(ICEBERG_SOURCES
     table_requirements.cc
     table_scan.cc
     table_update.cc
+    transaction.cc
     transaction_catalog.cc
     transform.cc
     transform_function.cc

src/iceberg/base_transaction.cc

@@ -41,7 +41,6 @@ configure_file(
 iceberg_include_dir = include_directories('..')
 iceberg_sources = files(
     'arrow_c_data_guard_internal.cc',
-    'base_transaction.cc',
     'catalog/memory/in_memory_catalog.cc',
     'expression/aggregate.cc',
     'expression/binder.cc',
@@ -95,6 +94,7 @@ iceberg_sources = files(
     'table_requirements.cc',
     'table_scan.cc',
     'table_update.cc',
+    'transaction.cc',
     'transaction_catalog.cc',
     'transform.cc',
     'transform_function.cc',

src/iceberg/base_transaction.h

@@ -19,14 +19,14 @@
 
 #include "iceberg/table.h"
 
-#include "iceberg/base_transaction.h"
 #include "iceberg/catalog.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/schema.h"
 #include "iceberg/sort_order.h"
 #include "iceberg/table_metadata.h"
 #include "iceberg/table_properties.h"
 #include "iceberg/table_scan.h"
+#include "iceberg/transaction.h"
 #include "iceberg/update/update_properties.h"
 #include "iceberg/util/macros.h"
 
@@ -114,8 +114,8 @@ std::unique_ptr<UpdateProperties> Table::UpdateProperties() const {
   return std::make_unique<iceberg::UpdateProperties>(identifier_, catalog_, metadata_);
 }
 
-std::unique_ptr<Transaction> Table::NewTransaction() const {
-  return std::make_unique<BaseTransaction>(shared_from_this(), catalog_);
+Result<std::unique_ptr<Transaction>> Table::NewTransaction() const {
+  return Transaction::Make(shared_from_this(), catalog_);
 }
 
 const std::shared_ptr<FileIO>& Table::io() const { return io_; }

src/iceberg/meson.build

@@ -26,6 +26,7 @@
 #include <vector>
 
 #include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
 #include "iceberg/snapshot.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/type_fwd.h"
@@ -121,8 +122,8 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
 
   /// \brief Create a new transaction for this table
   ///
-  /// \return a pointer to the new Transaction
-  virtual std::unique_ptr<Transaction> NewTransaction() const;
+  /// \return a new Transaction or an error if the transaction cannot be created
+  virtual Result<std::unique_ptr<Transaction>> NewTransaction() const;
 
   /// \brief Returns a FileIO to read and write table data and metadata files
   const std::shared_ptr<FileIO>& io() const;

src/iceberg/table.cc

@@ -17,8 +17,6 @@
  * under the License.
  */
 
-#include "iceberg/base_transaction.h"
-
 #include <unordered_map>
 
 #include <gtest/gtest.h>
@@ -30,6 +28,7 @@
 #include "iceberg/table_update.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/mock_catalog.h"
+#include "iceberg/transaction.h"
 #include "iceberg/update/update_properties.h"
 
 namespace iceberg {
@@ -47,13 +46,22 @@ class BaseTransactionTest : public ::testing::Test {
                                 "s3://bucket/table/metadata.json", nullptr, catalog_);
   }
 
+  std::unique_ptr<Transaction> NewTransaction() {
+    auto transaction_result = BaseTransaction::Make(table_, catalog_);
+    if (!transaction_result.has_value()) {
+      ADD_FAILURE() << "Failed to create transaction: "
+                    << transaction_result.error().message;
+    }
+    return std::move(transaction_result).value();
+  }
+
   TableIdentifier identifier_;
   std::shared_ptr<MockCatalog> catalog_;
   std::shared_ptr<Table> table_;
 };
 
 TEST_F(BaseTransactionTest, CommitSetPropertiesUsesCatalog) {
-  auto transaction = table_->NewTransaction();
+  auto transaction = NewTransaction();
   auto update_properties = transaction->NewUpdateProperties();
   EXPECT_TRUE(update_properties.has_value());
   update_properties.value()->Set("new-key", "new-value");
@@ -82,7 +90,7 @@ TEST_F(BaseTransactionTest, CommitSetPropertiesUsesCatalog) {
 }
 
 TEST_F(BaseTransactionTest, RemovePropertiesSkipsMissingKeys) {
-  auto transaction = table_->NewTransaction();
+  auto transaction = NewTransaction();
   auto update_properties = transaction->NewUpdateProperties();
   EXPECT_TRUE(update_properties.has_value());
   update_properties.value()->Remove("missing").Remove("existing");
@@ -108,7 +116,7 @@ TEST_F(BaseTransactionTest, RemovePropertiesSkipsMissingKeys) {
 }
 
 TEST_F(BaseTransactionTest, AggregatesMultiplePendingUpdates) {
-  auto transaction = table_->NewTransaction();
+  auto transaction = NewTransaction();
   auto update_properties = transaction->NewUpdateProperties();
   EXPECT_TRUE(update_properties.has_value());
   update_properties.value()->Set("new-key", "new-value");
@@ -147,11 +155,20 @@ TEST_F(BaseTransactionTest, AggregatesMultiplePendingUpdates) {
 }
 
 TEST_F(BaseTransactionTest, FailsIfUpdateNotCommitted) {
-  auto transaction = table_->NewTransaction();
+  auto transaction = NewTransaction();
   auto update_properties = transaction->NewUpdateProperties();
   EXPECT_TRUE(update_properties.has_value());
   update_properties.value()->Set("new-key", "new-value");
   EXPECT_THAT(transaction->CommitTransaction(), IsError(ErrorKind::kInvalidState));
 }
 
+TEST_F(BaseTransactionTest, NewTransactionFailsWithoutCatalog) {
+  auto metadata = std::make_shared<TableMetadata>();
+  auto table_without_catalog =
+      std::make_shared<Table>(identifier_, std::move(metadata),
+                              "s3://bucket/table/metadata.json", nullptr, nullptr);
+  EXPECT_THAT(table_without_catalog->NewTransaction(),
+              IsError(ErrorKind::kInvalidArgument));
+}
+
 }  // namespace iceberg