Spark 3.3: Add RemoveDanglingDeletes action

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

@@ -55,6 +55,17 @@ default DeleteFiles deleteFile(DataFile file) {
     return this;
   }
 
+  /**
+   * Delete a file tracked by a {@link DeleteFile} from the underlying table.
+   *
+   * @param file a DeleteFile to remove from the table
+   * @return this for method chaining
+   */
+  default DeleteFiles deleteFile(DeleteFile file) {
+    deleteFile(file.path());
+    return this;
+  }
+
   /**
    * Delete files that match an {@link Expression} on data rows from the table.
    *

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

@@ -64,4 +64,10 @@ default DeleteReachableFiles deleteReachableFiles(String metadataLocation) {
     throw new UnsupportedOperationException(
         this.getClass().getName() + " does not implement deleteReachableFiles");
   }
+
+  /** Instantiates an action to remove dangling delete files from current snapshot. */
+  default RemoveDanglingDeleteFiles removeDanglingDeleteFiles(Table table) {
+    throw new UnsupportedOperationException(
+        this.getClass().getName() + " does not implement removeDanglingDeleteFiles");
+  }
 }

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

@@ -0,0 +1,45 @@
+/*
+ * 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;
+
+/**
+ * An action that removes dangling delete files from the current snapshot. A delete file is dangling
+ * if its deletes no longer applies to any data file.
+ *
+ * <p>The following dangling delete files are removed:
+ *
+ * <ul>
+ *   <li>Position delete files with a sequence number less than that of any data file in the same
+ *       partition
+ *   <li>Equality delete files with a sequence number less than or equal to that of any data file in
+ *       the same partition
+ * </ul>
+ */
+public interface RemoveDanglingDeleteFiles
+    extends Action<RemoveDanglingDeleteFiles, RemoveDanglingDeleteFiles.Result> {
+
+  /** The action result that contains a summary of the execution. */
+  interface Result {
+    /** Removes representation of removed delete files. */
+    List<DeleteFile> removedDeleteFiles();
+  }
+}

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

@@ -54,6 +54,12 @@ public StreamingDelete deleteFile(DataFile file) {
     return this;
   }
 
+  @Override
+  public StreamingDelete deleteFile(DeleteFile file) {
+    delete(file);
+    return this;
+  }
+
   @Override
   public StreamingDelete deleteFromRowFilter(Expression expr) {
     deleteByRowFilter(expr);

core/src/main/java/org/apache/iceberg/actions/RemoveDanglingDeleteFilesActionResult.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.List;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+
+public class RemoveDanglingDeleteFilesActionResult implements RemoveDanglingDeleteFiles.Result {
+
+  private static final RemoveDanglingDeleteFilesActionResult EMPTY =
+      new RemoveDanglingDeleteFilesActionResult(ImmutableList.of());
+
+  private List<DeleteFile> removedDeleteFiles;
+
+  public RemoveDanglingDeleteFilesActionResult(List<DeleteFile> removeDeleteFiles) {
+    this.removedDeleteFiles = removeDeleteFiles;
+  }
+
+  public static RemoveDanglingDeleteFilesActionResult empty() {
+    return EMPTY;
+  }
+
+  @Override
+  public List<DeleteFile> removedDeleteFiles() {
+    return removedDeleteFiles;
+  }
+}

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java

@@ -30,6 +30,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import java.util.function.Supplier;
 import org.apache.iceberg.AllManifestsTable;
 import org.apache.iceberg.BaseTable;
@@ -58,20 +59,26 @@
 import org.apache.iceberg.spark.JobGroupUtils;
 import org.apache.iceberg.spark.SparkTableUtil;
 import org.apache.iceberg.spark.source.SerializableTableWithSize;
+import org.apache.iceberg.util.PropertyUtil;
 import org.apache.iceberg.util.Tasks;
 import org.apache.spark.SparkContext;
 import org.apache.spark.api.java.JavaSparkContext;
 import org.apache.spark.api.java.function.FlatMapFunction;
+import org.apache.spark.api.java.function.MapFunction;
 import org.apache.spark.broadcast.Broadcast;
 import org.apache.spark.sql.Column;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.internal.SQLConf;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 abstract class BaseSparkAction<ThisT> {
 
+  public static final String USE_CACHING = "use-caching";
+  public static final boolean USE_CACHING_DEFAULT = true;
+
   protected static final String MANIFEST = "Manifest";
   protected static final String MANIFEST_LIST = "Manifest List";
   protected static final String OTHERS = "Others";
@@ -361,4 +368,25 @@ static FileInfo toFileInfo(ContentFile<?> file) {
       return new FileInfo(file.path().toString(), file.content().toString());
     }
   }
+
+  protected <T, U> U withReusableDS(Dataset<T> ds, Function<Dataset<T>, U> func) {
+    Dataset<T> reusableDS;
+    boolean useCaching =
+        PropertyUtil.propertyAsBoolean(options(), USE_CACHING, USE_CACHING_DEFAULT);
+    if (useCaching) {
+      reusableDS = ds.cache();
+    } else {
+      int parallelism = SQLConf.get().numShufflePartitions();
+      reusableDS =
+          ds.repartition(parallelism).map((MapFunction<T, T>) value -> value, ds.exprEnc());
+    }
+
+    try {
+      return func.apply(reusableDS);
+    } finally {
+      if (useCaching) {
+        reusableDS.unpersist(false);
+      }
+    }
+  }
 }