diff --git a/presto-hive-common/src/main/java/org/apache/hadoop/fs/HadoopExtendedFileSystem.java b/presto-hive-common/src/main/java/org/apache/hadoop/fs/HadoopExtendedFileSystem.java
index 3bb417e55dd29..b44a237381682 100644
--- a/presto-hive-common/src/main/java/org/apache/hadoop/fs/HadoopExtendedFileSystem.java
+++ b/presto-hive-common/src/main/java/org/apache/hadoop/fs/HadoopExtendedFileSystem.java
@@ -65,6 +65,12 @@ public FileSystem getRawFileSystem()
         return fs;
     }
 
+    @Override
+    public String getScheme()
+    {
+        return fs.getScheme();
+    }
+
     @Override
     public URI getUri()
     {
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/util/HudiRealtimeSplitConverter.java b/presto-hive/src/main/java/com/facebook/presto/hive/util/HudiRealtimeSplitConverter.java
index 5868ae1a9fbfd..a5b60ef8e2a50 100644
--- a/presto-hive/src/main/java/com/facebook/presto/hive/util/HudiRealtimeSplitConverter.java
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/util/HudiRealtimeSplitConverter.java
@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.hive.util;
 
+import com.google.common.base.Splitter;
 import com.google.common.collect.ImmutableMap;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapred.FileSplit;
@@ -21,7 +22,6 @@
 import org.apache.hudi.hadoop.realtime.HoodieRealtimeFileSplit;
 
 import java.io.IOException;
-import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -37,6 +37,8 @@
 public class HudiRealtimeSplitConverter
         implements CustomSplitConverter
 {
+    private static final Splitter SPLITTER = Splitter.on(",").omitEmptyStrings();
+
     public static final String HUDI_DELTA_FILEPATHS_KEY = "hudi_delta_filepaths";
     public static final String HUDI_BASEPATH_KEY = "hudi_basepath";
     public static final String HUDI_MAX_COMMIT_TIME_KEY = "hudi_max_commit_time";
@@ -64,7 +66,7 @@ public Optional<FileSplit> recreateFileSplitWithCustomInfo(FileSplit split, Map<
         String customSplitClass = customSplitInfo.get(CUSTOM_FILE_SPLIT_CLASS_KEY);
         if (HoodieRealtimeFileSplit.class.getName().equals(customSplitClass)) {
             requireNonNull(customSplitInfo.get(HUDI_DELTA_FILEPATHS_KEY), "HUDI_DELTA_FILEPATHS_KEY is missing");
-            List<String> deltaLogPaths = Arrays.asList(customSplitInfo.get(HUDI_DELTA_FILEPATHS_KEY).split(","));
+            List<String> deltaLogPaths = SPLITTER.splitToList(customSplitInfo.get(HUDI_DELTA_FILEPATHS_KEY));
             List<HoodieLogFile> deltaLogFiles = deltaLogPaths.stream().map(p -> new HoodieLogFile(new Path(p))).collect(Collectors.toList());
             return Optional.of(new HoodieRealtimeFileSplit(
                     split,
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/util/TestCustomSplitConversionUtils.java b/presto-hive/src/test/java/com/facebook/presto/hive/util/TestCustomSplitConversionUtils.java
index fb65585e50b96..5b27c75390aca 100644
--- a/presto-hive/src/test/java/com/facebook/presto/hive/util/TestCustomSplitConversionUtils.java
+++ b/presto-hive/src/test/java/com/facebook/presto/hive/util/TestCustomSplitConversionUtils.java
@@ -13,6 +13,7 @@
  */
 package com.facebook.presto.hive.util;
 
+import com.google.common.collect.ImmutableList;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapred.FileSplit;
 import org.apache.hudi.common.model.HoodieLogFile;
@@ -64,6 +65,32 @@ public void testHudiRealtimeSplitConverterRoundTrip()
         assertEquals(expectedMaxCommitTime, recreatedSplit.getMaxCommitTime());
     }
 
+    @Test
+    public void testHudiRealtimeSplitConverterNoLogRoundTrip()
+            throws IOException
+    {
+        List<String> deltaLogPaths = ImmutableList.of();
+        List<HoodieLogFile> deltaLogFiles = ImmutableList.of();
+        String expectedMaxCommitTime = "max_commit_time";
+
+        FileSplit baseSplit = new FileSplit(FILE_PATH, SPLIT_START_POS, SPLIT_LENGTH, SPLIT_HOSTS);
+        FileSplit hudiSplit = new HoodieRealtimeFileSplit(baseSplit, BASE_PATH, deltaLogFiles, expectedMaxCommitTime, Option.empty());
+
+        // Test conversion of HudiSplit -> customSplitInfo
+        Map<String, String> customSplitInfo = CustomSplitConversionUtils.extractCustomSplitInfo(hudiSplit);
+
+        // Test conversion of (customSplitInfo + baseSplit) -> HudiSplit
+        HoodieRealtimeFileSplit recreatedSplit = (HoodieRealtimeFileSplit) CustomSplitConversionUtils.recreateSplitWithCustomInfo(baseSplit, customSplitInfo);
+
+        assertEquals(FILE_PATH, recreatedSplit.getPath());
+        assertEquals(SPLIT_START_POS, recreatedSplit.getStart());
+        assertEquals(SPLIT_LENGTH, recreatedSplit.getLength());
+        assertEquals(SPLIT_HOSTS, recreatedSplit.getLocations());
+        assertEquals(BASE_PATH, recreatedSplit.getBasePath());
+        assertEquals(deltaLogPaths, recreatedSplit.getDeltaLogPaths());
+        assertEquals(expectedMaxCommitTime, recreatedSplit.getMaxCommitTime());
+    }
+
     @Test
     public void testHudiBootstrapBaseFileSplitConverter()
             throws IOException