[HUDI-5253] HoodieMergeOnReadTableInputFormat could have duplicate records issue if it contains delta files while still splittable

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnMergeOnReadStorage.java

@@ -168,7 +168,6 @@ public void testLogCompactionOnMORTable() throws Exception {
     client.compact(compactionTimeStamp.get());
 
     prevCommitTime = compactionTimeStamp.get();
-    //TODO: Below commits are creating duplicates when all the tests are run together. but individually they are passing.
     for (int i = 0; i < 2; i++) {
       // Upsert
       newCommitTime = HoodieActiveTimeline.createNewInstantTime();

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

@@ -89,7 +89,7 @@ public boolean getBelongsToIncrementalQuery() {
   }
 
   public boolean isSplitable() {
-    return !toString().contains(".log") && !includeBootstrapFilePath();
+    return !toString().contains(".log") && deltaLogFiles.isEmpty() && !includeBootstrapFilePath();
   }
 
   public PathWithBootstrapFileStatus getPathWithBootstrapFileStatus() {

hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieMergeOnReadTableInputFormat.java

@@ -19,6 +19,7 @@
 
 package org.apache.hudi.hadoop.realtime;
 
+import org.apache.hudi.common.model.HoodieLogFile;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.hadoop.PathWithBootstrapFileStatus;
 
@@ -65,4 +66,16 @@ void pathNotSplitableForBootstrapScenario() throws IOException {
     rtPath.setPathWithBootstrapFileStatus(path);
     assertFalse(new HoodieMergeOnReadTableInputFormat().isSplitable(fs, rtPath), "Path for bootstrap should not be splitable.");
   }
+
+  @Test
+  void pathNotSplitableIfContainsDeltaFiles() throws IOException {
+    URI basePath = Files.createTempFile(tempDir, "target", ".parquet").toUri();
+    HoodieRealtimePath rtPath = new HoodieRealtimePath(new Path("foo"), "bar", basePath.toString(), Collections.emptyList(), "000", false, Option.empty());
+    assertTrue(new HoodieMergeOnReadTableInputFormat().isSplitable(fs, rtPath), "Path only contains the base file should be splittable");
+
+    URI logPath = Files.createTempFile(tempDir, ".test", ".log.4_1-149-180").toUri();
+    HoodieLogFile logFile = new HoodieLogFile(fs.getFileStatus(new Path(logPath)));
+    rtPath = new HoodieRealtimePath(new Path("foo"), "bar", basePath.toString(), Collections.singletonList(logFile), "000", false, Option.empty());
+    assertFalse(new HoodieMergeOnReadTableInputFormat().isSplitable(fs, rtPath), "Path contains log files should not be splittable.");
+  }
 }