Address delete validation review comments

Author: gty404

src/iceberg/manifest/manifest_filter_manager.cc

@@ -65,6 +65,25 @@ Result<std::string> FormatPartitionPath(const PartitionSpecsById& specs_by_id,
 
 }  // namespace
 
+size_t ManifestFilterManager::DeleteFileKeyHash::operator()(
+    const DeleteFileKey& key) const {
+  size_t hash = std::hash<std::string>{}(key.path);
+  auto combine = [&hash](const auto& value) {
+    size_t value_hash = value.has_value() ? std::hash<int64_t>{}(*value) : 0;
+    hash ^= value_hash + 0x9e3779b9 + (hash << 6) + (hash >> 2);
+  };
+  combine(key.content_offset);
+  combine(key.content_size_in_bytes);
+  return hash;
+}
+
+ManifestFilterManager::DeleteFileKey ManifestFilterManager::MakeDeleteFileKey(
+    const DataFile& file) {
+  return DeleteFileKey{.path = file.file_path,
+                       .content_offset = file.content_offset,
+                       .content_size_in_bytes = file.content_size_in_bytes};
+}
+
 ManifestFilterManager::ManifestFilterManager(ManifestContent content,
                                              std::shared_ptr<FileIO> file_io)
     : manifest_content_(content),
@@ -93,14 +112,19 @@ void ManifestFilterManager::DeleteFile(std::string_view path) {
 
 Status ManifestFilterManager::DeleteFile(std::shared_ptr<DataFile> file) {
   ICEBERG_PRECHECK(file != nullptr, "Cannot delete file: null");
-  delete_paths_.insert(file->file_path);
+  delete_file_keys_.insert(MakeDeleteFileKey(*file));
   return {};
 }
 
 const DataFileSet& ManifestFilterManager::FilesToBeDeleted() const {
   return delete_files_;
 }
 
+const std::vector<std::shared_ptr<DataFile>>& ManifestFilterManager::DeletedFiles()
+    const {
+  return deleted_files_;
+}
+
 void ManifestFilterManager::DropPartition(int32_t spec_id, PartitionValues partition) {
   drop_partitions_.add(spec_id, std::move(partition));
 }
@@ -113,7 +137,7 @@ void ManifestFilterManager::FailAnyDelete() { fail_any_delete_ = true; }
 
 bool ManifestFilterManager::ContainsDeletes() const {
   return HasRowFilterExpression(delete_expr_) || !delete_paths_.empty() ||
-         !drop_partitions_.empty();
+         !delete_file_keys_.empty() || !drop_partitions_.empty();
 }
 
 void ManifestFilterManager::DropDeleteFilesOlderThan(int64_t sequence_number) {
@@ -129,9 +153,8 @@ void ManifestFilterManager::RemoveDanglingDeletesFor(const DataFileSet& deleted_
 Result<bool> ManifestFilterManager::CanContainDroppedFiles(const ManifestFile&) const {
   // TODO(Guotao): Use the manifest descriptor to skip unrelated object-delete
   // manifests once object-delete partitions are tracked separately.
-  // Currently, DeleteFile(std::shared_ptr<DataFile>) degrades to a path-based delete,
-  // which forces scanning all manifests.
-  return !delete_paths_.empty() || !removed_data_file_paths_.empty();
+  return !delete_paths_.empty() || !delete_file_keys_.empty() ||
+         !removed_data_file_paths_.empty();
 }
 
 Result<bool> ManifestFilterManager::CanContainDroppedPartitions(
@@ -217,8 +240,9 @@ Result<bool> ManifestFilterManager::ShouldDelete(const ManifestEntry& entry,
   const DataFile& file = *entry.data_file;
   int32_t spec_id = file.partition_spec_id.value_or(manifest_spec_id);
 
-  // Path-based and partition-drop checks
+  // Path/object-based and partition-drop checks.
   if (delete_paths_.count(file.file_path) ||
+      delete_file_keys_.count(MakeDeleteFileKey(file)) ||
       drop_partitions_.contains(spec_id, file.partition)) {
     if (fail_any_delete_) {
       ICEBERG_ASSIGN_OR_RAISE(auto partition_path,
@@ -293,8 +317,7 @@ bool ManifestFilterManager::CanTrustManifestReferences(
 Result<ManifestFile> ManifestFilterManager::FilterManifest(
     const std::shared_ptr<Schema>& schema, const PartitionSpecsById& specs_by_id,
     const ManifestFile& manifest, bool trust_manifest_references,
-    const ManifestWriterFactory& writer_factory,
-    std::unordered_set<std::string>& found_paths) {
+    const ManifestWriterFactory& writer_factory, FoundDeletes& found_deletes) {
   ICEBERG_ASSIGN_OR_RAISE(
       auto can_contain_deleted_files,
       CanContainDeletedFiles(manifest, schema, specs_by_id, trust_manifest_references));
@@ -315,7 +338,7 @@ Result<ManifestFile> ManifestFilterManager::FilterManifest(
   }
 
   return FilterManifestWithDeletedFiles(entries, spec_id, schema, specs_by_id,
-                                        writer_factory, found_paths);
+                                        writer_factory, found_deletes);
 }
 
 Result<bool> ManifestFilterManager::ManifestHasDeletedFiles(
@@ -334,21 +357,30 @@ Result<bool> ManifestFilterManager::ManifestHasDeletedFiles(
 Result<ManifestFile> ManifestFilterManager::FilterManifestWithDeletedFiles(
     const std::vector<ManifestEntry>& entries, int32_t manifest_spec_id,
     const std::shared_ptr<Schema>& schema, const PartitionSpecsById& specs_by_id,
-    const ManifestWriterFactory& writer_factory,
-    std::unordered_set<std::string>& found_paths) {
+    const ManifestWriterFactory& writer_factory, FoundDeletes& found_deletes) {
   ICEBERG_ASSIGN_OR_RAISE(auto writer,
                           writer_factory(manifest_spec_id, manifest_content_));
   for (const auto& entry : entries) {
     ICEBERG_ASSIGN_OR_RAISE(auto should_delete,
                             ShouldDelete(entry, schema, specs_by_id, manifest_spec_id));
     if (should_delete) {
-      if (entry.data_file && delete_paths_.count(entry.data_file->file_path)) {
-        found_paths.insert(entry.data_file->file_path);
-      }
       if (entry.data_file) {
-        // TODO(Guotao): Track duplicate deletes and avoid full DataFile copies when
-        // summary generation can use lighter records.
-        delete_files_.insert(std::make_shared<DataFile>(*entry.data_file));
+        const auto key = MakeDeleteFileKey(*entry.data_file);
+        if (delete_paths_.count(entry.data_file->file_path)) {
+          found_deletes.paths.insert(entry.data_file->file_path);
+        }
+        if (delete_file_keys_.count(key)) {
+          found_deletes.files.insert(key);
+        }
+
+        auto file = std::make_shared<DataFile>(*entry.data_file);
+        delete_files_.insert(file);
+        auto [_, inserted] = deleted_file_keys_.insert(key);
+        if (inserted) {
+          deleted_files_.push_back(std::move(file));
+        } else {
+          ++duplicate_deletes_count_;
+        }
       }
       ICEBERG_RETURN_UNEXPECTED(writer->WriteDeletedEntry(entry));
     } else {
@@ -361,18 +393,24 @@ Result<ManifestFile> ManifestFilterManager::FilterManifestWithDeletedFiles(
 }
 
 Status ManifestFilterManager::ValidateRequiredDeletes(
-    const std::unordered_set<std::string>& found_paths) const {
+    const FoundDeletes& found_deletes) const {
   if (!fail_missing_delete_paths_) {
     return {};
   }
 
   std::string missing;
   for (const auto& path : delete_paths_) {
-    if (!found_paths.count(path)) {
+    if (!found_deletes.paths.count(path)) {
       if (!missing.empty()) missing += ", ";
       missing += path;
     }
   }
+  for (const auto& key : delete_file_keys_) {
+    if (!found_deletes.files.count(key)) {
+      if (!missing.empty()) missing += ", ";
+      missing += key.path;
+    }
+  }
   if (!missing.empty()) {
     return InvalidArgument("Missing delete paths: {}", missing);
   }
@@ -391,9 +429,12 @@ Result<std::vector<ManifestFile>> ManifestFilterManager::FilterManifests(
     const std::shared_ptr<Snapshot>& base_snapshot,
     const ManifestWriterFactory& writer_factory) {
   delete_files_.clear();
+  deleted_files_.clear();
+  deleted_file_keys_.clear();
+  duplicate_deletes_count_ = 0;
   replaced_manifests_count_ = 0;
   if (!base_snapshot) {
-    ICEBERG_RETURN_UNEXPECTED(ValidateRequiredDeletes({}));
+    ICEBERG_RETURN_UNEXPECTED(ValidateRequiredDeletes(FoundDeletes{}));
     return std::vector<ManifestFile>{};
   }
 
@@ -431,11 +472,14 @@ Result<std::vector<ManifestFile>> ManifestFilterManager::FilterManifests(
     }
   }
 
-  std::unordered_set<std::string> found_paths;
+  FoundDeletes found_deletes;
   delete_files_.clear();
+  deleted_files_.clear();
+  deleted_file_keys_.clear();
+  duplicate_deletes_count_ = 0;
   if (manifests.empty()) {
     replaced_manifests_count_ = 0;
-    ICEBERG_RETURN_UNEXPECTED(ValidateRequiredDeletes(found_paths));
+    ICEBERG_RETURN_UNEXPECTED(ValidateRequiredDeletes(found_deletes));
     return std::vector<ManifestFile>{};
   }
 
@@ -452,14 +496,14 @@ Result<std::vector<ManifestFile>> ManifestFilterManager::FilterManifests(
     ICEBERG_ASSIGN_OR_RAISE(
         auto filtered_manifest,
         FilterManifest(schema, specs_by_id, *manifest_ptr, trust_manifest_references,
-                       writer_factory, found_paths));
+                       writer_factory, found_deletes));
     if (filtered_manifest.manifest_path != manifest_ptr->manifest_path) {
       ++replaced_manifests_count_;
     }
     filtered.push_back(std::move(filtered_manifest));
   }
 
-  ICEBERG_RETURN_UNEXPECTED(ValidateRequiredDeletes(found_paths));
+  ICEBERG_RETURN_UNEXPECTED(ValidateRequiredDeletes(found_deletes));
   return filtered;
 }
 

src/iceberg/manifest/manifest_filter_manager.h

@@ -25,6 +25,7 @@
 
 #include <cstdint>
 #include <memory>
+#include <optional>
 #include <string>
 #include <unordered_map>
 #include <unordered_set>
@@ -97,6 +98,14 @@ class ICEBERG_EXPORT ManifestFilterManager {
   /// operations (e.g. RowDelta) to enumerate deleted data files for follow-up cleanup.
   const DataFileSet& FilesToBeDeleted() const;
 
+  /// \brief Returns content-file objects deleted by the most recent
+  /// FilterManifests() call, deduplicated by content-file identity.
+  const std::vector<std::shared_ptr<DataFile>>& DeletedFiles() const;
+
+  /// \brief Returns how many duplicate file deletes were found in the most recent
+  /// FilterManifests() call.
+  int32_t DuplicateDeletesCount() const { return duplicate_deletes_count_; }
+
   /// \brief Register a partition for dropping.
   ///
   /// Any manifest entry whose (spec_id, partition) pair matches will be marked DELETED.
@@ -205,12 +214,31 @@ class ICEBERG_EXPORT ManifestFilterManager {
   bool CanTrustManifestReferences(
       const std::vector<const ManifestFile*>& manifests) const;
 
+  struct DeleteFileKey {
+    std::string path;
+    std::optional<int64_t> content_offset;
+    std::optional<int64_t> content_size_in_bytes;
+
+    bool operator==(const DeleteFileKey& other) const = default;
+  };
+
+  struct DeleteFileKeyHash {
+    size_t operator()(const DeleteFileKey& key) const;
+  };
+
+  struct FoundDeletes {
+    std::unordered_set<std::string> paths;
+    std::unordered_set<DeleteFileKey, DeleteFileKeyHash> files;
+  };
+
+  static DeleteFileKey MakeDeleteFileKey(const DataFile& file);
+
   Result<ManifestFile> FilterManifest(const std::shared_ptr<Schema>& schema,
                                       const PartitionSpecsById& specs_by_id,
                                       const ManifestFile& manifest,
                                       bool trust_manifest_references,
                                       const ManifestWriterFactory& writer_factory,
-                                      std::unordered_set<std::string>& found_paths);
+                                      FoundDeletes& found_deletes);
 
   Result<bool> ManifestHasDeletedFiles(const std::vector<ManifestEntry>& entries,
                                        const std::shared_ptr<Schema>& schema,
@@ -220,11 +248,9 @@ class ICEBERG_EXPORT ManifestFilterManager {
   Result<ManifestFile> FilterManifestWithDeletedFiles(
       const std::vector<ManifestEntry>& entries, int32_t manifest_spec_id,
       const std::shared_ptr<Schema>& schema, const PartitionSpecsById& specs_by_id,
-      const ManifestWriterFactory& writer_factory,
-      std::unordered_set<std::string>& found_paths);
+      const ManifestWriterFactory& writer_factory, FoundDeletes& found_deletes);
 
-  Status ValidateRequiredDeletes(
-      const std::unordered_set<std::string>& found_paths) const;
+  Status ValidateRequiredDeletes(const FoundDeletes& found_deletes) const;
 
   /// \brief Get or create a ManifestEvaluator for the given spec.
   Result<ManifestEvaluator*> GetManifestEvaluator(const std::shared_ptr<Schema>& schema,
@@ -247,11 +273,15 @@ class ICEBERG_EXPORT ManifestFilterManager {
 
   std::shared_ptr<Expression> delete_expr_;
   std::unordered_set<std::string> delete_paths_;
+  std::unordered_set<DeleteFileKey, DeleteFileKeyHash> delete_file_keys_;
   DataFileSet delete_files_;
+  std::vector<std::shared_ptr<DataFile>> deleted_files_;
+  std::unordered_set<DeleteFileKey, DeleteFileKeyHash> deleted_file_keys_;
   PartitionSet drop_partitions_;
   bool fail_missing_delete_paths_{false};
   bool fail_any_delete_{false};
   bool case_sensitive_{true};
+  int32_t duplicate_deletes_count_{0};
 
   int32_t replaced_manifests_count_{0};
 

src/iceberg/test/manifest_filter_manager_test.cc

@@ -562,4 +562,84 @@ TEST_F(ManifestFilterManagerTest, RemoveDanglingDeletesForFiltersDanglingDV) {
   EXPECT_EQ(entries[0].status, ManifestStatus::kDeleted);
 }
 
+TEST_F(ManifestFilterManagerTest, DeleteFileObjectMatchesDeletionVectorByContent) {
+  auto metadata = *table_->metadata();
+  metadata.format_version = 3;
+  auto factory = MakeWriterFactory(metadata);
+
+  auto make_dv = [&](int64_t offset) {
+    auto f = std::make_shared<DataFile>();
+    f->content = DataFile::Content::kPositionDeletes;
+    f->file_path = table_location_ + "/delete/dv.puffin";
+    f->file_format = FileFormatType::kPuffin;
+    f->referenced_data_file =
+        std::format("{}/data/referenced-{}.parquet", table_location_, offset);
+    f->content_offset = offset;
+    f->content_size_in_bytes = 10;
+    f->partition = PartitionValues(std::vector<Literal>{Literal::Long(1L)});
+    f->file_size_in_bytes = 256;
+    f->record_count = 5;
+    f->partition_spec_id = spec_->spec_id();
+    return f;
+  };
+  auto dv0 = make_dv(0);
+  auto dv1 = make_dv(10);
+
+  auto manifest_path = std::format("{}/metadata/dv-manifest-{}.avro", table_location_,
+                                   manifest_counter_++);
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto manifest,
+      WriteDeleteManifest({{dv0, 3L}, {dv1, 3L}}, file_io_, metadata, manifest_path));
+
+  ManifestFilterManager mgr(ManifestContent::kDeletes, file_io_);
+  EXPECT_THAT(mgr.DeleteFile(dv0), IsOk());
+
+  std::vector<const ManifestFile*> manifests{&manifest};
+  ICEBERG_UNWRAP_OR_FAIL(auto schema, metadata.Schema());
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto result, mgr.FilterManifests(schema, SpecsById(metadata), manifests, factory));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto entries, ReadAllEntries(result, metadata));
+  ASSERT_EQ(entries.size(), 2U);
+  for (const auto& entry : entries) {
+    ASSERT_NE(entry.data_file, nullptr);
+    if (entry.data_file->content_offset == 0) {
+      EXPECT_EQ(entry.status, ManifestStatus::kDeleted);
+    } else {
+      EXPECT_EQ(entry.status, ManifestStatus::kExisting);
+    }
+  }
+}
+
+TEST_F(ManifestFilterManagerTest, DuplicateDeletesCountRepeatedDeletedFiles) {
+  auto* metadata = table_->metadata().get();
+  auto factory = MakeWriterFactory(*metadata);
+
+  auto del_file = std::make_shared<DataFile>();
+  del_file->content = DataFile::Content::kPositionDeletes;
+  del_file->file_path = table_location_ + "/delete/duplicate.parquet";
+  del_file->file_format = FileFormatType::kParquet;
+  del_file->partition = PartitionValues(std::vector<Literal>{Literal::Long(1L)});
+  del_file->file_size_in_bytes = 512;
+  del_file->record_count = 10;
+  del_file->partition_spec_id = spec_->spec_id();
+
+  auto manifest_path = std::format("{}/metadata/dup-manifest-{}.avro", table_location_,
+                                   manifest_counter_++);
+  ICEBERG_UNWRAP_OR_FAIL(auto manifest,
+                         WriteDeleteManifest({{del_file, 3L}, {del_file, 3L}}, file_io_,
+                                             *metadata, manifest_path));
+
+  ManifestFilterManager mgr(ManifestContent::kDeletes, file_io_);
+  EXPECT_THAT(mgr.DeleteFile(del_file), IsOk());
+
+  std::vector<const ManifestFile*> manifests{&manifest};
+  ICEBERG_UNWRAP_OR_FAIL(auto schema, metadata->Schema());
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto result, mgr.FilterManifests(schema, SpecsById(*metadata), manifests, factory));
+
+  EXPECT_EQ(mgr.DeletedFiles().size(), 1U);
+  EXPECT_EQ(mgr.DuplicateDeletesCount(), 1);
+}
+
 }  // namespace iceberg