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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
