Move deleted files to Hadoop trash if configured

As of now, the HadoopFileIO uses the Java delete
API, which always skips using a configured trash
directory. If the table's hadoop configuration
has trash enabled, we should use it. In the event
of file deletions, we aim to mimic existing behavior,
throwing an error when attempting to delete a directory.
In the single file delete case, this notably adds an
RPC call (to check if a path is a directory)
if the trash is enabled.

Author: Jordan Epstein

core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java

@@ -30,6 +30,7 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.fs.Trash;
 import org.apache.iceberg.exceptions.RuntimeIOException;
 import org.apache.iceberg.io.BulkDeletionFailureException;
 import org.apache.iceberg.io.DelegateFileIO;
@@ -47,6 +48,7 @@
 
 public class HadoopFileIO implements HadoopConfigurable, DelegateFileIO {
 
+  public static final String USE_TRASH_IF_CONFIGURED = "iceberg.hadoop.use-trash-if-configured";
   private static final Logger LOG = LoggerFactory.getLogger(HadoopFileIO.class);
   private static final String DELETE_FILE_PARALLELISM = "iceberg.hadoop.delete-file-parallelism";
   private static final String DELETE_FILE_POOL_NAME = "iceberg-hadoopfileio-delete";
@@ -102,7 +104,7 @@ public void deleteFile(String path) {
     Path toDelete = new Path(path);
     FileSystem fs = Util.getFs(toDelete, getConf());
     try {
-      fs.delete(toDelete, false /* not recursive */);
+      deletePath(fs, toDelete, false);
     } catch (IOException e) {
       throw new RuntimeIOException(e, "Failed to delete file: %s", path);
     }
@@ -167,7 +169,7 @@ public void deletePrefix(String prefix) {
     FileSystem fs = Util.getFs(prefixToDelete, getConf());
 
     try {
-      fs.delete(prefixToDelete, true /* recursive */);
+      deletePath(fs, prefixToDelete, true);
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     }
@@ -211,6 +213,27 @@ private ExecutorService executorService() {
     return executorService;
   }
 
+  private void deletePath(FileSystem fs, Path toDelete, boolean recursive) throws IOException {
+    if (!shouldUseTrashIfConfigured()) {
+      fs.delete(toDelete, recursive);
+      return;
+    }
+    Trash trash = new Trash(fs, getConf());
+    if (!trash.isEnabled()) {
+      fs.delete(toDelete, recursive);
+    } else {
+      if (!recursive && fs.getFileStatus(toDelete).isDirectory()) {
+        throw new IOException(
+            String.format("Cannot move directory %s to trash", toDelete.toString()));
+      }
+      trash.moveToTrash(toDelete);
+    }
+  }
+
+  private boolean shouldUseTrashIfConfigured() {
+    return Boolean.parseBoolean(getConf().get(USE_TRASH_IF_CONFIGURED, "false"));
+  }
+
   /**
    * This class is a simple adaptor to allow for using Hadoop's RemoteIterator as an Iterator.
    *

core/src/test/java/org/apache/iceberg/hadoop/TestHadoopFileIO.java

@@ -18,7 +18,10 @@
  */
 package org.apache.iceberg.hadoop;
 
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY;
+import static org.apache.iceberg.hadoop.HadoopFileIO.USE_TRASH_IF_CONFIGURED;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.io.File;
@@ -126,6 +129,47 @@ public void testDeletePrefix() {
         .hasMessageContaining("java.io.FileNotFoundException");
   }
 
+  @Test
+  public void testDeletePrefixWithTrashEnabled() throws IOException {
+    Configuration conf = new Configuration();
+    conf.set(FS_TRASH_INTERVAL_KEY, "60");
+    conf.set(USE_TRASH_IF_CONFIGURED, "true");
+    fs = FileSystem.getLocal(conf);
+
+    hadoopFileIO = new HadoopFileIO(conf);
+    Path parent = new Path(tempDir.toURI());
+
+    List<Integer> scaleSizes = Lists.newArrayList(1, 1000, 2500);
+
+    scaleSizes.parallelStream()
+        .forEach(
+            scale -> {
+              Path scalePath = new Path(parent, Integer.toString(scale));
+
+              List<Path> filesCreated = createRandomFiles(scalePath, scale);
+              hadoopFileIO.deletePrefix(scalePath.toUri().toString());
+
+              // Hadoop filesystem will throw if the path does not exist
+              assertThatThrownBy(
+                      () -> hadoopFileIO.listPrefix(scalePath.toUri().toString()).iterator())
+                  .isInstanceOf(UncheckedIOException.class)
+                  .hasMessageContaining("java.io.FileNotFoundException");
+              filesCreated.forEach(
+                  file -> {
+                    String fileSuffix = Path.getPathWithoutSchemeAndAuthority(file).toString();
+                    String trashPath =
+                        fs.getTrashRoot(scalePath).toString() + "/Current" + fileSuffix;
+                    assertThat(hadoopFileIO.newInputFile(trashPath).exists()).isTrue();
+                  });
+            });
+
+    hadoopFileIO.deletePrefix(parent.toUri().toString());
+    // Hadoop filesystem will throw if the path does not exist
+    assertThatThrownBy(() -> hadoopFileIO.listPrefix(parent.toUri().toString()).iterator())
+        .isInstanceOf(UncheckedIOException.class)
+        .hasMessageContaining("java.io.FileNotFoundException");
+  }
+
   @Test
   public void testDeleteFiles() {
     Path parent = new Path(tempDir.toURI());
@@ -136,6 +180,74 @@ public void testDeleteFiles() {
         file -> assertThat(hadoopFileIO.newInputFile(file.toString()).exists()).isFalse());
   }
 
+  @Test
+  public void testDeleteFilesWithTrashEnabled() throws IOException {
+    Configuration conf = new Configuration();
+    conf.set(FS_TRASH_INTERVAL_KEY, "60");
+    conf.set(USE_TRASH_IF_CONFIGURED, "true");
+    fs = FileSystem.getLocal(conf);
+
+    hadoopFileIO = new HadoopFileIO(conf);
+    Path parent = new Path(tempDir.toURI());
+    List<Path> filesCreated = createRandomFiles(parent, 10);
+    hadoopFileIO.deleteFiles(
+        filesCreated.stream().map(Path::toString).collect(Collectors.toList()));
+    filesCreated.forEach(
+        file -> assertThat(hadoopFileIO.newInputFile(file.toString()).exists()).isFalse());
+    filesCreated.forEach(
+        file -> {
+          String fileSuffix = Path.getPathWithoutSchemeAndAuthority(file).toString();
+          String trashPath = fs.getTrashRoot(parent).toString() + "/Current" + fileSuffix;
+          assertThat(hadoopFileIO.newInputFile(trashPath).exists()).isTrue();
+        });
+  }
+
+  @Test
+  public void testDeleteFilesWithoutIcebergTrashToggleDoesNotMoveToTrash() throws IOException {
+    Configuration conf = new Configuration();
+    conf.set(FS_TRASH_INTERVAL_KEY, "60");
+    fs = FileSystem.getLocal(conf);
+
+    hadoopFileIO = new HadoopFileIO(conf);
+    Path parent = new Path(tempDir.toURI());
+    List<Path> filesCreated = createRandomFiles(parent, 10);
+    hadoopFileIO.deleteFiles(
+        filesCreated.stream().map(Path::toString).collect(Collectors.toList()));
+    filesCreated.forEach(
+        file -> assertThat(hadoopFileIO.newInputFile(file.toString()).exists()).isFalse());
+    filesCreated.forEach(
+        file -> {
+          String fileSuffix = Path.getPathWithoutSchemeAndAuthority(file).toString();
+          String trashPath = fs.getTrashRoot(parent).toString() + "/Current" + fileSuffix;
+          assertThat(hadoopFileIO.newInputFile(trashPath).exists()).isFalse();
+        });
+  }
+
+  @Test
+  public void testDeleteDirectoryWithTrashEnabledShouldThrow() throws IOException {
+    Configuration conf = new Configuration();
+    conf.set(FS_TRASH_INTERVAL_KEY, "60");
+    conf.set(USE_TRASH_IF_CONFIGURED, "true");
+    fs = FileSystem.getLocal(conf);
+
+    hadoopFileIO = new HadoopFileIO(conf);
+    Path parent = new Path(tempDir.toURI());
+    assertThatThrownBy(() -> hadoopFileIO.deleteFile(parent.toString()))
+        .hasCauseInstanceOf(IOException.class);
+  }
+
+  @Test
+  public void testDeleteDirectoryWithTrashEnabledShouldNotThrow() throws IOException {
+    Configuration conf = new Configuration();
+    conf.set(FS_TRASH_INTERVAL_KEY, "60");
+    conf.set(USE_TRASH_IF_CONFIGURED, "true");
+    fs = FileSystem.getLocal(conf);
+
+    hadoopFileIO = new HadoopFileIO(conf);
+    Path parent = new Path(tempDir.toURI());
+    assertThatNoException().isThrownBy(() -> hadoopFileIO.deletePrefix(parent.toString()));
+  }
+
   @Test
   public void testDeleteFilesErrorHandling() {
     List<String> filesCreated =