[merge] hotfix_1.10_4.0.x_35168 -> tmp_1.10_release_4.1.x

resolve conflict

Author: 温天柱

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

@@ -30,9 +30,10 @@
 import java.net.URLClassLoader;
 
 /**
- *  local模式获取sql任务的执行计划
+ * local模式获取sql任务的执行计划
  * Date: 2020/2/17
  * Company: www.dtstack.com
+ *
  * @author maqi
  */
 public class GetPlan {
@@ -46,7 +47,9 @@ public static String getExecutionPlan(String[] args) {
             ParamsInfo paramsInfo = ExecuteProcessHelper.parseParams(args);
             paramsInfo.setGetPlan(true);
             ClassLoader envClassLoader = StreamExecutionEnvironment.class.getClassLoader();
-            ClassLoader plannerClassLoader = URLClassLoader.newInstance(new URL[0], envClassLoader);
+            ClassLoader plannerClassLoader = URLClassLoader.newInstance(
+                    paramsInfo.getJarUrlList().toArray(new URL[0]),
+                    envClassLoader);
             Thread.currentThread().setContextClassLoader(plannerClassLoader);
             StreamExecutionEnvironment env = ExecuteProcessHelper.getStreamExecution(paramsInfo);
             String executionPlan = env.getExecutionPlan();

core/src/main/java/com/dtstack/flink/sql/classloader/ClassLoaderManager.java

@@ -20,10 +20,12 @@
 
 import com.dtstack.flink.sql.util.PluginUtil;
 import com.dtstack.flink.sql.util.ReflectionUtils;
+import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.FileInputStream;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.net.URL;
@@ -71,9 +73,19 @@ private static DtClassLoader retrieveClassLoad(String pluginJarPath) {
         });
     }
 
-    private static DtClassLoader retrieveClassLoad(List<URL> jarUrls) {
+    public static DtClassLoader retrieveClassLoad(List<URL> jarUrls) {
         jarUrls.sort(Comparator.comparing(URL::toString));
-        String jarUrlkey = StringUtils.join(jarUrls, "_");
+
+        List<String> jarMd5s = new ArrayList<>(jarUrls.size());
+        for (URL jarUrl : jarUrls) {
+            try (FileInputStream inputStream = new FileInputStream(jarUrl.getPath())){
+                String jarMd5 = DigestUtils.md5Hex(inputStream);
+                jarMd5s.add(jarMd5);
+            } catch (Exception e) {
+                throw new RuntimeException("Exceptions appears when read file:" + e);
+            }
+        }
+        String jarUrlkey = StringUtils.join(jarMd5s, "_");
         return pluginClassLoader.computeIfAbsent(jarUrlkey, k -> {
             try {
                 URL[] urls = jarUrls.toArray(new URL[jarUrls.size()]);

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -55,6 +55,7 @@
 import org.apache.calcite.sql.SqlInsert;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.commons.io.Charsets;
+import org.apache.commons.lang3.SerializationUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
@@ -79,6 +80,7 @@
 import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -244,7 +246,11 @@ private static void sqlTranslation(String localSqlPluginPath,
             scope++;
         }
 
+        final Map<String, AbstractSideTableInfo> tmpTableMap = new HashMap<>();
         for (InsertSqlParser.SqlParseResult result : sqlTree.getExecSqlList()) {
+            // prevent current sql use last sql's sideTableInfo
+            sideTableMap.forEach((s, abstractSideTableInfo) -> tmpTableMap.put(s, SerializationUtils.clone(abstractSideTableInfo)));
+
             if (LOG.isInfoEnabled()) {
                 LOG.info("exe-sql:\n" + result.getExecSql());
             }
@@ -257,17 +263,17 @@ private static void sqlTranslation(String localSqlPluginPath,
                     SqlNode sqlNode = flinkPlanner.getParser().parse(realSql);
                     String tmpSql = ((SqlInsert) sqlNode).getSource().toString();
                     tmp.setExecSql(tmpSql);
-                    sideSqlExec.exec(tmp.getExecSql(), sideTableMap, tableEnv, registerTableCache, tmp, scope + "");
+                    sideSqlExec.exec(tmp.getExecSql(), tmpTableMap, tableEnv, registerTableCache, tmp, scope + "");
                 } else {
                     for (String sourceTable : result.getSourceTableList()) {
-                        if (sideTableMap.containsKey(sourceTable)) {
+                        if (tmpTableMap.containsKey(sourceTable)) {
                             isSide = true;
                             break;
                         }
                     }
                     if (isSide) {
                         //sql-dimensional table contains the dimension table of execution
-                        sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache, null, String.valueOf(scope));
+                        sideSqlExec.exec(result.getExecSql(), tmpTableMap, tableEnv, registerTableCache, null, String.valueOf(scope));
                     } else {
                         LOG.info("----------exec sql without dimension join-----------");
                         LOG.info("----------real sql exec is--------------------------\n{}", result.getExecSql());
@@ -280,26 +286,17 @@ private static void sqlTranslation(String localSqlPluginPath,
 
                 scope++;
             }
+            tmpTableMap.clear();
         }
     }
 
     public static void registerUserDefinedFunction(SqlTree sqlTree, List<URL> jarUrlList, TableEnvironment tableEnv, boolean getPlan)
             throws IllegalAccessException, InvocationTargetException {
         // udf和tableEnv须由同一个类加载器加载
-        ClassLoader levelClassLoader = tableEnv.getClass().getClassLoader();
         ClassLoader currentClassLoader = Thread.currentThread().getContextClassLoader();
-        URLClassLoader classLoader = null;
+        URLClassLoader classLoader = ClassLoaderManager.loadExtraJar(jarUrlList, (URLClassLoader) currentClassLoader);
         List<CreateFuncParser.SqlParserResult> funcList = sqlTree.getFunctionList();
         for (CreateFuncParser.SqlParserResult funcInfo : funcList) {
-            // 构建plan的情况下，udf和tableEnv不需要是同一个类加载器
-            if (getPlan) {
-                classLoader = ClassLoaderManager.loadExtraJar(jarUrlList, (URLClassLoader) currentClassLoader);
-            }
-
-            //classloader
-            if (classLoader == null) {
-                classLoader = ClassLoaderManager.loadExtraJar(jarUrlList, (URLClassLoader) levelClassLoader);
-            }
             FunctionManager.registerUDF(funcInfo.getType(), funcInfo.getClassName(), funcInfo.getName(), tableEnv, classLoader);
         }
     }

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

@@ -539,11 +539,25 @@ private static void replaceConditionNode(SqlNode selectNode, String oldTbName, S
 
             String tableName = sqlIdentifier.names.asList().get(0);
             String tableField = sqlIdentifier.names.asList().get(1);
-            String fieldKey = tableName + "_" + tableField;
+            String fieldKey = tableName + "." + tableField;
 
             if(tableName.equalsIgnoreCase(oldTbName)){
 
-                String newFieldName = fieldReplaceRef.get(fieldKey) == null ? tableField : fieldReplaceRef.get(fieldKey);
+                /*
+                 *  ****Before replace:*****
+                 *  fieldKey: b.department
+                 *  fieldReplaceRef : b.department -> a_b_0.department0
+                 *  oldFieldRef: a_b_0.department0
+                 *  oldTbName: b
+                 *  oldFieldName: department
+                 *  ****After replace:*****
+                 *  newTbName: a_b_0
+                 *  newFieldName: department0
+                 */
+                String oldFieldRef = fieldReplaceRef.get(fieldKey);
+                String newFieldName = (oldFieldRef != null && !StringUtils.substringAfter(oldFieldRef, ".").isEmpty()) ?
+                        StringUtils.substringAfter(oldFieldRef, ".") : tableField;
+
                 SqlIdentifier newField = ((SqlIdentifier)selectNode).setName(0, newTbName);
                 newField = newField.setName(1, newFieldName);
                 ((SqlIdentifier)selectNode).assignNamesFrom(newField);