Skip to content

Commit 6ebf1d4

Browse files
gty404wgtmac
gty404
authored and
wgtmac
committed
Address merging snapshot review comments
1 parent f5cf69a commit 6ebf1d4

4 files changed

Lines changed: 144 additions & 45 deletions

File tree

‎src/iceberg/manifest/manifest_filter_manager.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -133,7 +133,7 @@ class ICEBERG_EXPORT ManifestFilterManager {
133133
/// \brief Set the minimum data sequence number for delete files to retain.
134134
///
135135
/// Only valid for ManifestContent::kDeletes managers. Delete entries whose
136-
/// data_sequence_number is positive and less than \p sequence_number will be
136+
/// data_sequence_number is positive and less than sequence_number will be
137137
/// marked DELETED. This continuously removes delete files that cannot match
138138
/// any remaining data rows (i.e. all data written before that sequence number
139139
/// has itself been deleted).
@@ -146,7 +146,7 @@ class ICEBERG_EXPORT ManifestFilterManager {
146146
/// can be cleaned up.
147147
///
148148
/// Only valid for ManifestContent::kDeletes managers. For each DV whose
149-
/// referenced_data_file path appears in \p deleted_files, the DV entry is
149+
/// referenced_data_file path appears in deleted_files, the DV entry is
150150
/// marked DELETED because the data file it targets no longer exists.
151151
///
152152
/// \param deleted_files set of data files that have been marked for deletion

‎src/iceberg/test/merging_snapshot_update_test.cc‎

Lines changed: 76 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -139,14 +139,14 @@ class TestMergeAppend : public MergingSnapshotUpdate {
139139
return MergingSnapshotUpdate::ValidateDeletedDataFiles(
140140
metadata, starting_snapshot_id, partition_set, parent, std::move(io));
141141
}
142-
static Status ValidateAddedDVsForTest(const TableMetadata& metadata,
143-
int64_t starting_snapshot_id,
144-
std::shared_ptr<Expression> conflict_filter,
145-
const std::shared_ptr<Snapshot>& parent,
146-
std::shared_ptr<FileIO> io) {
147-
return MergingSnapshotUpdate::ValidateAddedDVs(metadata, starting_snapshot_id,
148-
std::move(conflict_filter), parent,
149-
std::move(io));
142+
static Status ValidateAddedDVsForTest(
143+
const TableMetadata& metadata, int64_t starting_snapshot_id,
144+
std::shared_ptr<Expression> conflict_filter,
145+
const std::unordered_set<std::string>& referenced_data_files,
146+
const std::shared_ptr<Snapshot>& parent, std::shared_ptr<FileIO> io) {
147+
return MergingSnapshotUpdate::ValidateAddedDVs(
148+
metadata, starting_snapshot_id, std::move(conflict_filter), referenced_data_files,
149+
parent, std::move(io));
150150
}
151151

152152
bool HasDataFiles() const { return AddsDataFiles(); }
@@ -371,8 +371,8 @@ TEST_F(MergingSnapshotUpdateTest, CommitNewDataFile) {
371371

372372
EXPECT_THAT(table_->Refresh(), IsOk());
373373
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
374-
EXPECT_EQ(snapshot->summary.at("added-data-files"), "1");
375-
EXPECT_EQ(snapshot->summary.at("added-records"), "100");
374+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "1");
375+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedRecords), "100");
376376
}
377377

378378
TEST_F(MergingSnapshotUpdateTest, CommitMultipleDataFiles) {
@@ -383,8 +383,8 @@ TEST_F(MergingSnapshotUpdateTest, CommitMultipleDataFiles) {
383383

384384
EXPECT_THAT(table_->Refresh(), IsOk());
385385
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
386-
EXPECT_EQ(snapshot->summary.at("added-data-files"), "2");
387-
EXPECT_EQ(snapshot->summary.at("added-records"), "200");
386+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "2");
387+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedRecords), "200");
388388
}
389389

390390
TEST_F(MergingSnapshotUpdateTest, CommitDataFileAndDeleteFile) {
@@ -398,7 +398,7 @@ TEST_F(MergingSnapshotUpdateTest, CommitDataFileAndDeleteFile) {
398398
EXPECT_THAT(table_->Refresh(), IsOk());
399399
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
400400
// Data file summary
401-
EXPECT_EQ(snapshot->summary.at("added-data-files"), "1");
401+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "1");
402402
}
403403

404404
TEST_F(MergingSnapshotUpdateTest, CommitPreservesExistingManifests) {
@@ -413,7 +413,7 @@ TEST_F(MergingSnapshotUpdateTest, CommitPreservesExistingManifests) {
413413
EXPECT_THAT(table_->Refresh(), IsOk());
414414
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
415415
// Both data files should be visible — 1 existing + 1 new
416-
EXPECT_EQ(snapshot->summary.at("total-data-files"), "2");
416+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kTotalDataFiles), "2");
417417
}
418418

419419
TEST_F(MergingSnapshotUpdateTest, CommitDeletesDataFile) {
@@ -426,8 +426,8 @@ TEST_F(MergingSnapshotUpdateTest, CommitDeletesDataFile) {
426426

427427
EXPECT_THAT(table_->Refresh(), IsOk());
428428
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
429-
EXPECT_EQ(snapshot->summary.at("total-data-files"), "0");
430-
EXPECT_EQ(snapshot->summary.at("deleted-data-files"), "1");
429+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kTotalDataFiles), "0");
430+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kDeletedDataFiles), "1");
431431
}
432432

433433
TEST_F(MergingSnapshotUpdateTest, SetNewDataFilesDataSequenceNumber) {
@@ -438,7 +438,19 @@ TEST_F(MergingSnapshotUpdateTest, SetNewDataFilesDataSequenceNumber) {
438438

439439
EXPECT_THAT(table_->Refresh(), IsOk());
440440
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
441-
EXPECT_EQ(snapshot->summary.at("added-data-files"), "1");
441+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "1");
442+
}
443+
444+
TEST_F(MergingSnapshotUpdateTest, CustomSummaryPropertySurvivesApplyRebuild) {
445+
ICEBERG_UNWRAP_OR_FAIL(auto op, NewMergeAppend());
446+
op->Set("custom-prop", "custom-value");
447+
EXPECT_THAT(op->AddFile(file_a_), IsOk());
448+
EXPECT_THAT(op->Commit(), IsOk());
449+
450+
EXPECT_THAT(table_->Refresh(), IsOk());
451+
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
452+
EXPECT_EQ(snapshot->summary.at("custom-prop"), "custom-value");
453+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "1");
442454
}
443455

444456
// -------------------------------------------------------------------------
@@ -694,8 +706,8 @@ TEST_F(MergingSnapshotUpdateTest, AppendManifestEmptyTable) {
694706
ICEBERG_UNWRAP_OR_FAIL(auto data_manifests, snapshot_cache.DataManifests(file_io_));
695707
ASSERT_EQ(data_manifests.size(), 1);
696708

697-
EXPECT_EQ(snapshot->summary.at("added-data-files"), "2");
698-
EXPECT_EQ(snapshot->summary.at("total-data-files"), "2");
709+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "2");
710+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kTotalDataFiles), "2");
699711
}
700712

701713
TEST_F(MergingSnapshotUpdateTest, AppendManifestWithDataFiles) {
@@ -714,7 +726,7 @@ TEST_F(MergingSnapshotUpdateTest, AppendManifestWithDataFiles) {
714726
ICEBERG_UNWRAP_OR_FAIL(auto data_manifests, snapshot_cache.DataManifests(file_io_));
715727
// Written manifest (file_b_) + appended manifest (file_a_, file_b_)
716728
EXPECT_EQ(data_manifests.size(), 2);
717-
EXPECT_EQ(snapshot->summary.at("added-data-files"), "3");
729+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "3");
718730
}
719731

720732
// -------------------------------------------------------------------------
@@ -742,7 +754,7 @@ TEST_F(MergingSnapshotUpdateTest, AppendManifestMergeWithMinCountOne) {
742754
ICEBERG_UNWRAP_OR_FAIL(auto data_manifests, snapshot_cache.DataManifests(file_io_));
743755
// Both manifests merged into one.
744756
EXPECT_EQ(data_manifests.size(), 1);
745-
EXPECT_EQ(snapshot->summary.at("added-data-files"), "3");
757+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "3");
746758
}
747759

748760
TEST_F(MergingSnapshotUpdateTest, AppendManifestDoNotMergeMinCount) {
@@ -772,7 +784,7 @@ TEST_F(MergingSnapshotUpdateTest, AppendManifestDoNotMergeMinCount) {
772784
ICEBERG_UNWRAP_OR_FAIL(auto data_manifests, snapshot_cache.DataManifests(file_io_));
773785
// Below min-count-to-merge threshold — all 3 pass through unchanged.
774786
EXPECT_EQ(data_manifests.size(), 3);
775-
EXPECT_EQ(snapshot->summary.at("added-data-files"), "3");
787+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "3");
776788
}
777789

778790
// -------------------------------------------------------------------------
@@ -799,7 +811,7 @@ TEST_F(MergingSnapshotUpdateTest, ManifestMergeMergesIntoOne) {
799811
SnapshotCache snapshot_cache(snapshot.get());
800812
ICEBERG_UNWRAP_OR_FAIL(auto data_manifests, snapshot_cache.DataManifests(file_io_));
801813
EXPECT_EQ(data_manifests.size(), 1);
802-
EXPECT_EQ(snapshot->summary.at("total-data-files"), "2");
814+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kTotalDataFiles), "2");
803815
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kManifestsReplaced), "1");
804816
}
805817

@@ -816,7 +828,7 @@ TEST_F(MergingSnapshotUpdateTest, ManifestMergeDoesNotMergeWhenBelowMinCount) {
816828
SnapshotCache snapshot_cache(snapshot.get());
817829
ICEBERG_UNWRAP_OR_FAIL(auto data_manifests, snapshot_cache.DataManifests(file_io_));
818830
EXPECT_EQ(data_manifests.size(), 2);
819-
EXPECT_EQ(snapshot->summary.at("total-data-files"), "2");
831+
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kTotalDataFiles), "2");
820832
}
821833

822834
TEST_F(MergingSnapshotUpdateTest, ManifestMergeDoesNotMergeWhenSizeTargetTooSmall) {
@@ -1147,12 +1159,51 @@ TEST_F(MergingSnapshotUpdateTest, ValidateAddedDVsDetectsConflict) {
11471159
metadata->current_snapshot_id = second_snapshot->snapshot_id;
11481160
metadata->last_sequence_number = second_snapshot->sequence_number;
11491161

1162+
const std::unordered_set<std::string> referenced_data_files{file_a_->file_path};
11501163
EXPECT_THAT(TestMergeAppend::ValidateAddedDVsForTest(
11511164
*metadata, first_snapshot->snapshot_id, Expressions::AlwaysTrue(),
1152-
second_snapshot, file_io_),
1165+
referenced_data_files, second_snapshot, file_io_),
11531166
IsError(ErrorKind::kInvalidArgument));
11541167
}
11551168

1169+
TEST_F(MergingSnapshotUpdateTest, ValidateAddedDVsIgnoresUnrelatedDVs) {
1170+
CommitFileA();
1171+
ICEBERG_UNWRAP_OR_FAIL(auto first_snapshot, table_->current_snapshot());
1172+
1173+
auto metadata = std::make_shared<TableMetadata>(*table_->metadata());
1174+
metadata->format_version = 3;
1175+
1176+
auto dv_file = MakeDeleteFile("/delete/dv_a.puffin", 1L);
1177+
dv_file->file_format = FileFormatType::kPuffin;
1178+
dv_file->referenced_data_file = file_a_->file_path;
1179+
dv_file->content_offset = 0;
1180+
dv_file->content_size_in_bytes = 10;
1181+
1182+
constexpr int64_t kSecondSnapshotId = 123456;
1183+
auto manifest_path = table_location_ + "/metadata/dv-unrelated.avro";
1184+
ICEBERG_UNWRAP_OR_FAIL(auto manifest,
1185+
WriteDeleteManifest(*metadata, manifest_path, {dv_file},
1186+
first_snapshot->sequence_number + 1));
1187+
manifest.added_snapshot_id = kSecondSnapshotId;
1188+
manifest.sequence_number = first_snapshot->sequence_number + 1;
1189+
manifest.min_sequence_number = first_snapshot->sequence_number + 1;
1190+
ICEBERG_UNWRAP_OR_FAIL(
1191+
auto second_snapshot,
1192+
MakeSyntheticSnapshot(DataOperation::kOverwrite, kSecondSnapshotId,
1193+
first_snapshot->snapshot_id,
1194+
first_snapshot->sequence_number + 1, {manifest}));
1195+
1196+
metadata->snapshots.push_back(second_snapshot);
1197+
metadata->current_snapshot_id = second_snapshot->snapshot_id;
1198+
metadata->last_sequence_number = second_snapshot->sequence_number;
1199+
1200+
const std::unordered_set<std::string> referenced_data_files{file_b_->file_path};
1201+
EXPECT_THAT(TestMergeAppend::ValidateAddedDVsForTest(
1202+
*metadata, first_snapshot->snapshot_id, Expressions::AlwaysTrue(),
1203+
referenced_data_files, second_snapshot, file_io_),
1204+
IsOk());
1205+
}
1206+
11561207
TEST_F(MergingSnapshotUpdateTest, ValidateDeletedDataFilesWithExpressionDetectsConflict) {
11571208
CommitFileA();
11581209
ICEBERG_UNWRAP_OR_FAIL(auto first_snapshot, table_->current_snapshot());

‎src/iceberg/update/merging_snapshot_update.cc‎

Lines changed: 55 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -274,9 +274,7 @@ Status MergingSnapshotUpdate::AddDeleteFile(std::shared_ptr<DataFile> file,
274274
if (!file->partition_spec_id.has_value()) {
275275
return InvalidArgument("Delete file must have a partition spec ID");
276276
}
277-
ICEBERG_ASSIGN_OR_RAISE(auto spec,
278-
base().PartitionSpecById(file->partition_spec_id.value()));
279-
(void)spec;
277+
ICEBERG_RETURN_UNEXPECTED(base().PartitionSpecById(file->partition_spec_id.value()));
280278
has_new_delete_files_ = true;
281279
new_delete_files_.push_back(PendingDeleteFile{
282280
.file = std::move(file), .data_sequence_number = std::move(data_sequence_number)});
@@ -336,6 +334,7 @@ void MergingSnapshotUpdate::CaseSensitive(bool case_sensitive) {
336334
}
337335

338336
void MergingSnapshotUpdate::Set(const std::string& property, const std::string& value) {
337+
custom_summary_properties_[property] = value;
339338
summary_builder().Set(property, value);
340339
}
341340

@@ -482,7 +481,8 @@ MergingSnapshotUpdate::NormalizeNewDeleteFiles() const {
482481
dv_by_referenced_data_file.emplace(*file->referenced_data_file, key);
483482
if (!inserted && it->second != key) {
484483
return NotImplemented(
485-
"Cannot merge multiple deletion vectors for referenced data file: {}",
484+
"Merging multiple deletion vectors is not supported yet for referenced "
485+
"data file: {}",
486486
*file->referenced_data_file);
487487
}
488488
}
@@ -560,6 +560,9 @@ Result<std::vector<ManifestFile>> MergingSnapshotUpdate::Apply(
560560
summary_builder().Merge(added_data_files_summary_);
561561
summary_builder().Merge(added_delete_files_summary_);
562562
summary_builder().Merge(appended_manifests_summary_);
563+
for (const auto& [property, value] : custom_summary_properties_) {
564+
summary_builder().Set(property, value);
565+
}
563566

564567
ICEBERG_ASSIGN_OR_RAISE(auto target_schema,
565568
SnapshotUtil::SchemaFor(metadata_to_update, target_branch()));
@@ -1069,22 +1072,61 @@ Result<std::unique_ptr<DeleteFileIndex>> MergingSnapshotUpdate::AddedDeleteFiles
10691072

10701073
Status MergingSnapshotUpdate::ValidateAddedDVs(
10711074
const TableMetadata& metadata, int64_t starting_snapshot_id,
1072-
std::shared_ptr<Expression> conflict_filter, const std::shared_ptr<Snapshot>& parent,
1073-
std::shared_ptr<FileIO> io) {
1074-
if (parent == nullptr || metadata.format_version < 3) {
1075+
std::shared_ptr<Expression> conflict_filter,
1076+
const std::unordered_set<std::string>& referenced_data_files,
1077+
const std::shared_ptr<Snapshot>& parent, std::shared_ptr<FileIO> io) {
1078+
if (parent == nullptr || referenced_data_files.empty() || metadata.format_version < 3) {
10751079
return {};
10761080
}
10771081

10781082
ICEBERG_ASSIGN_OR_RAISE(
1079-
auto deletes, AddedDeleteFiles(metadata, starting_snapshot_id,
1080-
std::move(conflict_filter), nullptr, parent, io));
1081-
for (const auto& delete_file : deletes->ReferencedDeleteFiles()) {
1082-
if (ContentFileUtil::IsDV(*delete_file)) {
1083-
return InvalidArgument("Cannot commit, found new deletion vector: {}",
1084-
ContentFileUtil::DVDesc(*delete_file));
1083+
auto history, ValidationHistory(metadata, parent->snapshot_id, starting_snapshot_id,
1084+
{DataOperation::kOverwrite, DataOperation::kDelete,
1085+
DataOperation::kReplace},
1086+
ManifestContent::kDeletes, io));
1087+
ICEBERG_ASSIGN_OR_RAISE(auto schema, metadata.Schema());
1088+
1089+
for (const auto& manifest : history.manifests) {
1090+
ICEBERG_ASSIGN_OR_RAISE(auto spec,
1091+
metadata.PartitionSpecById(manifest.partition_spec_id));
1092+
ICEBERG_ASSIGN_OR_RAISE(auto reader,
1093+
ManifestReader::Make(manifest, io, schema, spec));
1094+
if (conflict_filter != nullptr) {
1095+
reader->FilterRows(conflict_filter);
1096+
}
1097+
ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->LiveEntries());
1098+
1099+
for (const auto& entry : entries) {
1100+
if (entry.data_file == nullptr || !ContentFileUtil::IsDV(*entry.data_file) ||
1101+
!entry.data_file->referenced_data_file.has_value()) {
1102+
continue;
1103+
}
1104+
if (referenced_data_files.contains(*entry.data_file->referenced_data_file)) {
1105+
return InvalidArgument("Cannot commit, found new deletion vector: {}",
1106+
ContentFileUtil::DVDesc(*entry.data_file));
1107+
}
10851108
}
10861109
}
10871110
return {};
10881111
}
10891112

1113+
Status MergingSnapshotUpdate::ValidateAddedDVs(
1114+
const TableMetadata& metadata, int64_t starting_snapshot_id,
1115+
std::shared_ptr<Expression> conflict_filter, const std::shared_ptr<Snapshot>& parent,
1116+
std::shared_ptr<FileIO> io) const {
1117+
std::unordered_set<std::string> referenced_data_files;
1118+
for (const auto& pending_file : new_delete_files_) {
1119+
if (pending_file.file == nullptr || !ContentFileUtil::IsDV(*pending_file.file) ||
1120+
!pending_file.file->referenced_data_file.has_value()) {
1121+
continue;
1122+
}
1123+
referenced_data_files.insert(*pending_file.file->referenced_data_file);
1124+
}
1125+
if (referenced_data_files.empty()) {
1126+
return {};
1127+
}
1128+
return ValidateAddedDVs(metadata, starting_snapshot_id, std::move(conflict_filter),
1129+
referenced_data_files, parent, std::move(io));
1130+
}
1131+
10901132
} // namespace iceberg

‎src/iceberg/update/merging_snapshot_update.h‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -279,11 +279,11 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate {
279279
/// \brief Return an error if any snapshot in [starting_snapshot_id+1, parent]
280280
/// added a deletion vector that conflicts with DVs being written.
281281
///
282-
static Status ValidateAddedDVs(const TableMetadata& metadata,
283-
int64_t starting_snapshot_id,
284-
std::shared_ptr<Expression> conflict_filter,
285-
const std::shared_ptr<Snapshot>& parent,
286-
std::shared_ptr<FileIO> io);
282+
static Status ValidateAddedDVs(
283+
const TableMetadata& metadata, int64_t starting_snapshot_id,
284+
std::shared_ptr<Expression> conflict_filter,
285+
const std::unordered_set<std::string>& referenced_data_files,
286+
const std::shared_ptr<Snapshot>& parent, std::shared_ptr<FileIO> io);
287287

288288
private:
289289
struct PendingDeleteFile {
@@ -302,6 +302,11 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate {
302302
Status AddDeleteFile(std::shared_ptr<DataFile> file,
303303
std::optional<int64_t> data_sequence_number);
304304

305+
Status ValidateAddedDVs(const TableMetadata& metadata, int64_t starting_snapshot_id,
306+
std::shared_ptr<Expression> conflict_filter,
307+
const std::shared_ptr<Snapshot>& parent,
308+
std::shared_ptr<FileIO> io) const;
309+
305310
Result<std::vector<PendingDeleteFile>> NormalizeNewDeleteFiles() const;
306311

307312
/// \brief Write new data manifests for staged data files; caches the result.
@@ -320,6 +325,7 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate {
320325
SnapshotSummaryBuilder added_data_files_summary_;
321326
SnapshotSummaryBuilder added_delete_files_summary_;
322327
SnapshotSummaryBuilder appended_manifests_summary_;
328+
std::unordered_map<std::string, std::string> custom_summary_properties_;
323329

324330
ManifestFilterManager data_filter_manager_;
325331
ManifestFilterManager delete_filter_manager_;

0 commit comments

Comments
 (0)