align java model

Author: wgtmac

src/iceberg/manifest/manifest_filter_manager.cc

@@ -23,17 +23,13 @@
 /// Filters an existing snapshot's manifest list, marking data files as DELETED
 /// or EXISTING based on row-filter expressions, exact path deletes, and partition drops.
 
-#include <functional>
+#include <cstdint>
 #include <memory>
 #include <string>
 #include <unordered_map>
 #include <unordered_set>
 #include <vector>
 
-#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/iceberg_export.h"
 #include "iceberg/manifest/manifest_list.h"
 #include "iceberg/manifest/manifest_writer.h"
@@ -44,15 +40,6 @@
 
 namespace iceberg {
 
-/// \brief Factory type for creating ManifestWriter instances during filtering/merging.
-///
-/// The factory receives the partition spec ID (to look up the spec) and the manifest
-/// content type, and returns a new ManifestWriter ready for writing.  The caller
-/// (i.e. MergingSnapshotUpdate in PR2) captures metadata, FileIO, and snapshot ID
-/// inside the lambda.
-using ManifestWriterFactory = std::function<Result<std::unique_ptr<ManifestWriter>>(
-    int32_t spec_id, ManifestContent content)>;
-
 /// \brief Filters an existing snapshot's manifest list.
 ///
 /// The manager accumulates delete conditions incrementally, then applies them all
@@ -63,15 +50,16 @@ using ManifestWriterFactory = std::function<Result<std::unique_ptr<ManifestWrite
 /// The manager is content-agnostic: pass ManifestContent::kData to process data
 /// manifests, or ManifestContent::kDeletes to process delete manifests.
 ///
-/// \note For ManifestContent::kDeletes, this implementation applies
-/// path/partition/row-filter/object-level deletes.  Java's additional cleanup
-/// semantics (dropDeleteFilesOlderThan, removeDanglingDeletesFor / DV cleanup)
-/// are not yet implemented and are planned for a later layer.
+/// TODO(Guotao): For ManifestContent::kDeletes, implement cleanup for orphan delete files
+/// and dangling deletion vectors.
 ///
 /// \note This class is non-copyable and non-movable.
 class ICEBERG_EXPORT ManifestFilterManager {
  public:
+  using PartitionSpecsById = std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>;
+
   ManifestFilterManager(ManifestContent content, std::shared_ptr<FileIO> file_io);
+  ~ManifestFilterManager();
 
   ManifestFilterManager(const ManifestFilterManager&) = delete;
   ManifestFilterManager& operator=(const ManifestFilterManager&) = delete;
@@ -82,7 +70,7 @@ class ICEBERG_EXPORT ManifestFilterManager {
   /// expression will be marked DELETED.
   ///
   /// \param expr The expression to match files against
-  void DeleteByRowFilter(std::shared_ptr<Expression> expr);
+  Status DeleteByRowFilter(std::shared_ptr<Expression> expr);
 
   /// \brief Set whether row-filter field binding is case-sensitive.
   void CaseSensitive(bool case_sensitive);
@@ -98,11 +86,11 @@ class ICEBERG_EXPORT ManifestFilterManager {
   ///
   /// Any manifest entry whose file_path matches file->file_path will be marked
   /// DELETED.  The file object is retained in FilesToBeDeleted(), allowing callers
-  /// (e.g. RowDelta in PR5) to enumerate deleted file objects for DV cleanup.
+  /// to enumerate deleted file objects for follow-up delete-file cleanup.
   /// Duplicate registrations (same path) are silently ignored.
   ///
   /// \param file The data/delete file to delete (must not be null)
-  void DeleteFile(std::shared_ptr<DataFile> file);
+  Status DeleteFile(std::shared_ptr<DataFile> file);
 
   /// \brief Returns the set of file objects marked for deletion by this manager.
   ///
@@ -126,9 +114,6 @@ class ICEBERG_EXPORT ManifestFilterManager {
 
   /// \brief Set a flag that makes FilterManifests() return an error if any
   /// manifest entry matches a delete condition.
-  ///
-  /// Used by conflict-detection logic (e.g. OverwriteFiles in PR4) to detect
-  /// whether a concurrent delete would invalidate a set of newly written files.
   void FailAnyDelete();
 
   /// \brief Returns true if any delete condition has been registered.
@@ -147,59 +132,97 @@ class ICEBERG_EXPORT ManifestFilterManager {
       const TableMetadata& metadata, const std::shared_ptr<Snapshot>& base_snapshot,
       const ManifestWriterFactory& writer_factory);
 
- private:
-  struct DeleteExpr {
-    std::shared_ptr<Expression> expr;
-  };
+  /// \brief Apply all accumulated delete conditions to the provided manifests.
+  ///
+  /// This overload accepts only the context needed for filtering.  It is intended for
+  /// callers that already have the active schema, partition specs, and manifest list.
+  ///
+  /// \param schema Active schema to bind row-filter expressions and metrics evaluators
+  /// \param specs_by_id All partition specs keyed by spec ID
+  /// \param manifests Manifest descriptors to filter
+  /// \param writer_factory Factory to create new ManifestWriter instances
+  /// \return The filtered manifest list, or an error
+  Result<std::vector<ManifestFile>> FilterManifests(
+      const std::shared_ptr<Schema>& schema, const PartitionSpecsById& specs_by_id,
+      const std::vector<const ManifestFile*>& manifests,
+      const ManifestWriterFactory& writer_factory);
 
+ private:
   /// \brief Returns true if the manifest might contain files matching any expression.
-  bool CanContainExpressionDeletes(const ManifestFile& manifest,
-                                   const TableMetadata& metadata);
+  Result<bool> CanContainExpressionDeletes(const ManifestFile& manifest,
+                                           const std::shared_ptr<Schema>& schema,
+                                           const PartitionSpecsById& specs_by_id);
 
   /// \brief Returns true if the manifest might contain files in a dropped partition.
   ///
   /// Checks whether the manifest's partition_spec_id matches any spec_id registered
   /// via DropPartition().  Manifests from a different spec cannot contain the dropped
-  /// partition values.  A more precise implementation could also compare
-  /// partition_summaries bounds, but that requires decoding binary bounds against the
-  /// PartitionSpec, which is not yet available at this call site.
-  bool CanContainDroppedPartitions(const ManifestFile& manifest);
+  /// partition values.
+  Result<bool> CanContainDroppedPartitions(const ManifestFile& manifest) const;
 
   /// \brief Returns true if the manifest might contain path-deleted files.
-  bool CanContainDroppedFiles() const;
+  Result<bool> CanContainDroppedFiles(const ManifestFile& manifest) const;
 
   /// \brief Returns true if the manifest possibly contains any deleted file.
-  bool CanContainDeletedFiles(const ManifestFile& manifest,
-                              const TableMetadata& metadata);
-
-  /// \brief Get or create a ManifestEvaluator for the given spec and expression.
-  Result<ManifestEvaluator*> GetManifestEvaluator(const TableMetadata& metadata,
-                                                  int32_t spec_id, const DeleteExpr& de);
-
-  /// \brief Get or create a ResidualEvaluator for the given spec and expression.
-  Result<ResidualEvaluator*> GetResidualEvaluator(const TableMetadata& metadata,
-                                                  int32_t spec_id, const DeleteExpr& de);
+  Result<bool> CanContainDeletedFiles(const ManifestFile& manifest,
+                                      const std::shared_ptr<Schema>& schema,
+                                      const PartitionSpecsById& specs_by_id,
+                                      bool trust_manifest_references);
+
+  bool CanTrustManifestReferences(
+      const std::vector<const ManifestFile*>& manifests) const;
+
+  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);
+
+  Result<bool> ManifestHasDeletedFiles(const std::vector<ManifestEntry>& entries,
+                                       const std::shared_ptr<Schema>& schema,
+                                       const PartitionSpecsById& specs_by_id,
+                                       int32_t manifest_spec_id);
+
+  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);
+
+  Status ValidateRequiredDeletes(
+      const std::unordered_set<std::string>& found_paths) const;
+
+  /// \brief Get or create a ManifestEvaluator for the given spec.
+  Result<ManifestEvaluator*> GetManifestEvaluator(const std::shared_ptr<Schema>& schema,
+                                                  const PartitionSpecsById& specs_by_id,
+                                                  int32_t spec_id);
+
+  /// \brief Get or create a ResidualEvaluator for the given spec.
+  Result<ResidualEvaluator*> GetResidualEvaluator(const std::shared_ptr<Schema>& schema,
+                                                  const PartitionSpecsById& specs_by_id,
+                                                  int32_t spec_id);
 
   /// \brief Check whether a single entry should be deleted.
-  Result<bool> ShouldDelete(const ManifestEntry& entry, const TableMetadata& metadata,
+  Result<bool> ShouldDelete(const ManifestEntry& entry,
+                            const std::shared_ptr<Schema>& schema,
+                            const PartitionSpecsById& specs_by_id,
                             int32_t manifest_spec_id);
 
   const ManifestContent manifest_content_;
   std::shared_ptr<FileIO> file_io_;
 
-  std::vector<DeleteExpr> delete_exprs_;
+  std::shared_ptr<Expression> delete_expr_;
   std::unordered_set<std::string> delete_paths_;
   DataFileSet delete_files_;
   PartitionSet drop_partitions_;
   bool fail_missing_delete_paths_{false};
   bool fail_any_delete_{false};
   bool case_sensitive_{true};
 
-  // Cache: (spec_id, expr_index) → ManifestEvaluator
-  std::unordered_map<int32_t, std::vector<std::unique_ptr<ManifestEvaluator>>>
+  std::unordered_map<int32_t, std::unique_ptr<ManifestEvaluator>>
       manifest_evaluator_cache_;
-  // Cache: (spec_id, expr_index) → ResidualEvaluator
-  std::unordered_map<int32_t, std::vector<std::unique_ptr<ResidualEvaluator>>>
+  std::unordered_map<int32_t, std::unique_ptr<ResidualEvaluator>>
       residual_evaluator_cache_;
 };
 

src/iceberg/manifest/manifest_filter_manager.h

@@ -20,6 +20,8 @@
 #include "iceberg/manifest/manifest_merge_manager.h"
 
 #include <algorithm>
+#include <array>
+#include <iterator>
 #include <map>
 #include <ranges>
 #include <utility>
@@ -43,70 +45,76 @@ Result<std::vector<ManifestFile>> ManifestMergeManager::MergeManifests(
     const std::vector<ManifestFile>& new_manifests, int64_t snapshot_id,
     const TableMetadata& metadata, std::shared_ptr<FileIO> file_io,
     const ManifestWriterFactory& writer_factory) {
-  // Combine new then existing (new-first ordering is preserved in output)
-  std::vector<ManifestFile> all;
+  // Combine new then existing (new-first ordering is preserved in output).
+  auto to_manifest_ptr = [](const ManifestFile& manifest) { return &manifest; };
+  auto manifest_ranges = std::array{
+      new_manifests | std::views::transform(to_manifest_ptr),
+      existing_manifests | std::views::transform(to_manifest_ptr),
+  };
+
+  std::vector<const ManifestFile*> all;
   all.reserve(new_manifests.size() + existing_manifests.size());
-  all.insert(all.end(), new_manifests.begin(), new_manifests.end());
-  all.insert(all.end(), existing_manifests.begin(), existing_manifests.end());
+  std::ranges::copy(manifest_ranges | std::views::join, std::back_inserter(all));
 
   if (all.empty() || !merge_enabled_) {
-    return all;
+    return all |
+           std::views::transform([](const ManifestFile* manifest) { return *manifest; }) |
+           std::ranges::to<std::vector<ManifestFile>>();
   }
 
-  // Match Java's separate DataFileMergeManager/DeleteFileMergeManager behavior by
-  // tracking the first (newest) manifest independently per content type.
+  // Track the first (newest) manifest independently per content type.
   std::map<ManifestContent, const ManifestFile*> first_by_content;
-  for (const auto& manifest : all) {
-    first_by_content.try_emplace(manifest.content, &manifest);
-  }
+  std::ranges::for_each(all, [&first_by_content](const ManifestFile* manifest) {
+    first_by_content.try_emplace(manifest->content, manifest);
+  });
 
-  // Group manifests by (partition_spec_id, content) — never merge across specs or
-  // content types.  Use reverse spec ordering to match Java's reverse-TreeMap behaviour,
-  // which is observable in v3 tables where first-row IDs are assigned in output order.
+  // Group manifests by (partition_spec_id, content), never merging across specs or
+  // content types. Reverse spec ordering preserves v3 first-row-id assignment order.
   using GroupKey = std::pair<int32_t, ManifestContent>;
-  std::map<GroupKey, std::vector<ManifestFile>, std::greater<>> by_spec;
-  for (const auto& m : all) {
-    by_spec[{m.partition_spec_id, m.content}].push_back(m);
-  }
+  auto group_key = [](const ManifestFile* manifest) {
+    return GroupKey{manifest->partition_spec_id, manifest->content};
+  };
+
+  std::map<GroupKey, std::vector<const ManifestFile*>, std::greater<>> by_spec;
+  std::ranges::for_each(all, [&by_spec, &group_key](const ManifestFile* manifest) {
+    by_spec[group_key(manifest)].push_back(manifest);
+  });
 
   std::vector<ManifestFile> result;
   result.reserve(all.size());
   for (auto& [key, group] : by_spec) {
     const auto* first = first_by_content.at(key.second);
-    ICEBERG_ASSIGN_OR_RAISE(auto merged, MergeGroup(group, *first, snapshot_id, metadata,
+    ICEBERG_ASSIGN_OR_RAISE(auto merged, MergeGroup(group, first, snapshot_id, metadata,
                                                     file_io, writer_factory));
-    result.insert(result.end(), std::make_move_iterator(merged.begin()),
-                  std::make_move_iterator(merged.end()));
+    std::ranges::move(merged, std::back_inserter(result));
   }
   return result;
 }
 
 Result<std::vector<ManifestFile>> ManifestMergeManager::MergeGroup(
-    const std::vector<ManifestFile>& group, const ManifestFile& first,
+    const std::vector<const ManifestFile*>& group, const ManifestFile* first,
     int64_t snapshot_id, const TableMetadata& metadata, std::shared_ptr<FileIO> file_io,
     const ManifestWriterFactory& writer_factory) {
-  // Mirror Java's ListPacker.packEnd(group, ManifestFile::length) with lookback 1:
+  // Match packEnd(group, ManifestFile::length) with lookback 1:
   //   1. Process manifests in reverse order (oldest-first).
   //   2. Greedy forward-pack with lookback=1: emit the current bin when the next item
   //      doesn't fit, then start a new bin.
   //   3. Reverse each bin (restoring original item order within a bin).
   //   4. Reverse the bin list (newest manifest's bin ends up first).
-  // Effect: the newest manifest is in the first, possibly under-filled, bin — exactly
-  // what Java's comment describes ("the manifest that gets under-filled is the first one,
-  // which will be merged the next time").
-  std::vector<std::vector<ManifestFile>> bins;
-  std::vector<ManifestFile> current_bin;
+  // Effect: the newest manifest is in the first, possibly under-filled, bin.
+  std::vector<std::vector<const ManifestFile*>> bins;
+  std::vector<const ManifestFile*> current_bin;
   int64_t bin_size = 0;
 
-  for (const auto& manifest : std::views::reverse(group)) {
+  for (const auto* manifest : std::views::reverse(group)) {
     if (!current_bin.empty() &&
-        bin_size + manifest.manifest_length > target_size_bytes_) {
+        bin_size + manifest->manifest_length > target_size_bytes_) {
       bins.push_back(std::move(current_bin));
       current_bin.clear();
       bin_size = 0;
     }
     current_bin.push_back(manifest);
-    bin_size += manifest.manifest_length;
+    bin_size += manifest->manifest_length;
   }
   if (!current_bin.empty()) {
     bins.push_back(std::move(current_bin));
@@ -118,13 +126,17 @@ Result<std::vector<ManifestFile>> ManifestMergeManager::MergeGroup(
   std::ranges::reverse(bins);
 
   // Process each bin: if the bin contains the newest manifest and is too small,
-  // pass its contents through unchanged (mirrors Java's minCountToMerge logic).
+  // pass its contents through unchanged.
   std::vector<ManifestFile> result;
   result.reserve(group.size());
+  // TODO(Guotao): Flush independent bins in parallel and cache successful merged bins
+  // for commit retries.
   for (auto& bin : bins) {
     bool contains_first = std::ranges::find(bin, first) != bin.end();
     if (contains_first && std::cmp_less(bin.size(), min_count_to_merge_)) {
-      result.insert(result.end(), bin.begin(), bin.end());
+      for (const auto* manifest : bin) {
+        result.push_back(*manifest);
+      }
     } else {
       ICEBERG_ASSIGN_OR_RAISE(
           auto merged, FlushBin(bin, snapshot_id, metadata, file_io, writer_factory));
@@ -136,23 +148,23 @@ Result<std::vector<ManifestFile>> ManifestMergeManager::MergeGroup(
 }
 
 Result<ManifestFile> ManifestMergeManager::FlushBin(
-    const std::vector<ManifestFile>& bin, int64_t snapshot_id,
+    const std::vector<const ManifestFile*>& bin, int64_t snapshot_id,
     const TableMetadata& metadata, std::shared_ptr<FileIO> file_io,
     const ManifestWriterFactory& writer_factory) {
   // A single-manifest bin requires no merging.
-  if (bin.size() == 1) return bin[0];
+  if (bin.size() == 1) return *bin[0];
 
-  const ManifestFile& first = bin[0];
+  const ManifestFile& first = *bin[0];
   int32_t spec_id = first.partition_spec_id;
 
   ICEBERG_ASSIGN_OR_RAISE(auto schema, metadata.Schema());
   ICEBERG_ASSIGN_OR_RAISE(auto spec, metadata.PartitionSpecById(spec_id));
 
   ICEBERG_ASSIGN_OR_RAISE(auto writer, writer_factory(spec_id, first.content));
 
-  for (const auto& manifest : bin) {
+  for (const auto* manifest : bin) {
     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->Entries());
     for (const auto& entry : entries) {
       bool is_current =
@@ -175,4 +187,5 @@ Result<ManifestFile> ManifestMergeManager::FlushBin(
   ICEBERG_RETURN_UNEXPECTED(writer->Close());
   return writer->ToManifestFile();
 }
+
 }  // namespace iceberg