minor fixes and refine comments

Author: wgtmac

src/iceberg/test/merge_append_test.cc

@@ -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/overwrite_files_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());
 

src/iceberg/test/rewrite_files_test.cc

@@ -307,6 +307,44 @@ TEST_P(RewriteFilesFormatVersionTest, DeleteDataFileCopiesCallerFile) {
   EXPECT_THAT(result, HasErrorMessage("Missing required files to delete"));
 }
 
+TEST_P(RewriteFilesFormatVersionTest, AddDataFileRejectsDeleteFileContent) {
+  CommitFileA();
+
+  ICEBERG_UNWRAP_OR_FAIL(auto rw, NewRewriteFiles());
+  rw->DeleteDataFile(file_a_);
+  rw->AddDataFile(delete_file_a_);
+  auto result = rw->Commit();
+  EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+  EXPECT_THAT(result, HasErrorMessage("Invalid data file to add"));
+}
+
+TEST_P(RewriteFilesFormatVersionTest, RewriteDeleteFilesCopiesCallerFiles) {
+  AssumeFormatVersionAtLeast(2);
+  CommitFileA();
+
+  {
+    ICEBERG_UNWRAP_OR_FAIL(auto row_delta, table_->NewRowDelta());
+    row_delta->AddDeletes(delete_file_a_);
+    EXPECT_THAT(row_delta->Commit(), IsOk());
+    EXPECT_THAT(table_->Refresh(), IsOk());
+  }
+
+  ICEBERG_UNWRAP_OR_FAIL(auto after_delta_snapshot, table_->current_snapshot());
+  auto old_delete = std::make_shared<DataFile>(*delete_file_a_);
+  auto new_delete = std::make_shared<DataFile>(*rewritten_delete_file_a_);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto rw, NewRewriteFiles());
+  rw->ValidateFromSnapshot(after_delta_snapshot->snapshot_id);
+  rw->DeleteDeleteFile(old_delete);
+  rw->AddDeleteFile(new_delete);
+
+  old_delete->file_path = table_location_ + "/data/delete_a_mutated.parquet";
+  new_delete->content = DataFile::Content::kData;
+
+  EXPECT_THAT(rw->Commit(), IsOk());
+  EXPECT_THAT(table_->Refresh(), IsOk());
+}
+
 // Rewrite one of several data files, verifying only the target is affected.
 TEST_P(RewriteFilesFormatVersionTest, AddAndDeletePartialRewrite) {
   CommitFileA();

src/iceberg/update/merging_snapshot_update.cc

@@ -540,8 +540,10 @@ Status MergingSnapshotUpdate::AddDeleteFile(std::shared_ptr<DataFile> file,
     return InvalidArgument("Delete file must have a partition spec ID");
   }
   ICEBERG_RETURN_UNEXPECTED(base().PartitionSpecById(file->partition_spec_id.value()));
+
+  auto staged_file = std::make_shared<DataFile>(*file);
   has_new_delete_files_ = true;
-  PendingDeleteFile pending_file{.file = std::move(file),
+  PendingDeleteFile pending_file{.file = std::move(staged_file),
                                  .data_sequence_number = std::move(data_sequence_number)};
   if (ContentFileUtil::IsDV(*pending_file.file)) {
     ICEBERG_PRECHECK(pending_file.file->referenced_data_file.has_value(),
@@ -559,14 +561,16 @@ Status MergingSnapshotUpdate::DeleteDataFile(std::shared_ptr<DataFile> file) {
   if (!file) {
     return InvalidArgument("Cannot delete a null data file");
   }
-  return data_filter_manager_->DeleteFile(std::move(file));
+  auto staged_file = std::make_shared<DataFile>(*file);
+  return data_filter_manager_->DeleteFile(std::move(staged_file));
 }
 
 Status MergingSnapshotUpdate::DeleteDeleteFile(std::shared_ptr<DataFile> file) {
   if (!file) {
     return InvalidArgument("Cannot delete a null delete file");
   }
-  return delete_filter_manager_->DeleteFile(std::move(file));
+  auto staged_file = std::make_shared<DataFile>(*file);
+  return delete_filter_manager_->DeleteFile(std::move(staged_file));
 }
 
 Status MergingSnapshotUpdate::DeleteByPath(std::string_view path) {

src/iceberg/update/rewrite_files.cc

@@ -48,31 +48,49 @@ Result<std::unique_ptr<RewriteFiles>> RewriteFiles::Make(
 }
 
 RewriteFiles& RewriteFiles::DeleteDataFile(const std::shared_ptr<DataFile>& data_file) {
-  auto staged_file =
-      data_file == nullptr ? nullptr : std::make_shared<DataFile>(*data_file);
-  ICEBERG_BUILDER_RETURN_IF_ERROR(MergingSnapshotUpdate::DeleteDataFile(staged_file));
-  replaced_data_files_.insert(std::move(staged_file));
+  ICEBERG_BUILDER_CHECK(data_file != nullptr, "Invalid data file: null");
+  ICEBERG_BUILDER_CHECK(data_file->content == DataFile::Content::kData,
+                        "Invalid data file to delete: {} has delete-file content",
+                        data_file->file_path);
+  ICEBERG_BUILDER_RETURN_IF_ERROR(MergingSnapshotUpdate::DeleteDataFile(data_file));
+  replaced_data_files_.insert(std::make_shared<DataFile>(*data_file));
   return *this;
 }
 
 RewriteFiles& RewriteFiles::DeleteDeleteFile(
     const std::shared_ptr<DataFile>& delete_file) {
+  ICEBERG_BUILDER_CHECK(delete_file != nullptr, "Invalid delete file: null");
+  ICEBERG_BUILDER_CHECK(delete_file->content != DataFile::Content::kData,
+                        "Invalid delete file to delete: {} has data-file content",
+                        delete_file->file_path);
   ICEBERG_BUILDER_RETURN_IF_ERROR(MergingSnapshotUpdate::DeleteDeleteFile(delete_file));
   return *this;
 }
 
 RewriteFiles& RewriteFiles::AddDataFile(const std::shared_ptr<DataFile>& file) {
+  ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null");
+  ICEBERG_BUILDER_CHECK(file->content == DataFile::Content::kData,
+                        "Invalid data file to add: {} has delete-file content",
+                        file->file_path);
   ICEBERG_BUILDER_RETURN_IF_ERROR(MergingSnapshotUpdate::AddDataFile(file));
   return *this;
 }
 
 RewriteFiles& RewriteFiles::AddDeleteFile(const std::shared_ptr<DataFile>& delete_file) {
+  ICEBERG_BUILDER_CHECK(delete_file != nullptr, "Invalid delete file: null");
+  ICEBERG_BUILDER_CHECK(delete_file->content != DataFile::Content::kData,
+                        "Invalid delete file to add: {} has data-file content",
+                        delete_file->file_path);
   ICEBERG_BUILDER_RETURN_IF_ERROR(MergingSnapshotUpdate::AddDeleteFile(delete_file));
   return *this;
 }
 
 RewriteFiles& RewriteFiles::AddDeleteFile(const std::shared_ptr<DataFile>& delete_file,
                                           int64_t data_sequence_number) {
+  ICEBERG_BUILDER_CHECK(delete_file != nullptr, "Invalid delete file: null");
+  ICEBERG_BUILDER_CHECK(delete_file->content != DataFile::Content::kData,
+                        "Invalid delete file to add: {} has data-file content",
+                        delete_file->file_path);
   ICEBERG_BUILDER_RETURN_IF_ERROR(
       MergingSnapshotUpdate::AddDeleteFile(delete_file, data_sequence_number));
   return *this;
@@ -120,11 +138,6 @@ RewriteFiles& RewriteFiles::ValidateFromSnapshot(int64_t snapshot_id) {
   return *this;
 }
 
-RewriteFiles& RewriteFiles::ToBranch(const std::string& branch) {
-  SetTargetBranch(branch);
-  return *this;
-}
-
 std::string RewriteFiles::operation() { return DataOperation::kReplace; }
 
 void RewriteFiles::ValidateReplacedAndAddedFiles() {
@@ -153,18 +166,12 @@ void RewriteFiles::ValidateReplacedAndAddedFiles() {
 
 Status RewriteFiles::Validate(const TableMetadata& current_metadata,
                               const std::shared_ptr<Snapshot>& snapshot) {
-  // Step 1: Validate the replaced and added files invariants
-  ValidateReplacedAndAddedFiles();
   ICEBERG_RETURN_UNEXPECTED(CheckErrors());
+  ValidateReplacedAndAddedFiles();
 
-  // Step 2: If there are replaced data files, validate no new row-level deletes
-  // for those data files have been added concurrently
   if (!replaced_data_files_.empty()) {
-    // The instance method automatically determines ignore_equality_deletes
-    // based on whether SetDataSequenceNumber was called:
-    // - If data sequence number was set, equality deletes at higher sequence
-    //   numbers still correctly apply to the new files → ignore them.
-    // - If no sequence number was set, ALL new deletes are conflicts.
+    // If there are replaced data files, there cannot be any new row-level deletes
+    // for those data files.
     auto io = ctx_->table->io();
     ICEBERG_RETURN_UNEXPECTED(MergingSnapshotUpdate::ValidateNoNewDeletesForDataFiles(
         current_metadata, starting_snapshot_id_, replaced_data_files_, snapshot,

src/iceberg/update/rewrite_files.h

@@ -37,14 +37,14 @@ namespace iceberg {
 
 /// \brief API for replacing files in a table.
 ///
-/// This operation accumulates file additions and deletions, produces a new
-/// Snapshot of the changes, and commits that snapshot as the current.
+/// This API accumulates file additions and deletions, produces a new Snapshot
+/// of the changes, and commits that snapshot as the current.
 ///
 /// When committing, these changes will be applied to the latest table snapshot.
 /// Commit conflicts will be resolved by applying the changes to the new latest
 /// snapshot and reattempting the commit. If any of the deleted files are no
-/// longer in the latest snapshot when reattempting, the commit will throw a
-/// ValidationException.
+/// longer in the latest snapshot when reattempting, the commit will fail
+/// validation.
 ///
 /// Note that the new state of the table after each rewrite must be logically
 /// equivalent to the original table state.
@@ -63,7 +63,9 @@ class ICEBERG_EXPORT RewriteFiles : public MergingSnapshotUpdate {
   /// \brief Remove a data file from the current table state.
   ///
   /// This rewrite operation may change the size or layout of the data files.
-  /// The set of live data records must never change.
+  /// When applicable, it is also recommended to discard already deleted records
+  /// while rewriting data files. However, the set of live data records must
+  /// never change.
   ///
   /// \param data_file a rewritten data file
   /// \return this for method chaining
@@ -72,32 +74,47 @@ class ICEBERG_EXPORT RewriteFiles : public MergingSnapshotUpdate {
   /// \brief Remove a delete file from the table state.
   ///
   /// This rewrite operation may change the size or layout of the delete files.
-  /// The set of applicable delete records must never change.
+  /// When applicable, it is also recommended to discard delete records for files
+  /// that are no longer part of the table state. However, the set of applicable
+  /// delete records must never change.
   ///
   /// \param delete_file a rewritten delete file
   /// \return this for method chaining
   RewriteFiles& DeleteDeleteFile(const std::shared_ptr<DataFile>& delete_file);
 
-  /// \brief Add a new data file or delete file.
+  /// \brief Add a new data file.
   ///
-  /// This rewrite operation may change the size or layout of the files.
-  /// The set of live data records must never change.
+  /// This rewrite operation may change the size or layout of the data files.
+  /// When applicable, it is also recommended to discard already deleted records
+  /// while rewriting data files. However, the set of live data records must
+  /// never change.
   ///
-  /// \param file a new file (data or delete)
+  /// \param file a new data file
   /// \return this for method chaining
   RewriteFiles& AddDataFile(const std::shared_ptr<DataFile>& file);
 
   /// \brief Add a new delete file.
   ///
+  /// This rewrite operation may change the size or layout of the delete files.
+  /// When applicable, it is also recommended to discard delete records for files
+  /// that are no longer part of the table state. However, the set of applicable
+  /// delete records must never change.
+  ///
   /// \param delete_file a new delete file
   /// \return this for method chaining
   RewriteFiles& AddDeleteFile(const std::shared_ptr<DataFile>& delete_file);
 
   /// \brief Add a new delete file with the given data sequence number.
   ///
+  /// This rewrite operation may change the size or layout of the delete files.
+  /// When applicable, it is also recommended to discard delete records for files
+  /// that are no longer part of the table state. However, the set of applicable
+  /// delete records must never change.
+  ///
   /// To ensure equivalence in the set of applicable delete records, the
   /// sequence number of the delete file must be the max sequence number of
-  /// the delete files that it is replacing.
+  /// the delete files that it is replacing. Rewriting equality deletes that
+  /// belong to different sequence numbers is not allowed.
   ///
   /// \param delete_file a new delete file
   /// \param data_sequence_number data sequence number to append on the file
@@ -108,8 +125,8 @@ class ICEBERG_EXPORT RewriteFiles : public MergingSnapshotUpdate {
   /// \brief Configure the data sequence number for this rewrite operation.
   ///
   /// This data sequence number will be used for all new data files that are
-  /// added in this rewrite. This is helpful to avoid commit conflicts between
-  /// data compaction and adding equality deletes.
+  /// added in this rewrite. This method is helpful to avoid commit conflicts
+  /// between data compaction and adding equality deletes.
   ///
   /// \param sequence_number a data sequence number
   /// \return this for method chaining
@@ -119,8 +136,9 @@ class ICEBERG_EXPORT RewriteFiles : public MergingSnapshotUpdate {
   /// that contains the same data. The sequence number provided will be used for
   /// all the data files added.
   ///
-  /// \param files_to_delete files that will be replaced (deleted)
-  /// \param files_to_add files that will be added
+  /// \param files_to_delete files that will be replaced (deleted), cannot be
+  /// null or empty
+  /// \param files_to_add files that will be added, cannot be null or empty
   /// \param sequence_number sequence number to use for all data files added
   /// \return this for method chaining
   RewriteFiles& RewriteDataFiles(
@@ -152,12 +170,6 @@ class ICEBERG_EXPORT RewriteFiles : public MergingSnapshotUpdate {
   /// \return this for method chaining
   RewriteFiles& ValidateFromSnapshot(int64_t snapshot_id);
 
-  /// \brief Perform operations on a particular branch.
-  ///
-  /// \param branch name of a SnapshotRef of type branch
-  /// \return this for method chaining
-  RewriteFiles& ToBranch(const std::string& branch);
-
   std::string operation() override;
 
  protected:

src/iceberg/update/snapshot_manager.cc

@@ -93,7 +93,7 @@ SnapshotManager& SnapshotManager::CreateBranch(const std::string& name) {
   }
   ICEBERG_BUILDER_CHECK(!base.refs.contains(name), "Ref {} already exists", name);
   ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto fast_append, transaction_->NewFastAppend());
-  ICEBERG_BUILDER_RETURN_IF_ERROR(fast_append->SetTargetBranch(name).Commit());
+  ICEBERG_BUILDER_RETURN_IF_ERROR(fast_append->ToBranch(name).Commit());
   return *this;
 }
 

src/iceberg/update/snapshot_update.h

@@ -96,7 +96,7 @@ class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate {
   ///
   /// \param branch The name of a SnapshotRef of type branch.
   /// \return This update for method chaining.
-  auto& SetTargetBranch(this auto& self, const std::string& branch) {
+  auto& ToBranch(this auto& self, const std::string& branch) {
     if (branch.empty()) [[unlikely]] {
       return self.AddError(ErrorKind::kInvalidArgument, "Branch name cannot be empty");
     }