Merge branch '1.10_release_4.2.x' into 1.10_release

# Conflicts:
#	core/src/main/java/com/dtstack/flink/sql/table/AbstractTableInfo.java
#	core/src/main/java/com/dtstack/flink/sql/util/DtStringUtil.java
#	elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/Es6Util.java
#	impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaOutputFormat.java
#	pom.xml
#	rdb/rdb-core/src/main/java/com/dtstack/flink/sql/core/rdb/JdbcResourceCheck.java
#	rdb/rdb-core/src/main/java/com/dtstack/flink/sql/core/rdb/util/JdbcConnectionUtil.java
#	rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/JDBCUpsertOutputFormat.java
#	rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/table/RdbSinkParser.java

Author: FlechazoW

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -138,8 +138,7 @@ public static ParamsInfo parseParams(String[] args) throws Exception {
         }
 
         if (!pluginLoadMode.equalsIgnoreCase(EPluginLoadMode.LOCALTEST.name()) && Objects.isNull(dirtyProperties.get(PLUGIN_PATH_STR))) {
-            dirtyProperties.put(PLUGIN_PATH_STR,
-                    Objects.isNull(remoteSqlPluginPath) ? localSqlPluginPath : remoteSqlPluginPath);
+            dirtyProperties.put(PLUGIN_PATH_STR, localSqlPluginPath);
         }
 
         List<URL> jarUrlList = getExternalJarUrls(options.getAddjar());
@@ -179,7 +178,7 @@ public static StreamExecutionEnvironment getStreamExecution(ParamsInfo paramsInf
         StreamExecutionEnvironment env = ExecuteProcessHelper.getStreamExeEnv(paramsInfo.getConfProp(), paramsInfo.getDeployMode());
         StreamTableEnvironment tableEnv = getStreamTableEnv(env, paramsInfo.getConfProp());
 
-        ResourceCheck.NEED_CHECK = Boolean.parseBoolean(paramsInfo.getConfProp().getProperty(ResourceCheck.CHECK_STR, "true"));
+        ResourceCheck.NEED_CHECK = Boolean.parseBoolean(paramsInfo.getConfProp().getProperty(ResourceCheck.CHECK_STR, "false"));
 
         SqlParser.setLocalSqlPluginRoot(paramsInfo.getLocalSqlPluginPath());
         SqlTree sqlTree = SqlParser.parseSql(paramsInfo.getSql(), paramsInfo.getPluginLoadMode());

core/src/main/java/com/dtstack/flink/sql/resource/ResourceCheck.java

@@ -29,7 +29,7 @@
  * @description:资源检测
  **/
 public abstract class ResourceCheck {
-    public static Boolean NEED_CHECK = true;
+    public static Boolean NEED_CHECK = false;
     public static String CHECK_STR = "checkResource";
     protected static Logger LOG = LoggerFactory.getLogger(ResourceCheck.class);
     public String TABLE_TYPE_KEY = "tableType";

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

@@ -21,15 +21,13 @@
 import com.dtstack.flink.sql.dirtyManager.manager.DirtyKeys;
 import com.dtstack.flink.sql.outputformat.AbstractDtRichOutputFormat;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
-
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
 import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Properties;
 
 /**
  * Reason:
@@ -43,6 +41,9 @@ public abstract class AbstractTableInfo implements Serializable {
 
     public static final String PARALLELISM_KEY = "parallelism";
     public static final String ERROR_LIMIT = "errorLimit";
+    public static final Boolean DEFAULT_FALSE = false;
+    public static final Boolean DEFAULT_TRUE = true;
+    public static final Object DEFAULT_NULL = null;
     private final List<String> fieldList = Lists.newArrayList();
     private final List<String> fieldTypeList = Lists.newArrayList();
     private final List<Class> fieldClassList = Lists.newArrayList();

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

@@ -21,10 +21,11 @@
 package com.dtstack.flink.sql.util;
 
 import com.dtstack.flink.sql.enums.ColumnType;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Strings;
 import com.google.common.collect.Maps;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.flink.util.Preconditions;
 
 import java.math.BigDecimal;
 import java.sql.Timestamp;
@@ -46,7 +47,7 @@ public class DtStringUtil {
 
     private static final Pattern NO_VERSION_PATTERN = Pattern.compile("([a-zA-Z]+).*");
 
-    private static ObjectMapper objectMapper = new ObjectMapper();
+    private static final ObjectMapper objectMapper = new ObjectMapper();
 
     /**
      * Split the specified string delimiter --- ignored quotes delimiter
@@ -255,10 +256,12 @@ public static String col2string(Object column, String type) {
         return result.toString();
     }
 
-    public static String getPluginTypeWithoutVersion(String engineType){
+    public static String getPluginTypeWithoutVersion(String engineType) {
+        Preconditions.checkNotNull(engineType, "type can't be null!");
 
         Matcher matcher = NO_VERSION_PATTERN.matcher(engineType);
-        if(!matcher.find()){
+
+        if (!matcher.find()) {
             return engineType;
         }
 

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

@@ -263,4 +263,16 @@ public static Timestamp getTimestamp(Object obj) {
         throw new RuntimeException("not support type of " + obj.getClass() + " convert to Date.");
     }
 
+    public static Character getChar(Object obj) {
+        if (obj == null) {
+            return null;
+        }
+        if (obj instanceof Character) {
+            return (Character) obj;
+        }
+        if (obj instanceof String) {
+            return String.valueOf(obj).charAt(0);
+        }
+        throw new RuntimeException("not support type of " + obj.getClass() + " convert to Char");
+    }
 }

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

@@ -22,11 +22,9 @@
 
 import com.dtstack.flink.sql.dirtyManager.consumer.DirtyConsumerFactory;
 import com.dtstack.flink.sql.enums.EPluginLoadMode;
-import com.fasterxml.jackson.core.JsonGenerationException;
-import com.fasterxml.jackson.core.JsonParseException;
-import com.fasterxml.jackson.databind.JsonMappingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -128,7 +126,11 @@ public static Map<String,Object> objectToMap(Object obj) throws Exception{
         return objectMapper.readValue(objectMapper.writeValueAsBytes(obj), Map.class);
     }
 
-    public static <T> T jsonStrToObject(String jsonStr, Class<T> clazz) throws JsonParseException, JsonMappingException, JsonGenerationException, IOException{
+    public static String objectToString(Object obj) throws JsonProcessingException {
+        return objectMapper.writeValueAsString(obj);
+    }
+
+    public static <T> T jsonStrToObject(String jsonStr, Class<T> clazz) throws IOException{
         return  objectMapper.readValue(jsonStr, clazz);
     }
 

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

@@ -27,6 +27,7 @@
  */
 public class ThreadUtil {
     public static final Long DEFAULT_SLEEP_TIME = 10L;
+
     public static void sleepSeconds(long timeout) {
         try {
             TimeUnit.SECONDS.sleep(timeout);
@@ -50,4 +51,12 @@ public static void sleepMicroseconds(long timeout) {
             throw new RuntimeException(ie);
         }
     }
+
+    public static void sleepMilliseconds(long timeout) {
+        try {
+            TimeUnit.MILLISECONDS.sleep(timeout);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }

docs/plugin/filesource.md

@@ -0,0 +1,184 @@
+## 1.格式：
+
+```
+CREATE TABLE tableName(
+    colName colType,
+    ...
+ )WITH(
+    type ='file',
+    format = 'csv',
+    fieldDelimiter = ',',
+    fileName = 'xxxx',
+    filePath = 'xx/xxx',
+    location = 'local',
+    nullLiteral = 'null',
+    allowComment = 'true',
+    arrayElementDelimiter = ',',
+    quoteCharacter = '"',
+    escapeCharacter = '\\',
+    ignoreParseErrors = 'true',
+    hdfsSite = 'xxx/hdfs-site.xml',
+    coreSite = 'xxx/core-site.xml',
+    hdfsUser = 'root',
+    charsetName = 'UTF-8'
+ );
+```
+
+## 2.支持的格式
+
+支持 HDFS、 Local 支持 Csv、Json、Arvo 格式文件
+
+## 3.表结构定义
+
+|参数名称|含义|
+|----|---|
+| tableName | 在 sql 中使用的名称;即注册到flink-table-env上的名称|
+| colName | 列名称|
+| colType | 列类型 [colType支持的类型](../colType.md)|
+
+## 4.参数
+
+通用参数设置
+
+|参数名称|默认值|是否必填|参数说明|
+|----|---|---|---|
+|type|file|是|当前表的类型|
+|format|csv|是|文件格式，仅支持csv，json，Arvo类型|
+|fileName|无|是|文件名|
+|filePath|无|是|文件绝对路径|
+|location|local|是|文件存储介质，仅支持HDFS、Local|
+|charsetName|UTF-8|否|文件编码格式|
+
+### 4.1 Csv 参数设置
+
+|参数名称|默认值|是否必填|参数说明|
+|----|---|---|---|
+|ignoreParseErrors|true|否|是否忽略解析失败的数据|
+|fieldDelimiter|,|否|csv数据的分割符|
+|nullLiteral|"null"|否|填充csv数据中的null值|
+|allowComments|true|否||
+|arrayElementDelimiter|,|否||
+|quoteCharacter|"|否||
+|escapeCharacter|\|否||
+
+### 4.2 Arvo 参数说明
+
+|参数名称|默认值|是否必填|参数说明|
+|----|---|---|---|
+|avroFormat|无|是|在format = 'arvo'的情况下是必填项|
+
+### 4.3 HDFS 参数说明
+
+|参数名称|默认值|是否必填|参数说明|
+|----|---|---|---|
+|hdfsSite|${HADOOP_CONF_HOME}/hdfs-site.xml|是|hdfs-site.xml所在位置|
+|coreSite|${HADOOP_CONF_HOME}/core-site.xml|是|core-site.xml所在位置|
+|hdfsUser|root|否|HDFS访问用户，默认是[root]用户|
+
+### 4.4 Json 参数说明
+
+Json无特殊参数
+
+## 5.样例
+
+数据展示：
+
+csv
+
+```csv
+712382,1/1/2017 0:00,1/1/2017 0:03,223,7051,Wellesley St E / Yonge St Green P,7089,Church St  / Wood St,Member
+```
+
+json
+
+```json
+{
+  "trip_id": "712382",
+  "trip_start_time": "1/1/2017 0:00",
+  "trip_stop_time": "1/1/2017 0:03",
+  "trip_duration_seconds": "223",
+  "from_station_id": "7051",
+  "from_station_name": "Wellesley St E / Yonge St Green P",
+  "to_station_id": "7089",
+  "to_station_name": "Church St  / Wood St",
+  "user_type": "Member"
+},
+
+```
+
+### 5.1 csv
+
+```sql
+CREATE TABLE SourceOne
+(
+    trip_id               varchar,
+    trip_start_time       varchar,
+    trip_stop_time        varchar,
+    trip_duration_seconds varchar,
+    from_station_id       varchar,
+    from_station_name     varchar,
+    to_station_id         varchar,
+    to_station_name       varchar,
+    user_type             varchar
+) WITH (
+      type = 'file',
+      format = 'csv',
+      fieldDelimiter = ',',
+      fileName = '2017-Q1.csv',
+      filePath = '/data',
+      location = 'local',
+      charsetName = 'UTF-8'
+      );
+```
+
+### 5.2 json
+
+```sql
+CREATE TABLE SourceOne
+(
+    trip_id               varchar,
+    trip_start_time       varchar,
+    trip_stop_time        varchar,
+    trip_duration_seconds varchar,
+    from_station_id       varchar,
+    from_station_name     varchar,
+    to_station_id         varchar,
+    to_station_name       varchar,
+    user_type             varchar
+) WITH (
+      type = 'file',
+      format = 'json',
+      fieldDelimiter = ',',
+      fileName = '2017-Q1.json',
+      filePath = '/data',
+      charsetName = 'UTF-8'
+      );
+```
+
+### 5.3 HDFS
+
+```sql
+CREATE TABLE SourceOne
+(
+    trip_id               varchar,
+    trip_start_time       varchar,
+    trip_stop_time        varchar,
+    trip_duration_seconds varchar,
+    from_station_id       varchar,
+    from_station_name     varchar,
+    to_station_id         varchar,
+    to_station_name       varchar,
+    user_type             varchar
+) WITH (
+      type = 'file',
+      format = 'json',
+      fieldDelimiter = ',',
+      fileName = '2017-Q1.json',
+      filePath = 'hdfs://ns1/data',
+      location = 'hdfs',
+      hdfsSite = '/Users/wtz/dtstack/conf/yarn/kudu1/hdfs-site.xml',
+      coreSite = '/Users/wtz/dtstack/conf/yarn/kudu1/core-site.xml',
+      hdfsUser = 'admin',
+      charsetName = 'UTF-8'
+      );
+```