address minor issues

Author: wgtmac

src/iceberg/test/fast_append_test.cc

@@ -20,13 +20,20 @@
 #include "iceberg/update/fast_append.h"
 
 #include <format>
+#include <optional>
+#include <string>
+#include <vector>
 
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
 #include "iceberg/avro/avro_register.h"
+#include "iceberg/constants.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/table_metadata.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/test_resource.h"
@@ -72,6 +79,23 @@ class FastAppendTest : public UpdateTestBase {
     return data_file;
   }
 
+  Result<ManifestFile> WriteManifest(
+      const std::string& path, const std::vector<std::shared_ptr<DataFile>>& files) {
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto writer, ManifestWriter::MakeWriter(table_->metadata()->format_version,
+                                                kInvalidSnapshotId, path, file_io_, spec_,
+                                                schema_, ManifestContent::kData));
+    for (const auto& file : files) {
+      ManifestEntry entry;
+      entry.status = ManifestStatus::kAdded;
+      entry.snapshot_id = std::nullopt;
+      entry.data_file = file;
+      ICEBERG_RETURN_UNEXPECTED(writer->WriteAddedEntry(entry));
+    }
+    ICEBERG_RETURN_UNEXPECTED(writer->Close());
+    return writer->ToManifestFile();
+  }
+
   std::shared_ptr<PartitionSpec> spec_;
   std::shared_ptr<Schema> schema_;
   std::shared_ptr<DataFile> file_a_;
@@ -90,6 +114,9 @@ TEST_F(FastAppendTest, AppendDataFile) {
   EXPECT_EQ(snapshot->summary.at("added-data-files"), "1");
   EXPECT_EQ(snapshot->summary.at("added-records"), "100");
   EXPECT_EQ(snapshot->summary.at("added-files-size"), "1024");
+  EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kManifestsCreated), "1");
+  EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kManifestsKept), "0");
+  EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kManifestsReplaced), "0");
 }
 
 TEST_F(FastAppendTest, AppendMultipleDataFiles) {
@@ -172,6 +199,40 @@ TEST_F(FastAppendTest, FinalizeIgnoresCleanupDeleteFailure) {
               IsOk());
 }
 
+TEST_F(FastAppendTest, RetryCopiesAppendManifestAgain) {
+  table_->metadata()->format_version = 1;
+  const auto path = table_location_ + "/metadata/input.avro";
+  ICEBERG_UNWRAP_OR_FAIL(auto manifest, WriteManifest(path, {file_a_}));
+
+  std::shared_ptr<FastAppend> fast_append;
+  ICEBERG_UNWRAP_OR_FAIL(fast_append, table_->NewFastAppend());
+  std::vector<std::string> deleted_paths;
+  fast_append->DeleteWith([&](const std::string& deleted_path) {
+    deleted_paths.push_back(deleted_path);
+    return file_io_->DeleteFile(deleted_path);
+  });
+  fast_append->AppendManifest(manifest);
+
+  auto& update = static_cast<SnapshotUpdate&>(*fast_append);
+  // First Apply() copies the input manifest because v1 cannot inherit snapshot IDs.
+  ICEBERG_UNWRAP_OR_FAIL(auto first_apply, update.Apply());
+  SnapshotCache first_cache(first_apply.snapshot.get());
+  ICEBERG_UNWRAP_OR_FAIL(auto first_manifests, first_cache.Manifests(file_io_));
+  ASSERT_EQ(first_manifests.size(), 1U);
+  const auto first_rewritten_path = first_manifests[0].manifest_path;
+  EXPECT_NE(first_rewritten_path, path);
+
+  // Second Apply() simulates retry cleanup, then copies the original manifest again.
+  ICEBERG_UNWRAP_OR_FAIL(auto second_apply, update.Apply());
+  EXPECT_THAT(deleted_paths, testing::Contains(first_rewritten_path));
+
+  SnapshotCache second_cache(second_apply.snapshot.get());
+  ICEBERG_UNWRAP_OR_FAIL(auto second_manifests, second_cache.Manifests(file_io_));
+  ASSERT_EQ(second_manifests.size(), 1U);
+  EXPECT_NE(second_manifests[0].manifest_path, path);
+  EXPECT_NE(second_manifests[0].manifest_path, first_rewritten_path);
+}
+
 TEST_F(FastAppendTest, AppendDuplicateFile) {
   std::shared_ptr<FastAppend> fast_append;
   ICEBERG_UNWRAP_OR_FAIL(fast_append, table_->NewFastAppend());

src/iceberg/update/fast_append.cc

@@ -95,6 +95,8 @@ Result<std::vector<ManifestFile>> FastAppend::Apply(
   std::vector<ManifestFile> manifests;
 
   ICEBERG_ASSIGN_OR_RAISE(auto new_written_manifests, WriteNewManifests());
+  // A retry cleanup deletes copied append manifests and clears the rewritten
+  // list; rebuild them from the original appended manifests before re-applying.
   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,
@@ -132,6 +134,9 @@ Result<std::vector<ManifestFile>> FastAppend::Apply(
                      snapshot_manifests.end());
   }
 
+  manifest_count_summary_ =
+      BuildManifestCountSummary(manifests, /*replaced_manifests_count=*/0);
+
   return manifests;
 }
 
@@ -144,6 +149,7 @@ std::unordered_map<std::string, std::string> FastAppend::Summary() {
   for (const auto& [property, value] : custom_summary_properties_) {
     summary_.Set(property, value);
   }
+  summary_.Merge(manifest_count_summary_);
   return summary_.Build();
 }
 
@@ -179,9 +185,9 @@ Status FastAppend::CleanUncommitted(const std::unordered_set<std::string>& commi
 }
 
 bool FastAppend::CleanupAfterCommit() const {
-  // Cleanup after committing is disabled for FastAppend unless there are
-  // rewritten_append_manifests_ because:
-  // 1.) Appended manifests are never rewritten
+  // Cleanup after committing is disabled for FastAppend unless append manifests
+  // were copied or need to be copied on retry because:
+  // 1.) Directly 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() || !append_manifests_to_copy_.empty();

src/iceberg/update/fast_append.h

@@ -85,6 +85,7 @@ class ICEBERG_EXPORT FastAppend : public SnapshotUpdate {
   /// \brief Copy a manifest file with a new snapshot ID.
   ///
   /// \param manifest The manifest to copy
+  /// \param update_summary Whether to add copied entries to the append summary
   /// \return The copied manifest file
   Result<ManifestFile> CopyManifest(const ManifestFile& manifest, bool update_summary);
 
@@ -96,13 +97,18 @@ class ICEBERG_EXPORT FastAppend : public SnapshotUpdate {
  private:
   std::string table_name_;
   std::unordered_map<int32_t, DataFileSet> new_data_files_by_spec_;
+  // Stable input summaries for retry-safe summary_ rebuilds.
   SnapshotSummaryBuilder added_data_files_summary_;
   SnapshotSummaryBuilder appended_manifests_summary_;
+  // User-provided summary properties restored after summary_ rebuilds.
   std::unordered_map<std::string, std::string> custom_summary_properties_;
   std::vector<ManifestFile> append_manifests_;
+  // Original manifests kept to recreate copied manifests after retry cleanup.
   std::vector<ManifestFile> append_manifests_to_copy_;
   std::vector<ManifestFile> rewritten_append_manifests_;
   std::vector<ManifestFile> new_manifests_;
+  // Manifest count summary from the latest Apply() result.
+  SnapshotSummaryBuilder manifest_count_summary_;
   bool has_new_files_{false};
 };
 

src/iceberg/update/merging_snapshot_update.cc

@@ -944,26 +944,11 @@ Result<std::vector<ManifestFile>> MergingSnapshotUpdate::Apply(
   result.insert(result.end(), std::make_move_iterator(merged_deletes.begin()),
                 std::make_move_iterator(merged_deletes.end()));
 
-  // Manifest count summary: unassigned manifests count as neither created nor kept.
-  int32_t manifests_created = 0;
-  int32_t manifests_kept = 0;
-  for (const auto& m : result) {
-    if (m.added_snapshot_id == snapshot_id) {
-      ++manifests_created;
-    } else if (m.added_snapshot_id != kInvalidSnapshotId) {
-      ++manifests_kept;
-    }
-  }
   int32_t replaced_manifests_count = data_filter_manager_->ReplacedManifestsCount() +
                                      delete_filter_manager_->ReplacedManifestsCount() +
                                      data_merge_manager_->ReplacedManifestsCount() +
                                      delete_merge_manager_->ReplacedManifestsCount();
-  summary_builder().Set(SnapshotSummaryFields::kManifestsCreated,
-                        std::to_string(manifests_created));
-  summary_builder().Set(SnapshotSummaryFields::kManifestsKept,
-                        std::to_string(manifests_kept));
-  summary_builder().Set(SnapshotSummaryFields::kManifestsReplaced,
-                        std::to_string(replaced_manifests_count));
+  summary_builder().Merge(BuildManifestCountSummary(result, replaced_manifests_count));
 
   return result;
 }

src/iceberg/update/snapshot_update.cc

@@ -424,6 +424,28 @@ std::string SnapshotUpdate::ManifestListPath() {
   return ctx_->MetadataFileLocation(filename);
 }
 
+SnapshotSummaryBuilder SnapshotUpdate::BuildManifestCountSummary(
+    std::span<const ManifestFile> manifests, int32_t replaced_manifests_count) {
+  SnapshotSummaryBuilder summary;
+  int32_t manifests_created = 0;
+  int32_t manifests_kept = 0;
+  int64_t snapshot_id = SnapshotId();
+  for (const auto& manifest : manifests) {
+    if (manifest.added_snapshot_id == snapshot_id) {
+      ++manifests_created;
+    } else if (manifest.added_snapshot_id != kInvalidSnapshotId) {
+      ++manifests_kept;
+    }
+  }
+
+  summary.Set(SnapshotSummaryFields::kManifestsCreated,
+              std::to_string(manifests_created));
+  summary.Set(SnapshotSummaryFields::kManifestsKept, std::to_string(manifests_kept));
+  summary.Set(SnapshotSummaryFields::kManifestsReplaced,
+              std::to_string(replaced_manifests_count));
+  return summary;
+}
+
 std::string SnapshotUpdate::ManifestPath() {
   // Generate manifest path
   // Format: {metadata_location}/{uuid}-m{manifest_count}.avro

src/iceberg/update/snapshot_update.h

@@ -216,6 +216,8 @@ class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate {
   std::string ManifestPath();
   std::string ManifestListPath();
   SnapshotSummaryBuilder& summary_builder() { return summary_; }
+  SnapshotSummaryBuilder BuildManifestCountSummary(
+      std::span<const ManifestFile> manifests, int32_t replaced_manifests_count);
 
  private:
   /// \brief Returns the snapshot summary from the implementation and updates totals.