diff --git a/hudi-common/src/main/java/org/apache/hudi/metadata/FileSystemBackedTableMetadata.java b/hudi-common/src/main/java/org/apache/hudi/metadata/FileSystemBackedTableMetadata.java index b77bb12c49447..f029995ba0122 100644 --- a/hudi-common/src/main/java/org/apache/hudi/metadata/FileSystemBackedTableMetadata.java +++ b/hudi-common/src/main/java/org/apache/hudi/metadata/FileSystemBackedTableMetadata.java @@ -68,13 +68,14 @@ public FileStatus[] getAllFilesInPartition(Path partitionPath) throws IOExceptio @Override public List getAllPartitionPaths() throws IOException { - FileSystem fs = new Path(datasetBasePath).getFileSystem(hadoopConf.get()); + Path basePath = new Path(datasetBasePath); + FileSystem fs = basePath.getFileSystem(hadoopConf.get()); if (assumeDatePartitioning) { return FSUtils.getAllPartitionFoldersThreeLevelsDown(fs, datasetBasePath); } List pathsToList = new CopyOnWriteArrayList<>(); - pathsToList.add(new Path(datasetBasePath)); + pathsToList.add(basePath); List partitionPaths = new CopyOnWriteArrayList<>(); while (!pathsToList.isEmpty()) { @@ -82,27 +83,31 @@ public List getAllPartitionPaths() throws IOException { int listingParallelism = Math.min(DEFAULT_LISTING_PARALLELISM, pathsToList.size()); // List all directories in parallel - List dirToFileListing = engineContext.map(pathsToList, path -> { + List> dirToFileListing = engineContext.map(pathsToList, path -> { FileSystem fileSystem = path.getFileSystem(hadoopConf.get()); - return fileSystem.listStatus(path); + return Pair.of(path, fileSystem.listStatus(path)); }, listingParallelism); pathsToList.clear(); // if current dictionary contains PartitionMetadata, add it to result // if current dictionary does not contain PartitionMetadata, add it to queue - dirToFileListing.stream().flatMap(Arrays::stream).parallel() - .forEach(fileStatus -> { - if (fileStatus.isDirectory()) { - if (HoodiePartitionMetadata.hasPartitionMetadata(fs, fileStatus.getPath())) { - partitionPaths.add(FSUtils.getRelativePartitionPath(new Path(datasetBasePath), fileStatus.getPath())); - } else if (!fileStatus.getPath().getName().equals(HoodieTableMetaClient.METAFOLDER_NAME)) { - pathsToList.add(fileStatus.getPath()); - } - } else if (fileStatus.getPath().getName().startsWith(HoodiePartitionMetadata.HOODIE_PARTITION_METAFILE_PREFIX)) { - String partitionName = FSUtils.getRelativePartitionPath(new Path(datasetBasePath), fileStatus.getPath().getParent()); - partitionPaths.add(partitionName); - } - }); + dirToFileListing.forEach(p -> { + Option partitionMetaFile = Option.fromJavaOptional(Arrays.stream(p.getRight()).parallel() + .filter(fileStatus -> fileStatus.getPath().getName().startsWith(HoodiePartitionMetadata.HOODIE_PARTITION_METAFILE_PREFIX)) + .findFirst()); + + if (partitionMetaFile.isPresent()) { + // Is a partition. + String partitionName = FSUtils.getRelativePartitionPath(basePath, p.getLeft()); + partitionPaths.add(partitionName); + } else { + // Add sub-dirs to the queue + pathsToList.addAll(Arrays.stream(p.getRight()) + .filter(fileStatus -> fileStatus.isDirectory() && !fileStatus.getPath().getName().equals(HoodieTableMetaClient.METAFOLDER_NAME)) + .map(fileStatus -> fileStatus.getPath()) + .collect(Collectors.toList())); + } + }); } return partitionPaths; }