modify position

Author: todd5167

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

@@ -771,11 +771,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);
-        RowTypeInfo leftTableOutType = buildLeftTableOutType(leftTypeInfo);
-        adaptStream.getTransformation().setOutputType(leftTableOutType);
+
 
         //join side table before keyby ===> Reducing the size of each dimension table cache of async
         if(sideTableInfo.isPartitionedJoin()){
+            RowTypeInfo leftTableOutType = buildLeftTableOutType(leftTypeInfo);
+            adaptStream.getTransformation().setOutputType(leftTableOutType);
             List<String> leftJoinColList = getConditionFields(joinInfo.getCondition(), joinInfo.getLeftTableAlias(), sideTableInfo);
             String[] leftJoinColArr = leftJoinColList.toArray(new String[leftJoinColList.size()]);
             adaptStream = adaptStream.keyBy(leftJoinColArr);