indeedjcohorn commented on code in PR #16933:
URL: https://github.com/apache/iceberg/pull/16933#discussion_r3660199430
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -422,17 +511,59 @@ private Dataset<String> listedFileDS() {
"Cannot use prefix listing with FileIO {} which does not support
prefix operations.",
table.io());
- Predicate<org.apache.iceberg.io.FileInfo> predicate =
- fileInfo -> fileInfo.createdAtMillis() < olderThanTimestamp;
- FileSystemWalker.listDirRecursivelyWithFileIO(
- (SupportsPrefixOperations) table.io(),
+ if (!parallelPrefixListing()) {
+ // Driver-side fallback: listPrefix() recursively enumerates the
entire table on a single
+ // driver thread. Kept as an escape hatch behind
parallel-prefix-listing=false.
+ Predicate<org.apache.iceberg.io.FileInfo> predicate =
+ fileInfo -> fileInfo.createdAtMillis() < olderThanTimestamp;
+ FileSystemWalker.listDirRecursivelyWithFileIO(
+ (SupportsPrefixOperations) table.io(),
+ location,
+ table.specs(),
+ predicate,
+ matchingFiles::add);
+
+ JavaRDD<String> matchingFileRDD =
sparkContext().parallelize(matchingFiles, 1);
Review Comment:
FYI we believe hard coding to a single partition here(as it was previously)
might be leading to an OOM in some cases:
https://github.com/apache/iceberg/issues/17387
--
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]
