修改 timestamp 转换问题

Author: xuchao

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

@@ -40,6 +40,7 @@
 import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -112,7 +113,7 @@ private void initMetric() {
 
 
     protected Object convertTimeIndictorTypeInfo(Integer index, Object obj) {
-        boolean isTimeIndicatorTypeInfo = DataTypes.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(index).getClass());
+        boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(index).getClass());
 
         //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) {

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

@@ -3,6 +3,7 @@
 import org.apache.flink.table.dataformat.BaseRow;
 import org.apache.flink.table.dataformat.BinaryString;
 import org.apache.flink.table.dataformat.GenericRow;
+import org.apache.flink.table.dataformat.SqlTimestamp;
 import org.apache.flink.types.Row;
 
 import java.sql.Timestamp;
@@ -23,7 +24,8 @@ public static BaseRow convertToBaseRow(Row row){
             if(row.getField(i) instanceof String){
                 genericRow.setField(i, BinaryString.fromString((String)row.getField(i)));
             } else if(row.getField(i) instanceof Timestamp){
-                genericRow.setField(i, ((Timestamp)row.getField(i)).getTime());
+                SqlTimestamp newTimestamp = SqlTimestamp.fromTimestamp(((Timestamp)row.getField(i)));
+                genericRow.setField(i, newTimestamp);
             }else{
                 genericRow.setField(i, row.getField(i));
             }