fix: rewrite files review comments

Author: WZhuo

src/iceberg/table.cc

@@ -367,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

@@ -267,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/rewrite_files_test.cc

@@ -127,6 +127,30 @@ TEST_F(RewriteFilesTest, AddAndDelete) {
   EXPECT_EQ(std::stoll(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles)), 1);
 }
 
+TEST_F(RewriteFilesTest, DeleteDataFileCopiesCallerFile) {
+  CommitFileA();
+
+  const std::string original_path = file_a_->file_path;
+
+  ICEBERG_UNWRAP_OR_FAIL(auto rw, NewRewriteFiles());
+  rw->DeleteDataFile(file_a_);
+  rw->AddDataFile(rewritten_file_a_);
+
+  file_a_->file_path = file_b_->file_path;
+
+  EXPECT_THAT(rw->Commit(), IsOk());
+  EXPECT_THAT(table_->Refresh(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto rw_missing_original, NewRewriteFiles());
+  auto missing_file = std::make_shared<DataFile>(*file_a_);
+  missing_file->file_path = original_path;
+  rw_missing_original->DeleteDataFile(missing_file);
+  rw_missing_original->AddDataFile(file_b_);
+  auto result = rw_missing_original->Commit();
+  EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+  EXPECT_THAT(result, HasErrorMessage("Missing required files to delete"));
+}
+
 // Rewrite one of several data files, verifying only the target is affected.
 TEST_F(RewriteFilesTest, AddAndDeletePartialRewrite) {
   CommitFileA();

src/iceberg/test/table_test.cc

@@ -164,6 +164,7 @@ TEST(StaticTableTest, NewMutatingOperationsAreNotSupported) {
   EXPECT_THAT(table->NewDeleteFiles(), IsError(ErrorKind::kNotSupported));
   EXPECT_THAT(table->NewRowDelta(), IsError(ErrorKind::kNotSupported));
   EXPECT_THAT(table->NewOverwrite(), IsError(ErrorKind::kNotSupported));
+  EXPECT_THAT(table->NewRewriteFiles(), IsError(ErrorKind::kNotSupported));
   EXPECT_THAT(table->NewSnapshotManager(), IsError(ErrorKind::kNotSupported));
 }
 

src/iceberg/update/rewrite_files.cc

@@ -48,9 +48,10 @@ Result<std::unique_ptr<RewriteFiles>> RewriteFiles::Make(
 }
 
 RewriteFiles& RewriteFiles::DeleteDataFile(const std::shared_ptr<DataFile>& data_file) {
-  ICEBERG_BUILDER_RETURN_IF_ERROR(MergingSnapshotUpdate::DeleteDataFile(data_file));
-  // Track replaced data files for conflict detection
-  replaced_data_files_.insert(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));
   return *this;
 }