From 3f61d7f290e16d28191d87b063ee13ff04356dc3 Mon Sep 17 00:00:00 2001 From: Junwang Zhao Date: Sun, 2 Aug 2026 23:26:46 +0800 Subject: [PATCH] fix: enforce pending update lifecycle Pending updates need deterministic ownership and finalization across both standalone commits and explicit transactions. Register updates through shared ownership so transactions can retain and finalize them safely, and scope the temporary transaction binding used by standalone commits so it never escapes through TransactionContext. Finalize no-op commits and failed applies so every update reaches a terminal state and staged files are cleaned even when the caller never commits the transaction. During Transaction::Commit retries, defer eager finalization while updates are being reapplied; otherwise a retryable validation error would finalize the transaction and destroy staged state before RetryRunner can retry. Restore the retry lifecycle marker with RAII when commit exits or throws. Convert snapshot update factories and test helpers to shared_ptr to satisfy the ownership contract. Tests cover detached temporary transactions, cleared standalone bindings, rejection of unshared standalone updates, no-op and apply-failure finalization, staged-file cleanup, standalone retry reapplication, and lifecycle restoration after commit exceptions. --- src/iceberg/test/fast_append_test.cc | 20 +++ .../test/merging_snapshot_update_test.cc | 14 +- src/iceberg/test/replace_partitions_test.cc | 2 +- src/iceberg/test/transaction_test.cc | 131 ++++++++++++++++++ src/iceberg/transaction.cc | 79 +++++++---- src/iceberg/transaction.h | 13 +- src/iceberg/update/delete_files.cc | 4 +- src/iceberg/update/delete_files.h | 2 +- src/iceberg/update/fast_append.cc | 4 +- src/iceberg/update/fast_append.h | 2 +- src/iceberg/update/merge_append.cc | 4 +- src/iceberg/update/merge_append.h | 2 +- src/iceberg/update/pending_update.cc | 46 ++++-- src/iceberg/update/pending_update.h | 4 +- src/iceberg/update/replace_partitions.cc | 4 +- src/iceberg/update/replace_partitions.h | 2 +- src/iceberg/update/rewrite_files.cc | 4 +- src/iceberg/update/rewrite_files.h | 4 +- src/iceberg/update/row_delta.cc | 4 +- src/iceberg/update/row_delta.h | 2 +- 20 files changed, 282 insertions(+), 65 deletions(-) diff --git a/src/iceberg/test/fast_append_test.cc b/src/iceberg/test/fast_append_test.cc index f88d2e0118..15cc78c40d 100644 --- 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 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(*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 index 6684cf221c..d0819c6f39 100644 --- 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> Make(std::string table_name, + static Result> Make(std::string table_name, std::shared_ptr table) { ICEBERG_ASSIGN_OR_RAISE( auto ctx, TransactionContext::Make(std::move(table), TransactionKind::kUpdate)); - return std::unique_ptr( + return std::shared_ptr( new TestMergeAppend(std::move(table_name), std::move(ctx))); } @@ -256,11 +256,11 @@ class TestMergeAppend : public MergingSnapshotUpdate { class TestOverwriteUpdate : public MergingSnapshotUpdate { public: - static Result> Make(std::string table_name, + static Result> Make(std::string table_name, std::shared_ptr
table) { ICEBERG_ASSIGN_OR_RAISE( auto ctx, TransactionContext::Make(std::move(table), TransactionKind::kUpdate)); - return std::unique_ptr( + return std::shared_ptr( new TestOverwriteUpdate(std::move(table_name), std::move(ctx))); } @@ -336,11 +336,11 @@ class MergingSnapshotUpdateTest : public MinimalUpdateTestBase { return f; } - Result> NewMergeAppend() { + Result> NewMergeAppend() { return TestMergeAppend::Make(TableName(), table_); } - Result> NewOverwriteUpdate() { + Result> NewOverwriteUpdate() { return TestOverwriteUpdate::Make(TableName(), table_); } @@ -983,7 +983,7 @@ class MergingSnapshotUpdateV1Test : public UpdateTestBase { return f; } - Result> NewMergeAppend() { + Result> NewMergeAppend() { return TestMergeAppend::Make(TableName(), table_); } diff --git a/src/iceberg/test/replace_partitions_test.cc b/src/iceberg/test/replace_partitions_test.cc index 86f4f20a21..b8dfa7a0f7 100644 --- 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> NewReplace() { + Result> 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 index 3a13b7bc5d..0998beacd7 100644 --- a/src/iceberg/test/transaction_test.cc +++ b/src/iceberg/test/transaction_test.cc @@ -19,20 +19,38 @@ #include "iceberg/transaction.h" +#include +#include +#include +#include +#include + #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 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(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 update_counts; + ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_)) + .WillByDefault( + [this, &update_counts](const TableIdentifier&, + const std::vector>&, + const std::vector>& updates) + -> Result> { + 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> { + 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(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>&, + const std::vector>&) + -> Result> { + ++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 index 80d39c8a89..8004e0974e 100644 --- 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> Transaction::Make(std::shared_ptr
ta Result> Transaction::Make( std::shared_ptr ctx) { ICEBERG_PRECHECK(ctx != nullptr, "TransactionContext cannot be null"); - auto txn = std::shared_ptr(new Transaction(ctx)); - ctx->transaction = std::weak_ptr(txn); - return txn; + return std::shared_ptr(new Transaction(std::move(ctx))); } const std::shared_ptr
& 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& 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& 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> 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> commit_result = ctx_->table; + if (!updates.empty()) { + const auto& props = ctx_->table->properties(); + int32_t num_retries = + CanRetry() ? static_cast(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> { + 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(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> { - auto result = CommitOnce(is_first_attempt); - is_first_attempt = false; - return result; - }); - Result finalize_result = commit_result.has_value() ? Result(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> Transaction::Commit() { return ctx_->table; } +void Transaction::FinalizeUpdates(const Result& commit_result) { + if (finalized_) { + return; + } + finalized_ = true; + for (const auto& update : pending_updates_) { + std::ignore = update->Finalize(commit_result); + } +} + Result> Transaction::CommitOnce(bool is_first_attempt) { std::vector> requirements; diff --git a/src/iceberg/transaction.h b/src/iceberg/transaction.h index 007b1057e7..bb73a9b1b1 100644 --- a/src/iceberg/transaction.h +++ b/src/iceberg/transaction.h @@ -50,8 +50,8 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this> Make( std::shared_ptr ctx); @@ -166,6 +166,9 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this& commit_result); + private: friend class PendingUpdate; @@ -178,6 +181,12 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this> DeleteFiles::Make( +Result> DeleteFiles::Make( std::string table_name, std::shared_ptr ctx) { ICEBERG_PRECHECK(!table_name.empty(), "Table name cannot be empty"); ICEBERG_PRECHECK(ctx != nullptr, "Cannot create DeleteFiles without a context"); - return std::unique_ptr( + return std::shared_ptr( new DeleteFiles(std::move(table_name), std::move(ctx))); } diff --git a/src/iceberg/update/delete_files.h b/src/iceberg/update/delete_files.h index 7e567830e5..18756ff486 100644 --- a/src/iceberg/update/delete_files.h +++ b/src/iceberg/update/delete_files.h @@ -43,7 +43,7 @@ namespace iceberg { /// differently-normalized URIs are not considered matches. class ICEBERG_EXPORT DeleteFiles : public MergingSnapshotUpdate { public: - static Result> Make( + static Result> Make( std::string table_name, std::shared_ptr ctx); /// \brief Delete a file by path from the underlying table. diff --git a/src/iceberg/update/fast_append.cc b/src/iceberg/update/fast_append.cc index 4167387c7e..3e21c67c18 100644 --- a/src/iceberg/update/fast_append.cc +++ b/src/iceberg/update/fast_append.cc @@ -35,11 +35,11 @@ namespace iceberg { -Result> FastAppend::Make( +Result> FastAppend::Make( std::string table_name, std::shared_ptr ctx) { ICEBERG_PRECHECK(!table_name.empty(), "Table name cannot be empty"); ICEBERG_PRECHECK(ctx != nullptr, "Cannot create FastAppend without a context"); - return std::unique_ptr( + return std::shared_ptr( new FastAppend(std::move(table_name), std::move(ctx))); } diff --git a/src/iceberg/update/fast_append.h b/src/iceberg/update/fast_append.h index f5e1ceabae..c28c61cddc 100644 --- a/src/iceberg/update/fast_append.h +++ b/src/iceberg/update/fast_append.h @@ -51,7 +51,7 @@ class ICEBERG_EXPORT FastAppend : public SnapshotUpdate { /// \param table_name The name of the table /// \param ctx The transaction context to use for this update /// \return A Result containing the FastAppend instance or an error - static Result> Make( + static Result> Make( std::string table_name, std::shared_ptr ctx); /// \brief Append a DataFile to the table. diff --git a/src/iceberg/update/merge_append.cc b/src/iceberg/update/merge_append.cc index 70cd7b8e7a..fd6cc35ae0 100644 --- a/src/iceberg/update/merge_append.cc +++ b/src/iceberg/update/merge_append.cc @@ -29,11 +29,11 @@ namespace iceberg { -Result> MergeAppend::Make( +Result> MergeAppend::Make( std::string table_name, std::shared_ptr ctx) { ICEBERG_PRECHECK(!table_name.empty(), "Table name cannot be empty"); ICEBERG_PRECHECK(ctx != nullptr, "Cannot create MergeAppend without a context"); - return std::unique_ptr( + return std::shared_ptr( new MergeAppend(std::move(table_name), std::move(ctx))); } diff --git a/src/iceberg/update/merge_append.h b/src/iceberg/update/merge_append.h index cd4f4acbba..18503b3c17 100644 --- a/src/iceberg/update/merge_append.h +++ b/src/iceberg/update/merge_append.h @@ -47,7 +47,7 @@ class ICEBERG_EXPORT MergeAppend : public MergingSnapshotUpdate { /// \param table_name The name of the table /// \param ctx The transaction context to use for this update /// \return A Result containing the MergeAppend instance or an error - static Result> Make( + static Result> Make( std::string table_name, std::shared_ptr ctx); /// \brief Append a DataFile to the table. diff --git a/src/iceberg/update/pending_update.cc b/src/iceberg/update/pending_update.cc index 4b30006524..70388e1988 100644 --- a/src/iceberg/update/pending_update.cc +++ b/src/iceberg/update/pending_update.cc @@ -25,6 +25,23 @@ #include "iceberg/util/macros.h" namespace iceberg { +namespace { + +class ScopedTransactionBinding { + public: + ScopedTransactionBinding(TransactionContext& ctx, + const std::shared_ptr& txn) + : ctx_(ctx) { + ctx_.transaction = txn; + } + + ~ScopedTransactionBinding() { ctx_.transaction.reset(); } + + private: + TransactionContext& ctx_; +}; + +} // namespace PendingUpdate::PendingUpdate(std::shared_ptr ctx) : ctx_(std::move(ctx)) {} @@ -35,26 +52,39 @@ 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_)); + auto self = weak_from_this().lock(); + ICEBERG_PRECHECK(self != nullptr, "PendingUpdate must be owned by std::shared_ptr"); + ICEBERG_RETURN_UNEXPECTED(txn->AddUpdate(self)); + // Keep Transaction::Make(ctx_) detached, but expose this live transaction while + // Commit() runs so an internal retry can reapply through update->Commit(). + ScopedTransactionBinding binding(*ctx_, txn); + 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->committing_) { + // Finalize eagerly so a failed update cleans up its staged files even if the + // caller never commits the transaction. When the transaction is mid-commit, + // leave finalization to Transaction::Commit(): the failure may be retryable + // (e.g. RetryableValidationFailed from a stale sequence number), and + // finalizing here would destroy staged state before the retry runs. + txn->FinalizeUpdates(std::unexpected(apply_status.error())); + } + return apply_status; } Status PendingUpdate::Finalize( diff --git a/src/iceberg/update/pending_update.h b/src/iceberg/update/pending_update.h index 19998ddb3c..dc9e705aa5 100644 --- a/src/iceberg/update/pending_update.h +++ b/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 { 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. diff --git a/src/iceberg/update/replace_partitions.cc b/src/iceberg/update/replace_partitions.cc index 3d224375ce..4487751bf1 100644 --- a/src/iceberg/update/replace_partitions.cc +++ b/src/iceberg/update/replace_partitions.cc @@ -32,11 +32,11 @@ namespace iceberg { -Result> ReplacePartitions::Make( +Result> ReplacePartitions::Make( std::string table_name, std::shared_ptr ctx) { ICEBERG_PRECHECK(!table_name.empty(), "Table name cannot be empty"); ICEBERG_PRECHECK(ctx != nullptr, "Cannot create ReplacePartitions without a context"); - return std::unique_ptr( + return std::shared_ptr( new ReplacePartitions(std::move(table_name), std::move(ctx))); } diff --git a/src/iceberg/update/replace_partitions.h b/src/iceberg/update/replace_partitions.h index 40048ba5c3..ee57b61247 100644 --- a/src/iceberg/update/replace_partitions.h +++ b/src/iceberg/update/replace_partitions.h @@ -62,7 +62,7 @@ class ICEBERG_EXPORT ReplacePartitions : public MergingSnapshotUpdate { /// \param table_name The name of the table /// \param ctx The transaction context /// \return A Result containing the ReplacePartitions instance or an error - static Result> Make( + static Result> Make( std::string table_name, std::shared_ptr ctx); /// \brief Add a data file to the table. diff --git a/src/iceberg/update/rewrite_files.cc b/src/iceberg/update/rewrite_files.cc index b7fc048d35..3c743c01d8 100644 --- a/src/iceberg/update/rewrite_files.cc +++ b/src/iceberg/update/rewrite_files.cc @@ -39,11 +39,11 @@ RewriteFiles::RewriteFiles(std::string table_name, FailMissingDeletePaths(); } -Result> RewriteFiles::Make( +Result> RewriteFiles::Make( std::string table_name, std::shared_ptr ctx) { ICEBERG_PRECHECK(!table_name.empty(), "Table name cannot be empty"); ICEBERG_PRECHECK(ctx != nullptr, "Cannot create RewriteFiles without a context"); - return std::unique_ptr( + return std::shared_ptr( new RewriteFiles(std::move(table_name), std::move(ctx))); } diff --git a/src/iceberg/update/rewrite_files.h b/src/iceberg/update/rewrite_files.h index ce219c3aec..283091c131 100644 --- a/src/iceberg/update/rewrite_files.h +++ b/src/iceberg/update/rewrite_files.h @@ -54,8 +54,8 @@ class ICEBERG_EXPORT RewriteFiles : public MergingSnapshotUpdate { /// /// \param table_name The name of the table /// \param ctx The transaction context - /// \return A unique pointer to the new RewriteFiles operation - static Result> Make( + /// \return A shared pointer to the new RewriteFiles operation + static Result> Make( std::string table_name, std::shared_ptr ctx); ~RewriteFiles() override = default; diff --git a/src/iceberg/update/row_delta.cc b/src/iceberg/update/row_delta.cc index dd3f50c583..f239ea495d 100644 --- a/src/iceberg/update/row_delta.cc +++ b/src/iceberg/update/row_delta.cc @@ -38,11 +38,11 @@ namespace iceberg { -Result> RowDelta::Make( +Result> RowDelta::Make( std::string table_name, std::shared_ptr ctx) { ICEBERG_PRECHECK(!table_name.empty(), "Table name cannot be empty"); ICEBERG_PRECHECK(ctx != nullptr, "Cannot create RowDelta without a context"); - return std::unique_ptr(new RowDelta(std::move(table_name), std::move(ctx))); + return std::shared_ptr(new RowDelta(std::move(table_name), std::move(ctx))); } RowDelta::RowDelta(std::string table_name, std::shared_ptr ctx) diff --git a/src/iceberg/update/row_delta.h b/src/iceberg/update/row_delta.h index bf58a048c7..f6faf8c92e 100644 --- a/src/iceberg/update/row_delta.h +++ b/src/iceberg/update/row_delta.h @@ -47,7 +47,7 @@ namespace iceberg { class ICEBERG_EXPORT RowDelta : public MergingSnapshotUpdate { public: /// \brief Create a new RowDelta instance. - static Result> Make(std::string table_name, + static Result> Make(std::string table_name, std::shared_ptr ctx); /// \brief Add a DataFile to the table.