Merge pull request DTStack#397 from kyo-tom/hotfix_1.10_release_396

[hotfix-36211][ftp] 解决windows下中文目录的文件读取不到问题
[hotfix-36332][flinkx-core][hivewriter]解决gsonutil解析到18446744073702197459出错
[hotfix-36332][hive]修改hive解析json失败报错
[hotfix-36431][logminer]logminer支持解析toDate函数
[hotfix-36432][logminer]logminer 增加3f207768657265(? where)场景解析支持
[hotfix-387][logminer][docs] add standaLoneQuickStart docs
[hotfix-36476][kudureader]支持column配置为常量
[hotfix-36591][logminer]事务回滚日志delete where row_id=''转为正常业务字段,修改缓存逻辑，加载的日志文件从V$LOGMNR_LOGS查找出来并打印，加载日志文件SQL_START_LOGMINER_HAS_MAX_LIMIT 查找实时日志文件where条件修改,disConnect增加非空判断
[hotfix-36356][ftp]解决ftp在windows写入问题

Author: kanata163

flinkx-core/src/main/java/com/dtstack/flinkx/authenticate/KerberosUtil.java

@@ -148,6 +148,10 @@ public static String loadFile(Map<String, Object> kerberosConfig, String filePat
             LOG.info("will use local file:{}", filePath);
             checkFileExists(filePath);
         } else {
+            if(filePath.contains(SP)){
+                filePath = filePath.substring(filePath.lastIndexOf(SP) + 1);
+            }
+
             filePath = loadFromSftp(kerberosConfig, filePath);
         }
 
@@ -172,9 +176,6 @@ private static String loadFromSftp(Map<String, Object> config, String fileName){
         String localDirName = Md5Util.getMd5(remoteDir);
         String localDir = LOCAL_CACHE_DIR + SP + localDirName;
         localDir = createDir(localDir);
-        if (fileName.contains(SP)) {
-            fileName = fileName.substring(fileName.lastIndexOf(SP) + 1);
-        }
         String fileLocalPath = localDir + SP + fileName;
         // 更新sftp文件对应的local文件
         if (fileExists(fileLocalPath)) {

flinkx-core/src/main/java/com/dtstack/flinkx/decoder/JsonDecoder.java

@@ -17,7 +17,7 @@
  */
 package com.dtstack.flinkx.decoder;
 
-import com.dtstack.flinkx.util.GsonUtil;
+import com.dtstack.flinkx.util.MapUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -38,7 +38,7 @@ public class JsonDecoder implements IDecode {
     @Override
     public Map<String, Object> decode(final String message) {
         try {
-            Map<String, Object> event = GsonUtil.GSON.fromJson(message, GsonUtil.gsonMapTypeToken);
+            Map<String, Object> event =  MapUtil.jsonStrToObject(message,Map.class);
             if (!event.containsKey(KEY_MESSAGE)) {
                 event.put(KEY_MESSAGE, message);
             }

flinkx-core/src/main/java/com/dtstack/flinkx/util/GsonUtil.java

@@ -33,6 +33,7 @@
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.lang.reflect.Type;
+import java.math.BigInteger;
 import java.util.*;
 
 /**
@@ -107,7 +108,11 @@ public Object read(JsonReader in) throws IOException {
                                                     try {
                                                         return Integer.valueOf(s);
                                                     } catch (Exception e) {
-                                                        return Long.valueOf(s);
+                                                        try{
+                                                            return Long.valueOf(s);
+                                                        }catch (Exception e1){
+                                                            return new BigInteger(s);
+                                                        }
                                                     }
                                                 }
                                             case BOOLEAN:

flinkx-core/src/main/java/com/dtstack/flinkx/util/MapUtil.java

@@ -7,11 +7,15 @@
 import com.google.gson.internal.LinkedHashTreeMap;
 import com.google.gson.internal.LinkedTreeMap;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang3.StringUtils;
 
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 
+import static com.dtstack.flinkx.util.StringUtil.escapeExprSpecialWord;
+
 /**
  * Reason:
  * Date: 2019/8/9
@@ -61,4 +65,89 @@ public static String writeValueAsString(Object obj) throws JsonProcessingExcepti
         return objectMapper.writeValueAsString(obj);
     }
 
+    /**
+     * 根据key 以及切割键 获取真正的key，将key 和value放入data中
+     * @param key key
+     * @param fieldDelimiter 切割键
+     * @param value 值
+     * @param data 载体data
+     */
+    public static void buildMap(String key, String fieldDelimiter, Object value, Map<String, Object> data) {
+        String[] split = new String[1];
+        if (StringUtils.isBlank(fieldDelimiter)) {
+            split[0] = key;
+        } else {
+            split = key.split(escapeExprSpecialWord(fieldDelimiter));
+        }
+
+        if (split.length == 1) {
+            data.put(split[0], value);
+        } else {
+            Map<String, Object> temp = data;
+            for (int i = 0; i < split.length - 1; i++) {
+                if (temp.containsKey(split[i])) {
+                    if (temp.get(split[i]) instanceof HashMap) {
+                        temp = (HashMap) temp.get(split[i]);
+                    } else {
+                        throw new RuntimeException("build map failed ,data is " + GsonUtil.GSON.toJson(data) + " key is " + key);
+                    }
+                } else {
+                    Map hashMap = new HashMap(2);
+                    temp.put(split[i], hashMap);
+                    temp = hashMap;
+                }
+                if (i == split.length - 2) {
+                    temp.put(split[split.length - 1], value);
+                }
+            }
+        }
+    }
+
+
+    /**
+     * 根据指定的key从map里获取对应的值
+     * 如果key不存在 报错
+     *
+     * @param map            需要解析的map
+     * @param key            指定的key  key可以是嵌套的
+     * @param fieldDelimiter 嵌套key的分隔符
+     */
+    public static Object getValueByKey(Map<String, Object> map, String key, String fieldDelimiter) {
+        if (MapUtils.isEmpty(map)) {
+            throw new RuntimeException(key + " not exist  because map is empty");
+        }
+        Object o = null;
+        String[] split = new String[1];
+        if (StringUtils.isBlank(fieldDelimiter)) {
+            split[0] = key;
+        } else {
+            split = key.split(escapeExprSpecialWord(fieldDelimiter));
+        }
+
+        Map<String, Object> tempMap = map;
+        for (int i = 0; i < split.length; i++) {
+            o = getValue(tempMap, split[i]);
+            //仅仅代表这个key对应的值是null但是key还是存在的
+            if (o == null && i != split.length - 1) {
+                throw new RuntimeException(key + " on  [" + GsonUtil.GSON.toJson(map) + "]  is null");
+            }
+
+            if (i != split.length - 1) {
+                if (!(o instanceof Map)) {
+                    throw new RuntimeException("key " + key + " on " + map + " is not a json");
+                }
+                tempMap = (Map<String, Object>) o;
+            }
+        }
+        return o;
+    }
+
+
+    private static Object getValue(Map<String, Object> map, String key) {
+        if (!map.containsKey(key)) {
+            throw new RuntimeException(key + " not exist on  " + GsonUtil.GSON.toJson(map) );
+        }
+        return map.get(key);
+    }
+
 }

flinkx-core/src/main/java/com/dtstack/flinkx/util/StringUtil.java

@@ -96,6 +96,7 @@ public static Object string2col(String str, String type, SimpleDateFormat custom
                 break;
             case MEDIUMINT:
             case BIGINT:
+            case LONG:
                 ret = Long.valueOf(str.trim());
                 break;
             case FLOAT:
@@ -339,4 +340,23 @@ public static String splitIgnoreQuotaAndJoinByPoint(String table) {
         }
         return stringBuffer.toString();
     }
+
+
+    /**
+     * 转义正则特殊字符 （$()*+.[]?\^{},|）
+     *
+     * @param keyword 需要转义特殊字符串的文本
+     * @return 特殊字符串转义后的文本
+     */
+    public static String escapeExprSpecialWord(String keyword) {
+        if (StringUtils.isNotBlank(keyword)) {
+            String[] fbsArr = {"\\", "$", "(", ")", "*", "+", ".", "[", "]", "?", "^", "{", "}", "|"};
+            for (String key : fbsArr) {
+                if (keyword.contains(key)) {
+                    keyword = keyword.replace(key, "\\" + key);
+                }
+            }
+        }
+        return keyword;
+    }
 }

flinkx-ftp/flinkx-ftp-core/src/main/java/com/dtstack/flinkx/ftp/FtpConfig.java

@@ -19,6 +19,7 @@
 
 package com.dtstack.flinkx.ftp;
 
+import com.dtstack.flinkx.constants.ConstantValue;
 import org.codehaus.jackson.annotate.JsonIgnoreProperties;
 
 import java.io.Serializable;
@@ -60,6 +61,12 @@ public class FtpConfig implements Serializable {
 
     public long maxFileSize = 1024 * 1024 * 1024L;
 
+    /** ftp客户端编码格式 **/
+    public String controlEncoding = System.getProperty(ConstantValue.SYSTEM_PROPERTIES_KEY_FILE_ENCODING);
+
+    /** linux是否展示隐藏文件 **/
+    public boolean listHiddenFiles = true;
+
     public String getUsername() {
         return username;
     }
@@ -179,4 +186,20 @@ public long getMaxFileSize() {
     public void setMaxFileSize(long maxFileSize) {
         this.maxFileSize = maxFileSize;
     }
+
+    public String getControlEncoding() {
+        return controlEncoding;
+    }
+
+    public void setControlEncoding(String controlEncoding) {
+        this.controlEncoding = controlEncoding;
+    }
+
+    public boolean getListHiddenFiles() {
+        return listHiddenFiles;
+    }
+
+    public void setListHiddenFiles(boolean listHiddenFiles) {
+        this.listHiddenFiles = listHiddenFiles;
+    }
 }