partitioned join bug fix
@@ -46,6 +46,7 @@ import org.apache.calcite.sql.parser.SqlParseException;
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.apache.flink.types.Row;
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 @@ public class SideSqlExec {
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);
}
