resolve conflicts

simenliuxing · simenliuxing · commit fa0a24227e96 · 2020-08-03T16:38:22.000+08:00
diff --git a/core/pom.xml b/core/pom.xml
@@ -105,43 +105,12 @@
             <version>${guava.version}</version>
         </dependency>
 
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-cep-scala_2.11</artifactId>
-            <version>${flink.version}</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-scala_2.11</artifactId>
-            <version>${flink.version}</version>
-        </dependency>
-
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-yarn_2.11</artifactId>
-            <version>${flink.version}</version>
-            <exclusions>
-                <exclusion>
-                    <groupId>org.apache.flink</groupId>
-                    <artifactId>flink-shaded-hadoop2</artifactId>
-                </exclusion>
-            </exclusions>
-        </dependency>
-
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-shaded-hadoop2</artifactId>
             <version>2.7.5-1.8.1</version>
         </dependency>
 
-        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-statebackend-rocksdb -->
-        <dependency>
-            <groupId>org.apache.flink</groupId>
-            <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
-            <version>${flink.version}</version>
-        </dependency>
-
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>
@@ -154,7 +123,6 @@
             <artifactId>joda-time</artifactId>
             <version>2.5</version>
         </dependency>
-
     </dependencies>
 
     <build>
diff --git a/core/src/main/java/com/dtstack/flink/sql/parser/CreateTmpTableParser.java b/core/src/main/java/com/dtstack/flink/sql/parser/CreateTmpTableParser.java
@@ -21,7 +21,12 @@
 package com.dtstack.flink.sql.parser;
 
 import com.dtstack.flink.sql.util.DtStringUtil;
-import org.apache.calcite.sql.*;
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlJoin;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.SqlMatchRecognize;
 import com.google.common.collect.Lists;
 import java.util.List;
 import java.util.regex.Matcher;
diff --git a/core/src/main/java/com/dtstack/flink/sql/table/AbstractTableParser.java b/core/src/main/java/com/dtstack/flink/sql/table/AbstractTableParser.java
@@ -16,6 +16,8 @@
  * limitations under the License.
  */
 
+
+
 package com.dtstack.flink.sql.table;
 
 import com.dtstack.flink.sql.util.ClassUtil;
@@ -44,7 +46,7 @@ public abstract class AbstractTableParser {
     private static final String CHAR_TYPE_NO_LENGTH = "CHAR";
 
     private static Pattern primaryKeyPattern = Pattern.compile("(?i)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 nestJsonFieldKeyPattern = Pattern.compile("(?i)((@*\\S+\\.)*\\S+)\\s+(\\w+)\\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*)\\)$");
 
@@ -82,14 +84,13 @@ public boolean dealKeyPattern(String fieldRow, AbstractTableInfo tableInfo){
         return false;
     }
 
-    public void parseFieldsInfo(String fieldsInfo, AbstractTableInfo tableInfo) {
+    public void parseFieldsInfo(String fieldsInfo, AbstractTableInfo tableInfo){
 
         List<String> fieldRows = DtStringUtil.splitIgnoreQuota(fieldsInfo, ',');
-
-        for (String fieldRow : fieldRows) {
+        for(String fieldRow : fieldRows){
             fieldRow = fieldRow.trim();
 
-            if (StringUtils.isBlank(fieldRow)) {
+            if(StringUtils.isBlank(fieldRow)){
                 throw new RuntimeException(String.format("table [%s],exists field empty.", tableInfo.getName()));
             }
 
@@ -131,7 +132,7 @@ public void parseFieldsInfo(String fieldsInfo, AbstractTableInfo tableInfo) {
         tableInfo.finish();
     }
 
-    public void dealPrimaryKey(Matcher matcher, AbstractTableInfo tableInfo) {
+    public void dealPrimaryKey(Matcher matcher, AbstractTableInfo tableInfo){
         String primaryFields = matcher.group(1).trim();
         String[] splitArry = primaryFields.split(",");
         List<String> primaryKes = Lists.newArrayList(splitArry);
@@ -170,5 +171,4 @@ protected void addParserHandler(String parserName, Pattern pattern, ITableFieldD
         patternMap.put(parserName, pattern);
         handlerMap.put(parserName, handler);
     }
-
 }
diff --git a/elasticsearch6/elasticsearch6-side/elasticsearch6-all-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AllReqRow.java b/elasticsearch6/elasticsearch6-side/elasticsearch6-all-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AllReqRow.java
@@ -29,7 +29,6 @@
 import com.google.common.collect.Maps;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.table.dataformat.BaseRow;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
@@ -107,11 +106,9 @@ public Row fillData(Row input, Object sideInput) {
 
             //Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
             if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
-                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
-                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
+                obj = ((Timestamp) obj).getTime();
             }
 
-
             row.setField(entry.getKey(), obj);
         }
 
diff --git a/elasticsearch6/elasticsearch6-side/elasticsearch6-async-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AsyncReqRow.java b/elasticsearch6/elasticsearch6-side/elasticsearch6-async-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AsyncReqRow.java
@@ -19,7 +19,6 @@
 package com.dtstack.flink.sql.side.elasticsearch6;
 
 import com.dtstack.flink.sql.util.RowDataComplete;
-import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
@@ -54,7 +53,6 @@
 import java.sql.Timestamp;
 import java.util.List;
 import java.util.Map;
-import java.util.TimeZone;
 
 /**
  * @author yinxi
@@ -63,7 +61,6 @@
 public class Elasticsearch6AsyncReqRow extends BaseAsyncReqRow implements Serializable {
 
     private static final Logger LOG = LoggerFactory.getLogger(Elasticsearch6AsyncReqRow.class);
-    private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
     private transient RestHighLevelClient rhlClient;
     private SearchRequest searchRequest;
     private List<String> sqlJoinCompareOperate = Lists.newArrayList();
@@ -194,8 +191,7 @@ public Row fillData(Row input, Object line) {
             Object obj = input.getField(entry.getValue());
             boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
             if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
-                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
-                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
+                obj = ((Timestamp) obj).getTime();
             }
 
             row.setField(entry.getKey(), obj);
diff --git a/hbase/hbase-side/hbase-all-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAllReqRow.java b/hbase/hbase-side/hbase-all-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAllReqRow.java
@@ -29,8 +29,6 @@
 import com.dtstack.flink.sql.util.RowDataComplete;
 import org.apache.calcite.sql.JoinType;
 import org.apache.commons.collections.map.HashedMap;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.commons.lang.StringUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import com.google.common.collect.Maps;
 import org.apache.flink.table.dataformat.BaseRow;
@@ -40,7 +38,6 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
-import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.ConnectionFactory;
@@ -53,7 +50,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.File;
 import java.io.IOException;
 
 import java.security.PrivilegedAction;
@@ -104,7 +100,6 @@ public Row fillData(Row input, Object sideInput) {
             if (obj instanceof LocalDateTime && isTimeIndicatorTypeInfo) {
                 obj = Timestamp.valueOf(((LocalDateTime) obj));
             }
-
             row.setField(entry.getKey(), obj);
         }
 
diff --git a/hbase/hbase-side/hbase-side-core/src/main/java/com/dtstack/flink/sql/side/hbase/utils/HbaseConfigUtils.java b/hbase/hbase-side/hbase-side-core/src/main/java/com/dtstack/flink/sql/side/hbase/utils/HbaseConfigUtils.java
@@ -33,7 +33,6 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import java.util.UUID;
 
 /**
  *
diff --git a/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseConfigUtils.java b/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseConfigUtils.java
@@ -19,7 +19,6 @@
 package com.dtstack.flink.sql.sink.hbase;
 
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.security.HadoopKerberosName;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authentication.util.KerberosName;
 import org.slf4j.Logger;
diff --git a/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/table/HbaseSinkParser.java b/hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/table/HbaseSinkParser.java
@@ -129,7 +129,7 @@ public void parseFieldsInfo(String fieldsInfo, HbaseTableInfo tableInfo){
     }
 
     private Map<String, String> parseColumnFamily(Map<String, String> physicalFieldMap){
-        Map<String, String> columnFamiles = new LinkedHashMap<>();
+        Map<String, String> columnFamiles = Maps.newHashMap();
         physicalFieldMap.values().forEach(x -> {
             String[] columnFamily = StringUtils.split(x.trim(), ":");
             columnFamiles.put(x, columnFamily[1]);
diff --git a/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/AbstractKafkaSource.java b/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/AbstractKafkaSource.java
@@ -21,7 +21,6 @@
 import com.dtstack.flink.sql.source.IStreamSourceGener;
 import com.dtstack.flink.sql.source.kafka.enums.EKafkaOffset;
 import com.dtstack.flink.sql.source.kafka.table.KafkaSourceTableInfo;
-import com.dtstack.flink.sql.util.DataTypeUtils;
 import com.dtstack.flink.sql.util.DtStringUtil;
 import com.dtstack.flink.sql.util.PluginUtil;
 import org.apache.commons.lang3.StringUtils;
@@ -34,7 +33,6 @@
 import org.apache.flink.types.Row;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 
-import java.lang.reflect.Array;
 import java.util.Map;
 import java.util.Properties;
 import java.util.stream.Collectors;
@@ -74,16 +72,9 @@ protected String generateOperatorName(String tabName, String topicName) {
     }
 
     protected TypeInformation<Row> getRowTypeInformation(KafkaSourceTableInfo kafkaSourceTableInfo) {
-        String[] fieldTypes = kafkaSourceTableInfo.getFieldTypes();
         Class<?>[] fieldClasses = kafkaSourceTableInfo.getFieldClasses();
-        TypeInformation[] types =
-                IntStream.range(0, fieldClasses.length)
-                .mapToObj(i -> {
-                    if (fieldClasses[i].isArray()) {
-                        return DataTypeUtils.convertToArray(fieldTypes[i]);
-                    }
-                    return TypeInformation.of(fieldClasses[i]);
-                })
+        TypeInformation[] types = IntStream.range(0, fieldClasses.length)
+                .mapToObj(i -> TypeInformation.of(fieldClasses[i]))
                 .toArray(TypeInformation[]::new);
 
         return new RowTypeInfo(types, kafkaSourceTableInfo.getFields());
diff --git a/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java b/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java
@@ -50,6 +50,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         kafkaSourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
         kafkaSourceTableInfo.setTopicIsPattern(MathUtil.getBoolean(props.get(KafkaSourceTableInfo.TOPICISPATTERN_KEY.toLowerCase()), false));
         kafkaSourceTableInfo.setOffsetReset(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase(), EKafkaOffset.LATEST.name().toLowerCase())));
+        kafkaSourceTableInfo.setCharsetName(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.CHARSET_NAME_KEY.toLowerCase(),"UTF-8")));
 
         kafkaSourceTableInfo.setSchemaString(MathUtil.getString(props.get(KafkaSourceTableInfo.SCHEMA_STRING_KEY.toLowerCase())));
         kafkaSourceTableInfo.setFieldDelimiter(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.CSV_FIELD_DELIMITER_KEY.toLowerCase(), "|")));
diff --git a/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceTableInfo.java b/kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceTableInfo.java
@@ -19,7 +19,6 @@
 
 package com.dtstack.flink.sql.source.kafka.table;
 
-import com.dtstack.flink.sql.format.FormatType;
 import com.dtstack.flink.sql.table.AbstractSourceTableInfo;
 import com.google.common.base.Preconditions;
 
diff --git a/rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/JDBCTypeConvertUtils.java b/rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/JDBCTypeConvertUtils.java
@@ -18,7 +18,6 @@
 
 package com.dtstack.flink.sql.sink.rdb;
 
-import com.google.common.collect.Lists;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
diff --git a/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java b/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java
diff --git a/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/table/RedisSinkParser.java b/redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/table/RedisSinkParser.java

Original file line number	Diff line number	Diff line change
`@@ -33,7 +33,6 @@`
`33`	`33`	`import java.util.Arrays;`
`34`	`34`	`import java.util.List;`
`35`	`35`	`import java.util.Map;`
`36`		`-import java.util.UUID;`
`37`	`36`
`38`	`37`	`/**`
`39`	`38`	`*`
Original file line number	Diff line number	Diff line change
`@@ -129,7 +129,7 @@ public void parseFieldsInfo(String fieldsInfo, HbaseTableInfo tableInfo){`
`129`	`129`	`}`
`130`	`130`
`131`	`131`	`private Map<String, String> parseColumnFamily(Map<String, String> physicalFieldMap){`
`132`		`- Map<String, String> columnFamiles = new LinkedHashMap<>();`
	`132`	`+ Map<String, String> columnFamiles = Maps.newHashMap();`
`133`	`133`	`physicalFieldMap.values().forEach(x -> {`
`134`	`134`	`String[] columnFamily = StringUtils.split(x.trim(), ":");`
`135`	`135`	`columnFamiles.put(x, columnFamily[1]);`