modify sidesqlexc datastream type
@@ -595,9 +595,11 @@ public class SideSqlExec {
}
RowTypeInfo typeInfo = new RowTypeInfo(targetTable.getSchema().getTypes(), targetTable.getSchema().getColumnNames());
+
DataStream adaptStream = tableEnv.toRetractStream(targetTable, org.apache.flink.types.Row.class)
.map((Tuple2<Boolean, Row> f0) -> { return f0.f1; })
- .returns(typeInfo);
+ .returns(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());
