From 7d40f19f395460bc78a12ac452831a9f0393ab49 Mon Sep 17 00:00:00 2001
From: Wenning Ding <wenningd@amazon.com>
Date: Sat, 16 May 2020 13:39:53 -0700
Subject: [PATCH] HUDI-515 Resolve API conflict for Hive 2 & Hive 3

---
 .../hive/HoodieCombineHiveInputFormat.java    | 43 ++++++++++++++++---
 .../hudi/hive/testutils/HiveTestService.java  |  4 +-
 2 files changed, 39 insertions(+), 8 deletions(-)

diff --git a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/hive/HoodieCombineHiveInputFormat.java b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/hive/HoodieCombineHiveInputFormat.java
index 9f024e9d2..fd7ffb221 100644
--- a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/hive/HoodieCombineHiveInputFormat.java
+++ b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/hive/HoodieCombineHiveInputFormat.java
@@ -70,6 +70,7 @@ import org.apache.log4j.Logger;
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
+import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -137,7 +138,7 @@ public class HoodieCombineHiveInputFormat<K extends WritableComparable, V extend
     Set<Path> poolSet = new HashSet<>();
 
     for (Path path : paths) {
-      PartitionDesc part = HiveFileFormatUtils.getPartitionDescFromPathRecursively(pathToPartitionInfo, path,
+      PartitionDesc part = getPartitionFromPath(pathToPartitionInfo, path,
           IOPrepareCache.get().allocatePartitionDescMap());
       TableDesc tableDesc = part.getTableDesc();
       if ((tableDesc != null) && tableDesc.isNonNative()) {
@@ -375,6 +376,34 @@ public class HoodieCombineHiveInputFormat<K extends WritableComparable, V extend
     iss.addAll(Arrays.asList(combine.getSplits(currJob, 1)));
   }
 
+  /**
+   * HiveFileFormatUtils.getPartitionDescFromPathRecursively is no longer available since Hive 3.
+   * This method is to make it compatible with both Hive 2 and Hive 3.
+   * @param pathToPartitionInfo
+   * @param dir
+   * @param cacheMap
+   * @return
+   * @throws IOException
+   */
+  private static PartitionDesc getPartitionFromPath(Map<Path, PartitionDesc> pathToPartitionInfo, Path dir,
+      Map<Map<Path, PartitionDesc>, Map<Path, PartitionDesc>> cacheMap)
+      throws IOException {
+    Method method;
+    try {
+      Class<?> hiveUtilsClass = Class.forName("org.apache.hadoop.hive.ql.io.HiveFileFormatUtils");
+      try {
+        // HiveFileFormatUtils.getPartitionDescFromPathRecursively method only available in Hive 2.x
+        method = hiveUtilsClass.getMethod("getPartitionDescFromPathRecursively", Map.class, Path.class, Map.class);
+      } catch (NoSuchMethodException e) {
+        // HiveFileFormatUtils.getFromPathRecursively method only available in Hive 3.x
+        method = hiveUtilsClass.getMethod("getFromPathRecursively", Map.class, Path.class, Map.class);
+      }
+      return (PartitionDesc) method.invoke(null, pathToPartitionInfo, dir, cacheMap);
+    } catch (ReflectiveOperationException e) {
+      throw new IOException(e);
+    }
+  }
+
   /**
    * MOD - Just added this for visibility.
    */
@@ -568,8 +597,8 @@ public class HoodieCombineHiveInputFormat<K extends WritableComparable, V extend
         // CombinedSplit.
         Path[] ipaths = inputSplitShim.getPaths();
         if (ipaths.length > 0) {
-          PartitionDesc part = HiveFileFormatUtils.getPartitionDescFromPathRecursively(this.pathToPartitionInfo,
-              ipaths[0], IOPrepareCache.get().getPartitionDescMap());
+          PartitionDesc part = getPartitionFromPath(this.pathToPartitionInfo, ipaths[0],
+              IOPrepareCache.get().getPartitionDescMap());
           inputFormatClassName = part.getInputFileFormatClass().getName();
         }
       }
@@ -703,8 +732,8 @@ public class HoodieCombineHiveInputFormat<K extends WritableComparable, V extend
 
         // extract all the inputFormatClass names for each chunk in the
         // CombinedSplit.
-        PartitionDesc part = HiveFileFormatUtils.getPartitionDescFromPathRecursively(pathToPartitionInfo,
-            inputSplitShim.getPath(0), IOPrepareCache.get().getPartitionDescMap());
+        PartitionDesc part = getPartitionFromPath(pathToPartitionInfo, inputSplitShim.getPath(0),
+            IOPrepareCache.get().getPartitionDescMap());
 
         // create a new InputFormat instance if this is the first time to see
         // this class
@@ -957,8 +986,8 @@ public class HoodieCombineHiveInputFormat<K extends WritableComparable, V extend
     public Set<Integer> call() throws Exception {
       Set<Integer> nonCombinablePathIndices = new HashSet<Integer>();
       for (int i = 0; i < length; i++) {
-        PartitionDesc part = HiveFileFormatUtils.getPartitionDescFromPathRecursively(pathToPartitionInfo,
-            paths[i + start], IOPrepareCache.get().allocatePartitionDescMap());
+        PartitionDesc part = getPartitionFromPath(pathToPartitionInfo, paths[i + start],
+            IOPrepareCache.get().allocatePartitionDescMap());
         // Use HiveInputFormat if any of the paths is not splittable
         Class<? extends InputFormat> inputFormatClass = part.getInputFileFormatClass();
         InputFormat<WritableComparable, Writable> inputFormat = getInputFormatFromCache(inputFormatClass, conf);
diff --git a/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestService.java b/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestService.java
index e325094b9..2f9880304 100644
--- a/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestService.java
+++ b/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestService.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.HiveMetaStore;
 import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
 import org.apache.hadoop.hive.metastore.IHMSHandler;
+import org.apache.hadoop.hive.metastore.RetryingHMSHandler;
 import org.apache.hadoop.hive.metastore.TSetIpAddressProcessor;
 import org.apache.hadoop.hive.metastore.TUGIBasedProcessor;
 import org.apache.hadoop.hive.metastore.api.MetaException;
@@ -286,7 +287,8 @@ public class HiveTestService {
       TProcessor processor;
       TTransportFactory transFactory;
 
-      IHMSHandler handler = (IHMSHandler) HiveMetaStore.newRetryingHMSHandler("new db based metaserver", conf, true);
+      HiveMetaStore.HMSHandler baseHandler = new HiveMetaStore.HMSHandler("new db based metaserver", conf, false);
+      IHMSHandler handler = RetryingHMSHandler.getProxy(conf, baseHandler, true);
 
       if (conf.getBoolVar(HiveConf.ConfVars.METASTORE_EXECUTE_SET_UGI)) {
         transFactory = useFramedTransport