hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
/**
   * Filters data files for a snapshot queried table.
   * @param job
   * @param metadata
   * @param fileStatuses
   * @return
   */
Filters data files for a snapshot queried table.
Filters data files for a snapshot queried table .
public static List<FileStatus> filterFileStatusForSnapshotMode(
      JobConf job, HoodieTableMetaClient metadata, List<FileStatus> fileStatuses) throws IOException {
    FileStatus[] statuses = fileStatuses.toArray(new FileStatus[0]);
    if (LOG.isDebugEnabled()) {
      LOG.debug("Hoodie Metadata initialized with completed commit Ts as :" + metadata);
    }
    HoodieTimeline timeline = metadata.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
    TableFileSystemView.BaseFileOnlyView roView = new HoodieTableFileSystemView(metadata, timeline, statuses);
    List<HoodieBaseFile> filteredFiles = roView.getLatestBaseFiles().collect(Collectors.toList());
    LOG.info("Total paths to process after hoodie filter " + filteredFiles.size());
    List<FileStatus> returns = new ArrayList<>();
    for (HoodieBaseFile filteredFile : filteredFiles) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Processing latest hoodie file - " + filteredFile.getPath());
      }
      filteredFile = refreshFileStatus(job, filteredFile);
      returns.add(getFileStatus(filteredFile));
    }
    return returns;
  }
public static List < FileStatus > filterFileStatusForSnapshotMode ( JobConf job , HoodieTableMetaClient metadata , List < FileStatus > fileStatuses ) throws IOException { FileStatus [ ] statuses = fileStatuses . toArray ( new FileStatus [ 0 ] ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Hoodie Metadata initialized with completed commit Ts as :" + metadata ) ; } HoodieTimeline timeline = metadata . getActiveTimeline ( ) . getCommitsTimeline ( ) . filterCompletedInstants ( ) ; TableFileSystemView . BaseFileOnlyView roView = new HoodieTableFileSystemView ( metadata , timeline , statuses ) ; List < HoodieBaseFile > filteredFiles = roView . getLatestBaseFiles ( ) . collect ( Collectors . toList ( ) ) ; LOG . info ( "Total paths to process after hoodie filter " + filteredFiles . size ( ) ) ; List < FileStatus > returns = new ArrayList < > ( ) ; for ( HoodieBaseFile filteredFile : filteredFiles ) { if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Processing latest hoodie file - " + filteredFile . getPath ( ) ) ; } filteredFile = refreshFileStatus ( job , filteredFile ) ; returns . add ( getFileStatus ( filteredFile ) ) ; } return returns ; }
Filters data files for a snapshot queried table.
Filters data files for a snapshot queried table .
// Get all commits, delta commits, compactions, as all of them produce a base parquet file today // filter files on the latest commit found