hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
/**
   * Filter a list of FileStatus based on commitsToCheck for incremental view.
   * @param job
   * @param tableMetaClient
   * @param timeline
   * @param fileStatuses
   * @param commitsToCheck
   * @return
   */
Filter a list of FileStatus based on commitsToCheck for incremental view.
Filter a list of FileStatus based on commitsToCheck for incremental view .
public static List<FileStatus> filterIncrementalFileStatus(Job job, HoodieTableMetaClient tableMetaClient,
      HoodieTimeline timeline, FileStatus[] fileStatuses, List<HoodieInstant> commitsToCheck) throws IOException {
    TableFileSystemView.BaseFileOnlyView roView = new HoodieTableFileSystemView(tableMetaClient, timeline, fileStatuses);
    List<String> commitsList = commitsToCheck.stream().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
    List<HoodieBaseFile> filteredFiles = roView.getLatestBaseFilesInRange(commitsList).collect(Collectors.toList());
    List<FileStatus> returns = new ArrayList<>();
    for (HoodieBaseFile filteredFile : filteredFiles) {
      LOG.debug("Processing incremental hoodie file - " + filteredFile.getPath());
      filteredFile = refreshFileStatus(job.getConfiguration(), filteredFile);
      returns.add(getFileStatus(filteredFile));
    }
    LOG.info("Total paths to process after hoodie incremental filter " + filteredFiles.size());
    return returns;
  }
public static List < FileStatus > filterIncrementalFileStatus ( Job job , HoodieTableMetaClient tableMetaClient , HoodieTimeline timeline , FileStatus [ ] fileStatuses , List < HoodieInstant > commitsToCheck ) throws IOException { TableFileSystemView . BaseFileOnlyView roView = new HoodieTableFileSystemView ( tableMetaClient , timeline , fileStatuses ) ; List < String > commitsList = commitsToCheck . stream ( ) . map ( HoodieInstant :: getTimestamp ) . collect ( Collectors . toList ( ) ) ; List < HoodieBaseFile > filteredFiles = roView . getLatestBaseFilesInRange ( commitsList ) . collect ( Collectors . toList ( ) ) ; List < FileStatus > returns = new ArrayList < > ( ) ; for ( HoodieBaseFile filteredFile : filteredFiles ) { LOG . debug ( "Processing incremental hoodie file - " + filteredFile . getPath ( ) ) ; filteredFile = refreshFileStatus ( job . getConfiguration ( ) , filteredFile ) ; returns . add ( getFileStatus ( filteredFile ) ) ; } LOG . info ( "Total paths to process after hoodie incremental filter " + filteredFiles . size ( ) ) ; return returns ; }
Filter a list of FileStatus based on commitsToCheck for incremental view.
Filter a list of FileStatus based on commitsToCheck for incremental view .