feat: implement RewriteFiles operation (#751)

This PR implements the RewriteFiles operation for the Iceberg C++
library.

## Changes

- Added RewriteFiles update class in src/iceberg/update/rewrite_files.h
and .cc
- Added RewriteFiles to the type forward declarations in
src/iceberg/type_fwd.h
- Registered RewiteFiles as a merging snapshot update in
src/iceberg/table.h and .cc
- Added RewriteFiles support in transaction (src/iceberg/transaction.h
and .cc)
- Added comprehensive tests in src/iceberg/test/rewrite_files_test.cc
- Added merge test for RewriteFiles in
src/iceberg/test/merging_snapshot_update_test.cc

Author: WZhuo

src/iceberg/CMakeLists.txt

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

src/iceberg/meson.build

@@ -133,6 +133,7 @@ iceberg_sources = files(
     'update/merging_snapshot_update.cc',
     'update/overwrite_files.cc',
     'update/pending_update.cc',
+    'update/rewrite_files.cc',
     'update/row_delta.cc',
     'update/set_snapshot.cc',
     'update/snapshot_manager.cc',

src/iceberg/table.cc

@@ -36,6 +36,7 @@
 #include "iceberg/update/fast_append.h"
 #include "iceberg/update/merge_append.h"
 #include "iceberg/update/overwrite_files.h"
+#include "iceberg/update/rewrite_files.h"
 #include "iceberg/update/row_delta.h"
 #include "iceberg/update/set_snapshot.h"
 #include "iceberg/update/snapshot_manager.h"
@@ -245,6 +246,12 @@ Result<std::shared_ptr<OverwriteFiles>> Table::NewOverwrite() {
   return OverwriteFiles::Make(name().name, std::move(ctx));
 }
 
+Result<std::shared_ptr<RewriteFiles>> Table::NewRewriteFiles() {
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
+  return RewriteFiles::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));
@@ -360,6 +367,10 @@ Result<std::shared_ptr<OverwriteFiles>> StaticTable::NewOverwrite() {
   return NotSupported("Cannot create an overwrite for a static table");
 }
 
+Result<std::shared_ptr<RewriteFiles>> StaticTable::NewRewriteFiles() {
+  return NotSupported("Cannot create a rewrite files 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

@@ -188,6 +188,10 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   /// \brief Create a new OverwriteFiles to overwrite data files and commit the changes.
   virtual Result<std::shared_ptr<OverwriteFiles>> NewOverwrite();
 
+  /// \brief Create a new RewriteFiles to replace files in this table and commit the
+  /// changes.
+  virtual Result<std::shared_ptr<RewriteFiles>> NewRewriteFiles();
+
   /// \brief Create a new SnapshotManager to manage snapshots and snapshot references.
   virtual Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
 
@@ -263,6 +267,8 @@ class ICEBERG_EXPORT StaticTable : public Table {
 
   Result<std::shared_ptr<OverwriteFiles>> NewOverwrite() override;
 
+  Result<std::shared_ptr<RewriteFiles>> NewRewriteFiles() override;
+
   Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager() override;
 
  private:

src/iceberg/test/.meson.build.swp

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

src/iceberg/test/CMakeLists.txt

@@ -269,13 +269,13 @@ class MergeAppendTestBase : public UpdateTestBase {
   // both the main branch and a named branch without per-call boilerplate.
   Result<std::shared_ptr<MergeAppend>> NewBranchMergeAppend() {
     ICEBERG_ASSIGN_OR_RAISE(auto append, table_->NewMergeAppend());
-    append->SetTargetBranch(branch());
+    append->ToBranch(branch());
     return append;
   }
 
   Result<std::shared_ptr<FastAppend>> NewBranchFastAppend() {
     ICEBERG_ASSIGN_OR_RAISE(auto append, table_->NewFastAppend());
-    append->SetTargetBranch(branch());
+    append->ToBranch(branch());
     return append;
   }
 
@@ -1127,7 +1127,7 @@ TEST_P(MergeAppendTest, Failure) {
 
   ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
   ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewMergeAppend());
-  append->SetTargetBranch(branch());
+  append->ToBranch(branch());
   append->AppendFile(file_b_);
   EXPECT_THAT(append->Commit(), IsOk());
 
@@ -1166,7 +1166,7 @@ TEST_P(MergeAppendTest, AppendManifestCleanup) {
 
   ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
   ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewMergeAppend());
-  append->SetTargetBranch(branch());
+  append->ToBranch(branch());
   append->AppendManifest(manifest);
   EXPECT_THAT(append->Commit(), IsOk());
 
@@ -1215,7 +1215,7 @@ TEST_P(MergeAppendTest, Recovery) {
 
   ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
   ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewMergeAppend());
-  append->SetTargetBranch(branch());
+  append->ToBranch(branch());
   append->AppendFile(file_b_);
   EXPECT_THAT(append->Commit(), IsOk());
 
@@ -1336,7 +1336,7 @@ TEST_P(MergeAppendTest, AppendManifestFailureWithSnapshotIdInheritance) {
 TEST_P(MergeAppendTest, TransactionNewMergeAppendCommits) {
   ICEBERG_UNWRAP_OR_FAIL(auto txn, table_->NewTransaction());
   ICEBERG_UNWRAP_OR_FAIL(auto append, txn->NewMergeAppend());
-  append->SetTargetBranch(branch());
+  append->ToBranch(branch());
   append->AppendFile(file_a_);
   EXPECT_THAT(append->Commit(), IsOk());
   ICEBERG_UNWRAP_OR_FAIL(auto committed_table, txn->Commit());

src/iceberg/test/merge_append_test.cc

@@ -1450,6 +1450,43 @@ TEST_F(MergingSnapshotUpdateTest,
               IsError(ErrorKind::kValidationFailed));
 }
 
+TEST_F(MergingSnapshotUpdateTest,
+       ValidateNoNewDeletesForDataFilesIgnoresEqualityDeletesWhenFlagIsTrue) {
+  // This tests the behavior that RewriteFiles::SetDataSequenceNumber() and
+  // RewriteFiles::RewriteDataFiles() enable: when a data sequence number is
+  // set for rewritten data files, concurrent equality deletes at higher
+  // sequence numbers still apply to the new files and are NOT a conflict.
+  // Only position deletes should still fail (tested separately by
+  // ValidateNoNewDeletesForDataFilesFailsOnPositionDeleteWhenIgnoringEqualityDeletes).
+  CommitFileA();
+  ICEBERG_UNWRAP_OR_FAIL(auto first_snapshot, table_->current_snapshot());
+
+  auto del_file = MakeEqualityDeleteFile("/delete/del_a.parquet", 1L);
+  ICEBERG_UNWRAP_OR_FAIL(auto op, NewOverwriteUpdate());
+  EXPECT_THAT(op->AddDelete(del_file), IsOk());
+  const int64_t second_snapshot_id = op->GeneratedSnapshotId();
+  ICEBERG_UNWRAP_OR_FAIL(auto manifests, op->Apply(*table_->metadata(), first_snapshot));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto second_snapshot,
+      MakeSyntheticSnapshot(DataOperation::kOverwrite, second_snapshot_id,
+                            first_snapshot->snapshot_id,
+                            first_snapshot->sequence_number + 1, manifests));
+
+  auto metadata = std::make_shared<TableMetadata>(*table_->metadata());
+  metadata->snapshots.push_back(second_snapshot);
+  metadata->current_snapshot_id = second_snapshot->snapshot_id;
+  metadata->last_sequence_number = second_snapshot->sequence_number;
+
+  DataFileSet replaced_files;
+  replaced_files.insert(file_a_);
+  // With ignore_equality_deletes=true, concurrently-added equality deletes
+  // should NOT cause a conflict.
+  EXPECT_THAT(TestMergeAppend::ValidateNoNewDeletesForDataFilesForTest(
+                  *metadata, first_snapshot->snapshot_id, replaced_files, second_snapshot,
+                  file_io_, /*ignore_equality_deletes=*/true),
+              IsOk());
+}
+
 TEST_F(MergingSnapshotUpdateTest,
        ValidateNoNewDeletesForDataFilesUsesConfiguredCaseSensitivity) {
   CommitFileA();

src/iceberg/test/merging_snapshot_update_test.cc

@@ -243,7 +243,7 @@ TEST_F(OverwriteFilesTest, TargetBranch) {
   CommitFileA();
 
   ICEBERG_UNWRAP_OR_FAIL(auto op, NewOverwrite());
-  op->SetTargetBranch("audit");
+  op->ToBranch("audit");
   op->AddFile(file_b_);
   EXPECT_THAT(op->Commit(), IsOk());