[Feature][JsonSplit] fix process lineage bug (#5418)

* update taskParams/add task delayTime/fix conditionType bug

* update codeStyle for merge to dev

* fix process lineage bug

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

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

@@ -20,11 +20,11 @@ package org.apache.dolphinscheduler.api.service.impl;
 import org.apache.dolphinscheduler.api.enums.Status;
 import org.apache.dolphinscheduler.api.service.WorkFlowLineageService;
 import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.utils.StringUtils;
 import org.apache.dolphinscheduler.dao.entity.ProcessLineage;
 import org.apache.dolphinscheduler.dao.entity.Project;
 import org.apache.dolphinscheduler.dao.entity.WorkFlowLineage;
 import org.apache.dolphinscheduler.dao.entity.WorkFlowRelation;
-import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
 import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
 import org.apache.dolphinscheduler.dao.mapper.WorkFlowLineageMapper;
 
@@ -46,9 +46,6 @@ public class WorkFlowLineageServiceImpl extends BaseServiceImpl implements WorkF
     @Autowired
     private WorkFlowLineageMapper workFlowLineageMapper;
 
-    @Autowired
-    private ProcessDefinitionMapper processDefinitionMapper;
-
     @Autowired
     private ProjectMapper projectMapper;
 
@@ -66,48 +63,41 @@ public class WorkFlowLineageServiceImpl extends BaseServiceImpl implements WorkF
                              Set<WorkFlowRelation> workFlowRelations,
                              ProcessLineage processLineage) {
         List<ProcessLineage> relations = workFlowLineageMapper.queryCodeRelation(
-                processLineage.getPostTaskCode(), processLineage.getPostTaskVersion()
-                , processLineage.getProcessDefinitionCode(), processLineage.getProjectCode());
-
-        for (ProcessLineage relation : relations) {
-            if (relation.getProcessDefinitionCode() != null) {
-
-                relation.setPreTaskCode(processLineage.getPostTaskCode());
-                relation.setPreTaskVersion(processLineage.getPostTaskVersion());
-
-                WorkFlowLineage pre = workFlowLineageMapper
-                        .queryWorkFlowLineageByCode(processLineage.getProcessDefinitionCode(), processLineage.getProjectCode());
-                // sourceWorkFlowId = ""
-                if (!workFlowLineageMap.containsKey(pre.getWorkFlowId())) {
-                    workFlowLineageMap.put(pre.getWorkFlowId(), pre);
+                processLineage.getPostTaskCode(), processLineage.getPostTaskVersion(),
+                processLineage.getProcessDefinitionCode(), processLineage.getProjectCode());
+        if (!relations.isEmpty()) {
+            Set<Integer> preWorkFlowIds = new HashSet<>();
+            List<ProcessLineage> preRelations = workFlowLineageMapper.queryCodeRelation(
+                    processLineage.getPreTaskCode(), processLineage.getPreTaskVersion(),
+                    processLineage.getProcessDefinitionCode(), processLineage.getProjectCode());
+            for (ProcessLineage preRelation : preRelations) {
+                WorkFlowLineage pre = workFlowLineageMapper.queryWorkFlowLineageByCode(
+                        preRelation.getProcessDefinitionCode(), preRelation.getProjectCode());
+                preWorkFlowIds.add(pre.getWorkFlowId());
             }
-
-                WorkFlowLineage post = workFlowLineageMapper
-                        .queryWorkFlowLineageByCode(relation.getProcessDefinitionCode(), relation.getProjectCode());
-
-                if (workFlowLineageMap.containsKey(post.getWorkFlowId())) {
+            ProcessLineage postRelation = relations.get(0);
+            WorkFlowLineage post = workFlowLineageMapper.queryWorkFlowLineageByCode(
+                    postRelation.getProcessDefinitionCode(), postRelation.getProjectCode());
+            if (!workFlowLineageMap.containsKey(post.getWorkFlowId())) {
+                post.setSourceWorkFlowId(StringUtils.join(preWorkFlowIds, ","));
+                workFlowLineageMap.put(post.getWorkFlowId(), post);
+            } else {
                 WorkFlowLineage workFlowLineage = workFlowLineageMap.get(post.getWorkFlowId());
                 String sourceWorkFlowId = workFlowLineage.getSourceWorkFlowId();
                 if (sourceWorkFlowId.equals("")) {
-                        workFlowLineage.setSourceWorkFlowId(String.valueOf(pre.getWorkFlowId()));
+                    workFlowLineage.setSourceWorkFlowId(StringUtils.join(preWorkFlowIds, ","));
                 } else {
-                        workFlowLineage.setSourceWorkFlowId(sourceWorkFlowId + "," + pre.getWorkFlowId());
-
+                    if (!preWorkFlowIds.isEmpty()) {
+                        workFlowLineage.setSourceWorkFlowId(sourceWorkFlowId + "," + StringUtils.join(preWorkFlowIds, ","));
                     }
-
-                } else {
-                    post.setSourceWorkFlowId(String.valueOf(pre.getWorkFlowId()));
-                    workFlowLineageMap.put(post.getWorkFlowId(), post);
                 }
-
-                WorkFlowRelation workFlowRelation = new WorkFlowRelation();
-                workFlowRelation.setSourceWorkFlowId(pre.getWorkFlowId());
-                workFlowRelation.setTargetWorkFlowId(post.getWorkFlowId());
-                if (workFlowRelations.contains(workFlowRelation)) {
-                    continue;
             }
-                workFlowRelations.add(workFlowRelation);
-                getRelation(workFlowLineageMap, workFlowRelations, relation);
+            if (preWorkFlowIds.isEmpty()) {
+                workFlowRelations.add(new WorkFlowRelation(0, post.getWorkFlowId()));
+            } else {
+                for (Integer workFlowId : preWorkFlowIds) {
+                    workFlowRelations.add(new WorkFlowRelation(workFlowId, post.getWorkFlowId()));
+                }
             }
         }
     }

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/WorkFlowRelation.java

@@ -38,6 +38,14 @@ public class WorkFlowRelation {
         this.targetWorkFlowId = targetWorkFlowId;
     }
 
+    public WorkFlowRelation() {
+    }
+
+    public WorkFlowRelation(int sourceWorkFlowId, int targetWorkFlowId) {
+        this.sourceWorkFlowId = sourceWorkFlowId;
+        this.targetWorkFlowId = targetWorkFlowId;
+    }
+
     @Override
     public boolean equals(Object obj) {
         return obj instanceof WorkFlowRelation

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

@@ -59,7 +59,7 @@ public interface WorkFlowLineageMapper {
     /**
      * queryWorkFlowLineageByCode
      *
-     * @param processDefinitionCode processDefinitioncode
+     * @param processDefinitionCode processDefinitionCode
      * @param projectCode projectCode
      * @return WorkFlowLineage
      */

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

@@ -50,19 +50,18 @@
     </select>
 
     <select id="queryCodeRelation" resultType="org.apache.dolphinscheduler.dao.entity.ProcessLineage">
-        select project_code
+        select project_code,
                post_task_code,
                post_task_version,
                pre_task_code,
                pre_task_version,
                process_definition_code,
                process_definition_version
-         from t_ds_process_task_relation ptr
-         where ptr.pre_task_code=#{taskCode}
-                and ptr.pre_task_version=#{taskVersion}
-                and ptr.process_definition_code!=#{processDefinitionCode}
-                and ptr.project_code =#{projectCode}
-
+         from t_ds_process_task_relation
+         where post_task_code = #{taskCode}
+                and post_task_version = #{taskVersion}
+                and process_definition_code = #{processDefinitionCode}
+                and project_code = #{projectCode}
     </select>
 
     <select id="queryWorkFlowLineageByCode" resultType="org.apache.dolphinscheduler.dao.entity.WorkFlowLineage"