From c0435e5bd88ab8712942cfecf9e1fefa0d886773 Mon Sep 17 00:00:00 2001
From: Evan Sun <abzymeinsjtu@gmail.com>
Date: Wed, 3 Apr 2024 13:56:41 +0800
Subject: [PATCH] fix dependent task logic (#15795)

Co-authored-by: abzymeinsjtu <abzymeinsjtu@B-54Q8MD6R-0244.local>
---
 .../dao/mapper/ProcessInstanceMapper.java              |  6 ++++--
 .../dao/repository/ProcessInstanceDao.java             |  6 ++++--
 .../dao/repository/impl/ProcessInstanceDaoImpl.java    | 10 +++++++---
 .../dao/mapper/ProcessInstanceMapper.xml               |  9 +++++++--
 .../dao/mapper/ProcessInstanceMapperTest.java          |  3 ++-
 .../server/master/utils/DependentExecute.java          |  2 +-
 6 files changed, 25 insertions(+), 11 deletions(-)

diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java
index 20c259da9b..943b4ac2db 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java
@@ -188,13 +188,15 @@ public interface ProcessInstanceMapper extends BaseMapper<ProcessInstance> {
     /**
      * query last scheduler process instance
      *
-     * @param definitionCode definitionCode
+     * @param processDefinitionCode definitionCode
+     * @param taskDefinitionCode definitionCode
      * @param startTime      startTime
      * @param endTime        endTime
      * @param testFlag       testFlag
      * @return process instance
      */
-    ProcessInstance queryLastSchedulerProcess(@Param("processDefinitionCode") Long definitionCode,
+    ProcessInstance queryLastSchedulerProcess(@Param("processDefinitionCode") Long processDefinitionCode,
+                                              @Param("taskDefinitionCode") Long taskDefinitionCode,
                                               @Param("startTime") Date startTime,
                                               @Param("endTime") Date endTime,
                                               @Param("testFlag") int testFlag);
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/ProcessInstanceDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/ProcessInstanceDao.java
index 02703fe0f3..91c567e2e3 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/ProcessInstanceDao.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/ProcessInstanceDao.java
@@ -41,11 +41,13 @@ public interface ProcessInstanceDao extends IDao<ProcessInstance> {
     /**
      * find last scheduler process instance in the date interval
      *
-     * @param definitionCode definitionCode
+     * @param processDefinitionCode definitionCode
+     * @param taskDefinitionCode definitionCode
      * @param dateInterval   dateInterval
      * @return process instance
      */
-    ProcessInstance queryLastSchedulerProcessInterval(Long definitionCode, DateInterval dateInterval, int testFlag);
+    ProcessInstance queryLastSchedulerProcessInterval(Long processDefinitionCode, Long taskDefinitionCode,
+                                                      DateInterval dateInterval, int testFlag);
 
     /**
      * find last manual process instance interval
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/ProcessInstanceDaoImpl.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/ProcessInstanceDaoImpl.java
index 8ff93b3a9f..8e966c1850 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/ProcessInstanceDaoImpl.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/ProcessInstanceDaoImpl.java
@@ -67,14 +67,18 @@ public class ProcessInstanceDaoImpl extends BaseDao<ProcessInstance, ProcessInst
     /**
      * find last scheduler process instance in the date interval
      *
-     * @param definitionCode definitionCode
+     * @param processDefinitionCode definitionCode
+     * @param taskDefinitionCode definitionCode
      * @param dateInterval   dateInterval
      * @return process instance
      */
     @Override
-    public ProcessInstance queryLastSchedulerProcessInterval(Long definitionCode, DateInterval dateInterval,
+    public ProcessInstance queryLastSchedulerProcessInterval(Long processDefinitionCode, Long taskDefinitionCode,
+                                                             DateInterval dateInterval,
                                                              int testFlag) {
-        return mybatisMapper.queryLastSchedulerProcess(definitionCode,
+        return mybatisMapper.queryLastSchedulerProcess(
+                processDefinitionCode,
+                taskDefinitionCode,
                 dateInterval.getStartTime(),
                 dateInterval.getEndTime(),
                 testFlag);
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml
index 095e266ed5..b31bb571d7 100644
--- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml
+++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml
@@ -197,12 +197,17 @@
         order by start_time desc limit #{size}
     </select>
     <select id="queryLastSchedulerProcess" resultType="org.apache.dolphinscheduler.dao.entity.ProcessInstance">
-        select
+        select t1.* from (select
         <include refid="baseSql"/>
         from t_ds_process_instance
         where process_definition_code=#{processDefinitionCode} and test_flag=#{testFlag}
-        <if test="startTime!=null and endTime != null ">
+        <if test="startTime != null and endTime != null ">
             and schedule_time <![CDATA[ >= ]]> #{startTime} and schedule_time <![CDATA[ <= ]]> #{endTime}
+        </if>) as t1
+        <if test="taskDefinitionCode != null and taskDefinitionCode != 0 and taskDefinitionCode != -1">
+            inner join
+            t_ds_task_instance as t2
+            on t1.id = t2.process_instance_id and t2.task_code=#{taskDefinitionCode}
         </if>
         order by end_time desc limit 1
     </select>
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapperTest.java
index 39b8d04e4d..0aae0dce2b 100644
--- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapperTest.java
+++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapperTest.java
@@ -263,7 +263,8 @@ public class ProcessInstanceMapperTest extends BaseDaoTest {
         processInstanceMapper.updateById(processInstance);
 
         ProcessInstance processInstance1 =
-                processInstanceMapper.queryLastSchedulerProcess(processInstance.getProcessDefinitionCode(), null, null,
+                processInstanceMapper.queryLastSchedulerProcess(processInstance.getProcessDefinitionCode(), 0L, null,
+                        null,
                         processInstance.getTestFlag());
         Assertions.assertNotEquals(null, processInstance1);
         processInstanceMapper.deleteById(processInstance.getId());
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
index 15ab34de34..28f9fd682b 100644
--- 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
@@ -320,7 +320,7 @@ public class DependentExecute {
                                                     int testFlag) {
 
         ProcessInstance lastSchedulerProcess =
-                processInstanceDao.queryLastSchedulerProcessInterval(definitionCode, dateInterval, testFlag);
+                processInstanceDao.queryLastSchedulerProcessInterval(definitionCode, taskCode, dateInterval, testFlag);
 
         ProcessInstance lastManualProcess =
                 processInstanceDao.queryLastManualProcessInterval(definitionCode, taskCode, dateInterval, testFlag);