Core: ValidationException on metadata-only delete of certain delete files

core/src/main/java/org/apache/iceberg/ManifestFilterManager.java

@@ -350,18 +350,25 @@ private boolean canContainDeletedFiles(ManifestFile manifest) {
     return canContainExpressionDeletes || canContainDroppedPartitions || canContainDroppedFiles || canContainDropBySeq;
   }
 
-  @SuppressWarnings("CollectionUndefinedEquality")
+  @SuppressWarnings({"CollectionUndefinedEquality", "checkstyle:CyclomaticComplexity"})
   private boolean manifestHasDeletedFiles(PartitionAndMetricsEvaluator evaluator, ManifestReader<F> reader) {
     boolean isDelete = reader.isDeleteManifestReader();
     boolean hasDeletedFiles = false;
     for (ManifestEntry<F> entry : reader.liveEntries()) {
       F file = entry.file();
-      boolean fileDelete = deletePaths.contains(file.path()) ||
+      boolean markedForDelete = deletePaths.contains(file.path()) ||
           dropPartitions.contains(file.specId(), file.partition()) ||
           (isDelete && entry.sequenceNumber() > 0 && entry.sequenceNumber() < minSequenceNumber);
-      if (fileDelete || evaluator.rowsMightMatch(file)) {
+
+      boolean nonMatchingDeleteFile = !file.content().equals(FileContent.DATA) && !evaluator.rowsMustMatch(file);
+      if (!markedForDelete && nonMatchingDeleteFile) {
+        // not all DeleteFiles removal can be handled by metadata operation, skip in this case
+        continue;
+      }
+
+      if (markedForDelete || evaluator.rowsMightMatch(file)) {
         ValidationException.check(
-            fileDelete || evaluator.rowsMustMatch(file),
+            markedForDelete || evaluator.rowsMustMatch(file),
             "Cannot delete file where some, but not all, rows match filter %s: %s",
             this.deleteExpression, file.path());
 

spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java

@@ -22,6 +22,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -30,6 +31,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.DataFile;
 import org.apache.iceberg.RowLevelOperationMode;
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.Table;
@@ -39,6 +41,8 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.util.concurrent.MoreExecutors;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.iceberg.spark.data.TestHelpers;
 import org.apache.spark.SparkException;
 import org.apache.spark.sql.AnalysisException;
 import org.apache.spark.sql.Dataset;
@@ -80,6 +84,30 @@ public void removeTables() {
     sql("DROP TABLE IF EXISTS deleted_dep");
   }
 
+  @Test
+  public void testDeleteFileThenMetadataDelete() throws Exception {
+    Assume.assumeFalse("Avro does not support metadata delete", fileFormat.equals("avro"));
+    createAndInitUnpartitionedTable();
+
+    sql("INSERT INTO TABLE %s VALUES (1, 'hr'), (2, 'hardware'), (null, 'hr')", tableName);
+
+    // MOR mode: writes a delete file as null cannot be deleted by metadata
+    sql("DELETE FROM %s AS t WHERE t.id IS NULL", tableName);
+
+    // Metadata Delete
+    Table table = Spark3Util.loadIcebergTable(spark, tableName);
+    Set<DataFile> dataFilesBefore = TestHelpers.dataFiles(table);
+
+    sql("DELETE FROM %s AS t WHERE t.id = 1", tableName);
+
+    Set<DataFile> dataFilesAfter = TestHelpers.dataFiles(table);
+    Assert.assertTrue("Data file should have been removed", dataFilesBefore.size() > dataFilesAfter.size());
+
+    assertEquals("Should have expected rows",
+        ImmutableList.of(row(2, "hardware")),
+        sql("SELECT * FROM %s ORDER BY id", tableName));
+  }
+
   @Test
   public void testDeleteWithFalseCondition() {
     createAndInitUnpartitionedTable();

spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/data/TestHelpers.java

@@ -37,6 +37,7 @@
 import org.apache.arrow.vector.ValueVector;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericData.Record;
+import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.ManifestFile;
@@ -704,6 +705,16 @@ public static List<ManifestFile> deleteManifests(Table table) {
     return table.currentSnapshot().deleteManifests();
   }
 
+  public static Set<DataFile> dataFiles(Table table) {
+    Set<DataFile> dataFiles = Sets.newHashSet();
+
+    for (FileScanTask task : table.newScan().planFiles()) {
+      dataFiles.add(task.file());
+    }
+
+    return dataFiles;
+  }
+
   public static Set<DeleteFile> deleteFiles(Table table) {
     Set<DeleteFile> deleteFiles = Sets.newHashSet();