[Feature][admin,core] add flink pre-job udf; (#1128)

* add flink pre-job udf;
FIX SOME  BUG(udf flink sql explain)

* change code style

dlink-admin/src/main/java/com/dlink/service/impl/StudioServiceImpl.java

@@ -259,7 +259,7 @@ public class StudioServiceImpl implements StudioService {
         buildSession(config);
         process.infoSuccess();
         // To initialize java udf, but it has a bug in the product environment now.
-        // initUDF(config,studioExecuteDTO.getStatement());
+        config.setJarFiles(udfService.initUDF(studioExecuteDTO.getStatement()));
         process.start();
         JobManager jobManager = JobManager.buildPlanMode(config);
         List<SqlExplainResult> sqlExplainResults =
@@ -311,6 +311,7 @@ public class StudioServiceImpl implements StudioService {
         // If you are using explainSql | getStreamGraph | getJobPlan, make the dialect change to local.
         config.buildLocal();
         buildSession(config);
+        config.setJarFiles(udfService.initUDF(studioExecuteDTO.getStatement()));
         JobManager jobManager = JobManager.buildPlanMode(config);
         String planJson = jobManager.getJobPlanJson(studioExecuteDTO.getStatement());
         ObjectMapper mapper = new ObjectMapper();

dlink-core/src/main/java/com/dlink/job/JobManager.java

@@ -72,15 +72,20 @@ import org.apache.flink.yarn.configuration.YarnConfigOptions;
 
 import java.time.LocalDateTime;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.fasterxml.jackson.databind.node.ObjectNode;
 
+import cn.hutool.core.io.FileUtil;
+import cn.hutool.core.util.URLUtil;
+
 /**
  * JobManager
  *
@@ -168,6 +173,7 @@ public class JobManager {
         JobManager manager = new JobManager(config);
         manager.setPlanMode(true);
         manager.init();
+        manager.executor.initUDF(config.getJarFiles());
         ProcessContextHolder.getProcess().info("Build Flink plan mode success.");
         return manager;
     }
@@ -392,6 +398,7 @@ public class JobManager {
                             jobGraph.setSavepointRestoreSettings(
                                     SavepointRestoreSettings.forPath(config.getSavePointPath(), true));
                         }
+                        jobGraph.addJars(Arrays.stream(config.getJarFiles()).map(path -> URLUtil.getURL(FileUtil.file(path))).collect(Collectors.toList()));
                         gatewayResult = Gateway.build(config.getGatewayConfig()).submitJobGraph(jobGraph);
                     }
                     job.setResult(InsertResult.success(gatewayResult.getAppId()));
@@ -454,6 +461,7 @@ public class JobManager {
                 jobGraph.setSavepointRestoreSettings(SavepointRestoreSettings.forPath(config.getSavePointPath(), true));
             }
             // Perjob mode need to submit job graph.
+            jobGraph.addJars(Arrays.stream(config.getJarFiles()).map(path -> URLUtil.getURL(FileUtil.file(path))).collect(Collectors.toList()));
             gatewayResult = Gateway.build(config.getGatewayConfig()).submitJobGraph(jobGraph);
         }
         return gatewayResult;