Merge branch 'v1.5.0_dev_bugfix' into v1.5.0_dev

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

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -46,6 +46,7 @@
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.calcite.shaded.com.google.common.base.Preconditions;
 import org.apache.flink.calcite.shaded.com.google.common.base.Strings;
@@ -271,7 +272,10 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en
                 Table adaptTable = adaptSql == null ? table : tableEnv.sqlQuery(adaptSql);
 
                 RowTypeInfo typeInfo = new RowTypeInfo(adaptTable.getSchema().getTypes(), adaptTable.getSchema().getColumnNames());
-                DataStream adaptStream = tableEnv.toAppendStream(adaptTable, typeInfo);
+                DataStream adaptStream = tableEnv.toRetractStream(adaptTable, typeInfo)
+                        .map((Tuple2<Boolean, Row> f0) -> { return f0.f1; })
+                        .returns(typeInfo);
+
                 String fields = String.join(",", typeInfo.getFieldNames());
 
                 if(waterMarkerAssigner.checkNeedAssignWaterMarker(sourceTableInfo)){

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

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
- 
+
 
 package com.dtstack.flink.sql.side;
 
@@ -41,13 +41,15 @@
 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.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.calcite.shaded.com.google.common.collect.HashBasedTable;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.table.api.Table;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.types.Row;
 
 import java.util.*;
 
@@ -593,7 +595,11 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
         }
 
         RowTypeInfo typeInfo = new RowTypeInfo(targetTable.getSchema().getTypes(), targetTable.getSchema().getColumnNames());
-        DataStream adaptStream = tableEnv.toRetractStream(targetTable, org.apache.flink.types.Row.class);
+
+        DataStream adaptStream = tableEnv.toRetractStream(targetTable, org.apache.flink.types.Row.class)
+                                .map((Tuple2<Boolean, Row> f0) -> { return f0.f1; })
+                                .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());

core/src/main/java/com/dtstack/flink/sql/util/PluginUtil.java

@@ -111,15 +111,15 @@ public static URL getRemoteJarFilePath(String pluginType, String tableType, Stri
         String dirName = pluginType + tableType.toLowerCase();
         String prefix = String.format("%s-%s", pluginType, tableType.toLowerCase());
         String jarPath = remoteSqlRootDir + SP + dirName;
-        String jarName = getCoreJarFileName(jarPath, prefix);
+        String jarName = getCoreJarFileName(jarPath, prefix, false);
         return new URL("file:" + remoteSqlRootDir + SP + dirName + SP + jarName);
     }
 
     public static URL getRemoteSideJarFilePath(String pluginType, String sideOperator, String tableType, String remoteSqlRootDir) throws Exception {
         String dirName = pluginType + sideOperator + tableType.toLowerCase();
         String prefix = String.format("%s-%s-%s", pluginType, sideOperator, tableType.toLowerCase());
         String jarPath = remoteSqlRootDir + SP + dirName;
-        String jarName = getCoreJarFileName(jarPath, prefix);
+        String jarName = getCoreJarFileName(jarPath, prefix, false);
         return new URL("file:" + remoteSqlRootDir + SP + dirName + SP + jarName);
     }
 
@@ -144,7 +144,11 @@ public static void addPluginJar(String pluginDir, DtClassLoader classLoader) thr
         }
     }
 
-    public static String getCoreJarFileName (String path, String prefix) throws Exception {
+    public static String getCoreJarFileName (String path, String prefix, boolean existCheck) throws Exception {
+        if (!existCheck) {
+            return prefix.toLowerCase() + ".jar";
+        }
+
         String coreJarFileName = null;
         File pluginDir = new File(path);
         if (pluginDir.exists() && pluginDir.isDirectory()){

launcher/src/main/java/com/dtstack/flink/sql/launcher/LauncherMain.java

@@ -59,7 +59,7 @@ public class LauncherMain {
 
 
     private static String getLocalCoreJarPath(String localSqlRootJar) throws Exception {
-        String jarPath = PluginUtil.getCoreJarFileName(localSqlRootJar, CORE_JAR);
+        String jarPath = PluginUtil.getCoreJarFileName(localSqlRootJar, CORE_JAR, true);
         String corePath = localSqlRootJar + SP + jarPath;
         return corePath;
     }

launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/PerJobSubmitter.java

@@ -52,16 +52,16 @@ public class PerJobSubmitter {
     public static String submit(LauncherOptions launcherOptions, JobGraph jobGraph) throws Exception {
 
 		fillJobGraphClassPath(jobGraph);
-
-		String addjarPath = URLDecoder.decode(launcherOptions.getAddjar(), Charsets.UTF_8.toString());
-		if (StringUtils.isNotBlank(addjarPath) ){
+		if (!StringUtils.isBlank(launcherOptions.getAddjar())) {
+			String addjarPath = URLDecoder.decode(launcherOptions.getAddjar(), Charsets.UTF_8.toString());
 			List<String>  paths = getJarPaths(addjarPath);
-			paths.forEach( path ->{
+			paths.forEach( path -> {
 				jobGraph.addJar(new Path("file://" + path));
 			});
-
 		}
 
+
+
 		String confProp = launcherOptions.getConfProp();
         confProp = URLDecoder.decode(confProp, Charsets.UTF_8.toString());
         Properties confProperties = PluginUtil.jsonStrToObject(confProp, Properties.class);