fix: fast append should not contain deleted file during retry process

Author: HuaHuaY

src/iceberg/update/fast_append.cc

@@ -26,7 +26,7 @@
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/manifest/manifest_util_internal.h"
 #include "iceberg/snapshot.h"
-#include "iceberg/table.h"
+#include "iceberg/table.h"  // IWYU pragma: keep
 #include "iceberg/table_metadata.h"
 #include "iceberg/table_properties.h"
 #include "iceberg/transaction.h"
@@ -58,7 +58,7 @@ FastAppend& FastAppend::AppendFile(const std::shared_ptr<DataFile>& file) {
   auto [iter, inserted] = data_files.insert(file);
   if (inserted) {
     has_new_files_ = true;
-    ICEBERG_BUILDER_RETURN_IF_ERROR(summary_.AddedFile(*spec, *file));
+    ICEBERG_BUILDER_RETURN_IF_ERROR(added_data_files_summary_.AddedFile(*spec, *file));
   }
 
   return *this;
@@ -75,11 +75,13 @@ FastAppend& FastAppend::AppendManifest(const ManifestFile& manifest) {
                         "Sequence number must be assigned during commit");
 
   if (can_inherit_snapshot_id() && manifest.added_snapshot_id == kInvalidSnapshotId) {
-    summary_.AddedManifest(manifest);
+    appended_manifests_summary_.AddedManifest(manifest);
     append_manifests_.push_back(manifest);
   } else {
     // The manifest must be rewritten with this update's snapshot ID
-    ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest, CopyManifest(manifest));
+    ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto copied_manifest,
+                                     CopyManifest(manifest, /*update_summary=*/true));
+    append_manifests_to_copy_.push_back(manifest);
     rewritten_append_manifests_.push_back(std::move(copied_manifest));
   }
 
@@ -93,6 +95,14 @@ Result<std::vector<ManifestFile>> FastAppend::Apply(
   std::vector<ManifestFile> manifests;
 
   ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests());
+  if (rewritten_append_manifests_.empty() && !append_manifests_to_copy_.empty()) {
+    for (const auto& manifest : append_manifests_to_copy_) {
+      ICEBERG_ASSIGN_OR_RAISE(auto copied_manifest,
+                              CopyManifest(manifest, /*update_summary=*/false));
+      rewritten_append_manifests_.push_back(std::move(copied_manifest));
+    }
+  }
+
   manifests.reserve(new_written_manifests.size() + append_manifests_.size() +
                     rewritten_append_manifests_.size());
   if (!new_written_manifests.empty()) {
@@ -126,11 +136,23 @@ Result<std::vector<ManifestFile>> FastAppend::Apply(
 }
 
 std::unordered_map<std::string, std::string> FastAppend::Summary() {
+  summary_.Clear();
   summary_.SetPartitionSummaryLimit(
       base().properties.Get(TableProperties::kWritePartitionSummaryLimit));
+  summary_.Merge(added_data_files_summary_);
+  summary_.Merge(appended_manifests_summary_);
+  for (const auto& [property, value] : custom_summary_properties_) {
+    summary_.Set(property, value);
+  }
   return summary_.Build();
 }
 
+void FastAppend::SetSummaryProperty(const std::string& property,
+                                    const std::string& value) {
+  custom_summary_properties_[property] = value;
+  SnapshotUpdate::SetSummaryProperty(property, value);
+}
+
 Status FastAppend::CleanUncommitted(const std::unordered_set<std::string>& committed) {
   // Clean up new manifests that were written but not committed
   if (!new_manifests_.empty()) {
@@ -151,6 +173,7 @@ Status FastAppend::CleanUncommitted(const std::unordered_set<std::string>& commi
         std::ignore = DeleteFile(manifest.manifest_path);
       }
     }
+    rewritten_append_manifests_.clear();
   }
   return {};
 }
@@ -161,14 +184,15 @@ bool FastAppend::CleanupAfterCommit() const {
   // 1.) Appended manifests are never rewritten
   // 2.) Manifests which are written out as part of AppendFile are already cleaned
   //     up between commit attempts in WriteNewManifests
-  return !rewritten_append_manifests_.empty();
+  return !rewritten_append_manifests_.empty() || !append_manifests_to_copy_.empty();
 }
 
 Result<std::shared_ptr<PartitionSpec>> FastAppend::Spec(int32_t spec_id) {
   return base().PartitionSpecById(spec_id);
 }
 
-Result<ManifestFile> FastAppend::CopyManifest(const ManifestFile& manifest) {
+Result<ManifestFile> FastAppend::CopyManifest(const ManifestFile& manifest,
+                                              bool update_summary) {
   const TableMetadata& current = base();
   ICEBERG_ASSIGN_OR_RAISE(auto schema, current.Schema());
   ICEBERG_ASSIGN_OR_RAISE(auto spec,
@@ -180,7 +204,8 @@ Result<ManifestFile> FastAppend::CopyManifest(const ManifestFile& manifest) {
 
   // Copy the manifest with the new snapshot ID.
   return CopyAppendManifest(manifest, ctx_->table->io(), schema, spec, snapshot_id,
-                            new_manifest_path, current.format_version, &summary_);
+                            new_manifest_path, current.format_version,
+                            update_summary ? &appended_manifests_summary_ : nullptr);
 }
 
 Result<std::vector<ManifestFile>> FastAppend::WriteNewManifests() {

src/iceberg/update/fast_append.h

@@ -72,6 +72,7 @@ class ICEBERG_EXPORT FastAppend : public SnapshotUpdate {
       const TableMetadata& metadata_to_update,
       const std::shared_ptr<Snapshot>& snapshot) override;
   std::unordered_map<std::string, std::string> Summary() override;
+  void SetSummaryProperty(const std::string& property, const std::string& value) override;
   Status CleanUncommitted(const std::unordered_set<std::string>& committed) override;
   bool CleanupAfterCommit() const override;
 
@@ -85,7 +86,7 @@ class ICEBERG_EXPORT FastAppend : public SnapshotUpdate {
   ///
   /// \param manifest The manifest to copy
   /// \return The copied manifest file
-  Result<ManifestFile> CopyManifest(const ManifestFile& manifest);
+  Result<ManifestFile> CopyManifest(const ManifestFile& manifest, bool update_summary);
 
   /// \brief Write new manifests for the accumulated data files.
   ///
@@ -95,7 +96,11 @@ class ICEBERG_EXPORT FastAppend : public SnapshotUpdate {
  private:
   std::string table_name_;
   std::unordered_map<int32_t, DataFileSet> new_data_files_by_spec_;
+  SnapshotSummaryBuilder added_data_files_summary_;
+  SnapshotSummaryBuilder appended_manifests_summary_;
+  std::unordered_map<std::string, std::string> custom_summary_properties_;
   std::vector<ManifestFile> append_manifests_;
+  std::vector<ManifestFile> append_manifests_to_copy_;
   std::vector<ManifestFile> rewritten_append_manifests_;
   std::vector<ManifestFile> new_manifests_;
   bool has_new_files_{false};