diff --git a/pom.xml b/pom.xml
index f6d52d03c1cda..b4ca11dd36fb8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -69,7 +69,7 @@
         <dep.kafka.version>2.3.1</dep.kafka.version>
         <dep.druid.version>0.19.0</dep.druid.version>
         <dep.jaxb.version>2.3.1</dep.jaxb.version>
-        <dep.hudi.version>0.9.0</dep.hudi.version>
+        <dep.hudi.version>0.10.0-SNAPSHOT</dep.hudi.version>
         <dep.testcontainers.version>1.15.1</dep.testcontainers.version>
         <!--
           America/Bahia_Banderas has:
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/HadoopDirectoryLister.java b/presto-hive/src/main/java/com/facebook/presto/hive/HadoopDirectoryLister.java
index b5f65f28941a7..c6012fc865508 100644
--- a/presto-hive/src/main/java/com/facebook/presto/hive/HadoopDirectoryLister.java
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/HadoopDirectoryLister.java
@@ -45,7 +45,7 @@ public Iterator<HiveFileInfo> list(
                 p -> new HadoopFileInfoIterator(fileSystem.listLocatedStatus(p)),
                 namenodeStats,
                 hiveDirectoryContext.getNestedDirectoryPolicy(),
-                pathFilter);
+                Optional.of(pathFilter));
     }
 
     public static class HadoopFileInfoIterator
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/HiveClientConfig.java b/presto-hive/src/main/java/com/facebook/presto/hive/HiveClientConfig.java
index 0b8d0fa99f566..d182f861857f9 100644
--- a/presto-hive/src/main/java/com/facebook/presto/hive/HiveClientConfig.java
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/HiveClientConfig.java
@@ -197,6 +197,7 @@ public class HiveClientConfig
     private int materializedViewMissingPartitionsThreshold = 100;
 
     private boolean verboseRuntimeStatsEnabled;
+    private boolean hudiPreferMetadataToListFiles;
 
     public int getMaxInitialSplits()
     {
@@ -1682,4 +1683,17 @@ public int getMaterializedViewMissingPartitionsThreshold()
     {
         return this.materializedViewMissingPartitionsThreshold;
     }
+
+    @Config("hive.hudi-prefer-metadata-to-list-files")
+    @ConfigDescription("For Hudi tables prefer to fetch the list of file names and sizes from metadata rather than storage")
+    public HiveClientConfig setHudiPreferMetadataToListFiles(boolean hudiPreferMetadataToListFiles)
+    {
+        this.hudiPreferMetadataToListFiles = hudiPreferMetadataToListFiles;
+        return this;
+    }
+
+    public boolean isHudiPreferMetadataToListFiles()
+    {
+        return this.hudiPreferMetadataToListFiles;
+    }
 }
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/HivePageSourceProvider.java b/presto-hive/src/main/java/com/facebook/presto/hive/HivePageSourceProvider.java
index f257c3ab5eb3b..34d75952e057e 100644
--- a/presto-hive/src/main/java/com/facebook/presto/hive/HivePageSourceProvider.java
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/HivePageSourceProvider.java
@@ -71,6 +71,7 @@
 import static com.facebook.presto.hive.HivePageSourceProvider.ColumnMapping.toColumnHandles;
 import static com.facebook.presto.hive.HiveUtil.getPrefilledColumnValue;
 import static com.facebook.presto.hive.HiveUtil.parsePartitionValue;
+import static com.facebook.presto.hive.HiveUtil.shouldUseRecordReaderFromInputFormat;
 import static com.facebook.presto.hive.metastore.MetastoreUtil.getHiveSchema;
 import static com.facebook.presto.hive.metastore.MetastoreUtil.reconstructPartitionSchema;
 import static com.facebook.presto.spi.relation.ExpressionOptimizer.Level.OPTIMIZED;
@@ -408,42 +409,44 @@ public static Optional<ConnectorPageSource> createHivePageSource(
 
         Optional<BucketAdaptation> bucketAdaptation = bucketConversion.map(conversion -> toBucketAdaptation(conversion, regularAndInterimColumnMappings, tableBucketNumber, ColumnMapping::getIndex));
 
-        for (HiveBatchPageSourceFactory pageSourceFactory : pageSourceFactories) {
-            Optional<? extends ConnectorPageSource> pageSource = pageSourceFactory.createPageSource(
-                    configuration,
-                    session,
-                    path,
-                    start,
-                    length,
-                    fileSize,
-                    storage,
-                    tableName,
-                    tableParameters,
-                    toColumnHandles(regularAndInterimColumnMappings, true),
-                    effectivePredicate,
-                    hiveStorageTimeZone,
-                    hiveFileContext,
-                    encryptionInformation);
-            if (pageSource.isPresent()) {
-                HivePageSource hivePageSource = new HivePageSource(
-                        columnMappings,
-                        bucketAdaptation,
+        if (!shouldUseRecordReaderFromInputFormat(configuration, storage, customSplitInfo)) {
+            for (HiveBatchPageSourceFactory pageSourceFactory : pageSourceFactories) {
+                Optional<? extends ConnectorPageSource> pageSource = pageSourceFactory.createPageSource(
+                        configuration,
+                        session,
+                        path,
+                        start,
+                        length,
+                        fileSize,
+                        storage,
+                        tableName,
+                        tableParameters,
+                        toColumnHandles(regularAndInterimColumnMappings, true),
+                        effectivePredicate,
                         hiveStorageTimeZone,
-                        typeManager,
-                        pageSource.get());
-
-                if (isPushdownFilterEnabled) {
-                    return Optional.of(new FilteringPageSource(
+                        hiveFileContext,
+                        encryptionInformation);
+                if (pageSource.isPresent()) {
+                    HivePageSource hivePageSource = new HivePageSource(
                             columnMappings,
-                            effectivePredicate,
-                            remainingPredicate,
+                            bucketAdaptation,
+                            hiveStorageTimeZone,
                             typeManager,
-                            rowExpressionService,
-                            session,
-                            outputIndices,
-                            hivePageSource));
+                            pageSource.get());
+
+                    if (isPushdownFilterEnabled) {
+                        return Optional.of(new FilteringPageSource(
+                                columnMappings,
+                                effectivePredicate,
+                                remainingPredicate,
+                                typeManager,
+                                rowExpressionService,
+                                session,
+                                outputIndices,
+                                hivePageSource));
+                    }
+                    return Optional.of(hivePageSource);
                 }
-                return Optional.of(hivePageSource);
             }
         }
 
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/HiveSessionProperties.java b/presto-hive/src/main/java/com/facebook/presto/hive/HiveSessionProperties.java
index 9bf4b724d204c..e7a53bfa958eb 100644
--- a/presto-hive/src/main/java/com/facebook/presto/hive/HiveSessionProperties.java
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/HiveSessionProperties.java
@@ -129,6 +129,7 @@ public final class HiveSessionProperties
     public static final String VERBOSE_RUNTIME_STATS_ENABLED = "verbose_runtime_stats_enabled";
     private static final String DWRF_WRITER_STRIPE_CACHE_ENABLED = "dwrf_writer_stripe_cache_enabled";
     private static final String DWRF_WRITER_STRIPE_CACHE_SIZE = "dwrf_writer_stripe_cache_size";
+    private static final String HUDI_PREFER_METADATA_TO_LIST_FILES = "hudi_prefer_metadata_to_list_files";
 
     private final List<PropertyMetadata<?>> sessionProperties;
 
@@ -617,6 +618,11 @@ public HiveSessionProperties(HiveClientConfig hiveClientConfig, OrcFileWriterCon
                         DWRF_WRITER_STRIPE_CACHE_SIZE,
                         "Maximum size of DWRF stripe cache to be held in memory",
                         orcFileWriterConfig.getDwrfStripeCacheMaxSize(),
+                        false),
+                booleanProperty(
+                        HUDI_PREFER_METADATA_TO_LIST_FILES,
+                        "For Hudi tables prefer to fetch the list of files from its metadata",
+                        hiveClientConfig.isHudiPreferMetadataToListFiles(),
                         false));
     }
 
@@ -1077,4 +1083,9 @@ public static DataSize getDwrfWriterStripeCacheeMaxSize(ConnectorSession session
     {
         return session.getProperty(DWRF_WRITER_STRIPE_CACHE_SIZE, DataSize.class);
     }
+
+    public static boolean isHudiPreferMetadataToListFiles(ConnectorSession session)
+    {
+        return session.getProperty(HUDI_PREFER_METADATA_TO_LIST_FILES, Boolean.class);
+    }
 }
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/HiveUtil.java b/presto-hive/src/main/java/com/facebook/presto/hive/HiveUtil.java
index 236566b96d552..92c5acbc8c9c9 100644
--- a/presto-hive/src/main/java/com/facebook/presto/hive/HiveUtil.java
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/HiveUtil.java
@@ -33,7 +33,6 @@
 import com.facebook.presto.hive.metastore.Table;
 import com.facebook.presto.hive.pagefile.PageInputFormat;
 import com.facebook.presto.hive.util.FooterAwareRecordReader;
-import com.facebook.presto.hive.util.HudiRealtimeSplitConverter;
 import com.facebook.presto.orc.metadata.OrcType;
 import com.facebook.presto.spi.ColumnMetadata;
 import com.facebook.presto.spi.PrestoException;
@@ -76,6 +75,8 @@
 import org.apache.hadoop.mapred.Reporter;
 import org.apache.hadoop.mapred.TextInputFormat;
 import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hudi.hadoop.HoodieParquetInputFormat;
+import org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat;
 import org.apache.hudi.hadoop.realtime.HoodieRealtimeFileSplit;
 import org.joda.time.DateTimeZone;
 import org.joda.time.format.DateTimeFormat;
@@ -149,6 +150,7 @@
 import static com.facebook.presto.hive.util.ConfigurationUtils.copy;
 import static com.facebook.presto.hive.util.ConfigurationUtils.toJobConf;
 import static com.facebook.presto.hive.util.CustomSplitConversionUtils.recreateSplitWithCustomInfo;
+import static com.facebook.presto.hive.util.HudiRealtimeSplitConverter.CUSTOM_SPLIT_CLASS_KEY;
 import static com.facebook.presto.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
 import static com.facebook.presto.spi.StandardErrorCode.NOT_SUPPORTED;
 import static com.google.common.base.MoreObjects.firstNonNull;
@@ -196,6 +198,8 @@ public final class HiveUtil
     private static final int DECIMAL_SCALE_GROUP = 2;
 
     private static final String BIG_DECIMAL_POSTFIX = "BD";
+    private static final String USE_RECORD_READER_FROM_INPUT_FORMAT_ANNOTATION = "UseRecordReaderFromInputFormat";
+    private static final String USE_FILE_SPLITS_FROM_INPUT_FORMAT_ANNOTATION = "UseFileSplitsFromInputFormat";
 
     static {
         DateTimeParser[] timestampWithoutTimeZoneParser = {
@@ -295,7 +299,7 @@ private HiveUtil()
 
     private static boolean isHudiRealtimeSplit(Map<String, String> customSplitInfo)
     {
-        String customSplitClass = customSplitInfo.get(HudiRealtimeSplitConverter.CUSTOM_SPLIT_CLASS_KEY);
+        String customSplitClass = customSplitInfo.get(CUSTOM_SPLIT_CLASS_KEY);
         return HoodieRealtimeFileSplit.class.getName().equals(customSplitClass);
     }
 
@@ -366,13 +370,35 @@ static String getInputFormatName(Properties schema)
         return name;
     }
 
-    public static boolean shouldUseRecordReaderFromInputFormat(Configuration configuration, Storage storage)
+    public static boolean shouldUseRecordReaderFromInputFormat(Configuration configuration, Storage storage, Map<String, String> customSplitInfo)
     {
+        if (customSplitInfo == null || !customSplitInfo.containsKey(CUSTOM_SPLIT_CLASS_KEY)) {
+            return false;
+        }
+
         InputFormat<?, ?> inputFormat = HiveUtil.getInputFormat(configuration, storage.getStorageFormat().getInputFormat(), false);
         return Arrays.stream(inputFormat.getClass().getAnnotations())
                 .map(Annotation::annotationType)
                 .map(Class::getSimpleName)
-                .anyMatch(name -> name.equals("UseRecordReaderFromInputFormat"));
+                .anyMatch(USE_RECORD_READER_FROM_INPUT_FORMAT_ANNOTATION::equals);
+    }
+
+    static boolean shouldUseFileSplitsFromInputFormat(InputFormat<?, ?> inputFormat)
+    {
+        boolean hasUseSplitsAnnotation = Arrays.stream(inputFormat.getClass().getAnnotations())
+                .map(Annotation::annotationType)
+                .map(Class::getSimpleName)
+                .anyMatch(USE_FILE_SPLITS_FROM_INPUT_FORMAT_ANNOTATION::equals);
+
+        return hasUseSplitsAnnotation && !isHudiParquetInputFormat(inputFormat);
+    }
+
+    static boolean isHudiParquetInputFormat(InputFormat<?, ?> inputFormat)
+    {
+        if (inputFormat instanceof HoodieParquetRealtimeInputFormat) {
+            return false;
+        }
+        return inputFormat instanceof HoodieParquetInputFormat;
     }
 
     public static long parseHiveDate(String value)
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/HudiDirectoryLister.java b/presto-hive/src/main/java/com/facebook/presto/hive/HudiDirectoryLister.java
new file mode 100644
index 0000000000000..a31f67bf9e5cc
--- /dev/null
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/HudiDirectoryLister.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.hive;
+
+import com.facebook.airlift.log.Logger;
+import com.facebook.presto.hive.filesystem.ExtendedFileSystem;
+import com.facebook.presto.hive.metastore.Table;
+import com.facebook.presto.hive.util.HiveFileIterator;
+import com.facebook.presto.spi.ConnectorSession;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.BlockLocation;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.engine.HoodieLocalEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.view.FileSystemViewManager;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Optional;
+
+import static com.facebook.presto.hive.HiveFileInfo.createHiveFileInfo;
+import static com.facebook.presto.hive.HiveSessionProperties.isHudiPreferMetadataToListFiles;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_DEFAULT_PORT;
+
+public final class HudiDirectoryLister
+        implements DirectoryLister
+{
+    private static final Logger log = Logger.get(HudiDirectoryLister.class);
+
+    private final HoodieTableFileSystemView fileSystemView;
+
+    public HudiDirectoryLister(Configuration conf, ConnectorSession session, Table table)
+    {
+        log.info("Using Hudi Directory Lister.");
+        HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder()
+                .setConf(conf)
+                .setBasePath(table.getStorage().getLocation())
+                .build();
+        HoodieEngineContext engineContext = new HoodieLocalEngineContext(conf);
+        HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder()
+                .enable(isHudiPreferMetadataToListFiles(session))
+                .build();
+        this.fileSystemView = FileSystemViewManager.createInMemoryFileSystemView(engineContext, metaClient,
+                metadataConfig);
+    }
+
+    @Override
+    public Iterator<HiveFileInfo> list(ExtendedFileSystem fileSystem,
+            Table table,
+            Path path,
+            NamenodeStats namenodeStats,
+            PathFilter pathFilter,
+            HiveDirectoryContext hiveDirectoryContext)
+    {
+        log.debug("Listing path using Hudi directory lister: %s", path.toString());
+        return new HiveFileIterator(
+                path,
+                p -> new HudiFileInfoIterator(fileSystemView, fileSystem.listStatus(p), table.getStorage().getLocation(), p),
+                namenodeStats,
+                hiveDirectoryContext.getNestedDirectoryPolicy(),
+                Optional.empty());
+    }
+
+    public static class HudiFileInfoIterator
+            implements RemoteIterator<HiveFileInfo>
+    {
+        private final Iterator<HoodieBaseFile> hoodieBaseFileIterator;
+
+        public HudiFileInfoIterator(
+                HoodieTableFileSystemView fileSystemView,
+                FileStatus[] fileStatuses,
+                String tablePath,
+                Path directory)
+        {
+            fileSystemView.addFilesToView(fileStatuses);
+            String partition = FSUtils.getRelativePartitionPath(new Path(tablePath), directory);
+            this.hoodieBaseFileIterator = fileSystemView.fetchLatestBaseFiles(partition).iterator();
+        }
+
+        @Override
+        public boolean hasNext()
+        {
+            return hoodieBaseFileIterator.hasNext();
+        }
+
+        @Override
+        public HiveFileInfo next()
+                throws IOException
+        {
+            FileStatus fileStatus = hoodieBaseFileIterator.next().getFileStatus();
+            String[] name = new String[] {"localhost:" + DFS_DATANODE_DEFAULT_PORT};
+            String[] host = new String[] {"localhost"};
+            LocatedFileStatus hoodieFileStatus = new LocatedFileStatus(fileStatus,
+                    new BlockLocation[] {new BlockLocation(name, host, 0L, fileStatus.getLen())});
+            return createHiveFileInfo(hoodieFileStatus, Optional.empty());
+        }
+    }
+}
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/StoragePartitionLoader.java b/presto-hive/src/main/java/com/facebook/presto/hive/StoragePartitionLoader.java
index 9d698b899b878..c1ef12b6dcd79 100644
--- a/presto-hive/src/main/java/com/facebook/presto/hive/StoragePartitionLoader.java
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/StoragePartitionLoader.java
@@ -23,10 +23,7 @@
 import com.facebook.presto.spi.ConnectorSession;
 import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.SchemaTableName;
-import com.google.common.base.Function;
-import com.google.common.cache.CacheBuilder;
-import com.google.common.cache.CacheLoader;
-import com.google.common.cache.LoadingCache;
+import com.google.common.base.Strings;
 import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterators;
@@ -44,17 +41,12 @@
 import org.apache.hadoop.mapred.InputSplit;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.TextInputFormat;
-import org.apache.hudi.hadoop.HoodieParquetInputFormat;
-import org.apache.hudi.hadoop.HoodieROTablePathFilter;
-import org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat;
 
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
-import java.lang.annotation.Annotation;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Comparator;
 import java.util.Deque;
 import java.util.Iterator;
@@ -76,6 +68,8 @@
 import static com.facebook.presto.hive.HiveUtil.getFooterCount;
 import static com.facebook.presto.hive.HiveUtil.getHeaderCount;
 import static com.facebook.presto.hive.HiveUtil.getInputFormat;
+import static com.facebook.presto.hive.HiveUtil.isHudiParquetInputFormat;
+import static com.facebook.presto.hive.HiveUtil.shouldUseFileSplitsFromInputFormat;
 import static com.facebook.presto.hive.HiveWriterFactory.getBucketNumber;
 import static com.facebook.presto.hive.NestedDirectoryPolicy.FAIL;
 import static com.facebook.presto.hive.NestedDirectoryPolicy.IGNORED;
@@ -112,7 +106,6 @@ public class StoragePartitionLoader
     private final ConnectorSession session;
     private final Deque<Iterator<InternalHiveSplit>> fileIterators;
     private final boolean schedulerUsesHostAddresses;
-    private final LoadingCache<Configuration, HoodieROTablePathFilter> hoodiePathFilterLoadingCache;
     private final boolean partialAggregationsPushedDown;
 
     public StoragePartitionLoader(
@@ -134,15 +127,22 @@ public StoragePartitionLoader(
         this.session = requireNonNull(session, "session is null");
         this.hdfsEnvironment = requireNonNull(hdfsEnvironment, "hdfsEnvironment is null");
         this.namenodeStats = requireNonNull(namenodeStats, "namenodeStats is null");
-        this.directoryLister = requireNonNull(directoryLister, "directoryLister is null");
         this.recursiveDirWalkerEnabled = recursiveDirWalkerEnabled;
         this.hdfsContext = new HdfsContext(session, table.getDatabaseName(), table.getTableName(), table.getStorage().getLocation(), false);
         this.fileIterators = requireNonNull(fileIterators, "fileIterators is null");
         this.schedulerUsesHostAddresses = schedulerUsesHostAddresses;
-        this.hoodiePathFilterLoadingCache = CacheBuilder.newBuilder()
-                .maximumSize(1000)
-                .build(CacheLoader.from((Function<Configuration, HoodieROTablePathFilter>) HoodieROTablePathFilter::new));
         this.partialAggregationsPushedDown = partialAggregationsPushedDown;
+
+        DirectoryLister directoryListerOverride = null;
+        if (!Strings.isNullOrEmpty(table.getStorage().getLocation())) {
+            Configuration configuration = hdfsEnvironment.getConfiguration(hdfsContext, new Path(table.getStorage().getLocation()));
+            InputFormat<?, ?> inputFormat = getInputFormat(configuration, table.getStorage().getStorageFormat().getInputFormat(), false);
+            if (isHudiParquetInputFormat(inputFormat)) {
+                directoryListerOverride = new HudiDirectoryLister(configuration, session, table);
+            }
+        }
+        this.directoryLister = directoryListerOverride != null ? directoryListerOverride :
+                requireNonNull(directoryLister, "directoryLister is null");
     }
 
     @Override
@@ -249,7 +249,7 @@ public ListenableFuture<?> loadPartition(HivePartitionMetadata partition, HiveSp
                 schedulerUsesHostAddresses,
                 partition.getEncryptionInformation());
 
-        if (!isHudiParquetInputFormat(inputFormat) && shouldUseFileSplitsFromInputFormat(inputFormat)) {
+        if (shouldUseFileSplitsFromInputFormat(inputFormat)) {
             if (tableBucketInfo.isPresent()) {
                 throw new PrestoException(NOT_SUPPORTED, "Presto cannot read bucketed partition in an input format with UseFileSplitsFromInputFormat annotation: " + inputFormat.getClass().getSimpleName());
             }
@@ -261,7 +261,7 @@ public ListenableFuture<?> loadPartition(HivePartitionMetadata partition, HiveSp
 
             return addSplitsToSource(splits, splitFactory, hiveSplitSource, stopped);
         }
-        PathFilter pathFilter = isHudiParquetInputFormat(inputFormat) ? hoodiePathFilterLoadingCache.getUnchecked(configuration) : path1 -> true;
+        PathFilter pathFilter = path1 -> true;
         // S3 Select pushdown works at the granularity of individual S3 objects,
         // Partial aggregation pushdown works at the granularity of individual files
         // therefore we must not split files when either is enabled.
@@ -301,22 +301,6 @@ private ListenableFuture<?> addSplitsToSource(InputSplit[] targetSplits, Interna
         return lastResult;
     }
 
-    private static boolean isHudiParquetInputFormat(InputFormat<?, ?> inputFormat)
-    {
-        if (inputFormat instanceof HoodieParquetRealtimeInputFormat) {
-            return false;
-        }
-        return inputFormat instanceof HoodieParquetInputFormat;
-    }
-
-    private static boolean shouldUseFileSplitsFromInputFormat(InputFormat<?, ?> inputFormat)
-    {
-        return Arrays.stream(inputFormat.getClass().getAnnotations())
-                .map(Annotation::annotationType)
-                .map(Class::getSimpleName)
-                .anyMatch(name -> name.equals("UseFileSplitsFromInputFormat"));
-    }
-
     private Iterator<InternalHiveSplit> createInternalHiveSplitIterator(Path path, ExtendedFileSystem fileSystem, InternalHiveSplitFactory splitFactory, boolean splittable, PathFilter pathFilter, Optional<Partition> partition)
     {
         boolean cacheable = isUseListDirectoryCache(session);
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/parquet/ParquetPageSourceFactory.java b/presto-hive/src/main/java/com/facebook/presto/hive/parquet/ParquetPageSourceFactory.java
index 165eee54e62f5..505d4b2ca8553 100644
--- a/presto-hive/src/main/java/com/facebook/presto/hive/parquet/ParquetPageSourceFactory.java
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/parquet/ParquetPageSourceFactory.java
@@ -99,7 +99,6 @@
 import static com.facebook.presto.hive.HiveSessionProperties.isParquetBatchReaderVerificationEnabled;
 import static com.facebook.presto.hive.HiveSessionProperties.isParquetBatchReadsEnabled;
 import static com.facebook.presto.hive.HiveSessionProperties.isUseParquetColumnNames;
-import static com.facebook.presto.hive.HiveUtil.shouldUseRecordReaderFromInputFormat;
 import static com.facebook.presto.hive.parquet.HdfsParquetDataSource.buildHdfsParquetDataSource;
 import static com.facebook.presto.memory.context.AggregatedMemoryContext.newSimpleAggregatedMemoryContext;
 import static com.facebook.presto.parquet.ParquetTypeUtils.columnPathFromSubfield;
@@ -165,7 +164,7 @@ public Optional<? extends ConnectorPageSource> createPageSource(
             HiveFileContext hiveFileContext,
             Optional<EncryptionInformation> encryptionInformation)
     {
-        if (!PARQUET_SERDE_CLASS_NAMES.contains(storage.getStorageFormat().getSerDe()) || shouldUseRecordReaderFromInputFormat(configuration, storage)) {
+        if (!PARQUET_SERDE_CLASS_NAMES.contains(storage.getStorageFormat().getSerDe())) {
             return Optional.empty();
         }
 
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/util/HiveFileIterator.java b/presto-hive/src/main/java/com/facebook/presto/hive/util/HiveFileIterator.java
index 7749eb78148af..20fda0bc96ee7 100644
--- a/presto-hive/src/main/java/com/facebook/presto/hive/util/HiveFileIterator.java
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/util/HiveFileIterator.java
@@ -31,6 +31,7 @@
 import java.util.Collections;
 import java.util.Deque;
 import java.util.Iterator;
+import java.util.Optional;
 
 import static com.facebook.presto.hive.HiveErrorCode.HIVE_FILESYSTEM_ERROR;
 import static com.facebook.presto.hive.HiveErrorCode.HIVE_FILE_NOT_FOUND;
@@ -44,7 +45,7 @@ public class HiveFileIterator
     private final ListDirectoryOperation listDirectoryOperation;
     private final NamenodeStats namenodeStats;
     private final NestedDirectoryPolicy nestedDirectoryPolicy;
-    private final PathFilter pathFilter;
+    private final Optional<PathFilter> pathFilter;
 
     private Iterator<HiveFileInfo> remoteIterator = Collections.emptyIterator();
 
@@ -53,7 +54,7 @@ public HiveFileIterator(
             ListDirectoryOperation listDirectoryOperation,
             NamenodeStats namenodeStats,
             NestedDirectoryPolicy nestedDirectoryPolicy,
-            PathFilter pathFilter)
+            Optional<PathFilter> pathFilter)
     {
         paths.addLast(requireNonNull(path, "path is null"));
         this.listDirectoryOperation = requireNonNull(listDirectoryOperation, "listDirectoryOperation is null");
@@ -97,10 +98,14 @@ protected HiveFileInfo computeNext()
         }
     }
 
-    private Iterator<HiveFileInfo> getLocatedFileStatusRemoteIterator(Path path, PathFilter pathFilter)
+    private Iterator<HiveFileInfo> getLocatedFileStatusRemoteIterator(Path path, Optional<PathFilter> pathFilter)
     {
         try (TimeStat.BlockTimer ignored = namenodeStats.getListLocatedStatus().time()) {
-            return Iterators.filter(new FileStatusIterator(path, listDirectoryOperation, namenodeStats), input -> pathFilter.accept(input.getPath()));
+            FileStatusIterator statusIterator = new FileStatusIterator(path, listDirectoryOperation, namenodeStats);
+            if (!pathFilter.isPresent()) {
+                return statusIterator;
+            }
+            return Iterators.filter(new FileStatusIterator(path, listDirectoryOperation, namenodeStats), input -> pathFilter.get().accept(input.getPath()));
         }
     }
 
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 05d5dba5f0695..76d33f3215c75 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
@@ -34,9 +34,9 @@ public class HudiRealtimeSplitConverter
         implements CustomSplitConverter
 {
     public static final String CUSTOM_SPLIT_CLASS_KEY = "custom_split_class";
-    private static final String HUDI_DELTA_FILEPATHS_KEY = "hudi_delta_filepaths";
-    private static final String HUDI_BASEPATH_KEY = "hudi_basepath";
-    private static final String HUDI_MAX_COMMIT_TIME_KEY = "hudi_max_commit_time";
+    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";
 
     @Override
     public Optional<Map<String, String>> extractCustomSplitInfo(FileSplit split)
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/TestHiveUtil.java b/presto-hive/src/test/java/com/facebook/presto/hive/TestHiveUtil.java
index d6460ca340b33..520f5424d5469 100644
--- a/presto-hive/src/test/java/com/facebook/presto/hive/TestHiveUtil.java
+++ b/presto-hive/src/test/java/com/facebook/presto/hive/TestHiveUtil.java
@@ -24,6 +24,7 @@
 import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.serde2.thrift.ThriftDeserializer;
 import org.apache.hadoop.hive.serde2.thrift.test.IntString;
+import org.apache.hudi.hadoop.realtime.HoodieRealtimeFileSplit;
 import org.apache.thrift.protocol.TBinaryProtocol;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
@@ -44,11 +45,16 @@
 import static com.facebook.presto.hive.HiveUtil.shouldUseRecordReaderFromInputFormat;
 import static com.facebook.presto.hive.metastore.MetastoreUtil.toPartitionNamesAndValues;
 import static com.facebook.presto.hive.metastore.MetastoreUtil.toPartitionValues;
+import static com.facebook.presto.hive.util.HudiRealtimeSplitConverter.CUSTOM_SPLIT_CLASS_KEY;
+import static com.facebook.presto.hive.util.HudiRealtimeSplitConverter.HUDI_BASEPATH_KEY;
+import static com.facebook.presto.hive.util.HudiRealtimeSplitConverter.HUDI_DELTA_FILEPATHS_KEY;
+import static com.facebook.presto.hive.util.HudiRealtimeSplitConverter.HUDI_MAX_COMMIT_TIME_KEY;
 import static com.google.common.base.Preconditions.checkState;
 import static org.apache.hadoop.hive.serde.serdeConstants.SERIALIZATION_CLASS;
 import static org.apache.hadoop.hive.serde.serdeConstants.SERIALIZATION_FORMAT;
 import static org.apache.hadoop.hive.serde.serdeConstants.SERIALIZATION_LIB;
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 
 public class TestHiveUtil
@@ -124,10 +130,15 @@ public void testToPartitionNamesAndValues()
     public void testShouldUseRecordReaderFromInputFormat()
     {
         StorageFormat hudiStorageFormat = StorageFormat.create("parquet.hive.serde.ParquetHiveSerDe", "org.apache.hudi.hadoop.HoodieParquetInputFormat", "");
-        assertTrue(shouldUseRecordReaderFromInputFormat(new Configuration(), new Storage(hudiStorageFormat, "test", Optional.empty(), true, ImmutableMap.of(), ImmutableMap.of())));
+        assertFalse(shouldUseRecordReaderFromInputFormat(new Configuration(), new Storage(hudiStorageFormat, "test", Optional.empty(), true, ImmutableMap.of(), ImmutableMap.of()), ImmutableMap.of()));
 
         StorageFormat hudiRealtimeStorageFormat = StorageFormat.create("parquet.hive.serde.ParquetHiveSerDe", "org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat", "");
-        assertTrue(shouldUseRecordReaderFromInputFormat(new Configuration(), new Storage(hudiRealtimeStorageFormat, "test", Optional.empty(), true, ImmutableMap.of(), ImmutableMap.of())));
+        Map<String, String> customSplitInfo = ImmutableMap.of(
+                CUSTOM_SPLIT_CLASS_KEY, HoodieRealtimeFileSplit.class.getName(),
+                HUDI_BASEPATH_KEY, "/test/file.parquet",
+                HUDI_DELTA_FILEPATHS_KEY, "/test/.file_100.log",
+                HUDI_MAX_COMMIT_TIME_KEY, "100");
+        assertTrue(shouldUseRecordReaderFromInputFormat(new Configuration(), new Storage(hudiRealtimeStorageFormat, "test", Optional.empty(), true, ImmutableMap.of(), ImmutableMap.of()), customSplitInfo));
     }
 
     private static void assertToPartitionValues(String partitionName)
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/util/TestHiveFileIterator.java b/presto-hive/src/test/java/com/facebook/presto/hive/util/TestHiveFileIterator.java
index b1e332b43a758..a0b2a4c9b3f75 100644
--- a/presto-hive/src/test/java/com/facebook/presto/hive/util/TestHiveFileIterator.java
+++ b/presto-hive/src/test/java/com/facebook/presto/hive/util/TestHiveFileIterator.java
@@ -29,6 +29,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 
 import static com.facebook.presto.hive.NestedDirectoryPolicy.IGNORED;
 import static com.facebook.presto.hive.NestedDirectoryPolicy.RECURSE;
@@ -77,7 +78,7 @@ public void testDefaultPathFilterNoRecursion()
         createFiles(basePath, 5, false);
         Path rootPath = new Path("file://" + basePath + File.separator);
         PathFilter pathFilter = path -> true;
-        HiveFileIterator hiveFileIterator = new HiveFileIterator(rootPath, listDirectoryOperation, new NamenodeStats(), IGNORED, pathFilter);
+        HiveFileIterator hiveFileIterator = new HiveFileIterator(rootPath, listDirectoryOperation, new NamenodeStats(), IGNORED, Optional.of(pathFilter));
 
         int actualCount = Iterators.size(hiveFileIterator);
         assertEquals(actualCount, 8);
@@ -108,7 +109,7 @@ public void testDefaultPathFilterWithRecursion()
         createFiles(dir2, 4, false);
         Path rootPath = new Path("file://" + basePath + File.separator);
         PathFilter pathFilter = path -> true;
-        HiveFileIterator hiveFileIterator = new HiveFileIterator(rootPath, listDirectoryOperation, new NamenodeStats(), RECURSE, pathFilter);
+        HiveFileIterator hiveFileIterator = new HiveFileIterator(rootPath, listDirectoryOperation, new NamenodeStats(), RECURSE, Optional.of(pathFilter));
 
         int actualCount = Iterators.size(hiveFileIterator);
         assertEquals(actualCount, 20);
@@ -129,7 +130,7 @@ public void testPathFilterWithNoRecursion()
         createFiles(basePath, 5, false);
         Path rootPath = new Path("file://" + basePath + File.separator);
         PathFilter pathFilter = path -> path.getName().contains(PATH_FILTER_MATCHED_PREFIX);
-        HiveFileIterator hiveFileIterator = new HiveFileIterator(rootPath, listDirectoryOperation, new NamenodeStats(), IGNORED, pathFilter);
+        HiveFileIterator hiveFileIterator = new HiveFileIterator(rootPath, listDirectoryOperation, new NamenodeStats(), IGNORED, Optional.of(pathFilter));
 
         int actualCount = Iterators.size(hiveFileIterator);
         assertEquals(actualCount, 3);
@@ -160,7 +161,7 @@ public void testPathFilterWithRecursion()
         createFiles(dir2, 4, false);
         Path rootPath = new Path("file://" + basePath + File.separator);
         PathFilter pathFilter = path -> path.getName().contains(PATH_FILTER_MATCHED_PREFIX);
-        HiveFileIterator hiveFileIterator = new HiveFileIterator(rootPath, listDirectoryOperation, new NamenodeStats(), RECURSE, pathFilter);
+        HiveFileIterator hiveFileIterator = new HiveFileIterator(rootPath, listDirectoryOperation, new NamenodeStats(), RECURSE, Optional.of(pathFilter));
 
         int actualCount = Iterators.size(hiveFileIterator);
         assertEquals(actualCount, 9);