fix(manifest): align managers with Java review feedback

Squash the review-fix commits after 2b124ca3 and the current tracked
working-tree changes into one commit.

This aligns ManifestFilterManager and ManifestMergeManager more closely
with Java by:
- adding object-level DeleteFile support and FilesToBeDeleted tracking
- renaming DeletesFiles to ContainsDeletes and adding FailAnyDelete
- separating merge grouping by content and adding kDeletes coverage
- matching ListPacker.packEnd bin-packing behavior
- aligning defined public APIs and row-filter semantics with Java,
  including manager-wide CaseSensitive handling and residual-based
  metrics evaluation
- applying doc and reviewer follow-up fixes

Author: gty404

src/iceberg/manifest/manifest_filter_manager.cc

@@ -25,6 +25,8 @@
 
 #include "iceberg/expression/inclusive_metrics_evaluator.h"
 #include "iceberg/expression/manifest_evaluator.h"
+#include "iceberg/expression/residual_evaluator.h"
+#include "iceberg/expression/strict_metrics_evaluator.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/manifest/manifest_list.h"
 #include "iceberg/manifest/manifest_reader.h"
@@ -39,15 +41,29 @@ ManifestFilterManager::ManifestFilterManager(ManifestContent content,
                                              std::shared_ptr<FileIO> file_io)
     : manifest_content_(content), file_io_(std::move(file_io)) {}
 
-void ManifestFilterManager::DeleteByRowFilter(std::shared_ptr<Expression> expr,
-                                              bool case_sensitive) {
-  delete_exprs_.push_back({std::move(expr), case_sensitive});
+void ManifestFilterManager::DeleteByRowFilter(std::shared_ptr<Expression> expr) {
+  ICEBERG_CHECK_OR_DIE(expr != nullptr, "Cannot delete files using filter: null");
+  delete_exprs_.push_back({.expr = std::move(expr)});
+}
+
+void ManifestFilterManager::CaseSensitive(bool case_sensitive) {
+  case_sensitive_ = case_sensitive;
+  manifest_evaluator_cache_.clear();
+  residual_evaluator_cache_.clear();
 }
 
 void ManifestFilterManager::DeleteFile(std::string_view path) {
-  std::string p(path);
-  delete_paths_.insert(p);
-  pending_paths_.insert(std::move(p));
+  delete_paths_.insert(std::string(path));
+}
+
+void ManifestFilterManager::DeleteFile(std::shared_ptr<DataFile> file) {
+  ICEBERG_CHECK_OR_DIE(file != nullptr, "Cannot delete file: null");
+  delete_paths_.insert(file->file_path);
+  delete_files_.insert(std::move(file));
+}
+
+const DataFileSet& ManifestFilterManager::FilesToBeDeleted() const {
+  return delete_files_;
 }
 
 void ManifestFilterManager::DropPartition(int32_t spec_id, PartitionValues partition) {
@@ -58,7 +74,9 @@ void ManifestFilterManager::FailMissingDeletePaths() {
   fail_missing_delete_paths_ = true;
 }
 
-bool ManifestFilterManager::DeletesFiles() const {
+void ManifestFilterManager::FailAnyDelete() { fail_any_delete_ = true; }
+
+bool ManifestFilterManager::ContainsDeletes() const {
   return !delete_exprs_.empty() || !delete_paths_.empty() || !drop_partitions_.empty();
 }
 
@@ -68,20 +86,22 @@ bool ManifestFilterManager::CanContainDroppedFiles() const {
 
 bool ManifestFilterManager::CanContainDroppedPartitions(const ManifestFile& manifest) {
   if (drop_partitions_.empty()) return false;
-  // Without a partition filter helper, we conservatively say yes.
-  // A full implementation would use ManifestFileUtil::canContainAny; for now
-  // we open the manifest and let per-entry checks decide.
-  (void)manifest;
-  return true;
+  // Only manifests whose partition spec matches a registered drop can contain
+  // entries for that partition.  PartitionKey is pair<spec_id, values>.
+  int32_t spec_id = manifest.partition_spec_id;
+  for (const auto& key : drop_partitions_) {
+    if (key.first == spec_id) return true;
+  }
+  return false;
 }
 
 bool ManifestFilterManager::CanContainExpressionDeletes(const ManifestFile& manifest,
                                                         const TableMetadata& metadata) {
   if (delete_exprs_.empty()) return false;
   int32_t spec_id = manifest.partition_spec_id;
-  for (size_t i = 0; i < delete_exprs_.size(); ++i) {
-    auto* evaluator_ptr = GetManifestEvaluator(metadata, spec_id, delete_exprs_[i])
-                              .value_or(nullptr);
+  for (const auto& delete_expr : delete_exprs_) {
+    auto* evaluator_ptr =
+        GetManifestEvaluator(metadata, spec_id, delete_expr).value_or(nullptr);
     if (evaluator_ptr == nullptr) return true;  // conservative on error
     auto result = evaluator_ptr->Evaluate(manifest);
     if (!result.has_value() || result.value()) return true;
@@ -92,8 +112,12 @@ bool ManifestFilterManager::CanContainExpressionDeletes(const ManifestFile& mani
 bool ManifestFilterManager::CanContainDeletedFiles(const ManifestFile& manifest,
                                                    const TableMetadata& metadata) {
   // A manifest with no live files cannot contain files to delete.
-  bool has_live = (manifest.added_files_count.value_or(0) > 0) ||
-                  (manifest.existing_files_count.value_or(0) > 0);
+  // Null counts mean the count is unknown — treat conservatively as possibly non-zero
+  // (matches Java's ManifestFile.hasAddedFiles / hasExistingFiles behaviour).
+  bool has_live = !manifest.added_files_count.has_value() ||
+                  manifest.added_files_count.value() > 0 ||
+                  !manifest.existing_files_count.has_value() ||
+                  manifest.existing_files_count.value() > 0;
   if (!has_live) return false;
 
   return CanContainDroppedFiles() || CanContainExpressionDeletes(manifest, metadata) ||
@@ -111,51 +135,79 @@ Result<ManifestEvaluator*> ManifestFilterManager::GetManifestEvaluator(
     ICEBERG_ASSIGN_OR_RAISE(auto spec, metadata.PartitionSpecById(spec_id));
     ICEBERG_ASSIGN_OR_RAISE(auto schema, metadata.Schema());
     ICEBERG_ASSIGN_OR_RAISE(vec[idx], ManifestEvaluator::MakeRowFilter(
-                                          de.expr, spec, *schema, de.case_sensitive));
+                                          de.expr, spec, *schema, case_sensitive_));
   }
   return vec[idx].get();
 }
 
-Result<InclusiveMetricsEvaluator*> ManifestFilterManager::GetMetricsEvaluator(
+Result<ResidualEvaluator*> ManifestFilterManager::GetResidualEvaluator(
     const TableMetadata& metadata, int32_t spec_id, const DeleteExpr& de) {
-  auto& vec = metrics_evaluator_cache_[spec_id];
+  auto& vec = residual_evaluator_cache_[spec_id];
   size_t idx = &de - delete_exprs_.data();
   if (idx >= vec.size()) {
     vec.resize(delete_exprs_.size());
   }
   if (!vec[idx]) {
+    ICEBERG_ASSIGN_OR_RAISE(auto spec, metadata.PartitionSpecById(spec_id));
     ICEBERG_ASSIGN_OR_RAISE(auto schema, metadata.Schema());
-    ICEBERG_ASSIGN_OR_RAISE(vec[idx],
-                            InclusiveMetricsEvaluator::Make(de.expr, *schema,
-                                                            de.case_sensitive));
+    ICEBERG_ASSIGN_OR_RAISE(
+        vec[idx], ResidualEvaluator::Make(de.expr, *spec, *schema, case_sensitive_));
   }
   return vec[idx].get();
 }
 
-bool ManifestFilterManager::ShouldDelete(const ManifestEntry& entry,
-                                         const TableMetadata& metadata,
-                                         int32_t manifest_spec_id) {
+Result<bool> ManifestFilterManager::ShouldDelete(const ManifestEntry& entry,
+                                                 const TableMetadata& metadata,
+                                                 int32_t manifest_spec_id) {
   if (!entry.data_file) return false;
   const DataFile& file = *entry.data_file;
-
-  // Path-based check
-  if (delete_paths_.count(file.file_path)) {
-    pending_paths_.erase(file.file_path);
-    return true;
+  int32_t spec_id = file.partition_spec_id.value_or(manifest_spec_id);
+  std::shared_ptr<Schema> schema;
+  if (!delete_exprs_.empty()) {
+    ICEBERG_ASSIGN_OR_RAISE(schema, metadata.Schema());
   }
 
-  // Partition-drop check
-  int32_t spec_id = file.partition_spec_id.value_or(manifest_spec_id);
-  if (drop_partitions_.contains(spec_id, file.partition)) {
+  // Path-based and partition-drop checks
+  if (delete_paths_.count(file.file_path) ||
+      drop_partitions_.contains(spec_id, file.partition)) {
+    if (fail_any_delete_) {
+      return InvalidArgument("Operation would delete existing data: {}", file.file_path);
+    }
     return true;
   }
 
-  // Expression-based check (inclusive metrics)
+  // Expression-based check.
+  // Java semantics: compute a partition residual first, then use strict/inclusive
+  // metrics on that residual. Data manifests reject partial matches; delete manifests
+  // tolerate them because only data-file deletes require all-rows-match validation.
   for (const auto& de : delete_exprs_) {
-    auto* eval = GetMetricsEvaluator(metadata, spec_id, de).value_or(nullptr);
-    if (eval == nullptr) return true;  // conservative on error
-    auto result = eval->Evaluate(file);
-    if (!result.has_value() || result.value()) return true;
+    ICEBERG_ASSIGN_OR_RAISE(auto* residual_eval,
+                            GetResidualEvaluator(metadata, spec_id, de));
+    ICEBERG_ASSIGN_OR_RAISE(auto residual_expr,
+                            residual_eval->ResidualFor(file.partition));
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto strict_eval,
+        StrictMetricsEvaluator::Make(residual_expr, schema, case_sensitive_));
+    ICEBERG_ASSIGN_OR_RAISE(auto strict_match, strict_eval->Evaluate(file));
+    if (strict_match) {
+      if (fail_any_delete_) {
+        return InvalidArgument("Operation would delete existing data: {}",
+                               file.file_path);
+      }
+      return true;
+    }
+
+    ICEBERG_ASSIGN_OR_RAISE(auto incl_eval, InclusiveMetricsEvaluator::Make(
+                                                residual_expr, *schema, case_sensitive_));
+    ICEBERG_ASSIGN_OR_RAISE(auto incl_match, incl_eval->Evaluate(file));
+    if (incl_match) {
+      if (manifest_content_ == ManifestContent::kDeletes) {
+        continue;
+      }
+      return InvalidArgument(
+          "Cannot delete file where some, but not all, rows match filter: {}",
+          file.file_path);
+    }
   }
 
   return false;
@@ -164,12 +216,25 @@ bool ManifestFilterManager::ShouldDelete(const ManifestEntry& entry,
 Result<std::vector<ManifestFile>> ManifestFilterManager::FilterManifests(
     const TableMetadata& metadata, const std::shared_ptr<Snapshot>& base_snapshot,
     const ManifestWriterFactory& writer_factory) {
+  // Validate required delete paths before any early return — even an empty base
+  // snapshot must report missing required paths (matches Java's validateRequiredDeletes).
+  if (fail_missing_delete_paths_ && !delete_paths_.empty() && !base_snapshot) {
+    return InvalidArgument("Missing delete paths: {}", [&] {
+      std::string s;
+      for (const auto& p : delete_paths_) {
+        if (!s.empty()) s += ", ";
+        s += p;
+      }
+      return s;
+    }());
+  }
+
   // No base snapshot → nothing to filter
   if (!base_snapshot) return std::vector<ManifestFile>{};
 
   // Load the relevant manifests from the manifest list
-  ICEBERG_ASSIGN_OR_RAISE(auto list_reader,
-                           ManifestListReader::Make(base_snapshot->manifest_list, file_io_));
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto list_reader, ManifestListReader::Make(base_snapshot->manifest_list, file_io_));
   ICEBERG_ASSIGN_OR_RAISE(auto all_manifests, list_reader->Files());
 
   // Keep only manifests for this manager's content type
@@ -180,12 +245,16 @@ Result<std::vector<ManifestFile>> ManifestFilterManager::FilterManifests(
   }
 
   // No conditions registered → return unchanged
-  if (!DeletesFiles()) return manifests;
+  if (!ContainsDeletes()) return manifests;
 
   ICEBERG_ASSIGN_OR_RAISE(auto schema, metadata.Schema());
 
   std::vector<ManifestFile> result;
   result.reserve(manifests.size());
+  // Track which registered delete paths were actually found across all manifests.
+  // Using delete_paths_ as the immutable source of truth makes FilterManifests
+  // idempotent across commit retries (matches Java's validateRequiredDeletes design).
+  std::unordered_set<std::string> found_paths;
 
   for (const auto& manifest : manifests) {
     // Fast path: metadata skip
@@ -199,13 +268,14 @@ Result<std::vector<ManifestFile>> ManifestFilterManager::FilterManifests(
 
     // Read all live entries from the manifest
     ICEBERG_ASSIGN_OR_RAISE(auto reader,
-                             ManifestReader::Make(manifest, file_io_, schema, spec));
+                            ManifestReader::Make(manifest, file_io_, schema, spec));
     ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->LiveEntries());
 
     // Check whether any entry should be deleted
     bool has_deletes = false;
     for (const auto& entry : entries) {
-      if (ShouldDelete(entry, metadata, spec_id)) {
+      ICEBERG_ASSIGN_OR_RAISE(auto should_delete, ShouldDelete(entry, metadata, spec_id));
+      if (should_delete) {
         has_deletes = true;
         break;
       }
@@ -216,10 +286,17 @@ Result<std::vector<ManifestFile>> ManifestFilterManager::FilterManifests(
       continue;
     }
 
-    // Rewrite the manifest with deleted entries marked
+    // Rewrite the manifest with deleted entries marked; record found paths.
     ICEBERG_ASSIGN_OR_RAISE(auto writer, writer_factory(spec_id, manifest_content_));
     for (const auto& entry : entries) {
-      if (ShouldDelete(entry, metadata, spec_id)) {
+      ICEBERG_ASSIGN_OR_RAISE(auto should_delete, ShouldDelete(entry, metadata, 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) {
+          delete_files_.insert(std::make_shared<DataFile>(*entry.data_file));
+        }
         ICEBERG_RETURN_UNEXPECTED(writer->WriteDeletedEntry(entry));
       } else {
         ICEBERG_RETURN_UNEXPECTED(writer->WriteExistingEntry(entry));
@@ -230,14 +307,19 @@ Result<std::vector<ManifestFile>> ManifestFilterManager::FilterManifests(
     result.push_back(std::move(filtered_manifest));
   }
 
-  // Validate that all registered delete paths were found
-  if (fail_missing_delete_paths_ && !pending_paths_.empty()) {
+  // Validate that all registered delete paths were found.  Uses delete_paths_ (not a
+  // consumed set) so repeated calls on the same manager produce the same result.
+  if (fail_missing_delete_paths_) {
     std::string missing;
-    for (const auto& p : pending_paths_) {
-      if (!missing.empty()) missing += ", ";
-      missing += p;
+    for (const auto& p : delete_paths_) {
+      if (!found_paths.count(p)) {
+        if (!missing.empty()) missing += ", ";
+        missing += p;
+      }
+    }
+    if (!missing.empty()) {
+      return InvalidArgument("Missing delete paths: {}", missing);
     }
-    return InvalidArgument("Missing delete paths: {}", missing);
   }
 
   return result;