[Feature][JsonSplit] add task delete/update/switch method (#4654)

* add task query

* modify codestyle

* add task delete/update/swich method

* add task delete/update/swich method

* codestyle

* use updateById save task definition

* modify method name

Co-authored-by: JinyLeeChina <297062848@qq.com>

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessTaskRelationService.java

@@ -48,5 +48,77 @@ public interface ProcessTaskRelationService {
                                                   String conditionType,
                                                   String conditionParams);
 
+
+    /**
+     * query process task relation
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param processDefinitionCode process definition code
+     */
+    Map<String, Object> queryProcessTaskRelation(User loginUser,
+                                                 String projectName,
+                                                 Long processDefinitionCode);
+
+    /**
+     * delete process task relation
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param processDefinitionCode process definition code
+     */
+    Map<String, Object> deleteTaskDefinitionByProcessCode(User loginUser,
+                                                          String projectName,
+                                                          Long processDefinitionCode);
+
+    /**
+     * delete process task relation
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param preTaskCode pre task code
+     */
+    Map<String, Object> deleteTaskDefinitionByTaskCode(User loginUser,
+                                                       String projectName,
+                                                       Long preTaskCode);
+
+
+    /**
+     * update process task relation
+     *
+     * @param loginUser login user
+     * @param id process task relation id
+     * @param name relation name
+     * @param projectName process name
+     * @param processDefinitionCode process definition code
+     * @param preTaskCode pre task code
+     * @param postTaskCode post task code
+     * @param conditionType condition type
+     * @param conditionParams condition params
+     */
+    Map<String, Object> updateTaskDefinition(User loginUser,
+                                             int id,
+                                             String name,
+                                             String projectName,
+                                             Long processDefinitionCode,
+                                             Long preTaskCode,
+                                             Long postTaskCode,
+                                             String conditionType,
+                                             String conditionParams);
+
+
+    /**
+     * switch process task relation version
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param processTaskRelationId process task relation id
+     * @param version the version user want to switch
+     * @return switch process task relation version result code
+     */
+    Map<String, Object> switchVersion(User loginUser,
+                                      String projectName,
+                                      int processTaskRelationId,
+                                      int version);
 }
 

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TaskDefinitionService.java

@@ -47,5 +47,42 @@ public interface TaskDefinitionService {
     Map<String, Object> queryTaskDefinitionByName(User loginUser,
                                                   String projectName,
                                                   String taskName);
+
+    /**
+     * delete task definition
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param taskCode task code
+     */
+    Map<String, Object> deleteTaskDefinitionByCode(User loginUser,
+                                                   String projectName,
+                                                   Long taskCode);
+
+    /**
+     * update task definition
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param taskCode task code
+     * @param taskDefinitionJson task definition json
+     */
+    Map<String, Object> updateTaskDefinition(User loginUser,
+                                             String projectName,
+                                             Long taskCode,
+                                             String taskDefinitionJson);
+
+    /**
+     * update task definition
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param taskCode task code
+     * @param version the version user want to switch
+     */
+    Map<String, Object> switchVersion(User loginUser,
+                                      String projectName,
+                                      Long taskCode,
+                                      int version);
 }
 

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessTaskRelationServiceImpl.java

@@ -147,4 +147,80 @@ public class ProcessTaskRelationServiceImpl extends BaseService implements
             putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
         }
     }
+
+    /**
+     * query process task relation
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param processDefinitionCode process definition code
+     */
+    @Override
+    public Map<String, Object> queryProcessTaskRelation(User loginUser, String projectName, Long processDefinitionCode) {
+        return null;
+    }
+
+    /**
+     * delete process task relation
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param processDefinitionCode process definition code
+     */
+    @Override
+    public Map<String, Object> deleteTaskDefinitionByProcessCode(User loginUser, String projectName, Long processDefinitionCode) {
+        return null;
+    }
+
+    /**
+     * delete process task relation
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param preTaskCode pre task code
+     */
+    @Override
+    public Map<String, Object> deleteTaskDefinitionByTaskCode(User loginUser, String projectName, Long preTaskCode) {
+        return null;
+    }
+
+    /**
+     * update process task relation
+     *
+     * @param loginUser login user
+     * @param id process task relation id
+     * @param name relation name
+     * @param projectName process name
+     * @param processDefinitionCode process definition code
+     * @param preTaskCode pre task code
+     * @param postTaskCode post task code
+     * @param conditionType condition type
+     * @param conditionParams condition params
+     */
+    @Override
+    public Map<String, Object> updateTaskDefinition(User loginUser,
+                                                    int id,
+                                                    String name,
+                                                    String projectName,
+                                                    Long processDefinitionCode,
+                                                    Long preTaskCode,
+                                                    Long postTaskCode,
+                                                    String conditionType,
+                                                    String conditionParams) {
+        return null;
+    }
+
+    /**
+     * switch process task relation version
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param processTaskRelationId process task relation id
+     * @param version the version user want to switch
+     * @return switch process task relation version result code
+     */
+    @Override
+    public Map<String, Object> switchVersion(User loginUser, String projectName, int processTaskRelationId, int version) {
+        return null;
+    }
 }

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java

@@ -24,6 +24,7 @@ import org.apache.dolphinscheduler.api.service.TaskDefinitionService;
 import org.apache.dolphinscheduler.api.utils.CheckUtils;
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.Flag;
+import org.apache.dolphinscheduler.common.enums.ReleaseState;
 import org.apache.dolphinscheduler.common.enums.TaskType;
 import org.apache.dolphinscheduler.common.enums.TimeoutFlag;
 import org.apache.dolphinscheduler.common.model.TaskNode;
@@ -35,16 +36,21 @@ import org.apache.dolphinscheduler.common.utils.SnowFlakeUtils;
 import org.apache.dolphinscheduler.common.utils.SnowFlakeUtils.SnowFlakeException;
 import org.apache.dolphinscheduler.common.utils.StringUtils;
 import org.apache.dolphinscheduler.common.utils.TaskParametersUtils;
+import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
+import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelation;
 import org.apache.dolphinscheduler.dao.entity.Project;
 import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
 import org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog;
 import org.apache.dolphinscheduler.dao.entity.User;
+import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
+import org.apache.dolphinscheduler.dao.mapper.ProcessTaskRelationMapper;
 import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
 import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
 import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
 
 import java.util.Date;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -76,6 +82,11 @@ public class TaskDefinitionServiceImpl extends BaseService implements
     @Autowired
     private TaskDefinitionLogMapper taskDefinitionLogMapper;
 
+    @Autowired
+    private ProcessTaskRelationMapper processTaskRelationMapper;
+
+    @Autowired
+    private ProcessDefinitionMapper processDefinitionMapper;
 
     /**
      * create task definition
@@ -100,14 +111,8 @@ public class TaskDefinitionServiceImpl extends BaseService implements
         }
 
         TaskNode taskNode = JSONUtils.parseObject(taskDefinitionJson, TaskNode.class);
-        if (taskNode == null) {
-            logger.error("taskDefinitionJson is not valid json");
-            putMsg(result, Status.DATA_IS_NOT_VALID, taskDefinitionJson);
-            return result;
-        }
-        if (!CheckUtils.checkTaskNodeParameters(taskNode.getParams(), taskNode.getName())) {
-            logger.error("task node {} parameter invalid", taskNode.getName());
-            putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskNode.getName());
+        checkTaskNode(result, taskNode, taskDefinitionJson);
+        if (result.get(Constants.STATUS) != Status.SUCCESS) {
             return result;
         }
         long code = 0L;
@@ -129,8 +134,10 @@ public class TaskDefinitionServiceImpl extends BaseService implements
                 loginUser.getId(),
                 TaskType.of(taskNode.getType()),
                 taskNode.getParams(),
-                taskNode.isForbidden() ? Flag.NO : Flag.YES, taskNode.getTaskInstancePriority(),
-                taskNode.getWorkerGroup(), taskNode.getMaxRetryTimes(),
+                taskNode.isForbidden() ? Flag.NO : Flag.YES,
+                taskNode.getTaskInstancePriority(),
+                taskNode.getWorkerGroup(),
+                taskNode.getMaxRetryTimes(),
                 taskNode.getRetryInterval(),
                 taskNode.getTaskTimeoutParameter().getEnable() ? TimeoutFlag.OPEN : TimeoutFlag.CLOSE,
                 taskNode.getTaskTimeoutParameter().getStrategy(),
@@ -204,5 +211,188 @@ public class TaskDefinitionServiceImpl extends BaseService implements
         return result;
     }
 
+    /**
+     * delete task definition
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param taskCode task code
+     */
+    @Transactional(rollbackFor = RuntimeException.class)
+    @Override
+    public Map<String, Object> deleteTaskDefinitionByCode(User loginUser, String projectName, Long taskCode) {
+        Map<String, Object> result = new HashMap<>(5);
+        Project project = projectMapper.queryByName(projectName);
+
+        Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+        Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+        if (resultEnum != Status.SUCCESS) {
+            return checkResult;
+        }
+        checkTaskRelation(result, taskCode);
+        resultEnum = (Status) result.get(Constants.STATUS);
+        if (resultEnum != Status.SUCCESS) {
+            return result;
+        }
+        int delete = taskDefinitionMapper.deleteByCode(taskCode);
+        if (delete > 0) {
+            putMsg(result, Status.SUCCESS);
+        } else {
+            putMsg(result, Status.DELETE_PROCESS_DEFINE_BY_ID_ERROR);
+        }
+        return result;
+    }
+
+    /**
+     * update task definition
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param taskCode task code
+     * @param taskDefinitionJson task definition json
+     */
+    @Override
+    public Map<String, Object> updateTaskDefinition(User loginUser, String projectName, Long taskCode, String taskDefinitionJson) {
+        Map<String, Object> result = new HashMap<>(5);
+        Project project = projectMapper.queryByName(projectName);
+
+        Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+        Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+        if (resultEnum != Status.SUCCESS) {
+            return checkResult;
+        }
+        checkTaskRelation(result, taskCode);
+        if (result.get(Constants.STATUS) != Status.SUCCESS) {
+            return result;
+        }
+        TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+        if (taskDefinition == null) {
+            putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+            return result;
+        }
+        TaskNode taskNode = JSONUtils.parseObject(taskDefinitionJson, TaskNode.class);
+        checkTaskNode(result, taskNode, taskDefinitionJson);
+        if (result.get(Constants.STATUS) != Status.SUCCESS) {
+            return result;
+        }
+
+        List<TaskDefinitionLog> taskDefinitionLogs = taskDefinitionLogMapper.queryByDefinitionCode(taskCode);
+        int version = taskDefinitionLogs
+                .stream()
+                .map(TaskDefinitionLog::getVersion)
+                .max((x, y) -> x > y ? x : y)
+                .orElse(0) + 1;
+        Date now = new Date();
+        taskDefinition.setVersion(version);
+        taskDefinition.setCode(taskCode);
+        taskDefinition.setName(taskNode.getName());
+        taskDefinition.setDescription(taskNode.getDesc());
+        taskDefinition.setProjectCode(project.getCode());
+        taskDefinition.setUserId(loginUser.getId());
+        taskDefinition.setTaskType(TaskType.of(taskNode.getType()));
+        taskDefinition.setTaskParams(taskNode.getParams());
+        taskDefinition.setFlag(taskNode.isForbidden() ? Flag.NO : Flag.YES);
+        taskDefinition.setTaskPriority(taskNode.getTaskInstancePriority());
+        taskDefinition.setWorkerGroup(taskNode.getWorkerGroup());
+        taskDefinition.setFailRetryTimes(taskNode.getMaxRetryTimes());
+        taskDefinition.setFailRetryInterval(taskNode.getRetryInterval());
+        taskDefinition.setTimeoutFlag(taskNode.getTaskTimeoutParameter().getEnable() ? TimeoutFlag.OPEN : TimeoutFlag.CLOSE);
+        taskDefinition.setTaskTimeoutStrategy(taskNode.getTaskTimeoutParameter().getStrategy());
+        taskDefinition.setTimeout(taskNode.getTaskTimeoutParameter().getInterval());
+        taskDefinition.setUpdateTime(now);
+        taskDefinition.setResourceIds(getResourceIds(taskDefinition));
+        taskDefinitionMapper.updateById(taskDefinition);
+        // save task definition log
+        TaskDefinitionLog taskDefinitionLog = new TaskDefinitionLog();
+        taskDefinitionLog.set(taskDefinition);
+        taskDefinitionLog.setOperator(loginUser.getId());
+        taskDefinitionLog.setOperateTime(now);
+        taskDefinitionLogMapper.insert(taskDefinitionLog);
+        result.put(Constants.DATA_LIST, taskCode);
+        putMsg(result, Status.SUCCESS);
+        return result;
+    }
+
+    public void checkTaskRelation(Map<String, Object> result, Long taskCode) {
+        List<ProcessTaskRelation> processTaskRelationList = processTaskRelationMapper.queryByTaskCode(taskCode, taskCode);
+        if (!processTaskRelationList.isEmpty()) {
+            Set<Long> processDefinitionCodes = processTaskRelationList
+                    .stream()
+                    .map(ProcessTaskRelation::getProcessDefinitionCode)
+                    .collect(Collectors.toSet());
+            List<ProcessDefinition> processDefinitionList = processDefinitionMapper.queryByCodes(processDefinitionCodes);
+            // check process definition is already online
+            for (ProcessDefinition processDefinition : processDefinitionList) {
+                if (processDefinition.getReleaseState() == ReleaseState.ONLINE) {
+                    putMsg(result, Status.PROCESS_DEFINE_STATE_ONLINE, processDefinition.getCode());
+                    return;
+                }
+            }
+        }
+    }
+
+    public void checkTaskNode(Map<String, Object> result, TaskNode taskNode, String taskDefinitionJson) {
+        if (taskNode == null) {
+            logger.error("taskDefinitionJson is not valid json");
+            putMsg(result, Status.DATA_IS_NOT_VALID, taskDefinitionJson);
+            return;
+        }
+        if (!CheckUtils.checkTaskNodeParameters(taskNode.getParams(), taskNode.getName())) {
+            logger.error("task node {} parameter invalid", taskNode.getName());
+            putMsg(result, Status.PROCESS_NODE_S_PARAMETER_INVALID, taskNode.getName());
+        }
+    }
+
+    /**
+     * update task definition
+     *
+     * @param loginUser login user
+     * @param projectName project name
+     * @param taskCode task code
+     * @param version the version user want to switch
+     */
+    @Override
+    public Map<String, Object> switchVersion(User loginUser, String projectName, Long taskCode, int version) {
+        Map<String, Object> result = new HashMap<>(5);
+        Project project = projectMapper.queryByName(projectName);
+
+        Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName);
+        Status resultEnum = (Status) checkResult.get(Constants.STATUS);
+        if (resultEnum != Status.SUCCESS) {
+            return checkResult;
+        }
+        checkTaskRelation(result, taskCode);
+        if (result.get(Constants.STATUS) != Status.SUCCESS) {
+            return result;
+        }
+        TaskDefinition taskDefinition = taskDefinitionMapper.queryByDefinitionCode(taskCode);
+        if (taskDefinition == null) {
+            putMsg(result, Status.TASK_DEFINE_NOT_EXIST, taskCode);
+            return result;
+        }
+        TaskDefinitionLog taskDefinitionLog = taskDefinitionLogMapper.queryByDefinitionCodeAndVersion(taskCode, version);
+        taskDefinition.setVersion(version);
+        taskDefinition.setCode(taskCode);
+        taskDefinition.setName(taskDefinitionLog.getName());
+        taskDefinition.setDescription(taskDefinitionLog.getDescription());
+        taskDefinition.setProjectCode(taskDefinitionLog.getProjectCode());
+        taskDefinition.setUserId(loginUser.getId());
+        taskDefinition.setTaskType(taskDefinitionLog.getTaskType());
+        taskDefinition.setTaskParams(taskDefinitionLog.getTaskParams());
+        taskDefinition.setFlag(taskDefinitionLog.getFlag());
+        taskDefinition.setTaskPriority(taskDefinitionLog.getTaskPriority());
+        taskDefinition.setWorkerGroup(taskDefinitionLog.getWorkerGroup());
+        taskDefinition.setFailRetryTimes(taskDefinitionLog.getFailRetryTimes());
+        taskDefinition.setFailRetryInterval(taskDefinitionLog.getFailRetryInterval());
+        taskDefinition.setTimeoutFlag(taskDefinitionLog.getTimeoutFlag());
+        taskDefinition.setTaskTimeoutStrategy(taskDefinitionLog.getTaskTimeoutStrategy());
+        taskDefinition.setTimeout(taskDefinitionLog.getTimeout());
+        taskDefinition.setUpdateTime(new Date());
+        taskDefinition.setResourceIds(taskDefinitionLog.getResourceIds());
+        taskDefinitionMapper.updateById(taskDefinition);
+        result.put(Constants.DATA_LIST, taskCode);
+        putMsg(result, Status.SUCCESS);
+        return result;
+    }
 }
 

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapper.java

@@ -23,6 +23,7 @@ import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
 import org.apache.ibatis.annotations.MapKey;
 import org.apache.ibatis.annotations.Param;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 
@@ -41,7 +42,15 @@ public interface ProcessDefinitionMapper extends BaseMapper<ProcessDefinition> {
      * @return process definition
      */
     ProcessDefinition queryByCode(@Param("code") Long code);
-    
+
+    /**
+     * query process definition by code list
+     *
+     * @param codes codes
+     * @return process definition list
+     */
+    List<ProcessDefinition> queryByCodes(@Param("codes") Collection<Long> codes);
+
     /**
      * delete process definition by code
      *

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessTaskRelationMapper.java

@@ -34,7 +34,17 @@ public interface ProcessTaskRelationMapper extends BaseMapper<ProcessTaskRelatio
      * process task relation by processDefinitionCode
      *
      * @param processDefinitionCode processDefinitionCode
-     * @return task definition
+     * @return ProcessTaskRelation
      */
     List<ProcessTaskRelation> queryByProcessDefinitionCode(@Param("processDefinitionCode") String processDefinitionCode);
+
+    /**
+     * process task relation by taskCode
+     *
+     * @param preTaskCode preTaskCode
+     * @param postTaskCode postTaskCode
+     * @return ProcessTaskRelation
+     */
+    List<ProcessTaskRelation> queryByTaskCode(@Param("preTaskCode") Long preTaskCode,
+                                              @Param("postTaskCode") Long postTaskCode);
 }

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionLogMapper.java

@@ -47,4 +47,14 @@ public interface TaskDefinitionLogMapper extends BaseMapper<TaskDefinitionLog> {
      * @return task definition log list
      */
     List<TaskDefinitionLog> queryByDefinitionCode(@Param("taskDefinitionCode") long taskDefinitionCode);
+
+    /**
+     * query task definition log
+     *
+     * @param taskDefinitionCode taskDefinitionCode
+     * @param version version
+     * @return task definition log
+     */
+    TaskDefinitionLog queryByDefinitionCodeAndVersion(@Param("taskDefinitionCode") long taskDefinitionCode,
+                                                      @Param("version") int version);
 }

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.java

@@ -98,4 +98,12 @@ public interface TaskDefinitionMapper extends BaseMapper<TaskDefinition> {
      */
     @MapKey("id")
     List<Map<String, Object>> listResourcesByUser(@Param("userId") Integer userId);
+
+    /**
+     * delete task definition by code
+     *
+     * @param code code
+     * @return int
+     */
+    int deleteByCode(@Param("code") Long code);
 }

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapper.xml

@@ -44,6 +44,18 @@
         from t_ds_process_definition
         where code = #{code}
     </select>
+    <select id="queryByCodes" resultType="org.apache.dolphinscheduler.dao.entity.ProcessDefinition">
+        select
+        <include refid="baseSql"/>
+        from t_ds_process_definition
+        where 1 = 1
+        <if test="codes != null and codes.length != 0">
+            and code in
+            <foreach collection="codes" index="index" item="i" open="(" separator="," close=")">
+                #{i}
+            </foreach>
+        </if>
+    </select>
 
     <select id="queryByDefineName" resultType="org.apache.dolphinscheduler.dao.entity.ProcessDefinition">
         select pd.id, pd.name, pd.version, pd.release_state, pd.project_id, pd.user_id, pd.description,

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessTaskRelationMapper.xml

@@ -19,8 +19,8 @@
 <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
 <mapper namespace="org.apache.dolphinscheduler.dao.mapper.ProcessTaskRelationMapper">
     <sql id="baseSql">
-        id, `name`, version, project_code, process_definition_code, pre_project_code, pre_task_code, post_project_code,
-        post_task_code, condition_type, condition_params, create_time, update_time
+        id, `name`, version, project_code, process_definition_code, pre_task_code, post_task_code,
+        condition_type, condition_params, create_time, update_time
     </sql>
     <select id="queryByProcessDefinitionCode" resultType="org.apache.dolphinscheduler.dao.entity.ProcessTaskRelation">
         select
@@ -28,4 +28,13 @@
         from t_ds_process_task_relation
         WHERE process_definition_code = #{processDefinitionCode}
     </select>
+    <select id="queryByTaskCode" resultType="org.apache.dolphinscheduler.dao.entity.ProcessTaskRelation">
+        select
+        <include refid="baseSql"/>
+        from t_ds_process_task_relation
+        WHERE pre_task_code = #{preTaskCode}
+        <if test="postTaskCode != 0">
+            or post_task_code = #{postTaskCode}
+        </if>
+    </select>
 </mapper>

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionLogMapper.xml

@@ -18,6 +18,11 @@
 
 <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
 <mapper namespace="org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper">
+    <sql id="baseSql">
+        id, code, `name`, version, description, project_code, user_id, task_type, task_params, flag, task_priority,
+        worker_group, fail_retry_times, fail_retry_interval, timeout_flag, timeout_notify_strategy, timeout,
+        resource_ids, operator, operate_time, create_time, update_time
+    </sql>
     <select id="queryByDefinitionName" resultType="org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog">
         select td.id, td.code, td.name, td.version, td.description, td.project_code, td.user_id, td.task_type, td.task_params,
         td.flag, td.task_priority, td.worker_group, td.fail_retry_times, td.fail_retry_interval, td.timeout_flag,
@@ -30,13 +35,16 @@
         and td.name = #{taskDefinitionName}
     </select>
     <select id="queryByDefinitionCode" resultType="org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog">
-        select td.id, td.code, td.name, td.version, td.description, td.project_code, td.user_id, td.task_type, td.task_params,
-        td.flag, td.task_priority, td.worker_group, td.fail_retry_times, td.fail_retry_interval, td.timeout_flag,
-        td.timeout_notify_strategy, td.timeout, td.resource_ids, td.operator,td.operate_time, td.create_time, td.update_time,
-        u.user_name,p.name as project_name
-        from t_ds_task_definition_log td
-        JOIN t_ds_user u ON td.user_id = u.id
-        JOIN  t_ds_project p ON td.project_code = p.code
-        WHERE td.code = #{taskDefinitionCode}
+        select
+        <include refid="baseSql"/>
+        from t_ds_task_definition_log
+        WHERE code = #{taskDefinitionCode}
+    </select>
+    <select id="queryByDefinitionCodeAndVersion" resultType="org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog">
+        select
+        <include refid="baseSql"/>
+        from t_ds_task_definition_log
+        WHERE code = #{taskDefinitionCode}
+        and version = #{version}
     </select>
 </mapper>

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.xml

@@ -87,4 +87,8 @@
         FROM t_ds_task_definition
         WHERE user_id = #{userId} resource_ids is not null and resource_ids != ''
     </select>
+    <delete id="deleteByCode">
+        delete from t_ds_task_definition
+        where code = #{code}
+    </delete>
 </mapper>