Merge branch '1.8_impaka_type' into 'v1.8.0_dev'

impala 类型转换



See merge request !168

Author: toutian

core/pom.xml

@@ -116,13 +116,6 @@
             <version>${flink.version}</version>
         </dependency>
 
-        <!-- test for predicate  -->
-        <dependency>
-            <groupId>com.dtstack.flink</groupId>
-            <artifactId>sql.side.rdb</artifactId>
-            <version>1.0-SNAPSHOT</version>
-            <scope>test</scope>
-        </dependency>
 
     </dependencies>
 

core/src/test/java/com/dtstack/flink/sql/side/SidePredicatesParserTest.java

@@ -18,7 +18,6 @@
 
 package com.dtstack.flink.sql.side;
 
-import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.junit.Assert;
 import org.junit.Test;
@@ -55,7 +54,13 @@ public void testfillPredicatesForSideTable() throws SqlParseException {
                         "           MyTable.a='1' and s.d='1' and  s.d <> '3' and s.c LIKE '%xx%' and s.c in ('1','2') and s.c between '10' and '23' and s.d is not null\n";
 
 
-        SideTableInfo sideTableInfo = new RdbSideTableInfo();
+        SideTableInfo sideTableInfo = new SideTableInfo(){
+            @Override
+            public boolean check() {
+                return false;
+            }
+        };
+
         sideTableInfo.setName("sideTable");
 
         Map<String, SideTableInfo> sideTableMap = new HashMap<>();

impala/impala-side/impala-async-side/src/main/java/com/dtstack/flink/sql/side/impala/ImpalaAsyncReqRow.java

@@ -20,11 +20,9 @@
 
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
-import com.dtstack.flink.sql.side.SideInfo;
 import com.dtstack.flink.sql.side.SideTableInfo;
 import com.dtstack.flink.sql.side.impala.table.ImpalaSideTableInfo;
 import com.dtstack.flink.sql.side.rdb.async.RdbAsyncReqRow;
-import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
 import io.vertx.core.Vertx;
 import io.vertx.core.VertxOptions;
 import io.vertx.core.json.JsonObject;

impala/impala-side/impala-side-core/src/main/java/com/dtstack/flink/sql/side/impala/table/ImpalaSideParser.java

@@ -23,8 +23,14 @@
 import com.dtstack.flink.sql.util.MathUtil;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
-import java.io.IOException;
-import java.util.*;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
 
 /**
  * Reason:
@@ -110,8 +116,36 @@ public Map setPartitionFieldValues(String partitionfieldValuesStr){
             e.printStackTrace();
             throw new RuntimeException(e);
         }
+    }
 
+    @Override
+    public  Class dbTypeConvertToJavaType(String fieldType) {
+        switch (fieldType.toLowerCase()) {
+            case "boolean":
+                return Boolean.class;
+            case "char":
+                return Character.class;
+            case "double":
+                return Double.class;
+            case "float":
+                return Float.class;
+            case "tinyint":
+                return Byte.class;
+            case "smallint":
+                return Short.class;
+            case "int":
+                return Integer.class;
+            case "bigint":
+                return Long.class;
+            case "decimal":
+                return BigDecimal.class;
+            case "string":
+            case "varchar":
+                return String.class;
+            case "timestamp":
+                return Timestamp.class;
+        }
 
-
+        throw new RuntimeException("不支持 " + fieldType + " 类型");
     }
 }

impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/table/ImpalaSinkParser.java

@@ -22,6 +22,8 @@
 import com.dtstack.flink.sql.table.TableInfo;
 import com.dtstack.flink.sql.util.MathUtil;
 
+import java.math.BigDecimal;
+import java.sql.Timestamp;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -55,37 +57,68 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         impalaTableInfo.setSchema(MathUtil.getString(props.get(ImpalaTableInfo.SCHEMA_KEY.toLowerCase())));
 
         Integer authMech = MathUtil.getIntegerVal(props.get(ImpalaTableInfo.AUTHMECH_KEY.toLowerCase()));
-        authMech = authMech == null? 0 : authMech;
+        authMech = authMech == null ? 0 : authMech;
         impalaTableInfo.setAuthMech(authMech);
         List authMechs = Arrays.asList(new Integer[]{0, 1, 2, 3});
 
-        if (!authMechs.contains(authMech) ){
+        if (!authMechs.contains(authMech)) {
             throw new IllegalArgumentException("The value of authMech is illegal, Please select 0, 1, 2, 3");
         } else if (authMech == 1) {
             impalaTableInfo.setPrincipal(MathUtil.getString(props.get(ImpalaTableInfo.PRINCIPAL_KEY.toLowerCase())));
             impalaTableInfo.setKeyTabFilePath(MathUtil.getString(props.get(ImpalaTableInfo.KEYTABFILEPATH_KEY.toLowerCase())));
             impalaTableInfo.setKrb5FilePath(MathUtil.getString(props.get(ImpalaTableInfo.KRB5FILEPATH_KEY.toLowerCase())));
             String krbRealm = MathUtil.getString(props.get(ImpalaTableInfo.KRBREALM_KEY.toLowerCase()));
-            krbRealm = krbRealm == null? "HADOOP.COM" : krbRealm;
+            krbRealm = krbRealm == null ? "HADOOP.COM" : krbRealm;
             impalaTableInfo.setKrbRealm(krbRealm);
             impalaTableInfo.setKrbHostFQDN(MathUtil.getString(props.get(impalaTableInfo.KRBHOSTFQDN_KEY.toLowerCase())));
             impalaTableInfo.setKrbServiceName(MathUtil.getString(props.get(impalaTableInfo.KRBSERVICENAME_KEY.toLowerCase())));
-        } else if (authMech == 2 ) {
+        } else if (authMech == 2) {
             impalaTableInfo.setUserName(MathUtil.getString(props.get(ImpalaTableInfo.USER_NAME_KEY.toLowerCase())));
         } else if (authMech == 3) {
             impalaTableInfo.setUserName(MathUtil.getString(props.get(ImpalaTableInfo.USER_NAME_KEY.toLowerCase())));
             impalaTableInfo.setPassword(MathUtil.getString(props.get(ImpalaTableInfo.PASSWORD_KEY.toLowerCase())));
         }
 
-        String enablePartitionStr  = (String) props.get(ImpalaTableInfo.ENABLEPARITION_KEY.toLowerCase());
-        boolean enablePartition = MathUtil.getBoolean(enablePartitionStr == null? "false":enablePartitionStr);
+        String enablePartitionStr = (String) props.get(ImpalaTableInfo.ENABLEPARITION_KEY.toLowerCase());
+        boolean enablePartition = MathUtil.getBoolean(enablePartitionStr == null ? "false" : enablePartitionStr);
         impalaTableInfo.setEnablePartition(enablePartition);
-        if(enablePartition){
+        if (enablePartition) {
             String partitionFields = MathUtil.getString(props.get(ImpalaTableInfo.PARTITIONFIELDS_KEY.toLowerCase()));
             impalaTableInfo.setPartitionFields(partitionFields);
         }
 
         impalaTableInfo.check();
         return impalaTableInfo;
     }
+
+    @Override
+    public  Class dbTypeConvertToJavaType(String fieldType) {
+        switch (fieldType.toLowerCase()) {
+            case "boolean":
+                return Boolean.class;
+            case "char":
+                return Character.class;
+            case "double":
+                return Double.class;
+            case "float":
+                return Float.class;
+            case "tinyint":
+                return Byte.class;
+            case "smallint":
+                return Short.class;
+            case "int":
+                return Integer.class;
+            case "bigint":
+                return Long.class;
+            case "decimal":
+                return BigDecimal.class;
+            case "string":
+            case "varchar":
+                return String.class;
+            case "timestamp":
+                return Timestamp.class;
+        }
+
+        throw new RuntimeException("不支持 " + fieldType + " 类型");
+    }
 }