fix row delta delete validation

Co-authored-by: Codex <codex@openai.com>

Author: manuzhang

src/iceberg/test/row_delta_test.cc

@@ -274,7 +274,7 @@ TEST_F(RowDeltaTest, ValidateNoConflictingDeleteFilesFailsForConcurrentDelete) {
   EXPECT_THAT(result, HasErrorMessage(delete_file->file_path));
 }
 
-TEST_F(RowDeltaTest, ValidateDataFilesExistFailsForConcurrentDelete) {
+TEST_F(RowDeltaTest, ValidateDataFilesExistSkipsConcurrentDeleteByDefault) {
   CommitFileA();
   ICEBERG_UNWRAP_OR_FAIL(auto starting_snapshot, table_->current_snapshot());
 
@@ -294,6 +294,30 @@ TEST_F(RowDeltaTest, ValidateDataFilesExistFailsForConcurrentDelete) {
   row_delta->ValidateDataFilesExist(referenced_files);
   row_delta->AddDeletes(delete_file);
 
+  EXPECT_THAT(row_delta->Commit(), IsOk());
+}
+
+TEST_F(RowDeltaTest, ValidateDataFilesExistFailsForConcurrentDeleteWithValidateDeletedFiles) {
+  CommitFileA();
+  ICEBERG_UNWRAP_OR_FAIL(auto starting_snapshot, table_->current_snapshot());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto delete_files, table_->NewDeleteFiles());
+  delete_files->DeleteFile(file_a_);
+  EXPECT_THAT(delete_files->Commit(), IsOk());
+  EXPECT_THAT(table_->Refresh(), IsOk());
+
+  auto delete_file = MakeDeleteFile("/delete/file_a_pos_deletes.parquet",
+                                    /*partition_x=*/1L);
+  delete_file->referenced_data_file = file_a_->file_path;
+
+  std::shared_ptr<RowDelta> row_delta;
+  ICEBERG_UNWRAP_OR_FAIL(row_delta, table_->NewRowDelta());
+  row_delta->ValidateFromSnapshot(starting_snapshot->snapshot_id);
+  std::vector<std::string> referenced_files{file_a_->file_path};
+  row_delta->ValidateDataFilesExist(referenced_files);
+  row_delta->ValidateDeletedFiles();
+  row_delta->AddDeletes(delete_file);
+
   auto result = row_delta->Commit();
   EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
   EXPECT_THAT(result, HasErrorMessage("Cannot commit, missing data files"));

src/iceberg/update/row_delta.cc

@@ -143,7 +143,7 @@ Status RowDelta::Validate(const TableMetadata& current_metadata,
   if (!referenced_data_files_.empty()) {
     ICEBERG_RETURN_UNEXPECTED(MergingSnapshotUpdate::ValidateDataFilesExist(
         current_metadata, starting_snapshot_id_, referenced_data_files_,
-        /*skip_deletes=*/false, conflict_detection_filter_, snapshot, io,
+        /*skip_deletes=*/!validate_deletes_, conflict_detection_filter_, snapshot, io,
         IsCaseSensitive()));
   }
 

src/iceberg/update/row_delta.h

@@ -66,10 +66,16 @@ class ICEBERG_EXPORT RowDelta : public MergingSnapshotUpdate {
   RowDelta& CaseSensitive(bool case_sensitive);
 
   /// \brief Validate that referenced data files still exist.
+  ///
+  /// By default, this validation checks overwrite and replace commits. To apply
+  /// validation to delete commits, call ValidateDeletedFiles().
   RowDelta& ValidateDataFilesExist(std::span<const std::string> referenced_files);
 
-  /// \brief Fail if any requested data/delete-file removal is missing from
-  /// manifests when the table has a current snapshot.
+  /// \brief Enable validation for missing delete paths and delete-operation conflicts.
+  ///
+  /// This fails if any requested data/delete-file removal is missing from
+  /// manifests when the table has a current snapshot. It also makes
+  /// ValidateDataFilesExist() check delete-operation snapshots.
   RowDelta& ValidateDeletedFiles();
 
   /// \brief Set the conflict detection filter used by validation methods.