pvary commented on code in PR #12254:
URL: https://github.com/apache/iceberg/pull/12254#discussion_r2079339005
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java:
##########
@@ -303,39 +310,88 @@ private Dataset<String> listedFileDS() {
List<String> subDirs = Lists.newArrayList();
List<String> matchingFiles = Lists.newArrayList();
- Predicate<FileStatus> predicate = file -> file.getModificationTime() <
olderThanTimestamp;
PathFilter pathFilter =
PartitionAwareHiddenPathFilter.forSpecs(table.specs());
- // list at most MAX_DRIVER_LISTING_DEPTH levels and only dirs that have
- // less than MAX_DRIVER_LISTING_DIRECT_SUB_DIRS direct sub dirs on the
driver
- listDirRecursively(
- location,
- predicate,
- hadoopConf.value(),
- MAX_DRIVER_LISTING_DEPTH,
- MAX_DRIVER_LISTING_DIRECT_SUB_DIRS,
- subDirs,
- pathFilter,
- matchingFiles);
-
- JavaRDD<String> matchingFileRDD =
sparkContext().parallelize(matchingFiles, 1);
-
- if (subDirs.isEmpty()) {
+ if (usePrefixList) {
+ Preconditions.checkArgument(
+ table.io() instanceof SupportsPrefixOperations,
+ "Table file io should prefix operations when using prefix list.");
+
+ Predicate<org.apache.iceberg.io.FileInfo> predicate =
+ fileInfo -> fileInfo.createdAtMillis() < olderThanTimestamp;
+ listDirRecursivelyWithFileIO(
+ (SupportsPrefixOperations) table.io(), location, predicate,
pathFilter, matchingFiles);
+
+ JavaRDD<String> matchingFileRDD =
sparkContext().parallelize(matchingFiles, 1);
return spark().createDataset(matchingFileRDD.rdd(), Encoders.STRING());
+ } else {
+ Predicate<FileStatus> predicate = file -> file.getModificationTime() <
olderThanTimestamp;
+ // list at most MAX_DRIVER_LISTING_DEPTH levels and only dirs that have
+ // less than MAX_DRIVER_LISTING_DIRECT_SUB_DIRS direct sub dirs on the
driver
+ listDirRecursivelyWithHadoop(
+ location,
+ predicate,
+ hadoopConf.value(),
+ MAX_DRIVER_LISTING_DEPTH,
+ MAX_DRIVER_LISTING_DIRECT_SUB_DIRS,
+ subDirs,
+ pathFilter,
+ matchingFiles);
+
+ JavaRDD<String> matchingFileRDD =
sparkContext().parallelize(matchingFiles, 1);
+
+ if (subDirs.isEmpty()) {
+ return spark().createDataset(matchingFileRDD.rdd(), Encoders.STRING());
+ }
+
+ int parallelism = Math.min(subDirs.size(), listingParallelism);
+ JavaRDD<String> subDirRDD = sparkContext().parallelize(subDirs,
parallelism);
+
+ Broadcast<SerializableConfiguration> conf =
sparkContext().broadcast(hadoopConf);
+ ListDirsRecursively listDirs = new ListDirsRecursively(conf,
olderThanTimestamp, pathFilter);
+ JavaRDD<String> matchingLeafFileRDD = subDirRDD.mapPartitions(listDirs);
+
+ JavaRDD<String> completeMatchingFileRDD =
matchingFileRDD.union(matchingLeafFileRDD);
+ return spark().createDataset(completeMatchingFileRDD.rdd(),
Encoders.STRING());
}
+ }
- int parallelism = Math.min(subDirs.size(), listingParallelism);
- JavaRDD<String> subDirRDD = sparkContext().parallelize(subDirs,
parallelism);
+ private static void listDirRecursivelyWithFileIO(
+ SupportsPrefixOperations io,
+ String dir,
+ Predicate<org.apache.iceberg.io.FileInfo> predicate,
+ PathFilter pathFilter,
+ List<String> matchingFiles) {
+ String listPath = dir;
+ if (!dir.endsWith("/")) {
+ listPath = dir + "/";
+ }
- Broadcast<SerializableConfiguration> conf =
sparkContext().broadcast(hadoopConf);
- ListDirsRecursively listDirs = new ListDirsRecursively(conf,
olderThanTimestamp, pathFilter);
- JavaRDD<String> matchingLeafFileRDD = subDirRDD.mapPartitions(listDirs);
+ Iterable<org.apache.iceberg.io.FileInfo> files = io.listPrefix(listPath);
+ for (org.apache.iceberg.io.FileInfo file : files) {
+ Path path = new Path(file.location());
+ if (!isHiddenPath(dir, path, pathFilter) && predicate.test(file)) {
+ matchingFiles.add(file.location());
+ }
+ }
+ }
+
+ private static boolean isHiddenPath(String baseDir, Path path, PathFilter
pathFilter) {
Review Comment:
Do we have CRC files, or something which might not be part of the table
metadata, but needed by the FS?
--
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-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
