[HUDI-9482] Fix the issue where the file slice was wrongly filtered out

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkInsertOverwriteCommitActionExecutor.java

@@ -96,8 +96,8 @@ protected Map<String, List<String>> getPartitionToReplacedFileIds(HoodieWriteMet
   }
 
   protected List<String> getAllExistingFileIds(String partitionPath) {
-    // because new commit is not complete. it is safe to mark all existing file Ids as old files
-    return table.getSliceView().getLatestFileSlices(partitionPath).map(FileSlice::getFileId).distinct().collect(Collectors.toList());
+    // we should only fetch the latest merged file slices with committed data
+    return table.getSliceView().getLatestMergedFileSlicesBeforeOrOn(partitionPath, instantTime).map(FileSlice::getFileId).distinct().collect(Collectors.toList());
   }
 
   @Override

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieFileGroup.java

@@ -18,7 +18,9 @@
 
 package org.apache.hudi.common.model;
 
+import org.apache.hudi.common.table.HoodieTableVersion;
 import org.apache.hudi.common.table.timeline.CompletionTimeQueryView;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.hudi.common.util.Option;
@@ -119,6 +121,10 @@ public void addLogFile(CompletionTimeQueryView completionTimeQueryView, HoodieLo
   @VisibleForTesting
   public String getBaseInstantTime(CompletionTimeQueryView completionTimeQueryView, HoodieLogFile logFile) {
     if (fileSlices.isEmpty()) {
+      if (completionTimeQueryView.getTableVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT) && !completionTimeQueryView.isCompleted(logFile.getDeltaCommitTime())) {
+        // no base file in the file group and the smallest log file is still pending, we should use the INIT_INSTANT_TS for these uncertain cases.
+        return HoodieActiveTimeline.INIT_INSTANT_TS;
+      }
       // no base file in the file group, use the log file delta commit time.
       return logFile.getDeltaCommitTime();
     }
@@ -130,7 +136,6 @@ public String getBaseInstantTime(CompletionTimeQueryView completionTimeQueryView
           return commitTime;
         }
       }
-      // no base file that starts earlier than the log delta commit completion time,
       // use the log file delta commit time.
       return logFile.getDeltaCommitTime();
     }

hudi-common/src/main/java/org/apache/hudi/common/table/timeline/CompletionTimeQueryView.java

@@ -18,30 +18,39 @@
 
 package org.apache.hudi.common.table.timeline;
 
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
 import org.apache.hudi.common.table.log.InstantRange;
 import org.apache.hudi.common.util.Option;
 
+import java.io.Serializable;
 import java.util.List;
 import java.util.function.Function;
 
-public interface CompletionTimeQueryView extends AutoCloseable {
+public abstract class CompletionTimeQueryView implements AutoCloseable, Serializable {
 
-  boolean isCompleted(String beginInstantTime);
+  protected HoodieTableMetaClient metaClient;
+
+  public CompletionTimeQueryView(HoodieTableMetaClient metaClient) {
+    this.metaClient = metaClient;
+  }
+
+  public abstract boolean isCompleted(String beginInstantTime);
 
   /**
    * Returns whether the instant is archived.
    */
-  boolean isArchived(String instantTime);
+  public abstract boolean isArchived(String instantTime);
 
   /**
    * Returns whether the give instant time {@code instantTime} completed before the base instant {@code baseInstant}.
    */
-  boolean isCompletedBefore(String baseInstant, String instantTime);
+  public abstract boolean isCompletedBefore(String baseInstant, String instantTime);
 
   /**
    * Returns whether the given instant time {@code instantTime} is sliced after or on the base instant {@code baseInstant}.
    */
-  boolean isSlicedAfterOrOn(String baseInstant, String instantTime);
+  public abstract boolean isSlicedAfterOrOn(String baseInstant, String instantTime);
 
   /**
    * Get completion time with a base instant time as a reference to fix the compatibility.
@@ -51,7 +60,7 @@ public interface CompletionTimeQueryView extends AutoCloseable {
    *
    * @return Probability fixed completion time.
    */
-  Option<String> getCompletionTime(String baseInstant, String instantTime);
+  public abstract Option<String> getCompletionTime(String baseInstant, String instantTime);
 
   /**
    * Queries the completion time with given instant time.
@@ -60,7 +69,7 @@ public interface CompletionTimeQueryView extends AutoCloseable {
    *
    * @return The completion time if the instant finished or empty if it is still pending.
    */
-  Option<String> getCompletionTime(String beginTime);
+  public abstract Option<String> getCompletionTime(String beginTime);
 
   /**
    * Queries the instant times with given completion time range.
@@ -74,7 +83,7 @@ public interface CompletionTimeQueryView extends AutoCloseable {
    *
    * @return The sorted instant time list.
    */
-  List<String> getInstantTimes(
+  public abstract List<String> getInstantTimes(
       HoodieTimeline timeline,
       Option<String> startCompletionTime,
       Option<String> endCompletionTime,
@@ -90,7 +99,7 @@ List<String> getInstantTimes(
    *
    * @return The sorted instant time list.
    */
-  List<String> getInstantTimes(
+  public abstract List<String> getInstantTimes(
       String startCompletionTime,
       String endCompletionTime,
       Function<String, String> earliestInstantTimeFunc);
@@ -99,11 +108,15 @@ List<String> getInstantTimes(
    *  Get Cursor Instant
    * @return
    */
-  String getCursorInstant();
+  public abstract String getCursorInstant();
 
   /**
    * Return true if the table is empty.
    * @return
    */
-  boolean isEmptyTable();
+  public abstract boolean isEmptyTable();
+
+  public HoodieTableVersion getTableVersion() {
+    return metaClient.getTableConfig().getTableVersion();
+  }
 }

hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CompletionTimeQueryViewV1.java

@@ -28,7 +28,6 @@
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.VisibleForTesting;
 
-import java.io.Serializable;
 import java.time.Instant;
 import java.util.Date;
 import java.util.List;
@@ -39,15 +38,13 @@
 import static org.apache.hudi.common.table.timeline.InstantComparison.GREATER_THAN_OR_EQUALS;
 import static org.apache.hudi.common.table.timeline.InstantComparison.LESSER_THAN;
 
-public class CompletionTimeQueryViewV1 implements CompletionTimeQueryView, Serializable {
+public class CompletionTimeQueryViewV1 extends CompletionTimeQueryView {
   private static final long serialVersionUID = 1L;
 
   private static final long MILLI_SECONDS_IN_THREE_DAYS = 3 * 24 * 3600 * 1000;
 
   private static final long MILLI_SECONDS_IN_ONE_DAY = 24 * 3600 * 1000;
 
-  private final HoodieTableMetaClient metaClient;
-
   /**
    * Mapping from instant start time -> completion time.
    * Should be thread-safe data structure.
@@ -83,7 +80,7 @@ public CompletionTimeQueryViewV1(HoodieTableMetaClient metaClient) {
    * @param eagerLoadInstant The earliest instant time to eagerly load from, by default load last N days of completed instants.
    */
   public CompletionTimeQueryViewV1(HoodieTableMetaClient metaClient, String eagerLoadInstant) {
-    this.metaClient = metaClient;
+    super(metaClient);
     this.beginToCompletionInstantTimeMap = new ConcurrentHashMap<>();
     this.cursorInstant = InstantComparison.minInstant(eagerLoadInstant, metaClient.getActiveTimeline().firstInstant().map(HoodieInstant::requestedTime).orElse(""));
     // Note: use getWriteTimeline() to keep sync with the fs view visibleCommitsAndCompactionTimeline, see AbstractTableFileSystemView.refreshTimeline.

hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/CompletionTimeQueryViewV2.java

@@ -31,7 +31,6 @@
 
 import org.apache.avro.generic.GenericRecord;
 
-import java.io.Serializable;
 import java.time.Instant;
 import java.util.Collections;
 import java.util.Comparator;
@@ -53,7 +52,7 @@
 /**
  * Query view for instant completion time.
  */
-public class CompletionTimeQueryViewV2 implements CompletionTimeQueryView, Serializable {
+public class CompletionTimeQueryViewV2 extends CompletionTimeQueryView {
   private static final long serialVersionUID = 1L;
 
   private static final long MILLI_SECONDS_IN_THREE_DAYS = 3 * 24 * 3600 * 1000;
@@ -63,8 +62,6 @@ public class CompletionTimeQueryViewV2 implements CompletionTimeQueryView, Seria
   private static final Function<String, String> GET_INSTANT_ONE_DAY_BEFORE = instant ->
       HoodieInstantTimeGenerator.instantTimeMinusMillis(instant, MILLI_SECONDS_IN_ONE_DAY);
 
-  private final HoodieTableMetaClient metaClient;
-
   /**
    * Mapping from instant time -> completion time.
    * Should be thread-safe data structure.
@@ -100,7 +97,7 @@ public CompletionTimeQueryViewV2(HoodieTableMetaClient metaClient) {
    * @param eagerLoadInstant The earliest instant time to eagerly load from, by default load last N days of completed instants.
    */
   public CompletionTimeQueryViewV2(HoodieTableMetaClient metaClient, String eagerLoadInstant) {
-    this.metaClient = metaClient;
+    super(metaClient);
     this.instantTimeToCompletionTimeMap = new ConcurrentHashMap<>();
     this.cursorInstant = InstantComparison.minInstant(eagerLoadInstant, metaClient.getActiveTimeline().firstInstant().map(HoodieInstant::requestedTime).orElse(""));
     // Note: use getWriteTimeline() to keep sync with the fs view visibleCommitsAndCompactionTimeline, see AbstractTableFileSystemView.refreshTimeline.

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

@@ -590,6 +590,27 @@ private Stream<FileSlice> filterUncommittedFiles(FileSlice fileSlice, boolean in
     return Stream.of(fileSlice);
   }
 
+  /**
+   * Ignores the empty file-slice.
+   * When a slice is full of pending log files, we need to filter out this situation because in fact, the file group represented by this file slice has not committed any data either
+   * @param fileSlice File Slice
+   * @return Stream of FileSlice
+   */
+  private Stream<FileSlice> filterOutEmptyFileSlice(FileSlice fileSlice) {
+    // Check if the file slice is empty
+    //    a. the base instant is a pending compaction, still return the file slice
+    //    b. the base instant is a normal write, should filter out the file slice, because the entire file group is empty
+    if (fileSlice.isEmpty()) {
+      if (isFileSliceAfterPendingCompaction(fileSlice)) {
+        LOG.debug("File Slice ({}) is in pending compaction", fileSlice);
+      } else {
+        LOG.debug("File Slice ({}) is empty", fileSlice);
+        return Stream.of();
+      }
+    }
+    return Stream.of(fileSlice);
+  }
+
   /**
    * Ignores the uncommitted log files.
    *
@@ -867,6 +888,10 @@ public final Stream<FileSlice> getLatestFileSlices(String partitionStr) {
           .flatMap(slice -> tableVersion8AndAbove()
               ? this.filterUncommittedFiles(slice, true)
               : this.filterBaseFileAfterPendingCompaction(slice, true))
+          .flatMap(slice -> tableVersion8AndAbove()
+              // for table version 8 and above, we need to filter out empty file slices
+              ? this.filterOutEmptyFileSlice(slice)
+              : Stream.of(slice))
           .map(this::addBootstrapBaseFileIfPresent);
     } finally {
       readLock.unlock();
@@ -899,7 +924,11 @@ public final Stream<FileSlice> getLatestFileSlicesStateless(String partitionStr)
             .filter(Option::isPresent).map(Option::get)
             .flatMap(slice -> tableVersion8AndAbove()
                 ? this.filterUncommittedFiles(slice, true)
-                : this.filterBaseFileAfterPendingCompaction(slice, true));
+                : this.filterBaseFileAfterPendingCompaction(slice, true))
+            .flatMap(slice -> tableVersion8AndAbove()
+                // for table version 8 and above, we need to filter out empty file slices
+                ? this.filterOutEmptyFileSlice(slice)
+                : Stream.of(slice));
 
         if (bootstrapIndex.useIndex()) {
           final Map<HoodieFileGroupId, BootstrapBaseFileMapping> bootstrapBaseFileMappings = getBootstrapBaseFileMappings(partition);
@@ -930,9 +959,13 @@ public final Option<FileSlice> getLatestFileSlice(String partitionStr, String fi
         if (!fs.isPresent()) {
           return Option.empty();
         }
-        Stream<FileSlice> fileSlices = tableVersion8AndAbove()
+        Stream<FileSlice> fileSlices = (tableVersion8AndAbove()
             ? this.filterUncommittedFiles(fs.get(), true)
-            : this.filterBaseFileAfterPendingCompaction(fs.get(), true);
+            : this.filterBaseFileAfterPendingCompaction(fs.get(), true))
+            .flatMap(slice -> tableVersion8AndAbove()
+            // for table version 8 and above, we need to filter out empty file slices
+            ? this.filterOutEmptyFileSlice(slice)
+            : Stream.of(slice));
 
         return Option.ofNullable(fileSlices
             .map(this::addBootstrapBaseFileIfPresent)
@@ -1051,6 +1084,9 @@ public final Stream<FileSlice> getLatestMergedFileSlicesBeforeOrOn(String partit
               fileSlice = Option.of(fetchMergedFileSlice(fileGroup, tableVersion8AndAbove()
                       ? filterUncommittedLogs(fileSlice.get()) : fileSlice.get())
               );
+              if (fileSlice.isPresent() && tableVersion8AndAbove()) {
+                fileSlice = Option.fromJavaOptional(filterOutEmptyFileSlice(fileSlice.get()).findAny());
+              }
             }
             return fileSlice;
           }).filter(Option::isPresent).map(Option::get).map(this::addBootstrapBaseFileIfPresent);

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

@@ -27,6 +27,7 @@
 import org.apache.hudi.common.model.HoodieFileGroupId;
 import org.apache.hudi.common.model.HoodieLogFile;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.hudi.common.util.Option;
@@ -588,7 +589,10 @@ private static boolean shouldMergeFileSlice(FileSlice oldSlice, FileSlice newSli
     // 2. the old file slice has no compaction scheduled, while the new file slice does,
     //    if they have different base instant time, should not merge;
     // 3. the new file slice has compaction been scheduled, should not merge if old and new base instant time is different.
-    return isFileSliceWithoutCompactionBarrier(newSlice) || newSlice.getBaseInstantTime().equals(oldSlice.getBaseInstantTime());
+    // 4. the new file slice base log file still in pending, should merge.
+    return isFileSliceWithoutCompactionBarrier(newSlice)
+        || newSlice.getBaseInstantTime().equals(oldSlice.getBaseInstantTime())
+        || newSlice.getBaseInstantTime().equals(HoodieActiveTimeline.INIT_INSTANT_TS);
   }
 
   private static boolean isFileSliceWithoutCompactionBarrier(FileSlice fileSlice) {