Spark: support replace equality deletes to position deletes

api/src/main/java/org/apache/iceberg/RewriteFiles.java

@@ -42,4 +42,13 @@ public interface RewriteFiles extends SnapshotUpdate<RewriteFiles> {
    * @return this for method chaining
    */
   RewriteFiles rewriteFiles(Set<DataFile> filesToDelete, Set<DataFile> filesToAdd);
+
+  /**
+   * Add a rewrite that replaces one set of deletes with another that contains the same deleted rows.
+   *
+   * @param deletesToDelete files that will be replaced, cannot be null or empty.
+   * @param deletesToAdd files that will be added, cannot be null or empty.
+   * @return this for method chaining
+   */
+  RewriteFiles rewriteDeletes(Set<DeleteFile> deletesToDelete, Set<DeleteFile> deletesToAdd);
 }

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

@@ -57,4 +57,22 @@ public RewriteFiles rewriteFiles(Set<DataFile> filesToDelete, Set<DataFile> file
 
     return this;
   }
+
+  @Override
+  public RewriteFiles rewriteDeletes(Set<DeleteFile> deletesToDelete, Set<DeleteFile> deletesToAdd) {
+    Preconditions.checkArgument(deletesToDelete != null && !deletesToDelete.isEmpty(),
+        "Files to delete cannot be null or empty");
+    Preconditions.checkArgument(deletesToAdd != null && !deletesToAdd.isEmpty(),
+        "Files to add can not be null or empty");
+
+    for (DeleteFile toDelete : deletesToDelete) {
+      delete(toDelete);
+    }
+
+    for (DeleteFile toAdd : deletesToAdd) {
+      add(toAdd);
+    }
+
+    return this;
+  }
 }

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

@@ -24,6 +24,7 @@
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.PartitionKey;
 import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.encryption.EncryptedOutputFile;
 import org.apache.iceberg.encryption.EncryptionManager;
 
@@ -104,4 +105,13 @@ public EncryptedOutputFile newOutputFile(PartitionKey key) {
     OutputFile rawOutputFile = io.newOutputFile(newDataLocation);
     return encryptionManager.encrypt(rawOutputFile);
   }
+
+  /**
+   * Generates EncryptedOutputFile for PartitionedWriter.
+   */
+  public EncryptedOutputFile newOutputFile(StructLike partition) {
+    String newDataLocation = locations.newDataLocation(spec, partition, generateFilename());
+    OutputFile rawOutputFile = io.newOutputFile(newDataLocation);
+    return encryptionManager.encrypt(rawOutputFile);
+  }
 }

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

@@ -28,7 +28,7 @@
 import java.util.Set;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.deletes.PositionDeleteWriter;
 import org.apache.iceberg.encryption.EncryptedOutputFile;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -37,7 +37,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();
@@ -48,15 +48,15 @@ class SortedPosDeleteWriter<T> implements Closeable {
   private final FileAppenderFactory<T> appenderFactory;
   private final OutputFileFactory fileFactory;
   private final FileFormat format;
-  private final PartitionKey partition;
+  private final StructLike partition;
   private final long recordsNumThreshold;
 
   private int records = 0;
 
   SortedPosDeleteWriter(FileAppenderFactory<T> appenderFactory,
                         OutputFileFactory fileFactory,
                         FileFormat format,
-                        PartitionKey partition,
+                        StructLike partition,
                         long recordsNumThreshold) {
     this.appenderFactory = appenderFactory;
     this.fileFactory = fileFactory;
@@ -65,10 +65,10 @@ class SortedPosDeleteWriter<T> implements Closeable {
     this.recordsNumThreshold = recordsNumThreshold;
   }
 
-  SortedPosDeleteWriter(FileAppenderFactory<T> appenderFactory,
-                        OutputFileFactory fileFactory,
-                        FileFormat format,
-                        PartitionKey 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/ChainOrFilter.java

@@ -0,0 +1,42 @@
+/*
+ * 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.util;
+
+import java.util.List;
+import java.util.function.Predicate;
+
+public class ChainOrFilter<T> extends Filter<T> {
+  private final List<Predicate<T>> filters;
+
+  public ChainOrFilter(List<Predicate<T>> filters) {
+    this.filters = filters;
+  }
+
+  @Override
+  protected boolean shouldKeep(T item) {
+    for (Predicate<T> filter : filters) {
+      if (filter.test(item)) {
+        return true;
+      }
+    }
+
+    return false;
+  }
+}

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

@@ -23,6 +23,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Predicate;
 import org.apache.iceberg.Accessor;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
@@ -48,6 +49,8 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ChainOrFilter;
+import org.apache.iceberg.util.Filter;
 import org.apache.iceberg.util.StructLikeSet;
 import org.apache.iceberg.util.StructProjection;
 import org.apache.parquet.Preconditions;
@@ -110,7 +113,42 @@ public CloseableIterable<T> filter(CloseableIterable<T> records) {
     return applyEqDeletes(applyPosDeletes(records));
   }
 
-  private CloseableIterable<T> applyEqDeletes(CloseableIterable<T> records) {
+  public CloseableIterable<T> matchEqDeletes(CloseableIterable<T> records) {
+    if (eqDeletes.isEmpty()) {
+      return records;
+    }
+
+    Multimap<Set<Integer>, DeleteFile> filesByDeleteIds = Multimaps.newMultimap(Maps.newHashMap(), Lists::newArrayList);
+    for (DeleteFile delete : eqDeletes) {
+      filesByDeleteIds.put(Sets.newHashSet(delete.equalityFieldIds()), delete);
+    }
+
+    List<Predicate<T>> deleteSetFilters = Lists.newArrayList();
+    for (Map.Entry<Set<Integer>, Collection<DeleteFile>> entry : filesByDeleteIds.asMap().entrySet()) {
+      Set<Integer> ids = entry.getKey();
+      Iterable<DeleteFile> deletes = entry.getValue();
+
+      Schema deleteSchema = TypeUtil.select(requiredSchema, ids);
+
+      // a projection to select and reorder fields of the file schema to match the delete rows
+      StructProjection projectRow = StructProjection.create(requiredSchema, deleteSchema);
+
+      Iterable<CloseableIterable<Record>> deleteRecords = Iterables.transform(deletes,
+          delete -> openDeletes(delete, deleteSchema));
+      StructLikeSet deleteSet = Deletes.toEqualitySet(
+          // copy the delete records because they will be held in a set
+          CloseableIterable.transform(CloseableIterable.concat(deleteRecords), Record::copy),
+          deleteSchema.asStruct());
+
+      Predicate<T> predicate = record -> deleteSet.contains(projectRow.wrap(asStructLike(record)));
+      deleteSetFilters.add(predicate);
+    }
+
+    Filter<T> findDeleteRows = new ChainOrFilter<>(deleteSetFilters);
+    return findDeleteRows.filter(records);
+  }
+
+  protected CloseableIterable<T> applyEqDeletes(CloseableIterable<T> records) {
     if (eqDeletes.isEmpty()) {
       return records;
     }

spark/src/main/java/org/apache/iceberg/actions/Actions.java

@@ -82,6 +82,10 @@ public ExpireSnapshotsAction expireSnapshots() {
     return new ExpireSnapshotsAction(spark, table);
   }
 
+  public ReplaceDeleteAction replaceEqDeleteToPosDelete() {
+    return new ReplaceDeleteAction(spark, table);
+  }
+
   /**
    * Converts the provided table into an Iceberg table in place. The table will no longer be accessible by it's
    * previous implementation

spark/src/main/java/org/apache/iceberg/actions/DeleteRewriteActionResult.java

@@ -0,0 +1,41 @@
+/*
+ * 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.List;
+import org.apache.iceberg.DeleteFile;
+
+public class DeleteRewriteActionResult {
+  private List<DeleteFile> posDeletes;
+  private List<DeleteFile> eqDeletes;
+
+  public DeleteRewriteActionResult(List<DeleteFile> eqDeletes, List<DeleteFile> posDeletes) {
+    this.eqDeletes = eqDeletes;
+    this.posDeletes = posDeletes;
+  }
+
+  public List<DeleteFile> deletedFiles() {
+    return eqDeletes;
+  }
+
+  public List<DeleteFile> addedFiles() {
+    return posDeletes;
+  }
+}