fix: unify pending update lifecycle

Author: zhjwpku

src/iceberg/test/fast_append_test.cc

@@ -307,6 +307,26 @@ TEST_F(FastAppendTest, FinalizeIgnoresCleanupDeleteFailure) {
               IsOk());
 }
 
+TEST_F(FastAppendTest, TransactionApplyFailureCleansUpStagedFiles) {
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto fast_append, txn->NewFastAppend());
+  std::vector<std::string> deleted_paths;
+  fast_append->DeleteWith([&](const std::string& path) {
+    deleted_paths.push_back(path);
+    return file_io_->DeleteFile(path);
+  });
+  fast_append->AppendFile(file_a_);
+
+  EXPECT_THAT(static_cast<SnapshotUpdate&>(*fast_append).Apply(), IsOk());
+  fast_append->AppendFile(nullptr);
+
+  EXPECT_THAT(fast_append->Commit(), IsError(ErrorKind::kValidationFailed));
+  EXPECT_THAT(deleted_paths, ::testing::SizeIs(2U));
+  EXPECT_THAT(txn->Commit(),
+              ::testing::AllOf(IsError(ErrorKind::kValidationFailed),
+                               HasErrorMessage("Transaction already finalized")));
+}
+
 TEST_F(FastAppendTest, RetryCopiesAppendManifestAgain) {
   table_->metadata()->format_version = 1;
   const auto path = table_location_ + "/metadata/input.avro";

src/iceberg/test/merging_snapshot_update_test.cc

@@ -110,11 +110,11 @@ class ScopedPuffinDVIORegistry {
 /// \brief Concrete subclass of MergingSnapshotUpdate for testing.
 class TestMergeAppend : public MergingSnapshotUpdate {
  public:
-  static Result<std::unique_ptr<TestMergeAppend>> Make(std::string table_name,
+  static Result<std::shared_ptr<TestMergeAppend>> Make(std::string table_name,
                                                        std::shared_ptr<Table> table) {
     ICEBERG_ASSIGN_OR_RAISE(
         auto ctx, TransactionContext::Make(std::move(table), TransactionKind::kUpdate));
-    return std::unique_ptr<TestMergeAppend>(
+    return std::shared_ptr<TestMergeAppend>(
         new TestMergeAppend(std::move(table_name), std::move(ctx)));
   }
 
@@ -256,11 +256,11 @@ class TestMergeAppend : public MergingSnapshotUpdate {
 
 class TestOverwriteUpdate : public MergingSnapshotUpdate {
  public:
-  static Result<std::unique_ptr<TestOverwriteUpdate>> Make(std::string table_name,
+  static Result<std::shared_ptr<TestOverwriteUpdate>> Make(std::string table_name,
                                                            std::shared_ptr<Table> table) {
     ICEBERG_ASSIGN_OR_RAISE(
         auto ctx, TransactionContext::Make(std::move(table), TransactionKind::kUpdate));
-    return std::unique_ptr<TestOverwriteUpdate>(
+    return std::shared_ptr<TestOverwriteUpdate>(
         new TestOverwriteUpdate(std::move(table_name), std::move(ctx)));
   }
 
@@ -336,11 +336,11 @@ class MergingSnapshotUpdateTest : public MinimalUpdateTestBase {
     return f;
   }
 
-  Result<std::unique_ptr<TestMergeAppend>> NewMergeAppend() {
+  Result<std::shared_ptr<TestMergeAppend>> NewMergeAppend() {
     return TestMergeAppend::Make(TableName(), table_);
   }
 
-  Result<std::unique_ptr<TestOverwriteUpdate>> NewOverwriteUpdate() {
+  Result<std::shared_ptr<TestOverwriteUpdate>> NewOverwriteUpdate() {
     return TestOverwriteUpdate::Make(TableName(), table_);
   }
 
@@ -983,7 +983,7 @@ class MergingSnapshotUpdateV1Test : public UpdateTestBase {
     return f;
   }
 
-  Result<std::unique_ptr<TestMergeAppend>> NewMergeAppend() {
+  Result<std::shared_ptr<TestMergeAppend>> NewMergeAppend() {
     return TestMergeAppend::Make(TableName(), table_);
   }
 

src/iceberg/test/transaction_test.cc

@@ -19,14 +19,22 @@
 
 #include "iceberg/transaction.h"
 
+#include <format>
+#include <memory>
+#include <string>
+#include <vector>
+
 #include "iceberg/expression/expressions.h"
 #include "iceberg/expression/term.h"
 #include "iceberg/sort_order.h"
+#include "iceberg/table_metadata.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/mock_catalog.h"
 #include "iceberg/test/update_test_base.h"
 #include "iceberg/transform.h"
 #include "iceberg/type.h"
+#include "iceberg/update/fast_append.h"
+#include "iceberg/update/set_snapshot.h"
 #include "iceberg/update/update_properties.h"
 #include "iceberg/update/update_schema.h"
 #include "iceberg/update/update_sort_order.h"
@@ -46,6 +54,47 @@ TEST_F(TransactionTest, CommitEmptyTransaction) {
   EXPECT_THAT(txn->Commit(), IsOk());
 }
 
+TEST_F(TransactionTest, TemporaryTransactionDoesNotAttachToContext) {
+  ICEBERG_UNWRAP_OR_FAIL(auto ctx,
+                         TransactionContext::Make(table_, TransactionKind::kUpdate));
+  ASSERT_FALSE(ctx->transaction.has_value());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, Transaction::Make(ctx));
+
+  EXPECT_NE(txn, nullptr);
+  EXPECT_FALSE(ctx->transaction.has_value());
+}
+
+TEST_F(TransactionTest, StandaloneCommitRequiresSharedOwnership) {
+  ICEBERG_UNWRAP_OR_FAIL(auto ctx,
+                         TransactionContext::Make(table_, TransactionKind::kUpdate));
+  ICEBERG_UNWRAP_OR_FAIL(auto update, FastAppend::Make(table_->name().name, ctx));
+
+  EXPECT_THAT(update->Commit(),
+              ::testing::AllOf(
+                  IsError(ErrorKind::kInvalidArgument),
+                  HasErrorMessage("PendingUpdate must be owned by std::shared_ptr")));
+}
+
+TEST_F(TransactionTest, CommitNoOpUpdate) {
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewSetSnapshot());
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_THAT(txn->Commit(), IsOk());
+}
+
+TEST_F(TransactionTest, ApplyFailureFinalizesTransaction) {
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewFastAppend());
+  update->AppendFile(nullptr);
+
+  EXPECT_THAT(update->Commit(), IsError(ErrorKind::kValidationFailed));
+  EXPECT_THAT(txn->Commit(),
+              ::testing::AllOf(IsError(ErrorKind::kValidationFailed),
+                               HasErrorMessage("Transaction already finalized")));
+}
+
 TEST_F(TransactionTest, CommitTransactionWithPropertyUpdate) {
   ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
   ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewUpdateProperties());
@@ -151,6 +200,39 @@ TEST_F(TransactionRetryTest, CommitRetrySucceedsAfterConflict) {
   EXPECT_EQ(update_call_count, 2);
 }
 
+TEST_F(TransactionRetryTest, StandaloneCommitRetryReappliesUpdate) {
+  std::vector<size_t> update_counts;
+  ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))
+      .WillByDefault(
+          [this, &update_counts](const TableIdentifier&,
+                                 const std::vector<std::unique_ptr<TableRequirement>>&,
+                                 const std::vector<std::unique_ptr<TableUpdate>>& updates)
+              -> Result<std::shared_ptr<Table>> {
+            update_counts.push_back(updates.size());
+            if (update_counts.size() == 1) {
+              return CommitFailed("conflict on first attempt");
+            }
+            return Table::Make(mock_table_->name(), mock_table_->metadata(),
+                               std::string(mock_table_->metadata_file_location()),
+                               mock_table_->io(), mock_catalog_);
+          });
+  EXPECT_CALL(*mock_catalog_, LoadTable(::testing::_))
+      .WillOnce([this](const TableIdentifier&) -> Result<std::shared_ptr<Table>> {
+        auto builder = TableMetadataBuilder::BuildFrom(mock_table_->metadata().get());
+        ICEBERG_ASSIGN_OR_RAISE(auto metadata, builder->Build());
+        return Table::Make(
+            mock_table_->name(), std::shared_ptr<TableMetadata>(std::move(metadata)),
+            std::format("{}.refreshed", mock_table_->metadata_file_location()),
+            mock_table_->io(), mock_catalog_);
+      });
+
+  ICEBERG_UNWRAP_OR_FAIL(auto update, mock_table_->NewUpdateProperties());
+  update->Set("retry.test", "value");
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_THAT(update_counts, ::testing::ElementsAre(1U, 1U));
+}
+
 TEST_F(TransactionRetryTest, CommitRetryExhausted) {
   int update_call_count = 0;
   ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))

src/iceberg/transaction.cc

@@ -123,9 +123,7 @@ Result<std::shared_ptr<Transaction>> Transaction::Make(std::shared_ptr<Table> ta
 Result<std::shared_ptr<Transaction>> Transaction::Make(
     std::shared_ptr<TransactionContext> ctx) {
   ICEBERG_PRECHECK(ctx != nullptr, "TransactionContext cannot be null");
-  auto txn = std::shared_ptr<Transaction>(new Transaction(ctx));
-  ctx->transaction = std::weak_ptr<Transaction>(txn);
-  return txn;
+  return std::shared_ptr<Transaction>(new Transaction(std::move(ctx)));
 }
 
 const std::shared_ptr<Table>& Transaction::table() const { return ctx_->table; }
@@ -139,6 +137,8 @@ std::string Transaction::MetadataFileLocation(std::string_view filename) const {
 }
 
 Status Transaction::AddUpdate(const std::shared_ptr<PendingUpdate>& update) {
+  ICEBERG_CHECK(!committed_, "Cannot add update to a committed transaction");
+  ICEBERG_CHECK(!finalized_, "Cannot add update to a finalized transaction");
   ICEBERG_CHECK(last_update_committed_,
                 "Cannot add update when previous update is not committed");
 
@@ -148,6 +148,9 @@ Status Transaction::AddUpdate(const std::shared_ptr<PendingUpdate>& update) {
 }
 
 Status Transaction::Apply(PendingUpdate& update) {
+  ICEBERG_CHECK(!committed_, "Cannot apply update to a committed transaction");
+  ICEBERG_CHECK(!finalized_, "Cannot apply update to a finalized transaction");
+
   switch (update.kind()) {
     case PendingUpdate::Kind::kExpireSnapshots:
       ICEBERG_RETURN_UNEXPECTED(
@@ -359,40 +362,36 @@ Status Transaction::ApplyUpdatePartitionStatistics(UpdatePartitionStatistics& up
 
 Result<std::shared_ptr<Table>> Transaction::Commit() {
   ICEBERG_CHECK(!committed_, "Transaction already committed");
+  ICEBERG_CHECK(!finalized_, "Transaction already finalized");
   ICEBERG_CHECK(last_update_committed_,
                 "Cannot commit transaction when previous update is not committed");
 
   const auto& updates = ctx_->metadata_builder->changes();
-  if (updates.empty()) {
-    committed_ = true;
-    return ctx_->table;
+  Result<std::shared_ptr<Table>> commit_result = ctx_->table;
+  if (!updates.empty()) {
+    const auto& props = ctx_->table->properties();
+    int32_t num_retries =
+        CanRetry() ? static_cast<int32_t>(props.Get(TableProperties::kCommitNumRetries))
+                   : 0;
+    int32_t min_wait_ms = props.Get(TableProperties::kCommitMinRetryWaitMs);
+    int32_t max_wait_ms = props.Get(TableProperties::kCommitMaxRetryWaitMs);
+    int32_t total_timeout_ms = props.Get(TableProperties::kCommitTotalRetryTimeMs);
+
+    bool is_first_attempt = true;
+    commit_result =
+        MakeCommitRetryRunner(num_retries, min_wait_ms, max_wait_ms, total_timeout_ms)
+            .Run([this, &is_first_attempt]() -> Result<std::shared_ptr<Table>> {
+              auto result = CommitOnce(is_first_attempt);
+              is_first_attempt = false;
+              return result;
+            });
   }
 
-  const auto& props = ctx_->table->properties();
-  int32_t num_retries =
-      CanRetry() ? static_cast<int32_t>(props.Get(TableProperties::kCommitNumRetries))
-                 : 0;
-  int32_t min_wait_ms = props.Get(TableProperties::kCommitMinRetryWaitMs);
-  int32_t max_wait_ms = props.Get(TableProperties::kCommitMaxRetryWaitMs);
-  int32_t total_timeout_ms = props.Get(TableProperties::kCommitTotalRetryTimeMs);
-
-  bool is_first_attempt = true;
-  auto commit_result =
-      MakeCommitRetryRunner(num_retries, min_wait_ms, max_wait_ms, total_timeout_ms)
-          .Run([this, &is_first_attempt]() -> Result<std::shared_ptr<Table>> {
-            auto result = CommitOnce(is_first_attempt);
-            is_first_attempt = false;
-            return result;
-          });
-
   Result<const TableMetadata*> finalize_result =
       commit_result.has_value()
           ? Result<const TableMetadata*>(commit_result.value()->metadata().get())
           : std::unexpected(commit_result.error());
-
-  for (const auto& update : pending_updates_) {
-    std::ignore = update->Finalize(finalize_result);
-  }
+  FinalizeUpdates(finalize_result);
 
   ICEBERG_RETURN_UNEXPECTED(commit_result);
 
@@ -403,6 +402,16 @@ Result<std::shared_ptr<Table>> Transaction::Commit() {
   return ctx_->table;
 }
 
+void Transaction::FinalizeUpdates(const Result<const TableMetadata*>& commit_result) {
+  if (finalized_) {
+    return;
+  }
+  finalized_ = true;
+  for (const auto& update : pending_updates_) {
+    std::ignore = update->Finalize(commit_result);
+  }
+}
+
 Result<std::shared_ptr<Table>> Transaction::CommitOnce(bool is_first_attempt) {
   std::vector<std::unique_ptr<TableRequirement>> requirements;
 

src/iceberg/transaction.h

@@ -48,7 +48,10 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this<Transacti
   static Result<std::shared_ptr<Transaction>> Make(std::shared_ptr<Table> table,
                                                    TransactionKind kind);
 
-  /// \brief Create a transaction from an existing context (used by PendingUpdate::Commit)
+  /// \brief Create a detached transaction from an existing context.
+  ///
+  /// This overload is used by PendingUpdate::Commit for standalone updates and does not
+  /// attach the temporary transaction to the context.
   static Result<std::shared_ptr<Transaction>> Make(
       std::shared_ptr<TransactionContext> ctx);
 
@@ -163,6 +166,9 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this<Transacti
   /// \brief Whether this transaction can retry after a commit conflict.
   bool CanRetry() const;
 
+  /// \brief Finalize all registered updates exactly once.
+  void FinalizeUpdates(const Result<const TableMetadata*>& commit_result);
+
  private:
   friend class PendingUpdate;
 
@@ -174,6 +180,8 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this<Transacti
   bool last_update_committed_ = true;
   // Tracks if transaction has been committed to prevent double-commit
   bool committed_ = false;
+  // Tracks whether registered updates have reached a terminal state.
+  bool finalized_ = false;
 };
 
 /// \brief Shared context between Transaction and PendingUpdate instances.

src/iceberg/update/pending_update.cc

@@ -34,27 +34,33 @@ PendingUpdate::~PendingUpdate() = default;
 Status PendingUpdate::Commit() {
   if (!ctx_->transaction) {
     // Table-created path: no transaction exists yet, create a temporary one.
+    auto update = weak_from_this().lock();
+    ICEBERG_PRECHECK(update != nullptr, "PendingUpdate must be owned by std::shared_ptr");
+
     ICEBERG_ASSIGN_OR_RAISE(auto txn, Transaction::Make(ctx_));
+    ICEBERG_RETURN_UNEXPECTED(txn->AddUpdate(update));
+
     auto apply_status = txn->Apply(*this);
     if (!apply_status.has_value()) {
-      std::ignore = Finalize(std::unexpected(apply_status.error()));
+      txn->FinalizeUpdates(std::unexpected(apply_status.error()));
       return apply_status;
     }
 
     auto commit_result = txn->Commit();
-    if (!commit_result.has_value()) {
-      std::ignore = Finalize(std::unexpected(commit_result.error()));
-      return std::unexpected(commit_result.error());
-    }
-
-    std::ignore = Finalize(commit_result.value()->metadata().get());
+    ICEBERG_RETURN_UNEXPECTED(commit_result);
     return {};
   }
+
   auto txn = ctx_->transaction->lock();
   if (!txn) {
     return CommitFailed("Transaction has been destroyed");
   }
-  return txn->Apply(*this);
+
+  auto apply_status = txn->Apply(*this);
+  if (!apply_status.has_value()) {
+    txn->FinalizeUpdates(std::unexpected(apply_status.error()));
+  }
+  return apply_status;
 }
 
 Status PendingUpdate::Finalize(

src/iceberg/update/pending_update.h

@@ -38,7 +38,8 @@ namespace iceberg {
 ///
 /// \note Implementations are expected to use builder pattern and errors
 /// should be handled by the ErrorCollector base class.
-class ICEBERG_EXPORT PendingUpdate : public ErrorCollector {
+class ICEBERG_EXPORT PendingUpdate : public ErrorCollector,
+                                     public std::enable_shared_from_this<PendingUpdate> {
  public:
   enum class Kind : uint8_t {
     kExpireSnapshots,
@@ -66,6 +67,7 @@ class ICEBERG_EXPORT PendingUpdate : public ErrorCollector {
   ///         - ValidationFailed: if it cannot be applied to the current table metadata.
   ///         - CommitFailed: if it cannot be committed due to conflicts.
   ///         - CommitStateUnknown: unknown status, no cleanup should be done.
+  /// \note The update must be owned by a `std::shared_ptr` before calling Commit().
   virtual Status Commit();
 
   /// \brief Finalize the pending update.