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

@@ -21,6 +21,7 @@
 import com.dtstack.flink.sql.util.DtStringUtil;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Maps;
+import org.apache.commons.lang3.StringUtils;
 
 import java.util.List;
 import java.util.Map;
@@ -74,7 +75,7 @@ private Map<String, Object> parseProp(String propsStr){
             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();
+            String value = DtStringUtil.removeStartAndEndQuota(ss.get(1).trim());
             propMap.put(key, value);
         }
 

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

@@ -23,15 +23,16 @@
 import com.dtstack.flink.sql.util.ClassUtil;
 import com.dtstack.flink.sql.util.DtStringUtil;
 import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.tuple.Tuple2;
 
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 /**
  * Reason:
@@ -142,9 +143,11 @@ private Tuple2<String, String> extractType(String fieldRow, String tableName) {
     }
 
     public void dealPrimaryKey(Matcher matcher, AbstractTableInfo tableInfo) {
-        String primaryFields = matcher.group(1).trim();
-        String[] splitArray = primaryFields.split(",");
-        List<String> primaryKeys = Lists.newArrayList(splitArray);
+        String primaryFields = matcher.group(2).trim();
+        List<String> primaryKeys = Arrays
+                .stream(primaryFields.split(","))
+                .map(String::trim)
+                .collect(Collectors.toList());;
         tableInfo.setPrimaryKeys(primaryKeys);
     }
 

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

@@ -373,8 +373,7 @@ public static String getTableFullPath(String schema, String tableName) {
             return addQuoteForStr(tableName);
         }
 
-        String schemaAndTabName = addQuoteForStr(schema) + "." + addQuoteForStr(tableName);
-        return schemaAndTabName;
+        return addQuoteForStr(schema) + "." + addQuoteForStr(tableName);
     }
 
     /**
@@ -407,4 +406,9 @@ public static String getStartQuote() {
     public static String getEndQuote() {
         return "\"";
     }
+
+    public static String removeStartAndEndQuota(String str) {
+        String removeStart = StringUtils.removeStart(str, "'");
+        return StringUtils.removeEnd(removeStart, "'");
+    }
 }