[WIP] Eq to pos delete

.palantir/revapi.yml

@@ -873,6 +873,11 @@ acceptedBreaks:
       new: "method void org.apache.iceberg.encryption.Ciphers::<init>()"
       justification: "Static utility class - should not have public constructor"
   "1.4.0":
+    org.apache.iceberg:iceberg-api:
+    - code: "java.method.addedToInterface"
+      new: "method org.apache.iceberg.RewriteFiles org.apache.iceberg.RewriteFiles::rewriteDeletes(java.util.Set<org.apache.iceberg.DeleteFile>,\
+        \ java.util.Set<org.apache.iceberg.DeleteFile>)"
+      justification: "added for rewrite eq to pos delete"
     org.apache.iceberg:iceberg-core:
     - code: "java.field.serialVersionUIDChanged"
       new: "field org.apache.iceberg.util.SerializableMap<K, V>.serialVersionUID"

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

@@ -184,4 +184,14 @@ RewriteFiles rewriteFiles(
    * @return this for method chaining
    */
   RewriteFiles validateFromSnapshot(long snapshotId);
+
+  /**
+   * 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

@@ -125,6 +125,26 @@ public RewriteFiles validateFromSnapshot(long snapshotId) {
     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;
+  }
+
   @Override
   public BaseRewriteFiles toBranch(String branch) {
     targetBranch(branch);

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

@@ -22,7 +22,7 @@
 import org.apache.iceberg.SnapshotUpdate;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 
-abstract class BaseSnapshotUpdateAction<ThisT, R> extends BaseAction<ThisT, R>
+public abstract class BaseSnapshotUpdateAction<ThisT, R> extends BaseAction<ThisT, R>
     implements SnapshotUpdateAction<ThisT, R> {
 
   private final Map<String, String> summary = Maps.newHashMap();

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

@@ -64,7 +64,7 @@ public class OutputFileFactory {
    * @param operationId Third part of the file name
    * @param suffix Suffix part of the file name
    */
-  private OutputFileFactory(
+  public OutputFileFactory(
       PartitionSpec spec,
       FileFormat format,
       LocationProvider locations,

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 FileWriter<PositionDelete<T>, DeleteWriteResult> {
+public class SortedPosDeleteWriter<T> implements FileWriter<PositionDelete<T>, DeleteWriteResult> {
   private static final long DEFAULT_RECORDS_NUM_THRESHOLD = 100_000L;
 
   private final Map<CharSequenceWrapper, List<PosRow<T>>> posDeletes = Maps.newHashMap();
@@ -54,7 +54,7 @@ class SortedPosDeleteWriter<T> implements FileWriter<PositionDelete<T>, DeleteWr
   private boolean closed = false;
   private Throwable failure;
 
-  SortedPosDeleteWriter(
+  public SortedPosDeleteWriter(
       FileAppenderFactory<T> appenderFactory,
       OutputFileFactory fileFactory,
       FileFormat format,
@@ -67,7 +67,7 @@ class SortedPosDeleteWriter<T> implements FileWriter<PositionDelete<T>, DeleteWr
     this.recordsNumThreshold = recordsNumThreshold;
   }
 
-  SortedPosDeleteWriter(
+  public SortedPosDeleteWriter(
       FileAppenderFactory<T> appenderFactory,
       OutputFileFactory fileFactory,
       FileFormat format,

core/src/main/java/org/apache/iceberg/util/ChainOrFilter.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.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

@@ -51,6 +51,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.slf4j.Logger;
@@ -151,7 +153,45 @@ public CloseableIterable<T> filter(CloseableIterable<T> records) {
     return applyEqDeletes(applyPosDeletes(records));
   }
 
-  private List<Predicate<T>> applyEqDeletes() {
+  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 List<Predicate<T>> applyEqDeletes() {
     if (isInDeleteSets != null) {
       return isInDeleteSets;
     }

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

@@ -0,0 +1,40 @@
+/*
+ * 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.spark.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;
+  }
+}