merge 1.8

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java

@@ -45,7 +45,6 @@
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.commons.collections.CollectionUtils;
-import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
@@ -54,13 +53,14 @@
 import com.google.common.collect.Maps;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.table.api.Table;
-import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
@@ -128,7 +128,6 @@ public void exec(String sql, Map<String, SideTableInfo> sideTableMap, StreamTabl
                     System.out.println("----------real exec sql-----------" );
                     System.out.println(pollSqlNode.toString());
                     FlinkSQLExec.sqlUpdate(tableEnv, pollSqlNode.toString());
-//                    tableEnv.sqlUpdate(pollSqlNode.toString());
                     if(LOG.isInfoEnabled()){
                         LOG.info("exec sql: " + pollSqlNode.toString());
                     }
@@ -160,15 +159,12 @@ private void addAliasForFieldNode(SqlNode pollSqlNode, List<String> fieldList, H
                 SqlNode source = ((SqlInsert) pollSqlNode).getSource();
                 addAliasForFieldNode(source, fieldList, mappingTable);
                 break;
-
             case AS:
                 addAliasForFieldNode(((SqlBasicCall) pollSqlNode).getOperands()[0], fieldList, mappingTable);
                 break;
 
             case SELECT:
-
                 SqlNodeList selectList = ((SqlSelect) pollSqlNode).getSelectList();
-
                 selectList.getList().forEach(node -> {
                     if (node.getKind() == IDENTIFIER) {
                         SqlIdentifier sqlIdentifier = (SqlIdentifier) node;
@@ -183,15 +179,13 @@ private void addAliasForFieldNode(SqlNode pollSqlNode, List<String> fieldList, H
 
                     }
                 });
-
                 for (int i = 0; i < selectList.getList().size(); i++) {
                     SqlNode node = selectList.get(i);
                     if (node.getKind() == IDENTIFIER) {
                         SqlIdentifier sqlIdentifier = (SqlIdentifier) node;
                         if (sqlIdentifier.names.size() == 1) {
                             return;
                         }
-
                         String name = sqlIdentifier.names.get(1);
                         // avoid real field pv0 convert pv
                         if (name.endsWith("0") &&  !fieldList.contains(name) && !fieldList.contains(name.substring(0, name.length() - 1))) {
@@ -268,7 +262,7 @@ private RowTypeInfo buildLeftTableOutType(RowTypeInfo leftTypeInfo) {
 
     private TypeInformation convertTimeAttributeType(TypeInformation typeInformation) {
         if (typeInformation instanceof TimeIndicatorTypeInfo) {
-            return TypeInformation.of(Timestamp.class);
+            return TypeInformation.of(LocalDateTime.class);
         }
         return typeInformation;
     }

impala/pom.xml

@@ -42,6 +42,12 @@
             <version>2.6.12</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-common</artifactId>
+            <version>2.7.3</version>
+        </dependency>
+
     </dependencies>
 
 

kafka/kafka-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java

@@ -56,10 +56,12 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
     private static final long serialVersionUID = 2385115520960444192L;
 
     private AbstractFetcher<Row, ?> fetcher;
+    private TypeInformation<Row> typeInfo;
 
     private boolean firstMsg = true;
 
     public CustomerJsonDeserialization(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping, List<TableInfo.FieldExtraInfo> fieldExtraInfos){
+        this.typeInfo = typeInfo;
         this.jsonDataParser= new JsonDataParser(typeInfo,rowAndFieldMapping,fieldExtraInfos);
     }
 

kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java

@@ -60,10 +60,12 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
     private static final long serialVersionUID = 2385115520960444192L;
 
     private AbstractFetcher<Row, ?> fetcher;
+    private TypeInformation<Row> typeInfo;
 
     private boolean firstMsg = true;
 
     public CustomerJsonDeserialization(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping, List<TableInfo.FieldExtraInfo> fieldExtraInfos){
+        this.typeInfo = typeInfo;
         this.jsonDataParser= new JsonDataParser(typeInfo,rowAndFieldMapping,fieldExtraInfos);
     }
 
@@ -120,6 +122,11 @@ private static String partitionLagMetricName(TopicPartition tp) {
         return tp + ".records-lag";
     }
 
+    public void setFetcher(AbstractFetcher<Row, ?> fetcher) {
+        this.fetcher = fetcher;
+    }
+
+
 
     @Override
     public TypeInformation<Row> getProducedType() {

kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java

@@ -60,10 +60,12 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
     private static final long serialVersionUID = 2385115520960444192L;
 
     private AbstractFetcher<Row, ?> fetcher;
+    private TypeInformation<Row> typeInfo;
 
     private boolean firstMsg = true;
 
     public CustomerJsonDeserialization(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping, List<TableInfo.FieldExtraInfo> fieldExtraInfos){
+        this.typeInfo = typeInfo;
         this.jsonDataParser= new JsonDataParser(typeInfo,rowAndFieldMapping,fieldExtraInfos);
     }
 

kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java

@@ -60,11 +60,13 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
     private static final long serialVersionUID = 2385115520960444192L;
 
     private AbstractFetcher<Row, ?> fetcher;
+    private TypeInformation<Row> typeInfo;
 
     private boolean firstMsg = true;
 
-    public CustomerJsonDeserialization(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping, List<TableInfo.FieldExtraInfo> fieldExtraInfos){
-        this.jsonDataParser= new JsonDataParser(typeInfo,rowAndFieldMapping,fieldExtraInfos);
+    public CustomerJsonDeserialization(TypeInformation<Row> typeInfo, Map<String, String> rowAndFieldMapping, List<TableInfo.FieldExtraInfo> fieldExtraInfos) {
+        this.typeInfo = typeInfo;
+        this.jsonDataParser = new JsonDataParser(typeInfo, rowAndFieldMapping, fieldExtraInfos);
     }
 
     @Override