This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 7e81fa2d test: complete v3 row lineage test coverage (#869)
7e81fa2d is described below
commit 7e81fa2d96aeff0b9b19d78d9218739db2bb2947
Author: Gang Wu <[email protected]>
AuthorDate: Tue Aug 4 23:18:31 2026 +0800
test: complete v3 row lineage test coverage (#869)
- enable RewriteFiles V3 coverage and verify DV cleanup
- cover V2-to-V3 assignment and incremental/changelog lineage
- verify delete-aware readers preserve lineage
---
src/iceberg/test/file_scan_task_reader_test.cc | 89 +++++++++++-
src/iceberg/test/incremental_append_scan_test.cc | 52 +++++++
.../test/incremental_changelog_scan_test.cc | 98 ++++++++++++++
src/iceberg/test/merging_snapshot_update_test.cc | 99 ++++++++++++++
src/iceberg/test/rewrite_files_test.cc | 149 ++++++++++++++++++---
5 files changed, 459 insertions(+), 28 deletions(-)
diff --git a/src/iceberg/test/file_scan_task_reader_test.cc
b/src/iceberg/test/file_scan_task_reader_test.cc
index 2c2381c4..e8f49f10 100644
--- a/src/iceberg/test/file_scan_task_reader_test.cc
+++ b/src/iceberg/test/file_scan_task_reader_test.cc
@@ -113,6 +113,14 @@ class FileScanTaskReaderTest : public TempFileTestBase {
table_schema_->schema_id());
}
+ std::shared_ptr<Schema> RowLineageProjection() const {
+ return std::make_shared<Schema>(
+ std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()),
+ MetadataColumns::kRowId,
+ MetadataColumns::kLastUpdatedSequenceNumber},
+ table_schema_->schema_id());
+ }
+
Result<ExportedBatch> MakeBatch(const Schema& schema,
const std::string& json_data) const {
ICEBERG_ASSIGN_OR_RAISE(auto arrow_schema, MakeArrowSchema(schema));
@@ -409,17 +417,12 @@ TEST_F(FileScanTaskReaderTest,
ReadLastUpdatedFromDataSeq) {
data_file->first_row_id = 100L;
data_file->data_sequence_number = 5L;
FileScanTask task(data_file);
- auto projected_schema = std::make_shared<Schema>(
- std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()),
- MetadataColumns::kRowId,
- MetadataColumns::kLastUpdatedSequenceNumber},
- table_schema_->schema_id());
FileScanTaskReader::Options options{
.io = file_io_,
.table_schema = table_schema_,
.schemas = {table_schema_},
- .projected_schema = projected_schema,
+ .projected_schema = RowLineageProjection(),
};
ICEBERG_UNWRAP_OR_FAIL(auto reader,
FileScanTaskReader::Make(std::move(options)));
auto stream_result = reader->Open(task);
@@ -504,6 +507,80 @@ TEST_F(FileScanTaskReaderTest,
OpenWithEqualityDeletesAddsAndPrunesDeleteOnlyCol
ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[1, "Foo"], [3,
"Baz"]])"));
}
+TEST_F(FileScanTaskReaderTest, PositionDeletesPreserveRowLineage) {
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto data_file,
+ MakeDataFile(table_schema_,
+ R"([[1, "Foo", "blue"], [2, "Bar", "red"], [3, "Baz",
"green"]])"));
+ data_file->first_row_id = 100;
+ data_file->data_sequence_number = 5;
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto pos_delete,
MakePositionDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"),
+ {1}, data_file->file_path));
+ FileScanTask task(data_file, {pos_delete});
+
+ FileScanTaskReader::Options options{
+ .io = file_io_,
+ .table_schema = table_schema_,
+ .schemas = {table_schema_},
+ .projected_schema = RowLineageProjection(),
+ };
+ ICEBERG_UNWRAP_OR_FAIL(auto reader,
FileScanTaskReader::Make(std::move(options)));
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+
+ ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[1, 100, 5], [3, 102,
5]])"));
+}
+
+TEST_F(FileScanTaskReaderTest, DeletionVectorDeletesPreserveRowLineage) {
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto data_file,
+ MakeDataFile(table_schema_,
+ R"([[1, "Foo", "blue"], [2, "Bar", "red"], [3, "Baz",
"green"]])"));
+ data_file->first_row_id = 100;
+ data_file->data_sequence_number = 5;
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto deletion_vector,
+ MakeDeletionVectorFile(CreateNewTempFilePathWithSuffix(".puffin"), {1},
+ data_file->file_path));
+ FileScanTask task(data_file, {deletion_vector});
+
+ FileScanTaskReader::Options options{
+ .io = file_io_,
+ .table_schema = table_schema_,
+ .schemas = {table_schema_},
+ .projected_schema = RowLineageProjection(),
+ };
+ ICEBERG_UNWRAP_OR_FAIL(auto reader,
FileScanTaskReader::Make(std::move(options)));
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+
+ ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[1, 100, 5], [3, 102,
5]])"));
+}
+
+TEST_F(FileScanTaskReaderTest, EqualityDeletesPreserveRowLineage) {
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto data_file,
+ MakeDataFile(table_schema_,
+ R"([[1, "Foo", "blue"], [2, "Bar", "red"], [3, "Baz",
"green"]])"));
+ data_file->first_row_id = 100;
+ data_file->data_sequence_number = 5;
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto equality_delete,
+ MakeEqualityDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"),
table_schema_,
+ R"([[0, "unused", "red"]])", {3}));
+ FileScanTask task(data_file, {equality_delete});
+
+ FileScanTaskReader::Options options{
+ .io = file_io_,
+ .table_schema = table_schema_,
+ .schemas = {table_schema_},
+ .projected_schema = RowLineageProjection(),
+ };
+ ICEBERG_UNWRAP_OR_FAIL(auto reader,
FileScanTaskReader::Make(std::move(options)));
+ ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+
+ ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[1, 100, 5], [3, 102,
5]])"));
+}
+
TEST_F(FileScanTaskReaderTest,
OpenWithEqualityDeletesKeepsInputBatchWhenAllRowsAlive) {
ICEBERG_UNWRAP_OR_FAIL(
auto data_file,
diff --git a/src/iceberg/test/incremental_append_scan_test.cc
b/src/iceberg/test/incremental_append_scan_test.cc
index 044942ad..194b5115 100644
--- a/src/iceberg/test/incremental_append_scan_test.cc
+++ b/src/iceberg/test/incremental_append_scan_test.cc
@@ -20,6 +20,7 @@
#include <memory>
#include <optional>
#include <string>
+#include <unordered_map>
#include <vector>
#include <gmock/gmock.h>
@@ -588,6 +589,57 @@ TEST_P(IncrementalAppendScanTest, MultipleRootSnapshots) {
}
}
+TEST_P(IncrementalAppendScanTest, PlanRowLineage) {
+ if (GetParam() < 3) {
+ GTEST_SKIP() << "Row lineage is only assigned in v3 manifests";
+ }
+
+ auto snapshot_a =
+ MakeAppendSnapshot(3, 1000L, std::nullopt, 5L,
{"/path/to/file_a.parquet"});
+ snapshot_a->first_row_id = 0;
+ snapshot_a->added_rows = 1;
+
+ auto file_b = MakeDataFile("/path/to/file_b.parquet");
+ auto entry_b = MakeEntry(ManifestStatus::kAdded, 2000L, 7L, file_b);
+ auto manifest_b = WriteDataManifest(3, 2000L, {std::move(entry_b)});
+ manifest_b.first_row_id = 1;
+ auto manifest_list_b = WriteManifestList(3, 2000L, 1000L, 7L, {manifest_b});
+ auto snapshot_b = std::make_shared<Snapshot>(Snapshot{
+ .snapshot_id = 2000L,
+ .parent_snapshot_id = 1000L,
+ .sequence_number = 7L,
+ .timestamp_ms = TimePointMsFromUnixMs(1609459200000L + 2000),
+ .manifest_list = manifest_list_b,
+ .summary = {{"operation", "append"}},
+ .schema_id = schema_->schema_id(),
+ .first_row_id = 1L,
+ .added_rows = 1L,
+ });
+ auto metadata = MakeTableMetadata(
+ {snapshot_a, snapshot_b}, 2000L,
+ {{"main", std::make_shared<SnapshotRef>(SnapshotRef{
+ .snapshot_id = 2000L, .retention =
SnapshotRef::Branch{}})}});
+ metadata->next_row_id = 2;
+
+ ICEBERG_UNWRAP_OR_FAIL(auto builder,
MakeScanBuilder<IncrementalAppendScan>(metadata));
+ builder->FromSnapshot(1000L, /*inclusive=*/true).ToSnapshot(2000L);
+ ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
+ ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles());
+ ASSERT_EQ(tasks.size(), 2);
+
+ std::unordered_map<std::string, std::shared_ptr<FileScanTask>> tasks_by_path;
+ for (const auto& task : tasks) {
+ tasks_by_path.emplace(task->data_file()->file_path, task);
+ }
+ ASSERT_EQ(tasks_by_path.size(), 2);
+
EXPECT_EQ(tasks_by_path.at("/path/to/file_a.parquet")->data_file()->first_row_id,
0);
+ EXPECT_EQ(
+
tasks_by_path.at("/path/to/file_a.parquet")->data_file()->data_sequence_number,
5);
+
EXPECT_EQ(tasks_by_path.at("/path/to/file_b.parquet")->data_file()->first_row_id,
1);
+ EXPECT_EQ(
+
tasks_by_path.at("/path/to/file_b.parquet")->data_file()->data_sequence_number,
7);
+}
+
INSTANTIATE_TEST_SUITE_P(IncrementalAppendScanVersions,
IncrementalAppendScanTest,
testing::Values(1, 2, 3));
diff --git a/src/iceberg/test/incremental_changelog_scan_test.cc
b/src/iceberg/test/incremental_changelog_scan_test.cc
index 62f6d4fe..667f439b 100644
--- a/src/iceberg/test/incremental_changelog_scan_test.cc
+++ b/src/iceberg/test/incremental_changelog_scan_test.cc
@@ -454,6 +454,104 @@ TEST_P(IncrementalChangelogScanTest,
ManifestRewritesAreIgnored) {
EXPECT_EQ(insert_t3->data_file()->file_path, "/path/to/file_c.parquet");
}
+TEST_P(IncrementalChangelogScanTest, PlanAddedRowLineage) {
+ if (GetParam() < 3) {
+ GTEST_SKIP() << "Row lineage is only assigned in v3 manifests";
+ }
+
+ auto snapshot_a =
+ MakeAppendSnapshot(3, 1000L, std::nullopt, 5L,
{"/path/to/file_a.parquet"});
+ snapshot_a->first_row_id = 0;
+ snapshot_a->added_rows = 1;
+
+ auto file_b = MakeDataFile("/path/to/file_b.parquet");
+ auto entry_b = MakeEntry(ManifestStatus::kAdded, 2000L, 7L, file_b);
+ auto manifest_b = WriteDataManifest(3, 2000L, {std::move(entry_b)});
+ manifest_b.first_row_id = 1;
+ auto manifest_list_b = WriteManifestList(3, 2000L, 1000L, 7L, {manifest_b});
+ auto snapshot_b = std::make_shared<Snapshot>(Snapshot{
+ .snapshot_id = 2000L,
+ .parent_snapshot_id = 1000L,
+ .sequence_number = 7L,
+ .timestamp_ms = TimePointMsFromUnixMs(1609459200000L + 2000),
+ .manifest_list = manifest_list_b,
+ .summary = {{"operation", "append"}},
+ .schema_id = schema_->schema_id(),
+ .first_row_id = 1L,
+ .added_rows = 1L,
+ });
+ auto metadata = MakeTableMetadata(
+ {snapshot_a, snapshot_b}, 2000L,
+ {{"main", std::make_shared<SnapshotRef>(SnapshotRef{
+ .snapshot_id = 2000L, .retention =
SnapshotRef::Branch{}})}});
+ metadata->next_row_id = 2;
+
+ ICEBERG_UNWRAP_OR_FAIL(auto builder,
+ MakeScanBuilder<IncrementalChangelogScan>(metadata));
+ builder->FromSnapshot(1000L, /*inclusive=*/true).ToSnapshot(2000L);
+ ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
+ ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles());
+ ASSERT_EQ(tasks.size(), 2);
+ SortTasks(tasks);
+
+ auto added_a = std::dynamic_pointer_cast<AddedRowsScanTask>(tasks[0]);
+ ASSERT_NE(added_a, nullptr);
+ EXPECT_EQ(added_a->commit_snapshot_id(), 1000L);
+ EXPECT_EQ(added_a->data_file()->first_row_id, 0);
+ EXPECT_EQ(added_a->data_file()->data_sequence_number, 5);
+
+ auto added_b = std::dynamic_pointer_cast<AddedRowsScanTask>(tasks[1]);
+ ASSERT_NE(added_b, nullptr);
+ EXPECT_EQ(added_b->commit_snapshot_id(), 2000L);
+ EXPECT_EQ(added_b->data_file()->first_row_id, 1);
+ EXPECT_EQ(added_b->data_file()->data_sequence_number, 7);
+}
+
+TEST_P(IncrementalChangelogScanTest, PlanDeletedRowLineage) {
+ if (GetParam() < 3) {
+ GTEST_SKIP() << "Row lineage is only assigned in v3 manifests";
+ }
+
+ auto snapshot_a =
+ MakeAppendSnapshot(3, 1000L, std::nullopt, 5L,
{"/path/to/file_a.parquet"});
+
+ auto deleted_file = MakeDataFile("/path/to/file_a.parquet");
+ deleted_file->first_row_id = 10;
+ auto deleted_entry = MakeEntry(ManifestStatus::kDeleted,
/*snapshot_id=*/2000L,
+ /*sequence_number=*/5L, deleted_file);
+ auto delete_manifest =
+ WriteDataManifest(3, 2000L, {std::move(deleted_entry)},
unpartitioned_spec_);
+ auto manifest_list = WriteManifestList(3, 2000L, 1000L, 7L,
{delete_manifest});
+ auto snapshot_b = std::make_shared<Snapshot>(Snapshot{
+ .snapshot_id = 2000L,
+ .parent_snapshot_id = 1000L,
+ .sequence_number = 7L,
+ .timestamp_ms = TimePointMsFromUnixMs(1609459200000L + 2000),
+ .manifest_list = manifest_list,
+ .summary = {{"operation", "overwrite"}},
+ .schema_id = schema_->schema_id(),
+ .first_row_id = 1L,
+ .added_rows = 0L,
+ });
+ auto metadata = MakeTableMetadata(
+ {snapshot_a, snapshot_b}, 2000L,
+ {{"main", std::make_shared<SnapshotRef>(SnapshotRef{
+ .snapshot_id = 2000L, .retention =
SnapshotRef::Branch{}})}});
+
+ ICEBERG_UNWRAP_OR_FAIL(auto builder,
+ MakeScanBuilder<IncrementalChangelogScan>(metadata));
+ builder->FromSnapshot(1000L, /*inclusive=*/false).ToSnapshot(2000L);
+ ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
+ ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles());
+ ASSERT_EQ(tasks.size(), 1);
+
+ auto deleted = std::dynamic_pointer_cast<DeletedDataFileScanTask>(tasks[0]);
+ ASSERT_NE(deleted, nullptr);
+ EXPECT_EQ(deleted->data_file()->first_row_id, 10);
+ EXPECT_EQ(deleted->data_file()->data_sequence_number, 5);
+ EXPECT_EQ(deleted->commit_snapshot_id(), 2000L);
+}
+
TEST_P(IncrementalChangelogScanTest, DeleteFilesAreNotSupported) {
auto version = GetParam();
if (version < 2) {
diff --git a/src/iceberg/test/merging_snapshot_update_test.cc
b/src/iceberg/test/merging_snapshot_update_test.cc
index 6684cf22..1c83e486 100644
--- a/src/iceberg/test/merging_snapshot_update_test.cc
+++ b/src/iceberg/test/merging_snapshot_update_test.cc
@@ -52,6 +52,7 @@
#include "iceberg/transaction.h"
#include "iceberg/update/fast_append.h"
#include "iceberg/update/merge_append.h"
+#include "iceberg/update/row_delta.h"
#include "iceberg/update/snapshot_manager.h"
#include "iceberg/update/update_properties.h"
#include "iceberg/util/macros.h"
@@ -273,6 +274,7 @@ class TestOverwriteUpdate : public MergingSnapshotUpdate {
Status AddDelete(std::shared_ptr<DataFile> file, int64_t
data_sequence_number) {
return AddDeleteFile(std::move(file), data_sequence_number);
}
+ Status AddFile(std::shared_ptr<DataFile> file) { return
AddDataFile(std::move(file)); }
Status RemoveDataFile(std::shared_ptr<DataFile> file) {
return DeleteDataFile(std::move(file));
}
@@ -294,6 +296,7 @@ class MergingSnapshotUpdateTest : public
MinimalUpdateTestBase {
file_a_ = MakeDataFile("/data/file_a.parquet", /*partition_x=*/1L);
file_b_ = MakeDataFile("/data/file_b.parquet", /*partition_x=*/2L);
+ file_c_ = MakeDataFile("/data/file_c.parquet", /*partition_x=*/3L);
}
std::shared_ptr<DataFile> MakeDataFile(const std::string& path, int64_t
partition_x) {
@@ -371,6 +374,28 @@ class MergingSnapshotUpdateTest : public
MinimalUpdateTestBase {
EXPECT_THAT(table_->Refresh(), IsOk());
}
+ void CommitV2FilesBeforeUpgrade() {
+ ASSERT_EQ(table_->metadata()->format_version, 2);
+
+ ICEBERG_UNWRAP_OR_FAIL(auto append, table_->NewFastAppend());
+ append->AppendFile(file_a_).AppendFile(file_b_);
+ ASSERT_THAT(append->Commit(), IsOk());
+ ASSERT_THAT(table_->Refresh(), IsOk());
+
+ auto equality_delete =
+ MakeEqualityDeleteFile("/delete/upgrade_eq_delete.parquet", 1L);
+ ICEBERG_UNWRAP_OR_FAIL(auto delta, table_->NewRowDelta());
+ delta->AddDeletes(equality_delete);
+ ASSERT_THAT(delta->Commit(), IsOk());
+ ASSERT_THAT(table_->Refresh(), IsOk());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto overwrite, NewOverwriteUpdate());
+ ASSERT_THAT(overwrite->RemoveDataFile(file_b_), IsOk());
+ ASSERT_THAT(overwrite->AddFile(file_c_), IsOk());
+ ASSERT_THAT(overwrite->Commit(), IsOk());
+ ASSERT_THAT(table_->Refresh(), IsOk());
+ }
+
// Read all entries from a list of ManifestFiles.
Result<std::vector<ManifestEntry>> ReadAllEntries(
const std::vector<ManifestFile>& manifests, const TableMetadata&
metadata) {
@@ -487,6 +512,7 @@ class MergingSnapshotUpdateTest : public
MinimalUpdateTestBase {
std::shared_ptr<Schema> schema_;
std::shared_ptr<DataFile> file_a_;
std::shared_ptr<DataFile> file_b_;
+ std::shared_ptr<DataFile> file_c_;
};
// -------------------------------------------------------------------------
@@ -658,6 +684,79 @@ TEST_F(MergingSnapshotUpdateTest, V3RetryRowIds) {
std::make_optional(file_b_->record_count));
}
+TEST_F(MergingSnapshotUpdateTest, V3UpgradeLeavesExistingRowsUnassigned) {
+ CommitV2FilesBeforeUpgrade();
+
+ const auto v2_snapshot_id = table_->metadata()->current_snapshot_id;
+ UpgradeTableToV3();
+
+ EXPECT_EQ(table_->metadata()->next_row_id, 0);
+ for (const auto& snapshot : table_->metadata()->snapshots) {
+ EXPECT_EQ(snapshot->first_row_id, std::nullopt);
+ EXPECT_EQ(snapshot->added_rows, std::nullopt);
+ }
+ ICEBERG_UNWRAP_OR_FAIL(auto current, table_->current_snapshot());
+ EXPECT_EQ(current->snapshot_id, v2_snapshot_id);
+ EXPECT_EQ(current->first_row_id, std::nullopt);
+ EXPECT_EQ(current->added_rows, std::nullopt);
+
+ SnapshotCache snapshot_cache(current.get());
+ ICEBERG_UNWRAP_OR_FAIL(auto data_manifest_range,
+ snapshot_cache.DataManifests(file_io_));
+ std::vector<ManifestFile> data_manifests(data_manifest_range.begin(),
+ data_manifest_range.end());
+ ASSERT_FALSE(data_manifests.empty());
+ for (const auto& manifest : data_manifests) {
+ EXPECT_EQ(manifest.first_row_id, std::nullopt);
+ }
+
+ ICEBERG_UNWRAP_OR_FAIL(auto first_row_ids,
+ DataFileFirstRowIds(current, *table_->metadata()));
+ EXPECT_EQ(first_row_ids.at(file_a_->file_path), std::nullopt);
+ EXPECT_EQ(first_row_ids.at(file_b_->file_path), std::nullopt);
+ EXPECT_EQ(first_row_ids.at(file_c_->file_path), std::nullopt);
+
+ ICEBERG_UNWRAP_OR_FAIL(auto delete_manifest_range,
+ snapshot_cache.DeleteManifests(file_io_));
+ ASSERT_FALSE(delete_manifest_range.empty());
+ for (const auto& manifest : delete_manifest_range) {
+ EXPECT_EQ(manifest.first_row_id, std::nullopt);
+ }
+}
+
+TEST_F(MergingSnapshotUpdateTest,
V3FirstCommitAssignsExistingRowsAfterUpgrade) {
+ CommitV2FilesBeforeUpgrade();
+
+ UpgradeTableToV3();
+
+ ICEBERG_UNWRAP_OR_FAIL(auto append, table_->NewFastAppend());
+ ASSERT_THAT(append->Commit(), IsOk());
+ ASSERT_THAT(table_->Refresh(), IsOk());
+
+ const auto expected_rows = file_a_->record_count + file_c_->record_count;
+ ICEBERG_UNWRAP_OR_FAIL(auto assigned, table_->current_snapshot());
+ EXPECT_EQ(assigned->first_row_id, 0);
+ EXPECT_EQ(assigned->added_rows, expected_rows);
+ EXPECT_EQ(table_->metadata()->next_row_id, expected_rows);
+
+ ICEBERG_UNWRAP_OR_FAIL(auto first_row_ids,
+ DataFileFirstRowIds(assigned, *table_->metadata()));
+ ASSERT_TRUE(first_row_ids.at(file_a_->file_path).has_value());
+ ASSERT_TRUE(first_row_ids.at(file_c_->file_path).has_value());
+ EXPECT_EQ(first_row_ids.at(file_b_->file_path), std::nullopt);
+ EXPECT_NE(first_row_ids.at(file_a_->file_path),
first_row_ids.at(file_c_->file_path));
+ EXPECT_THAT((std::vector<int64_t>{*first_row_ids.at(file_a_->file_path),
+ *first_row_ids.at(file_c_->file_path)}),
+ ::testing::UnorderedElementsAre(0, file_a_->record_count));
+
+ SnapshotCache snapshot_cache(assigned.get());
+ ICEBERG_UNWRAP_OR_FAIL(auto delete_manifests,
snapshot_cache.DeleteManifests(file_io_));
+ ASSERT_FALSE(delete_manifests.empty());
+ for (const auto& manifest : delete_manifests) {
+ EXPECT_EQ(manifest.first_row_id, std::nullopt);
+ }
+}
+
TEST_F(MergingSnapshotUpdateTest, CommitMultipleDataFiles) {
ICEBERG_UNWRAP_OR_FAIL(auto op, NewMergeAppend());
EXPECT_THAT(op->AddFile(file_a_), IsOk());
diff --git a/src/iceberg/test/rewrite_files_test.cc
b/src/iceberg/test/rewrite_files_test.cc
index 7a825835..d6c3e7b4 100644
--- a/src/iceberg/test/rewrite_files_test.cc
+++ b/src/iceberg/test/rewrite_files_test.cc
@@ -23,6 +23,7 @@
#include <cstdint>
#include <memory>
#include <string>
+#include <string_view>
#include <vector>
#include <arrow/filesystem/filesystem.h>
@@ -64,9 +65,10 @@ class RewriteFilesTest : public MinimalUpdateTestBase {
MakeDataFile("/data/file_a_rewritten.parquet", /*partition_x=*/1L);
rewritten_file_b_ =
MakeDataFile("/data/file_b_rewritten.parquet", /*partition_x=*/2L);
- delete_file_a_ = MakeDeleteFile("/data/delete_a.parquet",
/*partition_x=*/1L);
- rewritten_delete_file_a_ =
- MakeDeleteFile("/data/delete_a_rewritten.parquet", /*partition_x=*/1L);
+ delete_file_a_ = MakePositionDeleteFile("/data/delete_a.parquet",
/*partition_x=*/1L,
+ file_a_->file_path);
+ rewritten_delete_file_a_ = MakePositionDeleteFile(
+ "/data/delete_a_rewritten.parquet", /*partition_x=*/1L,
file_a_->file_path);
eq_delete_file_ =
MakeEqualityDeleteFile("/data/eq_delete_a.parquet",
/*partition_x=*/1L);
}
@@ -83,18 +85,31 @@ class RewriteFilesTest : public MinimalUpdateTestBase {
return f;
}
- std::shared_ptr<DataFile> MakeDeleteFile(const std::string& path, int64_t
partition_x) {
- auto f = MakeDataFile(path, partition_x);
- f->content = DataFile::Content::kPositionDeletes;
- return f;
+ std::shared_ptr<DataFile> MakePositionDeleteFile(
+ const std::string& path, int64_t partition_x,
+ const std::string& referenced_data_file) {
+ auto file = MakeDataFile(path, partition_x);
+ file->content = DataFile::Content::kPositionDeletes;
+ if (table_->metadata()->format_version >= 3) {
+ constexpr std::string_view kParquetSuffix = ".parquet";
+ if (file->file_path.ends_with(kParquetSuffix)) {
+ file->file_path.replace(file->file_path.size() - kParquetSuffix.size(),
+ kParquetSuffix.size(), ".puffin");
+ }
+ file->file_format = FileFormatType::kPuffin;
+ file->referenced_data_file = referenced_data_file;
+ file->content_offset = 0;
+ file->content_size_in_bytes = 10;
+ }
+ return file;
}
std::shared_ptr<DataFile> MakeEqualityDeleteFile(const std::string& path,
int64_t partition_x) {
- auto f = MakeDeleteFile(path, partition_x);
- f->content = DataFile::Content::kEqualityDeletes;
- f->equality_ids = {1};
- return f;
+ auto file = MakeDataFile(path, partition_x);
+ file->content = DataFile::Content::kEqualityDeletes;
+ file->equality_ids = {1};
+ return file;
}
Result<std::shared_ptr<RewriteFiles>> NewRewriteFiles() {
@@ -311,6 +326,13 @@ TEST_P(RewriteFilesFormatVersionTest,
RewriteDeleteFilesCopiesCallerFiles) {
}
CommitFileA();
+ if (format_version() >= 3) {
+ EXPECT_TRUE(delete_file_a_->IsDeletionVector());
+ EXPECT_EQ(delete_file_a_->referenced_data_file, file_a_->file_path);
+ } else {
+ EXPECT_FALSE(delete_file_a_->IsDeletionVector());
+ }
+
{
ICEBERG_UNWRAP_OR_FAIL(auto row_delta, table_->NewRowDelta());
row_delta->AddDeletes(delete_file_a_);
@@ -544,9 +566,10 @@ TEST_P(RewriteFilesFormatVersionTest, DataSequenceNumber) {
GTEST_SKIP() << "Requires format version >= 2";
}
CommitFileA();
+ const auto data_sequence_number = table_->metadata()->last_sequence_number;
ICEBERG_UNWRAP_OR_FAIL(auto rw, NewRewriteFiles());
- rw->SetDataSequenceNumber(5);
+ rw->SetDataSequenceNumber(data_sequence_number);
rw->DeleteDataFile(file_a_);
rw->AddDataFile(rewritten_file_a_);
EXPECT_THAT(rw->Commit(), IsOk());
@@ -555,6 +578,23 @@ TEST_P(RewriteFilesFormatVersionTest, DataSequenceNumber) {
ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kOperation),
DataOperation::kReplace);
+
+ ICEBERG_UNWRAP_OR_FAIL(auto manifests, DataManifests(snapshot));
+ bool found_rewritten = false;
+ for (const auto& manifest : manifests) {
+ ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadAllEntries(std::span{&manifest,
1}));
+ for (const auto& entry : entries) {
+ if (entry.data_file != nullptr &&
+ entry.data_file->file_path == rewritten_file_a_->file_path) {
+ found_rewritten = true;
+ EXPECT_EQ(entry.status, ManifestStatus::kAdded);
+ EXPECT_EQ(entry.sequence_number, data_sequence_number);
+ EXPECT_EQ(manifest.sequence_number, snapshot->sequence_number);
+ }
+ }
+ }
+ EXPECT_TRUE(found_rewritten) << "Rewritten data file should be present";
+ EXPECT_EQ(snapshot->sequence_number,
table_->metadata()->last_sequence_number);
}
// Bulk rewrite with sequence number via the RewriteDataFiles convenience
method.
@@ -773,8 +813,10 @@ TEST_P(RewriteFilesFormatVersionTest,
ReplaceEqualityDeletesWithPositionDeletes)
DataOperation::kReplace);
EXPECT_EQ(
std::stoll(snapshot->summary.at(SnapshotSummaryFields::kRemovedEqDeleteFiles)),
1);
-
EXPECT_EQ(std::stoll(snapshot->summary.at(SnapshotSummaryFields::kAddedPosDeleteFiles)),
- 1);
+ const auto& added_delete_summary = format_version() >= 3
+ ? SnapshotSummaryFields::kAddedDVs
+ :
SnapshotSummaryFields::kAddedPosDeleteFiles;
+ EXPECT_EQ(std::stoll(snapshot->summary.at(added_delete_summary)), 1);
// Verify the delete manifest shows the eq delete as DELETED and pos delete
as ADDED
ICEBERG_UNWRAP_OR_FAIL(auto delete_manifests, DeleteManifests(snapshot));
@@ -792,7 +834,7 @@ TEST_P(RewriteFilesFormatVersionTest,
ReplaceEqualityDeletesWithPositionDeletes)
}
}
EXPECT_TRUE(found_deleted_eq) << "Equality delete should be marked DELETED";
- EXPECT_TRUE(found_added_pos) << "Position delete should be marked ADDED";
+ EXPECT_TRUE(found_added_pos) << "Position delete or DV should be marked
ADDED";
}
// Remove all deletes: create a data file and an associated equality delete,
@@ -987,16 +1029,79 @@ TEST_P(RewriteFilesFormatVersionTest,
RecoverWhenRewriteBothDataAndDeleteFiles)
1);
}
+TEST_F(RewriteFilesTest, V3RewriteSuppressesCallerFirstRowId) {
+ ICEBERG_UNWRAP_OR_FAIL(auto props, table_->NewUpdateProperties());
+ props->Set(TableProperties::kFormatVersion.key(), "3");
+ ASSERT_THAT(props->Commit(), IsOk());
+ ASSERT_THAT(table_->Refresh(), IsOk());
+
+ CommitFileA();
+ ASSERT_EQ(table_->metadata()->next_row_id, file_a_->record_count);
+
+ rewritten_file_a_->first_row_id = 9999;
+ ICEBERG_UNWRAP_OR_FAIL(auto rewrite, NewRewriteFiles());
+ rewrite->RewriteDataFiles({file_a_}, {rewritten_file_a_},
/*sequence_number=*/1);
+ ASSERT_THAT(rewrite->Commit(), IsOk());
+ ASSERT_THAT(table_->Refresh(), IsOk());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ ICEBERG_UNWRAP_OR_FAIL(auto manifests, DataManifests(snapshot));
+ ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadAllEntries(manifests));
+ auto rewritten_entry =
+ std::ranges::find_if(entries, [this](const ManifestEntry& entry) {
+ return entry.status == ManifestStatus::kAdded && entry.data_file !=
nullptr &&
+ entry.data_file->file_path == rewritten_file_a_->file_path;
+ });
+ ASSERT_NE(rewritten_entry, entries.end());
+ EXPECT_EQ(rewritten_entry->data_file->first_row_id, file_a_->record_count);
+ EXPECT_NE(rewritten_entry->data_file->first_row_id,
rewritten_file_a_->first_row_id);
+ EXPECT_EQ(table_->metadata()->next_row_id,
+ file_a_->record_count + rewritten_file_a_->record_count);
+}
+
+TEST_F(RewriteFilesTest, RemovingDataFileAlsoRemovesDV) {
+ ICEBERG_UNWRAP_OR_FAIL(auto props, table_->NewUpdateProperties());
+ props->Set(TableProperties::kFormatVersion.key(), "3");
+ ASSERT_THAT(props->Commit(), IsOk());
+ ASSERT_THAT(table_->Refresh(), IsOk());
+
+ auto dv_a =
+ MakePositionDeleteFile("/data/dv_a.puffin", /*partition_x=*/1L,
file_a_->file_path);
+ auto dv_b =
+ MakePositionDeleteFile("/data/dv_b.puffin", /*partition_x=*/2L,
file_b_->file_path);
+ ASSERT_TRUE(dv_a->IsDeletionVector());
+ ASSERT_TRUE(dv_b->IsDeletionVector());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto delta, table_->NewRowDelta());
+ delta->AddRows(file_a_).AddRows(file_b_).AddDeletes(dv_a).AddDeletes(dv_b);
+ ASSERT_THAT(delta->Commit(), IsOk());
+ ASSERT_THAT(table_->Refresh(), IsOk());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto base_snapshot, table_->current_snapshot());
+ ICEBERG_UNWRAP_OR_FAIL(auto rewrite, NewRewriteFiles());
+ rewrite->ValidateFromSnapshot(base_snapshot->snapshot_id);
+ rewrite->DeleteDataFile(file_a_);
+ ASSERT_THAT(rewrite->Commit(), IsOk());
+ ASSERT_THAT(table_->Refresh(), IsOk());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot());
+ ICEBERG_UNWRAP_OR_FAIL(auto delete_manifests, DeleteManifests(snapshot));
+ ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadAllEntries(delete_manifests));
+ ASSERT_EQ(entries.size(), 2);
+ EXPECT_TRUE(std::ranges::any_of(entries, [&dv_a](const ManifestEntry& entry)
{
+ return entry.status == ManifestStatus::kDeleted && entry.data_file !=
nullptr &&
+ entry.data_file->file_path == dv_a->file_path;
+ }));
+ EXPECT_TRUE(std::ranges::any_of(entries, [&dv_b](const ManifestEntry& entry)
{
+ return entry.status == ManifestStatus::kExisting && entry.data_file !=
nullptr &&
+ entry.data_file->file_path == dv_b->file_path;
+ }));
+}
+
// ============================================================================
// TODO(WZhuo): Tests blocked on missing infrastructure in iceberg-cpp.
// ============================================================================
//
-// TODO(RemovingDataFileAlsoRemovesDV):
-// Blocked by: format v3 DV auto-cleanup not yet supported.
-// Creates data+delete files via RowDelta (v3), rewrites with deleteFile.
-// Verifies the DV for the removed data file is automatically cleaned up.
-// Java guard: assumeThat(formatVersion).isGreaterThanOrEqualTo(3)
-//
// TODO(DeleteWithDuplicateEntriesInManifest):
// Blocked by: cannot yet append the same file twice to create duplicate
manifest
// entries. Appends FILE_A twice, then rewrites one copy. Verifies manifest
entry
@@ -1004,6 +1109,6 @@ TEST_P(RewriteFilesFormatVersionTest,
RecoverWhenRewriteBothDataAndDeleteFiles)
// Java guard: none (runs on all versions)
INSTANTIATE_TEST_SUITE_P(FormatVersions, RewriteFilesFormatVersionTest,
- ::testing::Values(int8_t{1}, int8_t{2}));
+ ::testing::Values(int8_t{1}, int8_t{2}, int8_t{3}));
} // namespace iceberg
