Merge branch 'v1.4.1'

Author: zoudaokoulife

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

@@ -21,6 +21,7 @@
 package com.dtstack.flink.sql;
 
 import com.dtstack.flink.sql.classloader.DtClassLoader;
+import com.dtstack.flink.sql.enums.ECacheType;
 import com.dtstack.flink.sql.parser.CreateFuncParser;
 import com.dtstack.flink.sql.parser.InsertSqlParser;
 import com.dtstack.flink.sql.side.SideSqlExec;
@@ -272,8 +273,9 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en
                 classPathSet.add( PluginUtil.getRemoteJarFilePath(tableInfo.getType(), TargetTableInfo.TARGET_SUFFIX, remoteSqlPluginPath));
             } else if(tableInfo instanceof SideTableInfo){
 
+                String sideOperator = ECacheType.ALL.name().equals(((SideTableInfo) tableInfo).getCacheType()) ? "all" : "async";
                 sideTableMap.put(tableInfo.getName(), (SideTableInfo) tableInfo);
-                classPathSet.add(PluginUtil.getRemoteJarFilePath(tableInfo.getType(), SideTableInfo.TARGET_SUFFIX, remoteSqlPluginPath));
+                classPathSet.add(PluginUtil.getRemoteSideJarFilePath(tableInfo.getType(), sideOperator, SideTableInfo.TARGET_SUFFIX, remoteSqlPluginPath));
             }else {
                 throw new RuntimeException("not support table type:" + tableInfo.getType());
             }

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

@@ -21,6 +21,7 @@
 package com.dtstack.flink.sql.side;
 
 import com.dtstack.flink.sql.classloader.DtClassLoader;
+import com.dtstack.flink.sql.enums.ECacheType;
 import com.dtstack.flink.sql.table.AbsSideTableParser;
 import com.dtstack.flink.sql.table.AbsTableParser;
 import com.dtstack.flink.sql.util.PluginUtil;
@@ -36,14 +37,11 @@ public class StreamSideFactory {
 
     private static final String CURR_TYPE = "side";
 
-    private static final String SIDE_DIR_TMPL = "%s%sside";
-
     public static AbsTableParser getSqlParser(String pluginType, String sqlRootDir, String cacheType) throws Exception {
 
-        cacheType = cacheType == null ? "async" : cacheType;
-        String sideDir = String.format(SIDE_DIR_TMPL, pluginType, cacheType);
+        String sideOperator = ECacheType.ALL.name().equals(cacheType) ? "all" : "async";
         ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
-        String pluginJarPath = PluginUtil.getJarFileDirPath(sideDir, sqlRootDir);
+        String pluginJarPath = PluginUtil.getSideJarFileDirPath(pluginType, sideOperator, "side", sqlRootDir);
 
         DtClassLoader dtClassLoader = (DtClassLoader) classLoader;
         PluginUtil.addPluginJar(pluginJarPath, dtClassLoader);

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

@@ -63,6 +63,17 @@ public static String getJarFileDirPath(String type, String sqlRootDir){
         return jarPath;
     }
 
+    public static String getSideJarFileDirPath(String pluginType, String sideOperator, String tableType, String sqlRootDir) throws MalformedURLException {
+        String dirName = sqlRootDir + SP + pluginType + sideOperator + tableType.toLowerCase();
+        File jarFile = new File(dirName);
+
+        if(!jarFile.exists()){
+            throw new RuntimeException(String.format("path %s not exists!!!", dirName));
+        }
+
+        return dirName;
+    }
+
     public static String getGenerClassName(String pluginTypeName, String type) throws IOException {
         String pluginClassName = upperCaseFirstChar(pluginTypeName) + upperCaseFirstChar(type);
         return CLASS_PRE_STR  + "." + type.toLowerCase() + "." + pluginTypeName + "." + pluginClassName;
@@ -94,12 +105,18 @@ public static Properties stringToProperties(String str) throws IOException{
         return properties;
     }
 
-    public static URL getRemoteJarFilePath(String pluginType, String tableType,String remoteSqlRootDir) throws MalformedURLException {
+    public static URL getRemoteJarFilePath(String pluginType, String tableType, String remoteSqlRootDir) throws MalformedURLException {
         String dirName = pluginType + tableType.toLowerCase();
         String jarName = String.format("%s-%s.jar", pluginType, tableType.toLowerCase());
         return new URL("file:" + remoteSqlRootDir + SP + dirName + SP + jarName);
     }
 
+    public static URL getRemoteSideJarFilePath(String pluginType, String sideOperator, String tableType, String remoteSqlRootDir) throws MalformedURLException {
+        String dirName = pluginType + sideOperator + tableType.toLowerCase();
+        String jarName = String.format("%s-%s-%s.jar", pluginType, sideOperator, tableType.toLowerCase());
+        return new URL("file:" + remoteSqlRootDir + SP + dirName + SP + jarName);
+    }
+
     public static String upperCaseFirstChar(String str){
         return str.substring(0, 1).toUpperCase() + str.substring(1);
     }