anuragmantri commented on code in PR #17790:
URL: https://github.com/apache/iceberg/pull/17790#discussion_r4262055574
##########
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:
I think `isHiddenPath` can miss hidden directories when `baseDir` ends in
`/`. It checks `parent.toString().contains(baseDir)`, but `Path.toString()`
never has a trailing slash, so the walk up stops one level early and a
directory like `_tmp` directly under the seed is never examined. Sub-prefixes
from `listPrefixWithDelimiter` always end in `/`, so seeding a later walk with
one hits this.
I tried it with `HadoopFileIO` on `part=1/{keep.parquet, _tmp/leak1.parquet,
.hidden/leak2.parquet}`. The seed `.../part=1` returns only `keep.parquet`, and
`.../part=1/` returns all three. Those files would then look like orphans.
Would normalizing with `new Path(baseDir).toString()` inside `isHiddenPath`
fix it? It might be worth a test that takes a sub-prefix from
`directoryConsumer` and walks it again, asserting that `_` and `.` directories
directly under it are skipped.
##########
azure/src/main/java/org/apache/iceberg/azure/adlsv2/ADLSFileIO.java:
##########
@@ -242,6 +243,66 @@ public Iterable<FileInfo> listPrefix(String prefix) {
};
}
+ @Override
+ public Iterable<PrefixListingPage> listPrefixWithDelimiter(String prefix,
String delimiter) {
+ if (!"/".equals(delimiter)) {
+ throw new UnsupportedOperationException(
+ String.format("Prefix listing with delimiter '%s' is not supported",
delimiter));
+ }
+
+ ADLSLocation location = new ADLSLocation(prefix);
+ String baseUri = toBaseUri(location);
+
+ ListPathsOptions options = new ListPathsOptions();
+ options.setPath(location.path());
+ options.setRecursive(false);
+
+ return () -> {
+ try {
+ return Streams.stream(client(location).listPaths(options,
null).iterableByPage())
+ .map(response -> createPrefixListingPage(baseUri,
response.getElements()))
+ .iterator();
+ } catch (DataLakeStorageException e) {
+ if (e.getStatusCode() != 404) {
+ throw e;
+ }
+
+ return Collections.emptyIterator();
+ }
+ };
+ }
+
+ @Override
+ public boolean supportsPrefixListingWithDelimiter(String prefix, String
delimiter) {
+ return "/".equals(delimiter);
+ }
+
+ private String toBaseUri(ADLSLocation location) {
+ return String.format(
+ "%s://%s@%s/", location.scheme(), location.container().orElse(""),
location.host());
Review Comment:
This overlaps with #18400, which also makes `listPrefix` return
fully-qualified URIs and adds `ADLSLocation.scheme()`, so the two PRs will
conflict. For a location without a container, `orElse("")` produces
`abfss://@host/...`, while #18400 produces `abfss://host/...`. `ADLSLocation`
accepts container-less URIs, so `remove_orphan_files` would hit a prefix
mismatch there.
Could the `listPrefix(prefix)` and `ADLSLocation` changes come out of this
PR, with #18400 landing first and the new method reusing its location helper?
##########
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(
Review Comment:
Could this public overload get a short Javadoc? It would help to say that
the iterator is lazy and which forms of `dir` are accepted (see the
trailing-slash comment below).
--
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]
