Spark: Add custom metric for number of deletes applied by a SparkScan

core/src/main/java/org/apache/iceberg/deletes/DeleteCounter.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.deletes;
+
+/** A counter to be used to count deletes as they are applied. */
+public class DeleteCounter {
+
+  private long count = 0L;
+
+  /** Increment the counter by one. */
+  public void increment() {
+    count++;
+  }
+
+  /** Return the current value of the counter. */
+  public long get() {
+    return count;
+  }
+}

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

@@ -62,6 +62,14 @@ public static <T> CloseableIterable<T> filter(
     return equalityFilter.filter(rows);
   }
 
+  /**
+   * Returns the same rows that are input, while marking the deleted ones.
+   *
+   * @param rows the rows to process
+   * @param isDeleted a predicate that determines if a row is deleted
+   * @param deleteMarker a function that marks a row as deleted
+   * @return the processed rows
+   */
   public static <T> CloseableIterable<T> markDeleted(
       CloseableIterable<T> rows, Predicate<T> isDeleted, Consumer<T> deleteMarker) {
     return CloseableIterable.transform(
@@ -70,13 +78,35 @@ public static <T> CloseableIterable<T> markDeleted(
           if (isDeleted.test(row)) {
             deleteMarker.accept(row);
           }
+
           return row;
         });
   }
 
+  /**
+   * Returns the remaining rows (the ones that are not deleted), while counting the deleted ones.
+   *
+   * @param rows the rows to process
+   * @param isDeleted a predicate that determines if a row is deleted
+   * @param counter a counter that counts deleted rows
+   * @return the processed rows
+   */
   public static <T> CloseableIterable<T> filterDeleted(
-      CloseableIterable<T> rows, Predicate<T> isDeleted) {
-    return CloseableIterable.filter(rows, isDeleted.negate());
+      CloseableIterable<T> rows, Predicate<T> isDeleted, DeleteCounter counter) {
+    Filter<T> remainingRowsFilter =
+        new Filter<T>() {
+          @Override
+          protected boolean shouldKeep(T item) {
+            boolean deleted = isDeleted.test(item);
+            if (deleted) {
+              counter.increment();
+            }
+
+            return !deleted;
+          }
+        };
+
+    return remainingRowsFilter.filter(rows);
   }
 
   public static StructLikeSet toEqualitySet(
@@ -116,7 +146,15 @@ public static <T> CloseableIterable<T> streamingFilter(
       CloseableIterable<T> rows,
       Function<T, Long> rowToPosition,
       CloseableIterable<Long> posDeletes) {
-    return new PositionStreamDeleteFilter<>(rows, rowToPosition, posDeletes);
+    return streamingFilter(rows, rowToPosition, posDeletes, new DeleteCounter());
+  }
+
+  public static <T> CloseableIterable<T> streamingFilter(
+      CloseableIterable<T> rows,
+      Function<T, Long> rowToPosition,
+      CloseableIterable<Long> posDeletes,
+      DeleteCounter counter) {
+    return new PositionStreamDeleteFilter<>(rows, rowToPosition, posDeletes, counter);
   }
 
   public static <T> CloseableIterable<T> streamingMarker(
@@ -215,19 +253,28 @@ boolean isDeleted(T row) {
   }
 
   private static class PositionStreamDeleteFilter<T> extends PositionStreamDeleteIterable<T> {
-    private PositionStreamDeleteFilter(
+    private final DeleteCounter counter;
+
+    PositionStreamDeleteFilter(
         CloseableIterable<T> rows,
         Function<T, Long> rowToPosition,
-        CloseableIterable<Long> deletePositions) {
+        CloseableIterable<Long> deletePositions,
+        DeleteCounter counter) {
       super(rows, rowToPosition, deletePositions);
+      this.counter = counter;
     }
 
     @Override
     protected CloseableIterator<T> applyDelete(CloseableIterator<T> items) {
       return new FilterIterator<T>(items) {
         @Override
         protected boolean shouldKeep(T item) {
-          return !isDeleted(item);
+          boolean deleted = isDeleted(item);
+          if (deleted) {
+            counter.increment();
+          }
+
+          return !deleted;
         }
       };
     }

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

@@ -33,6 +33,7 @@
 import org.apache.iceberg.data.avro.DataReader;
 import org.apache.iceberg.data.orc.GenericOrcReader;
 import org.apache.iceberg.data.parquet.GenericParquetReaders;
+import org.apache.iceberg.deletes.DeleteCounter;
 import org.apache.iceberg.deletes.Deletes;
 import org.apache.iceberg.deletes.PositionDeleteIndex;
 import org.apache.iceberg.expressions.Expressions;
@@ -52,8 +53,11 @@
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.StructLikeSet;
 import org.apache.iceberg.util.StructProjection;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public abstract class DeleteFilter<T> {
+  private static final Logger LOG = LoggerFactory.getLogger(DeleteFilter.class);
   private static final long DEFAULT_SET_FILTER_THRESHOLD = 100_000L;
   private static final Schema POS_DELETE_SCHEMA =
       new Schema(MetadataColumns.DELETE_FILE_PATH, MetadataColumns.DELETE_FILE_POS);
@@ -66,24 +70,32 @@ public abstract class DeleteFilter<T> {
   private final Accessor<StructLike> posAccessor;
   private final boolean hasIsDeletedColumn;
   private final int isDeletedColumnPosition;
+  private final DeleteCounter counter;
 
   private PositionDeleteIndex deleteRowPositions = null;
   private List<Predicate<T>> isInDeleteSets = null;
   private Predicate<T> eqDeleteRows = null;
 
   protected DeleteFilter(
-      String filePath, List<DeleteFile> deletes, Schema tableSchema, Schema requestedSchema) {
+      String filePath,
+      List<DeleteFile> deletes,
+      Schema tableSchema,
+      Schema requestedSchema,
+      DeleteCounter counter) {
     this.setFilterThreshold = DEFAULT_SET_FILTER_THRESHOLD;
     this.filePath = filePath;
+    this.counter = counter;
 
     ImmutableList.Builder<DeleteFile> posDeleteBuilder = ImmutableList.builder();
     ImmutableList.Builder<DeleteFile> eqDeleteBuilder = ImmutableList.builder();
     for (DeleteFile delete : deletes) {
       switch (delete.content()) {
         case POSITION_DELETES:
+          LOG.debug("Adding position delete file {} to filter", delete.path());
           posDeleteBuilder.add(delete);
           break;
         case EQUALITY_DELETES:
+          LOG.debug("Adding equality delete file {} to filter", delete.path());
           eqDeleteBuilder.add(delete);
           break;
         default:
@@ -101,6 +113,11 @@ protected DeleteFilter(
     this.isDeletedColumnPosition = requiredSchema.columns().indexOf(MetadataColumns.IS_DELETED);
   }
 
+  protected DeleteFilter(
+      String filePath, List<DeleteFile> deletes, Schema tableSchema, Schema requestedSchema) {
+    this(filePath, deletes, tableSchema, requestedSchema, new DeleteCounter());
+  }
+
   protected int columnIsDeletedPosition() {
     return isDeletedColumnPosition;
   }
@@ -117,6 +134,10 @@ public boolean hasEqDeletes() {
     return !eqDeletes.isEmpty();
   }
 
+  public void incrementDeleteCount() {
+    counter.increment();
+  }
+
   Accessor<StructLike> posAccessor() {
     return posAccessor;
   }
@@ -234,21 +255,23 @@ private CloseableIterable<T> applyPosDeletes(CloseableIterable<T> records) {
     return hasIsDeletedColumn
         ? Deletes.streamingMarker(
             records, this::pos, Deletes.deletePositions(filePath, deletes), this::markRowDeleted)
-        : Deletes.streamingFilter(records, this::pos, Deletes.deletePositions(filePath, deletes));
+        : Deletes.streamingFilter(
+            records, this::pos, Deletes.deletePositions(filePath, deletes), counter);
   }
 
   private CloseableIterable<T> createDeleteIterable(
       CloseableIterable<T> records, Predicate<T> isDeleted) {
     return hasIsDeletedColumn
         ? Deletes.markDeleted(records, isDeleted, this::markRowDeleted)
-        : Deletes.filterDeleted(records, isDeleted);
+        : Deletes.filterDeleted(records, isDeleted, counter);
   }
 
   private CloseableIterable<Record> openPosDeletes(DeleteFile file) {
     return openDeletes(file, POS_DELETE_SCHEMA);
   }
 
   private CloseableIterable<Record> openDeletes(DeleteFile deleteFile, Schema deleteSchema) {
+    LOG.trace("Opening delete file {}", deleteFile.path());
     InputFile input = getInputFile(deleteFile.path().toString());
     switch (deleteFile.format()) {
       case AVRO:

data/src/test/java/org/apache/iceberg/data/DeleteReadTests.java

@@ -171,6 +171,26 @@ protected boolean expectPruned() {
     return true;
   }
 
+  protected boolean countDeletes() {
+    return false;
+  }
+
+  /**
+   * This will only be called after calling rowSet(String, Table, String...), and only if
+   * countDeletes() is true.
+   */
+  protected long deleteCount() {
+    return 0L;
+  }
+
+  protected void checkDeleteCount(long expectedDeletes) {
+    if (countDeletes()) {
+      long actualDeletes = deleteCount();
+      Assert.assertEquals(
+          "Table should contain expected number of deletes", expectedDeletes, actualDeletes);
+    }
+  }
+
   @Test
   public void testEqualityDeletes() throws IOException {
     Schema deleteRowSchema = table.schema().select("data");
@@ -192,6 +212,7 @@ public void testEqualityDeletes() throws IOException {
     StructLikeSet actual = rowSet(tableName, table, "*");
 
     Assert.assertEquals("Table should contain expected rows", expected, actual);
+    checkDeleteCount(3L);
   }
 
   @Test
@@ -240,6 +261,7 @@ public void testEqualityDateDeletes() throws IOException {
     StructLikeSet actual = rowSet(dateTableName, dateTable, "*");
 
     Assert.assertEquals("Table should contain expected rows", expected, actual);
+    checkDeleteCount(3L);
   }
 
   @Test
@@ -270,6 +292,8 @@ public void testEqualityDeletesWithRequiredEqColumn() throws IOException {
       Assert.assertEquals(
           "Table should contain expected rows", expected, selectColumns(actual, "id"));
     }
+
+    checkDeleteCount(3L);
   }
 
   @Test
@@ -281,6 +305,8 @@ public void testEqualityDeletesSpanningMultipleDataFiles() throws IOException {
     this.dataFile =
         FileHelpers.writeDataFile(table, Files.localOutput(temp.newFile()), Row.of(0), records);
 
+    // At this point, the table has two data files, with 7 and 8 rows respectively, of which all but
+    // one are in duplicate.
     table.newAppend().appendFile(dataFile).commit();
 
     Schema deleteRowSchema = table.schema().select("data");
@@ -296,12 +322,14 @@ public void testEqualityDeletesSpanningMultipleDataFiles() throws IOException {
         FileHelpers.writeDeleteFile(
             table, Files.localOutput(temp.newFile()), Row.of(0), dataDeletes, deleteRowSchema);
 
+    // At this point, 3 rows in the first data file and 4 rows in the second data file are deleted.
     table.newRowDelta().addDeletes(eqDeletes).commit();
 
     StructLikeSet expected = rowSetWithoutIds(table, records, 29, 89, 122, 144);
     StructLikeSet actual = rowSet(tableName, table, "*");
 
     Assert.assertEquals("Table should contain expected rows", expected, actual);
+    checkDeleteCount(7L);
   }
 
   @Test
@@ -326,6 +354,7 @@ public void testPositionDeletes() throws IOException {
     StructLikeSet actual = rowSet(tableName, table, "*");
 
     Assert.assertEquals("Table should contain expected rows", expected, actual);
+    checkDeleteCount(3L);
   }
 
   @Test
@@ -363,6 +392,7 @@ public void testMultiplePosDeleteFiles() throws IOException {
     StructLikeSet actual = rowSet(tableName, table, "*");
 
     Assert.assertEquals("Table should contain expected rows", expected, actual);
+    checkDeleteCount(3L);
   }
 
   @Test
@@ -400,6 +430,7 @@ public void testMixedPositionAndEqualityDeletes() throws IOException {
     StructLikeSet actual = rowSet(tableName, table, "*");
 
     Assert.assertEquals("Table should contain expected rows", expected, actual);
+    checkDeleteCount(4L);
   }
 
   @Test
@@ -435,6 +466,7 @@ public void testMultipleEqualityDeleteSchemas() throws IOException {
     StructLikeSet actual = rowSet(tableName, table, "*");
 
     Assert.assertEquals("Table should contain expected rows", expected, actual);
+    checkDeleteCount(4L);
   }
 
   @Test
@@ -471,6 +503,7 @@ public void testEqualityDeleteByNull() throws IOException {
     StructLikeSet actual = rowSet(tableName, table, "*");
 
     Assert.assertEquals("Table should contain expected rows", expected, actual);
+    checkDeleteCount(1L);
   }
 
   private StructLikeSet selectColumns(StructLikeSet rows, String... columns) {

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchReader.java

@@ -181,8 +181,12 @@ Pair<int[], Integer> buildPosDelRowIdMapping(PositionDeleteIndex deletedRowPosit
         if (!deletedRowPositions.isDeleted(originalRowId + rowStartPosInBatch)) {
           posDelRowIdMapping[currentRowId] = originalRowId;
           currentRowId++;
-        } else if (hasIsDeletedColumn) {
-          isDeleted[originalRowId] = true;
+        } else {
+          if (hasIsDeletedColumn) {
+            isDeleted[originalRowId] = true;
+          }
+
+          deletes.incrementDeleteCount();
         }
         originalRowId++;
       }
@@ -203,6 +207,7 @@ int[] initEqDeleteRowIdMapping() {
           eqDeleteRowIdMapping[i] = i;
         }
       }
+
       return eqDeleteRowIdMapping;
     }
 
@@ -227,8 +232,12 @@ void applyEqDelete(ColumnarBatch columnarBatch) {
           // skip deleted rows by pointing to the next undeleted row Id
           rowIdMapping[currentRowId] = rowIdMapping[rowId];
           currentRowId++;
-        } else if (hasIsDeletedColumn) {
-          isDeleted[rowIdMapping[rowId]] = true;
+        } else {
+          if (hasIsDeletedColumn) {
+            isDeleted[rowIdMapping[rowId]] = true;
+          }
+
+          deletes.incrementDeleteCount();
         }
 
         rowId++;