hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
/**
   * Filter any specific instants that we do not want to process.
   * example timeline:
   *
   * t0 -> create bucket1.parquet
   * t1 -> create and append updates bucket1.log
   * t2 -> request compaction
   * t3 -> create bucket2.parquet
   *
   * if compaction at t2 takes a long time, incremental readers on RO tables can move to t3 and would skip updates in t1
   *
   * To workaround this problem, we want to stop returning data belonging to commits > t2.
   * After compaction is complete, incremental reader would see updates in t2, t3, so on.
   * @param timeline
   * @return
   */
Filter any specific instants that we do not want to process.
example timeline.

> create bucket1.parquet
t1 -> create and append updates bucket1.log
t2 -> request compaction
t3 -> create bucket2.parquet

if compaction at t2 takes a long time, incremental readers on RO tables can move to t3 and would skip updates in t1

To workaround this problem, we want to stop returning data belonging to commits > t2.
After compaction is complete, incremental reader would see updates in t2, t3, so on.
@param timeline
@return
Filter any specific instants that we do not want to process . example timeline . > create bucket1 . parquet t1 - > create and append updates bucket1 . log t2 - > request compaction t3 - > create bucket2 . parquet if compaction at t2 takes a long time incremental readers on RO tables can move to t3 and would skip updates in t1 To workaround this problem we want to stop returning data belonging to commits > t2 . After compaction is complete incremental reader would see updates in t2 t3 so on . @param timeline @return
public static HoodieDefaultTimeline filterInstantsTimeline(HoodieDefaultTimeline timeline) {
    HoodieDefaultTimeline commitsAndCompactionTimeline = timeline.getCommitsAndCompactionTimeline();
    Option<HoodieInstant> pendingCompactionInstant = commitsAndCompactionTimeline
        .filterPendingCompactionTimeline().firstInstant();
    if (pendingCompactionInstant.isPresent()) {
      HoodieDefaultTimeline instantsTimeline = commitsAndCompactionTimeline
          .findInstantsBefore(pendingCompactionInstant.get().getTimestamp());
      int numCommitsFilteredByCompaction = commitsAndCompactionTimeline.getCommitsTimeline().countInstants()
          - instantsTimeline.getCommitsTimeline().countInstants();
      LOG.info("Earliest pending compaction instant is: " + pendingCompactionInstant.get().getTimestamp()
          + " skipping " + numCommitsFilteredByCompaction + " commits");
      return instantsTimeline;
    } else {
      return timeline;
    }
  }
public static HoodieDefaultTimeline filterInstantsTimeline ( HoodieDefaultTimeline timeline ) { HoodieDefaultTimeline commitsAndCompactionTimeline = timeline . getCommitsAndCompactionTimeline ( ) ; Option < HoodieInstant > pendingCompactionInstant = commitsAndCompactionTimeline . filterPendingCompactionTimeline ( ) . firstInstant ( ) ; if ( pendingCompactionInstant . isPresent ( ) ) { HoodieDefaultTimeline instantsTimeline = commitsAndCompactionTimeline . findInstantsBefore ( pendingCompactionInstant . get ( ) . getTimestamp ( ) ) ; int numCommitsFilteredByCompaction = commitsAndCompactionTimeline . getCommitsTimeline ( ) . countInstants ( ) - instantsTimeline . getCommitsTimeline ( ) . countInstants ( ) ; LOG . info ( "Earliest pending compaction instant is: " + pendingCompactionInstant . get ( ) . getTimestamp ( ) + " skipping " + numCommitsFilteredByCompaction + " commits" ) ; return instantsTimeline ; } else { return timeline ; } }
Filter any specific instants that we do not want to process.
Filter any specific instants that we do not want to process .