From 7f505eb5a4adb8d02c95bd1b978e64df4cc4d5f6 Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Mon, 17 Aug 2026 08:57:56 +0000 Subject: [PATCH 1/3] feat(logging): log transaction commit retries and final outcome Transaction::Commit runs through a retry runner but was completely silent, so operators could not tell whether a commit was retrying on a transient conflict or had failed permanently. This is the first real adoption of the logging component in the commit path. - WARN on each genuine retry (the runner only re-invokes the task when it decides to retry, so attempt > 1 marks a real retry), carrying the prior error. - INFO when a commit finally succeeds after > 1 attempt. - ERROR when retries are exhausted, with the attempt count and final error. Tests (TransactionRetryTest, via a CapturingLogger installed with ScopedDefaultLogger): assert the retry WARN + success INFO on a retry-then-succeed commit, and the exhaustion ERROR on an always-conflicting commit. Co-authored-by: Isaac --- src/iceberg/test/transaction_test.cc | 85 ++++++++++++++++++++++++++++ src/iceberg/transaction.cc | 26 ++++++++- 2 files changed, 110 insertions(+), 1 deletion(-) diff --git a/src/iceberg/test/transaction_test.cc b/src/iceberg/test/transaction_test.cc index 3a13b7bc5..46a645bd2 100644 --- a/src/iceberg/test/transaction_test.cc +++ b/src/iceberg/test/transaction_test.cc @@ -21,7 +21,9 @@ #include "iceberg/expression/expressions.h" #include "iceberg/expression/term.h" +#include "iceberg/logging/log_level.h" #include "iceberg/sort_order.h" +#include "iceberg/test/logging_test_helpers.h" #include "iceberg/test/matchers.h" #include "iceberg/test/mock_catalog.h" #include "iceberg/test/update_test_base.h" @@ -173,6 +175,89 @@ TEST_F(TransactionRetryTest, CommitRetryExhausted) { EXPECT_EQ(update_call_count, 5); } +namespace { +// True if any captured record has the given level and a message containing `needle`. +bool HasRecord(const std::vector& records, LogLevel level, + std::string_view needle) { + for (const auto& record : records) { + if (record.level == level && record.message.find(needle) != std::string::npos) { + return true; + } + } + return false; +} +} // namespace + +// A commit that succeeds after one retryable conflict emits a WARN for the retry +// (carrying the prior error) and an INFO for the eventual success. +TEST_F(TransactionRetryTest, CommitRetryEmitsRetryAndSuccessLogs) { + auto capturing = std::make_shared(); + capturing->SetLevel(LogLevel::kTrace); + ScopedDefaultLogger guard(capturing); + + int update_call_count = 0; + ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_)) + .WillByDefault([this, &update_call_count]( + const TableIdentifier&, + const std::vector>&, + const std::vector>&) + -> Result> { + ++update_call_count; + if (update_call_count == 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_); + }); + + ICEBERG_UNWRAP_OR_FAIL(auto txn, mock_table_->NewTransaction()); + ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewUpdateProperties()); + update->Set("retry.test", "value"); + EXPECT_THAT(update->Commit(), IsOk()); + EXPECT_THAT(txn->Commit(), IsOk()); + + auto records = capturing->records(); + EXPECT_TRUE( + HasRecord(records, LogLevel::kWarn, "Retrying transaction commit (attempt 2)")) + << "expected a retry WARN"; + EXPECT_TRUE(HasRecord(records, LogLevel::kWarn, "conflict on first attempt")) + << "retry WARN should carry the prior error"; + EXPECT_TRUE(HasRecord(records, LogLevel::kInfo, "succeeded after 2 attempts")) + << "expected a success INFO"; +} + +// A commit that exhausts its retries emits an ERROR with the attempt count and the +// final error. +TEST_F(TransactionRetryTest, CommitRetryExhaustedEmitsErrorLog) { + auto capturing = std::make_shared(); + capturing->SetLevel(LogLevel::kTrace); + ScopedDefaultLogger guard(capturing); + + ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_)) + .WillByDefault([](const TableIdentifier&, + const std::vector>&, + const std::vector>&) + -> Result> { + return CommitFailed("always conflicts"); + }); + + ICEBERG_UNWRAP_OR_FAIL(auto txn, mock_table_->NewTransaction()); + ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewUpdateProperties()); + update->Set("retry.test", "value"); + EXPECT_THAT(update->Commit(), IsOk()); + EXPECT_THAT(txn->Commit(), IsError(ErrorKind::kCommitFailed)); + + auto records = capturing->records(); + EXPECT_TRUE(HasRecord(records, LogLevel::kError, "failed after 5 attempt(s)")) + << "expected a final ERROR with the attempt count"; + EXPECT_TRUE(HasRecord(records, LogLevel::kError, "always conflicts")) + << "final ERROR should carry the last error"; + // Retries 2..5 each log a WARN. + EXPECT_TRUE( + HasRecord(records, LogLevel::kWarn, "Retrying transaction commit (attempt 5)")); +} + TEST_F(TransactionRetryTest, CommitNonRetryableErrorStopsImmediately) { 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..f617ce394 100644 --- a/src/iceberg/transaction.cc +++ b/src/iceberg/transaction.cc @@ -20,9 +20,11 @@ #include #include +#include #include "iceberg/catalog.h" #include "iceberg/location_provider.h" +#include "iceberg/logging/log_macros.h" #include "iceberg/schema.h" #include "iceberg/snapshot.h" #include "iceberg/statistics_file.h" @@ -376,14 +378,36 @@ Result> Transaction::Commit() { int32_t total_timeout_ms = props.Get(TableProperties::kCommitTotalRetryTimeMs); bool is_first_attempt = true; + int32_t attempt = 0; + std::string last_error; auto commit_result = MakeCommitRetryRunner(num_retries, min_wait_ms, max_wait_ms, total_timeout_ms) - .Run([this, &is_first_attempt]() -> Result> { + .Run([this, &is_first_attempt, &attempt, + &last_error]() -> Result> { + ++attempt; + // The runner only re-invokes this task when it has decided to retry, so + // attempt > 1 here means a genuine retry after a retryable failure. + if (attempt > 1) { + ICEBERG_LOG_WARN("Retrying transaction commit (attempt {}) after: {}", + attempt, last_error); + } auto result = CommitOnce(is_first_attempt); is_first_attempt = false; + if (!result.has_value()) { + last_error = result.error().message; + } return result; }); + if (commit_result.has_value()) { + if (attempt > 1) { + ICEBERG_LOG_INFO("Transaction commit succeeded after {} attempts", attempt); + } + } else { + ICEBERG_LOG_ERROR("Transaction commit failed after {} attempt(s): {}", attempt, + commit_result.error().message); + } + Result finalize_result = commit_result.has_value() ? Result(commit_result.value()->metadata().get()) From f9cf1290e69e057a54ea0ad50461283c9a8f8585 Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Mon, 17 Aug 2026 09:09:21 +0000 Subject: [PATCH 2/3] feat(logging): log commit success and snapshot additions Extend the commit-path logging beyond retries: - Transaction::Commit now logs an INFO on every successful commit (previously only after a retry). When the commit advanced the current snapshot (a data commit) the message names the snapshot id and operation; metadata-only commits report a plain success. - TableMetadataBuilder::AddSnapshot logs a DEBUG naming the snapshot id and sequence number when a snapshot is added to the metadata. Tests: single-attempt commit emits the success INFO with no retry WARN (TransactionRetryTest.CommitSuccessEmitsInfoLog); AddSnapshot emits the DEBUG (TableMetadataBuilderTest.AddSnapshotEmitsDebugLog). Co-authored-by: Isaac --- src/iceberg/table_metadata.cc | 3 ++ .../test/table_metadata_builder_test.cc | 25 ++++++++++++++++ src/iceberg/test/transaction_test.cc | 30 +++++++++++++++++++ src/iceberg/transaction.cc | 20 ++++++++++++- 4 files changed, 77 insertions(+), 1 deletion(-) diff --git a/src/iceberg/table_metadata.cc b/src/iceberg/table_metadata.cc index 0763c4fe6..e3f650b2f 100644 --- a/src/iceberg/table_metadata.cc +++ b/src/iceberg/table_metadata.cc @@ -38,6 +38,7 @@ #include "iceberg/exception.h" #include "iceberg/file_io.h" #include "iceberg/json_serde_internal.h" +#include "iceberg/logging/log_macros.h" #include "iceberg/metrics_config.h" #include "iceberg/partition_field.h" #include "iceberg/partition_spec.h" @@ -1106,6 +1107,8 @@ Status TableMetadataBuilder::Impl::AddSnapshot(std::shared_ptr snapsho metadata_.next_row_id += add_rows.value(); } + ICEBERG_LOG_DEBUG("Added snapshot {} (sequence number {}) to table metadata", + snapshot->snapshot_id, snapshot->sequence_number); return {}; } diff --git a/src/iceberg/test/table_metadata_builder_test.cc b/src/iceberg/test/table_metadata_builder_test.cc index 0d10722bb..9ebbfc327 100644 --- a/src/iceberg/test/table_metadata_builder_test.cc +++ b/src/iceberg/test/table_metadata_builder_test.cc @@ -24,6 +24,7 @@ #include #include +#include "iceberg/logging/log_level.h" #include "iceberg/partition_spec.h" #include "iceberg/result.h" #include "iceberg/schema.h" @@ -33,6 +34,7 @@ #include "iceberg/table_metadata.h" #include "iceberg/table_properties.h" #include "iceberg/table_update.h" +#include "iceberg/test/logging_test_helpers.h" #include "iceberg/test/matchers.h" #include "iceberg/transform.h" #include "iceberg/type.h" @@ -1185,6 +1187,29 @@ TEST(TableMetadataBuilderTest, RemoveSchemasAfterSchemaChange) { ASSERT_THAT(builder->Build(), HasErrorMessage("Cannot remove current schema: 1")); } +// Adding a snapshot to the builder emits a DEBUG record naming the snapshot. +TEST(TableMetadataBuilderTest, AddSnapshotEmitsDebugLog) { + auto capturing = std::make_shared(); + capturing->SetLevel(LogLevel::kTrace); + ScopedDefaultLogger guard(capturing); + + auto base = CreateBaseMetadata(); + auto builder = TableMetadataBuilder::BuildFrom(base.get()); + builder->AddSnapshot( + std::make_shared(Snapshot{.snapshot_id = 42, .sequence_number = 7})); + ICEBERG_UNWRAP_OR_FAIL(auto metadata, builder->Build()); + + bool found = false; + for (const auto& record : capturing->records()) { + if (record.level == LogLevel::kDebug && + record.message.find("Added snapshot 42") != std::string::npos) { + found = true; + break; + } + } + EXPECT_TRUE(found) << "expected a DEBUG record naming the added snapshot"; +} + TEST(TableMetadataBuilderTest, RemoveSnapshotRef) { auto base = CreateBaseMetadata(); auto builder = TableMetadataBuilder::BuildFrom(base.get()); diff --git a/src/iceberg/test/transaction_test.cc b/src/iceberg/test/transaction_test.cc index 46a645bd2..5da46a8a6 100644 --- a/src/iceberg/test/transaction_test.cc +++ b/src/iceberg/test/transaction_test.cc @@ -258,6 +258,36 @@ TEST_F(TransactionRetryTest, CommitRetryExhaustedEmitsErrorLog) { HasRecord(records, LogLevel::kWarn, "Retrying transaction commit (attempt 5)")); } +// A commit that succeeds on the first attempt emits a plain success INFO (no +// "after N attempts"). This is the single-attempt case that was previously silent. +TEST_F(TransactionRetryTest, CommitSuccessEmitsInfoLog) { + auto capturing = std::make_shared(); + capturing->SetLevel(LogLevel::kTrace); + ScopedDefaultLogger guard(capturing); + + ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_)) + .WillByDefault([this](const TableIdentifier&, + const std::vector>&, + const std::vector>&) + -> Result> { + return Table::Make(mock_table_->name(), mock_table_->metadata(), + std::string(mock_table_->metadata_file_location()), + mock_table_->io(), mock_catalog_); + }); + + ICEBERG_UNWRAP_OR_FAIL(auto txn, mock_table_->NewTransaction()); + ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewUpdateProperties()); + update->Set("retry.test", "value"); + EXPECT_THAT(update->Commit(), IsOk()); + EXPECT_THAT(txn->Commit(), IsOk()); + + auto records = capturing->records(); + EXPECT_TRUE(HasRecord(records, LogLevel::kInfo, "Transaction commit succeeded")) + << "expected a success INFO on a single-attempt commit"; + // No retry happened, so there must be no retry WARN. + EXPECT_FALSE(HasRecord(records, LogLevel::kWarn, "Retrying transaction commit")); +} + TEST_F(TransactionRetryTest, CommitNonRetryableErrorStopsImmediately) { 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 f617ce394..df978908b 100644 --- a/src/iceberg/transaction.cc +++ b/src/iceberg/transaction.cc @@ -377,6 +377,9 @@ Result> Transaction::Commit() { int32_t max_wait_ms = props.Get(TableProperties::kCommitMaxRetryWaitMs); int32_t total_timeout_ms = props.Get(TableProperties::kCommitTotalRetryTimeMs); + // Snapshot id before the commit, to detect whether this commit advanced it (a + // data commit) versus a metadata-only commit that adds no snapshot. + const int64_t base_current_snapshot_id = ctx_->table->metadata()->current_snapshot_id; bool is_first_attempt = true; int32_t attempt = 0; std::string last_error; @@ -400,8 +403,23 @@ Result> Transaction::Commit() { }); if (commit_result.has_value()) { + // Name the resulting snapshot only when this commit produced one (current + // snapshot advanced); metadata-only commits report a plain success. + std::string detail; + if (auto snapshot = commit_result.value()->metadata()->Snapshot(); + snapshot.has_value() && + snapshot.value()->snapshot_id != base_current_snapshot_id) { + const auto& summary = snapshot.value()->summary; + auto op = summary.find(SnapshotSummaryFields::kOperation); + detail = + std::format(": committed snapshot {} (op={})", snapshot.value()->snapshot_id, + op != summary.end() ? op->second : "unknown"); + } if (attempt > 1) { - ICEBERG_LOG_INFO("Transaction commit succeeded after {} attempts", attempt); + ICEBERG_LOG_INFO("Transaction commit succeeded after {} attempts{}", attempt, + detail); + } else { + ICEBERG_LOG_INFO("Transaction commit succeeded{}", detail); } } else { ICEBERG_LOG_ERROR("Transaction commit failed after {} attempt(s): {}", attempt, From f757a00e622ce891b7cc6db46d70deab54d33e4c Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Sat, 5 Sep 2026 05:04:14 +0000 Subject: [PATCH 3/3] fix(logging): address commit lifecycle review feedback --- src/iceberg/table_metadata.cc | 3 - src/iceberg/test/fast_append_test.cc | 27 +++++++ .../test/table_metadata_builder_test.cc | 25 ------ src/iceberg/test/transaction_test.cc | 81 +++++++++++++++++-- src/iceberg/transaction.cc | 45 +++++++---- 5 files changed, 129 insertions(+), 52 deletions(-) diff --git a/src/iceberg/table_metadata.cc b/src/iceberg/table_metadata.cc index e3f650b2f..0763c4fe6 100644 --- a/src/iceberg/table_metadata.cc +++ b/src/iceberg/table_metadata.cc @@ -38,7 +38,6 @@ #include "iceberg/exception.h" #include "iceberg/file_io.h" #include "iceberg/json_serde_internal.h" -#include "iceberg/logging/log_macros.h" #include "iceberg/metrics_config.h" #include "iceberg/partition_field.h" #include "iceberg/partition_spec.h" @@ -1107,8 +1106,6 @@ Status TableMetadataBuilder::Impl::AddSnapshot(std::shared_ptr snapsho metadata_.next_row_id += add_rows.value(); } - ICEBERG_LOG_DEBUG("Added snapshot {} (sequence number {}) to table metadata", - snapshot->snapshot_id, snapshot->sequence_number); return {}; } diff --git a/src/iceberg/test/fast_append_test.cc b/src/iceberg/test/fast_append_test.cc index f88d2e011..120917dd1 100644 --- a/src/iceberg/test/fast_append_test.cc +++ b/src/iceberg/test/fast_append_test.cc @@ -34,6 +34,7 @@ #include "iceberg/avro/avro_register.h" #include "iceberg/constants.h" +#include "iceberg/logging/log_level.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/manifest/manifest_reader.h" #include "iceberg/manifest/manifest_writer.h" @@ -46,6 +47,7 @@ #include "iceberg/table_metadata.h" #include "iceberg/table_properties.h" #include "iceberg/test/executor.h" +#include "iceberg/test/logging_test_helpers.h" #include "iceberg/test/matchers.h" #include "iceberg/test/mock_catalog.h" #include "iceberg/test/update_test_base.h" @@ -178,6 +180,31 @@ TEST_F(FastAppendTest, AppendDataFile) { EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kManifestsReplaced), "0"); } +TEST_F(FastAppendTest, StageOnlyCommitLogNamesAddedSnapshot) { + auto capturing = std::make_shared(); + capturing->SetLevel(LogLevel::kTrace); + ScopedDefaultLogger guard(capturing); + + ICEBERG_UNWRAP_OR_FAIL(auto fast_append, table_->NewFastAppend()); + fast_append->StageOnly(); + fast_append->AppendFile(file_a_); + ASSERT_THAT(fast_append->Commit(), IsOk()); + ASSERT_THAT(table_->Refresh(), IsOk()); + + ASSERT_FALSE(table_->metadata()->snapshots.empty()); + const auto snapshot_id = table_->metadata()->snapshots.back()->snapshot_id; + bool found = false; + for (const auto& record : capturing->records()) { + if (record.level == LogLevel::kInfo && + record.message.find(std::format("committed snapshot {}", snapshot_id)) != + std::string::npos) { + found = true; + break; + } + } + EXPECT_TRUE(found) << "expected the staged snapshot in the commit success log"; +} + TEST_F(FastAppendTest, AppendMultipleDataFiles) { std::shared_ptr fast_append; ICEBERG_UNWRAP_OR_FAIL(fast_append, table_->NewFastAppend()); diff --git a/src/iceberg/test/table_metadata_builder_test.cc b/src/iceberg/test/table_metadata_builder_test.cc index 9ebbfc327..0d10722bb 100644 --- a/src/iceberg/test/table_metadata_builder_test.cc +++ b/src/iceberg/test/table_metadata_builder_test.cc @@ -24,7 +24,6 @@ #include #include -#include "iceberg/logging/log_level.h" #include "iceberg/partition_spec.h" #include "iceberg/result.h" #include "iceberg/schema.h" @@ -34,7 +33,6 @@ #include "iceberg/table_metadata.h" #include "iceberg/table_properties.h" #include "iceberg/table_update.h" -#include "iceberg/test/logging_test_helpers.h" #include "iceberg/test/matchers.h" #include "iceberg/transform.h" #include "iceberg/type.h" @@ -1187,29 +1185,6 @@ TEST(TableMetadataBuilderTest, RemoveSchemasAfterSchemaChange) { ASSERT_THAT(builder->Build(), HasErrorMessage("Cannot remove current schema: 1")); } -// Adding a snapshot to the builder emits a DEBUG record naming the snapshot. -TEST(TableMetadataBuilderTest, AddSnapshotEmitsDebugLog) { - auto capturing = std::make_shared(); - capturing->SetLevel(LogLevel::kTrace); - ScopedDefaultLogger guard(capturing); - - auto base = CreateBaseMetadata(); - auto builder = TableMetadataBuilder::BuildFrom(base.get()); - builder->AddSnapshot( - std::make_shared(Snapshot{.snapshot_id = 42, .sequence_number = 7})); - ICEBERG_UNWRAP_OR_FAIL(auto metadata, builder->Build()); - - bool found = false; - for (const auto& record : capturing->records()) { - if (record.level == LogLevel::kDebug && - record.message.find("Added snapshot 42") != std::string::npos) { - found = true; - break; - } - } - EXPECT_TRUE(found) << "expected a DEBUG record naming the added snapshot"; -} - TEST(TableMetadataBuilderTest, RemoveSnapshotRef) { auto base = CreateBaseMetadata(); auto builder = TableMetadataBuilder::BuildFrom(base.get()); diff --git a/src/iceberg/test/transaction_test.cc b/src/iceberg/test/transaction_test.cc index 5da46a8a6..a8a9f80ca 100644 --- a/src/iceberg/test/transaction_test.cc +++ b/src/iceberg/test/transaction_test.cc @@ -22,7 +22,9 @@ #include "iceberg/expression/expressions.h" #include "iceberg/expression/term.h" #include "iceberg/logging/log_level.h" +#include "iceberg/snapshot.h" #include "iceberg/sort_order.h" +#include "iceberg/table_metadata.h" #include "iceberg/test/logging_test_helpers.h" #include "iceberg/test/matchers.h" #include "iceberg/test/mock_catalog.h" @@ -227,9 +229,68 @@ TEST_F(TransactionRetryTest, CommitRetryEmitsRetryAndSuccessLogs) { << "expected a success INFO"; } -// A commit that exhausts its retries emits an ERROR with the attempt count and the -// final error. -TEST_F(TransactionRetryTest, CommitRetryExhaustedEmitsErrorLog) { +// A metadata-only retry must not attribute a snapshot committed concurrently by +// another writer to this transaction. +TEST_F(TransactionRetryTest, MetadataOnlyRetryDoesNotLogConcurrentSnapshot) { + auto capturing = std::make_shared(); + capturing->SetLevel(LogLevel::kTrace); + ScopedDefaultLogger guard(capturing); + + constexpr int64_t kConcurrentSnapshotId = 987654321; + auto metadata_builder = TableMetadataBuilder::BuildFrom(mock_table_->metadata().get()); + auto concurrent_snapshot = std::make_shared(Snapshot{ + .snapshot_id = kConcurrentSnapshotId, + .parent_snapshot_id = mock_table_->metadata()->current_snapshot_id, + .sequence_number = mock_table_->metadata()->last_sequence_number + 1, + .timestamp_ms = TimePointMs{}, + .manifest_list = "concurrent-manifest-list.avro", + .summary = {{SnapshotSummaryFields::kOperation, "append"}}, + }); + metadata_builder->SetBranchSnapshot(concurrent_snapshot, + std::string(SnapshotRef::kMainBranch)); + ICEBERG_UNWRAP_OR_FAIL(auto concurrent_metadata, metadata_builder->Build()); + auto concurrent_metadata_ptr = + std::shared_ptr(std::move(concurrent_metadata)); + const std::string concurrent_metadata_location = "concurrent.metadata.json"; + + ON_CALL(*mock_catalog_, LoadTable(::testing::_)) + .WillByDefault([this, concurrent_metadata_ptr, &concurrent_metadata_location]( + const TableIdentifier&) -> Result> { + return Table::Make(mock_table_->name(), concurrent_metadata_ptr, + concurrent_metadata_location, mock_table_->io(), + mock_catalog_); + }); + + int update_call_count = 0; + ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_)) + .WillByDefault( + [this, concurrent_metadata_ptr, &concurrent_metadata_location, + &update_call_count](const TableIdentifier&, + const std::vector>&, + const std::vector>&) + -> Result> { + if (++update_call_count == 1) { + return CommitFailed("conflict on first attempt"); + } + return Table::Make(mock_table_->name(), concurrent_metadata_ptr, + concurrent_metadata_location, mock_table_->io(), + mock_catalog_); + }); + + ICEBERG_UNWRAP_OR_FAIL(auto txn, mock_table_->NewTransaction()); + ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewUpdateProperties()); + update->Set("retry.test", "value"); + ASSERT_THAT(update->Commit(), IsOk()); + ASSERT_THAT(txn->Commit(), IsOk()); + + EXPECT_FALSE(HasRecord(capturing->records(), LogLevel::kInfo, + std::to_string(kConcurrentSnapshotId))) + << "metadata-only commit attributed the concurrent snapshot to itself"; +} + +// A commit that exhausts its retries returns the final error without emitting a +// generic ERROR log. Genuine retry attempts still emit WARN records. +TEST_F(TransactionRetryTest, CommitRetryExhaustedDoesNotEmitErrorLog) { auto capturing = std::make_shared(); capturing->SetLevel(LogLevel::kTrace); ScopedDefaultLogger guard(capturing); @@ -249,10 +310,8 @@ TEST_F(TransactionRetryTest, CommitRetryExhaustedEmitsErrorLog) { EXPECT_THAT(txn->Commit(), IsError(ErrorKind::kCommitFailed)); auto records = capturing->records(); - EXPECT_TRUE(HasRecord(records, LogLevel::kError, "failed after 5 attempt(s)")) - << "expected a final ERROR with the attempt count"; - EXPECT_TRUE(HasRecord(records, LogLevel::kError, "always conflicts")) - << "final ERROR should carry the last error"; + EXPECT_FALSE(HasRecord(records, LogLevel::kError, "")) + << "the final commit error should be propagated without a generic ERROR log"; // Retries 2..5 each log a WARN. EXPECT_TRUE( HasRecord(records, LogLevel::kWarn, "Retrying transaction commit (attempt 5)")); @@ -288,7 +347,11 @@ TEST_F(TransactionRetryTest, CommitSuccessEmitsInfoLog) { EXPECT_FALSE(HasRecord(records, LogLevel::kWarn, "Retrying transaction commit")); } -TEST_F(TransactionRetryTest, CommitNonRetryableErrorStopsImmediately) { +TEST_F(TransactionRetryTest, CommitStateUnknownStopsImmediatelyWithoutErrorLog) { + auto capturing = std::make_shared(); + capturing->SetLevel(LogLevel::kTrace); + ScopedDefaultLogger guard(capturing); + int update_call_count = 0; ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_)) .WillByDefault( @@ -308,6 +371,8 @@ TEST_F(TransactionRetryTest, CommitNonRetryableErrorStopsImmediately) { auto result = txn->Commit(); EXPECT_THAT(result, IsError(ErrorKind::kCommitStateUnknown)); EXPECT_EQ(update_call_count, 1); // Should not retry + EXPECT_FALSE(HasRecord(capturing->records(), LogLevel::kError, "")) + << "an unknown commit state must not be logged as a confirmed failure"; } TEST_F(TransactionRetryTest, CreateTransactionDoesNotRetry) { diff --git a/src/iceberg/transaction.cc b/src/iceberg/transaction.cc index df978908b..b722197ae 100644 --- a/src/iceberg/transaction.cc +++ b/src/iceberg/transaction.cc @@ -19,6 +19,7 @@ #include "iceberg/transaction.h" #include +#include #include #include @@ -377,9 +378,6 @@ Result> Transaction::Commit() { int32_t max_wait_ms = props.Get(TableProperties::kCommitMaxRetryWaitMs); int32_t total_timeout_ms = props.Get(TableProperties::kCommitTotalRetryTimeMs); - // Snapshot id before the commit, to detect whether this commit advanced it (a - // data commit) versus a metadata-only commit that adds no snapshot. - const int64_t base_current_snapshot_id = ctx_->table->metadata()->current_snapshot_id; bool is_first_attempt = true; int32_t attempt = 0; std::string last_error; @@ -403,17 +401,35 @@ Result> Transaction::Commit() { }); if (commit_result.has_value()) { - // Name the resulting snapshot only when this commit produced one (current - // snapshot advanced); metadata-only commits report a plain success. + // The builder contains only changes made by the successful attempt. Inspecting + // AddSnapshot changes avoids attributing a concurrent writer's snapshot to this + // transaction and also detects snapshots committed with StageOnly or ToBranch. std::string detail; - if (auto snapshot = commit_result.value()->metadata()->Snapshot(); - snapshot.has_value() && - snapshot.value()->snapshot_id != base_current_snapshot_id) { - const auto& summary = snapshot.value()->summary; - auto op = summary.find(SnapshotSummaryFields::kOperation); - detail = - std::format(": committed snapshot {} (op={})", snapshot.value()->snapshot_id, - op != summary.end() ? op->second : "unknown"); + const auto& changes = ctx_->metadata_builder->changes(); + size_t added_snapshot_count = 0; + for (const auto& change : changes) { + added_snapshot_count += change->kind() == TableUpdate::Kind::kAddSnapshot; + } + if (added_snapshot_count > 0) { + detail.reserve(32 + added_snapshot_count * 48); + std::format_to(std::back_inserter(detail), ": committed snapshot{} ", + added_snapshot_count == 1 ? "" : "s"); + + size_t appended_snapshot_count = 0; + for (const auto& change : changes) { + if (change->kind() != TableUpdate::Kind::kAddSnapshot) { + continue; + } + const auto& snapshot = + internal::checked_cast(*change).snapshot(); + if (appended_snapshot_count++ > 0) { + detail += ", "; + } + const auto& summary = snapshot->summary; + auto op = summary.find(SnapshotSummaryFields::kOperation); + std::format_to(std::back_inserter(detail), "{} (op={})", snapshot->snapshot_id, + op != summary.end() ? op->second : "unknown"); + } } if (attempt > 1) { ICEBERG_LOG_INFO("Transaction commit succeeded after {} attempts{}", attempt, @@ -421,9 +437,6 @@ Result> Transaction::Commit() { } else { ICEBERG_LOG_INFO("Transaction commit succeeded{}", detail); } - } else { - ICEBERG_LOG_ERROR("Transaction commit failed after {} attempt(s): {}", attempt, - commit_result.error().message); } Result finalize_result =