[HUDI-4508] Repair the exception when reading optimized query for mor in hive and presto/trino, if the fileSlice with log file has not base file.

Author: 5herhom

hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieCopyOnWriteTableInputFormat.java

@@ -35,6 +35,7 @@
 import org.apache.hudi.common.engine.HoodieLocalEngineContext;
 import org.apache.hudi.common.model.FileSlice;
 import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieLogFile;
 import org.apache.hudi.common.model.HoodieTableQueryType;
 import org.apache.hudi.common.table.HoodieTableConfig;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -253,6 +254,7 @@ private List<FileStatus> listStatusForSnapshotMode(JobConf job,
           partitionedFileSlices.values()
               .stream()
               .flatMap(Collection::stream)
+              .filter(fileSlice -> checkIfValidFileSlice(fileSlice))
               .map(fileSlice -> createFileStatusUnchecked(fileSlice, fileIndex, virtualKeyInfoOpt))
               .collect(Collectors.toList())
       );
@@ -261,6 +263,20 @@ private List<FileStatus> listStatusForSnapshotMode(JobConf job,
     return targetFiles;
   }
 
+  private boolean checkIfValidFileSlice(FileSlice fileSlice) {
+    Option<HoodieBaseFile> baseFileOpt = fileSlice.getBaseFile();
+    Option<HoodieLogFile> latestLogFileOpt = fileSlice.getLatestLogFile();
+
+    if (baseFileOpt.isPresent()) {
+      return true;
+    } else if (latestLogFileOpt.isPresent()) {
+      // It happens when reading optimized query to mor.
+      return false;
+    } else {
+      throw new IllegalStateException("Invalid state: base-file has to be present");
+    }
+  }
+
   private void validate(List<FileStatus> targetFiles, List<FileStatus> legacyFileStatuses) {
     List<FileStatus> diff = CollectionUtils.diff(targetFiles, legacyFileStatuses);
     checkState(diff.isEmpty(), "Should be empty");

hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieMergeOnReadTableInputFormat.java

@@ -103,6 +103,17 @@ protected FileStatus createFileStatusUnchecked(FileSlice fileSlice, HiveHoodieTa
     }
   }
 
+  private boolean checkIfValidFileSlice(FileSlice fileSlice) {
+    Option<HoodieBaseFile> baseFileOpt = fileSlice.getBaseFile();
+    Option<HoodieLogFile> latestLogFileOpt = fileSlice.getLatestLogFile();
+
+    if (baseFileOpt.isPresent() || latestLogFileOpt.isPresent()) {
+      return true;
+    } else {
+      throw new IllegalStateException("Invalid state: either base-file or log-file has to be present");
+    }
+  }
+
   /**
    * Keep the logic of mor_incr_view as same as spark datasource.
    * Step1: Get list of commits to be fetched based on start commit and max commits(for snapshot max commits is -1).