解决kudu类型转换的问题和去掉share包

Author: xiuzhu9527

kudu/kudu-side/kudu-all-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAllReqRow.java

@@ -8,8 +8,8 @@
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
-import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Collector;

kudu/kudu-side/kudu-all-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAllSideInfo.java

@@ -10,7 +10,7 @@
 import org.apache.calcite.sql.SqlNode;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
+import com.google.common.collect.Lists;
 
 import java.util.List;
 

kudu/kudu-side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncReqRow.java

@@ -8,9 +8,9 @@
 import com.stumbleupon.async.Deferred;
 import io.vertx.core.json.JsonArray;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
+import com.google.common.collect.Lists;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
+import com.google.common.collect.Maps;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;

kudu/kudu-side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncSideInfo.java

@@ -10,7 +10,7 @@
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
+import com.google.common.collect.Lists;
 
 import java.util.List;
 

kudu/kudu-side/kudu-side-core/src/main/java/com/dtstack/flink/sql/side/kudu/table/KuduSideParser.java

@@ -4,6 +4,9 @@
 import com.dtstack.flink.sql.table.TableInfo;
 import com.dtstack.flink.sql.util.MathUtil;
 
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
 import java.util.Map;
 
 import static com.dtstack.flink.sql.table.TableInfo.PARALLELISM_KEY;
@@ -71,4 +74,40 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         return kuduSideTableInfo;
 
     }
+
+    @Override
+    public Class dbTypeConvertToJavaType(String fieldType) {
+
+        switch (fieldType.toLowerCase()) {
+            case "boolean":
+            case "bool":
+                return Boolean.class;
+            case "int8":
+                return Byte.class;
+            case "int16":
+                return Short.class;
+            case "int":
+            case "int32":
+                return Integer.class;
+            case "long":
+            case "int64":
+                return Long.class;
+            case "varchar":
+            case "binary":
+            case "string":
+                return String.class;
+            case "float":
+                return Float.class;
+            case "double":
+                return Double.class;
+            case "date":
+                return Date.class;
+            case "unixtime_micros":
+                return Timestamp.class;
+            case "decimal":
+                return BigDecimal.class;
+        }
+
+        throw new RuntimeException("不支持 " + fieldType + " 类型");
+    }
 }

kudu/kudu-side/kudu-side-core/src/main/java/com/dtstack/flink/sql/side/kudu/table/KuduSideTableInfo.java

@@ -1,7 +1,7 @@
 package com.dtstack.flink.sql.side.kudu.table;
 
 import com.dtstack.flink.sql.side.SideTableInfo;
-import org.apache.flink.calcite.shaded.com.google.common.base.Preconditions;
+import com.google.common.base.Preconditions;
 
 public class KuduSideTableInfo extends SideTableInfo {
 

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/table/KuduSinkParser.java

@@ -5,6 +5,9 @@
 import com.dtstack.flink.sql.table.TableInfo;
 import com.dtstack.flink.sql.util.MathUtil;
 
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
 import java.util.Map;
 
 import static com.dtstack.flink.sql.table.TableInfo.PARALLELISM_KEY;
@@ -51,4 +54,39 @@ private KuduOutputFormat.WriteMode transWriteMode(String writeMode) {
                 return KuduOutputFormat.WriteMode.UPSERT;
         }
     }
+
+    @Override
+    public Class dbTypeConvertToJavaType(String fieldType) {
+        switch (fieldType.toLowerCase()) {
+            case "boolean":
+            case "bool":
+                return Boolean.class;
+            case "int8":
+                return Byte.class;
+            case "int16":
+                return Short.class;
+            case "int":
+            case "int32":
+                return Integer.class;
+            case "long":
+            case "int64":
+                return Long.class;
+            case "varchar":
+            case "binary":
+            case "string":
+                return String.class;
+            case "float":
+                return Float.class;
+            case "double":
+                return Double.class;
+            case "date":
+                return Date.class;
+            case "unixtime_micros":
+                return Timestamp.class;
+            case "decimal":
+                return BigDecimal.class;
+        }
+
+        throw new RuntimeException("不支持 " + fieldType + " 类型");
+    }
 }

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/table/KuduTableInfo.java

@@ -2,7 +2,7 @@
 
 import com.dtstack.flink.sql.sink.kudu.KuduOutputFormat;
 import com.dtstack.flink.sql.table.TargetTableInfo;
-import org.apache.flink.calcite.shaded.com.google.common.base.Preconditions;
+import com.google.common.base.Preconditions;
 
 public class KuduTableInfo extends TargetTableInfo {