public static Option<Set<String>> getBaseAndLogFilePathsFromTimeline(
      HoodieTimeline timeline, HoodieInstant instant) throws IOException {
    if (!instant.isCompleted()) {
      throw new HoodieException("Cannot get base and log file paths from "
          + "instant not completed: " + instant.getTimestamp());
    }

    switch (instant.getAction()) {
      case COMMIT_ACTION:
      case DELTA_COMMIT_ACTION:
        final HoodieCommitMetadata commitMetadata =
            HoodieCommitMetadata.fromBytes(
                timeline.getInstantDetails(instant).get(), HoodieCommitMetadata.class);
        return Option.of(commitMetadata.getPartitionToWriteStats().values().stream().flatMap(List::stream)
            .map(HoodieWriteStat::getPath).collect(Collectors.toSet()));
      case REPLACE_COMMIT_ACTION:
      case CLUSTERING_ACTION:
        final HoodieReplaceCommitMetadata replaceCommitMetadata =
            HoodieReplaceCommitMetadata.fromBytes(
                timeline.getInstantDetails(instant).get(), HoodieReplaceCommitMetadata.class);
        return Option.of(replaceCommitMetadata.getPartitionToWriteStats().values().stream().flatMap(List::stream)
            .map(HoodieWriteStat::getPath).collect(Collectors.toSet()));
      default:
        return Option.empty();
    }
  }
@Test
  public void testGetBaseAndLogFilePathsFromTimeline() throws IOException {
    setupTimelineInFS();
    HoodieTimeline timeline = metaClient.getActiveTimeline();
    HoodieInstant commitInstant = new HoodieInstant(
        HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "001");
    HoodieInstant inflightInstant = new HoodieInstant(
        HoodieInstant.State.INFLIGHT, HoodieTimeline.COMMIT_ACTION, "005");
    HoodieInstant compactionInstant = new HoodieInstant(
        HoodieInstant.State.COMPLETED, HoodieTimeline.COMPACTION_ACTION, "006");

    Map<String, List<Pair<String, String>>> partitionToFileIdAndNameMap =
        instantInfoMap.get(commitInstant.getTimestamp());
    Set<String> expectedPaths = partitionToFileIdAndNameMap.entrySet().stream()
        .flatMap(entry ->
            entry.getValue().stream()
                .map(fileInfo -> new Path(entry.getKey(), fileInfo.getValue()).toString())
                .collect(Collectors.toList())
                .stream()
        ).collect(Collectors.toSet());
    assertEquals(Option.of(expectedPaths),
        RepairUtils.getBaseAndLogFilePathsFromTimeline(timeline, commitInstant));
    assertThrows(HoodieException.class,
        () -> RepairUtils.getBaseAndLogFilePathsFromTimeline(timeline, inflightInstant));
    assertEquals(Option.empty(),
        RepairUtils.getBaseAndLogFilePathsFromTimeline(timeline, compactionInstant));
  }