apache · zhjwpku · Aug 2, 2026
diff --git a/src/iceberg/test/fast_append_test.cc b/src/iceberg/test/fast_append_test.cc
@@ -315,6 +315,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";

diff --git a/src/iceberg/test/merging_snapshot_update_test.cc b/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_);
   }
 

diff --git a/src/iceberg/test/replace_partitions_test.cc b/src/iceberg/test/replace_partitions_test.cc
@@ -216,7 +216,7 @@ class ReplacePartitionsTest : public UpdateTestBase {
     return paths;
   }
 
-  Result<std::unique_ptr<ReplacePartitions>> NewReplace() {
+  Result<std::shared_ptr<ReplacePartitions>> NewReplace() {
     ICEBERG_ASSIGN_OR_RAISE(auto ctx,
                             TransactionContext::Make(table_, TransactionKind::kUpdate));
     return ReplacePartitions::Make(TableName(), std::move(ctx));

diff --git a/src/iceberg/test/transaction_test.cc b/src/iceberg/test/transaction_test.cc
@@ -19,20 +19,38 @@
 
 #include "iceberg/transaction.h"
 
+#include <format>
+#include <memory>
+#include <stdexcept>
+#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"
 
 namespace iceberg {
 
+class TestPendingUpdate final : public PendingUpdate {
+ public:
+  explicit TestPendingUpdate(std::shared_ptr<TransactionContext> ctx)
+      : PendingUpdate(std::move(ctx)) {}
+
+  Kind kind() const override { return Kind::kUpdateProperties; }
+  bool IsRetryable() const override { return true; }
+};
+
 class TransactionTest : public UpdateTestBase {};
 
 TEST_F(TransactionTest, CreateTransaction) {
@@ -46,6 +64,58 @@ 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));
+  auto update = std::make_unique<TestPendingUpdate>(ctx);
+
+  EXPECT_THAT(update->Commit(),
+              ::testing::AllOf(
+                  IsError(ErrorKind::kInvalidArgument),
+                  HasErrorMessage("PendingUpdate must be owned by std::shared_ptr")));
+  EXPECT_FALSE(ctx->transaction.has_value());
+}
+
+TEST_F(TransactionTest, StandaloneCommitClearsTemporaryTransactionBinding) {
+  ICEBERG_UNWRAP_OR_FAIL(auto ctx,
+                         TransactionContext::Make(table_, TransactionKind::kUpdate));
+  ICEBERG_UNWRAP_OR_FAIL(auto update, UpdateProperties::Make(ctx));
+  update->Set("standalone.property", "standalone.value");
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_FALSE(ctx->transaction.has_value());
+}
+
+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 +221,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::_))
@@ -195,6 +298,34 @@ TEST_F(TransactionRetryTest, CommitNonRetryableErrorStopsImmediately) {
   EXPECT_EQ(update_call_count, 1);  // Should not retry
 }
 
+TEST_F(TransactionRetryTest, CommitExceptionRestoresLifecycleState) {
+  int update_call_count = 0;
+  ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))
+      .WillByDefault(
+          [&update_call_count](const TableIdentifier&,
+                               const std::vector<std::unique_ptr<TableRequirement>>&,
+                               const std::vector<std::unique_ptr<TableUpdate>>&)
+              -> Result<std::shared_ptr<Table>> {
+            ++update_call_count;
+            throw std::runtime_error("injected catalog failure");
+          });
+
+  ICEBERG_UNWRAP_OR_FAIL(auto txn, mock_table_->NewTransaction());
+  ICEBERG_UNWRAP_OR_FAIL(auto properties, txn->NewUpdateProperties());
+  properties->Set("exception.test", "value");
+  EXPECT_THAT(properties->Commit(), IsOk());
+
+  EXPECT_THROW(std::ignore = txn->Commit(), std::runtime_error);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewFastAppend());
+  append->AppendFile(nullptr);
+  EXPECT_THAT(append->Commit(), IsError(ErrorKind::kValidationFailed));
+  EXPECT_THAT(txn->Commit(),
+              ::testing::AllOf(IsError(ErrorKind::kValidationFailed),
+                               HasErrorMessage("Transaction already finalized")));
+  EXPECT_EQ(update_call_count, 1);
+}
+
 TEST_F(TransactionRetryTest, CreateTransactionDoesNotRetry) {
   int update_call_count = 0;
   ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))

diff --git a/src/iceberg/transaction.cc b/src/iceberg/transaction.cc
@@ -57,6 +57,21 @@
 #include "iceberg/util/retry_util.h"
 
 namespace iceberg {
+namespace {
+
+class ScopedTrue {
+ public:
+  explicit ScopedTrue(bool& value) : value_(value) { value_ = true; }
+  ~ScopedTrue() { value_ = false; }
+
+  ScopedTrue(const ScopedTrue&) = delete;
+  ScopedTrue& operator=(const ScopedTrue&) = delete;
+
+ private:
+  bool& value_;
+};
+
+}  // namespace
 
 // ---------------------------------------------------------------------------
 // TransactionContext
@@ -122,9 +137,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; }
@@ -138,6 +151,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");
 
@@ -147,6 +162,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(
@@ -358,40 +376,37 @@ 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;
+    ScopedTrue committing(committing_);
+    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);
 
@@ -402,6 +417,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;