[fix-33847][core][kafka] add array and map type in the kafka sink.

Author: a49a

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

@@ -334,9 +334,7 @@ public static Set<URL> registerTable(SqlTree sqlTree,
             } else if (tableInfo instanceof AbstractTargetTableInfo) {
 
                 TableSink tableSink = StreamSinkFactory.getTableSink((AbstractTargetTableInfo) tableInfo, localSqlPluginPath, pluginLoadMode);
-                TypeInformation[] flinkTypes = FunctionManager.transformTypes(tableInfo.getFieldClasses());
-                tableEnv.registerTableSink(tableInfo.getName(), tableInfo.getFields(), flinkTypes, tableSink);
-
+                tableEnv.registerTableSink(tableInfo.getName(), tableSink);
                 URL sinkTablePathUrl = PluginUtil.buildSourceAndSinkPathByLoadMode(tableInfo.getType(), AbstractTargetTableInfo.TARGET_SUFFIX, localSqlPluginPath, remoteSqlPluginPath, pluginLoadMode);
                 pluginClassPathSets.add(sinkTablePathUrl);
             } else if (tableInfo instanceof AbstractSideTableInfo) {

core/src/main/java/com/dtstack/flink/sql/function/FunctionManager.java

@@ -124,13 +124,4 @@ public static void registerAggregateUDF(String classPath, String funcName, Table
         }
     }
 
-
-    public static TypeInformation[] transformTypes(Class[] fieldTypes) {
-        TypeInformation[] types = new TypeInformation[fieldTypes.length];
-        for (int i = 0; i < fieldTypes.length; i++) {
-            types[i] = TypeInformation.of(fieldTypes[i]);
-        }
-
-        return types;
-    }
 }

core/src/main/java/com/dtstack/flink/sql/util/ClassUtil.java

@@ -25,6 +25,8 @@
 import java.sql.Date;
 import java.sql.Time;
 import java.sql.Timestamp;
+import java.util.HashMap;
+import java.util.Map;
 
 /**
  * Reason: TODO ADD REASON(可选)
@@ -41,6 +43,10 @@ public static Class<?> stringConvertClass(String str) {
         if (lowerStr.startsWith("array")) {
             return Array.newInstance(Integer.class, 0).getClass();
         }
+        if (lowerStr.startsWith("map")) {
+            Map m = new HashMap();
+            return m.getClass();
+        }
 
         switch (lowerStr) {
             case "boolean":

core/src/main/java/com/dtstack/flink/sql/util/DataTypeUtils.java

@@ -20,6 +20,7 @@
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
@@ -30,6 +31,8 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import static org.apache.commons.lang3.StringUtils.split;
+
 /**
  * @program: flink.sql
  * @author: wuren
@@ -39,28 +42,13 @@ public class DataTypeUtils {
 
     private final static Pattern COMPOSITE_TYPE_PATTERN = Pattern.compile("(.+?)<(.+)>");
     private final static String ARRAY = "ARRAY";
+    private final static String MAP = "MAP";
     private final static String ROW = "ROW";
     private final static char FIELD_DELIMITER = ',';
     private final static char TYPE_DELIMITER = ' ';
 
     private DataTypeUtils() {}
 
-    /**
-     * 现在只支持ARRAY类型后续可以加入 MAP等类型
-     * @param compositeTypeString
-     * @return
-     */
-    public static TypeInformation convertToCompositeType(String compositeTypeString) {
-        Matcher matcher = matchCompositeType(compositeTypeString);
-        final String errorMsg = "type " + compositeTypeString + "is not support!";
-        Preconditions.checkState(matcher.find(), errorMsg);
-
-        String normalizedType = normalizeType(matcher.group(1));
-        Preconditions.checkState(ARRAY.equals(normalizedType), errorMsg);
-
-        return convertToArray(compositeTypeString);
-    }
-
     /**
      * 目前ARRAY里只支持ROW和其他基本类型
      * @param arrayTypeString
@@ -86,6 +74,30 @@ public static TypeInformation convertToArray(String arrayTypeString) {
         return Types.OBJECT_ARRAY(elementType);
     }
 
+    /**
+     * 目前Map里只支持基本类型
+     * @param mapTypeString
+     * @return
+     */
+    public static TypeInformation convertToMap(String mapTypeString) {
+        Matcher matcher = matchCompositeType(mapTypeString);
+        final String errorMsg =  mapTypeString + "convert to map type error!";
+        Preconditions.checkState(matcher.find(), errorMsg);
+
+        String normalizedType = normalizeType(matcher.group(1));
+        Preconditions.checkState(MAP.equals(normalizedType), errorMsg);
+
+        String kvTypeString = matcher.group(2);
+        String[] kvTypeStringList = StringUtils.split(kvTypeString, ",");
+        final String mapTypeErrorMsg = "There can only be key and value two types in map declaration.";
+        Preconditions.checkState(kvTypeStringList.length == 2, mapTypeErrorMsg);
+        String keyTypeString = normalizeType(kvTypeStringList[0]);
+        String valueTypeString = normalizeType(kvTypeStringList[1]);
+        TypeInformation keyType = convertToAtomicType(keyTypeString);
+        TypeInformation valueType = convertToAtomicType(valueTypeString);
+        return Types.MAP(keyType, valueType);
+    }
+
     /**
      * 目前ROW里只支持基本类型
      * @param rowTypeString
@@ -104,6 +116,7 @@ public static RowTypeInfo convertToRow(String rowTypeString) {
         return new RowTypeInfo(info.f0, info.f1);
     }
 
+
     private static Tuple2<TypeInformation[], String[]> genFieldInfo(Iterable<String> fieldInfoStrs) {
         ArrayList<TypeInformation> types = Lists.newArrayList();
         ArrayList<String> fieldNames = Lists.newArrayList();

docs/plugin/kafkaSink.md

@@ -221,3 +221,29 @@ into
     from
             MyTable a
 ```
+## MAP类型示例
+目前Kafka Sink支持Map类型
+```sql
+CREATE TABLE ods(
+    id INT,
+    name STRING
+) WITH (
+    ...
+);
+
+CREATE TABLE dwd (
+    id INT,
+    dids MAP<STRING, INT>>
+) WITH (
+    type ='kafka',
+    bootstrapServers ='localhost:9092',
+    offsetReset ='latest',
+    groupId='wuren_foo',
+    topic ='luna_foo',
+    parallelism ='1'
+);
+
+INSERT INTO dwd
+    SELECT ods.id, MAP['foo', 1, 'bar', 2] AS dids
+    FROM ods;
+```

docs/pluginsInfo.md

@@ -1,9 +1,9 @@
 ### 1 插件列表
 #### 1.1 源表插件
 * [kafka 源表插件](plugin/kafkaSource.md)
-* [kafka 结果表插件](plugin/kafkaSink.md)
 
 #### 1.2 结果表插件
+* [kafka 结果表插件](plugin/kafkaSink.md)
 * [elasticsearch 结果表插件](plugin/elasticsearchSink.md)
 * [hbase 结果表插件](plugin/hbaseSink.md)
 * [mysql 结果表插件](plugin/mysqlSink.md)

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/AbstractKafkaSink.java

@@ -21,6 +21,7 @@
 import com.dtstack.flink.sql.enums.EUpdateMode;
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
 import com.dtstack.flink.sql.sink.kafka.table.KafkaSinkTableInfo;
+import com.dtstack.flink.sql.util.DataTypeUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -38,6 +39,7 @@
 import org.apache.flink.util.Preconditions;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 
+import java.util.HashMap;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.stream.IntStream;
@@ -77,11 +79,21 @@ protected Properties getKafkaProperties(KafkaSinkTableInfo KafkaSinkTableInfo) {
         }
         return props;
     }
-
+    // TODO Source有相同的方法日后可以合并
     protected TypeInformation[] getTypeInformations(KafkaSinkTableInfo kafka11SinkTableInfo) {
+        String[] fieldTypes = kafka11SinkTableInfo.getFieldTypes();
         Class<?>[] fieldClasses = kafka11SinkTableInfo.getFieldClasses();
         TypeInformation[] types = IntStream.range(0, fieldClasses.length)
-                .mapToObj(i -> TypeInformation.of(fieldClasses[i]))
+                .mapToObj(
+                    i -> {
+                        if (fieldClasses[i].isArray()) {
+                            return DataTypeUtils.convertToArray(fieldTypes[i]);
+                        }
+                        if (fieldClasses[i] == new HashMap().getClass()) {
+                            return DataTypeUtils.convertToMap(fieldTypes[i]);
+                        }
+                        return TypeInformation.of(fieldClasses[i]);
+                    })
                 .toArray(TypeInformation[]::new);
         return types;
     }