【feat】tableSource 添加对 charset 的支持

Author: FlechazoW

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -38,7 +38,6 @@
 import org.apache.flink.types.Row;
 
 import com.dtstack.flink.sql.classloader.ClassLoaderManager;
-import com.dtstack.flink.sql.constrant.ConfigConstrant;
 import com.dtstack.flink.sql.enums.ClusterMode;
 import com.dtstack.flink.sql.enums.ECacheType;
 import com.dtstack.flink.sql.enums.EPluginLoadMode;

core/src/main/java/com/dtstack/flink/sql/format/dtnest/DtNestRowDeserializationSchema.java

@@ -43,7 +43,7 @@
 
 /**
  * source data parse to json format
- *
+ * <p>
  * Date: 2019/12/12
  * Company: www.dtstack.com
  *
@@ -58,9 +58,12 @@ public class DtNestRowDeserializationSchema extends AbstractDeserializationSchem
 
     private final String[] fieldNames;
     private final TypeInformation<?>[] fieldTypes;
-    private List<TableInfo.FieldExtraInfo> fieldExtraInfos;
+    private List<AbstractTableInfo.FieldExtraInfo> fieldExtraInfos;
+    private String charsetName;
 
-    public DtNestRowDeserializationSchema(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping, List<TableInfo.FieldExtraInfo> fieldExtraInfos) {
+    public DtNestRowDeserializationSchema(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping,
+                                          List<AbstractTableInfo.FieldExtraInfo> fieldExtraInfos,
+                                          String charsetName) {
         this.fieldNames = ((RowTypeInfo) typeInfo).getFieldNames();
         this.fieldTypes = ((RowTypeInfo) typeInfo).getFieldTypes();
         this.rowAndFieldMapping = rowAndFieldMapping;
@@ -99,7 +102,7 @@ public Row deserialize(byte[] message) throws IOException {
         }
     }
 
-    private void parseTree(JsonNode jsonNode, String prefix){
+    private void parseTree(JsonNode jsonNode, String prefix) {
         if (jsonNode.isArray()) {
             ArrayNode array = (ArrayNode) jsonNode;
             for (int i = 0; i < array.size(); i++) {
@@ -118,15 +121,15 @@ private void parseTree(JsonNode jsonNode, String prefix){
             return;
         }
         Iterator<String> iterator = jsonNode.fieldNames();
-        while (iterator.hasNext()){
+        while (iterator.hasNext()) {
             String next = iterator.next();
             JsonNode child = jsonNode.get(next);
             String nodeKey = getNodeKey(prefix, next);
 
             nodeAndJsonNodeMapping.put(nodeKey, child);
-            if(child.isArray()){
+            if (child.isArray()) {
                 parseTree(child, nodeKey);
-            }else {
+            } else {
                 parseTree(child, nodeKey);
             }
         }
@@ -137,8 +140,8 @@ private JsonNode getIgnoreCase(String key) {
         return nodeAndJsonNodeMapping.get(nodeMappingKey);
     }
 
-    private String getNodeKey(String prefix, String nodeName){
-        if(Strings.isNullOrEmpty(prefix)){
+    private String getNodeKey(String prefix, String nodeName) {
+        if (Strings.isNullOrEmpty(prefix)) {
             return nodeName;
         }
         return prefix + "." + nodeName;
@@ -162,15 +165,15 @@ private Object convert(JsonNode node, TypeInformation<?> info) {
             } else {
                 return node.asText();
             }
-        }  else if (info.getTypeClass().equals(Types.SQL_DATE.getTypeClass())) {
+        } else if (info.getTypeClass().equals(Types.SQL_DATE.getTypeClass())) {
             return Date.valueOf(node.asText());
         } else if (info.getTypeClass().equals(Types.SQL_TIME.getTypeClass())) {
             // local zone
             return Time.valueOf(node.asText());
         } else if (info.getTypeClass().equals(Types.SQL_TIMESTAMP.getTypeClass())) {
             // local zone
             return Timestamp.valueOf(node.asText());
-        }  else {
+        } else {
             // for types that were specified without JSON schema
             // e.g. POJOs
             try {

kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceTableInfo.java

@@ -19,6 +19,7 @@
 
 package com.dtstack.flink.sql.source.kafka.table;
 
+import com.dtstack.flink.sql.format.FormatType;
 import com.dtstack.flink.sql.table.AbstractSourceTableInfo;
 import com.google.common.base.Preconditions;
 
@@ -71,6 +72,7 @@ public class KafkaSourceTableInfo extends AbstractSourceTableInfo {
 
     public Map<String, String> kafkaParams = new HashMap<>();
 
+    public String charsetName;
 
     public String getBootstrapServers() {
         return bootstrapServers;
@@ -148,11 +150,19 @@ public void setFieldDelimiter(String fieldDelimiter) {
         this.fieldDelimiter = fieldDelimiter;
     }
 
-    @Override
-    public boolean check() {
-        Preconditions.checkNotNull(getType(), "kafka of type is required");
-        Preconditions.checkNotNull(bootstrapServers, "kafka of bootstrapServers is required");
-        Preconditions.checkNotNull(topic, "kafka of topic is required");
-        return false;
-    }
+	public String getCharsetName() {
+		return charsetName;
+	}
+
+	public void setCharsetName(String charsetName) {
+		this.charsetName = charsetName;
+	}
+
+	@Override
+	public boolean check() {
+		Preconditions.checkNotNull(getType(), "kafka of type is required");
+		Preconditions.checkNotNull(bootstrapServers, "kafka of bootstrapServers is required");
+		Preconditions.checkNotNull(topic, "kafka of topic is required");
+		return false;
+	}
 }