use retract stream

Author: todd5167

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

@@ -593,9 +593,7 @@ 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);
         //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());