retract stream convert type

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -46,6 +46,7 @@
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.calcite.shaded.com.google.common.base.Preconditions;
 import org.apache.flink.calcite.shaded.com.google.common.base.Strings;
@@ -271,7 +272,10 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en
                 Table adaptTable = adaptSql == null ? table : tableEnv.sqlQuery(adaptSql);
 
                 RowTypeInfo typeInfo = new RowTypeInfo(adaptTable.getSchema().getTypes(), adaptTable.getSchema().getColumnNames());
-                DataStream adaptStream = tableEnv.toAppendStream(adaptTable, typeInfo);
+                DataStream adaptStream = tableEnv.toRetractStream(adaptTable, typeInfo)
+                        .map((Tuple2<Boolean, Row> f0) -> { return f0.f1; })
+                        .returns(typeInfo);
+
                 String fields = String.join(",", typeInfo.getFieldNames());
 
                 if(waterMarkerAssigner.checkNeedAssignWaterMarker(sourceTableInfo)){

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

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
- 
+
 
 package com.dtstack.flink.sql.side;
 
@@ -41,13 +41,15 @@
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.calcite.shaded.com.google.common.collect.HashBasedTable;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
 import org.apache.flink.calcite.shaded.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.java.StreamTableEnvironment;
+import org.apache.flink.types.Row;
 
 import java.util.*;
 
@@ -593,9 +595,9 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
         }
 
         RowTypeInfo typeInfo = new RowTypeInfo(targetTable.getSchema().getTypes(), targetTable.getSchema().getColumnNames());
-        //DataStream adaptStream = tableEnv.toRetractStream(targetTable, org.apache.flink.types.Row.class);
-        DataStream adaptStream = tableEnv.toAppendStream(targetTable, org.apache.flink.types.Row.class);
-
+        DataStream adaptStream = tableEnv.toRetractStream(targetTable, org.apache.flink.types.Row.class)
+                                .map((Tuple2<Boolean, Row> f0) -> { return f0.f1; })
+                                .returns(typeInfo);
         //join side table before keyby ===> Reducing the size of each dimension table cache of async
         if(sideTableInfo.isPartitionedJoin()){
             List<String> leftJoinColList = getConditionFields(joinInfo.getCondition(), joinInfo.getLeftTableAlias());