Skip to content
Open
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/iceberg/table_metadata.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -1106,6 +1107,8 @@ Status TableMetadataBuilder::Impl::AddSnapshot(std::shared_ptr<Snapshot> snapsho
metadata_.next_row_id += add_rows.value();
}

ICEBERG_LOG_DEBUG("Added snapshot {} (sequence number {}) to table metadata",
Comment thread
kamcheungting-db marked this conversation as resolved.
Outdated
snapshot->snapshot_id, snapshot->sequence_number);
return {};
}

Expand Down
25 changes: 25 additions & 0 deletions src/iceberg/test/table_metadata_builder_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
#include <gmock/gmock.h>
#include <gtest/gtest.h>

#include "iceberg/logging/log_level.h"
#include "iceberg/partition_spec.h"
#include "iceberg/result.h"
#include "iceberg/schema.h"
Expand All @@ -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"
Expand Down Expand Up @@ -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<CapturingLogger>();
capturing->SetLevel(LogLevel::kTrace);
ScopedDefaultLogger guard(capturing);

auto base = CreateBaseMetadata();
auto builder = TableMetadataBuilder::BuildFrom(base.get());
builder->AddSnapshot(
std::make_shared<Snapshot>(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());
Expand Down
115 changes: 115 additions & 0 deletions src/iceberg/test/transaction_test.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -173,6 +175,119 @@ 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<LogMessage>& 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<CapturingLogger>();
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<std::unique_ptr<TableRequirement>>&,
const std::vector<std::unique_ptr<TableUpdate>>&)
-> Result<std::shared_ptr<Table>> {
++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<CapturingLogger>();
capturing->SetLevel(LogLevel::kTrace);
ScopedDefaultLogger guard(capturing);

ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))
.WillByDefault([](const TableIdentifier&,
const std::vector<std::unique_ptr<TableRequirement>>&,
const std::vector<std::unique_ptr<TableUpdate>>&)
-> Result<std::shared_ptr<Table>> {
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)"));
}

// 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<CapturingLogger>();
capturing->SetLevel(LogLevel::kTrace);
ScopedDefaultLogger guard(capturing);

ON_CALL(*mock_catalog_, UpdateTable(::testing::_, ::testing::_, ::testing::_))
.WillByDefault([this](const TableIdentifier&,
const std::vector<std::unique_ptr<TableRequirement>>&,
const std::vector<std::unique_ptr<TableUpdate>>&)
-> Result<std::shared_ptr<Table>> {
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::_))
Expand Down
44 changes: 43 additions & 1 deletion src/iceberg/transaction.cc
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,11 @@

#include <format>
#include <memory>
#include <string>

#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"
Expand Down Expand Up @@ -375,15 +377,55 @@ Result<std::shared_ptr<Table>> 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;
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>> {
.Run([this, &is_first_attempt, &attempt,
&last_error]() -> Result<std::shared_ptr<Table>> {
++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()) {
// 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) {
Comment thread
kamcheungting-db marked this conversation as resolved.
Outdated
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,
detail);
} else {
ICEBERG_LOG_INFO("Transaction commit succeeded{}", detail);
}
} else {
ICEBERG_LOG_ERROR("Transaction commit failed after {} attempt(s): {}", attempt,
Comment thread
kamcheungting-db marked this conversation as resolved.
Outdated
commit_result.error().message);
}

Result<const TableMetadata*> finalize_result =
commit_result.has_value()
? Result<const TableMetadata*>(commit_result.value()->metadata().get())
Expand Down
Loading