From 193cb46eca80c5c5569ef94cd0ac189e7a3cac96 Mon Sep 17 00:00:00 2001
From: Udit Mehrotra <udit.mehrotra90@gmail.com>
Date: Mon, 7 Dec 2020 16:09:29 -0800
Subject: [PATCH] Support listing of Hudi files through its metadata

---
 pom.xml                                       |  42 ++++++-
 presto-hive/pom.xml                           |   6 +
 .../presto/hive/HiveClientConfig.java         |  28 +++++
 .../presto/hive/HiveSessionProperties.java    |  22 ++++
 .../com/facebook/presto/hive/HiveUtil.java    |  11 ++
 .../presto/hive/HudiDirectoryLister.java      | 110 ++++++++++++++++++
 .../presto/hive/StoragePartitionLoader.java   |  19 +--
 .../presto/hive/TestHiveClientConfig.java     |  10 +-
 8 files changed, 238 insertions(+), 10 deletions(-)
 create mode 100644 presto-hive/src/main/java/com/facebook/presto/hive/HudiDirectoryLister.java

diff --git a/pom.xml b/pom.xml
index 8eaa9226998cd..e32f232b4af64 100644
--- a/pom.xml
+++ b/pom.xml
@@ -67,7 +67,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.5.3</dep.hudi.version>
+        <dep.hudi.version>0.7.0-rc1</dep.hudi.version>
         <!--
           America/Bahia_Banderas has:
            - offset change since 1970 (offset Jan 1970: -08:00, offset Jan 2018: -06:00)
@@ -1029,11 +1029,51 @@
                 </exclusions>
             </dependency>
 
+            <dependency>
+                <groupId>org.apache.hudi</groupId>
+                <artifactId>hudi-common</artifactId>
+                <version>${dep.hudi.version}</version>
+                <exclusions>
+                    <exclusion>
+                        <groupId>org.apache.hbase</groupId>
+                        <artifactId>hbase-server</artifactId>
+                    </exclusion>
+                    <exclusion>
+                        <groupId>com.fasterxml.jackson.core</groupId>
+                        <artifactId>jackson-annotations</artifactId>
+                    </exclusion>
+                    <exclusion>
+                        <groupId>com.fasterxml.jackson.core</groupId>
+                        <artifactId>jackson-databind</artifactId>
+                    </exclusion>
+                    <exclusion>
+                        <groupId>org.apache.httpcomponents</groupId>
+                        <artifactId>httpclient</artifactId>
+                    </exclusion>
+                    <exclusion>
+                        <groupId>org.apache.httpcomponents</groupId>
+                        <artifactId>fluent-hc</artifactId>
+                    </exclusion>
+                    <exclusion>
+                        <groupId>org.rocksdb</groupId>
+                        <artifactId>rocksdbjni</artifactId>
+                    </exclusion>
+                    <exclusion>
+                        <groupId>com.esotericsoftware</groupId>
+                        <artifactId>kryo-shaded</artifactId>
+                    </exclusion>
+                </exclusions>
+            </dependency>
+
             <dependency>
                 <groupId>org.apache.hudi</groupId>
                 <artifactId>hudi-hadoop-mr</artifactId>
                 <version>${dep.hudi.version}</version>
                 <exclusions>
+                    <exclusion>
+                        <groupId>org.apache.hbase</groupId>
+                        <artifactId>hbase-server</artifactId>
+                    </exclusion>
                     <exclusion>
                         <groupId>org.objenesis</groupId>
                         <artifactId>objenesis</artifactId>
diff --git a/presto-hive/pom.xml b/presto-hive/pom.xml
index a6aaf1479af47..165049d49f6f3 100644
--- a/presto-hive/pom.xml
+++ b/presto-hive/pom.xml
@@ -85,6 +85,12 @@
             </exclusions>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.hudi</groupId>
+            <artifactId>hudi-common</artifactId>
+            <version>${dep.hudi.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.hudi</groupId>
             <artifactId>hudi-hadoop-mr</artifactId>
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 e6fd16f09c7c2..35cce1131f873 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
@@ -184,6 +184,8 @@ public class HiveClientConfig
     private boolean fileRenamingEnabled;
     private boolean preferManifestToListFiles;
     private boolean manifestVerificationEnabled;
+    private boolean preferMetadataToListHudiFiles;
+    private boolean hudiMetadataVerificationEnabled;
 
     public int getMaxInitialSplits()
     {
@@ -1552,4 +1554,30 @@ public boolean isManifestVerificationEnabled()
     {
         return this.manifestVerificationEnabled;
     }
+
+    @Config("hive.prefer-metadata-to-list-hudi-files")
+    @ConfigDescription("For Hudi tables prefer to fetch the list of file names and sizes from metadata rather than storage")
+    public HiveClientConfig setPreferMetadataToListHudiFiles(boolean preferMetadataToListHudiFiles)
+    {
+        this.preferMetadataToListHudiFiles = preferMetadataToListHudiFiles;
+        return this;
+    }
+
+    public boolean isPreferMetadataToListHudiFiles()
+    {
+        return this.preferMetadataToListHudiFiles;
+    }
+
+    @Config("hive.hudi-metadata-verification-enabled")
+    @ConfigDescription("Enable verification of file names and sizes in Hudi metadata")
+    public HiveClientConfig setHudiMetadataVerificationEnabled(boolean hudiMetadataVerificationEnabled)
+    {
+        this.hudiMetadataVerificationEnabled = hudiMetadataVerificationEnabled;
+        return this;
+    }
+
+    public boolean isHudiMetadataVerificationEnabled()
+    {
+        return this.hudiMetadataVerificationEnabled;
+    }
 }
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 6f8eea4518688..b536042ff9c5a 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
@@ -113,6 +113,8 @@ public final class HiveSessionProperties
     public static final String FILE_RENAMING_ENABLED = "file_renaming_enabled";
     public static final String PREFER_MANIFESTS_TO_LIST_FILES = "prefer_manifests_to_list_files";
     public static final String MANIFEST_VERIFICATION_ENABLED = "manifest_verification_enabled";
+    public static final String PREFER_METADATA_TO_LIST_HUDI_FILES = "prefer_metadata_to_list_hudi_files";
+    public static final String HUDI_METADATA_VERIFICATION_ENABLED = "hudi_metadata_verification_enabled";
 
     private final List<PropertyMetadata<?>> sessionProperties;
 
@@ -532,6 +534,16 @@ public HiveSessionProperties(HiveClientConfig hiveClientConfig, OrcFileWriterCon
                         MANIFEST_VERIFICATION_ENABLED,
                         "Enable manifest verification",
                         hiveClientConfig.isManifestVerificationEnabled(),
+                        false),
+                booleanProperty(
+                        PREFER_METADATA_TO_LIST_HUDI_FILES,
+                        "For Hudi tables prefer to fetch the list of files from its metadata",
+                        hiveClientConfig.isPreferMetadataToListHudiFiles(),
+                        false),
+                booleanProperty(
+                        HUDI_METADATA_VERIFICATION_ENABLED,
+                        "Verify file listing maintained in Hudi table metadata against the file system",
+                        hiveClientConfig.isHudiMetadataVerificationEnabled(),
                         false));
     }
 
@@ -932,4 +944,14 @@ public static boolean isManifestVerificationEnabled(ConnectorSession session)
     {
         return session.getProperty(MANIFEST_VERIFICATION_ENABLED, Boolean.class);
     }
+
+    public static boolean isPreferMetadataToListHudiFiles(ConnectorSession session)
+    {
+        return session.getProperty(PREFER_METADATA_TO_LIST_HUDI_FILES, Boolean.class);
+    }
+
+    public static boolean isHudiMetadataVerificationEnabled(ConnectorSession session)
+    {
+        return session.getProperty(HUDI_METADATA_VERIFICATION_ENABLED, 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 88d27cfd5a3d7..8bc4813d3cf41 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
@@ -75,6 +75,7 @@
 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.HoodieRealtimeFileSplit;
 import org.joda.time.DateTimeZone;
 import org.joda.time.format.DateTimeFormat;
@@ -287,6 +288,16 @@ private static boolean isHudiRealtimeSplit(Map<String, String> customSplitInfo)
         return HoodieRealtimeFileSplit.class.getName().equals(customSplitClass);
     }
 
+    /**
+     * Checks whether a table is a Hudi table based on configured InputFormat.
+     * @param inputFormat InputFormat configured for the table
+     * @return Boolean indicating whether the table is a Hudi table or not.
+     */
+    static boolean isHudiTable(InputFormat<?, ?> inputFormat)
+    {
+        return inputFormat instanceof HoodieParquetInputFormat;
+    }
+
     public static void setReadColumns(Configuration configuration, List<Integer> readHiveColumnIndexes)
     {
         configuration.set(READ_COLUMN_IDS_CONF_STR, Joiner.on(',').join(readHiveColumnIndexes));
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..1e35fe8321690
--- /dev/null
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/HudiDirectoryLister.java
@@ -0,0 +1,110 @@
+/*
+ * 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.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.isHudiMetadataVerificationEnabled;
+import static com.facebook.presto.hive.HiveSessionProperties.isPreferMetadataToListHudiFiles;
+
+public final class HudiDirectoryLister
+        implements DirectoryLister
+{
+    private HoodieTableFileSystemView fileSystemView;
+
+    public HudiDirectoryLister(Configuration conf,
+                               ConnectorSession session,
+                               Table table)
+    {
+        System.out.println("Uditt: Initializing Hudi Directory Lister...");
+        HoodieTableMetaClient metaClient = new HoodieTableMetaClient(conf, table.getStorage().getLocation());
+        HoodieEngineContext engineContext = new HoodieLocalEngineContext(conf);
+        HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder()
+                .enable(isPreferMetadataToListHudiFiles(session))
+                .validate(isHudiMetadataVerificationEnabled(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)
+    {
+        return new HiveFileIterator(
+                path,
+                p -> new HudiFileInfoIterator(fileSystemView, table.getStorage().getLocation(), p),
+                namenodeStats,
+                hiveDirectoryContext.getNestedDirectoryPolicy(),
+                pathFilter);
+    }
+
+    public static class HudiFileInfoIterator
+            implements RemoteIterator<HiveFileInfo>
+    {
+        private final Iterator<HoodieBaseFile> hoodieBaseFileIterator;
+
+        public HudiFileInfoIterator(HoodieTableFileSystemView fileSystemView,
+                                    String tablePath,
+                                    Path directory)
+        {
+            String partition = FSUtils.getRelativePartitionPath(new Path(tablePath), directory);
+            this.hoodieBaseFileIterator = fileSystemView.getLatestBaseFiles(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:50010"};
+            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 e363e7a338a78..bd2377ee39d88 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,7 +23,6 @@
 import com.facebook.presto.spi.ConnectorSession;
 import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.SchemaTableName;
-import com.google.common.base.Suppliers;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterators;
 import com.google.common.io.CharStreams;
@@ -40,7 +39,6 @@
 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;
@@ -57,7 +55,6 @@
 import java.util.Optional;
 import java.util.Properties;
 import java.util.function.IntPredicate;
-import java.util.function.Supplier;
 
 import static com.facebook.presto.hive.HiveBucketing.getVirtualBucketNumber;
 import static com.facebook.presto.hive.HiveColumnHandle.pathColumnHandle;
@@ -104,7 +101,6 @@ public class StoragePartitionLoader
     private final ConnectorSession session;
     private final Deque<Iterator<InternalHiveSplit>> fileIterators;
     private final boolean schedulerUsesHostAddresses;
-    private final Supplier<HoodieROTablePathFilter> hoodiePathFilterSupplier;
     private final boolean partialAggregationsPushedDown;
 
     public StoragePartitionLoader(
@@ -126,13 +122,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());
         this.fileIterators = requireNonNull(fileIterators, "fileIterators is null");
         this.schedulerUsesHostAddresses = schedulerUsesHostAddresses;
-        this.hoodiePathFilterSupplier = Suppliers.memoize(HoodieROTablePathFilter::new)::get;
         this.partialAggregationsPushedDown = partialAggregationsPushedDown;
+
+        Configuration configuration = hdfsEnvironment.getConfiguration(hdfsContext,
+                new Path(table.getStorage().getLocation()));
+        InputFormat<?, ?> inputFormat = HiveUtil.getInputFormat(configuration,
+                table.getStorage().getStorageFormat().getInputFormat(), false);
+        if (HiveUtil.isHudiTable(inputFormat)) {
+            this.directoryLister = new HudiDirectoryLister(configuration, session, table);
+        }
+        else {
+            this.directoryLister = requireNonNull(directoryLister, "directoryLister is null");
+        }
     }
 
     @Override
@@ -234,7 +239,7 @@ public ListenableFuture<?> loadPartition(HivePartitionMetadata partition, HiveSp
 
             return addSplitsToSource(splits, splitFactory, hiveSplitSource, stopped);
         }
-        PathFilter pathFilter = isHudiParquetInputFormat(inputFormat) ? hoodiePathFilterSupplier.get() : 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.
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/TestHiveClientConfig.java b/presto-hive/src/test/java/com/facebook/presto/hive/TestHiveClientConfig.java
index 034d0f7f1095a..385cdc6a0670f 100644
--- a/presto-hive/src/test/java/com/facebook/presto/hive/TestHiveClientConfig.java
+++ b/presto-hive/src/test/java/com/facebook/presto/hive/TestHiveClientConfig.java
@@ -146,7 +146,9 @@ public void testDefaults()
                 .setPartialAggregationPushdownForVariableLengthDatatypesEnabled(false)
                 .setFileRenamingEnabled(false)
                 .setPreferManifestsToListFiles(false)
-                .setManifestVerificationEnabled(false));
+                .setManifestVerificationEnabled(false)
+                .setPreferMetadataToListHudiFiles(false)
+                .setHudiMetadataVerificationEnabled(false));
     }
 
     @Test
@@ -255,6 +257,8 @@ public void testExplicitPropertyMappings()
                 .put("hive.file_renaming_enabled", "true")
                 .put("hive.prefer-manifests-to-list-files", "true")
                 .put("hive.manifest-verification-enabled", "true")
+                .put("hive.prefer-metadata-to-list-hudi-files", "true")
+                .put("hive.hudi-metadata-verification-enabled", "true")
                 .build();
 
         HiveClientConfig expected = new HiveClientConfig()
@@ -359,7 +363,9 @@ public void testExplicitPropertyMappings()
                 .setPartialAggregationPushdownForVariableLengthDatatypesEnabled(true)
                 .setFileRenamingEnabled(true)
                 .setPreferManifestsToListFiles(true)
-                .setManifestVerificationEnabled(true);
+                .setManifestVerificationEnabled(true)
+                .setPreferMetadataToListHudiFiles(true)
+                .setHudiMetadataVerificationEnabled(true);
 
         ConfigAssertions.assertFullMapping(properties, expected);
     }