[Fix-3236][DagHelper]getFlowNodeListPost/getFlowNodeListPre time complexity O(N^N) (#3249)

* getFlowNodeListPost/getFlowNodeListPre time complexity is O(N^N); This might solve the problem

Change-Id: I07e45c831c7df0089a94a003ca8a2c07e9746892

* getFlowNodeListPost/getFlowNodeListPre time complexity is O(N^N); This might solve the problem

Change-Id: I96f5a8c0ba18656de8981f513470cfdb6bc2f467
Co-authored-by: Nwangyan.61 <wangyan.61@bytedance.com>

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/DagHelper.java

@@ -98,14 +98,16 @@ public class DagHelper {
                 List<TaskNode> childNodeList = new ArrayList<>();
                 if (startNode == null) {
                     logger.error("start node name [{}] is not in task node list [{}] ",
-                        startNodeName,
-                        taskNodeList
+                            startNodeName,
+                            taskNodeList
                     );
                     continue;
                 } else if (TaskDependType.TASK_POST == taskDependType) {
-                    childNodeList = getFlowNodeListPost(startNode, taskNodeList);
+                    List<String> visitedNodeNameList = new ArrayList<>();
+                    childNodeList = getFlowNodeListPost(startNode, taskNodeList, visitedNodeNameList);
                 } else if (TaskDependType.TASK_PRE == taskDependType) {
-                    childNodeList = getFlowNodeListPre(startNode, recoveryNodeNameList, taskNodeList);
+                    List<String> visitedNodeNameList = new ArrayList<>();
+                    childNodeList = getFlowNodeListPre(startNode, recoveryNodeNameList, taskNodeList, visitedNodeNameList);
                 } else {
                     childNodeList.add(startNode);
                 }
@@ -128,14 +130,19 @@ public class DagHelper {
      * @param taskNodeList taskNodeList
      * @return task node list
      */
-    private static List<TaskNode> getFlowNodeListPost(TaskNode startNode, List<TaskNode> taskNodeList) {
+    private static List<TaskNode> getFlowNodeListPost(TaskNode startNode, List<TaskNode> taskNodeList, List<String> visitedNodeNameList) {
         List<TaskNode> resultList = new ArrayList<>();
         for (TaskNode taskNode : taskNodeList) {
             List<String> depList = taskNode.getDepList();
-            if (null != depList && null != startNode && depList.contains(startNode.getName())) {
-                resultList.addAll(getFlowNodeListPost(taskNode, taskNodeList));
+            if (null != depList && null != startNode && depList.contains(startNode.getName()) && !visitedNodeNameList.contains(taskNode.getName())) {
+                resultList.addAll(getFlowNodeListPost(taskNode, taskNodeList, visitedNodeNameList));
             }
         }
+        // why add (startNode != null) condition? for SonarCloud Quality Gate passed
+        if (null != startNode) {
+            visitedNodeNameList.add(startNode.getName());
+        }
+
         resultList.add(startNode);
         return resultList;
     }
@@ -148,7 +155,7 @@ public class DagHelper {
      * @param taskNodeList taskNodeList
      * @return task node list
      */
-    private static List<TaskNode> getFlowNodeListPre(TaskNode startNode, List<String> recoveryNodeNameList, List<TaskNode> taskNodeList) {
+    private static List<TaskNode> getFlowNodeListPre(TaskNode startNode, List<String> recoveryNodeNameList, List<TaskNode> taskNodeList, List<String> visitedNodeNameList) {
 
         List<TaskNode> resultList = new ArrayList<>();
 
@@ -158,16 +165,23 @@ public class DagHelper {
             resultList.add(startNode);
         }
         if (CollectionUtils.isEmpty(depList)) {
+            if (null != startNode) {
+                visitedNodeNameList.add(startNode.getName());
+            }
             return resultList;
         }
         for (String depNodeName : depList) {
             TaskNode start = findNodeByName(taskNodeList, depNodeName);
             if (recoveryNodeNameList.contains(depNodeName)) {
                 resultList.add(start);
-            } else {
-                resultList.addAll(getFlowNodeListPre(start, recoveryNodeNameList, taskNodeList));
+            } else if (!visitedNodeNameList.contains(depNodeName)) {
+                resultList.addAll(getFlowNodeListPre(start, recoveryNodeNameList, taskNodeList, visitedNodeNameList));
             }
         }
+        // why add (startNode != null) condition? for SonarCloud Quality Gate passed
+        if (null != startNode) {
+            visitedNodeNameList.add(startNode.getName());
+        }
         return resultList;
     }
 
@@ -369,7 +383,7 @@ public class DagHelper {
      */
     public static boolean haveConditionsAfterNode(String parentNodeName,
                                                   DAG<String, TaskNode, TaskNodeRelation> dag
-                                            ){
+    ){
         boolean result = false;
         Set<String> subsequentNodes = dag.getSubsequentNodes(parentNodeName);
         if(CollectionUtils.isEmpty(subsequentNodes)){