[Improvement-17057][Master]Check if the WorkflowGrapth is a dag at constructor(#17057)

Author: yingh0ng

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraph.java

@@ -19,16 +19,15 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import org.apache.dolphinscheduler.common.graph.DAG;
 import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
 import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelation;
 
-import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Queue;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -61,57 +60,18 @@ public WorkflowGraph(List<WorkflowTaskRelation> workflowTaskRelations, List<Task
     }
 
     private void checkIfDAG(List<WorkflowTaskRelation> workflowTaskRelations, List<TaskDefinition> taskDefinitions) {
-        // If topology-sort-result`s size less than taskDefinitions`s size, then not a DAG
-        Map<Long, List<Long>> preTaskCodeMap = workflowTaskRelations
-                .stream()
-                .collect(Collectors.groupingBy(WorkflowTaskRelation::getPostTaskCode,
-                        Collectors.mapping(WorkflowTaskRelation::getPreTaskCode, Collectors.toList())));
-        Map<Long, List<Long>> postTaskCodeMap = workflowTaskRelations
-                .stream()
-                .collect(Collectors.groupingBy(WorkflowTaskRelation::getPreTaskCode,
-                        Collectors.mapping(WorkflowTaskRelation::getPostTaskCode, Collectors.toList())));
-
-        // build in-degree count
-        Map<Long, Integer> inDegreeCount = new HashMap<>();
+        DAG<Long, TaskDefinition, WorkflowTaskRelation> graph = new DAG<>();
+        // Fill the vertices
         for (TaskDefinition taskDefinition : taskDefinitions) {
-            List<Long> preTasks = preTaskCodeMap.get(taskDefinition.getCode());
-            if (preTasks == null) {
-                inDegreeCount.put(taskDefinition.getCode(), 0);
-            } else {
-                inDegreeCount.put(taskDefinition.getCode(), preTasks.size());
-            }
-        }
-
-        // Adds the task with zero-in-degree to the queue
-        Set<Long> visitTable = new HashSet<>();
-        Queue<Long> queue = new ArrayDeque<>();
-        for (Map.Entry<Long, Integer> entry : inDegreeCount.entrySet()) {
-            if (entry.getValue() == 0 && visitTable.add(entry.getKey())) {
-                queue.offer(entry.getKey());
-            }
+            graph.addNode(taskDefinition.getCode(), taskDefinition);
         }
-
-        // topology sort
-        Set<Long> resultTable = new HashSet<>();
-        while (!queue.isEmpty()) {
-            Long taskCode = queue.poll();
-            resultTable.add(taskCode);
-
-            List<Long> postCodes = postTaskCodeMap.get(taskCode);
-            if (postCodes == null) {
-                continue;
+        // Fill edge relations
+        for (WorkflowTaskRelation relation : workflowTaskRelations) {
+            long preTaskCode = relation.getPreTaskCode();
+            // When exist a ring cycle, then not a DAG.
+            if (preTaskCode != 0 && !graph.addEdge(preTaskCode, relation.getPostTaskCode())) {
+                throw new IllegalArgumentException("The workflow graph is not a DAG");
             }
-            for (Long postCode : postCodes) {
-                inDegreeCount.put(postCode, inDegreeCount.get(postCode) - 1);
-
-                if (inDegreeCount.get(postCode) == 0) {
-                    queue.offer(postCode);
-                }
-            }
-        }
-
-        if (resultTable.size() < taskDefinitions.size()) {
-            throw new IllegalArgumentException("The workflow task relation is not a DAG");
         }
     }
 

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/graph/WorkflowGraphCheckIfDAGTest.java

@@ -43,6 +43,11 @@ public void checkIfDAGSingleTaskNoException() {
         TaskDefinition taskDefinition = new TaskDefinition(1L, 1);
         taskDefinitions.add(taskDefinition);
 
+        WorkflowTaskRelation relation = new WorkflowTaskRelation();
+        relation.setPreTaskCode(0);
+        relation.setPostTaskCode(1);
+        workflowTaskRelations.add(relation);
+
         Assertions.assertDoesNotThrow(() -> new WorkflowGraph(workflowTaskRelations, taskDefinitions));
     }