Support Hive 3.x

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-hive/pom.xml

@@ -27,8 +27,8 @@
     <artifactId>geaflow-dsl-connector-hive</artifactId>
 
     <properties>
-        <hive.version>2.3.6</hive.version>
-        <hadoop.version>2.7.4</hadoop.version>
+        <hive.version>3.1.2</hive.version>
+        <hadoop.version>3.1.0</hadoop.version>
         <hiverunner.version>4.0.0</hiverunner.version>
         <reflections.version>0.9.10</reflections.version>
         <jetty.test.version>7.6.0.v20120127</jetty.test.version>
@@ -251,10 +251,6 @@
                     <groupId>org.apache.calcite</groupId>
                     <artifactId>calcite-core</artifactId>
                 </exclusion>
-                <exclusion>
-                    <groupId>org.apache.calcite</groupId>
-                    <artifactId>calcite-druid</artifactId>
-                </exclusion>
                 <exclusion>
                     <groupId>org.apache.calcite.avatica</groupId>
                     <artifactId>avatica</artifactId>
@@ -366,5 +362,10 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <artifactId>hadoop-hdfs</artifactId>
+            <groupId>org.apache.hadoop</groupId>
+            <version>${hadoop.version}</version>
+        </dependency>
     </dependencies>
 </project>

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-hive/src/main/java/com/antgroup/geaflow/dsl/connector/hive/adapter/Hive3Adapter.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2023 AntGroup CO., Ltd.
+ *
+ * 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.
+ */
+
+package com.antgroup.geaflow.dsl.connector.hive.adapter;
+
+import com.antgroup.geaflow.dsl.common.exception.GeaFlowDSLException;
+import java.lang.reflect.Method;
+import java.util.Properties;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.RetryingMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Table;
+
+public class Hive3Adapter implements HiveVersionAdapter {
+
+    private final String version;
+    private static final String METHOD_NAME_GET_PROXY = "getProxy";
+    private static final String METHOD_NAME_GET_TABLE_META_DATA = "getTableMetadata";
+
+    public Hive3Adapter(String version) {
+        this.version = version;
+    }
+
+    @Override
+    public String version() {
+        return version;
+    }
+
+    @Override
+    public IMetaStoreClient createMetaSoreClient(HiveConf hiveConf) {
+        try {
+            Method method = RetryingMetaStoreClient.class
+                .getMethod(METHOD_NAME_GET_PROXY, Configuration.class, Boolean.TYPE);
+            return (IMetaStoreClient) method.invoke(null, hiveConf, true);
+        } catch (Exception ex) {
+            throw new RuntimeException("Failed to create Hive Metastore client", ex);
+        }
+    }
+
+    @Override
+    public Properties getTableMetadata(Table table) {
+        try {
+            Class metaStoreUtilsClass = Class.forName("org.apache.hadoop.hive.metastore.utils.MetaStoreUtils");
+            Method method =
+                metaStoreUtilsClass.getMethod(METHOD_NAME_GET_TABLE_META_DATA, Table.class);
+            return (Properties) method.invoke(null, table);
+        } catch (Exception e) {
+            throw new GeaFlowDSLException(e);
+        }
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-hive/src/main/java/com/antgroup/geaflow/dsl/connector/hive/adapter/HiveVersionAdapters.java

@@ -37,6 +37,14 @@ public class HiveVersionAdapters {
 
     public static final String HIVE_239 = "2.3.9";
 
+    public static final String HIVE_300 = "3.0.0";
+
+    public static final String HIVE_310 = "3.1.0";
+
+    public static final String HIVE_311 = "3.1.1";
+
+    public static final String HIVE_312 = "3.1.2";
+
     public static HiveVersionAdapter get() {
         Package myPackage = HiveVersionAnnotation.class.getPackage();
         HiveVersionAnnotation version = myPackage.getAnnotation(HiveVersionAnnotation.class);
@@ -51,6 +59,11 @@ public static HiveVersionAdapter get() {
             case HIVE_237:
             case HIVE_239:
                 return new Hive23Adapter(version.version());
+            case HIVE_300:
+            case HIVE_310:
+            case HIVE_311:
+            case HIVE_312:
+                return new Hive3Adapter(version.version());
             default:
                 throw new GeaFlowDSLException("Hive version: {} is not supported.", version.version());
         }

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-hive/src/test/java/com/antgroup/geaflow/dsl/connector/hive/BaseHiveTest.java

@@ -74,6 +74,8 @@ private HiveConf createHiveConf(Configuration conf, String hiveLocation) {
         conf.set("hive.metastore.schema.verification", "false");
         conf.set("datanucleus.autoCreateSchema", "true");
         conf.set("datanucleus.fixedDatastore", "false");
+        conf.set("datanucleus.schema.autoCreateAll", "true");
+        conf.set("hive.stats.autogather","false");
 
         String scratchDir = FileUtil.concatPath(hiveLocation, "scratch");
         conf.set(ConfVars.SCRATCHDIR.varname, scratchDir);