[fix] udf class conflict

Author: FlechazoW

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

@@ -23,6 +23,8 @@
 import com.dtstack.flink.sql.exec.ParamsInfo;
 import org.apache.commons.lang.exception.ExceptionUtils;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  *  local模式获取sql任务的执行计划
@@ -32,15 +34,19 @@
  */
 public class GetPlan {
 
+    private static final Logger LOG = LoggerFactory.getLogger(GetPlan.class);
+
     public static String getExecutionPlan(String[] args) {
         try {
             long start = System.currentTimeMillis();
             ParamsInfo paramsInfo = ExecuteProcessHelper.parseParams(args);
+            paramsInfo.setGetPlan(true);
             StreamExecutionEnvironment env = ExecuteProcessHelper.getStreamExecution(paramsInfo);
             String executionPlan = env.getExecutionPlan();
             long end = System.currentTimeMillis();
             return ApiResult.createSuccessResultJsonStr(executionPlan, end - start);
         } catch (Exception e) {
+            LOG.error("Get plan error", e);
             return ApiResult.createErrorResultJsonStr(ExceptionUtils.getFullStackTrace(e));
         }
     }

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

@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.exec;
 
+import com.dtstack.flink.sql.GetPlan;
 import com.dtstack.flink.sql.parser.CreateFuncParser;
 import com.dtstack.flink.sql.parser.CreateTmpTableParser;
 import com.dtstack.flink.sql.parser.FlinkPlanner;
@@ -74,6 +75,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Properties;
 import java.util.Set;
 import java.util.TimeZone;
@@ -158,7 +160,7 @@ public static StreamExecutionEnvironment getStreamExecution(ParamsInfo paramsInf
         Map<String, Table> registerTableCache = Maps.newHashMap();
 
         //register udf
-        ExecuteProcessHelper.registerUserDefinedFunction(sqlTree, paramsInfo.getJarUrlList(), tableEnv);
+        ExecuteProcessHelper.registerUserDefinedFunction(sqlTree, paramsInfo.getJarUrlList(), tableEnv, paramsInfo.isGetPlan());
         //register table schema
         Set<URL> classPathSets = ExecuteProcessHelper.registerTable(sqlTree, env, tableEnv, paramsInfo.getLocalSqlPluginPath(),
                 paramsInfo.getRemoteSqlPluginPath(), paramsInfo.getPluginLoadMode(), sideTableMap, registerTableCache);
@@ -243,13 +245,19 @@ private static void sqlTranslation(String localSqlPluginPath,
         }
     }
 
-    public static void registerUserDefinedFunction(SqlTree sqlTree, List<URL> jarUrlList, TableEnvironment tableEnv)
+    public static void registerUserDefinedFunction(SqlTree sqlTree, List<URL> jarUrlList, TableEnvironment tableEnv, boolean getPlan)
             throws IllegalAccessException, InvocationTargetException {
         // udf和tableEnv须由同一个类加载器加载
         ClassLoader levelClassLoader = tableEnv.getClass().getClassLoader();
         URLClassLoader classLoader = null;
         List<CreateFuncParser.SqlParserResult> funcList = sqlTree.getFunctionList();
         for (CreateFuncParser.SqlParserResult funcInfo : funcList) {
+            // 构建plan的情况下，udf和tableEnv不需要是同一个类加载器
+            if (getPlan) {
+                URL[] urls = jarUrlList.toArray(new URL[0]);
+                classLoader = URLClassLoader.newInstance(urls);
+            }
+
             //classloader
             if (classLoader == null) {
                 classLoader = ClassLoaderManager.loadExtraJar(jarUrlList, (URLClassLoader) levelClassLoader);

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

@@ -21,6 +21,7 @@
 
 import java.net.URL;
 import java.util.List;
+import java.util.Objects;
 import java.util.Properties;
 
 /**
@@ -39,6 +40,7 @@ public class ParamsInfo {
     private String pluginLoadMode;
     private String deployMode;
     private Properties confProp;
+    private boolean getPlan = false;
 
     public ParamsInfo(String sql, String name, List<URL> jarUrlList, String localSqlPluginPath,
                       String remoteSqlPluginPath, String pluginLoadMode, String deployMode, Properties confProp) {
@@ -52,6 +54,14 @@ public ParamsInfo(String sql, String name, List<URL> jarUrlList, String localSql
         this.confProp = confProp;
     }
 
+    public boolean isGetPlan() {
+        return getPlan;
+    }
+
+    public void setGetPlan(boolean getPlan) {
+        this.getPlan = getPlan;
+    }
+
     public String getSql() {
         return sql;
     }