修改flink.1.10数据类型转换问题

Author: dapeng

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

@@ -24,6 +24,7 @@
 import com.dtstack.flink.sql.parser.InsertSqlParser;
 import com.dtstack.flink.sql.parser.SqlParser;
 import com.dtstack.flink.sql.parser.SqlTree;
+import com.dtstack.flink.sql.util.TypeInfoDataTypeConverter;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.streaming.api.datastream.DataStream;
@@ -62,6 +63,8 @@
 import org.apache.calcite.sql.SqlNode;
 import org.apache.commons.io.Charsets;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.utils.TypeConversions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -78,6 +81,7 @@
 import java.util.Set;
 import java.util.TimeZone;
 import java.util.ArrayList;
+import java.util.stream.Stream;
 
 /**
  *  任务执行时的流程方法
@@ -287,7 +291,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());
@@ -380,4 +384,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);
+    }
 }