refactor: introduce TransactionContext to decouple Transaction and PendingUpdate (#591)

Add TransactionContext to own the shared state (table, metadata_builder,
kind) between Transaction and PendingUpdate. Both now hold a
shared_ptr<TransactionContext> instead of PendingUpdate holding a
weak_ptr<Transaction>.

This fixes two issues:
- pending_updates_ was weak_ptr, so dropping a PendingUpdate would
silently break Finalize/retry; now Transaction holds shared_ptr
- Table::New*() no longer creates a temporary Transaction; it creates a
TransactionContext directly and passes it to the PendingUpdate, removing
the circular dependency

Also clean up related redundancy:
- Hoist Transaction::Kind to a standalone enum class TransactionKind
- Remove Transaction::kind_ (duplicate of ctx_->kind)
- Remove auto_commit machinery; PendingUpdate::Commit() now calls
txn->Commit() explicitly on the table-created path
- TransactionContext::Make returns Result to propagate null table errors

Author: wgtmac

src/iceberg/catalog/memory/in_memory_catalog.cc

@@ -500,8 +500,7 @@ Result<std::shared_ptr<Transaction>> InMemoryCatalog::StageCreateTable(
   ICEBERG_ASSIGN_OR_RAISE(
       auto table, StagedTable::Make(identifier, std::move(table_metadata), "", file_io_,
                                     shared_from_this()));
-  return Transaction::Make(std::move(table), Transaction::Kind::kCreate,
-                           /* auto_commit */ false);
+  return Transaction::Make(std::move(table), TransactionKind::kCreate);
 }
 
 Result<bool> InMemoryCatalog::TableExists(const TableIdentifier& identifier) const {

src/iceberg/catalog/rest/rest_catalog.cc

@@ -384,8 +384,7 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
                           StagedTable::Make(identifier, std::move(result.metadata),
                                             std::move(result.metadata_location), file_io_,
                                             shared_from_this()));
-  return Transaction::Make(std::move(staged_table), Transaction::Kind::kCreate,
-                           /*auto_commit=*/false);
+  return Transaction::Make(std::move(staged_table), TransactionKind::kCreate);
 }
 
 Status RestCatalog::DropTable(const TableIdentifier& identifier, bool purge) {

src/iceberg/table.cc

@@ -32,11 +32,16 @@
 #include "iceberg/table_scan.h"
 #include "iceberg/transaction.h"
 #include "iceberg/update/expire_snapshots.h"
+#include "iceberg/update/fast_append.h"
+#include "iceberg/update/set_snapshot.h"
 #include "iceberg/update/snapshot_manager.h"
+#include "iceberg/update/update_location.h"
 #include "iceberg/update/update_partition_spec.h"
 #include "iceberg/update/update_partition_statistics.h"
 #include "iceberg/update/update_properties.h"
 #include "iceberg/update/update_schema.h"
+#include "iceberg/update/update_snapshot_reference.h"
+#include "iceberg/update/update_sort_order.h"
 #include "iceberg/update/update_statistics.h"
 #include "iceberg/util/macros.h"
 
@@ -166,71 +171,61 @@ Table::NewIncrementalChangelogScan() const {
 Result<std::shared_ptr<Transaction>> Table::NewTransaction() {
   // Create a brand new transaction object for the table. Users are expected to commit the
   // transaction manually.
-  return Transaction::Make(shared_from_this(), Transaction::Kind::kUpdate,
-                           /*auto_commit=*/false);
+  return Transaction::Make(shared_from_this(), TransactionKind::kUpdate);
 }
 
 Result<std::shared_ptr<UpdatePartitionSpec>> Table::NewUpdatePartitionSpec() {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto transaction, Transaction::Make(shared_from_this(), Transaction::Kind::kUpdate,
-                                          /*auto_commit=*/true));
-  return transaction->NewUpdatePartitionSpec();
+      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
+  return UpdatePartitionSpec::Make(std::move(ctx));
 }
 
 Result<std::shared_ptr<UpdateProperties>> Table::NewUpdateProperties() {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto transaction, Transaction::Make(shared_from_this(), Transaction::Kind::kUpdate,
-                                          /*auto_commit=*/true));
-  return transaction->NewUpdateProperties();
+      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
+  return UpdateProperties::Make(std::move(ctx));
 }
 
 Result<std::shared_ptr<UpdateSortOrder>> Table::NewUpdateSortOrder() {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto transaction, Transaction::Make(shared_from_this(), Transaction::Kind::kUpdate,
-                                          /*auto_commit=*/true));
-  return transaction->NewUpdateSortOrder();
+      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
+  return UpdateSortOrder::Make(std::move(ctx));
 }
 
 Result<std::shared_ptr<UpdateSchema>> Table::NewUpdateSchema() {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto transaction, Transaction::Make(shared_from_this(), Transaction::Kind::kUpdate,
-                                          /*auto_commit=*/true));
-  return transaction->NewUpdateSchema();
+      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
+  return UpdateSchema::Make(std::move(ctx));
 }
 
 Result<std::shared_ptr<ExpireSnapshots>> Table::NewExpireSnapshots() {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto transaction, Transaction::Make(shared_from_this(), Transaction::Kind::kUpdate,
-                                          /*auto_commit=*/true));
-  return transaction->NewExpireSnapshots();
+      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
+  return ExpireSnapshots::Make(std::move(ctx));
 }
 
 Result<std::shared_ptr<UpdateLocation>> Table::NewUpdateLocation() {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto transaction, Transaction::Make(shared_from_this(), Transaction::Kind::kUpdate,
-                                          /*auto_commit=*/true));
-  return transaction->NewUpdateLocation();
+      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
+  return UpdateLocation::Make(std::move(ctx));
 }
 
 Result<std::shared_ptr<FastAppend>> Table::NewFastAppend() {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto transaction, Transaction::Make(shared_from_this(), Transaction::Kind::kUpdate,
-                                          /*auto_commit=*/true));
-  return transaction->NewFastAppend();
+      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
+  return FastAppend::Make(name().name, std::move(ctx));
 }
 
 Result<std::shared_ptr<UpdateStatistics>> Table::NewUpdateStatistics() {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto transaction, Transaction::Make(shared_from_this(), Transaction::Kind::kUpdate,
-                                          /*auto_commit=*/true));
-  return transaction->NewUpdateStatistics();
+      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
+  return UpdateStatistics::Make(std::move(ctx));
 }
 
 Result<std::shared_ptr<UpdatePartitionStatistics>> Table::NewUpdatePartitionStatistics() {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto transaction, Transaction::Make(shared_from_this(), Transaction::Kind::kUpdate,
-                                          /*auto_commit=*/true));
-  return transaction->NewUpdatePartitionStatistics();
+      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
+  return UpdatePartitionStatistics::Make(std::move(ctx));
 }
 
 Result<std::shared_ptr<SnapshotManager>> Table::NewSnapshotManager() {

src/iceberg/test/update_partition_spec_test.cc

@@ -232,14 +232,12 @@ class UpdatePartitionSpecTest : public ::testing::TestWithParam<int8_t> {
   // Helper to create UpdatePartitionSpec from a table
   std::shared_ptr<UpdatePartitionSpec> CreateUpdateFromTable(
       std::shared_ptr<Table> table) {
-    auto transaction_result =
-        Transaction::Make(table, Transaction::Kind::kUpdate, /*auto_commit=*/false);
-    if (!transaction_result.has_value()) {
-      ADD_FAILURE() << "Failed to create transaction: "
-                    << transaction_result.error().message;
+    auto ctx_result = TransactionContext::Make(table, TransactionKind::kUpdate);
+    if (!ctx_result.has_value()) {
+      ADD_FAILURE() << "Failed to create context: " << ctx_result.error().message;
       return nullptr;
     }
-    auto update_result = UpdatePartitionSpec::Make(transaction_result.value());
+    auto update_result = UpdatePartitionSpec::Make(std::move(ctx_result.value()));
     if (!update_result.has_value()) {
       ADD_FAILURE() << "Failed to create UpdatePartitionSpec: "
                     << update_result.error().message;