Use SupportsPrefixOperations for Remove OrphanFile Procedure on Spark 3.5

Author: ismailsimsek

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

@@ -293,28 +293,37 @@ private Dataset<FileURI> validFileIdentDS() {
 
   private Dataset<FileURI> actualFileIdentDS() {
     StringToFileURI toFileURI = new StringToFileURI(equalSchemes, equalAuthorities);
+    Dataset<String> dataList;
     if (compareToFileList == null) {
-      return toFileURI.apply(listedFileDS());
+      dataList =
+          table.io() instanceof SupportsPrefixOperations ? listWithPrefix() : listWithoutPrefix();
     } else {
-      return toFileURI.apply(filteredCompareToFileList());
+      dataList = filteredCompareToFileList();
     }
+
+    return toFileURI.apply(dataList);
   }
 
-  private Dataset<String> listWithPrefix() {
+  @VisibleForTesting
+  Dataset<String> listWithPrefix() {
     List<String> matchingFiles = Lists.newArrayList();
+    PathFilter pathFilter = PartitionAwareHiddenPathFilter.forSpecs(table.specs());
+
     Iterator<org.apache.iceberg.io.FileInfo> iterator =
         ((SupportsPrefixOperations) table.io()).listPrefix(location).iterator();
     while (iterator.hasNext()) {
       org.apache.iceberg.io.FileInfo fileInfo = iterator.next();
-      if (fileInfo.createdAtMillis() < olderThanTimestamp) {
+      if (fileInfo.createdAtMillis() < olderThanTimestamp
+          && pathFilter.accept(new Path(fileInfo.location()))) {
         matchingFiles.add(fileInfo.location());
       }
     }
     JavaRDD<String> matchingFileRDD = sparkContext().parallelize(matchingFiles, 1);
     return spark().createDataset(matchingFileRDD.rdd(), Encoders.STRING());
   }
 
-  private Dataset<String> listWithoutPrefix() {
+  @VisibleForTesting
+  Dataset<String> listWithoutPrefix() {
     List<String> subDirs = Lists.newArrayList();
     List<String> matchingFiles = Lists.newArrayList();
 
@@ -349,14 +358,6 @@ private Dataset<String> listWithoutPrefix() {
     return spark().createDataset(completeMatchingFileRDD.rdd(), Encoders.STRING());
   }
 
-  private Dataset<String> listedFileDS() {
-    if (table.io() instanceof SupportsPrefixOperations) {
-      return listWithPrefix();
-    } else {
-      return listWithoutPrefix();
-    }
-  }
-
   private static void listDirRecursively(
       String dir,
       Predicate<FileStatus> predicate,

spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveOrphanFilesAction.java

@@ -539,9 +539,12 @@ public void testHiddenPartitionPaths() {
     waitUntilAfter(System.currentTimeMillis());
 
     SparkActions actions = SparkActions.get();
+    DeleteOrphanFilesSparkAction action =
+        actions.deleteOrphanFiles(table).olderThan(System.currentTimeMillis());
+    // test list methods by directly instantiating the action
+    assertThatDatasetsAreEqualIgnoringOrder(action.listWithPrefix(), action.listWithoutPrefix());
 
-    DeleteOrphanFiles.Result result =
-        actions.deleteOrphanFiles(table).olderThan(System.currentTimeMillis()).execute();
+    DeleteOrphanFiles.Result result = action.execute();
 
     assertThat(result.orphanFileLocations()).as("Should delete 2 files").hasSize(2);
   }
@@ -575,9 +578,12 @@ public void testHiddenPartitionPathsWithPartitionEvolution() {
     waitUntilAfter(System.currentTimeMillis());
 
     SparkActions actions = SparkActions.get();
+    DeleteOrphanFilesSparkAction action =
+        actions.deleteOrphanFiles(table).olderThan(System.currentTimeMillis());
+    // test list methods by directly instantiating the action
+    assertThatDatasetsAreEqualIgnoringOrder(action.listWithPrefix(), action.listWithoutPrefix());
 
-    DeleteOrphanFiles.Result result =
-        actions.deleteOrphanFiles(table).olderThan(System.currentTimeMillis()).execute();
+    DeleteOrphanFiles.Result result = action.execute();
 
     assertThat(result.orphanFileLocations()).as("Should delete 2 files").hasSize(2);
   }
@@ -605,17 +611,23 @@ public void testHiddenPathsStartingWithPartitionNamesAreIgnored() throws IOExcep
     Path dataPath = new Path(tableLocation + "/data");
     FileSystem fs = dataPath.getFileSystem(spark.sessionState().newHadoopConf());
     Path pathToFileInHiddenFolder = new Path(dataPath, "_c2_trunc/file.txt");
-    fs.createNewFile(pathToFileInHiddenFolder);
+    fs.createNewFile(new Path(dataPath, "_c2_trunc/file.txt"));
+    Path pathToFileInHiddenFolder2 = new Path(dataPath, "_c2_trunc/subfolder/file.txt");
+    fs.createNewFile(pathToFileInHiddenFolder2);
 
     waitUntilAfter(System.currentTimeMillis());
 
     SparkActions actions = SparkActions.get();
+    DeleteOrphanFilesSparkAction action =
+        actions.deleteOrphanFiles(table).olderThan(System.currentTimeMillis());
+    // test list methods by directly instantiating the action
+    assertThatDatasetsAreEqualIgnoringOrder(action.listWithPrefix(), action.listWithoutPrefix());
 
-    DeleteOrphanFiles.Result result =
-        actions.deleteOrphanFiles(table).olderThan(System.currentTimeMillis()).execute();
+    DeleteOrphanFiles.Result result = action.execute();
 
     assertThat(result.orphanFileLocations()).as("Should delete 0 files").isEmpty();
     assertThat(fs.exists(pathToFileInHiddenFolder)).isTrue();
+    assertThat(fs.exists(pathToFileInHiddenFolder2)).isTrue();
   }
 
   private List<String> snapshotFiles(long snapshotId) {
@@ -675,12 +687,10 @@ public void testRemoveOrphanFilesWithRelativeFilePath() throws IOException {
     waitUntilAfter(System.currentTimeMillis());
 
     SparkActions actions = SparkActions.get();
-    DeleteOrphanFiles.Result result =
-        actions
-            .deleteOrphanFiles(table)
-            .olderThan(System.currentTimeMillis())
-            .deleteWith(s -> {})
-            .execute();
+    DeleteOrphanFilesSparkAction action =
+        actions.deleteOrphanFiles(table).olderThan(System.currentTimeMillis()).deleteWith(s -> {});
+    assertThatDatasetsAreEqualIgnoringOrder(action.listWithPrefix(), action.listWithoutPrefix());
+    DeleteOrphanFiles.Result result = action.execute();
     assertThat(result.orphanFileLocations())
         .as("Action should find 1 file")
         .isEqualTo(invalidFiles);
@@ -713,8 +723,11 @@ public void testRemoveOrphanFilesWithHadoopCatalog() throws InterruptedException
 
     table.refresh();
 
-    DeleteOrphanFiles.Result result =
-        SparkActions.get().deleteOrphanFiles(table).olderThan(System.currentTimeMillis()).execute();
+    DeleteOrphanFilesSparkAction action =
+        SparkActions.get().deleteOrphanFiles(table).olderThan(System.currentTimeMillis());
+    assertThatDatasetsAreEqualIgnoringOrder(action.listWithPrefix(), action.listWithoutPrefix());
+
+    DeleteOrphanFiles.Result result = action.execute();
 
     assertThat(result.orphanFileLocations()).as("Should delete only 1 file").hasSize(1);
 
@@ -854,12 +867,14 @@ public void testCompareToFileList() throws IOException {
         .as("Invalid file should be present")
         .isTrue();
 
-    DeleteOrphanFiles.Result result3 =
+    DeleteOrphanFilesSparkAction action3 =
         actions
             .deleteOrphanFiles(table)
             .compareToFileList(compareToFileList)
-            .olderThan(System.currentTimeMillis())
-            .execute();
+            .olderThan(System.currentTimeMillis());
+    assertThatDatasetsAreEqualIgnoringOrder(action3.listWithPrefix(), action3.listWithoutPrefix());
+
+    DeleteOrphanFiles.Result result3 = action3.execute();
     assertThat(result3.orphanFileLocations())
         .as("Action should delete 1 file")
         .isEqualTo(invalidFilePaths);
@@ -885,12 +900,14 @@ public void testCompareToFileList() throws IOException {
             .withColumnRenamed("filePath", "file_path")
             .withColumnRenamed("lastModified", "last_modified");
 
-    DeleteOrphanFiles.Result result4 =
+    DeleteOrphanFilesSparkAction action4 =
         actions
             .deleteOrphanFiles(table)
             .compareToFileList(compareToFileListWithOutsideLocation)
-            .deleteWith(s -> {})
-            .execute();
+            .deleteWith(s -> {});
+    assertThatDatasetsAreEqualIgnoringOrder(action4.listWithPrefix(), action4.listWithoutPrefix());
+
+    DeleteOrphanFiles.Result result4 = action4.execute();
     assertThat(result4.orphanFileLocations()).as("Action should find nothing").isEmpty();
   }
 
@@ -1100,4 +1117,11 @@ private void executeTest(
             spark, toFileUri.apply(actualFileDS), toFileUri.apply(validFileDS), mode);
     assertThat(orphanFiles).isEqualTo(expectedOrphanFiles);
   }
+
+  private void assertThatDatasetsAreEqualIgnoringOrder(
+      Dataset<String> actual, Dataset<String> expected) {
+    assertThat(actual.collectAsList())
+        .as("same as")
+        .containsExactlyInAnyOrderElementsOf(expected.collectAsList());
+  }
 }