Skip to content

fix: enforce pending update lifecycle #868

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zhjwpku wants to merge 5 commits into
base: main
Choose a base branch
from
Open

fix: enforce pending update lifecycle #868

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d281388
fix: enforce pending update lifecycle
zhjwpku Aug 2, 2026
5e90b57
update misleading docs.
zhjwpku Aug 31, 2026
b508d03
fix: make transaction updates replayable and cleanup-safe
zhjwpku Sep 9, 2026
a990fb6
fix: resolve transaction update CI build failures
zhjwpku Sep 10, 2026
309f4a5
test: use std::array for schema identifier names
zhjwpku Sep 10, 2026
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
148 changes: 117 additions & 31 deletions src/iceberg/test/expire_snapshots_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,9 @@
#include "iceberg/test/executor.h"
#include "iceberg/test/matchers.h"
#include "iceberg/test/update_test_base.h"
#include "iceberg/transaction.h"
#include "iceberg/update/fast_append.h"
#include "iceberg/update/set_snapshot.h"

namespace iceberg {

Expand Down Expand Up @@ -221,23 +224,23 @@ class ExpireSnapshotsCleanupTest : public UpdateTestBase {

TEST_F(ExpireSnapshotsTest, DefaultExpireByAge) {
ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Validate());
EXPECT_EQ(result.snapshot_ids_to_remove.size(), 1);
EXPECT_EQ(result.snapshot_ids_to_remove.at(0), 3051729675574597004);
}

TEST_F(ExpireSnapshotsTest, KeepAll) {
ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
update->RetainLast(2);
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Validate());
EXPECT_TRUE(result.snapshot_ids_to_remove.empty());
EXPECT_TRUE(result.refs_to_remove.empty());
}

TEST_F(ExpireSnapshotsTest, ExpireById) {
ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
update->ExpireSnapshotId(3051729675574597004);
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Validate());
EXPECT_EQ(result.snapshot_ids_to_remove.size(), 1);
EXPECT_EQ(result.snapshot_ids_to_remove.at(0), 3051729675574597004);
}
Expand All @@ -247,7 +250,7 @@ TEST_F(ExpireSnapshotsTest, ExpireByIdOverridesRetainLast) {
update->RetainLast(2);
update->ExpireSnapshotId(3051729675574597004);

ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Validate());
EXPECT_THAT(result.snapshot_ids_to_remove, testing::ElementsAre(3051729675574597004));
}

Expand All @@ -262,7 +265,7 @@ TEST_F(ExpireSnapshotsTest, ExpireOlderThan) {
for (const auto& test_case : test_cases) {
ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
update->ExpireOlderThan(test_case.expire_older_than);
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Validate());
EXPECT_EQ(result.snapshot_ids_to_remove.size(), test_case.expected_num_expired);
}
}
Expand All @@ -286,26 +289,20 @@ TEST_F(ExpireSnapshotsCleanupTest, RetainsUnreferencedSnapshotAtExpireThreshold)
ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
update->ExpireOlderThan(expire_at_ms);

ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Validate());
EXPECT_THAT(result.snapshot_ids_to_remove,
testing::Not(testing::Contains(unreferenced_snapshot_id)));
}

TEST_F(ExpireSnapshotsTest, FinalizeRequiresCommittedMetadata) {
TEST_F(ExpireSnapshotsTest, ValidateDoesNotScheduleCleanup) {
std::vector<std::string> deleted_files;
ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewExpireSnapshots());
update->DeleteWith(
[&deleted_files](const std::string& path) { deleted_files.push_back(path); });

// Apply first so apply_result_ is cached
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Validate());
EXPECT_EQ(result.snapshot_ids_to_remove.size(), 1);

// A successful finalize now requires the committed metadata from the catalog.
auto finalize_status = update->Finalize(static_cast<const TableMetadata*>(nullptr));
EXPECT_THAT(finalize_status, IsError(ErrorKind::kInvalidArgument));
EXPECT_THAT(finalize_status,
HasErrorMessage("Missing committed table metadata for cleanup"));
EXPECT_THAT(txn->Abort(), IsOk());
EXPECT_TRUE(deleted_files.empty());
}

Expand All @@ -316,28 +313,24 @@ TEST_F(ExpireSnapshotsTest, CleanupNoneSkipsDeletion) {
update->DeleteWith(
[&deleted_files](const std::string& path) { deleted_files.push_back(path); });

ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Validate());
EXPECT_EQ(result.snapshot_ids_to_remove.size(), 1);

// With kNone cleanup level, Finalize should skip all file deletion
auto finalize_status = update->Finalize(static_cast<const TableMetadata*>(nullptr));
auto finalize_status = update->Commit();
EXPECT_THAT(finalize_status, IsOk());
EXPECT_TRUE(deleted_files.empty());
}

TEST_F(ExpireSnapshotsTest, FinalizeSkippedOnCommitError) {
TEST_F(ExpireSnapshotsTest, AbortSkipsExpirationDeletion) {
std::vector<std::string> deleted_files;
ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
ICEBERG_UNWRAP_OR_FAIL(auto update, txn->NewExpireSnapshots());
update->DeleteWith(
[&deleted_files](const std::string& path) { deleted_files.push_back(path); });

ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
EXPECT_EQ(result.snapshot_ids_to_remove.size(), 1);

// Simulate a commit failure - Finalize should not delete any files
auto finalize_status = update->Finalize(Result<const TableMetadata*>(std::unexpected(
Error{.kind = ErrorKind::kCommitFailed, .message = "simulated failure"})));
EXPECT_THAT(finalize_status, IsOk());
EXPECT_THAT(update->Commit(), IsOk());
EXPECT_THAT(txn->Abort(), IsOk());
EXPECT_THAT(txn->Abort(), IsOk());
EXPECT_TRUE(deleted_files.empty());
}

Expand All @@ -348,11 +341,11 @@ TEST_F(ExpireSnapshotsTest, FinalizeSkipsWhenNothingExpired) {
update->DeleteWith(
[&deleted_files](const std::string& path) { deleted_files.push_back(path); });

ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
ICEBERG_UNWRAP_OR_FAIL(auto result, update->Validate());
EXPECT_TRUE(result.snapshot_ids_to_remove.empty());

// No snapshots expired, so Finalize should not delete any files
auto finalize_status = update->Finalize(static_cast<const TableMetadata*>(nullptr));
auto finalize_status = update->Commit();
EXPECT_THAT(finalize_status, IsOk());
EXPECT_TRUE(deleted_files.empty());
}
Expand Down Expand Up @@ -1016,4 +1009,97 @@ TEST_F(ExpireSnapshotsCleanupTest, CommitIgnoresMalformedSourceSnapshotIdCleanup
EXPECT_EQ(committed_metadata->snapshots.at(0)->snapshot_id, kCurrentSnapshotId);
}

class ExpirationLifecycleTest
: public ExpireSnapshotsCleanupTest,
public ::testing::WithParamInterface<std::tuple<CleanupLevel, bool, bool, bool>> {};

TEST_P(ExpirationLifecycleTest, LaterReferencesSuppressPhysicalDeletion) {
auto [cleanup_level, custom_delete, reattach, retry] = GetParam();
const auto data_path = table_location_ + "/data/reattached.parquet";
const auto manifest_path = table_location_ + "/metadata/expired.avro";
const auto expired_list = table_location_ + "/metadata/expired-list.avro";
const auto current_list = table_location_ + "/metadata/current-list.avro";
auto data_file = MakeDataFile(data_path);
data_file->partition_spec_id = DefaultSpec()->spec_id();
ASSERT_THAT(file_io_->WriteFile(data_path, "data"), IsOk());
auto manifest = WriteDataManifest(manifest_path, kExpiredSnapshotId,
{MakeEntry(ManifestStatus::kAdded, kExpiredSnapshotId,
kExpiredSequenceNumber, data_file)});
WriteManifestList(expired_list, kExpiredSnapshotId, 0, kExpiredSequenceNumber,
{manifest});
WriteManifestList(current_list, kCurrentSnapshotId, kExpiredSnapshotId,
kCurrentSequenceNumber, {});
RewriteTableWithManifestLists(expired_list, current_list);
if (retry) {
FailCommits(1);
}
ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
ICEBERG_UNWRAP_OR_FAIL(auto expire, txn->NewExpireSnapshots());
expire->ExpireSnapshotId(kExpiredSnapshotId).CleanupLevel(cleanup_level);
int deletes = 0;
if (custom_delete) {
expire->DeleteWith([&](const std::string&) { ++deletes; });
}
ASSERT_THAT(expire->Commit(), IsOk());
if (reattach) {
// FastAppend inherits the conservative default capability from PendingUpdate.
ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewFastAppend());
append->AppendFile(data_file);
ASSERT_THAT(append->Commit(), IsOk());
} else {
ICEBERG_UNWRAP_OR_FAIL(auto noop, txn->NewSetSnapshot());
noop->SetCurrentSnapshot(kCurrentSnapshotId);
ASSERT_THAT(noop->Commit(), IsOk());
}
ASSERT_THAT(txn->Commit(), IsOk());
EXPECT_FALSE(ReloadMetadata()->SnapshotById(kExpiredSnapshotId).has_value());
EXPECT_EQ(deletes, 0);
for (const auto& path : {data_path, manifest_path, expired_list}) {
EXPECT_THAT(file_io_->ReadFile(path, std::nullopt), IsOk());
}
EXPECT_THAT(txn->Commit(), IsError(ErrorKind::kValidationFailed));
EXPECT_THAT(txn->Abort(), IsError(ErrorKind::kValidationFailed));
EXPECT_EQ(deletes, 0);
}

INSTANTIATE_TEST_SUITE_P(
CleanupPolicies, ExpirationLifecycleTest,
::testing::Combine(::testing::Values(CleanupLevel::kAll, CleanupLevel::kMetadataOnly),
::testing::Bool(), ::testing::Bool(), ::testing::Bool()));

TEST_F(ExpireSnapshotsCleanupTest, RetryRecomputesExpirationAgainstRefreshedMetadata) {
const auto expired_list = table_location_ + "/metadata/expire-before-retry.avro";
const auto current_list = table_location_ + "/metadata/current-before-retry.avro";
WriteManifestList(expired_list, kExpiredSnapshotId, 0, kExpiredSequenceNumber, {});
WriteManifestList(current_list, kCurrentSnapshotId, kExpiredSnapshotId,
kCurrentSequenceNumber, {});
RewriteTableWithManifestLists(expired_list, current_list);
auto data_file = MakeDataFile(table_location_ + "/data/concurrent.parquet");
data_file->partition_spec_id = DefaultSpec()->spec_id();
FailCommits(1, [&](int attempt) {
if (attempt == 0) {
ICEBERG_UNWRAP_OR_FAIL(auto latest, catalog_->LoadTable(table_ident_));
ICEBERG_UNWRAP_OR_FAIL(auto append, latest->NewFastAppend());
append->AppendFile(data_file);
ASSERT_THAT(append->Commit(), IsOk());
}
});
ICEBERG_UNWRAP_OR_FAIL(auto expire, table_->NewExpireSnapshots());
expire
->ExpireOlderThan(
(CurrentTimePointMs() + std::chrono::hours(1)).time_since_epoch().count())
.RetainLast(1);
std::vector<std::string> deleted;
expire->DeleteWith([&](const std::string& path) { deleted.push_back(path); });
ICEBERG_UNWRAP_OR_FAIL(auto preview, expire->Validate());
EXPECT_THAT(preview.snapshot_ids_to_remove, ::testing::ElementsAre(kExpiredSnapshotId));
ASSERT_THAT(expire->Commit(), IsOk());
auto metadata = ReloadMetadata();
EXPECT_FALSE(metadata->SnapshotById(kExpiredSnapshotId).has_value());
EXPECT_FALSE(metadata->SnapshotById(kCurrentSnapshotId).has_value());
EXPECT_EQ(metadata->snapshots.size(), 1U);
EXPECT_THAT(deleted, ::testing::Contains(expired_list));
EXPECT_THAT(deleted, ::testing::Contains(current_list));
}

} // namespace iceberg
Loading
Loading