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/exec/ExecuteProcessHelper.java

@@ -42,6 +42,7 @@
 import com.dtstack.flink.sql.table.AbstractTargetTableInfo;
 import com.dtstack.flink.sql.util.DtStringUtil;
 import com.dtstack.flink.sql.util.PluginUtil;
+import com.dtstack.flink.sql.util.TypeInfoDataTypeConverter;
 import com.dtstack.flink.sql.watermarker.WaterMarkerAssigner;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Preconditions;
@@ -64,6 +65,7 @@
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 import org.apache.flink.table.api.java.internal.StreamTableEnvironmentImpl;
 import org.apache.flink.table.sinks.TableSink;
+import org.apache.flink.table.types.DataType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -80,6 +82,7 @@
 import java.util.Properties;
 import java.util.Set;
 import java.util.TimeZone;
+import java.util.stream.Stream;
 
 /**
  *  任务执行时的流程方法
@@ -295,7 +298,7 @@ public static Set<URL> registerTable(SqlTree sqlTree, StreamExecutionEnvironment
                 String adaptSql = sourceTableInfo.getAdaptSelectSql();
                 Table adaptTable = adaptSql == null ? table : tableEnv.sqlQuery(adaptSql);
 
-                RowTypeInfo typeInfo = new RowTypeInfo(adaptTable.getSchema().getFieldTypes(), adaptTable.getSchema().getFieldNames());
+                RowTypeInfo typeInfo = new RowTypeInfo(fromDataTypeToLegacyInfo(adaptTable.getSchema().getFieldDataTypes()), adaptTable.getSchema().getFieldNames());
                 DataStream adaptStream = tableEnv.toAppendStream(adaptTable, typeInfo);
 
                 String fields = String.join(",", typeInfo.getFieldNames());
@@ -389,4 +392,10 @@ private static void timeZoneCheck(String timeZone) {
             throw new IllegalArgumentException(String.format(" timezone of %s is Incorrect!", timeZone));
         }
     }
+
+    private static TypeInformation<?>[] fromDataTypeToLegacyInfo(DataType[] dataType) {
+        return Stream.of(dataType)
+                .map(TypeInfoDataTypeConverter::toLegacyTypeInfo)
+                .toArray(TypeInformation[]::new);
+    }
 }

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

@@ -37,6 +37,7 @@
 import java.time.LocalDateTime;
 import java.util.Map;
 import java.util.TimeZone;
+import java.util.TimeZone;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -45,6 +46,7 @@
  * Reason:
  * Date: 2018/9/18
  * Company: www.dtstack.com
+ *
  * @author xuchao
  */
 
@@ -86,7 +88,8 @@ protected Object convertTimeIndictorTypeInfo(Integer index, Object obj) {
 
         //Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
         if (obj instanceof LocalDateTime && isTimeIndicatorTypeInfo) {
-            obj = Timestamp.valueOf(((LocalDateTime) obj));
+            //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+            obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
         }
         return obj;
     }

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

@@ -52,6 +52,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.TimeZone;
 import java.util.concurrent.ScheduledFuture;
 
 /**
@@ -71,6 +72,7 @@ public abstract class BaseAsyncReqRow extends RichAsyncFunction<Row, BaseRow> im
     private int timeOutNum = 0;
     protected BaseSideInfo sideInfo;
     protected transient Counter parseErrorRecords;
+    private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
 
     public BaseAsyncReqRow(BaseSideInfo sideInfo) {
         this.sideInfo = sideInfo;
@@ -117,7 +119,8 @@ protected Object convertTimeIndictorTypeInfo(Integer index, Object obj) {
 
         //Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
         if (obj instanceof LocalDateTime && isTimeIndicatorTypeInfo) {
-            obj = Timestamp.valueOf(((LocalDateTime) obj));
+            //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+            obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
         }
         return obj;
     }

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

@@ -16,14 +16,18 @@
  * limitations under the License.
  */
 
- 
+
 
 package com.dtstack.flink.sql.table;
 
 import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import org.apache.flink.util.StringUtils;
 
+import java.util.ArrayList;
 import java.util.Map;
+import java.util.TimeZone;
 
 /**
  * Reason:
@@ -36,6 +40,10 @@ public abstract class AbstractSourceTableInfo extends AbstractTableInfo {
 
     public static final String SOURCE_SUFFIX = "Source";
 
+    public static final String TIME_ZONE_KEY = "timezone";
+
+    private String timeZone = TimeZone.getDefault().getID();
+
     private String eventTimeField;
 
     private Integer maxOutOrderness = 10;
@@ -101,4 +109,23 @@ public String getAdaptSelectSql(){
     public String getAdaptName(){
         return getName() + "_adapt";
     }
+
+    public String getTimeZone() {
+        return timeZone;
+    }
+
+    public void setTimeZone(String timeZone) {
+        if (StringUtils.isNullOrWhitespaceOnly(timeZone)){
+            return;
+        }
+        timeZoneCheck(timeZone);
+        this.timeZone = timeZone;
+    }
+
+    private void timeZoneCheck(String timeZone) {
+        ArrayList<String> zones = Lists.newArrayList(TimeZone.getAvailableIDs());
+        if (!zones.contains(timeZone)){
+            throw  new IllegalArgumentException(" timezone is Incorrect!");
+        }
+    }
 }