[fix] time transform error in view

Author: FlechazoW

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

@@ -409,10 +409,9 @@ private void joinFun(Object pollObj,
 
         RowTypeInfo typeInfo = new RowTypeInfo(targetTable.getSchema().getFieldTypes(), targetTable.getSchema().getFieldNames());
 
-        DataStream adaptStream = tableEnv.toRetractStream(targetTable, Row.class)
+        DataStream adaptStream = tableEnv.toRetractStream(targetTable, typeInfo)
                 .filter(f -> f.f0)
-                .map(f -> f.f1)
-                .returns(Row.class);
+                .map(f -> f.f1);
 
         //join side table before keyby ===> Reducing the size of each dimension table cache of async
         if (sideTableInfo.isPartitionedJoin()) {

localTest/src/main/java/com/dtstack/flink/sql/localTest/LocalTest.java

@@ -50,7 +50,7 @@ public static void main(String[] args) throws Exception {
         setLogLevel("INFO");
 
         List<String> propertiesList = new ArrayList<>();
-        String sqlPath = "/Users/wtz4680/Desktop/flinkStreamSQL/sql/rowtimeprint.sql";
+        String sqlPath = "/Users/wtz/dtstack/job/flinkStreamSQL/temp2.sql";
         Map<String, Object> conf = new HashMap<>();
         JSONObject properties = new JSONObject();
 

pom.xml

@@ -36,7 +36,7 @@
         <module>polardb</module>
         <module>oceanbase</module>
         <module>tidb</module>
-        <!--        <module>localTest</module>-->
+        <module>localTest</module>
 
     </modules>
 

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java

@@ -48,7 +48,10 @@
 import java.time.Instant;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -117,7 +120,8 @@ protected void preInvoke(Row input, ResultFuture<BaseRow> resultFuture) {
     public void handleAsyncInvoke(Map<String, Object> inputParams, Row input, ResultFuture<BaseRow> resultFuture) throws Exception {
 
         AtomicLong networkLogCounter = new AtomicLong(0L);
-        while (!connectionStatus.get()) {//network is unhealth
+        //network is unhealthy
+        while (!connectionStatus.get()) {
             if (networkLogCounter.getAndIncrement() % 1000 == 0) {
                 LOG.info("network unhealth to block task");
             }