@Test public void testProcessFiles() throws Exception { List<String> folders = Arrays.asList("2016/04/15", "2016/05/16", ".hoodie/.temp/2/2016/04/15", ".hoodie/.temp/2/2016/05/16"); folders.forEach(f -> { try { metaClient.getFs().mkdirs(new Path(new Path(basePath), f)); } catch (IOException e) { throw new HoodieException(e); } }); List<String> files = Arrays.asList("2016/04/15/1_1-0-1_20190528120000.parquet", "2016/05/16/2_1-0-1_20190528120000.parquet", ".hoodie/.temp/2/2016/05/16/2_1-0-1_20190528120000.parquet", ".hoodie/.temp/2/2016/04/15/1_1-0-1_20190528120000.parquet"); files.forEach(f -> { try { metaClient.getFs().create(new Path(new Path(basePath), f)); } catch (IOException e) { throw new HoodieException(e); } }); final List<String> collected = new ArrayList<>(); FSUtils.processFiles(metaClient.getFs(), basePath, (status) -> { collected.add(status.getPath().toString()); return true; }, true); assertTrue(collected.stream().noneMatch(s -> s.contains(HoodieTableMetaClient.METAFOLDER_NAME)), "Hoodie MetaFolder MUST be skipped but got :" + collected); assertEquals(2, collected.size()); final List<String> collected2 = new ArrayList<>(); FSUtils.processFiles(metaClient.getFs(), basePath, (status) -> { collected2.add(status.getPath().toString()); return true; }, false); assertFalse(collected2.stream().noneMatch(s -> s.contains(HoodieTableMetaClient.METAFOLDER_NAME)), "Hoodie MetaFolder will be present :" + collected2); assertEquals(5, collected2.size(), "Collected=" + collected2); }
public static void processFiles(FileSystem fs, String basePathStr, Function<FileStatus, Boolean> consumer, boolean excludeMetaFolder) throws IOException { PathFilter pathFilter = excludeMetaFolder ? getExcludeMetaPathFilter() : ALLOW_ALL_FILTER; FileStatus[] topLevelStatuses = fs.listStatus(new Path(basePathStr)); for (FileStatus child : topLevelStatuses) { if (child.isFile()) { boolean success = consumer.apply(child); if (!success) { throw new HoodieException("Failed to process file-status=" + child); } } else if (pathFilter.accept(child.getPath())) { RemoteIterator<LocatedFileStatus> itr = fs.listFiles(child.getPath(), true); while (itr.hasNext()) { FileStatus status = itr.next(); boolean success = consumer.apply(status); if (!success) { throw new HoodieException("Failed to process file-status=" + status); } } } } }
FSUtils { public static void processFiles(FileSystem fs, String basePathStr, Function<FileStatus, Boolean> consumer, boolean excludeMetaFolder) throws IOException { PathFilter pathFilter = excludeMetaFolder ? getExcludeMetaPathFilter() : ALLOW_ALL_FILTER; FileStatus[] topLevelStatuses = fs.listStatus(new Path(basePathStr)); for (FileStatus child : topLevelStatuses) { if (child.isFile()) { boolean success = consumer.apply(child); if (!success) { throw new HoodieException("Failed to process file-status=" + child); } } else if (pathFilter.accept(child.getPath())) { RemoteIterator<LocatedFileStatus> itr = fs.listFiles(child.getPath(), true); while (itr.hasNext()) { FileStatus status = itr.next(); boolean success = consumer.apply(status); if (!success) { throw new HoodieException("Failed to process file-status=" + status); } } } } } }
FSUtils { public static void processFiles(FileSystem fs, String basePathStr, Function<FileStatus, Boolean> consumer, boolean excludeMetaFolder) throws IOException { PathFilter pathFilter = excludeMetaFolder ? getExcludeMetaPathFilter() : ALLOW_ALL_FILTER; FileStatus[] topLevelStatuses = fs.listStatus(new Path(basePathStr)); for (FileStatus child : topLevelStatuses) { if (child.isFile()) { boolean success = consumer.apply(child); if (!success) { throw new HoodieException("Failed to process file-status=" + child); } } else if (pathFilter.accept(child.getPath())) { RemoteIterator<LocatedFileStatus> itr = fs.listFiles(child.getPath(), true); while (itr.hasNext()) { FileStatus status = itr.next(); boolean success = consumer.apply(status); if (!success) { throw new HoodieException("Failed to process file-status=" + status); } } } } }  }
FSUtils { public static void processFiles(FileSystem fs, String basePathStr, Function<FileStatus, Boolean> consumer, boolean excludeMetaFolder) throws IOException { PathFilter pathFilter = excludeMetaFolder ? getExcludeMetaPathFilter() : ALLOW_ALL_FILTER; FileStatus[] topLevelStatuses = fs.listStatus(new Path(basePathStr)); for (FileStatus child : topLevelStatuses) { if (child.isFile()) { boolean success = consumer.apply(child); if (!success) { throw new HoodieException("Failed to process file-status=" + child); } } else if (pathFilter.accept(child.getPath())) { RemoteIterator<LocatedFileStatus> itr = fs.listFiles(child.getPath(), true); while (itr.hasNext()) { FileStatus status = itr.next(); boolean success = consumer.apply(status); if (!success) { throw new HoodieException("Failed to process file-status=" + status); } } } } }  static Configuration prepareHadoopConf(Configuration conf); static FileSystem getFs(String path, Configuration conf); static FileSystem getFs(String path, Configuration conf, boolean localByDefault); static Path addSchemeIfLocalPath(String path); static String makeWriteToken(int taskPartitionId, int stageId, long taskAttemptId); static String makeDataFileName(String instantTime, String writeToken, String fileId); static String makeDataFileName(String instantTime, String writeToken, String fileId, String fileExtension); static String makeBootstrapIndexFileName(String instantTime, String fileId, String ext); static String maskWithoutFileId(String instantTime, int taskPartitionId); static String getCommitFromCommitFile(String commitFileName); static String getCommitTime(String fullFileName); static long getFileSize(FileSystem fs, Path path); static String getFileId(String fullFileName); static List<String> getAllPartitionFoldersThreeLevelsDown(FileSystem fs, String basePath); static String getRelativePartitionPath(Path basePath, Path fullPartitionPath); static List<String> getAllFoldersWithPartitionMetaFile(FileSystem fs, String basePathStr); static void processFiles(FileSystem fs, String basePathStr, Function<FileStatus, Boolean> consumer,
                                  boolean excludeMetaFolder); static List<String> getAllPartitionPaths(FileSystem fs, String basePathStr, boolean assumeDatePartitioning); static String getFileExtension(String fullName); static String createNewFileIdPfx(); static String getFileExtensionFromLog(Path logPath); static String getFileIdFromLogPath(Path path); static String getFileIdFromFilePath(Path filePath); static String getBaseCommitTimeFromLogPath(Path path); static Integer getTaskPartitionIdFromLogPath(Path path); static String getWriteTokenFromLogPath(Path path); static Integer getStageIdFromLogPath(Path path); static Integer getTaskAttemptIdFromLogPath(Path path); static int getFileVersionFromLog(Path logPath); static String makeLogFileName(String fileId, String logFileExtension, String baseCommitTime, int version,
      String writeToken); static boolean isLogFile(Path logPath); static Option<HoodieLogFile> getLatestLogFile(Stream<HoodieLogFile> logFiles); static Stream<HoodieLogFile> getAllLogFiles(FileSystem fs, Path partitionPath, final String fileId,
      final String logFileExtension, final String baseCommitTime); static Option<Pair<Integer, String>> getLatestLogVersion(FileSystem fs, Path partitionPath,
      final String fileId, final String logFileExtension, final String baseCommitTime); static int computeNextLogVersion(FileSystem fs, Path partitionPath, final String fileId,
      final String logFileExtension, final String baseCommitTime); static int getDefaultBufferSize(final FileSystem fs); static Short getDefaultReplication(FileSystem fs, Path path); static boolean recoverDFSFileLease(final DistributedFileSystem dfs, final Path p); static void deleteOlderCleanMetaFiles(FileSystem fs, String metaPath, Stream<HoodieInstant> instants); static void deleteOlderRollbackMetaFiles(FileSystem fs, String metaPath, Stream<HoodieInstant> instants); static void deleteInstantFile(FileSystem fs, String metaPath, HoodieInstant instant); static void createPathIfNotExists(FileSystem fs, Path partitionPath); static Long getSizeInMB(long sizeInBytes); static Path getPartitionPath(String basePath, String partitionPath); static Path getPartitionPath(Path basePath, String partitionPath); static String getDFSFullPartitionPath(FileSystem fs, Path fullPartitionPath); static boolean isGCSInputStream(FSDataInputStream inputStream); static Configuration registerFileSystem(Path file, Configuration conf); static HoodieWrapperFileSystem getFs(String path, SerializableConfiguration hadoopConf,
      ConsistencyGuardConfig consistencyGuardConfig); static List<FileStatus> getGlobStatusExcludingMetaFolder(FileSystem fs, Path globPath); }
FSUtils { public static void processFiles(FileSystem fs, String basePathStr, Function<FileStatus, Boolean> consumer, boolean excludeMetaFolder) throws IOException { PathFilter pathFilter = excludeMetaFolder ? getExcludeMetaPathFilter() : ALLOW_ALL_FILTER; FileStatus[] topLevelStatuses = fs.listStatus(new Path(basePathStr)); for (FileStatus child : topLevelStatuses) { if (child.isFile()) { boolean success = consumer.apply(child); if (!success) { throw new HoodieException("Failed to process file-status=" + child); } } else if (pathFilter.accept(child.getPath())) { RemoteIterator<LocatedFileStatus> itr = fs.listFiles(child.getPath(), true); while (itr.hasNext()) { FileStatus status = itr.next(); boolean success = consumer.apply(status); if (!success) { throw new HoodieException("Failed to process file-status=" + status); } } } } }  static Configuration prepareHadoopConf(Configuration conf); static FileSystem getFs(String path, Configuration conf); static FileSystem getFs(String path, Configuration conf, boolean localByDefault); static Path addSchemeIfLocalPath(String path); static String makeWriteToken(int taskPartitionId, int stageId, long taskAttemptId); static String makeDataFileName(String instantTime, String writeToken, String fileId); static String makeDataFileName(String instantTime, String writeToken, String fileId, String fileExtension); static String makeBootstrapIndexFileName(String instantTime, String fileId, String ext); static String maskWithoutFileId(String instantTime, int taskPartitionId); static String getCommitFromCommitFile(String commitFileName); static String getCommitTime(String fullFileName); static long getFileSize(FileSystem fs, Path path); static String getFileId(String fullFileName); static List<String> getAllPartitionFoldersThreeLevelsDown(FileSystem fs, String basePath); static String getRelativePartitionPath(Path basePath, Path fullPartitionPath); static List<String> getAllFoldersWithPartitionMetaFile(FileSystem fs, String basePathStr); static void processFiles(FileSystem fs, String basePathStr, Function<FileStatus, Boolean> consumer,
                                  boolean excludeMetaFolder); static List<String> getAllPartitionPaths(FileSystem fs, String basePathStr, boolean assumeDatePartitioning); static String getFileExtension(String fullName); static String createNewFileIdPfx(); static String getFileExtensionFromLog(Path logPath); static String getFileIdFromLogPath(Path path); static String getFileIdFromFilePath(Path filePath); static String getBaseCommitTimeFromLogPath(Path path); static Integer getTaskPartitionIdFromLogPath(Path path); static String getWriteTokenFromLogPath(Path path); static Integer getStageIdFromLogPath(Path path); static Integer getTaskAttemptIdFromLogPath(Path path); static int getFileVersionFromLog(Path logPath); static String makeLogFileName(String fileId, String logFileExtension, String baseCommitTime, int version,
      String writeToken); static boolean isLogFile(Path logPath); static Option<HoodieLogFile> getLatestLogFile(Stream<HoodieLogFile> logFiles); static Stream<HoodieLogFile> getAllLogFiles(FileSystem fs, Path partitionPath, final String fileId,
      final String logFileExtension, final String baseCommitTime); static Option<Pair<Integer, String>> getLatestLogVersion(FileSystem fs, Path partitionPath,
      final String fileId, final String logFileExtension, final String baseCommitTime); static int computeNextLogVersion(FileSystem fs, Path partitionPath, final String fileId,
      final String logFileExtension, final String baseCommitTime); static int getDefaultBufferSize(final FileSystem fs); static Short getDefaultReplication(FileSystem fs, Path path); static boolean recoverDFSFileLease(final DistributedFileSystem dfs, final Path p); static void deleteOlderCleanMetaFiles(FileSystem fs, String metaPath, Stream<HoodieInstant> instants); static void deleteOlderRollbackMetaFiles(FileSystem fs, String metaPath, Stream<HoodieInstant> instants); static void deleteInstantFile(FileSystem fs, String metaPath, HoodieInstant instant); static void createPathIfNotExists(FileSystem fs, Path partitionPath); static Long getSizeInMB(long sizeInBytes); static Path getPartitionPath(String basePath, String partitionPath); static Path getPartitionPath(Path basePath, String partitionPath); static String getDFSFullPartitionPath(FileSystem fs, Path fullPartitionPath); static boolean isGCSInputStream(FSDataInputStream inputStream); static Configuration registerFileSystem(Path file, Configuration conf); static HoodieWrapperFileSystem getFs(String path, SerializableConfiguration hadoopConf,
      ConsistencyGuardConfig consistencyGuardConfig); static List<FileStatus> getGlobStatusExcludingMetaFolder(FileSystem fs, Path globPath);  }