Draft use multiple listing when its ObjectStoreLocationProvider

Author: ismailsimsek

api/src/main/java/org/apache/iceberg/io/LocationProvider.java

@@ -45,4 +45,6 @@ public interface LocationProvider extends Serializable {
    * @return a fully-qualified location URI for a data file
    */
   String newDataLocation(PartitionSpec spec, StructLike partitionData, String filename);
+
+  String dataLocationRoot();
 }

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

@@ -102,9 +102,14 @@ public String newDataLocation(PartitionSpec spec, StructLike partitionData, Stri
     public String newDataLocation(String filename) {
       return String.format("%s/%s", dataLocation, filename);
     }
+
+    @Override
+    public String dataLocationRoot() {
+      return dataLocation;
+    }
   }
 
-  static class ObjectStoreLocationProvider implements LocationProvider {
+  public static class ObjectStoreLocationProvider implements LocationProvider {
 
     private static final HashFunction HASH_FUNC = Hashing.murmur3_32_fixed();
     // Length of entropy generated in the file location
@@ -228,5 +233,10 @@ private String dirsFromHash(String hash) {
 
       return hashWithDirs.toString();
     }
+
+    @Override
+    public String dataLocationRoot() {
+      return storageLocation;
+    }
   }
 }

core/src/test/java/org/apache/iceberg/TestLocationProvider.java

@@ -57,6 +57,11 @@ public String newDataLocation(String filename) {
     public String newDataLocation(PartitionSpec spec, StructLike partitionData, String filename) {
       throw new RuntimeException("Test custom provider does not expect any invocation");
     }
+
+    @Override
+    public String dataLocationRoot() {
+      return tableLocation;
+    }
   }
 
   // publicly visible for testing to be dynamically loaded
@@ -72,6 +77,11 @@ public String newDataLocation(String filename) {
     public String newDataLocation(PartitionSpec spec, StructLike partitionData, String filename) {
       throw new RuntimeException("Test custom provider does not expect any invocation");
     }
+
+    @Override
+    public String dataLocationRoot() {
+      return "";
+    }
   }
 
   // publicly visible for testing to be dynamically loaded
@@ -88,6 +98,11 @@ public String newDataLocation(String filename) {
     public String newDataLocation(PartitionSpec spec, StructLike partitionData, String filename) {
       throw new RuntimeException("Invalid provider should have not been instantiated!");
     }
+
+    @Override
+    public String dataLocationRoot() {
+      return "";
+    }
   }
 
   // publicly visible for testing to be dynamically loaded

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

@@ -43,6 +43,7 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.PathFilter;
+import org.apache.iceberg.LocationProviders;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.actions.DeleteOrphanFiles;
@@ -306,25 +307,46 @@ private Dataset<FileURI> actualFileIdentDS() {
   }
 
   @VisibleForTesting
-  Dataset<String> listWithPrefix() {
+  List<String> listLocationWithPrefix(String location, PathFilter pathFilter) {
     List<String> matchingFiles = Lists.newArrayList();
-    // listPrefix only returns files. so we additionally need to check parent folders for each file
-    // in following example file itself is not filtered out,
-    // but it should be excluded due to its parent folder: `_c2_trunc`
-    // "/data/_c2_trunc/file.txt"
-    PathFilter pathFilter = PartitionAwareHiddenPathFilter.forSpecs(table.specs(), true);
-
     Iterator<org.apache.iceberg.io.FileInfo> iterator =
         ((SupportsPrefixOperations) table.io()).listPrefix(location).iterator();
     while (iterator.hasNext()) {
       org.apache.iceberg.io.FileInfo fileInfo = iterator.next();
-      // NOTE: check the path relative to table location. To avoid checking un necessary root
-      // folders
+      // NOTE: To avoid checking un necessary root folders, check the path relative to table
+      // location.
       Path relativeFilePath = new Path(fileInfo.location().replace(location, ""));
       if (fileInfo.createdAtMillis() < olderThanTimestamp && pathFilter.accept(relativeFilePath)) {
         matchingFiles.add(fileInfo.location());
       }
     }
+    return matchingFiles;
+  }
+
+  @VisibleForTesting
+  Dataset<String> listWithPrefix() {
+    List<String> matchingFiles = Lists.newArrayList();
+    PathFilter pathFilter = PartitionAwareHiddenPathFilter.forSpecs(table.specs(), true);
+
+    if (table.locationProvider() instanceof LocationProviders.ObjectStoreLocationProvider) {
+      // ObjectStoreLocationProvider generates hierarchical prefixes in a binary fashion
+      // (0000/, 0001/, 0010/, 0011/, ...).
+      // This allows us to parallelize listing operations across these prefixes.
+      List<String> prefixes =
+          List.of(
+              "/0000", "/0001", "/0010", "/0011", "/0100", "/0101", "/0110", "/0111", "/1000",
+              "/1001", "/1010", "/1011", "/1100", "/1101", "/1110", "/1111");
+
+      String tableDataLocationRoot = table.locationProvider().dataLocationRoot();
+      for (String prefix : prefixes) {
+        List<String> result = listLocationWithPrefix(tableDataLocationRoot + prefix, pathFilter);
+        matchingFiles.addAll(result);
+      }
+
+    } else {
+      matchingFiles.addAll(listLocationWithPrefix(location, pathFilter));
+    }
+
     JavaRDD<String> matchingFileRDD = sparkContext().parallelize(matchingFiles, 1);
     return spark().createDataset(matchingFileRDD.rdd(), Encoders.STRING());
   }

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

@@ -60,6 +60,7 @@
 import org.apache.iceberg.actions.DeleteOrphanFiles;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.hadoop.HadoopCatalog;
 import org.apache.iceberg.hadoop.HadoopTables;
@@ -515,6 +516,7 @@ public void testManyLeafPartitions() {
 
   @TestTemplate
   public void testHiddenPartitionPaths() {
+    //
     Schema schema =
         new Schema(
             optional(1, "c1", Types.IntegerType.get()),
@@ -1090,6 +1092,27 @@ public void testRemoveOrphanFileActionWithDeleteMode() {
         DeleteOrphanFiles.PrefixMismatchMode.DELETE);
   }
 
+  @TestTemplate
+  public void testLocationProvider() {
+    ImmutableMap<String, String> props =
+        ImmutableMap.of(
+            TableProperties.FORMAT_VERSION,
+            String.valueOf(formatVersion),
+            TableProperties.OBJECT_STORE_ENABLED,
+            "true");
+    System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).truncate("c2", 2).identity("c3").build();
+    Table table = TABLES.create(SCHEMA, spec, props, tableLocation);
+    System.out.println(">: " + table.location());
+    System.out.println(">: " + table.locationProvider().newDataLocation("test"));
+    GenericRecord record =
+        GenericRecord.create(SCHEMA).copy(Map.of("c1", "54321", "c2", "bbbb", "c3", "cccc"));
+    System.out.println(">: " + table.locationProvider().newDataLocation(spec, record, "test"));
+    System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
+    assertThat(table.locationProvider().newDataLocation("test")).as("equals").isEqualTo("xxx");
+    assertThat(table.locationProvider().getClass().toString()).as("equals").isEqualTo("xxx");
+  }
+
   protected String randomName(String prefix) {
     return prefix + UUID.randomUUID().toString().replace("-", "");
   }