Merge remote-tracking branch 'origin/1.10_release_4.0.x' into 1.10_release_4.0.x

Author: simenliuxing

core/src/main/java/com/dtstack/flink/sql/parser/CreateTableParser.java

@@ -19,6 +19,7 @@
 package com.dtstack.flink.sql.parser;
 
 import com.dtstack.flink.sql.util.DtStringUtil;
+import com.google.common.base.Preconditions;
 import com.google.common.collect.Maps;
 
 import java.util.List;
@@ -71,6 +72,7 @@ private Map<String, Object> parseProp(String propsStr){
         Map<String, Object> propMap = Maps.newHashMap();
         for (String str : strings) {
             List<String> ss = DtStringUtil.splitIgnoreQuota(str, '=');
+            Preconditions.checkState(ss.size() == 2, str + " Format error");
             String key = ss.get(0).trim();
             String value = ss.get(1).trim().replaceAll("'", "").trim();
             propMap.put(key, value);

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

@@ -46,7 +46,7 @@ public abstract class AbstractTableParser {
     private static final String NEST_JSON_FIELD_KEY = "nestFieldKey";
     private static final String CHAR_TYPE_NO_LENGTH = "CHAR";
 
-    private static Pattern primaryKeyPattern = Pattern.compile("(?i)PRIMARY\\s+KEY\\s*\\((.*)\\)");
+    private static Pattern primaryKeyPattern = Pattern.compile("(?i)(^\\s*)PRIMARY\\s+KEY\\s*\\((.*)\\)");
     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*)\\)$");