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 f88d2e011..15cc78c40 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 6684cf221..d0819c6f3 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 86f4f20a2..b8dfa7a0f 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 3a13b7bc5..0998beacd 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 80d39c8a8..8004e0974 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 007b1057e..bb73a9b1b 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 7e567830e..18756ff48 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 4167387c7..3e21c67c1 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 f5e1ceaba..c28c61cdd 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 70cd7b8e7..fd6cc35ae 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 cd4f4acbb..18503b3c1 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 4b3000652..70388e198 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 19998ddb3..dc9e705aa 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 3d224375c..4487751bf 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 40048ba5c..ee57b6124 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 b7fc048d3..3c743c01d 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 ce219c3ae..283091c13 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 dd3f50c58..f239ea495 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 bf58a048c..f6faf8c92 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.