[Fix-17239][Dependent] Dependent check get wrong result in manual running execution type (#17240)

SbloodyS · web-flow · commit 5f64badb4577 · 2025-06-05T09:53:58.000+08:00
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/WorkflowInstanceMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/WorkflowInstanceMapper.java
@@ -65,6 +65,7 @@ List<WorkflowInstance> queryByHostAndStatus(@Param("host") String host,
      */
     List<WorkflowInstance> queryMainWorkflowByHostAndStatus(@Param("host") String host,
                                                             @Param("states") int[] stateArray);
+
     /**
      * query workflow instance host by stateArray
      *
@@ -108,15 +109,15 @@ List<WorkflowInstance> queryByWorkerGroupNameAndStatus(@Param("workerGroupName")
     /**
      * workflow instance page
      *
-     * @param page                  page
-     * @param projectCode           projectCode
+     * @param page                   page
+     * @param projectCode            projectCode
      * @param workflowDefinitionCode workflowDefinitionCode
-     * @param searchVal             searchVal
-     * @param executorName          executorName
-     * @param statusArray           statusArray
-     * @param host                  host
-     * @param startTime             startTime
-     * @param endTime               endTime
+     * @param searchVal              searchVal
+     * @param executorName           executorName
+     * @param statusArray            statusArray
+     * @param host                   host
+     * @param startTime              startTime
+     * @param endTime                endTime
      * @return workflow instance page
      */
     IPage<WorkflowInstance> queryWorkflowInstanceListPaging(Page<WorkflowInstance> page,
@@ -186,7 +187,7 @@ List<WorkflowInstanceStatusCountDto> countWorkflowInstanceStateByProjectCodes(
      * query workflow instance by workflowDefinitionCode
      *
      * @param workflowDefinitionCode workflowDefinitionCode
-     * @param size                  size
+     * @param size                   size
      * @return workflow instance list
      */
     List<WorkflowInstance> queryByWorkflowDefinitionCode(@Param("workflowDefinitionCode") Long workflowDefinitionCode,
@@ -196,10 +197,10 @@ List<WorkflowInstance> queryByWorkflowDefinitionCode(@Param("workflowDefinitionC
      * query last scheduler workflow instance
      *
      * @param workflowDefinitionCode definitionCode
-     * @param taskDefinitionCode    definitionCode
-     * @param startTime             startTime
-     * @param endTime               endTime
-     * @param testFlag              testFlag
+     * @param taskDefinitionCode     definitionCode
+     * @param startTime              startTime
+     * @param endTime                endTime
+     * @param testFlag               testFlag
      * @return workflow instance
      */
     WorkflowInstance queryLastSchedulerWorkflow(@Param("workflowDefinitionCode") Long workflowDefinitionCode,
@@ -212,10 +213,10 @@ WorkflowInstance queryLastSchedulerWorkflow(@Param("workflowDefinitionCode") Lon
      * query last manual workflow instance
      *
      * @param workflowDefinitionCode workflowDefinitionCode
-     * @param taskCode       taskCode
-     * @param startTime      startTime
-     * @param endTime        endTime
-     * @param testFlag       testFlag
+     * @param taskCode               taskCode
+     * @param startTime              startTime
+     * @param endTime                endTime
+     * @param testFlag               testFlag
      * @return workflow instance
      */
     WorkflowInstance queryLastManualWorkflow(@Param("workflowDefinitionCode") Long workflowDefinitionCode,
@@ -224,6 +225,11 @@ WorkflowInstance queryLastManualWorkflow(@Param("workflowDefinitionCode") Long w
                                              @Param("endTime") Date endTime,
                                              @Param("testFlag") int testFlag);
 
+    WorkflowInstance queryLastRunningWorkflow(@Param("workflowDefinitionCode") Long workflowDefinitionCode,
+                                              @Param("startTime") Date startTime,
+                                              @Param("endTime") Date endTime,
+                                              @Param("states") int[] stateArray);
+
     /**
      * query first schedule workflow instance
      *
@@ -261,7 +267,7 @@ List<WorkflowInstance> queryTopNWorkflowInstance(@Param("size") int size,
      * query workflow instance by workflowDefinitionCode and stateArray
      *
      * @param workflowDefinitionCode workflowDefinitionCode
-     * @param states                states array
+     * @param states                 states array
      * @return workflow instance list
      */
 
@@ -275,12 +281,12 @@ List<WorkflowInstance> queryByWorkflowCodeVersionStatus(@Param("workflowDefiniti
     /**
      * Filter workflow instance
      *
-     * @param page                  page
+     * @param page                   page
      * @param workflowDefinitionCode workflowDefinitionCode
-     * @param name                  name
-     * @param host                  host
-     * @param startTime             startTime
-     * @param endTime               endTime
+     * @param name                   name
+     * @param host                   host
+     * @param startTime              startTime
+     * @param endTime                endTime
      * @return workflow instance IPage
      */
     IPage<WorkflowInstance> queryWorkflowInstanceListV2Paging(Page<WorkflowInstance> page,
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/WorkflowInstanceDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/WorkflowInstanceDao.java
@@ -68,6 +68,8 @@ WorkflowInstance queryLastSchedulerWorkflowInterval(Long workflowDefinitionCode,
     WorkflowInstance queryLastManualWorkflowInterval(Long definitionCode, Long taskCode, DateInterval dateInterval,
                                                      int testFlag);
 
+    WorkflowInstance queryLastRunningWorkflowInterval(Long definitionCode, DateInterval dateInterval);
+
     /**
      * query first schedule workflow instance
      *
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/WorkflowInstanceDaoImpl.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/WorkflowInstanceDaoImpl.java
@@ -120,6 +120,16 @@ public WorkflowInstance queryLastManualWorkflowInterval(Long definitionCode, Lon
                 testFlag);
     }
 
+    @Override
+    public WorkflowInstance queryLastRunningWorkflowInterval(Long definitionCode, DateInterval dateInterval) {
+        int[] runningStateArray = new int[]{WorkflowExecutionStatus.SUBMITTED_SUCCESS.ordinal(),
+                WorkflowExecutionStatus.RUNNING_EXECUTION.ordinal(),
+                WorkflowExecutionStatus.READY_PAUSE.ordinal(),
+                WorkflowExecutionStatus.READY_STOP.ordinal()};
+        return mybatisMapper.queryLastRunningWorkflow(definitionCode, dateInterval.getStartTime(),
+                dateInterval.getEndTime(), runningStateArray);
+    }
+
     /**
      * query first schedule process instance
      *
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkflowInstanceMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkflowInstanceMapper.xml
@@ -249,6 +249,24 @@
         order by t1.end_time desc limit 1
     </select>
 
+    <select id="queryLastRunningWorkflow" resultType="org.apache.dolphinscheduler.dao.entity.WorkflowInstance">
+        select
+        <include refid="baseSql"/>
+        from t_ds_workflow_instance
+        where workflow_definition_code=#{workflowDefinitionCode}
+        <if test="states !=null and states.length != 0">
+            and state in
+            <foreach collection="states" item="i" index="index" open="(" separator="," close=")">
+                #{i}
+            </foreach>
+        </if>
+        <if test="startTime!=null and endTime != null ">
+            and ((schedule_time <![CDATA[ >= ]]> #{startTime} and schedule_time <![CDATA[ <= ]]> #{endTime})
+            or (start_time <![CDATA[ >= ]]> #{startTime} and start_time <![CDATA[ <= ]]> #{endTime}))
+        </if>
+        order by start_time desc limit 1
+    </select>
+
     <select id="queryFirstScheduleWorkflowInstance" resultType="org.apache.dolphinscheduler.dao.entity.WorkflowInstance">
         select
         <include refid="baseSql"/>
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/utils/DependentExecute.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/utils/DependentExecute.java
@@ -155,7 +155,7 @@ private DependResult calculateResultForTasks(DependentItem dependentItem,
         DependResult result = DependResult.FAILED;
         for (DateInterval dateInterval : dateIntervals) {
             WorkflowInstance workflowInstance =
-                    findLastWorkflowInterval(dependentItem.getDefinitionCode(), dependentItem.getDepTaskCode(),
+                    findDependentWorkflowCandidate(dependentItem.getDefinitionCode(), dependentItem.getDepTaskCode(),
                             dateInterval, testFlag);
             if (workflowInstance == null) {
                 return DependResult.WAITING;
@@ -314,17 +314,24 @@ private void addItemVarPool(String varPoolStr, Long endTime) {
     }
 
     /**
-     * find the last one workflow instance that :
-     * 1. manual run and finish between the interval
-     * 2. schedule run and schedule time between the interval
+     * find the last one workflow instance that:
+     * 1. running workflow instance in the date interval
+     * 2. manual run and finish between the interval
+     * 3. schedule run and schedule time between the interval
      *
      * @param definitionCode definition code
      * @param taskCode task code
      * @param dateInterval   date interval
      * @return workflowInstance
      */
-    private WorkflowInstance findLastWorkflowInterval(Long definitionCode, Long taskCode, DateInterval dateInterval,
-                                                      int testFlag) {
+    private WorkflowInstance findDependentWorkflowCandidate(Long definitionCode, Long taskCode,
+                                                            DateInterval dateInterval,
+                                                            int testFlag) {
+        WorkflowInstance runningWorkflow =
+                workflowInstanceDao.queryLastRunningWorkflowInterval(definitionCode, dateInterval);
+        if (runningWorkflow != null) {
+            return runningWorkflow;
+        }
 
         WorkflowInstance lastSchedulerWorkflowInstance =
                 workflowInstanceDao.queryLastSchedulerWorkflowInterval(definitionCode, taskCode, dateInterval,

Original file line number	Diff line number	Diff line change
`@@ -68,6 +68,8 @@ WorkflowInstance queryLastSchedulerWorkflowInterval(Long workflowDefinitionCode,`
`68`	`68`	`WorkflowInstance queryLastManualWorkflowInterval(Long definitionCode, Long taskCode, DateInterval dateInterval,`
`69`	`69`	`int testFlag);`
`70`	`70`
	`71`	`+ WorkflowInstance queryLastRunningWorkflowInterval(Long definitionCode, DateInterval dateInterval);`
	`72`	`+`
`71`	`73`	`/**`
`72`	`74`	`* query first schedule workflow instance`
`73`	`75`	`*`