feat: add IncrementalFileCleanup strategy and dispatch in ExpireSnapshots::Finalize (#648)

shangxinli · claude · web-flow · commit 3e7b20ab8555 · 2026-05-20T23:22:49.000+08:00
Mirrors Java's IncrementalFileCleanup for the linear-ancestry case: each
manifest is attributed to its writer snapshot, so two passes are enough
instead of the full reachability scan. Cherry-pick protection via
SnapshotSummaryFields::kSourceSnapshotId is preserved.

Finalize() now picks IncrementalFileCleanup when the expiration is
"simple" (no explicit snapshot IDs, no removed snapshots outside the
current main ancestry, and no retained snapshots outside the current
main ancestry), and falls back to ReachableFileCleanup otherwise. The
dispatch matches Java RemoveSnapshots.cleanExpiredSnapshots.

Two existing cleanup tests (DeletesExpiredFiles,
IgnoresExpiredDeleteManifestReadFailures) used an empty current manifest
list, which is an unreachable-orphan scenario that only
ReachableFileCleanup can resolve. They now call ExpireSnapshotId() to
force the reachable path, which keeps their original intent and matches
Java behavior. New tests cover both dispatch branches.

---------

Co-authored-by: shangxinli &lt;shangxinli@users.noreply.github.com&gt;
Co-authored-by: Claude Opus 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/src/iceberg/manifest/manifest_group.cc b/src/iceberg/manifest/manifest_group.cc
@@ -262,14 +262,8 @@ Result<std::vector<ManifestEntry>> ManifestGroup::Entries() {
 
 Result<std::unique_ptr<ManifestReader>> ManifestGroup::MakeReader(
     const ManifestFile& manifest) {
-  auto spec_it = specs_by_id_.find(manifest.partition_spec_id);
-  if (spec_it == specs_by_id_.end()) {
-    return InvalidArgument("Partition spec {} not found for manifest {}",
-                           manifest.partition_spec_id, manifest.manifest_path);
-  }
-
   ICEBERG_ASSIGN_OR_RAISE(auto reader,
-                          ManifestReader::Make(manifest, io_, schema_, spec_it->second));
+                          ManifestReader::Make(manifest, io_, schema_, specs_by_id_));
 
   reader->FilterRows(data_filter_)
       .FilterPartitions(partition_filter_)
diff --git a/src/iceberg/manifest/manifest_reader.cc b/src/iceberg/manifest/manifest_reader.cc
@@ -21,6 +21,7 @@
 
 #include <algorithm>
 #include <memory>
+#include <optional>
 #include <ranges>
 #include <type_traits>
 #include <unordered_set>
@@ -998,6 +999,19 @@ Result<std::unique_ptr<ManifestReader>> ManifestReader::Make(
       manifest.first_row_id);
 }
 
+Result<std::unique_ptr<ManifestReader>> ManifestReader::Make(
+    const ManifestFile& manifest, std::shared_ptr<FileIO> file_io,
+    std::shared_ptr<Schema> schema,
+    const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& specs_by_id) {
+  auto spec_it = specs_by_id.find(manifest.partition_spec_id);
+  if (spec_it == specs_by_id.end() || spec_it->second == nullptr) {
+    return InvalidArgument("Partition spec {} not found for manifest {}",
+                           manifest.partition_spec_id, manifest.manifest_path);
+  }
+  auto spec = spec_it->second;
+  return Make(manifest, std::move(file_io), std::move(schema), std::move(spec));
+}
+
 Result<std::unique_ptr<ManifestReader>> ManifestReader::Make(
     std::string_view manifest_location, std::optional<int64_t> manifest_length,
     std::shared_ptr<FileIO> file_io, std::shared_ptr<Schema> schema,
diff --git a/src/iceberg/manifest/manifest_reader.h b/src/iceberg/manifest/manifest_reader.h
@@ -92,6 +92,17 @@ class ICEBERG_EXPORT ManifestReader {
       const ManifestFile& manifest, std::shared_ptr<FileIO> file_io,
       std::shared_ptr<Schema> schema, std::shared_ptr<PartitionSpec> spec);
 
+  /// \brief Creates a reader for a manifest file using specs keyed by ID.
+  /// \param manifest A ManifestFile object containing metadata about the manifest.
+  /// \param file_io File IO implementation to use.
+  /// \param schema Schema used to bind the partition type.
+  /// \param specs_by_id Mapping of partition spec ID to PartitionSpec.
+  /// \return A Result containing the reader or an error.
+  static Result<std::unique_ptr<ManifestReader>> Make(
+      const ManifestFile& manifest, std::shared_ptr<FileIO> file_io,
+      std::shared_ptr<Schema> schema,
+      const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>& specs_by_id);
+
   /// \brief Creates a reader for a manifest file.
   /// \param manifest_location Path to the manifest file.
   /// \param manifest_length Length of the manifest file.
diff --git a/src/iceberg/test/expire_snapshots_test.cc b/src/iceberg/test/expire_snapshots_test.cc
@@ -28,6 +28,9 @@
 #include "iceberg/avro/avro_register.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/manifest/manifest_writer.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/snapshot.h"
 #include "iceberg/statistics_file.h"
 #include "iceberg/table_metadata.h"
 #include "iceberg/test/matchers.h"
@@ -135,6 +138,13 @@ class ExpireSnapshotsCleanupTest : public UpdateTestBase {
     return manifest_result.value();
   }
 
+  ManifestFile AssignManifestSequenceNumber(ManifestFile manifest,
+                                            int64_t sequence_number) const {
+    manifest.sequence_number = sequence_number;
+    manifest.min_sequence_number = sequence_number;
+    return manifest;
+  }
+
   ManifestFile WriteDeleteManifest(const std::string& path, int64_t snapshot_id,
                                    std::vector<ManifestEntry> entries) {
     auto writer_result = ManifestWriter::MakeWriter(
@@ -227,6 +237,15 @@ TEST_F(ExpireSnapshotsTest, ExpireById) {
   EXPECT_EQ(result.snapshot_ids_to_remove.at(0), 3051729675574597004);
 }
 
+TEST_F(ExpireSnapshotsTest, ExpireByIdOverridesRetainLast) {
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  update->RetainLast(2);
+  update->ExpireSnapshotId(3051729675574597004);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
+  EXPECT_THAT(result.snapshot_ids_to_remove, testing::ElementsAre(3051729675574597004));
+}
+
 TEST_F(ExpireSnapshotsTest, ExpireOlderThan) {
   struct TestCase {
     int64_t expire_older_than;
@@ -243,6 +262,30 @@ TEST_F(ExpireSnapshotsTest, ExpireOlderThan) {
   }
 }
 
+TEST_F(ExpireSnapshotsCleanupTest, RetainsUnreferencedSnapshotAtExpireThreshold) {
+  const int64_t unreferenced_snapshot_id = 4055729675574597004;
+  const int64_t expire_at_ms = 1515100955770;
+
+  auto metadata = ReloadMetadata();
+  metadata->snapshots.push_back(std::make_shared<Snapshot>(Snapshot{
+      .snapshot_id = unreferenced_snapshot_id,
+      .parent_snapshot_id = std::nullopt,
+      .sequence_number = 2,
+      .timestamp_ms = TimePointMsFromUnixMs(expire_at_ms),
+      .manifest_list = table_location_ + "/metadata/unreferenced.avro",
+      .summary = {{SnapshotSummaryFields::kOperation, "append"}},
+      .schema_id = metadata->current_schema_id,
+  }));
+  RewriteTable(std::move(metadata));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  update->ExpireOlderThan(expire_at_ms);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto result, update->Apply());
+  EXPECT_THAT(result.snapshot_ids_to_remove,
+              testing::Not(testing::Contains(unreferenced_snapshot_id)));
+}
+
 TEST_F(ExpireSnapshotsTest, FinalizeRequiresCommittedMetadata) {
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
@@ -350,6 +393,8 @@ TEST_F(ExpireSnapshotsCleanupTest, IgnoresExpiredDeleteManifestReadFailures) {
 
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  // Force the reachable path.
+  update->ExpireSnapshotId(kExpiredSnapshotId);
   update->DeleteWith(
       [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
 
@@ -388,6 +433,7 @@ TEST_F(ExpireSnapshotsCleanupTest, DeletesExpiredFiles) {
 
   std::vector<std::string> deleted_files;
   ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  update->ExpireSnapshotId(kExpiredSnapshotId);
   update->DeleteWith(
       [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
 
@@ -573,4 +619,216 @@ TEST_F(ExpireSnapshotsCleanupTest, KeepsReusedPartitionStats) {
   EXPECT_THAT(deleted_files, testing::Not(testing::Contains(reused_statistics_path)));
 }
 
+TEST_F(ExpireSnapshotsCleanupTest, IncrementalDispatchPreservesAncestorAddedFiles) {
+  const auto expired_data_file_path = table_location_ + "/data/expired-data.parquet";
+  const auto expired_data_manifest_path = table_location_ + "/metadata/expired-data.avro";
+  const auto expired_manifest_list_path =
+      table_location_ + "/metadata/expired-manifest-list.avro";
+  const auto current_manifest_list_path =
+      table_location_ + "/metadata/current-manifest-list.avro";
+
+  auto expired_data_manifest = WriteDataManifest(
+      expired_data_manifest_path, kExpiredSnapshotId,
+      {MakeEntry(ManifestStatus::kAdded, kExpiredSnapshotId, kExpiredSequenceNumber,
+                 MakeDataFile(expired_data_file_path))});
+  WriteManifestList(expired_manifest_list_path, kExpiredSnapshotId,
+                    /*parent_snapshot_id=*/0, kExpiredSequenceNumber,
+                    {expired_data_manifest});
+  WriteManifestList(current_manifest_list_path, kCurrentSnapshotId, kExpiredSnapshotId,
+                    kCurrentSequenceNumber, {});
+  RewriteTableWithManifestLists(expired_manifest_list_path, current_manifest_list_path);
+
+  std::vector<std::string> deleted_files;
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  update->DeleteWith(
+      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_THAT(deleted_files, testing::Contains(expired_data_manifest_path));
+  EXPECT_THAT(deleted_files, testing::Contains(expired_manifest_list_path));
+  EXPECT_THAT(deleted_files, testing::Not(testing::Contains(expired_data_file_path)));
+}
+
+TEST_F(ExpireSnapshotsCleanupTest, IncrementalDeletesExpiredDeletedEntries) {
+  const auto deleted_data_file_path =
+      table_location_ + "/data/deleted-by-expired.parquet";
+  const auto delete_manifest_path =
+      table_location_ + "/metadata/expired-delete-entry.avro";
+  const auto expired_manifest_list_path =
+      table_location_ + "/metadata/expired-deleted-entry-ml.avro";
+  const auto current_manifest_list_path =
+      table_location_ + "/metadata/current-deleted-entry-ml.avro";
+
+  auto delete_manifest = WriteDataManifest(
+      delete_manifest_path, kExpiredSnapshotId,
+      {MakeEntry(ManifestStatus::kDeleted, kExpiredSnapshotId, kExpiredSequenceNumber,
+                 MakeDataFile(deleted_data_file_path))});
+  delete_manifest =
+      AssignManifestSequenceNumber(std::move(delete_manifest), kExpiredSequenceNumber);
+  WriteManifestList(expired_manifest_list_path, kExpiredSnapshotId,
+                    /*parent_snapshot_id=*/0, kExpiredSequenceNumber, {delete_manifest});
+  WriteManifestList(current_manifest_list_path, kCurrentSnapshotId, kExpiredSnapshotId,
+                    kCurrentSequenceNumber, {delete_manifest});
+  RewriteTableWithManifestLists(expired_manifest_list_path, current_manifest_list_path);
+
+  std::vector<std::string> deleted_files;
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  update->DeleteWith(
+      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_THAT(deleted_files, testing::Contains(deleted_data_file_path));
+  EXPECT_THAT(deleted_files, testing::Contains(expired_manifest_list_path));
+  EXPECT_THAT(deleted_files, testing::Not(testing::Contains(delete_manifest_path)));
+}
+
+TEST_F(ExpireSnapshotsCleanupTest, ReachableDispatchDeletesUnreachableData) {
+  const auto expired_data_file_path = table_location_ + "/data/expired-data.parquet";
+  const auto expired_data_manifest_path = table_location_ + "/metadata/expired-data.avro";
+  const auto expired_manifest_list_path =
+      table_location_ + "/metadata/expired-manifest-list.avro";
+  const auto current_manifest_list_path =
+      table_location_ + "/metadata/current-manifest-list.avro";
+
+  auto expired_data_manifest = WriteDataManifest(
+      expired_data_manifest_path, kExpiredSnapshotId,
+      {MakeEntry(ManifestStatus::kAdded, kExpiredSnapshotId, kExpiredSequenceNumber,
+                 MakeDataFile(expired_data_file_path))});
+  WriteManifestList(expired_manifest_list_path, kExpiredSnapshotId,
+                    /*parent_snapshot_id=*/0, kExpiredSequenceNumber,
+                    {expired_data_manifest});
+  WriteManifestList(current_manifest_list_path, kCurrentSnapshotId, kExpiredSnapshotId,
+                    kCurrentSequenceNumber, {});
+  RewriteTableWithManifestLists(expired_manifest_list_path, current_manifest_list_path);
+
+  std::vector<std::string> deleted_files;
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  update->ExpireSnapshotId(kExpiredSnapshotId);
+  update->DeleteWith(
+      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_THAT(deleted_files, testing::UnorderedElementsAre(expired_data_file_path,
+                                                           expired_data_manifest_path,
+                                                           expired_manifest_list_path));
+}
+
+TEST_F(ExpireSnapshotsCleanupTest, IncrementalSkipsCherryPickedSnapshotCleanup) {
+  const auto picked_data_file_path = table_location_ + "/data/picked-data.parquet";
+  const auto picked_manifest_path = table_location_ + "/metadata/picked-data.avro";
+  const auto expired_manifest_list_path =
+      table_location_ + "/metadata/expired-picked-ml.avro";
+  const auto current_manifest_list_path =
+      table_location_ + "/metadata/current-picked-ml.avro";
+
+  auto picked_manifest = WriteDataManifest(
+      picked_manifest_path, kExpiredSnapshotId,
+      {MakeEntry(ManifestStatus::kAdded, kExpiredSnapshotId, kExpiredSequenceNumber,
+                 MakeDataFile(picked_data_file_path))});
+  picked_manifest =
+      AssignManifestSequenceNumber(std::move(picked_manifest), kExpiredSequenceNumber);
+  WriteManifestList(expired_manifest_list_path, kExpiredSnapshotId,
+                    /*parent_snapshot_id=*/0, kExpiredSequenceNumber, {picked_manifest});
+  WriteManifestList(current_manifest_list_path, kCurrentSnapshotId, kExpiredSnapshotId,
+                    kCurrentSequenceNumber, {picked_manifest});
+
+  auto metadata = ReloadMetadata();
+  ASSERT_EQ(metadata->snapshots.size(), 2);
+  metadata->snapshots.at(0)->manifest_list = expired_manifest_list_path;
+  metadata->snapshots.at(1)->manifest_list = current_manifest_list_path;
+  metadata->snapshots.at(1)->summary[SnapshotSummaryFields::kSourceSnapshotId] =
+      std::to_string(kExpiredSnapshotId);
+  RewriteTable(std::move(metadata));
+
+  std::vector<std::string> deleted_files;
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  update->DeleteWith(
+      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_TRUE(deleted_files.empty());
+  auto committed_metadata = ReloadMetadata();
+  EXPECT_EQ(committed_metadata->snapshots.size(), 1);
+  EXPECT_EQ(committed_metadata->snapshots.at(0)->snapshot_id, kCurrentSnapshotId);
+}
+
+TEST_F(ExpireSnapshotsCleanupTest, ReachableCleanupFailsClosedOnUnbindableExpiredSpec) {
+  const auto expired_data_file_path = table_location_ + "/data/expired-data.parquet";
+  const auto expired_data_manifest_path = table_location_ + "/metadata/expired-data.avro";
+  const auto expired_manifest_list_path =
+      table_location_ + "/metadata/expired-manifest-list.avro";
+  const auto current_manifest_list_path =
+      table_location_ + "/metadata/current-manifest-list.avro";
+
+  auto expired_data_manifest = WriteDataManifest(
+      expired_data_manifest_path, kExpiredSnapshotId,
+      {MakeEntry(ManifestStatus::kAdded, kExpiredSnapshotId, kExpiredSequenceNumber,
+                 MakeDataFile(expired_data_file_path))});
+  WriteManifestList(expired_manifest_list_path, kExpiredSnapshotId,
+                    /*parent_snapshot_id=*/0, kExpiredSequenceNumber,
+                    {expired_data_manifest});
+  WriteManifestList(current_manifest_list_path, kCurrentSnapshotId, kExpiredSnapshotId,
+                    kCurrentSequenceNumber, {});
+
+  auto metadata = ReloadMetadata();
+  ASSERT_EQ(metadata->snapshots.size(), 2);
+  metadata->snapshots.at(0)->manifest_list = expired_manifest_list_path;
+  metadata->snapshots.at(1)->manifest_list = current_manifest_list_path;
+  ICEBERG_UNWRAP_OR_FAIL(auto retained_spec, PartitionSpec::Make(/*spec_id=*/1, {}));
+  metadata->partition_specs.push_back(
+      std::shared_ptr<PartitionSpec>(std::move(retained_spec)));
+  metadata->default_spec_id = 1;
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto retained_schema,
+      Schema::Make(std::vector<SchemaField>{SchemaField::MakeRequired(2, "y", int64()),
+                                            SchemaField::MakeRequired(3, "z", int64())},
+                   /*schema_id=*/2, std::vector<int32_t>{}));
+  metadata->schemas.push_back(std::shared_ptr<Schema>(std::move(retained_schema)));
+  metadata->current_schema_id = 2;
+  metadata->snapshots.at(1)->schema_id = 2;
+  RewriteTable(std::move(metadata));
+
+  std::vector<std::string> deleted_files;
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  update->ExpireSnapshotId(kExpiredSnapshotId);
+  update->CleanExpiredMetadata(true);
+  update->DeleteWith(
+      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_THAT(deleted_files, testing::UnorderedElementsAre(expired_data_manifest_path,
+                                                           expired_manifest_list_path));
+  EXPECT_THAT(deleted_files, testing::Not(testing::Contains(expired_data_file_path)));
+}
+
+TEST_F(ExpireSnapshotsCleanupTest, CommitIgnoresMalformedSourceSnapshotIdCleanup) {
+  const auto expired_manifest_list_path =
+      table_location_ + "/metadata/expired-malformed-ml.avro";
+  const auto current_manifest_list_path =
+      table_location_ + "/metadata/current-malformed-ml.avro";
+  WriteManifestList(expired_manifest_list_path, kExpiredSnapshotId,
+                    /*parent_snapshot_id=*/0, kExpiredSequenceNumber, {});
+  WriteManifestList(current_manifest_list_path, kCurrentSnapshotId, kExpiredSnapshotId,
+                    kCurrentSequenceNumber, {});
+
+  auto metadata = ReloadMetadata();
+  ASSERT_EQ(metadata->snapshots.size(), 2);
+  metadata->snapshots.at(0)->manifest_list = expired_manifest_list_path;
+  metadata->snapshots.at(1)->manifest_list = current_manifest_list_path;
+  metadata->snapshots.at(1)->summary[SnapshotSummaryFields::kSourceSnapshotId] =
+      "not-a-number";
+  RewriteTable(std::move(metadata));
+
+  std::vector<std::string> deleted_files;
+  ICEBERG_UNWRAP_OR_FAIL(auto update, table_->NewExpireSnapshots());
+  update->DeleteWith(
+      [&deleted_files](const std::string& path) { deleted_files.push_back(path); });
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_TRUE(deleted_files.empty());
+  auto committed_metadata = ReloadMetadata();
+  EXPECT_EQ(committed_metadata->snapshots.size(), 1);
+  EXPECT_EQ(committed_metadata->snapshots.at(0)->snapshot_id, kCurrentSnapshotId);
+}
+
 }  // namespace iceberg
diff --git a/src/iceberg/update/expire_snapshots.cc b/src/iceberg/update/expire_snapshots.cc
diff --git a/src/iceberg/update/expire_snapshots.h b/src/iceberg/update/expire_snapshots.h
