Skip to content

Commit 9bf1ba1

Browse files
wgtmacwgtmac
authored
feat: retry stale v3 snapshot row-lineage validation (#794)
Add a retryable validation error kind and use it for add-snapshot stale sequence-number and stale first-row-id checks, matching Java Iceberg's RetryableValidationException behavior. Include the new retryable validation kind in commit retry policy, while preserving normal validation failures for mixed/non-retryable builder errors. Add focused v3 row-lineage tests for multi-file assignment, branch commits, retry reassignment, stale snapshot validation, and delete-manifest null first_row_id handling.
1 parent 9fb4a2b commit 9bf1ba1

8 files changed

Lines changed: 256 additions & 17 deletions

‎src/iceberg/result.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,7 @@ enum class ErrorKind {
6161
kNotImplemented,
6262
kNotSupported,
6363
kRestError,
64+
kRetryableValidationFailed,
6465
kServiceUnavailable,
6566
kTokenExpired,
6667
kUnknownError,
@@ -127,6 +128,7 @@ DEFINE_ERROR_FUNCTION(NotFound)
127128
DEFINE_ERROR_FUNCTION(NotImplemented)
128129
DEFINE_ERROR_FUNCTION(NotSupported)
129130
DEFINE_ERROR_FUNCTION(RestError)
131+
DEFINE_ERROR_FUNCTION(RetryableValidationFailed)
130132
DEFINE_ERROR_FUNCTION(ServiceUnavailable)
131133
DEFINE_ERROR_FUNCTION(TokenExpired)
132134
DEFINE_ERROR_FUNCTION(UnknownError)

‎src/iceberg/table_metadata.cc‎

Lines changed: 12 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1071,12 +1071,13 @@ Status TableMetadataBuilder::Impl::AddSnapshot(std::shared_ptr<Snapshot> snapsho
10711071
"Attempting to add a snapshot before a sort order is added");
10721072
ICEBERG_CHECK(!snapshots_by_id_.contains(snapshot->snapshot_id),
10731073
"Snapshot already exists for id: {}", snapshot->snapshot_id);
1074-
ICEBERG_CHECK(
1075-
metadata_.format_version == 1 ||
1076-
snapshot->sequence_number > metadata_.last_sequence_number ||
1077-
!snapshot->parent_snapshot_id.has_value(),
1078-
"Cannot add snapshot with sequence number {} older than last sequence number {}",
1079-
snapshot->sequence_number, metadata_.last_sequence_number);
1074+
if (metadata_.format_version != 1 &&
1075+
snapshot->sequence_number <= metadata_.last_sequence_number &&
1076+
snapshot->parent_snapshot_id.has_value()) {
1077+
return RetryableValidationFailed(
1078+
"Cannot add snapshot with sequence number {} older than last sequence number {}",
1079+
snapshot->sequence_number, metadata_.last_sequence_number);
1080+
}
10801081

10811082
metadata_.last_sequence_number = snapshot->sequence_number;
10821083
metadata_.snapshots.push_back(snapshot);
@@ -1087,10 +1088,11 @@ Status TableMetadataBuilder::Impl::AddSnapshot(std::shared_ptr<Snapshot> snapsho
10871088
ICEBERG_ASSIGN_OR_RAISE(auto first_row_id, snapshot->FirstRowId());
10881089
ICEBERG_CHECK(first_row_id.has_value(),
10891090
"Cannot add a snapshot: first-row-id is null");
1090-
ICEBERG_CHECK(
1091-
first_row_id.value() >= metadata_.next_row_id,
1092-
"Cannot add a snapshot, first-row-id is behind table next-row-id: {} < {}",
1093-
first_row_id.value(), metadata_.next_row_id);
1091+
if (first_row_id.value() < metadata_.next_row_id) {
1092+
return RetryableValidationFailed(
1093+
"Cannot add a snapshot, first-row-id is behind table next-row-id: {} < {}",
1094+
first_row_id.value(), metadata_.next_row_id);
1095+
}
10941096

10951097
ICEBERG_ASSIGN_OR_RAISE(auto add_rows, snapshot->AddedRows());
10961098
ICEBERG_CHECK(add_rows.has_value(), "Cannot add a snapshot: added-rows is null");

‎src/iceberg/test/manifest_list_versions_test.cc‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -328,6 +328,15 @@ TEST_F(TestManifestListVersions, TestV3WriteMixedRowIdAssignment) {
328328
std::make_optional(kSnapshotFirstRowId + kAddedRows + kExistingRows));
329329
}
330330

331+
TEST_F(TestManifestListVersions, TestV3DeleteRowIdNull) {
332+
const auto manifest_list_path =
333+
WriteManifestList(/*format_version=*/3, kSnapshotFirstRowId, {kDeleteManifest});
334+
335+
auto manifest = ReadManifestList(manifest_list_path);
336+
EXPECT_EQ(manifest.content, ManifestContent::kDeletes);
337+
EXPECT_FALSE(manifest.first_row_id.has_value());
338+
}
339+
331340
TEST_F(TestManifestListVersions, TestV1ForwardCompatibility) {
332341
std::string manifest_list_path =
333342
WriteManifestList(/*format_version=*/1, kSnapshotFirstRowId, {kTestManifest});

‎src/iceberg/test/merging_snapshot_update_test.cc‎

Lines changed: 132 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
#include <optional>
2525
#include <ranges>
2626
#include <string>
27+
#include <unordered_map>
2728
#include <unordered_set>
2829
#include <vector>
2930

@@ -45,9 +46,12 @@
4546
#include "iceberg/table_properties.h"
4647
#include "iceberg/test/executor.h"
4748
#include "iceberg/test/matchers.h"
49+
#include "iceberg/test/retry.h"
4850
#include "iceberg/test/update_test_base.h"
4951
#include "iceberg/transaction.h"
5052
#include "iceberg/update/fast_append.h"
53+
#include "iceberg/update/merge_append.h"
54+
#include "iceberg/update/snapshot_manager.h"
5155
#include "iceberg/update/update_properties.h"
5256
#include "iceberg/util/macros.h"
5357

@@ -276,6 +280,13 @@ class MergingSnapshotUpdateTest : public MinimalUpdateTestBase {
276280
return TestOverwriteUpdate::Make(TableName(), table_);
277281
}
278282

283+
void UpgradeTableToV3() {
284+
ICEBERG_UNWRAP_OR_FAIL(auto props, table_->NewUpdateProperties());
285+
props->Set(TableProperties::kFormatVersion.key(), "3");
286+
EXPECT_THAT(props->Commit(), IsOk());
287+
EXPECT_THAT(table_->Refresh(), IsOk());
288+
}
289+
279290
void SetTableFormatVersion(int8_t format_version) {
280291
table_->metadata()->format_version = format_version;
281292
}
@@ -311,6 +322,22 @@ class MergingSnapshotUpdateTest : public MinimalUpdateTestBase {
311322
return result;
312323
}
313324

325+
Result<std::unordered_map<std::string, std::optional<int64_t>>> DataFileFirstRowIds(
326+
const std::shared_ptr<Snapshot>& snapshot, const TableMetadata& metadata) {
327+
SnapshotCache snapshot_cache(snapshot.get());
328+
ICEBERG_ASSIGN_OR_RAISE(auto manifest_range, snapshot_cache.DataManifests(file_io_));
329+
std::vector<ManifestFile> manifests(manifest_range.begin(), manifest_range.end());
330+
ICEBERG_ASSIGN_OR_RAISE(auto entries, ReadAllEntries(manifests, metadata));
331+
332+
std::unordered_map<std::string, std::optional<int64_t>> first_row_ids;
333+
for (const auto& entry : entries) {
334+
if (entry.data_file != nullptr) {
335+
first_row_ids.emplace(entry.data_file->file_path, entry.data_file->first_row_id);
336+
}
337+
}
338+
return first_row_ids;
339+
}
340+
314341
// Write a manifest file containing the given data files.
315342
// Returns a ManifestFile with added_snapshot_id = kInvalidSnapshotId so it
316343
// is eligible for snapshot ID inheritance.
@@ -448,10 +475,7 @@ TEST_F(MergingSnapshotUpdateTest, CommitNewDataFile) {
448475
}
449476

450477
TEST_F(MergingSnapshotUpdateTest, CommitV3NewDataFileAssignsRowLineage) {
451-
ICEBERG_UNWRAP_OR_FAIL(auto props, table_->NewUpdateProperties());
452-
props->Set(TableProperties::kFormatVersion.key(), "3");
453-
EXPECT_THAT(props->Commit(), IsOk());
454-
EXPECT_THAT(table_->Refresh(), IsOk());
478+
UpgradeTableToV3();
455479

456480
ICEBERG_UNWRAP_OR_FAIL(auto op, NewMergeAppend());
457481
EXPECT_THAT(op->AddFile(file_a_), IsOk());
@@ -466,6 +490,110 @@ TEST_F(MergingSnapshotUpdateTest, CommitV3NewDataFileAssignsRowLineage) {
466490
EXPECT_EQ(table_->metadata()->next_row_id, 100);
467491
}
468492

493+
TEST_F(MergingSnapshotUpdateTest, V3MultiFileRowIds) {
494+
UpgradeTableToV3();
495+
496+
ICEBERG_UNWRAP_OR_FAIL(auto op, NewMergeAppend());
497+
EXPECT_THAT(op->AddFile(file_a_), IsOk());
498+
EXPECT_THAT(op->AddFile(file_b_), IsOk());
499+
EXPECT_THAT(op->Commit(), IsOk());
500+
501+
EXPECT_THAT(table_->Refresh(), IsOk());
502+
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
503+
ICEBERG_UNWRAP_OR_FAIL(auto first_row_id, snapshot->FirstRowId());
504+
ICEBERG_UNWRAP_OR_FAIL(auto added_rows, snapshot->AddedRows());
505+
EXPECT_EQ(first_row_id, std::make_optional<int64_t>(0));
506+
EXPECT_EQ(added_rows, std::make_optional<int64_t>(200));
507+
EXPECT_EQ(table_->metadata()->next_row_id, 200);
508+
509+
ICEBERG_UNWRAP_OR_FAIL(auto first_row_ids,
510+
DataFileFirstRowIds(snapshot, *table_->metadata()));
511+
EXPECT_EQ(first_row_ids.at(file_a_->file_path), std::make_optional<int64_t>(0));
512+
EXPECT_EQ(first_row_ids.at(file_b_->file_path), std::make_optional<int64_t>(100));
513+
}
514+
515+
TEST_F(MergingSnapshotUpdateTest, V3BranchRowIds) {
516+
UpgradeTableToV3();
517+
CommitFileA();
518+
519+
ICEBERG_UNWRAP_OR_FAIL(auto starting_snapshot, table_->current_snapshot());
520+
const auto starting_snapshot_id = starting_snapshot->snapshot_id;
521+
const auto starting_next_row_id = table_->metadata()->next_row_id;
522+
523+
ICEBERG_UNWRAP_OR_FAIL(auto manager, table_->NewSnapshotManager());
524+
manager->CreateBranch("audit", starting_snapshot_id);
525+
EXPECT_THAT(manager->Commit(), IsOk());
526+
EXPECT_THAT(table_->Refresh(), IsOk());
527+
528+
ICEBERG_UNWRAP_OR_FAIL(auto op, NewMergeAppend());
529+
op->SetTargetBranch("audit");
530+
EXPECT_THAT(op->AddFile(file_b_), IsOk());
531+
EXPECT_THAT(op->Commit(), IsOk());
532+
533+
EXPECT_THAT(table_->Refresh(), IsOk());
534+
EXPECT_EQ(table_->metadata()->next_row_id,
535+
starting_next_row_id + file_b_->record_count);
536+
ICEBERG_UNWRAP_OR_FAIL(auto current_snapshot, table_->current_snapshot());
537+
EXPECT_EQ(current_snapshot->snapshot_id, starting_snapshot_id);
538+
539+
auto ref_it = table_->metadata()->refs.find("audit");
540+
ASSERT_NE(ref_it, table_->metadata()->refs.end());
541+
ICEBERG_UNWRAP_OR_FAIL(auto branch_snapshot,
542+
table_->metadata()->SnapshotById(ref_it->second->snapshot_id));
543+
ICEBERG_UNWRAP_OR_FAIL(auto first_row_id, branch_snapshot->FirstRowId());
544+
ICEBERG_UNWRAP_OR_FAIL(auto added_rows, branch_snapshot->AddedRows());
545+
EXPECT_EQ(first_row_id, std::make_optional(starting_next_row_id));
546+
EXPECT_EQ(added_rows, std::make_optional(file_b_->record_count));
547+
548+
ICEBERG_UNWRAP_OR_FAIL(auto first_row_ids,
549+
DataFileFirstRowIds(branch_snapshot, *table_->metadata()));
550+
EXPECT_EQ(first_row_ids.at(file_a_->file_path), std::make_optional<int64_t>(0));
551+
EXPECT_EQ(first_row_ids.at(file_b_->file_path),
552+
std::make_optional(starting_next_row_id));
553+
}
554+
555+
// A staged v3 append must reassign row IDs when a concurrent commit advances next_row_id.
556+
TEST_F(MergingSnapshotUpdateTest, V3RetryRowIds) {
557+
UpgradeTableToV3();
558+
test::FakeRetryEnvironment fake_retry;
559+
ScopedRetryTestHooks retry_hooks(fake_retry.hooks());
560+
561+
// Stage file_a_ with first_row_id = 0, but do not commit the transaction yet.
562+
ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
563+
ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewMergeAppend());
564+
append->AppendFile(file_a_);
565+
EXPECT_THAT(append->Commit(), IsOk());
566+
567+
ICEBERG_UNWRAP_OR_FAIL(auto staged_snapshot, txn->current().Snapshot());
568+
ICEBERG_UNWRAP_OR_FAIL(auto staged_first_row_id, staged_snapshot->FirstRowId());
569+
EXPECT_EQ(staged_first_row_id, std::make_optional<int64_t>(0));
570+
571+
// Commit file_b_ first, advancing table next_row_id and making file_a_'s row IDs stale.
572+
ICEBERG_UNWRAP_OR_FAIL(auto concurrent, table_->NewMergeAppend());
573+
concurrent->AppendFile(file_b_);
574+
EXPECT_THAT(concurrent->Commit(), IsOk());
575+
auto after_concurrent = ReloadMetadata();
576+
EXPECT_EQ(after_concurrent->next_row_id, file_b_->record_count);
577+
578+
// The original transaction must retry and reassign file_a_ after file_b_'s row IDs.
579+
ICEBERG_UNWRAP_OR_FAIL(auto committed_table, txn->Commit());
580+
EXPECT_FALSE(fake_retry.sleep_durations().empty());
581+
EXPECT_EQ(committed_table->metadata()->next_row_id,
582+
file_b_->record_count + file_a_->record_count);
583+
584+
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, committed_table->current_snapshot());
585+
ICEBERG_UNWRAP_OR_FAIL(auto first_row_id, snapshot->FirstRowId());
586+
ICEBERG_UNWRAP_OR_FAIL(auto added_rows, snapshot->AddedRows());
587+
EXPECT_EQ(first_row_id, std::make_optional(file_b_->record_count));
588+
EXPECT_EQ(added_rows, std::make_optional(file_a_->record_count));
589+
590+
ICEBERG_UNWRAP_OR_FAIL(auto first_row_ids,
591+
DataFileFirstRowIds(snapshot, *committed_table->metadata()));
592+
EXPECT_EQ(first_row_ids.at(file_b_->file_path), std::make_optional<int64_t>(0));
593+
EXPECT_EQ(first_row_ids.at(file_a_->file_path),
594+
std::make_optional(file_b_->record_count));
595+
}
596+
469597
TEST_F(MergingSnapshotUpdateTest, CommitMultipleDataFiles) {
470598
ICEBERG_UNWRAP_OR_FAIL(auto op, NewMergeAppend());
471599
EXPECT_THAT(op->AddFile(file_a_), IsOk());

‎src/iceberg/test/retry_util_test.cc‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -485,6 +485,27 @@ TEST(RetryRunnerTest, MakeCommitRetryRunnerRetriesCommitFailed) {
485485
EXPECT_EQ(attempts, 3);
486486
}
487487

488+
TEST(RetryRunnerTest, RetriesRetryableValidation) {
489+
int call_count = 0;
490+
int32_t attempts = 0;
491+
492+
auto result = MakeCommitRetryRunner(3, 1, 10, 5000)
493+
.Run(
494+
[&]() -> Result<int> {
495+
++call_count;
496+
if (call_count <= 2) {
497+
return RetryableValidationFailed("stale");
498+
}
499+
return 99;
500+
},
501+
&attempts);
502+
503+
EXPECT_THAT(result, IsOk());
504+
EXPECT_EQ(*result, 99);
505+
EXPECT_EQ(call_count, 3);
506+
EXPECT_EQ(attempts, 3);
507+
}
508+
488509
TEST(RetryRunnerTest, OnlyRetryOnMultipleErrorKinds) {
489510
int call_count = 0;
490511
int32_t attempts = 0;

‎src/iceberg/test/table_update_test.cc‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -460,4 +460,74 @@ TEST(TableUpdateTest, SetSnapshotRefRejectsTagForMainBranch) {
460460
EXPECT_THAT(result, HasErrorMessage("Cannot set main to a tag, it must be a branch"));
461461
}
462462

463+
TEST(TableUpdateTest, V3SnapshotRows) {
464+
auto base = CreateBaseMetadata();
465+
base->format_version = 3;
466+
base->next_row_id = 5;
467+
auto builder = TableMetadataBuilder::BuildFrom(base.get());
468+
469+
ICEBERG_UNWRAP_OR_FAIL(
470+
auto snapshot,
471+
Snapshot::Make(/*sequence_number=*/1, /*snapshot_id=*/123,
472+
/*parent_snapshot_id=*/std::nullopt, TimePointMsFromUnixMs(2000000),
473+
DataOperation::kAppend,
474+
/*summary=*/{}, base->current_schema_id,
475+
"s3://bucket/manifest-list.avro",
476+
/*first_row_id=*/5,
477+
/*added_rows=*/7));
478+
table::AddSnapshot update(std::shared_ptr<Snapshot>(std::move(snapshot)));
479+
update.ApplyTo(*builder);
480+
481+
ICEBERG_UNWRAP_OR_FAIL(auto metadata, builder->Build());
482+
EXPECT_EQ(metadata->next_row_id, 12);
483+
}
484+
485+
TEST(TableUpdateTest, V3StaleRowId) {
486+
auto base = CreateBaseMetadata();
487+
base->format_version = 3;
488+
base->next_row_id = 5;
489+
auto builder = TableMetadataBuilder::BuildFrom(base.get());
490+
491+
ICEBERG_UNWRAP_OR_FAIL(
492+
auto snapshot,
493+
Snapshot::Make(/*sequence_number=*/1, /*snapshot_id=*/123,
494+
/*parent_snapshot_id=*/std::nullopt, TimePointMsFromUnixMs(2000000),
495+
DataOperation::kAppend,
496+
/*summary=*/{}, base->current_schema_id,
497+
"s3://bucket/manifest-list.avro",
498+
/*first_row_id=*/4,
499+
/*added_rows=*/1));
500+
table::AddSnapshot update(std::shared_ptr<Snapshot>(std::move(snapshot)));
501+
update.ApplyTo(*builder);
502+
503+
auto result = builder->Build();
504+
ASSERT_THAT(result, IsError(ErrorKind::kRetryableValidationFailed));
505+
EXPECT_THAT(
506+
result,
507+
HasErrorMessage("Cannot add a snapshot, first-row-id is behind table next-row-id"));
508+
}
509+
510+
TEST(TableUpdateTest, V3StaleSequence) {
511+
auto base = CreateBaseMetadata();
512+
base->format_version = 3;
513+
base->last_sequence_number = 5;
514+
auto builder = TableMetadataBuilder::BuildFrom(base.get());
515+
516+
ICEBERG_UNWRAP_OR_FAIL(
517+
auto snapshot,
518+
Snapshot::Make(/*sequence_number=*/5, /*snapshot_id=*/123,
519+
/*parent_snapshot_id=*/1, TimePointMsFromUnixMs(2000000),
520+
DataOperation::kAppend,
521+
/*summary=*/{}, base->current_schema_id,
522+
"s3://bucket/manifest-list.avro",
523+
/*first_row_id=*/0,
524+
/*added_rows=*/1));
525+
table::AddSnapshot update(std::shared_ptr<Snapshot>(std::move(snapshot)));
526+
update.ApplyTo(*builder);
527+
528+
auto result = builder->Build();
529+
ASSERT_THAT(result, IsError(ErrorKind::kRetryableValidationFailed));
530+
EXPECT_THAT(result, HasErrorMessage("Cannot add snapshot with sequence number"));
531+
}
532+
463533
} // namespace iceberg

‎src/iceberg/util/error_collector.h‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -159,14 +159,20 @@ class ICEBERG_EXPORT ErrorCollector {
159159
/// in Build(), Apply(), or Commit() methods) to validate that no errors
160160
/// were accumulated during the builder method calls.
161161
///
162-
/// \return Status indicating success if no errors, or a ValidationFailed
163-
/// error with all accumulated error messages
162+
/// \return Status indicating success if no errors, a RetryableValidationFailed if
163+
/// all accumulated errors are retryable validations, or a ValidationFailed
164+
/// error with all accumulated error messages otherwise
164165
[[nodiscard]] Status CheckErrors() const {
165166
if (!errors_.empty()) {
166167
std::string error_msg = "Validation failed due to the following errors:\n";
168+
bool all_retryable = true;
167169
for (const auto& [kind, message] : errors_) {
170+
all_retryable &= kind == ErrorKind::kRetryableValidationFailed;
168171
error_msg += " - " + message + "\n";
169172
}
173+
if (all_retryable) {
174+
return RetryableValidationFailed("{}", error_msg);
175+
}
170176
return ValidationFailed("{}", error_msg);
171177
}
172178
return {};

‎src/iceberg/util/retry_util.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -198,7 +198,8 @@ class RetryRunner : private detail::RetryRunnerBase {
198198
ICEBERG_EXPORT inline auto MakeCommitRetryRunner(int32_t num_retries, int32_t min_wait_ms,
199199
int32_t max_wait_ms,
200200
int32_t total_timeout_ms) {
201-
return RetryRunner<retry::OnlyRetryOn<ErrorKind::kCommitFailed>>(
201+
return RetryRunner<retry::OnlyRetryOn<ErrorKind::kCommitFailed,
202+
ErrorKind::kRetryableValidationFailed>>(
202203
RetryConfig{.num_retries = num_retries,
203204
.min_wait_ms = min_wait_ms,
204205
.max_wait_ms = max_wait_ms,

0 commit comments

Comments
 (0)