side parser always need nestJson as same as source

Author: jiemotongxue

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

@@ -38,9 +38,7 @@ public abstract class AbsSourceParser extends AbsTableParser {
     private static final String VIRTUAL_KEY = "virtualFieldKey";
     private static final String WATERMARK_KEY = "waterMarkKey";
     private static final String NOTNULL_KEY = "notNullKey";
-    private static final String NEST_JSON_FIELD_KEY = "nestFieldKey";
 
-    private static Pattern nestJsonFieldKeyPattern = Pattern.compile("(?i)((@*\\S+\\.)*\\S+)\\s+(\\w+)\\s+AS\\s+(\\w+)(\\s+NOT\\s+NULL)?$");
     private static Pattern virtualFieldKeyPattern = Pattern.compile("(?i)^(\\S+\\([^\\)]+\\))\\s+AS\\s+(\\w+)$");
     private static Pattern waterMarkKeyPattern = Pattern.compile("(?i)^\\s*WATERMARK\\s+FOR\\s+(\\S+)\\s+AS\\s+withOffset\\(\\s*(\\S+)\\s*,\\s*(\\d+)\\s*\\)$");
     private static Pattern notNullKeyPattern = Pattern.compile("(?i)^(\\w+)\\s+(\\w+)\\s+NOT\\s+NULL?$");
@@ -49,7 +47,6 @@ public AbsSourceParser() {
         addParserHandler(VIRTUAL_KEY, virtualFieldKeyPattern, this::dealVirtualField);
         addParserHandler(WATERMARK_KEY, waterMarkKeyPattern, this::dealWaterMark);
         addParserHandler(NOTNULL_KEY, notNullKeyPattern, this::dealNotNull);
-        addParserHandler(NEST_JSON_FIELD_KEY, nestJsonFieldKeyPattern, this::dealNestField);
     }
 
     protected void dealVirtualField(Matcher matcher, TableInfo tableInfo){
@@ -82,24 +79,4 @@ protected void dealNotNull(Matcher matcher, TableInfo tableInfo) {
         tableInfo.addFieldExtraInfo(fieldExtraInfo);
     }
 
-    /**
-     * add parser for alias field
-     * @param matcher
-     * @param tableInfo
-     */
-    protected void dealNestField(Matcher matcher, TableInfo tableInfo) {
-        String physicalField = matcher.group(1);
-        String fieldType = matcher.group(3);
-        String mappingField = matcher.group(4);
-        Class fieldClass= dbTypeConvertToJavaType(fieldType);
-        boolean notNull = matcher.group(5) != null;
-        TableInfo.FieldExtraInfo fieldExtraInfo = new TableInfo.FieldExtraInfo();
-        fieldExtraInfo.setNotNull(notNull);
-
-        tableInfo.addPhysicalMappings(mappingField, physicalField);
-        tableInfo.addField(mappingField);
-        tableInfo.addFieldClass(fieldClass);
-        tableInfo.addFieldType(fieldType);
-        tableInfo.addFieldExtraInfo(fieldExtraInfo);
-    }
 }

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

@@ -40,15 +40,18 @@
 public abstract class AbsTableParser {
 
     private static final String PRIMARY_KEY = "primaryKey";
+    private static final String NEST_JSON_FIELD_KEY = "nestFieldKey";
 
     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 Map<String, Pattern> patternMap = Maps.newHashMap();
 
     private Map<String, ITableFieldDealHandler> handlerMap = Maps.newHashMap();
 
     public AbsTableParser() {
         addParserHandler(PRIMARY_KEY, primaryKeyPattern, this::dealPrimaryKey);
+        addParserHandler(NEST_JSON_FIELD_KEY, nestJsonFieldKeyPattern, this::dealNestField);
     }
 
     protected boolean fieldNameNeedsUpperCase() {
@@ -116,6 +119,27 @@ public void dealPrimaryKey(Matcher matcher, TableInfo tableInfo){
         tableInfo.setPrimaryKeys(primaryKes);
     }
 
+    /**
+     * add parser for alias field
+     * @param matcher
+     * @param tableInfo
+     */
+    protected void dealNestField(Matcher matcher, TableInfo tableInfo) {
+        String physicalField = matcher.group(1);
+        String fieldType = matcher.group(3);
+        String mappingField = matcher.group(4);
+        Class fieldClass= dbTypeConvertToJavaType(fieldType);
+        boolean notNull = matcher.group(5) != null;
+        TableInfo.FieldExtraInfo fieldExtraInfo = new TableInfo.FieldExtraInfo();
+        fieldExtraInfo.setNotNull(notNull);
+
+        tableInfo.addPhysicalMappings(mappingField, physicalField);
+        tableInfo.addField(mappingField);
+        tableInfo.addFieldClass(fieldClass);
+        tableInfo.addFieldType(fieldType);
+        tableInfo.addFieldExtraInfo(fieldExtraInfo);
+    }
+
     public Class dbTypeConvertToJavaType(String fieldType) {
         return ClassUtil.stringConvertClass(fieldType);
     }