Spark: Add an action to rewrite equality deletes

api/src/main/java/org/apache/iceberg/actions/ActionsProvider.java

@@ -74,4 +74,11 @@ default ExpireSnapshots expireSnapshots(Table table) {
   default DeleteReachableFiles deleteReachableFiles(String metadataLocation) {
     throw new UnsupportedOperationException(this.getClass().getName() + " does not implement deleteReachableFiles");
   }
+
+  /**
+   * Instantiates an action to rewrite deletes.
+   */
+  default RewriteDeletes rewriteDeletes(Table table) {
+    throw new UnsupportedOperationException(this.getClass().getName() + " does not implement rewriteDeletes");
+  }
 }

api/src/main/java/org/apache/iceberg/actions/RewriteDeletes.java

@@ -0,0 +1,48 @@
+/*
+ * 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.actions;
+
+import java.util.Set;
+import org.apache.iceberg.DeleteFile;
+
+public interface RewriteDeletes extends SnapshotUpdate<RewriteDeletes, RewriteDeletes.Result> {
+
+  /**
+   * Set the implementation class name of rewrite delete strategy
+   *
+   * @return this for method chaining
+   */
+  RewriteDeletes strategy(String strategyImpl);
+
+  /**
+   * The action result that contains a summary of the execution.
+   */
+  interface Result {
+    /**
+     * Returns the delete files to rewrite.
+     */
+    Set<DeleteFile> deletedFiles();
+
+    /**
+     * Returns the added delete files.
+     */
+    Set<DeleteFile> addedFiles();
+  }
+}

core/src/main/java/org/apache/iceberg/actions/BaseRewriteDeletesResult.java

@@ -0,0 +1,44 @@
+/*
+ * 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.actions;
+
+import java.util.Set;
+import org.apache.iceberg.DeleteFile;
+
+public class BaseRewriteDeletesResult implements RewriteDeletes.Result {
+
+  private final Set<DeleteFile> deletesToReplace;
+  private final Set<DeleteFile> deletesToAdd;
+
+  public BaseRewriteDeletesResult(Set<DeleteFile> deletesToReplace, Set<DeleteFile> deletesToAdd) {
+    this.deletesToReplace = deletesToReplace;
+    this.deletesToAdd = deletesToAdd;
+  }
+
+  @Override
+  public Set<DeleteFile> deletedFiles() {
+    return deletesToReplace;
+  }
+
+  @Override
+  public Set<DeleteFile> addedFiles() {
+    return deletesToAdd;
+  }
+}

core/src/main/java/org/apache/iceberg/actions/RewriteDeleteStrategy.java

@@ -0,0 +1,76 @@
+/*
+ * 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.actions;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Table;
+
+public interface RewriteDeleteStrategy {
+
+  /**
+   * Returns the name of this rewrite deletes strategy
+   */
+  String name();
+
+  /**
+   * Returns the table being modified by this rewrite strategy
+   */
+  Table table();
+
+  /**
+   * Select the deletes to rewrite.
+   *
+   * @param dataFiles iterable of FileScanTasks for data files in a given partition
+   * @return iterable of original delete file to be replaced.
+   */
+  Iterable<DeleteFile> selectDeletesToRewrite(Iterable<FileScanTask> dataFiles);
+
+  /**
+   * Define how to rewrite the deletes.
+   *
+   * @param deleteFilesToRewrite a group of files to be rewritten together
+   * @return iterable of delete files used to replace the original delete files.
+   */
+  Set<DeleteFile> rewriteDeletes(Set<DeleteFile> deleteFilesToRewrite);
+
+  /**
+   * Groups file scans into lists which will be processed in a single executable unit. Each group will end up being
+   * committed as an independent set of changes. This creates the jobs which will eventually be run as by the underlying
+   * Action.
+   *
+   * @param deleteFiles iterable of DeleteFile to be rewritten
+   * @return iterable of lists of FileScanTasks which will be processed together
+   */
+  Iterable<Set<DeleteFile>> planDeleteGroups(Iterable<DeleteFile> deleteFiles);
+
+  /**
+   * Sets options to be used with this strategy
+   */
+  RewriteDeleteStrategy options(Map<String, String> options);
+
+  /**
+   * Returns a set of options which this rewrite strategy can use. This is an allowed-list and any options not
+   * specified here will be rejected at runtime.
+   */
+  Set<String> validOptions();
+}

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

@@ -24,6 +24,7 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.Set;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import org.apache.iceberg.Accessor;
 import org.apache.iceberg.MetadataColumns;
@@ -113,6 +114,13 @@ public static <T> CloseableIterable<T> streamingFilter(CloseableIterable<T> rows
     return new PositionStreamDeleteFilter<>(rows, rowToPosition, posDeletes);
   }
 
+  public static <T> CloseableIterable<T> streamingDeletedRowMarker(CloseableIterable<T> rows,
+                                                                   Function<T, Long> rowToPosition,
+                                                                   CloseableIterable<Long> posDeletes,
+                                                                   Consumer<T> deleteMarker) {
+    return new PositionStreamDeletedRowMarker<>(rows, rowToPosition, posDeletes, deleteMarker);
+  }
+
   public static CloseableIterable<Long> deletePositions(CharSequence dataLocation,
                                                         CloseableIterable<StructLike> deleteFile) {
     return deletePositions(dataLocation, ImmutableList.of(deleteFile));
@@ -176,7 +184,7 @@ public CloseableIterator<T> iterator() {
 
       CloseableIterator<T> iter;
       if (deletePosIterator.hasNext()) {
-        iter = new PositionFilterIterator(rows.iterator(), deletePosIterator);
+        iter = positionIterator(rows.iterator(), deletePosIterator);
       } else {
         iter = rows.iterator();
         try {
@@ -191,7 +199,12 @@ public CloseableIterator<T> iterator() {
       return iter;
     }
 
-    private class PositionFilterIterator extends FilterIterator<T> {
+    protected FilterIterator<T> positionIterator(CloseableIterator<T> items,
+                                                 CloseableIterator<Long> newDeletePositions) {
+      return new PositionFilterIterator(items, newDeletePositions);
+    }
+
+    protected class PositionFilterIterator extends FilterIterator<T> {
       private final CloseableIterator<Long> deletePosIterator;
       private long nextDeletePos;
 
@@ -233,6 +246,37 @@ public void close() {
     }
   }
 
+  static class PositionStreamDeletedRowMarker<T> extends PositionStreamDeleteFilter<T> {
+    private final Consumer<T> deleteMarker;
+
+    private PositionStreamDeletedRowMarker(CloseableIterable<T> rows, Function<T, Long> extractPos,
+                                           CloseableIterable<Long> deletePositions,
+                                           Consumer<T> deleteMarker) {
+      super(rows, extractPos, deletePositions);
+      this.deleteMarker = deleteMarker;
+    }
+
+    @Override
+    protected FilterIterator<T> positionIterator(CloseableIterator<T> items,
+                                                 CloseableIterator<Long> deletePositions) {
+      return new PositionMarkerIterator(items, deletePositions);
+    }
+
+    private class PositionMarkerIterator extends PositionFilterIterator {
+      private PositionMarkerIterator(CloseableIterator<T> items, CloseableIterator<Long> deletePositions) {
+        super(items, deletePositions);
+      }
+
+      @Override
+      protected boolean shouldKeep(T row) {
+        if (!super.shouldKeep(row)) {
+          deleteMarker.accept(row);
+        }
+        return true;
+      }
+    }
+  }
+
   private static class DataFileFilter<T extends StructLike> extends Filter<T> {
     private static final Comparator<CharSequence> CHARSEQ_COMPARATOR = Comparators.charSequences();
     private final CharSequence dataLocation;

core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java

@@ -36,7 +36,7 @@
 import org.apache.iceberg.util.CharSequenceSet;
 import org.apache.iceberg.util.CharSequenceWrapper;
 
-class SortedPosDeleteWriter<T> implements Closeable {
+public class SortedPosDeleteWriter<T> implements Closeable {
   private static final long DEFAULT_RECORDS_NUM_THRESHOLD = 100_000L;
 
   private final Map<CharSequenceWrapper, List<PosRow<T>>> posDeletes = Maps.newHashMap();
@@ -52,22 +52,22 @@ class SortedPosDeleteWriter<T> implements Closeable {
 
   private int records = 0;
 
-  SortedPosDeleteWriter(FileAppenderFactory<T> appenderFactory,
-                        OutputFileFactory fileFactory,
-                        FileFormat format,
-                        StructLike partition,
-                        long recordsNumThreshold) {
+  public SortedPosDeleteWriter(FileAppenderFactory<T> appenderFactory,
+                               OutputFileFactory fileFactory,
+                               FileFormat format,
+                               StructLike partition,
+                               long recordsNumThreshold) {
     this.appenderFactory = appenderFactory;
     this.fileFactory = fileFactory;
     this.format = format;
     this.partition = partition;
     this.recordsNumThreshold = recordsNumThreshold;
   }
 
-  SortedPosDeleteWriter(FileAppenderFactory<T> appenderFactory,
-                        OutputFileFactory fileFactory,
-                        FileFormat format,
-                        StructLike partition) {
+  public SortedPosDeleteWriter(FileAppenderFactory<T> appenderFactory,
+                               OutputFileFactory fileFactory,
+                               FileFormat format,
+                               StructLike partition) {
     this(appenderFactory, fileFactory, format, partition, DEFAULT_RECORDS_NUM_THRESHOLD);
   }
 

core/src/main/java/org/apache/iceberg/util/TableScanUtil.java

@@ -19,16 +19,29 @@
 
 package org.apache.iceberg.util;
 
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Map;
 import java.util.function.Function;
 import org.apache.iceberg.BaseCombinedScanTask;
 import org.apache.iceberg.CombinedScanTask;
 import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.ListMultimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class TableScanUtil {
 
+  private static final Logger LOG = LoggerFactory.getLogger(TableScanUtil.class);
+
   private TableScanUtil() {
   }
 
@@ -64,4 +77,21 @@ public static CloseableIterable<CombinedScanTask> planTasks(CloseableIterable<Fi
             splitFiles),
         BaseCombinedScanTask::new);
   }
+
+  public static Map<StructLikeWrapper, Collection<FileScanTask>> groupTasksByPartition(
+      PartitionSpec spec,
+      CloseableIterator<FileScanTask> tasksIter) {
+    ListMultimap<StructLikeWrapper, FileScanTask> tasksGroupedByPartition = Multimaps.newListMultimap(
+        Maps.newHashMap(), Lists::newArrayList);
+    try (CloseableIterator<FileScanTask> iterator = tasksIter) {
+      iterator.forEachRemaining(task -> {
+        StructLikeWrapper structLike = StructLikeWrapper.forType(spec.partitionType()).set(task.file().partition());
+        tasksGroupedByPartition.put(structLike, task);
+      });
+    } catch (IOException e) {
+      LOG.warn("Failed to close task iterator", e);
+    }
+
+    return tasksGroupedByPartition.asMap();
+  }
 }