feat: add rewrite manifests

Author: zhjwpku

src/iceberg/CMakeLists.txt

@@ -110,6 +110,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/manifest/rolling_manifest_writer.cc

@@ -54,6 +54,13 @@ Status RollingManifestWriter::WriteExistingEntry(
   return {};
 }
 
+Status RollingManifestWriter::WriteExistingEntry(const ManifestEntry& entry) {
+  ICEBERG_ASSIGN_OR_RAISE(auto* writer, CurrentWriter());
+  ICEBERG_RETURN_UNEXPECTED(writer->WriteExistingEntry(entry));
+  current_file_rows_++;
+  return {};
+}
+
 Status RollingManifestWriter::WriteDeletedEntry(
     std::shared_ptr<DataFile> file, int64_t data_sequence_number,
     std::optional<int64_t> file_sequence_number) {

src/iceberg/manifest/rolling_manifest_writer.h

@@ -76,6 +76,9 @@ class ICEBERG_EXPORT RollingManifestWriter {
                             int64_t data_sequence_number,
                             std::optional<int64_t> file_sequence_number = std::nullopt);
 
+  /// \brief Add an existing entry while preserving snapshot and sequence fields.
+  Status WriteExistingEntry(const ManifestEntry& entry);
+
   /// \brief Add a delete entry for a file.
   ///
   /// \param file a deleted data file

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/meson.build

@@ -135,6 +135,7 @@ iceberg_sources = files(
     '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/table.cc

@@ -37,6 +37,7 @@
 #include "iceberg/update/merge_append.h"
 #include "iceberg/update/overwrite_files.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"
@@ -252,6 +253,12 @@ Result<std::shared_ptr<RewriteFiles>> Table::NewRewriteFiles() {
   return RewriteFiles::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));
@@ -371,6 +378,10 @@ Result<std::shared_ptr<RewriteFiles>> StaticTable::NewRewriteFiles() {
   return NotSupported("Cannot create a rewrite files 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

@@ -192,6 +192,9 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   /// changes.
   virtual Result<std::shared_ptr<RewriteFiles>> NewRewriteFiles();
 
+  /// \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();
 
@@ -269,6 +272,8 @@ class ICEBERG_EXPORT StaticTable : public Table {
 
   Result<std::shared_ptr<RewriteFiles>> NewRewriteFiles() override;
 
+  Result<std::shared_ptr<RewriteManifests>> NewRewriteManifests() override;
+
   Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager() override;
 
  private:

src/iceberg/test/CMakeLists.txt

@@ -236,6 +236,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

src/iceberg/test/merge_append_test.cc

@@ -45,13 +45,11 @@
 #include "iceberg/table_properties.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/mock_catalog.h"
-#include "iceberg/test/test_resource.h"
 #include "iceberg/test/update_test_base.h"
 #include "iceberg/transaction.h"
 #include "iceberg/update/fast_append.h"
 #include "iceberg/update/update_partition_spec.h"
 #include "iceberg/update/update_properties.h"
-#include "iceberg/util/uuid.h"
 
 namespace iceberg {
 
@@ -68,12 +66,21 @@ class MergeAppendTestBase : public UpdateTestBase {
 
   std::string TableName() const override { return "minimal_table"; }
 
-  void SetUp() override {
-    table_ident_ = TableIdentifier{.name = TableName()};
-    table_location_ = "/warehouse/" + TableName();
+  std::string MetadataResource() const override {
+    switch (format_version()) {
+      case 1:
+        return "TableMetadataV1Valid.json";
+      case 2:
+        return "TableMetadataV2ValidMinimal.json";
+      case 3:
+        return "TableMetadataV3ValidMinimal.json";
+      default:
+        return "TableMetadataV2ValidMinimal.json";
+    }
+  }
 
-    InitializeFileIO();
-    RegisterMinimalTable(format_version());
+  void SetUp() override {
+    UpdateTestBase::SetUp();
 
     ICEBERG_UNWRAP_OR_FAIL(spec_, table_->spec());
     ICEBERG_UNWRAP_OR_FAIL(schema_, table_->schema());
@@ -84,21 +91,6 @@ class MergeAppendTestBase : public UpdateTestBase {
     file_d_ = MakeDataFile("/data/file_d.parquet", /*partition_x=*/4L);
   }
 
-  void RegisterMinimalTable(int8_t format_version) {
-    auto metadata_location = std::format("{}/metadata/00001-{}.metadata.json",
-                                         table_location_, Uuid::GenerateV7().ToString());
-    ICEBERG_UNWRAP_OR_FAIL(
-        auto metadata, ReadTableMetadataFromResource("TableMetadataV2ValidMinimal.json"));
-    metadata->format_version = format_version;
-    metadata->location = table_location_;
-    metadata->next_row_id = TableMetadata::kInitialRowId;
-
-    ASSERT_THAT(TableMetadataUtil::Write(*file_io_, metadata_location, *metadata),
-                IsOk());
-    ICEBERG_UNWRAP_OR_FAIL(table_,
-                           catalog_->RegisterTable(table_ident_, metadata_location));
-  }
-
   virtual int8_t format_version() const {
     return TableMetadata::kDefaultTableFormatVersion;
   }