partitioned join bug fix

Author: todd5167

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

@@ -46,6 +46,7 @@
 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.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import com.google.common.collect.HashBasedTable;
@@ -58,7 +59,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.Serializable;
 import java.util.*;
+import java.util.stream.Collectors;
 
 import static org.apache.calcite.sql.SqlKind.*;
 
@@ -743,12 +746,12 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
         DataStream adaptStream = tableEnv.toRetractStream(targetTable, org.apache.flink.types.Row.class)
                                 .map((Tuple2<Boolean, Row> f0) -> { return f0.f1; })
                                 .returns(Row.class);
+        adaptStream.getTransformation().setOutputType(leftTypeInfo);
 
         //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(), sideTableInfo);
-            String[] leftJoinColArr = new String[leftJoinColList.size()];
-            leftJoinColArr = leftJoinColList.toArray(leftJoinColArr);
+            String[] leftJoinColArr = leftJoinColList.toArray(new String[leftJoinColList.size()]);
             adaptStream = adaptStream.keyBy(leftJoinColArr);
         }