feat: add rewrite manifests

src/iceberg/CMakeLists.txt

@@ -129,6 +129,7 @@ set(ICEBERG_SOURCES
     update/overwrite_files.cc
     update/pending_update.cc
     update/rewrite_files.cc
+    update/rewrite_manifests.cc
     update/row_delta.cc
     update/set_snapshot.cc
     update/snapshot_manager.cc

src/iceberg/cpp20_compatibility_internal.h

@@ -149,6 +149,7 @@
 #include "iceberg/update/pending_update.h"
 #include "iceberg/update/replace_partitions.h"
 #include "iceberg/update/rewrite_files.h"
+#include "iceberg/update/rewrite_manifests.h"
 #include "iceberg/update/row_delta.h"
 #include "iceberg/update/set_snapshot.h"
 #include "iceberg/update/snapshot_manager.h"

src/iceberg/manifest/manifest_util.cc

@@ -19,7 +19,9 @@
 
 #include <memory>
 #include <optional>
+#include <vector>
 
+#include "iceberg/file_reader.h"
 #include "iceberg/inheritable_metadata.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/manifest/manifest_reader.h"
@@ -28,10 +30,36 @@
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
 #include "iceberg/snapshot.h"
+#include "iceberg/table_metadata.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/string_util.h"
 
 namespace iceberg {
 
+Result<int8_t> ReadManifestFormatVersion(const ManifestFile& manifest,
+                                         const std::shared_ptr<FileIO>& file_io) {
+  ReaderOptions options{.path = manifest.manifest_path,
+                        .length = manifest.manifest_length,
+                        .io = file_io,
+                        .projection = std::make_shared<Schema>(
+                            std::vector<SchemaField>{ManifestEntry::kStatus})};
+  ICEBERG_ASSIGN_OR_RAISE(auto reader,
+                          ReaderFactoryRegistry::Open(FileFormatType::kAvro, options));
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata, reader->Metadata());
+  ICEBERG_RETURN_UNEXPECTED(reader->Close());
+  auto version = metadata.find("format-version");
+  if (version == metadata.end()) {
+    return int8_t{1};
+  }
+  ICEBERG_ASSIGN_OR_RAISE(auto format_version,
+                          StringUtils::ParseNumber<int8_t>(version->second));
+  ICEBERG_CHECK(format_version >= 1 &&
+                    format_version <= TableMetadata::kSupportedTableFormatVersion,
+                "Unsupported manifest format version {} in {}", format_version,
+                manifest.manifest_path);
+  return format_version;
+}
+
 Result<ManifestFile> CopyAppendManifest(
     const ManifestFile& manifest, const std::shared_ptr<FileIO>& file_io,
     const std::shared_ptr<Schema>& schema, const std::shared_ptr<PartitionSpec>& spec,
@@ -68,4 +96,43 @@ Result<ManifestFile> CopyAppendManifest(
   return writer->ToManifestFile();
 }
 
+Result<ManifestFile> CopyRewriteManifest(
+    const ManifestFile& manifest, const std::shared_ptr<FileIO>& file_io,
+    const std::shared_ptr<Schema>& schema, const std::shared_ptr<PartitionSpec>& spec,
+    int64_t snapshot_id, const std::string& output_path, int8_t format_version) {
+  ICEBERG_ASSIGN_OR_RAISE(auto source_format_version,
+                          ReadManifestFormatVersion(manifest, file_io));
+  // Require explicit snapshot IDs and preserve them in the copied entries.
+  ICEBERG_ASSIGN_OR_RAISE(auto inheritable_metadata, InheritableMetadataFactory::Empty());
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto reader, ManifestReader::Make(
+                       manifest.manifest_path, manifest.manifest_length, file_io, schema,
+                       spec, std::move(inheritable_metadata), manifest.first_row_id,
+                       /*is_committed=*/false));
+  ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->Entries());
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto writer,
+      ManifestWriter::MakeWriter(format_version, snapshot_id, output_path, file_io, spec,
+                                 schema, manifest.content, manifest.first_row_id));
+  for (auto& entry : entries) {
+    // A rewritten added manifest may only contain existing entries.
+    if (entry.status == ManifestStatus::kAdded) {
+      return ValidationFailed("Cannot add manifest with added files");
+    }
+    if (entry.status == ManifestStatus::kDeleted) {
+      return ValidationFailed("Cannot add manifest with deleted files");
+    }
+    if (source_format_version == 1) {
+      // V1 does not persist sequence numbers. Materialize its legacy defaults
+      // before writing EXISTING entries into a v2 or v3 replacement manifest.
+      entry.sequence_number = entry.sequence_number.value_or(0);
+      entry.file_sequence_number = entry.file_sequence_number.value_or(0);
+    }
+    ICEBERG_RETURN_UNEXPECTED(writer->WriteExistingEntry(entry));
+  }
+  ICEBERG_RETURN_UNEXPECTED(writer->Close());
+  return writer->ToManifestFile();
+}
+
 }  // namespace iceberg

src/iceberg/manifest/manifest_util_internal.h

@@ -32,6 +32,11 @@
 
 namespace iceberg {
 
+/// \brief Read the format version recorded in a manifest's file metadata.
+/// Missing format-version metadata identifies a v1 manifest.
+ICEBERG_EXPORT Result<int8_t> ReadManifestFormatVersion(
+    const ManifestFile& manifest, const std::shared_ptr<FileIO>& file_io);
+
 /// \brief Copy an append manifest with a new snapshot ID.
 ///
 /// This function copies a manifest file that contains only ADDED entries,
@@ -53,4 +58,13 @@ ICEBERG_EXPORT Result<ManifestFile> CopyAppendManifest(
     int64_t snapshot_id, const std::string& output_path, int8_t format_version,
     SnapshotSummaryBuilder* summary_builder = nullptr);
 
+/// \brief Copy a manifest containing only EXISTING entries, preserving their
+/// snapshot IDs, sequence numbers, and row IDs. Missing v1 sequence numbers are
+/// materialized as zero. The caller owns output_path and must clean it up if
+/// copying or committing fails.
+ICEBERG_EXPORT Result<ManifestFile> CopyRewriteManifest(
+    const ManifestFile& manifest, const std::shared_ptr<FileIO>& file_io,
+    const std::shared_ptr<Schema>& schema, const std::shared_ptr<PartitionSpec>& spec,
+    int64_t snapshot_id, const std::string& output_path, int8_t format_version);
+
 }  // namespace iceberg

src/iceberg/manifest/manifest_writer.cc

@@ -31,6 +31,7 @@
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
 #include "iceberg/table_metadata.h"
+#include "iceberg/util/error_util_internal.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg {
@@ -46,7 +47,13 @@ ManifestWriter::ManifestWriter(std::unique_ptr<Writer> writer,
       partition_summary_(
           std::make_unique<PartitionSummary>(*adapter_->partition_type())) {}
 
-ManifestWriter::~ManifestWriter() = default;
+ManifestWriter::~ManifestWriter() {
+  if (!closed_) {
+    // Release the output on failure without flushing uncommitted adapter entries.
+    internal::LogAndIgnoreFailure("Manifest writer cleanup",
+                                  [this] { return writer_->Close(); });
+  }
+}
 
 Status ManifestWriter::WriteAddedEntry(std::shared_ptr<DataFile> file,
                                        std::optional<int64_t> data_sequence_number) {

src/iceberg/manifest/manifest_writer.h

@@ -39,6 +39,8 @@ namespace iceberg {
 /// \brief Write manifest entries to a manifest file.
 class ICEBERG_EXPORT ManifestWriter {
  public:
+  /// Close the underlying output on destruction. Call Close() explicitly to flush
+  /// all entries and report write or close errors before committing the manifest.
   ~ManifestWriter();
 
   /// \brief Write the entry that all its fields are populated correctly.

src/iceberg/manifest/v3_metadata.cc

@@ -194,13 +194,20 @@ Status ManifestFileAdapterV3::Init() {
 }
 
 Status ManifestFileAdapterV3::Append(const ManifestFile& file) {
-  ICEBERG_RETURN_UNEXPECTED(AppendInternal(file));
   if (WrapFirstRowId(file)) {
     if (!next_row_id_.has_value()) {
       return InvalidManifestList("Missing next-row-id for file: {}", file.manifest_path);
     }
-    next_row_id_ = next_row_id_.value() + file.existing_rows_count.value_or(0) +
-                   file.added_rows_count.value_or(0);
+    if (!file.existing_rows_count.has_value() || !file.added_rows_count.has_value()) {
+      return InvalidManifestList("Missing row counts for file: {}", file.manifest_path);
+    }
+  }
+
+  ICEBERG_RETURN_UNEXPECTED(AppendInternal(file));
+
+  if (WrapFirstRowId(file)) {
+    next_row_id_ = next_row_id_.value() + file.existing_rows_count.value() +
+                   file.added_rows_count.value();
   }
   return {};
 }

src/iceberg/table.cc

@@ -38,6 +38,7 @@
 #include "iceberg/update/overwrite_files.h"
 #include "iceberg/update/replace_partitions.h"
 #include "iceberg/update/rewrite_files.h"
+#include "iceberg/update/rewrite_manifests.h"
 #include "iceberg/update/row_delta.h"
 #include "iceberg/update/set_snapshot.h"
 #include "iceberg/update/snapshot_manager.h"
@@ -266,6 +267,12 @@ Result<std::shared_ptr<ReplacePartitions>> Table::NewReplacePartitions() {
   return ReplacePartitions::Make(name().name, std::move(ctx));
 }
 
+Result<std::shared_ptr<RewriteManifests>> Table::NewRewriteManifests() {
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
+  return RewriteManifests::Make(name().name, std::move(ctx));
+}
+
 Result<std::shared_ptr<UpdateStatistics>> Table::NewUpdateStatistics() {
   ICEBERG_ASSIGN_OR_RAISE(
       auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
@@ -390,6 +397,10 @@ Result<std::shared_ptr<ReplacePartitions>> StaticTable::NewReplacePartitions() {
   return NotSupported("Cannot replace partitions for a static table");
 }
 
+Result<std::shared_ptr<RewriteManifests>> StaticTable::NewRewriteManifests() {
+  return NotSupported("Cannot create a rewrite manifests for a static table");
+}
+
 Result<std::shared_ptr<SnapshotManager>> StaticTable::NewSnapshotManager() {
   return NotSupported("Cannot create a snapshot manager for a static table");
 }

src/iceberg/table.h

@@ -209,6 +209,9 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   /// the changes.
   virtual Result<std::shared_ptr<ReplacePartitions>> NewReplacePartitions();
 
+  /// \brief Create a new RewriteManifests to rewrite manifest layout.
+  virtual Result<std::shared_ptr<RewriteManifests>> NewRewriteManifests();
+
   /// \brief Create a new SnapshotManager to manage snapshots and snapshot references.
   virtual Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
 
@@ -293,6 +296,8 @@ class ICEBERG_EXPORT StaticTable : public Table {
 
   Result<std::shared_ptr<ReplacePartitions>> NewReplacePartitions() override;
 
+  Result<std::shared_ptr<RewriteManifests>> NewRewriteManifests() override;
+
   Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager() override;
 
  private:

src/iceberg/test/CMakeLists.txt

@@ -243,6 +243,7 @@ if(ICEBERG_BUILD_BUNDLE)
                    merging_snapshot_update_test.cc
                    name_mapping_update_test.cc
                    rewrite_files_test.cc
+                   rewrite_manifests_test.cc
                    row_delta_test.cc
                    snapshot_manager_test.cc
                    transaction_test.cc