fix(transaction): re-apply table-created updates on commit retry

src/iceberg/test/transaction_test.cc

@@ -151,6 +151,50 @@ TEST_F(TransactionRetryTest, CommitRetrySucceedsAfterConflict) {
   EXPECT_EQ(update_call_count, 2);
 }
 
+TEST_F(TransactionRetryTest, TableCreatedUpdateIsReappliedOnRetry) {
+  // A concurrent commit moves the metadata location, so the retry's Refresh() swaps
+  // metadata and CommitOnce rebuilds the builder by re-applying registered updates;
+  // with an unchanged location the rebuild never runs and this regression cannot fire.
+  ON_CALL(*mock_catalog_, LoadTable(::testing::_))
+      .WillByDefault([this](const TableIdentifier&) -> Result<std::shared_ptr<Table>> {
+        // A distinct metadata object at a new location: Refresh() must actually swap
+        // metadata, or CommitOnce keeps the stale builder and never re-applies updates.
+        return Table::Make(table_->name(),
+                           std::make_shared<TableMetadata>(*table_->metadata()),
+                           std::string(table_->metadata_file_location()) + "-refreshed",
+                           table_->io(), mock_catalog_);
+      });
+
+  int update_call_count = 0;
+  size_t second_attempt_update_count = 0;
+  ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))
+      .WillByDefault([this, &update_call_count, &second_attempt_update_count](
+                         const TableIdentifier&,
+                         const std::vector<std::unique_ptr<TableRequirement>>&,
+                         const std::vector<std::unique_ptr<TableUpdate>>& updates)
+                         -> Result<std::shared_ptr<Table>> {
+        ++update_call_count;
+        if (update_call_count == 1) {
+          return CommitFailed("conflict on first attempt");
+        }
+        second_attempt_update_count = updates.size();
+        return Table::Make(mock_table_->name(), mock_table_->metadata(),
+                           std::string(mock_table_->metadata_file_location()),
+                           mock_table_->io(), mock_catalog_);
+      });
+
+  // Created from the table, not from a transaction: the temporary transaction inside
+  // PendingUpdate::Commit must re-apply this update when the first attempt conflicts,
+  // instead of committing an empty change set and reporting success.
+  ICEBERG_UNWRAP_OR_FAIL(auto update, mock_table_->NewUpdateProperties());
+  update->Set("retry.test", "value");
+  EXPECT_THAT(update->Commit(), IsOk());
+
+  EXPECT_EQ(update_call_count, 2);
+  EXPECT_GT(second_attempt_update_count, 0)
+      << "retry posted an empty change set: the pending update was dropped";
+}
+
 TEST_F(TransactionRetryTest, CommitRetryExhausted) {
   int update_call_count = 0;
   ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))

src/iceberg/transaction.cc

@@ -358,6 +358,9 @@ Result<std::shared_ptr<Table>> Transaction::Commit() {
 
   const auto& updates = ctx_->metadata_builder->changes();
   if (updates.empty()) {
+    for (const auto& update : pending_updates_) {
+      std::ignore = update->Finalize(ctx_->table->metadata().get());
+    }
     committed_ = true;
     return ctx_->table;
   }

src/iceberg/update/pending_update.cc

@@ -35,19 +35,24 @@ Status PendingUpdate::Commit() {
   if (!ctx_->transaction) {
     // Table-created path: no transaction exists yet, create a temporary one.
     ICEBERG_ASSIGN_OR_RAISE(auto txn, Transaction::Make(ctx_));
+    // Register with the transaction so a commit retry re-applies this update onto the
+    // refreshed metadata; without it a retry rebuilds the metadata with no changes and
+    // reports success while silently dropping the update. The transaction does not
+    // outlive this call, so the non-owning handle is safe.
+    ICEBERG_RETURN_UNEXPECTED(
+        txn->AddUpdate(std::shared_ptr<PendingUpdate>(this, [](PendingUpdate*) {})));
     auto apply_status = txn->Apply(*this);
     if (!apply_status.has_value()) {
       std::ignore = Finalize(std::unexpected(apply_status.error()));
       return apply_status;
     }
 
+    // The transaction finalizes its registered updates, this one included; finalizing
+    // here as well would repeat cleanup and file-deletion callbacks.
     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());
     return {};
   }
   auto txn = ctx_->transaction->lock();