Re: [PR] API, AWS, Azure, Core, GCP: Add delimiter-based prefix listing to FileIO [iceberg]

["tom-s-powell (via GitHub)" <gi...@apache.org>]

tom-s-powell commented on code in PR #17790:
URL: https://github.com/apache/iceberg/pull/17790#discussion_r4332668555


##########
core/src/main/java/org/apache/iceberg/util/FileSystemWalker.java:
##########
@@ -63,18 +67,134 @@ public static void listDirRecursivelyWithFileIO(
       Map<Integer, PartitionSpec> specs,
       Predicate<FileInfo> filter,
       Consumer<String> fileConsumer) {
+    listDirRecursivelyWithFileIO(io, dir, specs, filter).forEachRemaining(fileConsumer);
+  }
+
+  public static Iterator<String> listDirRecursivelyWithFileIO(
+      SupportsPrefixOperations io,
+      String dir,
+      Map<Integer, PartitionSpec> specs,
+      Predicate<FileInfo> filter) {
     PathFilter pathFilter = PartitionAwareHiddenPathFilter.forSpecs(specs);
     String listPath = dir;
     if (!dir.endsWith("/")) {
       listPath = dir + "/";
     }
 
-    Iterable<FileInfo> files = io.listPrefix(listPath);
-    for (FileInfo file : files) {
-      Path path = new Path(file.location());
-      if (!isHiddenPath(dir, path, pathFilter) && filter.test(file)) {
-        fileConsumer.accept(file.location());
+    Iterator<FileInfo> files = io.listPrefix(listPath).iterator();
+    return Iterators.transform(
+        Iterators.filter(
+            files,
+            file -> {
+              Path path = new Path(file.location());
+              return !isHiddenPath(dir, path, pathFilter) && filter.test(file);
+            }),
+        FileInfo::location);
+  }
+
+  /**
+   * Recursively lists files in the specified directory that satisfy the given conditions. Use
+   * {@link PartitionAwareHiddenPathFilter} to filter out hidden paths.
+   *
+   * <p>Provides the same depth-and-fan-out controls as {@link #listDirRecursivelyWithHadoop}:
+   *
+   * <ul>
+   *   <li>Stops traversal when the maximum recursion depth is reached and adds the current location
+   *       to the pending list via {@code directoryConsumer}.
+   *   <li>Stops traversal when the number of direct sub-prefixes at a level exceeds the threshold
+   *       and adds those sub-prefixes to the pending list.
+   * </ul>
+   *
+   * @param io FileIO implementation that supports prefix listing
+   * @param dir the starting prefix to traverse
+   * @param specs partition specs used to preserve partition-name-based hidden paths
+   * @param filter file filter; only files satisfying this condition will be collected
+   * @param maxDepth maximum recursion depth
+   * @param maxDirectSubDirs upper limit of sub-prefixes that can be processed directly
+   * @param directoryConsumer consumer for sub-prefixes that were not expanded further
+   * @param fileConsumer consumer for qualifying file locations
+   */
+  public static void listDirRecursivelyWithFileIO(
+      SupportsPrefixOperations io,
+      String dir,
+      Map<Integer, PartitionSpec> specs,
+      Predicate<FileInfo> filter,
+      int maxDepth,
+      int maxDirectSubDirs,
+      Consumer<String> directoryConsumer,
+      Consumer<String> fileConsumer) {
+    PathFilter pathFilter = PartitionAwareHiddenPathFilter.forSpecs(specs);
+    listDirRecursivelyWithFileIO(
+        io,
+        dir,
+        dir,
+        pathFilter,
+        filter,
+        maxDepth,
+        maxDirectSubDirs,
+        directoryConsumer,
+        fileConsumer);
+  }
+
+  private static void listDirRecursivelyWithFileIO(
+      SupportsPrefixOperations io,
+      String baseDir,
+      String dir,
+      PathFilter pathFilter,
+      Predicate<FileInfo> filter,
+      int maxDepth,
+      int maxDirectSubDirs,
+      Consumer<String> directoryConsumer,
+      Consumer<String> fileConsumer) {
+    if (maxDepth <= 0) {
+      directoryConsumer.accept(dir);
+      return;
+    }
+
+    String listPath = dir.endsWith("/") ? dir : dir + "/";
+    Preconditions.checkArgument(
+        io.supportsPrefixListingWithDelimiter(listPath, "/"),
+        "FileIO does not support prefix listing with '/' delimiter: %s",
+        io.getClass().getName());
+    Iterable<PrefixListingPage> listing = io.listPrefixWithDelimiter(listPath, "/");
+
+    List<String> subDirs = Lists.newArrayList();
+    boolean maxDirectSubDirsExceeded = false;
+    for (PrefixListingPage page : listing) {
+      for (String subPrefix : page.subPrefixes()) {
+        if (!isHiddenPath(baseDir, new Path(subPrefix), pathFilter)) {

Review Comment:
   Fixed, thanks for flagging



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org