Skip to content
Open
Show file tree
Hide file tree
Changes from all 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 @@ -1099,6 +1100,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",
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) {
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,
commit_result.error().message);
}

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