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;
@@ -102,7 +103,15 @@ public void deleteFile(String path) {
     Path toDelete = new Path(path);
     FileSystem fs = Util.getFs(toDelete, getConf());
     try {
-      fs.delete(toDelete, false /* not recursive */);
+      Trash trash = new Trash(fs, getConf());
+      if (!trash.isEnabled()) {
+        fs.delete(toDelete, false /* not recursive */);
+      } else {
+        if (fs.getFileStatus(toDelete).isDirectory()) {
+          throw new IOException(String.format("Cannot delete move directory %s to trash", path));
+        }
+        trash.moveToTrash(toDelete);
+      }
     } catch (IOException e) {
       throw new RuntimeIOException(e, "Failed to delete file: %s", path);
     }
@@ -167,7 +176,12 @@ public void deletePrefix(String prefix) {
     FileSystem fs = Util.getFs(prefixToDelete, getConf());
 
     try {
-      fs.delete(prefixToDelete, true /* recursive */);
+      Trash trash = new Trash(fs, getConf());
+      if (!trash.isEnabled()) {
+        fs.delete(prefixToDelete, true /* recursive */);
+      } else {
+        trash.moveToTrash(prefixToDelete);
+      }
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     }

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

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.hadoop;
 
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
@@ -30,6 +31,7 @@
 import java.util.UUID;
 import java.util.Vector;
 import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -126,6 +128,48 @@ public void testDeletePrefix() {
         .hasMessageContaining("java.io.FileNotFoundException");
   }
 
+  @Test
+  public void testDeletePrefixWithTrashEnabled() 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<Integer> scaleSizes = Lists.newArrayList(1, 1000, 2500);
+
+    scaleSizes.parallelStream()
+        .forEach(
+            scale -> {
+              Path scalePath = new Path(parent, Integer.toString(scale));
+
+              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");
+              assertThat(
+                      StreamSupport.stream(
+                              hadoopFileIO
+                                  .listPrefix(
+                                      fs.getTrashRoot(scalePath).toString() + "/Current" + tempDir)
+                                  .spliterator(),
+                              false)
+                          .count())
+                  .isEqualTo((Long) (long) (int) scale);
+            });
+
+    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,29 @@ 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");
+    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());
+    assertThat(
+            StreamSupport.stream(
+                    hadoopFileIO
+                        .listPrefix(fs.getTrashRoot(parent).toString() + "/Current" + tempDir)
+                        .spliterator(),
+                    false)
+                .count())
+        .isEqualTo(10L);
+  }
+
   @Test
   public void testDeleteFilesErrorHandling() {
     List<String> filesCreated =