[HUDI-4078][HUDI-FLINK]BootstrapOperator contains the pending compact…

hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java

@@ -644,14 +644,17 @@ public final Stream<FileSlice> getLatestUnCompactedFileSlices(String partitionSt
 
   @Override
   public final Stream<FileSlice> getLatestFileSlicesBeforeOrOn(String partitionStr, String maxCommitTime,
-      boolean includeFileSlicesInPendingCompaction) {
+      boolean includeFileSlicesInPendingCompaction, boolean includeFilesInPendingCompaction) {
     try {
       readLock.lock();
       String partitionPath = formatPartitionKey(partitionStr);
       ensurePartitionLoadedCorrectly(partitionPath);
       Stream<FileSlice> fileSliceStream = fetchLatestFileSlicesBeforeOrOn(partitionPath, maxCommitTime)
           .filter(slice -> !isFileGroupReplacedBeforeOrOn(slice.getFileGroupId(), maxCommitTime));
       if (includeFileSlicesInPendingCompaction) {
+        if (includeFilesInPendingCompaction) {
+          return fileSliceStream.map(this::addBootstrapBaseFileIfPresent);
+        }
         return fileSliceStream.map(this::filterBaseFileAfterPendingCompaction).map(this::addBootstrapBaseFileIfPresent);
       } else {
         return fileSliceStream.filter(fs -> !isPendingCompactionScheduledForFileId(fs.getFileGroupId()))

hudi-common/src/main/java/org/apache/hudi/common/table/view/PriorityBasedFileSystemView.java

@@ -178,7 +178,7 @@ public Stream<FileSlice> getLatestUnCompactedFileSlices(String partitionPath) {
 
   @Override
   public Stream<FileSlice> getLatestFileSlicesBeforeOrOn(String partitionPath, String maxCommitTime,
-      boolean includeFileSlicesInPendingCompaction) {
+      boolean includeFileSlicesInPendingCompaction, boolean includeFilesInPendingCompaction) {
     return execute(partitionPath, maxCommitTime, includeFileSlicesInPendingCompaction,
         preferredView::getLatestFileSlicesBeforeOrOn, secondaryView::getLatestFileSlicesBeforeOrOn);
   }

hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java

@@ -312,7 +312,7 @@ public Stream<FileSlice> getLatestUnCompactedFileSlices(String partitionPath) {
 
   @Override
   public Stream<FileSlice> getLatestFileSlicesBeforeOrOn(String partitionPath, String maxCommitTime,
-      boolean includeFileSlicesInPendingCompaction) {
+      boolean includeFileSlicesInPendingCompaction, boolean includeFilesInPendingCompaction) {
     Map<String, String> paramsMap = getParamsWithAdditionalParams(partitionPath,
         new String[] {MAX_INSTANT_PARAM, INCLUDE_FILES_IN_PENDING_COMPACTION_PARAM},
         new String[] {maxCommitTime, String.valueOf(includeFileSlicesInPendingCompaction)});

hudi-common/src/main/java/org/apache/hudi/common/table/view/TableFileSystemView.java

@@ -106,15 +106,21 @@ interface SliceViewWithLatestSlice {
      */
     Stream<FileSlice> getLatestUnCompactedFileSlices(String partitionPath);
 
+    default Stream<FileSlice> getLatestFileSlicesBeforeOrOn(String partitionPath, String maxCommitTime,
+        boolean includeFileSlicesInPendingCompaction) {
+      return getLatestFileSlicesBeforeOrOn(partitionPath, maxCommitTime, includeFileSlicesInPendingCompaction, false);
+    }
+
     /**
      * Stream all latest file slices in given partition with precondition that commitTime(file) before maxCommitTime.
      *
      * @param partitionPath Partition path
      * @param maxCommitTime Max Instant Time
      * @param includeFileSlicesInPendingCompaction include file-slices that are in pending compaction
+     * @param includeFilesInPendingCompaction include files that are in pending compaction
      */
     Stream<FileSlice> getLatestFileSlicesBeforeOrOn(String partitionPath, String maxCommitTime,
-        boolean includeFileSlicesInPendingCompaction);
+        boolean includeFileSlicesInPendingCompaction, boolean includeFilesInPendingCompaction);
 
     /**
      * Stream all "merged" file-slices before on an instant time If a file-group has a pending compaction request, the
@@ -124,7 +130,7 @@ Stream<FileSlice> getLatestFileSlicesBeforeOrOn(String partitionPath, String max
      * @param maxInstantTime Max Instant Time
      * @return
      */
-    public Stream<FileSlice> getLatestMergedFileSlicesBeforeOrOn(String partitionPath, String maxInstantTime);
+    Stream<FileSlice> getLatestMergedFileSlicesBeforeOrOn(String partitionPath, String maxInstantTime);
 
     /**
      * Stream all the latest file slices, in the given range.

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/BootstrapOperator.java

@@ -198,7 +198,7 @@ protected void loadRecords(String partitionPath) throws Exception {
       Schema schema = new TableSchemaResolver(this.hoodieTable.getMetaClient()).getTableAvroSchema();
 
       List<FileSlice> fileSlices = this.hoodieTable.getSliceView()
-          .getLatestFileSlicesBeforeOrOn(partitionPath, latestCommitTime.get().getTimestamp(), true)
+          .getLatestFileSlicesBeforeOrOn(partitionPath, latestCommitTime.get().getTimestamp(), true, true)
           .collect(toList());
 
       for (FileSlice fileSlice : fileSlices) {