kafka array analysis

Author: simenliuxing

core/src/main/java/com/dtstack/flink/sql/table/AbstractTableParser.java

@@ -46,7 +46,7 @@ public abstract class AbstractTableParser {
     private static final String CHAR_TYPE_NO_LENGTH = "CHAR";
 
     private static Pattern primaryKeyPattern = Pattern.compile("(?i)PRIMARY\\s+KEY\\s*\\((.*)\\)");
-    private static Pattern nestJsonFieldKeyPattern = Pattern.compile("(?i)((@*\\S+\\.)*\\S+)\\s+(\\w+)\\s+AS\\s+(\\w+)(\\s+NOT\\s+NULL)?$");
+    private static Pattern nestJsonFieldKeyPattern = Pattern.compile("(?i)((@*\\S+\\.)*\\S+)\\s+(.+)\\s+AS\\s+(\\w+)(\\s+NOT\\s+NULL)?$");
     private static Pattern physicalFieldFunPattern = Pattern.compile("\\w+\\((\\w+)\\)$");
     private static Pattern charTypePattern = Pattern.compile("(?i)CHAR\\((\\d*)\\)$");
 

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

@@ -21,6 +21,7 @@
 import com.dtstack.flink.sql.source.IStreamSourceGener;
 import com.dtstack.flink.sql.source.kafka.enums.EKafkaOffset;
 import com.dtstack.flink.sql.source.kafka.table.KafkaSourceTableInfo;
+import com.dtstack.flink.sql.util.DataTypeUtils;
 import com.dtstack.flink.sql.util.DtStringUtil;
 import com.dtstack.flink.sql.util.PluginUtil;
 import org.apache.commons.lang3.StringUtils;
@@ -72,10 +73,18 @@ protected String generateOperatorName(String tabName, String topicName) {
     }
 
     protected TypeInformation<Row> getRowTypeInformation(KafkaSourceTableInfo kafkaSourceTableInfo) {
+        String[] fieldTypes = kafkaSourceTableInfo.getFieldTypes();
         Class<?>[] fieldClasses = kafkaSourceTableInfo.getFieldClasses();
-        TypeInformation[] types = IntStream.range(0, fieldClasses.length)
-                .mapToObj(i -> TypeInformation.of(fieldClasses[i]))
-                .toArray(TypeInformation[]::new);
+        TypeInformation[] types =
+                IntStream.range(0, fieldClasses.length)
+                        .mapToObj(i -> {
+                            if (fieldClasses[i].isArray()) {
+                                return DataTypeUtils.convertToArray(fieldTypes[i]);
+                            }
+                            return TypeInformation.of(fieldClasses[i]);
+                        })
+                        .toArray(TypeInformation[]::new);
+
 
         return new RowTypeInfo(types, kafkaSourceTableInfo.getFields());
     }