Merge remote-tracking branch 'origin/v1.8.0_dev' into v1.8.0_dev

Author: simenliuxing

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

@@ -747,7 +747,7 @@ 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);
+        //adaptStream.getTransformation().setOutputType(leftTypeInfo);
 
         //join side table before keyby ===> Reducing the size of each dimension table cache of async
         if(sideTableInfo.isPartitionedJoin()){

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduSink.java

@@ -9,6 +9,7 @@
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.api.java.typeutils.TupleTypeInfo;
 import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
 import org.apache.flink.streaming.api.functions.sink.OutputFormatSinkFunction;
 import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
 import org.apache.flink.table.sinks.RetractStreamTableSink;
@@ -63,7 +64,11 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
                 .setFieldTypes(this.fieldTypes);
         KuduOutputFormat kuduOutputFormat = builder.finish();
         RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(kuduOutputFormat);
-        dataStream.addSink(richSinkFunction);
+        DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction);
+        dataStreamSink.name(tableName);
+        if (parallelism > 0) {
+            dataStreamSink.setParallelism(parallelism);
+        }
     }
 
     @Override